Attempt to implemnt global thread pool #4018

dbms/programs/benchmark/Benchmark.cpp

@@ -11,7 +11,7 @@
 #include <Poco/File.h>
 #include <Poco/Util/Application.h>
 #include <Common/Stopwatch.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <AggregateFunctions/ReservoirSampler.h>
 #include <AggregateFunctions/registerAggregateFunctions.h>
 #include <boost/program_options.hpp>

dbms/programs/copier/ClusterCopier.cpp

@@ -18,7 +18,7 @@
 #include <pcg_random.hpp>
 
 #include <common/logger_useful.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <daemon/OwnPatternFormatter.h>
 
 #include <Common/Exception.h>

dbms/programs/performance-test/PerformanceTest.cpp

@@ -23,7 +23,7 @@
 #include <IO/ConnectionTimeouts.h>
 #include <IO/UseSSL.h>
 #include <Interpreters/Settings.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <common/getMemoryAmount.h>
 #include <Poco/AutoPtr.h>
 #include <Poco/Exception.h>

dbms/src/Common/tests/CMakeLists.txt

@@ -53,6 +53,12 @@ target_link_libraries (thread_creation_latency PRIVATE clickhouse_common_io)
 add_executable (thread_pool thread_pool.cpp)
 target_link_libraries (thread_pool PRIVATE clickhouse_common_io)
 
+add_executable (thread_pool_2 thread_pool_2.cpp)
+target_link_libraries (thread_pool_2 PRIVATE clickhouse_common_io)
+
+add_executable (multi_version multi_version.cpp)
+target_link_libraries (multi_version PRIVATE clickhouse_common_io)
+
 add_executable (array_cache array_cache.cpp)
 target_link_libraries (array_cache PRIVATE clickhouse_common_io)
 

dbms/src/Common/tests/gtest_rw_lock.cpp

@@ -8,7 +8,7 @@
 #include <Common/RWLock.h>
 #include <Common/Stopwatch.h>
 #include <common/Types.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <random>
 #include <pcg_random.hpp>
 #include <thread>

dbms/src/Common/tests/parallel_aggregation.cpp

@@ -16,7 +16,7 @@
 #include <Compression/CompressedReadBuffer.h>
 
 #include <Common/Stopwatch.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 
 
 using Key = UInt64;

dbms/src/Common/tests/parallel_aggregation2.cpp

@@ -16,7 +16,7 @@
 #include <Compression/CompressedReadBuffer.h>
 
 #include <Common/Stopwatch.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 
 
 using Key = UInt64;

dbms/src/Common/tests/thread_creation_latency.cpp

@@ -5,7 +5,7 @@
 
 #include <Common/Stopwatch.h>
 #include <Common/Exception.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 
 
 int x = 0;

dbms/src/Common/tests/thread_pool.cpp

