Updated to v3.4.0

BS_thread_pool.hpp

@@ -3,39 +3,44 @@
 /**
  * @file BS_thread_pool.hpp
  * @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
- * @version 3.1.0
- * @date 2022-07-13
- * @copyright Copyright (c) 2022 Barak Shoshany. Licensed under the MIT license. If you found this project useful, please consider starring it on GitHub! If you use this library in software of any kind, please provide a link to the GitHub repository https://github.com/bshoshany/thread-pool in the source code and documentation. If you use this library in published research, please cite it as follows: Barak Shoshany, "A C++17 Thread Pool for High-Performance Scientific Computing", doi:10.5281/zenodo.4742687, arXiv:2105.00613 (May 2021)
+ * @version 3.4.0
+ * @date 2023-05-12
+ * @copyright Copyright (c) 2023 Barak Shoshany. Licensed under the MIT license. If you found this project useful, please consider starring it on GitHub! If you use this library in software of any kind, please provide a link to the GitHub repository https://github.com/bshoshany/thread-pool in the source code and documentation. If you use this library in published research, please cite it as follows: Barak Shoshany, "A C++17 Thread Pool for High-Performance Scientific Computing", doi:10.5281/zenodo.4742687, arXiv:2105.00613 (May 2021)
  *
- * @brief BS::thread_pool: a fast, lightweight, and easy-to-use C++17 thread pool library. This header file contains the entire library, including the main BS::thread_pool class and the helper classes BS::multi_future, BS:synced_stream, and BS::timer.
+ * @brief BS::thread_pool: a fast, lightweight, and easy-to-use C++17 thread pool library. This header file contains the entire library, including the main BS::thread_pool class and the helper classes BS::multi_future, BS::blocks, BS:synced_stream, and BS::timer.
  */
 
-#define BS_THREAD_POOL_VERSION "v3.1.0 (2022-07-13)"
+#define BS_THREAD_POOL_VERSION "v3.4.0 (2023-05-12)"
 
 #include <atomic>             // std::atomic
 #include <chrono>             // std::chrono
 #include <condition_variable> // std::condition_variable
 #include <exception>          // std::current_exception
-#include <functional>         // std::function
+#include <functional>         // std::bind, std::function, std::invoke
 #include <future>             // std::future, std::promise
-#include <iostream>           // std::cout, std::ostream
+#include <iostream>           // std::cout, std::endl, std::flush, std::ostream
 #include <memory>             // std::make_shared, std::make_unique, std::shared_ptr, std::unique_ptr
 #include <mutex>              // std::mutex, std::scoped_lock, std::unique_lock
 #include <queue>              // std::queue
 #include <thread>             // std::thread
-#include <type_traits>        // std::common_type_t, std::decay_t, std::is_void_v, std::invoke_result_t
-#include <utility>            // std::move, std::swap
+#include <type_traits>        // std::common_type_t, std::conditional_t, std::decay_t, std::invoke_result_t, std::is_void_v
+#include <utility>            // std::forward, std::move, std::swap
 #include <vector>             // std::vector
 
 namespace BS
 {
+/**
+ * @brief A convenient shorthand for the type of std::thread::hardware_concurrency(). Should evaluate to unsigned int.
+ */
 using concurrency_t = std::invoke_result_t<decltype(std::thread::hardware_concurrency)>;
 
 // ============================================================================================= //
-//                                    Begin class multi_future                                    //
+//                                    Begin class multi_future                                   //
 
 /**
  * @brief A helper class to facilitate waiting for and/or getting the results of multiple futures at once.
+ *
+ * @tparam T The return type of the futures.
  */
 template <typename T>
 class [[nodiscard]] multi_future
@@ -46,19 +51,59 @@ public:
      *
      * @param num_futures_ The desired number of futures to store.
      */
-    multi_future(const size_t num_futures_ = 0) : f(num_futures_) {}
+    multi_future(const size_t num_futures_ = 0) : futures(num_futures_) {}
 
     /**
-     * @brief Get the results from all the futures stored in this multi_future object.
+     * @brief Get the results from all the futures stored in this multi_future object, rethrowing any stored exceptions.
      *
-     * @return A vector containing the results.
+     * @return If the futures return void, this function returns void as well. Otherwise, it returns a vector containing the results.
      */
-    [[nodiscard]] std::vector<T> get()
+    [[nodiscard]] std::conditional_t<std::is_void_v<T>, void, std::vector<T>> get()
     {
-        std::vector<T> results(f.size());
-        for (size_t i = 0; i < f.size(); ++i)
-            results[i] = f[i].get();
-        return results;
+        if constexpr (std::is_void_v<T>)
+        {
+            for (size_t i = 0; i < futures.size(); ++i)
+                futures[i].get();
+            return;
+        }
+        else
+        {
+            std::vector<T> results(futures.size());
+            for (size_t i = 0; i < futures.size(); ++i)
+                results[i] = futures[i].get();
+            return results;
+        }
+    }
+
+    /**
+     * @brief Get a reference to one of the futures stored in this multi_future object.
+     *
+     * @param i The index of the desired future.
+     * @return The future.
+     */
+    [[nodiscard]] std::future<T>& operator[](const size_t i)
+    {
+        return futures[i];
+    }
+
+    /**
+     * @brief Append a future to this multi_future object.
+     *
+     * @param future The future to append.
+     */
+    void push_back(std::future<T> future)
+    {
+        futures.push_back(std::move(future));
+    }
+
+    /**
+     * @brief Get the number of futures stored in this multi_future object.
+     *
+     * @return The number of futures.
+     */
+    [[nodiscard]] size_t size() const
+    {
+        return futures.size();
     }
 
     /**
@@ -66,17 +111,124 @@ public:
      */
     void wait() const
     {
-        for (size_t i = 0; i < f.size(); ++i)
-            f[i].wait();
+        for (size_t i = 0; i < futures.size(); ++i)
+            futures[i].wait();
     }
 
+private:
     /**
      * @brief A vector to store the futures.
      */
-    std::vector<std::future<T>> f;
+    std::vector<std::future<T>> futures;
 };
 
-//                                     End class multi_future                                     //
+//                                     End class multi_future                                    //
+// ============================================================================================= //
+
+// ============================================================================================= //
+//                                       Begin class blocks                                      //
+
+/**
+ * @brief A helper class to divide a range into blocks. Used by parallelize_loop() and push_loop().
+ *
+ * @tparam T1 The type of the first index in the range. Should be a signed or unsigned integer.
+ * @tparam T2 The type of the index after the last index in the range. Should be a signed or unsigned integer. If T1 is not the same as T2, a common type will be automatically inferred.
+ * @tparam T The common type of T1 and T2.
+ */
+template <typename T1, typename T2, typename T = std::common_type_t<T1, T2>>
+class [[nodiscard]] blocks
+{
+public:
+    /**
+     * @brief Construct a blocks object with the given specifications.
+     *
+     * @param first_index_ The first index in the range.
+     * @param index_after_last_ The index after the last index in the range.
+     * @param num_blocks_ The desired number of blocks to divide the range into.
+     */
+    blocks(const T1 first_index_, const T2 index_after_last_, const size_t num_blocks_) : first_index(static_cast<T>(first_index_)), index_after_last(static_cast<T>(index_after_last_)), num_blocks(num_blocks_)
+    {
+        if (index_after_last < first_index)
+            std::swap(index_after_last, first_index);
+        total_size = static_cast<size_t>(index_after_last - first_index);
+        block_size = static_cast<size_t>(total_size / num_blocks);
+        if (block_size == 0)
+        {
+            block_size = 1;
+            num_blocks = (total_size > 1) ? total_size : 1;
+        }
+    }
+
+    /**
+     * @brief Get the first index of a block.
+     *
+     * @param i The block number.
+     * @return The first index.
+     */
+    [[nodiscard]] T start(const size_t i) const
+    {
+        return static_cast<T>(i * block_size) + first_index;
+    }
+
+    /**
+     * @brief Get the index after the last index of a block.
+     *
+     * @param i The block number.
+     * @return The index after the last index.
+     */
+    [[nodiscard]] T end(const size_t i) const
+    {
+        return (i == num_blocks - 1) ? index_after_last : (static_cast<T>((i + 1) * block_size) + first_index);
+    }
+
+    /**
+     * @brief Get the number of blocks. Note that this may be different than the desired number of blocks that was passed to the constructor.
+     *
+     * @return The number of blocks.
+     */
+    [[nodiscard]] size_t get_num_blocks() const
+    {
+        return num_blocks;
+    }
+
+    /**
+     * @brief Get the total number of indices in the range.
+     *
+     * @return The total number of indices.
+     */
+    [[nodiscard]] size_t get_total_size() const
+    {
+        return total_size;
+    }
+
+private:
+    /**
+     * @brief The size of each block (except possibly the last block).
+     */
+    size_t block_size = 0;
+
+    /**
+     * @brief The first index in the range.
+     */
+    T first_index = 0;
+
+    /**
+     * @brief The index after the last index in the range.
+     */
+    T index_after_last = 0;
+
+    /**
+     * @brief The number of blocks.
+     */
+    size_t num_blocks = 0;
+
+    /**
+     * @brief The total number of indices in the range.
+     */
+    size_t total_size = 0;
+};
+
+//                                        End class blocks                                       //
 // ============================================================================================= //
 
 // ============================================================================================= //
