Improve work_queue thread pool: correctness fixes, performance, and tests

src/YgorThreadPool.h

@@ -1,13 +1,13 @@
-//Thread_Pool.h.
+//YgorThreadPool.h.
 
 #pragma once
 
 #include <mutex>
 #include <condition_variable>
 #include <thread>
 #include <atomic>
-#include <algorithm>
 #include <list>
+#include <type_traits>
 
 
 // Multi-threaded work queue for offloading processing tasks.
@@ -33,7 +33,7 @@ class work_queue {
         std::unique_lock<std::mutex> lock(this->queue_mutex);
 
         // Exercise the condition variables and mutexes, ensuring they are initialized by the implementation.
-        // This should efectively evaluate to a no-op, but also helps suppress false-positive warning messages in
+        // This should effectively evaluate to a no-op, but also helps suppress false-positive warning messages in
         // Valgrind's DRD tool, i.e., 'not a condition variable', and other tools.
         this->new_task_notifier.notify_all(); // No threads waiting, so nothing to notify.
         this->end_task_notifier.notify_all(); // No threads waiting, so nothing to notify.
@@ -49,58 +49,54 @@ class work_queue {
             this->worker_threads.emplace_back(
                 [this](){
                     // Continually check the queue and wait on the condition variable.
-                    bool l_should_quit = false;
-                    while(!l_should_quit){
+                    while(true){
 
                         std::unique_lock<std::mutex> lock(this->queue_mutex);
-                        while( !(l_should_quit = this->should_quit.load())
+                        while( !this->should_quit.load()
                                && this->queue.empty() ){
 
                             // Waiting releases the lock, which allows for work to be submitted.
                             //
                             // Note: spurious notifications are OK, since the queue will be empty and the worker will return to
                             // waiting on the condition variable.
-//                            this->new_task_notifier.wait(lock);
-                            this->new_task_notifier.wait_for(lock, std::chrono::seconds(2) ); // No notifiers, so no signal to receive.
+                            this->new_task_notifier.wait_for(lock, std::chrono::seconds(2) );
+                        }
+
+                        if(this->queue.empty()){
+                            return;
                         }
 
                         // Assume ownership of only the first item in the queue (FIFO).
-                        std::list<T> l_queue;
-                        if(!this->queue.empty()) l_queue.splice( std::end(l_queue), this->queue, std::begin(this->queue) );
-
-                        //// Assume ownership of all available items in the queue.
-                        //std::list<T> l_queue;
-                        //l_queue.swap( this->queue );
-
-                        // Perform the work in FIFO order.
+                        auto task = std::move(this->queue.front());
+                        this->queue.pop_front();
+
                         lock.unlock();
-                        for(const auto &user_f : l_queue){
-                            try{
-                                if(user_f){
-                                    user_f();
-                                }
-                            }catch(const std::exception &){};
-
-                            lock.lock();
-                            this->end_task_notifier.notify_all();
-                            lock.unlock();
-                        }
+
+                        // Perform the work.
+                        try{
+                            if constexpr(std::is_constructible_v<bool, const T &>){
+                                if(task) task();
+                            }else{
+                                task();
+                            }
+                        }catch(const std::exception &){}
+                         catch(...){};
+
+                        this->end_task_notifier.notify_one();
                     }
                 }
             );
         }
     }
 
     void submit_task(T f){
-        std::lock_guard<std::mutex> lock(this->queue_mutex);
-        this->queue.push_back(std::move(f));
-
-        // Note: it's not strictly necessary to lock the mutex before notifying, but it's possible it could lead to a data
-        // race or use-after-free. If nothing else, locking suppresses warnings of a 'possible' data race in thread sanitizers.
-        // See discussion and some links at
-        // https://stackoverflow.com/questions/17101922/do-i-have-to-acquire-lock-before-calling-condition-variable-notify-one
-        // Also note that this can potentially lead to a performance downgrade; in practice, most pthread
-        // implementations will detect and mitigate the issue.
+        {
+            std::lock_guard<std::mutex> lock(this->queue_mutex);
+            this->queue.push_back(std::move(f));
+        }
+        // Note: notifying without the mutex held avoids the situation where a notified thread wakes and immediately
+        // blocks waiting for the mutex to be released. This is safe; condition_variable::notify_one does not require
+        // the associated mutex to be held.
         this->new_task_notifier.notify_one();
         return;
     }
@@ -119,12 +115,9 @@ class work_queue {
         //
         // We rely on a condition variable to signal when tasks are completed, but fallback on occasional polling in
         // case there are any races to avoid waiting forever.
-        bool l_should_quit = false;
-        while(!l_should_quit){
-
+        {
             std::unique_lock<std::mutex> lock(this->queue_mutex);
-            while( !(l_should_quit = this->should_quit.load())
-                   && !this->queue.empty() ){
+            while( !this->queue.empty() ){
 
                 // Waiting releases the lock while waiting, which still allows for outstanding work to be completed.
                 //
@@ -133,15 +126,10 @@ class work_queue {
                 this->end_task_notifier.wait_for(lock, std::chrono::milliseconds(2000) );
             }
 
-            if(!l_should_quit){
-                this->should_quit.store(true);
-                this->new_task_notifier.notify_all(); // notify threads to wake up and 'notice' they need to terminate.
-                lock.unlock();
-                for(auto &wt : this->worker_threads) wt.join();
-            }
-            break;
+            this->should_quit.store(true);
         }
+        this->new_task_notifier.notify_all(); // notify threads to wake up and 'notice' they need to terminate.
+        for(auto &wt : this->worker_threads) wt.join();
     }
 };
 
-

tests2/YgorThreadPool.cc

@@ -0,0 +1,143 @@
+#include <atomic>
+#include <chrono>
+#include <functional>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+#include <mutex>
+
+#include <YgorThreadPool.h>
+
+#include "doctest/doctest.h"
+
+
+TEST_CASE( "work_queue" ){
+
+    SUBCASE("basic task execution"){
+        std::atomic<int> counter{0};
+        {
+            work_queue<std::function<void()>> pool(2);
+            pool.submit_task([&counter](){ ++counter; });
+            pool.submit_task([&counter](){ ++counter; });
+            pool.submit_task([&counter](){ ++counter; });
+        }
+        REQUIRE(counter.load() == 3);
+    }
+
+    SUBCASE("single worker sequential FIFO order"){
+        std::vector<int> results;
+        std::mutex results_mutex;
+        {
+            work_queue<std::function<void()>> pool(1);
+            for(int i = 0; i < 10; ++i){
+                pool.submit_task([i, &results, &results_mutex](){
+                    std::lock_guard<std::mutex> lock(results_mutex);
+                    results.push_back(i);
+                });
+            }
+        }
+        REQUIRE(results.size() == 10);
+        for(int i = 0; i < 10; ++i){
+            REQUIRE(results[static_cast<size_t>(i)] == i);
+        }
+    }
+
+    SUBCASE("handles std::exception without crashing"){
+        std::atomic<int> counter{0};
+        {
+            work_queue<std::function<void()>> pool(2);
+            pool.submit_task([](){ throw std::runtime_error("test"); });
+            pool.submit_task([&counter](){ ++counter; });
+        }
+        REQUIRE(counter.load() == 1);
+    }
+
+    SUBCASE("handles non-std exceptions without crashing"){
+        std::atomic<int> counter{0};
+        {
+            work_queue<std::function<void()>> pool(2);
+            pool.submit_task([](){ throw 42; });
+            pool.submit_task([&counter](){ ++counter; });
+        }
+        REQUIRE(counter.load() == 1);
+    }
+
+    SUBCASE("clear_tasks removes pending tasks"){
+        std::atomic<int> counter{0};
+        work_queue<std::function<void()>> pool(1);
+
+        // Submit a blocking task to hold the single worker.
+        std::atomic<bool> blocker{true};
+        std::atomic<bool> started{false};
+        pool.submit_task([&blocker, &started](){
+            started.store(true);
+            while(blocker.load()) std::this_thread::sleep_for(std::chrono::milliseconds(10));
+        });
+
+        // Wait for the blocking task to start.
+        while(!started.load()) std::this_thread::sleep_for(std::chrono::milliseconds(1));
+
+        // Submit tasks that should be clearable (worker is busy).
+        pool.submit_task([&counter](){ ++counter; });
+        pool.submit_task([&counter](){ ++counter; });
+        pool.submit_task([&counter](){ ++counter; });
+
+        auto cleared = pool.clear_tasks();
+        REQUIRE(cleared.size() == 3);
+
+        // Release the blocker so the destructor can join.
+        blocker.store(false);
+    }
+
+    SUBCASE("many tasks with many workers"){
+        const int N = 1000;
+        std::atomic<int> counter{0};
+        {
+            work_queue<std::function<void()>> pool(8);
+            for(int i = 0; i < N; ++i){
+                pool.submit_task([&counter](){ ++counter; });
+            }
+        }
+        REQUIRE(counter.load() == N);
+    }
+
+    SUBCASE("handles empty function gracefully"){
+        {
+            work_queue<std::function<void()>> pool(2);
+            std::function<void()> empty_fn;
+            pool.submit_task(empty_fn);
+        }
+        // Should not crash.
+        REQUIRE(true);
+    }
+
+    SUBCASE("default constructor uses hardware concurrency"){
+        std::atomic<int> counter{0};
+        {
+            work_queue<std::function<void()>> pool;
+            pool.submit_task([&counter](){ ++counter; });
+        }
+        REQUIRE(counter.load() == 1);
+    }
+
+    SUBCASE("zero workers defaults to hardware concurrency"){
+        std::atomic<int> counter{0};
+        {
+            work_queue<std::function<void()>> pool(0);
+            pool.submit_task([&counter](){ ++counter; });
+        }
+        REQUIRE(counter.load() == 1);
+    }
+
+    SUBCASE("tasks submitted from within tasks"){
+        std::atomic<int> counter{0};
+        {
+            work_queue<std::function<void()>> pool(2);
+            pool.submit_task([&pool, &counter](){
+                ++counter;
+                pool.submit_task([&counter](){ ++counter; });
+            });
+        }
+        REQUIRE(counter.load() == 2);
+    }
+}

tests2/compile_and_run.sh

@@ -60,6 +60,7 @@ g++ \
   YgorStatsConditionalForests.cc \
   YgorStatsStochasticForests.cc \
   YgorString.cc \
+  YgorThreadPool.cc \
   YgorTime/*.cc \
   \
   -o run_tests \