@@ -1,4 +1,4 @@
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 
 /** Reproduces bug in ThreadPool.
   * It get stuck if we call 'wait' many times from many other threads simultaneously.

dbms/src/DataStreams/AsynchronousBlockInputStream.h

@@ -5,7 +5,7 @@
 #include <DataStreams/IProfilingBlockInputStream.h>
 #include <Common/setThreadName.h>
 #include <Common/CurrentMetrics.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <Common/MemoryTracker.h>
 #include <Poco/Ext/ThreadNumber.h>
 

dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.h

@@ -4,7 +4,7 @@
 #include <DataStreams/IProfilingBlockInputStream.h>
 #include <Common/ConcurrentBoundedQueue.h>
 #include <Common/CurrentThread.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <condition_variable>
 
 

dbms/src/DataStreams/PushingToViewsBlockOutputStream.cpp

@@ -5,7 +5,7 @@
 #include <Common/CurrentThread.h>
 #include <Common/setThreadName.h>
 #include <Common/getNumberOfPhysicalCPUCores.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.h>
 
 namespace DB

dbms/src/Databases/DatabaseFactory.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <Databases/IDatabase.h>
 
 

dbms/src/Databases/DatabaseOrdinary.cpp

@@ -10,7 +10,7 @@
 #include <Common/escapeForFileName.h>
 #include <Common/StringUtils/StringUtils.h>
 #include <Common/Stopwatch.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <Parsers/ASTCreateQuery.h>
 #include <Parsers/parseQuery.h>
 #include <Parsers/ParserCreateQuery.h>

dbms/src/Databases/IDatabase.h

@@ -8,7 +8,7 @@
 #include <functional>
 #include <Poco/File.h>
 #include <Common/escapeForFileName.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <Interpreters/Context.h>
 
 

dbms/src/IO/AsynchronousWriteBuffer.h

@@ -4,7 +4,7 @@
 
 #include <vector>
 
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <IO/WriteBuffer.h>
 
 

dbms/src/Interpreters/Aggregator.h

@@ -12,7 +12,7 @@
 #include <Common/Arena.h>
 #include <Common/HashTable/HashMap.h>
 #include <Common/HashTable/TwoLevelHashMap.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <Common/UInt128.h>
 #include <Common/LRUCache.h>
 

dbms/src/Interpreters/Compiler.h

@@ -15,7 +15,7 @@
 #include <Common/Exception.h>
 #include <Common/UInt128.h>
 #include <Common/SharedLibrary.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 
 namespace DB
 {

dbms/src/Interpreters/Context.h

@@ -10,7 +10,7 @@
 #include <optional>
 
 #include <Common/config.h>
-#include <common/MultiVersion.h>
+#include <Common/MultiVersion.h>
 #include <Common/LRUCache.h>
 #include <Core/Types.h>
 #include <Core/NamesAndTypes.h>

dbms/src/Interpreters/EmbeddedDictionaries.h

@@ -2,7 +2,7 @@
 
 #include <thread>
 #include <functional>
-#include <common/MultiVersion.h>
+#include <Common/MultiVersion.h>
 #include <Poco/Event.h>
 
 

dbms/src/Interpreters/InterpreterCreateQuery.h

@@ -2,7 +2,7 @@
 
 #include <Interpreters/IInterpreter.h>
 #include <Storages/ColumnsDescription.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 
 
 namespace DB

dbms/src/Interpreters/InterpreterSystemQuery.cpp

@@ -4,7 +4,7 @@
 #include <Common/config.h>
 #include <Common/typeid_cast.h>
 #include <Common/getNumberOfPhysicalCPUCores.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <Interpreters/Context.h>
 #include <Interpreters/ExternalDictionaries.h>
 #include <Interpreters/EmbeddedDictionaries.h>

dbms/src/Interpreters/loadMetadata.cpp

@@ -2,7 +2,7 @@
 #include <thread>
 #include <future>
 
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 
 #include <Poco/DirectoryIterator.h>
 #include <Poco/FileStream.h>

dbms/src/Interpreters/tests/internal_iotop.cpp

@@ -1,4 +1,4 @@
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <IO/ReadBufferFromFile.h>
 #include <IO/WriteBufferFromFile.h>
 #include <IO/copyData.h>

dbms/src/Storages/Distributed/DistributedBlockOutputStream.h

@@ -4,7 +4,7 @@
 #include <DataStreams/IBlockOutputStream.h>
 #include <Core/Block.h>
 #include <Common/Throttler.h>
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 #include <atomic>
 #include <memory>
 #include <chrono>

dbms/src/Storages/StorageDictionary.h

@@ -2,7 +2,7 @@
 
 #include <Storages/IStorage.h>
 #include <Core/Defines.h>
-#include <common/MultiVersion.h>
+#include <Common/MultiVersion.h>
 #include <ext/shared_ptr_helper.h>
 #include <IO/WriteBufferFromString.h>
 #include <IO/Operators.h>

dbms/src/Storages/StorageReplicatedMergeTree.cpp

@@ -44,7 +44,7 @@
 
 #include <Poco/DirectoryIterator.h>
 
-#include <common/ThreadPool.h>
+#include <Common/ThreadPool.h>
 
 #include <ext/range.h>
 #include <ext/scope_guard.h>

libs/libcommon/CMakeLists.txt

@@ -18,7 +18,6 @@ add_library (common ${LINK_MODE}
     src/mremap.cpp
     src/JSON.cpp
     src/getMemoryAmount.cpp
-    src/ThreadPool.cpp
     src/demangle.cpp
     src/SetTerminalEcho.cpp
 
@@ -34,11 +33,9 @@ add_library (common ${LINK_MODE}
     include/common/mremap.h
     include/common/likely.h
     include/common/logger_useful.h
-    include/common/MultiVersion.h
     include/common/strong_typedef.h
     include/common/JSON.h
     include/common/getMemoryAmount.h
-    include/common/ThreadPool.h
     include/common/demangle.h
     include/common/SetTerminalEcho.h
     include/common/find_symbols.h

libs/libcommon/include/common/MultiVersion.h

@@ -1,56 +0,0 @@
-#pragma once
-
-#include <mutex>
-#include <memory>
-
-
-/** Allow to store and read-only usage of an object in several threads,
-  *  and to atomically replace an object in another thread.
-  * The replacement is atomic and reading threads can work with different versions of an object.
-  *
-  * Usage:
-  *  MultiVersion<T> x;
-  * - on data update:
-  *  x.set(new value);
-  * - on read-only usage:
-  * {
-  *     MultiVersion<T>::Version current_version = x.get();
-  *     // use *current_version
-  * }   // now we finish own current version; if the version is outdated and no one else is using it - it will be destroyed.
-  *
-  * All methods are thread-safe.
-  */
-template <typename T>
-class MultiVersion
-{
-public:
-    /// Version of object for usage. shared_ptr manage lifetime of version.
-    using Version = std::shared_ptr<const T>;
-
-    /// Default initialization - by nullptr.
-    MultiVersion() = default;
-
-    MultiVersion(std::unique_ptr<const T> && value)
-    {
-        set(std::move(value));
-    }
-
-    /// Obtain current version for read-only usage. Returns shared_ptr, that manages lifetime of version.
-    Version get() const
-    {
-        /// NOTE: is it possible to lock-free replace of shared_ptr?
-        std::lock_guard lock(mutex);
-        return current_version;
-    }
-
-    /// Update an object with new version.
-    void set(std::unique_ptr<const T> && value)
-    {
-        std::lock_guard lock(mutex);
-        current_version = std::move(value);
-    }
-
-private:
-    Version current_version;
-    mutable std::mutex mutex;
-};