@@ -103,7 +255,7 @@ public:
     }
 
     /**
-     * @brief Destruct the thread pool. Waits for all tasks to complete, then destroys all threads. Note that if the variable paused is set to true, then any tasks still in the queue will never be executed.
+     * @brief Destruct the thread pool. Waits for all tasks to complete, then destroys all threads. Note that if the pool is paused, then any tasks still in the queue will never be executed.
      */
     ~thread_pool()
     {
@@ -158,7 +310,17 @@ public:
     }
 
     /**
-     * @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue.
+     * @brief Check whether the pool is currently paused.
+     *
+     * @return true if the pool is paused, false if it is not paused.
+     */
+    [[nodiscard]] bool is_paused() const
+    {
+        return paused;
+    }
+
+    /**
+     * @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. Returns a multi_future object that contains the futures for all of the blocks.
      *
      * @tparam F The type of the function to loop through.
      * @tparam T1 The type of the first index in the loop. Should be a signed or unsigned integer.
@@ -166,58 +328,108 @@ public:
      * @tparam T The common type of T1 and T2.
      * @tparam R The return value of the loop function F (can be void).
      * @param first_index The first index in the loop.
-     * @param index_after_last The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = first_index; i < index_after_last; ++i)". Note that if first_index == index_after_last, no blocks will be submitted.
+     * @param index_after_last The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = first_index; i < index_after_last; ++i)". Note that if index_after_last == first_index, no blocks will be submitted.
      * @param loop The function to loop through. Will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. loop(start, end) should typically involve a loop of the form "for (T i = start; i < end; ++i)".
      * @param num_blocks The maximum number of blocks to split the loop into. The default is to use the number of threads in the pool.
-     * @return A multi_future object that can be used to wait for all the blocks to finish. If the loop function returns a value, the multi_future object can be used to obtain the values returned by each block.
+     * @return A multi_future object that can be used to wait for all the blocks to finish. If the loop function returns a value, the multi_future object can also be used to obtain the values returned by each block.
      */
     template <typename F, typename T1, typename T2, typename T = std::common_type_t<T1, T2>, typename R = std::invoke_result_t<std::decay_t<F>, T, T>>
