From 42e9d0f5205453f68a87a7efb4ed0a64d4cf2903 Mon Sep 17 00:00:00 2001
From: Paul T <developer.paul.123@gmail.com>
Date: Thu, 2 Dec 2021 10:24:18 -0500
Subject: [PATCH 1/5] Improvements and Simplifications to ThreadPool

Removed type definition of thread pool for thread_pool_impl. Simplified class by removing queue template parameter. Switched to using std::deque for task_queue objects so that we don't have to use unique_ptrs for those objects. Made pool non-copyable. Enqueue parameters by value for version that returns a future. Removed other unused type definitions.
---
 include/thread_pool/thread_pool.h | 68 +++++++++++++++----------------
 1 file changed, 33 insertions(+), 35 deletions(-)

diff --git a/include/thread_pool/thread_pool.h b/include/thread_pool/thread_pool.h
index 16bc49a..317de8e 100644
--- a/include/thread_pool/thread_pool.h
+++ b/include/thread_pool/thread_pool.h
@@ -19,7 +19,7 @@ namespace dp {
             return std::forward<T>(v);
         }
 
-        // Bind F and args... into a nullary one-shot lambda. Lambda captures by value.
+        // bind F and parameter pack into a nullary one shot. Lambda captures by value.
         template <typename... Args, typename F>
         auto bind(F &&f, Args &&...args) {
             return [f = decay_copy(std::forward<F>(f)),
@@ -40,51 +40,49 @@ namespace dp {
         static_assert(detail::is_valid_queue<dp::thread_safe_queue<int>>);
     }  // namespace detail
 
-    template <template <class T> class Queue, typename FunctionType = std::function<void()>>
+    template <typename FunctionType = std::function<void()>>
     requires std::invocable<FunctionType> &&
-        std::is_same_v<void, std::invoke_result_t<FunctionType>> &&
-        detail::is_valid_queue<Queue<FunctionType>>
-    class thread_pool_impl {
+        std::is_same_v<void, std::invoke_result_t<FunctionType>>
+    class thread_pool {
       public:
-        thread_pool_impl(
-            const unsigned int &number_of_threads = std::thread::hardware_concurrency()) {
+        thread_pool(const unsigned int &number_of_threads = std::thread::hardware_concurrency())
+            : queues_(number_of_threads) {
             for (std::size_t i = 0; i < number_of_threads; ++i) {
-                queues_.push_back(std::make_unique<task_pair>());
                 threads_.emplace_back([&, id = i](std::stop_token stop_tok) {
                     do {
                         // check if we have task
-                        if (queues_[id]->tasks.empty()) {
+                        if (queues_[id].tasks.empty()) {
                             // no tasks, so we wait instead of spinning
-                            queues_[id]->semaphore.acquire();
+                            queues_[id].semaphore.acquire();
                         }
 
                         // ensure we have a task before getting task
                         // since the dtor releases the semaphore as well
-                        if (!queues_[id]->tasks.empty()) {
+                        if (!queues_[id].tasks.empty()) {
                             // get the task
-                            auto &task = queues_[id]->tasks.front();
+                            auto &task = queues_[id].tasks.front();
                             // invoke the task
                             std::invoke(std::move(task));
                             // remove task from the queue
-                            queues_[id]->tasks.pop();
+                            queues_[id].tasks.pop();
                         }
                     } while (!stop_tok.stop_requested());
                 });
             }
         }
 
-        ~thread_pool_impl() {
+        ~thread_pool() {
             // stop all threads
             for (std::size_t i = 0; i < threads_.size(); ++i) {
                 threads_[i].request_stop();
-                queues_[i]->semaphore.release();
+                queues_[i].semaphore.release();
                 threads_[i].join();
             }
         }
 
         /// thread pool is non-copyable
-        thread_pool_impl(const thread_pool_impl &) = delete;
-        thread_pool_impl &operator=(const thread_pool_impl &) = delete;
+        thread_pool(const thread_pool &) = delete;
+        thread_pool &operator=(const thread_pool &) = delete;
 
         /**
          * @brief Enqueue a task into the thread pool that returns a result.
@@ -98,11 +96,21 @@ namespace dp {
         template <typename Function, typename... Args,
                   typename ReturnType = std::invoke_result_t<Function &&, Args &&...>>
         requires std::invocable<Function, Args...>
-        [[nodiscard]] std::future<ReturnType> enqueue(Function &&f, Args &&...args) {
-            // use shared promise here so that we don't break the promise later
+        [[nodiscard]] std::future<ReturnType> enqueue(Function f, Args ...args) {
+            /*
+             * use shared promise here so that we don't break the promise later (until C++23)
+             *
+             * with C++23 we can do the following:
+             *
+             * std::promise<ReturnType> promise;
+             * auto future = promise.get_future();
+             * auto task = [func = std::move(f), ... largs = std::move(args),
+                              promise = std::move(promise)]() mutable {...};
+             */
             auto shared_promise = std::make_shared<std::promise<ReturnType>>();
             auto task = [func = std::move(f), ... largs = std::move(args),
                          promise = shared_promise]() { promise->set_value(func(largs...)); };
+
             // get the future before enqueuing the task
             auto future = shared_promise->get_future();
             // enqueue the task
@@ -125,33 +133,23 @@ namespace dp {
         }
 
       private:
-        using semaphore_type = std::binary_semaphore;
-        using task_type = FunctionType;
-        struct task_pair {
-            semaphore_type semaphore{0};
-            Queue<task_type> tasks{};
+        struct task_queue {
+            std::binary_semaphore semaphore{0};
+            dp::thread_safe_queue<FunctionType> tasks{};
         };
 
         template <typename Function>
         void enqueue_task(Function &&f) {
             const std::size_t i = count_++ % queues_.size();
-            queues_[i]->tasks.push(std::forward<Function>(f));
-            queues_[i]->semaphore.release();
+            queues_[i].tasks.push(std::forward<Function>(f));
+            queues_[i].semaphore.release();
         }
 
         std::vector<std::jthread> threads_;
-        // have to use unique_ptr here because std::binary_semaphore is not move/copy
-        // assignable/constructible
-        std::vector<std::unique_ptr<task_pair>> queues_;
+        std::deque<task_queue> queues_;
         std::size_t count_ = 0;
     };
 
-    /**
-     * @brief Thread pool class capable of queuing detached tasks and value returning tasks.
-     * @details This is a default alias for the dp::thread_pool_impl
-     */
-    using thread_pool = thread_pool_impl<dp::thread_safe_queue>;
-
     /**
      * @example mandelbrot/source/main.cpp
      * Example showing how to use thread pool with tasks that return a value. Outputs a PPM image of

From ffb5ed7a5f60fffa10eb8161dd2a085d5e649f9d Mon Sep 17 00:00:00 2001
From: Paul T <developer.paul.123@gmail.com>
Date: Thu, 2 Dec 2021 10:25:20 -0500
Subject: [PATCH 2/5] Improvements to thread_safe_queue

Fix to push function to properly use the lock guard before notifying with the condition variable. Switched to using std::deque directly instead of std::queue for the underlying data.
---
 include/thread_pool/thread_safe_queue.h | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/include/thread_pool/thread_safe_queue.h b/include/thread_pool/thread_safe_queue.h
index cb02ed4..4fefcbb 100644
--- a/include/thread_pool/thread_safe_queue.h
+++ b/include/thread_pool/thread_safe_queue.h
@@ -1,22 +1,26 @@
 #pragma once
 
 #include <condition_variable>
+#include <deque>
 #include <mutex>
-#include <queue>
 
 namespace dp {
     template <typename T>
     class thread_safe_queue {
       public:
         using value_type = T;
-        using size_type = typename std::queue<T>::size_type;
+        using size_type = typename std::deque<T>::size_type;
 
         thread_safe_queue() = default;
+
         void push(T&& value) {
-            std::lock_guard lock(mutex_);
-            data_.push(std::forward<T>(value));
+            {
+                std::lock_guard lock(mutex_);
+                data_.push_back(std::forward<T>(value));
+            }
             condition_variable_.notify_all();
         }
+
         bool empty() {
             std::lock_guard lock(mutex_);
             return data_.empty();
@@ -42,12 +46,12 @@ namespace dp {
         void pop() {
             std::unique_lock lock(mutex_);
             condition_variable_.wait(lock, [this] { return !data_.empty(); });
-            data_.pop();
+            data_.pop_front();
         }
 
       private:
         using mutex_type = std::mutex;
-        std::queue<T> data_;
+        std::deque<T> data_;
         mutable mutex_type mutex_{};
         std::condition_variable condition_variable_{};
     };

From 5491b564aecb32b6f3fa865be8817ebe9c2710bd Mon Sep 17 00:00:00 2001
From: Paul T <developer.paul.123@gmail.com>
Date: Thu, 2 Dec 2021 10:26:04 -0500
Subject: [PATCH 3/5] Add New Test

Renamed old test and added new test to ensure that work completes before the thread pool gets destroyed.
---
 test/source/thread_pool.cpp | 21 +++++++++++++++++++--
 1 file changed, 19 insertions(+), 2 deletions(-)

diff --git a/test/source/thread_pool.cpp b/test/source/thread_pool.cpp
index 02275d3..3a975c7 100644
--- a/test/source/thread_pool.cpp
+++ b/test/source/thread_pool.cpp
@@ -3,9 +3,8 @@
 
 #include <string>
 
-TEST_CASE("Basic Return Types") {
+TEST_CASE("Basic task return types") {
     dp::thread_pool pool(2);
-    // TODO
     auto future_value = pool.enqueue([](const int& value) { return value; }, 30);
     auto future_negative = pool.enqueue([](int x) -> int { return x - 20; }, 3);
 
@@ -30,3 +29,21 @@ TEST_CASE("Ensure input params are properly passed") {
         CHECK(j == futures[j].get());
     }
 }
+
+TEST_CASE("Ensure work completes upon destruction") {
+    std::atomic<int> counter;
+    std::vector<std::future<int>> futures;
+    const auto total_tasks = 20;
+    {
+        dp::thread_pool pool(4);
+        for (auto i = 0; i < total_tasks; i++) {
+            auto task = [index = i, &counter]() {
+                counter++;
+                return index;
+            };
+            futures.push_back(pool.enqueue(task));
+        }
+    }
+
+    CHECK_EQ(counter.load(), total_tasks);
+}

From 087127339bd5d7f76967e4d7cd6dcf498e657f84 Mon Sep 17 00:00:00 2001
From: Paul T <developer.paul.123@gmail.com>
Date: Thu, 2 Dec 2021 10:37:28 -0500
Subject: [PATCH 4/5] Remove Unused Concept and Format Fixes

---
 include/thread_pool/thread_pool.h | 12 +-----------
 1 file changed, 1 insertion(+), 11 deletions(-)

diff --git a/include/thread_pool/thread_pool.h b/include/thread_pool/thread_pool.h
index 317de8e..e524165 100644
--- a/include/thread_pool/thread_pool.h
+++ b/include/thread_pool/thread_pool.h
@@ -28,16 +28,6 @@ namespace dp {
             };
         }
 
-        template <class Queue, class U = typename Queue::value_type>
-        concept is_valid_queue = requires(Queue q) {
-            { q.empty() } -> std::convertible_to<bool>;
-            { q.front() } -> std::convertible_to<U &>;
-            { q.back() } -> std::convertible_to<U &>;
-            q.pop();
-        };
-
-        static_assert(detail::is_valid_queue<std::queue<int>>);
-        static_assert(detail::is_valid_queue<dp::thread_safe_queue<int>>);
     }  // namespace detail
 
     template <typename FunctionType = std::function<void()>>
@@ -96,7 +86,7 @@ namespace dp {
         template <typename Function, typename... Args,
                   typename ReturnType = std::invoke_result_t<Function &&, Args &&...>>
         requires std::invocable<Function, Args...>
-        [[nodiscard]] std::future<ReturnType> enqueue(Function f, Args ...args) {
+        [[nodiscard]] std::future<ReturnType> enqueue(Function f, Args... args) {
             /*
              * use shared promise here so that we don't break the promise later (until C++23)
              *

From b7490c37f9a3a4dd545a41b24c613f78baeb9c10 Mon Sep 17 00:00:00 2001
From: Paul T <developer.paul.123@gmail.com>
Date: Tue, 14 Dec 2021 13:20:56 -0500
Subject: [PATCH 5/5] Ensure Tasks Complete

Added way to ensure that tasks complete before destroying the thread pool.
---
 include/thread_pool/thread_pool.h | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/include/thread_pool/thread_pool.h b/include/thread_pool/thread_pool.h
index e524165..0edb41f 100644
--- a/include/thread_pool/thread_pool.h
+++ b/include/thread_pool/thread_pool.h
@@ -53,6 +53,8 @@ namespace dp {
                             auto &task = queues_[id].tasks.front();
                             // invoke the task
                             std::invoke(std::move(task));
+                            // decrement in-flight counter
+                            --in_flight_;
                             // remove task from the queue
                             queues_[id].tasks.pop();
                         }
@@ -62,6 +64,11 @@ namespace dp {
         }
 
         ~thread_pool() {
+            // wait for tasks to complete first
+            do {
+                std::this_thread::yield();
+            } while (in_flight_ > 0);
+
             // stop all threads
             for (std::size_t i = 0; i < threads_.size(); ++i) {
                 threads_[i].request_stop();
@@ -131,6 +138,7 @@ namespace dp {
         template <typename Function>
         void enqueue_task(Function &&f) {
             const std::size_t i = count_++ % queues_.size();
+            ++in_flight_;
             queues_[i].tasks.push(std::forward<Function>(f));
             queues_[i].semaphore.release();
         }
@@ -138,6 +146,7 @@ namespace dp {
         std::vector<std::jthread> threads_;
         std::deque<task_queue> queues_;
         std::size_t count_ = 0;
+        std::atomic<int64_t> in_flight_{0};
     };
 
     /**