libs/libcommon/include/common/ThreadPool.h

@@ -1,151 +0,0 @@
-#pragma once
-
-#include <cstdint>
-#include <thread>
-#include <mutex>
-#include <condition_variable>
-#include <functional>
-#include <queue>
-#include <vector>
-#include <ext/singleton.h>
-
-
-/** Very simple thread pool similar to boost::threadpool.
-  * Advantages:
-  * - catches exceptions and rethrows on wait.
-  */
-
-template <typename Thread>
-class ThreadPoolImpl
-{
-public:
-    using Job = std::function<void()>;
-
-    /// Size is constant. Up to num_threads are created on demand and then run until shutdown.
-    explicit ThreadPoolImpl(size_t num_threads);
-
-    /// queue_size - maximum number of running plus scheduled jobs. It can be greater than num_threads. Zero means unlimited.
-    ThreadPoolImpl(size_t num_threads, size_t queue_size);
-
-    /// Add new job. Locks until number of active jobs is less than maximum or exception in one of threads was thrown.
-    /// If an exception in some thread was thrown, method silently returns, and exception will be rethrown only on call to 'wait' function.
-    /// Priority: greater is higher.
-    void schedule(Job job, int priority = 0);
-
-    /// Wait for all currently active jobs to be done.
-    /// You may call schedule and wait many times in arbitary order.
-    /// If any thread was throw an exception, first exception will be rethrown from this method,
-    ///  and exception will be cleared.
-    void wait();
-
-    /// Waits for all threads. Doesn't rethrow exceptions (use 'wait' method to rethrow exceptions).
-    /// You should not destroy object while calling schedule or wait methods from another threads.
-    ~ThreadPoolImpl();
-
-    size_t size() const { return num_threads; }
-
-    /// Returns number of running and scheduled jobs.
-    size_t active() const;
-
-private:
-    mutable std::mutex mutex;
-    std::condition_variable job_finished;
-    std::condition_variable new_job_or_shutdown;
-
-    const size_t num_threads;
-    const size_t queue_size;
-
-    size_t active_jobs = 0;
-    bool shutdown = false;
-
-    struct JobWithPriority
-    {
-        Job job;
-        int priority;
-
-        JobWithPriority(Job job, int priority)
-            : job(job), priority(priority) {}
-
-        bool operator< (const JobWithPriority & rhs) const
-        {
-            return priority < rhs.priority;
-        }
-    };
-
-    std::priority_queue<JobWithPriority> jobs;
-    std::vector<Thread> threads;
-    std::exception_ptr first_exception;
-
-
-    void worker();
-
-    void finalize();
-};
-
-
-using FreeThreadPool = ThreadPoolImpl<std::thread>;
-
-class GlobalThreadPool : public FreeThreadPool, public ext::singleton<GlobalThreadPool>
-{
-public:
-    GlobalThreadPool() : FreeThreadPool(10000) {}   /// TODO: global blocking limit may lead to deadlocks.
-};
-
-class ThreadFromGlobalPool
-{
-public:
-    ThreadFromGlobalPool() {}
-
-    ThreadFromGlobalPool(std::function<void()> func)
-    {
-        mutex = std::make_unique<std::mutex>();
-        /// The function object must be copyable, so we wrap lock_guard in shared_ptr.
-        GlobalThreadPool::instance().schedule([lock = std::make_shared<std::lock_guard<std::mutex>>(*mutex), func = std::move(func)] { func(); });
-    }
-
-    ThreadFromGlobalPool(ThreadFromGlobalPool && rhs)
-    {
-        *this = std::move(rhs);
-    }
-
-    ThreadFromGlobalPool & operator=(ThreadFromGlobalPool && rhs)
-    {
-        if (mutex)
-            std::terminate();
-        mutex = std::move(rhs.mutex);
-        return *this;
-    }
-
-    ~ThreadFromGlobalPool()
-    {
-        if (mutex)
-            std::terminate();
-    }
-
-    void join()
-    {
-        {
-            std::lock_guard lock(*mutex);
-        }
-        mutex.reset();
-    }
-private:
-    std::unique_ptr<std::mutex> mutex;  /// Object must be moveable.
-};
-
-using ThreadPool = ThreadPoolImpl<ThreadFromGlobalPool>;
-
-
-/// Allows to save first catched exception in jobs and postpone its rethrow.
-class ExceptionHandler
-{
-public:
-    void setException(std::exception_ptr && exception);
-    void throwIfException();
-
-private:
-    std::exception_ptr first_exception;
-    std::mutex mutex;
-};
-
-ThreadPool::Job createExceptionHandledJob(ThreadPool::Job job, ExceptionHandler & handler);