-    [[nodiscard]] multi_future<R> parallelize_loop(const T1& first_index, const T2& index_after_last, const F& loop, size_t num_blocks = 0)
+    [[nodiscard]] multi_future<R> parallelize_loop(const T1 first_index, const T2 index_after_last, F&& loop, const size_t num_blocks = 0)
     {
-        T first_index_T = static_cast<T>(first_index);
-        T index_after_last_T = static_cast<T>(index_after_last);
-        if (first_index_T == index_after_last_T)
+        blocks blks(first_index, index_after_last, num_blocks ? num_blocks : thread_count);
+        if (blks.get_total_size() > 0)
+        {
+            multi_future<R> mf(blks.get_num_blocks());
+            for (size_t i = 0; i < blks.get_num_blocks(); ++i)
+                mf[i] = submit(std::forward<F>(loop), blks.start(i), blks.end(i));
+            return mf;
+        }
+        else
+        {
             return multi_future<R>();
-        if (index_after_last_T < first_index_T)
-            std::swap(index_after_last_T, first_index_T);
-        if (num_blocks == 0)
-            num_blocks = thread_count;
-        const size_t total_size = static_cast<size_t>(index_after_last_T - first_index_T);
-        size_t block_size = static_cast<size_t>(total_size / num_blocks);
-        if (block_size == 0)
-        {
-            block_size = 1;
-            num_blocks = total_size > 1 ? total_size : 1;
         }
-        multi_future<R> mf(num_blocks);
-        for (size_t i = 0; i < num_blocks; ++i)
-        {
-            const T start = (static_cast<T>(i * block_size) + first_index_T);
-            const T end = (i == num_blocks - 1) ? index_after_last_T : (static_cast<T>((i + 1) * block_size) + first_index_T);
-            mf.f[i] = submit(loop, start, end);
-        }
-        return mf;
     }
 
     /**
-     * @brief Push a function with zero or more arguments, but no return value, into the task queue.
+     * @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. Returns a multi_future object that contains the futures for all of the blocks. This overload is used for the special case where the first index is 0.
+     *
+     * @tparam F The type of the function to loop through.
+     * @tparam T The type of the loop indices. Should be a signed or unsigned integer.
+     * @tparam R The return value of the loop function F (can be void).
+     * @param index_after_last The index after the last index in the loop. The loop will iterate from 0 to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = 0; i < index_after_last; ++i)". Note that if index_after_last == 0, no blocks will be submitted.
+     * @param loop The function to loop through. Will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. loop(start, end) should typically involve a loop of the form "for (T i = start; i < end; ++i)".
+     * @param num_blocks The maximum number of blocks to split the loop into. The default is to use the number of threads in the pool.
+     * @return A multi_future object that can be used to wait for all the blocks to finish. If the loop function returns a value, the multi_future object can also be used to obtain the values returned by each block.
+     */
+    template <typename F, typename T, typename R = std::invoke_result_t<std::decay_t<F>, T, T>>
+    [[nodiscard]] multi_future<R> parallelize_loop(const T index_after_last, F&& loop, const size_t num_blocks = 0)
+    {
+        return parallelize_loop(0, index_after_last, std::forward<F>(loop), num_blocks);
+    }
+
+    /**
+     * @brief Pause the pool. The workers will temporarily stop retrieving new tasks out of the queue, although any tasks already executed will keep running until they are finished.
+     */
+    void pause()
+    {
+        paused = true;
+    }
+
+    /**
+     * @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. Does not return a multi_future, so the user must use wait_for_tasks() or some other method to ensure that the loop finishes executing, otherwise bad things will happen.
+     *
+     * @tparam F The type of the function to loop through.
+     * @tparam T1 The type of the first index in the loop. Should be a signed or unsigned integer.
+     * @tparam T2 The type of the index after the last index in the loop. Should be a signed or unsigned integer. If T1 is not the same as T2, a common type will be automatically inferred.
+     * @tparam T The common type of T1 and T2.
+     * @param first_index The first index in the loop.
+     * @param index_after_last The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = first_index; i < index_after_last; ++i)". Note that if index_after_last == first_index, no blocks will be submitted.
+     * @param loop The function to loop through. Will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. loop(start, end) should typically involve a loop of the form "for (T i = start; i < end; ++i)".
+     * @param num_blocks The maximum number of blocks to split the loop into. The default is to use the number of threads in the pool.
+     */
+    template <typename F, typename T1, typename T2, typename T = std::common_type_t<T1, T2>>
+    void push_loop(const T1 first_index, const T2 index_after_last, F&& loop, const size_t num_blocks = 0)
+    {
+        blocks blks(first_index, index_after_last, num_blocks ? num_blocks : thread_count);
+        if (blks.get_total_size() > 0)
+        {
+            for (size_t i = 0; i < blks.get_num_blocks(); ++i)
+                push_task(std::forward<F>(loop), blks.start(i), blks.end(i));
+        }
+    }
+
+    /**
+     * @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. Does not return a multi_future, so the user must use wait_for_tasks() or some other method to ensure that the loop finishes executing, otherwise bad things will happen. This overload is used for the special case where the first index is 0.
+     *
+     * @tparam F The type of the function to loop through.
+     * @tparam T The type of the loop indices. Should be a signed or unsigned integer.
+     * @param index_after_last The index after the last index in the loop. The loop will iterate from 0 to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = 0; i < index_after_last; ++i)". Note that if index_after_last == 0, no blocks will be submitted.
+     * @param loop The function to loop through. Will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. loop(start, end) should typically involve a loop of the form "for (T i = start; i < end; ++i)".
+     * @param num_blocks The maximum number of blocks to split the loop into. The default is to use the number of threads in the pool.
+     */
+    template <typename F, typename T>
+    void push_loop(const T index_after_last, F&& loop, const size_t num_blocks = 0)
+    {
+        push_loop(0, index_after_last, std::forward<F>(loop), num_blocks);
+    }
+
+    /**
+     * @brief Push a function with zero or more arguments, but no return value, into the task queue. Does not return a future, so the user must use wait_for_tasks() or some other method to ensure that the task finishes executing, otherwise bad things will happen.
      *
      * @tparam F The type of the function.
      * @tparam A The types of the arguments.
      * @param task The function to push.
-     * @param args The arguments to pass to the function.
+     * @param args The zero or more arguments to pass to the function. Note that if the task is a class member function, the first argument must be a pointer to the object, i.e. &object (or this), followed by the actual arguments.
      */
     template <typename F, typename... A>
-    void push_task(const F& task, const A&... args)
+    void push_task(F&& task, A&&... args)
     {
+        std::function<void()> task_function = std::bind(std::forward<F>(task), std::forward<A>(args)...);
         {
             const std::scoped_lock tasks_lock(tasks_mutex);
-            if constexpr (sizeof...(args) == 0)
-                tasks.push(std::function<void()>(task));
-            else
-                tasks.push(std::function<void()>([task, args...] { task(args...); }));
+            tasks.push(task_function);
+            ++tasks_total;
         }
-        ++tasks_total;
         task_available_cv.notify_one();
     }
 
@@ -245,26 +457,27 @@ public:
      * @tparam A The types of the zero or more arguments to pass to the function.
      * @tparam R The return type of the function (can be void).
      * @param task The function to submit.
-     * @param args The zero or more arguments to pass to the function.
+     * @param args The zero or more arguments to pass to the function. Note that if the task is a class member function, the first argument must be a pointer to the object, i.e. &object (or this), followed by the actual arguments.
      * @return A future to be used later to wait for the function to finish executing and/or obtain its returned value if it has one.
      */
     template <typename F, typename... A, typename R = std::invoke_result_t<std::decay_t<F>, std::decay_t<A>...>>
-    [[nodiscard]] std::future<R> submit(const F& task, const A&... args)
+    [[nodiscard]] std::future<R> submit(F&& task, A&&... args)
     {
+        std::function<R()> task_function = std::bind(std::forward<F>(task), std::forward<A>(args)...);
         std::shared_ptr<std::promise<R>> task_promise = std::make_shared<std::promise<R>>();
         push_task(
-            [task, args..., task_promise]
+            [task_function, task_promise]
             {
                 try
                 {
                     if constexpr (std::is_void_v<R>)
                     {
-                        task(args...);
+                        std::invoke(task_function);
                         task_promise->set_value();
                     }
                     else
                     {
-                        task_promise->set_value(task(args...));
+                        task_promise->set_value(std::invoke(task_function));
                     }
                 }
                 catch (...)
@@ -281,25 +494,71 @@ public:
         return task_promise->get_future();
     }
 
+    /**
+     * @brief Unpause the pool. The workers will resume retrieving new tasks out of the queue.
+     */
+    void unpause()
+    {
+        paused = false;
+    }
+
     /**
      * @brief Wait for tasks to be completed. Normally, this function waits for all tasks, both those that are currently running in the threads and those that are still waiting in the queue. However, if the pool is paused, this function only waits for the currently running tasks (otherwise it would wait forever). Note: To wait for just one specific task, use submit() instead, and call the wait() member function of the generated future.
      */
     void wait_for_tasks()
     {
-        waiting = true;
-        std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
-        task_done_cv.wait(tasks_lock, [this] { return (tasks_total == (paused ? tasks.size() : 0)); });
-        waiting = false;
+        if (!waiting)
+        {
+            waiting = true;
+            std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
+            task_done_cv.wait(tasks_lock, [this] { return (tasks_total == (paused ? tasks.size() : 0)); });
+            waiting = false;
+        }
     }
 
-    // ===========
-    // Public data
-    // ===========
+    /**
+     * @brief Wait for tasks to be completed, but stop waiting after the specified duration has passed.
+     *
+     * @tparam R An arithmetic type representing the number of ticks to wait.
+     * @tparam P An std::ratio representing the length of each tick in seconds.
+     * @param duration The time duration to wait.
+     * @return true if finished waiting before the duration expired, false if timed out or the pool is already waiting. In other words, returns false if and only if tasks are still running.
+     */
+    template <typename R, typename P>
+    bool wait_for_tasks_duration(const std::chrono::duration<R, P>& duration)
+    {
+        if (!waiting)
+        {
+            waiting = true;
+            std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
+            const bool status = task_done_cv.wait_for(tasks_lock, duration, [this] { return (tasks_total == (paused ? tasks.size() : 0)); });
+            waiting = false;
+            return status;
+        }
+        return false;
+    }
 
     /**
-     * @brief An atomic variable indicating whether the workers should pause. When set to true, the workers temporarily stop retrieving new tasks out of the queue, although any tasks already executed will keep running until they are finished. Set to false again to resume retrieving tasks.
+     * @brief Wait for tasks to be completed, but stop waiting after the specified time point has been reached.
+     *
+     * @tparam C The type of the clock used to measure time.
+     * @tparam D An std::chrono::duration type used to indicate the time point.
+     * @param timeout_time The time point at which to stop waiting.
+     * @return true if finished waiting before the time point was reached, false if timed out or the pool is already waiting. In other words, returns false if and only if tasks are still running.
      */
-    std::atomic<bool> paused = false;
+    template <typename C, typename D>
+    bool wait_for_tasks_until(const std::chrono::time_point<C, D>& timeout_time)
+    {
+        if (!waiting)
+        {
+            waiting = true;
+            std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
+            const bool status = task_done_cv.wait_until(tasks_lock, timeout_time, [this] { return (tasks_total == (paused ? tasks.size() : 0)); });
+            waiting = false;
+            return status;
+        }
+        return false;
+    }
 
 private:
     // ========================
@@ -324,7 +583,10 @@ private:
     void destroy_threads()
     {
         running = false;
-        task_available_cv.notify_all();
+        {
+            const std::scoped_lock tasks_lock(tasks_mutex);
+            task_available_cv.notify_all();
+        }
         for (concurrency_t i = 0; i < thread_count; ++i)
         {
             threads[i].join();
@@ -359,7 +621,7 @@ private:
         {
             std::function<void()> task;
             std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
-            task_available_cv.wait(tasks_lock, [&] { return !tasks.empty() || !running; });
+            task_available_cv.wait(tasks_lock, [this] { return !tasks.empty() || !running; });
             if (running && !paused)
             {
                 task = std::move(tasks.front());
@@ -378,6 +640,11 @@ private:
     // Private data
     // ============
 
+    /**
+     * @brief An atomic variable indicating whether the workers should pause. When set to true, the workers temporarily stop retrieving new tasks out of the queue, although any tasks already executed will keep running until they are finished. When set to false again, the workers resume retrieving tasks.
+     */
+    std::atomic<bool> paused = false;
+
     /**
      * @brief An atomic variable indicating to the workers to keep running. When set to false, the workers permanently stop working.
      */
@@ -450,10 +717,10 @@ public:
      * @param items The items to print.
      */
     template <typename... T>
-    void print(const T&... items)
+    void print(T&&... items)
     {
         const std::scoped_lock lock(stream_mutex);
-        (out_stream << ... << items);
+        (out_stream << ... << std::forward<T>(items));
     }
 
     /**
@@ -463,11 +730,21 @@ public:
      * @param items The items to print.
      */
     template <typename... T>
-    void println(const T&... items)
+    void println(T&&... items)
     {
-        print(items..., '\n');
+        print(std::forward<T>(items)..., '\n');
     }
 
+    /**
+     * @brief A stream manipulator to pass to a synced_stream (an explicit cast of std::endl). Prints a newline character to the stream, and then flushes it. Should only be used if flushing is desired, otherwise '\n' should be used instead.
+     */
+    inline static std::ostream& (&endl)(std::ostream&) = static_cast<std::ostream& (&)(std::ostream&)>(std::endl);
+
+    /**
+     * @brief A stream manipulator to pass to a synced_stream (an explicit cast of std::flush). Used to flush the stream.
+     */
+    inline static std::ostream& (&flush)(std::ostream&) = static_cast<std::ostream& (&)(std::ostream&)>(std::flush);
+
 private:
     /**
      * @brief The output stream to print to.

BS_thread_pool_light.hpp

@@ -0,0 +1,331 @@
+#pragma once
+
+/**
+ * @file BS_thread_pool_light.hpp
+ * @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
+ * @version 3.4.0
+ * @date 2023-05-12
+ * @copyright Copyright (c) 2023 Barak Shoshany. Licensed under the MIT license. If you found this project useful, please consider starring it on GitHub! If you use this library in software of any kind, please provide a link to the GitHub repository https://github.com/bshoshany/thread-pool in the source code and documentation. If you use this library in published research, please cite it as follows: Barak Shoshany, "A C++17 Thread Pool for High-Performance Scientific Computing", doi:10.5281/zenodo.4742687, arXiv:2105.00613 (May 2021)
+ *
+ * @brief BS::thread_pool_light: a fast, lightweight, and easy-to-use C++17 thread pool library. This header file contains a light version of the main library, for use when advanced features are not needed.
+ */
+
+#define BS_THREAD_POOL_LIGHT_VERSION "v3.4.0 (2023-05-12)"
+
+#include <atomic>             // std::atomic
+#include <condition_variable> // std::condition_variable
+#include <exception>          // std::current_exception
+#include <functional>         // std::bind, std::function, std::invoke
+#include <future>             // std::future, std::promise
+#include <memory>             // std::make_shared, std::make_unique, std::shared_ptr, std::unique_ptr
+#include <mutex>              // std::mutex, std::scoped_lock, std::unique_lock
+#include <queue>              // std::queue
+#include <thread>             // std::thread
+#include <type_traits>        // std::common_type_t, std::decay_t, std::invoke_result_t, std::is_void_v
+#include <utility>            // std::forward, std::move, std::swap
+
+namespace BS
+{
+/**
+ * @brief A convenient shorthand for the type of std::thread::hardware_concurrency(). Should evaluate to unsigned int.
+ */
+using concurrency_t = std::invoke_result_t<decltype(std::thread::hardware_concurrency)>;
+
+/**
+ * @brief A fast, lightweight, and easy-to-use C++17 thread pool class. This is a lighter version of the main thread pool class.
+ */
+class [[nodiscard]] thread_pool_light
+{
+public:
+    // ============================
+    // Constructors and destructors
+    // ============================
+
+    /**
+     * @brief Construct a new thread pool.
+     *
+     * @param thread_count_ The number of threads to use. The default value is the total number of hardware threads available, as reported by the implementation. This is usually determined by the number of cores in the CPU. If a core is hyperthreaded, it will count as two threads.
+     */
+    thread_pool_light(const concurrency_t thread_count_ = 0) : thread_count(determine_thread_count(thread_count_)), threads(std::make_unique<std::thread[]>(determine_thread_count(thread_count_)))
+    {
+        create_threads();
+    }
+
+    /**
+     * @brief Destruct the thread pool. Waits for all tasks to complete, then destroys all threads.
+     */
+    ~thread_pool_light()
+    {
+        wait_for_tasks();
+        destroy_threads();
+    }
+
+    // =======================
+    // Public member functions
+    // =======================
+
+    /**
+     * @brief Get the number of threads in the pool.
+     *
+     * @return The number of threads.
+     */
+    [[nodiscard]] concurrency_t get_thread_count() const
+    {
+        return thread_count;
+    }
+
+    /**
+     * @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. The user must use wait_for_tasks() or some other method to ensure that the loop finishes executing, otherwise bad things will happen.
+     *
+     * @tparam F The type of the function to loop through.
+     * @tparam T1 The type of the first index in the loop. Should be a signed or unsigned integer.
+     * @tparam T2 The type of the index after the last index in the loop. Should be a signed or unsigned integer. If T1 is not the same as T2, a common type will be automatically inferred.
+     * @tparam T The common type of T1 and T2.
+     * @param first_index The first index in the loop.
+     * @param index_after_last The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = first_index; i < index_after_last; ++i)". Note that if index_after_last == first_index, no blocks will be submitted.
+     * @param loop The function to loop through. Will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. loop(start, end) should typically involve a loop of the form "for (T i = start; i < end; ++i)".
+     * @param num_blocks The maximum number of blocks to split the loop into. The default is to use the number of threads in the pool.
+     */
+    template <typename F, typename T1, typename T2, typename T = std::common_type_t<T1, T2>>
+    void push_loop(T1 first_index_, T2 index_after_last_, F&& loop, size_t num_blocks = 0)
+    {
+        T first_index = static_cast<T>(first_index_);
+        T index_after_last = static_cast<T>(index_after_last_);
+        if (num_blocks == 0)
+            num_blocks = thread_count;
+        if (index_after_last < first_index)
+            std::swap(index_after_last, first_index);
+        size_t total_size = static_cast<size_t>(index_after_last - first_index);
+        size_t block_size = static_cast<size_t>(total_size / num_blocks);
+        if (block_size == 0)
+        {
+            block_size = 1;
+            num_blocks = (total_size > 1) ? total_size : 1;
+        }
+        if (total_size > 0)
+        {
+            for (size_t i = 0; i < num_blocks; ++i)
+                push_task(std::forward<F>(loop), static_cast<T>(i * block_size) + first_index, (i == num_blocks - 1) ? index_after_last : (static_cast<T>((i + 1) * block_size) + first_index));
+        }
+    }
+
+    /**
+     * @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. The user must use wait_for_tasks() or some other method to ensure that the loop finishes executing, otherwise bad things will happen. This overload is used for the special case where the first index is 0.
+     *
+     * @tparam F The type of the function to loop through.
+     * @tparam T The type of the loop indices. Should be a signed or unsigned integer.
+     * @param index_after_last The index after the last index in the loop. The loop will iterate from 0 to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = 0; i < index_after_last; ++i)". Note that if index_after_last == 0, no blocks will be submitted.
+     * @param loop The function to loop through. Will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. loop(start, end) should typically involve a loop of the form "for (T i = start; i < end; ++i)".
+     * @param num_blocks The maximum number of blocks to split the loop into. The default is to use the number of threads in the pool.
+     */
+    template <typename F, typename T>
+    void push_loop(const T index_after_last, F&& loop, const size_t num_blocks = 0)
+    {
+        push_loop(0, index_after_last, std::forward<F>(loop), num_blocks);
+    }
+
+    /**
+     * @brief Push a function with zero or more arguments, but no return value, into the task queue. Does not return a future, so the user must use wait_for_tasks() or some other method to ensure that the task finishes executing, otherwise bad things will happen.
+     *
+     * @tparam F The type of the function.
+     * @tparam A The types of the arguments.
+     * @param task The function to push.
+     * @param args The zero or more arguments to pass to the function. Note that if the task is a class member function, the first argument must be a pointer to the object, i.e. &object (or this), followed by the actual arguments.
+     */
+    template <typename F, typename... A>
+    void push_task(F&& task, A&&... args)
+    {
+        std::function<void()> task_function = std::bind(std::forward<F>(task), std::forward<A>(args)...);
+        {
+            const std::scoped_lock tasks_lock(tasks_mutex);
+            tasks.push(task_function);
+            ++tasks_total;
+        }
+        task_available_cv.notify_one();
+    }
+
+    /**
+     * @brief Submit a function with zero or more arguments into the task queue. If the function has a return value, get a future for the eventual returned value. If the function has no return value, get an std::future<void> which can be used to wait until the task finishes.
+     *
+     * @tparam F The type of the function.
+     * @tparam A The types of the zero or more arguments to pass to the function.
+     * @tparam R The return type of the function (can be void).
+     * @param task The function to submit.
+     * @param args The zero or more arguments to pass to the function. Note that if the task is a class member function, the first argument must be a pointer to the object, i.e. &object (or this), followed by the actual arguments.
+     * @return A future to be used later to wait for the function to finish executing and/or obtain its returned value if it has one.
+     */
+    template <typename F, typename... A, typename R = std::invoke_result_t<std::decay_t<F>, std::decay_t<A>...>>
+    [[nodiscard]] std::future<R> submit(F&& task, A&&... args)
+    {
+        std::function<R()> task_function = std::bind(std::forward<F>(task), std::forward<A>(args)...);
+        std::shared_ptr<std::promise<R>> task_promise = std::make_shared<std::promise<R>>();
+        push_task(
+            [task_function, task_promise]
+            {
+                try
+                {
+                    if constexpr (std::is_void_v<R>)
+                    {
+                        std::invoke(task_function);
+                        task_promise->set_value();
+                    }
+                    else
+                    {
+                        task_promise->set_value(std::invoke(task_function));
+                    }
+                }
+                catch (...)
+                {
+                    try
+                    {
+                        task_promise->set_exception(std::current_exception());
+                    }
+                    catch (...)
+                    {
+                    }
+                }
+            });
+        return task_promise->get_future();
+    }
+
+    /**
+     * @brief Wait for tasks to be completed. Normally, this function waits for all tasks, both those that are currently running in the threads and those that are still waiting in the queue. Note: To wait for just one specific task, use submit() instead, and call the wait() member function of the generated future.
+     */
+    void wait_for_tasks()
+    {
+        if (!waiting)
+        {
+            waiting = true;
+            std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
+            task_done_cv.wait(tasks_lock, [this] { return (tasks_total == 0); });
+            waiting = false;
+        }
+    }
+
+private:
+    // ========================
+    // Private member functions
+    // ========================
+
+    /**
+     * @brief Create the threads in the pool and assign a worker to each thread.
+     */
+    void create_threads()
+    {
+        running = true;
+        for (concurrency_t i = 0; i < thread_count; ++i)
+        {
+            threads[i] = std::thread(&thread_pool_light::worker, this);
+        }
+    }
+
+    /**
+     * @brief Destroy the threads in the pool.
+     */
+    void destroy_threads()
+    {
+        running = false;
+        {
+            const std::scoped_lock tasks_lock(tasks_mutex);
+            task_available_cv.notify_all();
+        }
+        for (concurrency_t i = 0; i < thread_count; ++i)
+        {
+            threads[i].join();
+        }
+    }
+
+    /**
+     * @brief Determine how many threads the pool should have, based on the parameter passed to the constructor.
+     *
+     * @param thread_count_ The parameter passed to the constructor. If the parameter is a positive number, then the pool will be created with this number of threads. If the parameter is non-positive, or a parameter was not supplied (in which case it will have the default value of 0), then the pool will be created with the total number of hardware threads available, as obtained from std::thread::hardware_concurrency(). If the latter returns a non-positive number for some reason, then the pool will be created with just one thread.
+     * @return The number of threads to use for constructing the pool.
+     */
+    [[nodiscard]] concurrency_t determine_thread_count(const concurrency_t thread_count_)
+    {
+        if (thread_count_ > 0)
+            return thread_count_;
+        else
+        {
+            if (std::thread::hardware_concurrency() > 0)
+                return std::thread::hardware_concurrency();
+            else
+                return 1;
+        }
+    }
+
+    /**
+     * @brief A worker function to be assigned to each thread in the pool. Waits until it is notified by push_task() that a task is available, and then retrieves the task from the queue and executes it. Once the task finishes, the worker notifies wait_for_tasks() in case it is waiting.
+     */
+    void worker()
+    {
+        while (running)
+        {
+            std::function<void()> task;
+            std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
+            task_available_cv.wait(tasks_lock, [this] { return !tasks.empty() || !running; });
+            if (running)
+            {
+                task = std::move(tasks.front());
+                tasks.pop();
+                tasks_lock.unlock();
+                task();
+                tasks_lock.lock();
+                --tasks_total;
+                if (waiting)
+                    task_done_cv.notify_one();
+            }
+        }
+    }
+
+    // ============
+    // Private data
+    // ============
+
+    /**
+     * @brief An atomic variable indicating to the workers to keep running. When set to false, the workers permanently stop working.
+     */
+    std::atomic<bool> running = false;
+
+    /**
+     * @brief A condition variable used to notify worker() that a new task has become available.
+     */
+    std::condition_variable task_available_cv = {};
+
+    /**
+     * @brief A condition variable used to notify wait_for_tasks() that a tasks is done.
+     */
+    std::condition_variable task_done_cv = {};
+
+    /**
+     * @brief A queue of tasks to be executed by the threads.
+     */
+    std::queue<std::function<void()>> tasks = {};
+
+    /**
+     * @brief An atomic variable to keep track of the total number of unfinished tasks - either still in the queue, or running in a thread.
+     */
+    std::atomic<size_t> tasks_total = 0;
+
+    /**
+     * @brief A mutex to synchronize access to the task queue by different threads.
+     */
+    mutable std::mutex tasks_mutex = {};
+
+    /**
+     * @brief The number of threads in the pool.
+     */
+    concurrency_t thread_count = 0;
+
+    /**
+     * @brief A smart pointer to manage the memory allocated for the threads.
+     */
+    std::unique_ptr<std::thread[]> threads = nullptr;
+
+    /**
+     * @brief An atomic variable indicating that wait_for_tasks() is active and expects to be notified whenever a task is done.
+     */
+    std::atomic<bool> waiting = false;
+};
+
+} // namespace BS

BS_thread_pool_light_test.cpp

@@ -0,0 +1,657 @@
+/**
+ * @file BS_thread_pool_light_test.cpp
+ * @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
+ * @version 3.4.0
+ * @date 2023-05-12
+ * @copyright Copyright (c) 2023 Barak Shoshany. Licensed under the MIT license. If you found this project useful, please consider starring it on GitHub! If you use this library in software of any kind, please provide a link to the GitHub repository https://github.com/bshoshany/thread-pool in the source code and documentation. If you use this library in published research, please cite it as follows: Barak Shoshany, "A C++17 Thread Pool for High-Performance Scientific Computing", doi:10.5281/zenodo.4742687, arXiv:2105.00613 (May 2021)
+ *
+ * @brief BS::thread_pool_light: a fast, lightweight, and easy-to-use C++17 thread pool library. This program tests all aspects of the light version of the main library, but is not needed in order to use the library.
+ */
+
+#include <algorithm>          // std::min, std::sort, std::unique
+#include <atomic>             // std::atomic
+#include <chrono>             // std::chrono
+#include <cmath>              // std::abs
+#include <condition_variable> // std::condition_variable
+#include <cstdlib>            // std::quick_exit
+#include <exception>          // std::exception
+#include <future>             // std::future
+#include <iostream>           // std::cout
+#include <memory>             // std::make_unique, std::unique_ptr
+#include <mutex>              // std::mutex, std::scoped_lock, std::unique_lock
+#include <random>             // std::mt19937_64, std::random_device, std::uniform_int_distribution
+#include <stdexcept>          // std::runtime_error
+#include <string>             // std::string, std::to_string
+#include <thread>             // std::this_thread, std::thread
+#include <utility>            // std::forward
+#include <vector>             // std::vector
+
+// Include the header file for the thread pool library.
+#include "BS_thread_pool_light.hpp"
+
+// ================
+// Global variables
+// ================
+
+// A global thread pool object to be used throughout the test.
+BS::thread_pool_light pool;
+
+// A global random_device object to be used to seed some random number generators.
+std::random_device rd;
+
+// A global variable to measure how many checks succeeded.
+size_t tests_succeeded = 0;
+
+// A global variable to measure how many checks failed.
+size_t tests_failed = 0;
+
+// ================
+// Helper functions
+// ================
+
+/**
+ * @brief Print any number of items into std::cout.
+ *
+ * @tparam T The types of the items.
+ * @param items The items to print.
+ */
+template <typename... T>
+void print(T&&... items)
+{
+    (std::cout << ... << std::forward<T>(items));
+}
+
+/**
+ * @brief Print any number of items into std::cout, followed by a newline character.
+ *
+ * @tparam T The types of the items.
+ * @param items The items to print.
+ */
+template <typename... T>
+void println(T&&... items)
+{
+    print(std::forward<T>(items)..., '\n');
+}
+
+/**
+ * @brief Print a stylized header.
+ *
+ * @param text The text of the header. Will appear between two lines.
+ * @param symbol The symbol to use for the lines. Default is '='.
+ */
+void print_header(const std::string& text, const char symbol = '=')
+{
+    println();
+    println(std::string(text.length(), symbol));
+    println(text);
+    println(std::string(text.length(), symbol));
+}
+
+/**
+ * @brief Check if a condition is met, report the result, and keep count of the total number of successes and failures.
+ *
+ * @param condition The condition to check.
+ */
+void check(const bool condition)
+{
+    if (condition)
+    {
+        println("-> PASSED!");
+        ++tests_succeeded;
+    }
+    else
+    {
+        println("-> FAILED!");
+        ++tests_failed;
+    }
+}
+
+/**
+ * @brief Check if the expected result has been obtained, report the result, and keep count of the total number of successes and failures.
+ *
+ * @param condition The condition to check.
+ */
+template <typename T1, typename T2>
+void check(const T1 expected, const T2 obtained)
+{
+    print("Expected: ", expected, ", obtained: ", obtained);
+    if (expected == obtained)
+    {
+        println(" -> PASSED!");
+        ++tests_succeeded;
+    }
+    else
+    {
+        println(" -> FAILED!");
+        ++tests_failed;
+    }
+}
+
+// =========================================
+// Functions to verify the number of threads
+// =========================================
+
+/**
+ * @brief Count the number of unique threads in the pool. Submits a number of tasks equal to twice the thread count into the pool. Each task stores the ID of the thread running it, and then waits until released by the main thread. This ensures that each thread in the pool runs at least one task. The number of unique thread IDs is then counted from the stored IDs.
+ */
+BS::concurrency_t count_unique_threads()
+{
+    std::condition_variable ID_cv, total_cv;
+    std::mutex ID_mutex, total_mutex;
+    {
+        const BS::concurrency_t num_tasks = pool.get_thread_count() * 2;
+        std::vector<std::thread::id> thread_IDs(num_tasks);
+        std::unique_lock<std::mutex> total_lock(total_mutex);
+        BS::concurrency_t total_count = 0;
+        bool ID_release = false;
+        pool.wait_for_tasks();
+        for (std::thread::id& id : thread_IDs)
+            pool.push_task(
+                [&total_count, &id, &ID_release, &ID_cv, &total_cv, &ID_mutex, &total_mutex]
+                {
+                    id = std::this_thread::get_id();
+                    {
+                        const std::scoped_lock total_lock_local(total_mutex);
+                        ++total_count;
+                    }
+                    total_cv.notify_one();
+                    std::unique_lock<std::mutex> ID_lock_local(ID_mutex);
+                    ID_cv.wait(ID_lock_local, [&ID_release] { return ID_release; });
+                });
+        total_cv.wait(total_lock, [&total_count] { return total_count == pool.get_thread_count(); });
+        {
+            const std::scoped_lock ID_lock(ID_mutex);
+            ID_release = true;
+        }
+        ID_cv.notify_all();
+        total_cv.wait(total_lock, [&total_count, &num_tasks] { return total_count == num_tasks; });
+        pool.wait_for_tasks();
+        std::sort(thread_IDs.begin(), thread_IDs.end());
+        return static_cast<BS::concurrency_t>(std::unique(thread_IDs.begin(), thread_IDs.end()) - thread_IDs.begin());
+    }
+}
+
+/**
+ * @brief Check that the constructor works.
+ */
+void check_constructor()
+{
+    println("Checking that the thread pool reports a number of threads equal to the hardware concurrency...");
+    check(std::thread::hardware_concurrency(), pool.get_thread_count());
+    println("Checking that the manually counted number of unique thread IDs is equal to the reported number of threads...");
+    check(pool.get_thread_count(), count_unique_threads());
+}
+
+// =======================================
+// Functions to verify submission of tasks
+// =======================================
+
+/**
+ * @brief Check that push_task() works.
+ */
+void check_push_task()
+{
+    println("Checking that push_task() works for a function with no arguments or return value...");
+    {
+        bool flag = false;
+        pool.push_task([&flag] { flag = true; });
+        pool.wait_for_tasks();
+        check(flag);
+    }
+    println("Checking that push_task() works for a function with one argument and no return value...");
+    {
+        bool flag = false;
+        pool.push_task([](bool* flag_) { *flag_ = true; }, &flag);
+        pool.wait_for_tasks();
+        check(flag);
+    }
+    println("Checking that push_task() works for a function with two arguments and no return value...");
+    {
+        bool flag1 = false;
+        bool flag2 = false;
+        pool.push_task([](bool* flag1_, bool* flag2_) { *flag1_ = *flag2_ = true; }, &flag1, &flag2);
+        pool.wait_for_tasks();
+        check(flag1 && flag2);
+    }
+}
+
+/**
+ * @brief Check that submit() works.
+ */
+void check_submit()
+{
+    println("Checking that submit() works for a function with no arguments or return value...");
+    {
+        bool flag = false;
+        pool.submit([&flag] { flag = true; }).wait();
+        check(flag);
+    }
+    println("Checking that submit() works for a function with one argument and no return value...");
+    {
+        bool flag = false;
+        pool.submit([](bool* flag_) { *flag_ = true; }, &flag).wait();
+        check(flag);
+    }
+    println("Checking that submit() works for a function with two arguments and no return value...");
+    {
+        bool flag1 = false;
+        bool flag2 = false;
+        pool.submit([](bool* flag1_, bool* flag2_) { *flag1_ = *flag2_ = true; }, &flag1, &flag2).wait();
+        check(flag1 && flag2);
+    }
+    println("Checking that submit() works for a function with no arguments and a return value...");
+    {
+        bool flag = false;
+        std::future<int> flag_future = pool.submit(
+            [&flag]
+            {
+                flag = true;
+                return 42;
+            });
+        check(flag_future.get() == 42 && flag);
+    }
+    println("Checking that submit() works for a function with one argument and a return value...");
+    {
+        bool flag = false;
+        std::future<int> flag_future = pool.submit(
+            [](bool* flag_)
+            {
+                *flag_ = true;
+                return 42;
+            },
+            &flag);
+        check(flag_future.get() == 42 && flag);
+    }
+    println("Checking that submit() works for a function with two arguments and a return value...");
+    {
+        bool flag1 = false;
+        bool flag2 = false;
+        std::future<int> flag_future = pool.submit(
+            [](bool* flag1_, bool* flag2_)
+            {
+                *flag1_ = *flag2_ = true;
+                return 42;
+            },
+            &flag1, &flag2);
+        check(flag_future.get() == 42 && flag1 && flag2);
+    }
+}
+
+class flag_class
+{
+public:
+    void set_flag_no_args()
+    {
+        flag = true;
+    }
+
+    void set_flag_one_arg(const bool arg)
+    {
+        flag = arg;
+    }
+
+    int set_flag_no_args_return()
+    {
+        flag = true;
+        return 42;
+    }
+
+    int set_flag_one_arg_return(const bool arg)
+    {
+        flag = arg;
+        return 42;
+    }
+
+    bool get_flag() const
+    {
+        return flag;
+    }
+
+    void push_test_flag_no_args()
+    {
+        pool.push_task(&flag_class::set_flag_no_args, this);
+        pool.wait_for_tasks();
+        check(get_flag());
+    }
+
+    void push_test_flag_one_arg()
+    {
+        pool.push_task(&flag_class::set_flag_one_arg, this, true);
+        pool.wait_for_tasks();
+        check(get_flag());
+    }
+
+    void submit_test_flag_no_args()
+    {
+        pool.submit(&flag_class::set_flag_no_args, this).wait();
+        check(get_flag());
+    }
+
+    void submit_test_flag_one_arg()
+    {
+        pool.submit(&flag_class::set_flag_one_arg, this, true).wait();
+        check(get_flag());
+    }
+
+    void submit_test_flag_no_args_return()
+    {
+        std::future<int> flag_future = pool.submit(&flag_class::set_flag_no_args_return, this);
+        check(flag_future.get() == 42 && get_flag());
+    }
+
+    void submit_test_flag_one_arg_return()
+    {
+        std::future<int> flag_future = pool.submit(&flag_class::set_flag_one_arg_return, this, true);
+        check(flag_future.get() == 42 && get_flag());
+    }
+
+private:
+    bool flag = false;
+};
+
+/**
+ * @brief Check that submitting member functions works.
+ */
+void check_member_function()
+{
+    println("Checking that push_task() works for a member function with no arguments or return value...");
+    {
+        flag_class flag;
+        pool.push_task(&flag_class::set_flag_no_args, &flag);
+        pool.wait_for_tasks();
+        check(flag.get_flag());
+    }
+    println("Checking that push_task() works for a member function with one argument and no return value...");
+    {
+        flag_class flag;
+        pool.push_task(&flag_class::set_flag_one_arg, &flag, true);
+        pool.wait_for_tasks();
+        check(flag.get_flag());
+    }
+    println("Checking that submit() works for a member function with no arguments or return value...");
+    {
+        flag_class flag;
+        pool.submit(&flag_class::set_flag_no_args, &flag).wait();
+        check(flag.get_flag());
+    }
+    println("Checking that submit() works for a member function with one argument and no return value...");
+    {
+        flag_class flag;
+        pool.submit(&flag_class::set_flag_one_arg, &flag, true).wait();
+        check(flag.get_flag());
+    }
+    println("Checking that submit() works for a member function with no arguments and a return value...");
+    {
+        flag_class flag;
+        std::future<int> flag_future = pool.submit(&flag_class::set_flag_no_args_return, &flag);
+        check(flag_future.get() == 42 && flag.get_flag());
+    }
+    println("Checking that submit() works for a member function with one argument and a return value...");
+    {
+        flag_class flag;
+        std::future<int> flag_future = pool.submit(&flag_class::set_flag_one_arg_return, &flag, true);
+        check(flag_future.get() == 42 && flag.get_flag());
+    }
+}
+
+/**
+ * @brief Check that submitting member functions within an object works.
+ */
+void check_member_function_within_object()
+{
+    println("Checking that push_task() works within an object for a member function with no arguments or return value...");
+    {
+        flag_class flag;
+        flag.push_test_flag_no_args();
+    }
+    println("Checking that push_task() works within an object for a member function with one argument and no return value...");
+    {
+        flag_class flag;
+        flag.push_test_flag_one_arg();
+    }
+    println("Checking that submit() works within an object for a member function with no arguments or return value...");
+    {
+        flag_class flag;
+        flag.submit_test_flag_no_args();
+    }
+    println("Checking that submit() works within an object for a member function with one argument and no return value...");
+    {
+        flag_class flag;
+        flag.submit_test_flag_one_arg();
+    }
+    println("Checking that submit() works within an object for a member function with no arguments and a return value...");
+    {
+        flag_class flag;
+        flag.submit_test_flag_no_args_return();
+    }
+    println("Checking that submit() works within an object for a member function with one argument and a return value...");
+    {
+        flag_class flag;
+        flag.submit_test_flag_one_arg_return();
+    }
+}
+
+/**
+ * @brief Check that wait_for_tasks() works.
+ */
+void check_wait_for_tasks()
+{
+    const BS::concurrency_t n = pool.get_thread_count() * 10;
+    std::unique_ptr<std::atomic<bool>[]> flags = std::make_unique<std::atomic<bool>[]>(n);
+    for (BS::concurrency_t i = 0; i < n; ++i)
+        pool.push_task(
+            [&flags, i]
+            {
+                std::this_thread::sleep_for(std::chrono::milliseconds(10));
+                flags[i] = true;
+            });
+    println("Waiting for tasks...");
+    pool.wait_for_tasks();
+    bool all_flags = true;
+    for (BS::concurrency_t i = 0; i < n; ++i)
+        all_flags = all_flags && flags[i];
+    check(all_flags);
+}
+
+// ========================================
+// Functions to verify loop parallelization
+// ========================================
+
+/**
+ * @brief Check that push_loop() works for a specific range of indices split over a specific number of tasks, with no return value.
+ *
+ * @param random_start The first index in the loop.
+ * @param random_end The last index in the loop plus 1.
+ * @param num_tasks The number of tasks.
+ */
+template <typename T>
+void check_push_loop_no_return(const int64_t random_start, T random_end, const BS::concurrency_t num_tasks)
+{
+    if (random_start == random_end)
+        ++random_end;
+    println("Verifying that push_loop() from ", random_start, " to ", random_end, " with ", num_tasks, num_tasks == 1 ? " task" : " tasks", " modifies all indices...");
+    const size_t num_indices = static_cast<size_t>(std::abs(random_end - random_start));
+    const int64_t offset = std::min<int64_t>(random_start, static_cast<int64_t>(random_end));
+    std::unique_ptr<std::atomic<bool>[]> flags = std::make_unique<std::atomic<bool>[]>(num_indices);
+    const auto loop = [&flags, offset](const int64_t start, const int64_t end)
+    {
+        for (int64_t i = start; i < end; ++i)
+            flags[static_cast<size_t>(i - offset)] = true;
+    };
+    if (random_start == 0)
+        pool.push_loop(random_end, loop, num_tasks);
+    else
+        pool.push_loop(random_start, random_end, loop, num_tasks);
+    pool.wait_for_tasks();
+    bool all_flags = true;
+    for (size_t i = 0; i < num_indices; ++i)
+        all_flags = all_flags && flags[i];
+    check(all_flags);
+}
+
+/**
+ * @brief Check that push_loop() works using several different random values for the range of indices and number of tasks.
+ */
+void check_push_loop()
+{
+    std::mt19937_64 mt(rd());
+    std::uniform_int_distribution<int64_t> index_dist(-1000000, 1000000);
+    std::uniform_int_distribution<BS::concurrency_t> task_dist(1, pool.get_thread_count());
+    constexpr uint64_t n = 10;
+    for (uint64_t i = 0; i < n; ++i)
+        check_push_loop_no_return(index_dist(mt), index_dist(mt), task_dist(mt));
+    println("Verifying that push_loop() with identical start and end indices does nothing...");
+    bool flag = true;
+    const int64_t index = index_dist(mt);
+    pool.push_loop(index, index, [&flag](const int64_t, const int64_t) { flag = false; });
+    pool.wait_for_tasks();
+    check(flag);
+    println("Trying push_loop() with start and end indices of different types:");
+    const int64_t start = index_dist(mt);
+    const uint32_t end = static_cast<uint32_t>(std::abs(index_dist(mt)));
+    check_push_loop_no_return(start, end, task_dist(mt));
+    println("Trying the overload for push_loop() for the case where the first index is equal to 0:");
+    check_push_loop_no_return(0, index_dist(mt), task_dist(mt));
+}
+
+// ======================================
+// Functions to verify exception handling
+// ======================================
+
+/**
+ * @brief Check that exception handling works.
+ */
+void check_exceptions()
+{
+    println("Checking that exceptions are forwarded correctly by submit()...");
+    bool caught = false;
+    auto throws = []
+    {
+        println("Throwing exception...");
+        throw std::runtime_error("Exception thrown!");
+    };
+    std::future<void> my_future = pool.submit(throws);
+    try
+    {
+        my_future.get();
+    }
+    catch (const std::exception& e)
+    {
+        if (e.what() == std::string("Exception thrown!"))
+            caught = true;
+    }
+    check(caught);
+}
+
+// =====================================
+// Functions to verify vector operations
+// =====================================
+
+/**
+ * @brief Check that parallelized vector operations work as expected by calculating the sum of two randomized vectors of a specific size in two ways, single-threaded and multithreaded, and comparing the results.
+ */
+void check_vector_of_size(const size_t vector_size, const BS::concurrency_t num_tasks)
+{
+    std::vector<int64_t> vector_1(vector_size);
+    std::vector<int64_t> vector_2(vector_size);
+    std::mt19937_64 mt(rd());
+    std::uniform_int_distribution<int64_t> vector_dist(-1000000, 1000000);
+    for (size_t i = 0; i < vector_size; ++i)
+    {
+        vector_1[i] = vector_dist(mt);
+        vector_2[i] = vector_dist(mt);
+    }
+    println("Adding two vectors with ", vector_size, " elements using ", num_tasks, " tasks...");
+    std::vector<int64_t> sum_single(vector_size);
+    for (size_t i = 0; i < vector_size; ++i)
+        sum_single[i] = vector_1[i] + vector_2[i];
+    std::vector<int64_t> sum_multi(vector_size);
+    pool.push_loop(
+        0, vector_size,
+        [&sum_multi, &vector_1, &vector_2](const size_t start, const size_t end)
+        {
+            for (size_t i = start; i < end; ++i)
+                sum_multi[i] = vector_1[i] + vector_2[i];
+        },
+        num_tasks);
+    pool.wait_for_tasks();
+    bool vectors_equal = true;
+    for (size_t i = 0; i < vector_size; ++i)
+        vectors_equal = vectors_equal && (sum_single[i] == sum_multi[i]);
+    check(vectors_equal);
+}
+
+/**
+ * @brief Check that parallelized vector operations work as expected by calculating the sum of two randomized vectors in two ways, single-threaded and multithreaded, and comparing the results.
+ */
+void check_vectors()
+{
+    std::mt19937_64 mt(rd());
+    std::uniform_int_distribution<size_t> size_dist(0, 1000000);
+    std::uniform_int_distribution<BS::concurrency_t> task_dist(1, pool.get_thread_count());
+    for (size_t i = 0; i < 10; ++i)
+        check_vector_of_size(size_dist(mt), task_dist(mt));
+}
+
+// ==================
+// Main test function
+// ==================
+
+/**
+ * @brief Test that various aspects of the library are working as expected.
+ */
+void do_tests()
+{
+    print_header("Checking that the constructor works:");
+    check_constructor();
+
+    print_header("Checking that push_task() works:");
+    check_push_task();
+
+    print_header("Checking that submit() works:");
+    check_submit();
+
+    print_header("Checking that submitting member functions works:");
+    check_member_function();
+
+    print_header("Checking that submitting member functions from within an object works:");
+    check_member_function_within_object();
+
+    print_header("Checking that wait_for_tasks() works...");
+    check_wait_for_tasks();
+
+    print_header("Checking that push_loop() works:");
+    check_push_loop();
+
+    print_header("Checking that exception handling works:");
+    check_exceptions();
+
+    print_header("Testing that vector operations produce the expected results:");
+    check_vectors();
+}
+
+int main()
+{
+    println("BS::thread_pool_light: a fast, lightweight, and easy-to-use C++17 thread pool library");
+    println("(c) 2023 Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)");
+    println("GitHub: https://github.com/bshoshany/thread-pool\n");
+
+    println("Thread pool library version is ", BS_THREAD_POOL_LIGHT_VERSION, ".");
+    println("Hardware concurrency is ", std::thread::hardware_concurrency(), ".");
+
+    println("Important: Please do not run any other applications, especially multithreaded applications, in parallel with this test!");
+
+    do_tests();
+
+    if (tests_failed == 0)
+    {
+        print_header("SUCCESS: Passed all " + std::to_string(tests_succeeded) + " checks!", '+');
+        return 0;
+    }
+    else
+    {
+        print_header("FAILURE: Passed " + std::to_string(tests_succeeded) + " checks, but failed " + std::to_string(tests_failed) + "!", '+');
+        println("\nPlease submit a bug report at https://github.com/bshoshany/thread-pool/issues including the exact specifications of your system (OS, CPU, compiler, etc.) and the generated log file.");
+        std::quick_exit(static_cast<int>(tests_failed));
+    }
+}

CHANGELOG.md

@@ -10,9 +10,15 @@
 
 # `BS::thread_pool`: a fast, lightweight, and easy-to-use C++17 thread pool library
 
-By Barak Shoshany ([baraksh@gmail.com](mailto:baraksh@gmail.com)) ([https://baraksh.com/](https://baraksh.com/))
+By Barak Shoshany\
+Email: <baraksh@gmail.com>\
+Website: <https://baraksh.com/>\
+GitHub: <https://github.com/bshoshany>
 
 * [Version history](#version-history)
+    * [v3.4.0 (2023-05-12)](#v340-2023-05-12)
+    * [v3.3.0 (2022-08-03)](#v330-2022-08-03)
+    * [v3.2.0 (2022-07-28)](#v320-2022-07-28)
     * [v3.1.0 (2022-07-13)](#v310-2022-07-13)
     * [v3.0.0 (2022-05-30)](#v300-2022-05-30)
     * [v2.0.0 (2021-08-14)](#v200-2021-08-14)
@@ -29,6 +35,73 @@ By Barak Shoshany ([baraksh@gmail.com](mailto:baraksh@gmail.com)) ([https://bara
 
 ## Version history
 
+### v3.4.0 (2023-05-12)
+
+* `BS_thread_pool.hpp` and `BS_thread_pool_light.hpp`:
+    * Resolved an issue which could have caused `tasks_total` to not be synchronized in some cases. See [#70](https://github.com/bshoshany/thread-pool/pull/70).
+    * Resolved a deadlock which could rarely be caused when the pool was destructed or reset. See [#93](https://github.com/bshoshany/thread-pool/pull/93), [#100](https://github.com/bshoshany/thread-pool/pull/100), [#107](https://github.com/bshoshany/thread-pool/pull/107), and [#108](https://github.com/bshoshany/thread-pool/pull/108).
+    * Resolved a deadlock which could be caused when `wait_for_tasks()` was called more than once.
+    * Two new member functions have been added to the non-light version: `wait_for_tasks_duration()` and `wait_for_tasks_until()`. They allow waiting for the tasks to complete, but with a timeout. `wait_for_tasks_duration()` will stop waiting after the specified duration has passed, and `wait_for_tasks_until()` will stop waiting after the specified time point has been reached.
+    * Renamed `BS_THREAD_POOL_VERSION` in `BS_thread_pool_light.hpp` to `BS_THREAD_POOL_LIGHT_VERSION` and removed the `[light]` tag. This allows including both header files in the same program in case we want to use both the light and non-light thread pools simultaneously.
+* `BS_thread_pool_test.cpp` and `BS_thread_pool_light_test.cpp`:
+    * Fixed an issue that caused a compilation error when using MSVC and including `Windows.h`. See [#72](https://github.com/bshoshany/thread-pool/pull/72).
+    * The number and size of the vectors in the performance test (`BS_thread_pool_test.cpp` only) are now guaranteed to be multiples of the number of threads, for optimal performance.
+    * In `count_unique_threads()`, moved the condition variables and mutexes to the function scope to prevent cluttering the global scope.
+    * Three new tests have been added to `BS_thread_pool_test.cpp` to check the deadlocks issue that were resolved in this release (see above). The tests rely on the new wait for tasks with timeout feature, so they are not available in the light version.
+        * One test checks for deadlocks when calling `wait_for_tasks()` more than once.
+        * Two tests check for deadlocks when destructing and resetting the pool respectively. They are turned off by default, since they take a long time to complete, but can be turned on by setting `enable_long_deadlock_tests` to `true`.
+    * Two new tests have been added to the non-light version to check the new member functions `wait_for_tasks_duration()` and `wait_for_tasks_until()`.
+    * The test programs now return the number of failed tests upon exit, instead of just 1 if any number of tests failed, which was the case in previous versions. Also, if any tests failed, `std::quick_exit()` is invoked instead of `return`, to avoid getting stuck due to any lingering tasks or deadlocks.
+* `README.md`:
+    * Added documentation for the two new member functions, `wait_for_tasks_duration()` and `wait_for_tasks_until()`.
+    * Fixed Markdown rendering incorrectly on Visual Studio. See [#77](https://github.com/bshoshany/thread-pool/pull/77).
+    * The sample performance tests are now taken from a 40-core / 80-thread dual-CPU computing node, which is a more typical use case for high-performance scientific software.
+
+### v3.3.0 (2022-08-03)
+
+* `BS_thread_pool.hpp`:
+    * The public member variable `paused` of `BS::thread_pool` has been made private for future-proofing (in case future versions implement a more involved pausing mechanism) and better encapsulation. It is now accessible only via the `pause()`, `unpause()`, and `is_paused()` member functions. In other words:
+        * Replace `pool.paused = true` with `pool.pause()`.
+        * Replace `pool.paused = false` with `pool.unpause()`.
+        * Replace `if (pool.paused)` (or similar) with `if (pool.is_paused())`.
+    * The public member variable `f` of `BS::multi_future` has been renamed to `futures` for clarity, and has been made private for encapsulation and simplification purposes. Instead of operating on the vector `futures` itself, you can now use the `[]` operator of the `BS::multi_future` to access the future at a specific index directly, or the `push_back()` member function to append a new future to the list. The `size()` member function tells you how many futures are currently stored in the object.
+    * The explicit casts of `std::endl` and `std::flush`, added in v3.2.0 to enable flushing a `BS::synced_stream`, caused ODR (One Definition Rule) violations if `BS_thread_pool.hpp` was included in two different translation units, since they were mistakenly not defined as `inline`. To fix this, I decided to make them static members of `BS::synced_stream` instead of global variables, which also makes the code better organized in my opinion. These objects can now be accessed as `BS::synced_stream::endl` and `BS::synced_stream::flush`. I also added an example for how to use them in `README.md`. See [#64](https://github.com/bshoshany/thread-pool/issues/64).
+* `BS_thread_pool_light.hpp`:
+    * This package started out as a very lightweight thread pool, but over time has expanded to include many additional features, and at the time of writing it has a total of 340 lines of code, including all the helper classes. Therefore, I have decided to bundle a light version of the thread pool in a separate and stand-alone header file, `BS_thread_pool_light.hpp`, with only 170 lines of code (half the size of the full package). This file does not contain any of the helper classes, only a new `BS::thread_pool_light` class, which is a minimal thread pool with only the 5 most basic member functions:
+        * `get_thread_count()`
+        * `push_loop()`
+        * `push_task()`
+        * `submit()`
+        * `wait_for_tasks()`
+    * A separate test program `BS_thread_pool_light_test.cpp` tests only the features of the lightweight `BS::thread_pool_light` class. In the spirit of minimalism, it does not generate a log file and does not do any benchmarks.
+    * To be perfectly clear, each header file is 100% stand-alone. If you wish to use the full package, you only need `BS_thread_pool.hpp`, and if you wish to use the light version, you only need `BS_thread_pool_light.hpp`. Only a single header file needs to be included in your project.
+
+### v3.2.0 (2022-07-28)
+
+* `BS_thread_pool.hpp`:
+    * Main `BS::thread_pool` class:
+        * Added a new member function, `push_loop()`, which does the same thing as `parallelize_loop()`, except that it does not return a `BS::multi_future` with the futures for each block. Just like `push_task()` vs. `submit()`, this avoids the overhead of creating the futures, but the user must use `wait_for_tasks()` or some other method to ensure that the loop finishes executing, otherwise bad things will happen.
+        * `push_task()` and `submit()` now utilize perfect forwarding in order to support more types of tasks - in particular member functions, which in previous versions could not be submitted unless wrapped in a lambda. To submit a member function, use the syntax `submit(&class::function, &object, args)`. More information can be found in `README.md`. See [#9](https://github.com/bshoshany/thread-pool/issues/9).
+        * `push_loop()` and `parallelize_loop()` now have overloads where the first argument (the first index in the loop) is omitted, in which case it is assumed to be 0. This is for convenience, as the case where the first index is 0 is very common.
+    * Helper classes:
+        * `BS::synced_stream` now utilizes perfect forwarding in the member functions `print()` and `println()`.
+        * Previously, it was impossible to pass the flushing manipulators `std::endl` and `std::flush` to `print()` and `println()`, since the compiler could not figure out which template specializations to use. The new objects `BS::endl` and `BS::flush` are explicit casts of these manipulators, whose sole purpose is to enable passing them to `print()` and `println()`.
+        * `BS::multi_future::get()` now rethrows exceptions generated by the futures, even if the futures return `void`. See [#62](https://github.com/bshoshany/thread-pool/pull/62).
+        * Added a new helper class, `BS::blocks`, which is used by `parallelize_loop()` and `push_loop()` to divide a range into blocks. This class is not documented in `README.md`, as it most likely will not be of interest to most users, but it is still publicly available, in case you want to parallelize something manually but still benefit from the built-in algorithm for splitting a range into blocks.
+* `BS_thread_pool_test.cpp`:
+    * Added plenty of new tests for the new features described above.
+    * Fixed a bug in `count_unique_threads()` that caused it to get stuck on certain systems.
+    * `dual_println()` now also flushes the stream using `BS::endl`, so that if the test gets stuck, the log file will still contain everything up to that point. (Note: It is a common misconception that `std::endl` and `'\n'` are interchangeable. `std::endl` not only prints a newline character, it also flushes the stream, which is not always desirable, as it may reduce performance.)
+    * The performance test has been modified as follows:
+        * Instead of generating random vectors using `std::mersenne_twister_engine`, which proved to be inconsistent across different compilers and systems, the test now generates each element via an arbitrarily-chosen numerical operation. In my testing, this provided much more consistent results.
+        * Instead of using a hard-coded vector size, a suitable vector size is now determined dynamically at runtime.
+        * Instead of using `parallelize_loop()`, the test now uses the new `push_loop()` function to squeeze out a bit more performance.
+        * Instead of setting the test parameters to achieve a fixed single-threaded mean execution time of 300 ms, the test now aims to achieve a fixed multi-threaded mean execution time of 50 ms when the number of blocks is equal to the number of threads. This allows for more reliable results on very fast CPUs with a very large number of threads, where the mean execution time when using all the threads could previously be below a statistically significant value.
+        * The number of vectors is now restricted to be a multiple of the number of threads, so that the blocks are always all of the same size.
+* `README.md`:
+    * Added instructions and examples for the new features described above.
+    * Rewrote the documentation for `parallelize_loop()` to make it clearer.
+
 ### v3.1.0 (2022-07-13)
 
 * `BS_thread_pool.hpp`:

LICENSE.txt

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2022 Barak Shoshany
+Copyright (c) 2023 Barak Shoshany
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal