Implemented runtime thread count resizing for execution pool

CMakeLists.txt

@@ -38,13 +38,17 @@ include_directories(execq PRIVATE
 
 
 ### execq unit-tests ###
-
 if (EXECQ_TESTING_ENABLE)
+    if (MSVC)
+        set(gtest_force_shared_crt ON CACHE BOOL "" FORCE)
+    endif()
+
     set(TEST_SOURCES
         tests/ExecqTestUtil.h
         tests/CancelTokenProviderTest.cpp
         tests/ExecutionStreamTest.cpp
         tests/ExecutionQueueTest.cpp
+        tests/ExecutionPoolTest.cpp
         tests/TaskExecutionQueueTest.cpp
         tests/TaskProviderListTest.cpp
     )

include/execq/internal/ExecutionPool.h

@@ -42,10 +42,14 @@ namespace execq
 
         virtual bool notifyOneWorker() = 0;
         virtual void notifyAllWorkers() = 0;
+
+        virtual void setThreadCount(uint32_t threadCount) = 0;
     };
 
     namespace impl
     {
+        using ThreadWorkers = std::vector<std::unique_ptr<IThreadWorker>>;
+
         class ExecutionPool: public IExecutionPool
         {
         public:
@@ -56,18 +60,26 @@ namespace execq
 
             virtual bool notifyOneWorker() final;
             virtual void notifyAllWorkers() final;
-
+
+            virtual void setThreadCount(uint32_t threadCount) final;
+
         private:
-            std::atomic_bool m_valid { true };
+            bool shouldWorkerExit();
+            void extractFinishedWorkers(ThreadWorkers& workers);
+
+            std::atomic_uint64_t m_threadCount {0};
+
             TaskProviderList m_providerGroup;
 
-            std::vector<std::unique_ptr<IThreadWorker>> m_workers;
+            mutable std::mutex m_workersMutex;
+            ThreadWorkers m_workers;
+            const IThreadWorkerFactory& m_workerFactory;
         };
 
 
         namespace details
         {
-            bool NotifyWorkers(const std::vector<std::unique_ptr<IThreadWorker>>& workers, const bool single);
+            bool NotifyWorkers(const ThreadWorkers& workers, const bool single);
         }
     }
 }

include/execq/internal/ThreadWorker.h

@@ -35,6 +35,7 @@ namespace execq
     namespace impl
     {
         using Task = std::packaged_task<void()>;
+        using ThreadStopCb = std::function<bool()>;
         class ITaskProvider
         {
         public:
@@ -50,6 +51,7 @@ namespace execq
             virtual ~IThreadWorker() = default;
 
             virtual bool notifyWorker() = 0;
+            virtual bool finished() { return false; }
         };
 
 
@@ -60,7 +62,7 @@ namespace execq
 
             virtual ~IThreadWorkerFactory() = default;
 
-            virtual std::unique_ptr<impl::IThreadWorker> createWorker(impl::ITaskProvider& provider) const = 0;
+            virtual std::unique_ptr<impl::IThreadWorker> createWorker(impl::ITaskProvider& provider, ThreadStopCb cb = nullptr) const = 0;
         };
     }
 }

src/ExecutionPool.cpp

@@ -24,11 +24,34 @@
 
 #include "ExecutionPool.h"
 
+namespace
+{
+static constexpr uint64_t packThreadsCount(uint32_t target, uint32_t current)
+{
+    return (uint64_t(target) << 32) | uint64_t(current);
+}
+static constexpr uint32_t unpackTarget(uint64_t threadCount)
+{
+    return uint32_t(threadCount >> 32);
+}
+static constexpr uint32_t unpackCurrent(uint64_t threadCount)
+{
+    return uint32_t(threadCount & 0xFFFFFFFFu);
+}
+}
+
 execq::impl::ExecutionPool::ExecutionPool(const uint32_t threadCount, const IThreadWorkerFactory& workerFactory)
+    : m_workerFactory { workerFactory }
 {
+    m_threadCount.store(packThreadsCount(threadCount, threadCount));
+
+    std::lock_guard<std::mutex> lock(m_workersMutex);
+    m_workers.reserve(threadCount);
+
     for (uint32_t i = 0; i < threadCount; i++)
     {
-        m_workers.emplace_back(workerFactory.createWorker(m_providerGroup));
+        ThreadStopCb cb = [this]() { return this->shouldWorkerExit(); };
+        m_workers.emplace_back(workerFactory.createWorker(m_providerGroup, cb));
     }
 }
 
@@ -44,17 +67,106 @@ void execq::impl::ExecutionPool::removeProvider(ITaskProvider& provider)
 
 bool execq::impl::ExecutionPool::notifyOneWorker()
 {
+    std::lock_guard<std::mutex> lock(m_workersMutex);
     return details::NotifyWorkers(m_workers, true);
 }
 
 void execq::impl::ExecutionPool::notifyAllWorkers()
 {
+    std::lock_guard<std::mutex> lock(m_workersMutex);
     details::NotifyWorkers(m_workers, false);
 }
+void execq::impl::ExecutionPool::setThreadCount(uint32_t threadCount)
+{
+    if (threadCount < 2)
+        return;
+
+    uint32_t toAdd = 0;
+    bool shrinking = false;
+
+    for (;;) {
+        uint64_t tc = m_threadCount.load();
+        auto cur = unpackCurrent(tc);
+
+        if (threadCount > cur) {
+            uint32_t localToAdd = threadCount - cur;
+            auto desired = packThreadsCount(threadCount, threadCount);
+
+            if (m_threadCount.compare_exchange_weak(tc, desired)) {
+                toAdd = localToAdd;
+                shrinking = false;
+                break;
+            }
+        } else {
+            auto desired = packThreadsCount(threadCount, cur);
+
+            if (m_threadCount.compare_exchange_weak(tc, desired)) {
+                toAdd = 0;
+                shrinking = (threadCount < cur);
+                break;
+            }
+        }
+    }
+
+    ThreadWorkers toDestroy;
+
+    {
+        std::lock_guard<std::mutex> lock(m_workersMutex);
+
+        extractFinishedWorkers(toDestroy);
+
+        if (toAdd > 0) {
+            m_workers.reserve(m_workers.size() + toAdd);
+
+            for (uint32_t i = 0; i < toAdd; ++i) {
+                ThreadStopCb cb = [this]() { return this->shouldWorkerExit(); };
+                m_workers.emplace_back(m_workerFactory.createWorker(m_providerGroup, cb));
+            }
+        }
+
+        if (shrinking || toAdd > 0) {
+            details::NotifyWorkers(m_workers, false);
+        }
+    }
+
+    toDestroy.clear();
+}
+
+void execq::impl::ExecutionPool::extractFinishedWorkers(ThreadWorkers& workers)
+{
+    auto it = m_workers.begin();
+    while (it != m_workers.end()) {
+        if ((*it)->finished()) {
+            workers.emplace_back(std::move(*it));
+            it = m_workers.erase(it);
+        } else {
+            ++it;
+        }
+    }
+}
+
+bool execq::impl::ExecutionPool::shouldWorkerExit()
+{
+    for (;;)
+    {
+        uint64_t threadsCount = m_threadCount.load();
+        auto target  = unpackTarget(threadsCount);
+        auto current = unpackCurrent(threadsCount);
+
+        if (current <= target)
+            return false;
+
+        auto desired = packThreadsCount(target, current - 1);
+
+        if (m_threadCount.compare_exchange_weak(threadsCount, desired))
+            return true;
+    }
+}
+
 
 // Details
 
-bool execq::impl::details::NotifyWorkers(const std::vector<std::unique_ptr<IThreadWorker>>& workers, const bool single)
+bool execq::impl::details::NotifyWorkers(const ThreadWorkers& workers, const bool single)
 {
     bool notified = false;
     for (const auto& worker : workers)

src/ThreadWorker.cpp

@@ -31,10 +31,11 @@ namespace execq
         class ThreadWorker: public IThreadWorker
         {
         public:
-            explicit ThreadWorker(ITaskProvider& provider);
+            explicit ThreadWorker(ITaskProvider& provider, ThreadStopCb cb = nullptr);
             virtual ~ThreadWorker();
 
             virtual bool notifyWorker() final;
+            virtual bool finished() final;
 
         private:
             void threadMain();
@@ -43,11 +44,13 @@ namespace execq
         private:
             std::atomic_bool m_shouldQuit { false };
             std::atomic_bool m_checkNextTask { false };
+            std::atomic_bool m_finished { false };
             std::condition_variable m_condition;
             std::mutex m_mutex;
             std::unique_ptr<std::thread> m_thread;
 
             ITaskProvider& m_provider;
+            ThreadStopCb m_stopCb = nullptr;
         };
     }
 }
@@ -57,18 +60,19 @@ std::shared_ptr<const execq::impl::IThreadWorkerFactory> execq::impl::IThreadWor
     class ThreadWorkerFactory: public IThreadWorkerFactory
     {
     public:
-        virtual std::unique_ptr<IThreadWorker> createWorker(ITaskProvider& provider) const final
+        virtual std::unique_ptr<IThreadWorker> createWorker(ITaskProvider& provider, ThreadStopCb cb = nullptr) const final
         {
-            return std::unique_ptr<IThreadWorker>(new ThreadWorker(provider));
+            return std::unique_ptr<IThreadWorker>(new ThreadWorker(provider, cb));
         }
     };
 
     static std::shared_ptr<IThreadWorkerFactory> s_factory = std::make_shared<ThreadWorkerFactory>();
     return s_factory;
 }
 
-execq::impl::ThreadWorker::ThreadWorker(ITaskProvider& provider)
-: m_provider(provider)
+execq::impl::ThreadWorker::ThreadWorker(ITaskProvider& provider, ThreadStopCb cb)
+    : m_provider(provider)
+    , m_stopCb(std::move(cb))
 {}
 
 execq::impl::ThreadWorker::~ThreadWorker()
@@ -83,7 +87,7 @@ execq::impl::ThreadWorker::~ThreadWorker()
 bool execq::impl::ThreadWorker::notifyWorker()
 {
     std::lock_guard<std::mutex> lock(m_mutex);
-    if (m_checkNextTask)
+    if (m_finished || m_checkNextTask)
     {
         return false;
     }
@@ -99,6 +103,11 @@ bool execq::impl::ThreadWorker::notifyWorker()
     return true;
 }
 
+bool execq::impl::ThreadWorker::finished()
+{
+    return m_finished.load();
+}
+
 void execq::impl::ThreadWorker::shutdown()
 {
     std::lock_guard<std::mutex> lock(m_mutex);
@@ -114,6 +123,9 @@ void execq::impl::ThreadWorker::threadMain()
         {
             break;
         }
+
+        if (m_stopCb && m_stopCb())
+            break;
 
         m_checkNextTask = false;
         Task task = m_provider.nextTask();
@@ -136,4 +148,6 @@ void execq::impl::ThreadWorker::threadMain()
 
         m_condition.wait(lock);
     }
+
+    m_finished.store(true);
 }

tests/ExecqTestUtil.h

@@ -40,12 +40,14 @@ namespace execq
 
             MOCK_METHOD0(notifyOneWorker, bool());
             MOCK_METHOD0(notifyAllWorkers, void());
+
+            MOCK_METHOD1(setThreadCount, void(uint32_t threadCount));
         };
 
         class MockThreadWorkerFactory: public execq::impl::IThreadWorkerFactory
         {
         public:
-            MOCK_CONST_METHOD1(createWorker, std::unique_ptr<execq::impl::IThreadWorker>(execq::impl::ITaskProvider& provider));
+            MOCK_CONST_METHOD2(createWorker, std::unique_ptr<execq::impl::IThreadWorker>(execq::impl::ITaskProvider& provider, execq::impl::ThreadStopCb cb));
         };
 
         class MockThreadWorker: public execq::impl::IThreadWorker