libs/libcommon/src/ThreadPool.cpp

@@ -1,169 +0,0 @@
-#include <common/ThreadPool.h>
-#include <iostream>
-
-
-template <typename Thread>
-ThreadPoolImpl<Thread>::ThreadPoolImpl(size_t num_threads)
-    : ThreadPoolImpl(num_threads, num_threads)
-{
-}
-
-template <typename Thread>
-ThreadPoolImpl<Thread>::ThreadPoolImpl(size_t num_threads, size_t queue_size)
-    : num_threads(num_threads), queue_size(queue_size)
-{
-    threads.reserve(num_threads);
-}
-
-template <typename Thread>
-void ThreadPoolImpl<Thread>::schedule(Job job, int priority)
-{
-    {
-        std::unique_lock<std::mutex> lock(mutex);
-        job_finished.wait(lock, [this] { return !queue_size || active_jobs < queue_size || shutdown; });
-        if (shutdown)
-            return;
-
-        jobs.emplace(std::move(job), priority);
-        ++active_jobs;
-
-        if (threads.size() < std::min(num_threads, active_jobs))
-            threads.emplace_back([this] { worker(); });
-    }
-    new_job_or_shutdown.notify_one();
-}
-
-template <typename Thread>
-void ThreadPoolImpl<Thread>::wait()
-{
-    {
-        std::unique_lock<std::mutex> lock(mutex);
-        job_finished.wait(lock, [this] { return active_jobs == 0; });
-
-        if (first_exception)
-        {
-            std::exception_ptr exception;
-            std::swap(exception, first_exception);
-            std::rethrow_exception(exception);
-        }
-    }
-}
-
-template <typename Thread>
-ThreadPoolImpl<Thread>::~ThreadPoolImpl()
-{
-    finalize();
-}
-
-template <typename Thread>
-void ThreadPoolImpl<Thread>::finalize()
-{
-    {
-        std::unique_lock<std::mutex> lock(mutex);
-        shutdown = true;
-    }
-
-    new_job_or_shutdown.notify_all();
-
-    for (auto & thread : threads)
-        thread.join();
-
-    threads.clear();
-}
-
-template <typename Thread>
-size_t ThreadPoolImpl<Thread>::active() const
-{
-    std::unique_lock<std::mutex> lock(mutex);
-    return active_jobs;
-}
-
-template <typename Thread>
-void ThreadPoolImpl<Thread>::worker()
-{
-    while (true)
-    {
-        Job job;
-        bool need_shutdown = false;
-
-        {
-            std::unique_lock<std::mutex> lock(mutex);
-            new_job_or_shutdown.wait(lock, [this] { return shutdown || !jobs.empty(); });
-            need_shutdown = shutdown;
-
-            if (!jobs.empty())
-            {
-                job = jobs.top().job;
-                jobs.pop();
-            }
-            else
-            {
-                return;
-            }
-        }
-
-        if (!need_shutdown)
-        {
-            try
-            {
-                job();
-            }
-            catch (...)
-            {
-                {
-                    std::unique_lock<std::mutex> lock(mutex);
-                    if (!first_exception)
-                        first_exception = std::current_exception();
-                    shutdown = true;
-                    --active_jobs;
-                }
-                job_finished.notify_all();
-                new_job_or_shutdown.notify_all();
-                return;
-            }
-        }
-
-        {
-            std::unique_lock<std::mutex> lock(mutex);
-            --active_jobs;
-        }
-
-        job_finished.notify_all();
-    }
-}
-
-
-template class ThreadPoolImpl<std::thread>;
-template class ThreadPoolImpl<ThreadFromGlobalPool>;
-
-
-void ExceptionHandler::setException(std::exception_ptr && exception)
-{
-    std::unique_lock<std::mutex> lock(mutex);
-    if (!first_exception)
-        first_exception = std::move(exception);
-}
-
-void ExceptionHandler::throwIfException()
-{
-    std::unique_lock<std::mutex> lock(mutex);
-    if (first_exception)
-        std::rethrow_exception(first_exception);
-}
-
-
-ThreadPool::Job createExceptionHandledJob(ThreadPool::Job job, ExceptionHandler & handler)
-{
-    return [job{std::move(job)}, &handler] ()
-    {
-        try
-        {
-            job();
-        }
-        catch (...)
-        {
-            handler.setException(std::current_exception());
-        }
-    };
-}
-

