Join threads from global thread pool before exiting from main

programs/benchmark/Benchmark.cpp

@@ -12,6 +12,8 @@
 #include <Common/Config/parseConnectionCredentials.h>
 #include <Common/Stopwatch.h>
 #include <Common/ThreadPool.h>
+#include <IO/SharedThreadPools.h>
+#include <Common/scope_guard_safe.h>
 #include <AggregateFunctions/ReservoirSampler.h>
 #include <AggregateFunctions/registerAggregateFunctions.h>
 #include <Client/ClientBaseHelpers.h>
@@ -898,6 +900,13 @@ int mainEntryClickHouseBenchmark(int argc, char ** argv)
     using namespace DB;
     bool print_stacktrace = false;
 
+    /// Join global-pool threads before the statics they may have accessed are destroyed.
+    /// That way, accesses happen-before destruction.
+    SCOPE_EXIT_SAFE({
+        DB::StaticThreadPool::shutdownAll();
+        GlobalThreadPool::shutdown();
+    });
+
     try
     {
         using boost::program_options::value;

programs/compressor/Compressor.cpp

@@ -22,6 +22,8 @@
 #include <Compression/CompressionFactory.h>
 #include <Common/TerminalSize.h>
 #include <Common/ThreadPool.h>
+#include <IO/SharedThreadPools.h>
+#include <Common/scope_guard_safe.h>
 #include <Common/CurrentMetrics.h>
 #include <Core/Defines.h>
 
@@ -76,6 +78,14 @@ int mainEntryClickHouseCompressor(int argc, char ** argv)
     namespace po = boost::program_options;
 
     bool print_stacktrace = false;
+
+    /// Join global-pool threads before the statics they may have accessed are destroyed.
+    /// That way, accesses happen-before destruction.
+    SCOPE_EXIT_SAFE({
+        DB::StaticThreadPool::shutdownAll();
+        GlobalThreadPool::shutdown();
+    });
+
     try
     {
         po::options_description desc = createOptionsDescription("Allowed options", getTerminalWidth());

programs/disks/DisksApp.cpp

@@ -32,6 +32,8 @@
 #include <Utils.h>
 #include <Server/CloudPlacementInfo.h>
 #include <IO/SharedThreadPools.h>
+#include <Common/ThreadPool.h>
+#include <Common/scope_guard_safe.h>
 
 #include <Poco/FileChannel.h>
 
@@ -627,6 +629,13 @@ void DisksApp::runInteractive()
 int mainEntryClickHouseDisks(int argc, char ** argv);
 int mainEntryClickHouseDisks(int argc, char ** argv)
 {
+    /// Join global-pool threads before the statics they may have accessed are destroyed.
+    /// That way, accesses happen-before destruction.
+    SCOPE_EXIT_SAFE({
+        DB::StaticThreadPool::shutdownAll();
+        GlobalThreadPool::shutdown();
+    });
+
     try
     {
         DB::DisksApp app;

programs/keeper-bench/KeeperBench.cpp

@@ -3,6 +3,9 @@
 #include <Runner.h>
 #include <Common/Exception.h>
 #include <Common/TerminalSize.h>
+#include <Common/ThreadPool.h>
+#include <IO/SharedThreadPools.h>
+#include <Common/scope_guard_safe.h>
 #include <Core/Types.h>
 #include <boost/program_options/variables_map.hpp>
 
@@ -30,6 +33,13 @@ int mainEntryClickHouseKeeperBench(int argc, char ** argv)
     //Poco::Logger::root().setChannel(channel);
     //Poco::Logger::root().setLevel("trace");
 
+    /// Join global-pool threads before the statics they may have accessed are destroyed.
+    /// That way, accesses happen-before destruction.
+    SCOPE_EXIT_SAFE({
+        DB::StaticThreadPool::shutdownAll();
+        GlobalThreadPool::shutdown();
+    });
+
     try
     {
         using boost::program_options::value;

programs/keeper/Keeper.cpp

@@ -17,6 +17,7 @@
 #include <Common/ErrorHandlers.h>
 #include <Common/assertProcessUserMatchesDataOwner.h>
 #include <Common/makeSocketAddress.h>
+#include <IO/SharedThreadPools.h>
 #include <Server/waitServersToFinish.h>
 #include <Server/CloudPlacementInfo.h>
 #include <base/getMemoryAmount.h>
@@ -388,6 +389,7 @@ try
     SCOPE_EXIT({
         Stopwatch watch;
         LOG_INFO(log, "Waiting for background threads");
+        DB::StaticThreadPool::shutdownAll();
         GlobalThreadPool::instance().shutdown();
         LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds());
     });

programs/local/LocalServer.cpp

@@ -38,6 +38,7 @@
 #include <Common/TLDListsHolder.h>
 #include <Common/quoteString.h>
 #include <Common/ThreadPool.h>
+#include <Common/scope_guard_safe.h>
 #include <Common/CurrentMetrics.h>
 #include <Common/NamedCollections/NamedCollectionsFactory.h>
 #include <Common/Jemalloc.h>
@@ -1406,6 +1407,13 @@ int mainEntryClickHouseLocal(int argc, char ** argv)
 {
     DB::MainThreadStatus::getInstance();
 
+    /// Join global-pool threads before the statics they may have accessed are destroyed.
+    /// That way, accesses happen-before destruction.
+    SCOPE_EXIT_SAFE({
+        DB::StaticThreadPool::shutdownAll();
+        GlobalThreadPool::shutdown();
+    });
+
     try
     {
         DB::LocalServer app;

programs/obfuscator/Obfuscator.cpp

@@ -51,6 +51,9 @@
 #include <boost/container/flat_map.hpp>
 #include <Common/TerminalSize.h>
 #include <Common/ErrnoException.h>
+#include <Common/ThreadPool.h>
+#include <IO/SharedThreadPools.h>
+#include <Common/scope_guard_safe.h>
 #include <bit>
 
 
@@ -1216,6 +1219,13 @@ try
     using namespace DB;
     namespace po = boost::program_options;
 
+    /// Join global-pool threads before the statics they may have accessed are destroyed.
+    /// That way, accesses happen-before destruction.
+    SCOPE_EXIT_SAFE({
+        DB::StaticThreadPool::shutdownAll();
+        GlobalThreadPool::shutdown();
+    });
+
     registerFormats();
 
     po::options_description description = createOptionsDescription("Options", getTerminalWidth());

programs/server/Server.cpp

@@ -1522,6 +1522,7 @@ try
     SCOPE_EXIT_SAFE({
         Stopwatch watch;
         LOG_INFO(log, "Waiting for background threads");
+        DB::StaticThreadPool::shutdownAll();
         GlobalThreadPool::instance().shutdown();
         LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds());
     });

src/Common/tests/gtest_main.cpp

@@ -2,6 +2,9 @@
 
 #include <base/sanitizer_options.h>
 
+#include <Common/ThreadPool.h>
+#include <Common/scope_guard_safe.h>
+#include <IO/SharedThreadPools.h>
 #include <Common/tests/gtest_global_context.h>
 
 class ContextEnvironment : public testing::Environment
@@ -13,6 +16,13 @@ public:
 
 int main(int argc, char ** argv)
 {
+    /// Join global-pool threads before the statics they may have accessed are destroyed.
+    /// That way, accesses happen-before destruction.
+    SCOPE_EXIT_SAFE({
+        DB::StaticThreadPool::shutdownAll();
+        GlobalThreadPool::shutdown();
+    });
+
     testing::InitGoogleTest(&argc, argv);
 
     auto & options = getTestCommandLineOptions();

src/IO/SharedThreadPools.cpp

@@ -106,6 +106,12 @@ bool StaticThreadPool::isInitialized() const
     return instance.operator bool();
 }
 
+void StaticThreadPool::shutdown()
+{
+    std::lock_guard lock(mutex);
+    instance.reset();
+}
+
 void StaticThreadPool::reloadConfiguration(size_t max_threads, size_t max_free_threads, size_t queue_size)
 {
     if (!instance)
@@ -163,77 +169,22 @@ void StaticThreadPool::setMaxTurboThreads(size_t max_threads_turbo_)
         instance->setMaxThreads(max_threads_turbo);
 }
 
-StaticThreadPool & getIOThreadPool()
-{
-    static StaticThreadPool instance("IOThreadPool", CurrentMetrics::IOThreads, CurrentMetrics::IOThreadsActive, CurrentMetrics::IOThreadsScheduled);
-    return instance;
+/// NOLINTBEGIN(bugprone-macro-parentheses)
+#define DEFINE_STATIC_THREAD_POOL_GETTER(SUFFIX, NAME, METRIC) \
+StaticThreadPool & get##SUFFIX##ThreadPool() \
+{ \
+    static StaticThreadPool instance(NAME, CurrentMetrics::METRIC##Threads, CurrentMetrics::METRIC##ThreadsActive, CurrentMetrics::METRIC##ThreadsScheduled); \
+    return instance; \
 }
+/// NOLINTEND(bugprone-macro-parentheses)
+APPLY_FOR_STATIC_THREAD_POOLS(DEFINE_STATIC_THREAD_POOL_GETTER)
+#undef DEFINE_STATIC_THREAD_POOL_GETTER
 
-StaticThreadPool & getBackupsIOThreadPool()
+void StaticThreadPool::shutdownAll()
 {
-    static StaticThreadPool instance("BackupsIOThreadPool", CurrentMetrics::BackupsIOThreads, CurrentMetrics::BackupsIOThreadsActive, CurrentMetrics::BackupsIOThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getFetchPartitionThreadPool()
-{
-    static StaticThreadPool instance("MergeTreeFetchPartitionThreadPool", CurrentMetrics::MergeTreeFetchPartitionThreads, CurrentMetrics::MergeTreeFetchPartitionThreadsActive, CurrentMetrics::MergeTreeFetchPartitionThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getActivePartsLoadingThreadPool()
-{
-    static StaticThreadPool instance("MergeTreePartsLoaderThreadPool", CurrentMetrics::MergeTreePartsLoaderThreads, CurrentMetrics::MergeTreePartsLoaderThreadsActive, CurrentMetrics::MergeTreePartsLoaderThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getSnapshotCommitThreadPool()
-{
-    static StaticThreadPool instance("MergeTreeSnapshotCommitThreadPool", CurrentMetrics::MergeTreeSnapshotCommitThreads, CurrentMetrics::MergeTreeSnapshotCommitThreadsActive, CurrentMetrics::MergeTreeSnapshotCommitThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getPartsCleaningThreadPool()
-{
-    static StaticThreadPool instance("MergeTreePartsCleanerThreadPool", CurrentMetrics::MergeTreePartsCleanerThreads, CurrentMetrics::MergeTreePartsCleanerThreadsActive, CurrentMetrics::MergeTreePartsCleanerThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getOutdatedPartsLoadingThreadPool()
-{
-    static StaticThreadPool instance("MergeTreeOutdatedPartsLoaderThreadPool", CurrentMetrics::MergeTreeOutdatedPartsLoaderThreads, CurrentMetrics::MergeTreeOutdatedPartsLoaderThreadsActive, CurrentMetrics::MergeTreeOutdatedPartsLoaderThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getUnexpectedPartsLoadingThreadPool()
-{
-    static StaticThreadPool instance("MergeTreeUnexpectedPartsLoaderThreadPool", CurrentMetrics::MergeTreeUnexpectedPartsLoaderThreads, CurrentMetrics::MergeTreeUnexpectedPartsLoaderThreadsActive, CurrentMetrics::MergeTreeUnexpectedPartsLoaderThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getDatabaseReplicatedCreateTablesThreadPool()
-{
-    static StaticThreadPool instance("CreateTablesThreadPool", CurrentMetrics::DatabaseReplicatedCreateTablesThreads, CurrentMetrics::DatabaseReplicatedCreateTablesThreadsActive, CurrentMetrics::DatabaseReplicatedCreateTablesThreadsScheduled);
-    return instance;
-}
-
-/// ThreadPool used for dropping tables.
-StaticThreadPool & getDatabaseCatalogDropTablesThreadPool()
-{
-    static StaticThreadPool instance("DropTablesThreadPool", CurrentMetrics::DatabaseCatalogThreads, CurrentMetrics::DatabaseCatalogThreadsActive, CurrentMetrics::DatabaseCatalogThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getMergeTreePrefixesDeserializationThreadPool()
-{
-    static StaticThreadPool instance("MergeTreePrefixesDeserializationThreadPool", CurrentMetrics::MergeTreeSubcolumnsReaderThreads, CurrentMetrics::MergeTreeSubcolumnsReaderThreadsActive, CurrentMetrics::MergeTreeSubcolumnsReaderThreadsScheduled);
-    return instance;
-}
-
-StaticThreadPool & getFormatParsingThreadPool()
-{
-    static StaticThreadPool instance("FormatParsingThreadPool", CurrentMetrics::FormatParsingThreads, CurrentMetrics::FormatParsingThreadsActive, CurrentMetrics::FormatParsingThreadsScheduled);
-    return instance;
+#define SHUTDOWN_STATIC_THREAD_POOL(SUFFIX, NAME, METRIC) get##SUFFIX##ThreadPool().shutdown();
+    APPLY_FOR_STATIC_THREAD_POOLS(SHUTDOWN_STATIC_THREAD_POOL)
+#undef SHUTDOWN_STATIC_THREAD_POOL
 }
 
 }

src/IO/SharedThreadPools.h

@@ -27,6 +27,9 @@ public:
     bool isInitialized() const;
     void reloadConfiguration(size_t max_threads, size_t max_free_threads, size_t queue_size);
 
+    void shutdown();
+    static void shutdownAll();
+
     /// Server and clickhouse-local initialize all thread pools on startup, with settings from config.
     /// Client and misc tools may initialize the pools they use lazily using this method instead.
     void initializeWithDefaultSettingsIfNotInitialized();
@@ -54,41 +57,22 @@ private:
     void initializeImpl(size_t max_threads, size_t max_free_threads, size_t queue_size);
 };
 
-/// ThreadPool used for the IO.
-StaticThreadPool & getIOThreadPool();
+#define APPLY_FOR_STATIC_THREAD_POOLS(M) \
+    M(IO, "IOThreadPool", IO) \
+    M(BackupsIO, "BackupsIOThreadPool", BackupsIO) \
+    M(FetchPartition, "MergeTreeFetchPartitionThreadPool", MergeTreeFetchPartition) \
+    M(ActivePartsLoading, "MergeTreePartsLoaderThreadPool", MergeTreePartsLoader) \
+    M(SnapshotCommit, "MergeTreeSnapshotCommitThreadPool", MergeTreeSnapshotCommit) \
+    M(PartsCleaning, "MergeTreePartsCleanerThreadPool", MergeTreePartsCleaner) \
+    M(OutdatedPartsLoading, "MergeTreeOutdatedPartsLoaderThreadPool", MergeTreeOutdatedPartsLoader) \
+    M(UnexpectedPartsLoading, "MergeTreeUnexpectedPartsLoaderThreadPool", MergeTreeUnexpectedPartsLoader) \
+    M(DatabaseReplicatedCreateTables, "CreateTablesThreadPool", DatabaseReplicatedCreateTables) \
+    M(DatabaseCatalogDropTables, "DropTablesThreadPool", DatabaseCatalog) \
+    M(MergeTreePrefixesDeserialization, "MergeTreePrefixesDeserializationThreadPool", MergeTreeSubcolumnsReader) \
+    M(FormatParsing, "FormatParsingThreadPool", FormatParsing)
 
-/// ThreadPool used for the Backup IO.
-StaticThreadPool & getBackupsIOThreadPool();
-
-/// ThreadPool used for FETCH PARTITION
-StaticThreadPool & getFetchPartitionThreadPool();
-
-/// ThreadPool used for the loading of Outdated data parts for MergeTree tables.
-StaticThreadPool & getActivePartsLoadingThreadPool();
-
-/// ThreadPool used for commit snapshot.
-StaticThreadPool & getSnapshotCommitThreadPool();
-
-/// ThreadPool used for deleting data parts for MergeTree tables.
-StaticThreadPool & getPartsCleaningThreadPool();
-
-/// This ThreadPool is used for the loading of Outdated data parts for MergeTree tables.
-/// Normally we will just load Outdated data parts concurrently in background, but in
-/// case when we need to synchronously wait for the loading to be finished, we can increase
-/// the number of threads by calling enableTurboMode() :-)
-StaticThreadPool & getOutdatedPartsLoadingThreadPool();
-
-StaticThreadPool & getUnexpectedPartsLoadingThreadPool();
-
-/// ThreadPool used for creating tables in DatabaseReplicated.
-StaticThreadPool & getDatabaseReplicatedCreateTablesThreadPool();
-
-/// ThreadPool used for dropping tables.
-StaticThreadPool & getDatabaseCatalogDropTablesThreadPool();
-
-/// ThreadPool used for parallel prefixes deserialization of subcolumns in Wide MergeTree parts.
-StaticThreadPool & getMergeTreePrefixesDeserializationThreadPool();
-
-StaticThreadPool & getFormatParsingThreadPool();
+#define DECLARE_STATIC_THREAD_POOL_GETTER(SUFFIX, NAME, METRIC) StaticThreadPool & get##SUFFIX##ThreadPool();
+APPLY_FOR_STATIC_THREAD_POOLS(DECLARE_STATIC_THREAD_POOL_GETTER)
+#undef DECLARE_STATIC_THREAD_POOL_GETTER
 
 }

tests/integration/test_thread_pool_free_size_shutdown/configs/thread_pools.xml

@@ -0,0 +1,8 @@
+<clickhouse>
+    <!-- Keep one idle worker in every configurable static thread pool, so they retain a
+         worker parked on the global thread pool after use. -->
+    <max_io_thread_pool_free_size>1</max_io_thread_pool_free_size>
+    <max_backups_io_thread_pool_free_size>1</max_backups_io_thread_pool_free_size>
+    <max_format_parsing_thread_pool_free_size>1</max_format_parsing_thread_pool_free_size>
+    <max_prefixes_deserialization_thread_pool_free_size>1</max_prefixes_deserialization_thread_pool_free_size>
+</clickhouse>

tests/integration/test_thread_pool_free_size_shutdown/test.py

@@ -0,0 +1,63 @@
+import pytest
+
+from helpers.cluster import ClickHouseCluster
+from helpers.config_cluster import minio_secret_key
+
+cluster = ClickHouseCluster(__file__)
+node = cluster.add_instance(
+    "node",
+    main_configs=["configs/thread_pools.xml"],
+    with_minio=True,
+    stay_alive=True,
+)
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+        yield cluster
+    finally:
+        cluster.shutdown()
+
+
+# With max_*_thread_pool_free_size > 0 static thread pools keep threads.
+# Meaning, we should always destroy static thread pools before calling GlobalThreadPool::shutdown
+# in binaries that allow to configure max_*_thread_pool_free_size (i.e. server and local).
+# These tests are verifying exactly that.
+
+
+def test_server_graceful_shutdown_with_idle_pool_threads(started_cluster):
+    def backup_destination(name):
+        return (
+            f"S3('http://{cluster.minio_host}:{cluster.minio_port}/{cluster.minio_bucket}/{name}',"
+            f" 'minio', '{minio_secret_key}')"
+        )
+
+    # Force the usage of a static thread pool.
+    node.query("CREATE TABLE t (a UInt64) ENGINE = MergeTree ORDER BY tuple()")
+    node.query("INSERT INTO t VALUES (1)")
+    node.query(f"BACKUP TABLE t TO {backup_destination('backup_server')}")
+
+    node.stop_clickhouse()
+    assert node.contains_in_log("Background threads finished")
+    node.start_clickhouse()
+
+
+def test_local_graceful_shutdown_with_idle_pool_threads(started_cluster):
+    # Force the usage of a static thread pool in local.
+    node.exec_in_container(
+        [
+            "clickhouse",
+            "local",
+            "--config-file",
+            "/etc/clickhouse-server/config.d/thread_pools.xml",
+            "--input_format_parallel_parsing=1",
+            "--max_parsing_threads=4",
+            "--query",
+            "SELECT number FROM numbers(200000) INTO OUTFILE '/tmp/t.csv' FORMAT CSV;"
+            " CREATE TABLE t (a UInt64) ENGINE = MergeTree ORDER BY tuple();"
+            " INSERT INTO t FROM INFILE '/tmp/t.csv' FORMAT CSV",
+        ],
+        timeout=10,
+    )