Correctly wait background threads

There are some places that make it possible to trigger use-after-free
from threads because some global variable had been destroyed before, for
example some logger.

Signed-off-by: Azat Khuzhin <a.khuzhin@semrush.com>

programs/keeper/Keeper.cpp

@@ -328,6 +328,13 @@ try
         config().getUInt("max_thread_pool_free_size", 1000),
         config().getUInt("thread_pool_queue_size", 10000)
     );
+    /// Wait for all threads to avoid possible use-after-free (for example logging objects can be already destroyed).
+    SCOPE_EXIT({
+        Stopwatch watch;
+        LOG_INFO(log, "Waiting for background threads");
+        GlobalThreadPool::instance().shutdown();
+        LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds());
+    });
 
     static ServerErrorHandler error_handler;
     Poco::ErrorHandler::set(&error_handler);

programs/server/Server.cpp

@@ -697,6 +697,14 @@ try
         server_settings.max_thread_pool_size,
         server_settings.max_thread_pool_free_size,
         server_settings.thread_pool_queue_size);
+    /// Wait for all threads to avoid possible use-after-free (for example logging objects can be already destroyed).
+    SCOPE_EXIT({
+        Stopwatch watch;
+        LOG_INFO(log, "Waiting for background threads");
+        GlobalThreadPool::instance().shutdown();
+        LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds());
+    });
+
 
 #if USE_AZURE_BLOB_STORAGE
     /// It makes sense to deinitialize libxml after joining of all threads

src/Common/ThreadPool.cpp

@@ -500,3 +500,10 @@ GlobalThreadPool & GlobalThreadPool::instance()
 
     return *the_instance;
 }
+void GlobalThreadPool::shutdown()
+{
+    if (the_instance)
+    {
+        the_instance->finalize();
+    }
+}

src/Common/ThreadPool.h

@@ -109,6 +109,8 @@ public:
     void addOnDestroyCallback(OnDestroyCallback && callback);
 
 private:
+    friend class GlobalThreadPool;
+
     mutable std::mutex mutex;
     std::condition_variable job_finished;
     std::condition_variable new_job_or_shutdown;
@@ -205,6 +207,7 @@ class GlobalThreadPool : public FreeThreadPool, private boost::noncopyable
 public:
     static void initialize(size_t max_threads = 10000, size_t max_free_threads = 1000, size_t queue_size = 10000);
     static GlobalThreadPool & instance();
+    static void shutdown();
 };