libs/libcommon/src/tests/CMakeLists.txt

@@ -5,7 +5,6 @@ add_executable (date_lut2 date_lut2.cpp)
 add_executable (date_lut3 date_lut3.cpp)
 add_executable (date_lut4 date_lut4.cpp)
 add_executable (date_lut_default_timezone date_lut_default_timezone.cpp)
-add_executable (multi_version multi_version.cpp)
 add_executable (local_date_time_comparison local_date_time_comparison.cpp)
 add_executable (realloc-perf allocator.cpp)
 
@@ -16,7 +15,6 @@ target_link_libraries (date_lut2 common ${PLATFORM_LIBS})
 target_link_libraries (date_lut3 common ${PLATFORM_LIBS})
 target_link_libraries (date_lut4 common ${PLATFORM_LIBS})
 target_link_libraries (date_lut_default_timezone common ${PLATFORM_LIBS})
-target_link_libraries (multi_version common)
 target_link_libraries (local_date_time_comparison common)
 target_link_libraries (realloc-perf common)
 add_check(multi_version)

libs/libcommon/src/tests/multi_version.cpp

@@ -1,56 +0,0 @@
-#include <string.h>
-#include <iostream>
-#include <common/ThreadPool.h>
-#include <functional>
-#include <common/MultiVersion.h>
-#include <Poco/Exception.h>
-
-
-using T = std::string;
-using MV = MultiVersion<T>;
-using Results = std::vector<T>;
-
-
-void thread1(MV & x, T & result)
-{
-    MV::Version v = x.get();
-    result = *v;
-}
-
-void thread2(MV & x, const char * result)
-{
-    x.set(std::make_unique<T>(result));
-}
-
-
-int main(int argc, char ** argv)
-{
-    try
-    {
-        const char * s1 = "Hello!";
-        const char * s2 = "Goodbye!";
-
-        size_t n = 1000;
-        MV x(std::make_unique<T>(s1));
-        Results results(n);
-
-        ThreadPool tp(8);
-        for (size_t i = 0; i < n; ++i)
-        {
-            tp.schedule(std::bind(thread1, std::ref(x), std::ref(results[i])));
-            tp.schedule(std::bind(thread2, std::ref(x), (rand() % 2) ? s1 : s2));
-        }
-        tp.wait();
-
-        for (size_t i = 0; i < n; ++i)
-            std::cerr << results[i] << " ";
-        std::cerr << std::endl;
-    }
-    catch (const Poco::Exception & e)
-    {
-        std::cerr << e.message() << std::endl;
-        throw;
-    }
-
-    return 0;
-}