Updated to v3.1.0

2026-07-29 15:03:00 +04:00 · 2022-07-13 19:50:44 -04:00
parent 9d43f5d05d
commit cca27608ee
4 changed files with 261 additions and 155 deletions
@@ -1,9 +1,9 @@
 /**
 * @file BS_thread_pool_test.cpp
 * @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
- * @version 3.0.0
- * @date 2022-05-30
- * @copyright Copyright (c) 2022 Barak Shoshany. Licensed under the MIT license. 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.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)
 *
 * @brief BS::thread_pool: a fast, lightweight, and easy-to-use C++17 thread pool library. This program tests all aspects of the library, but is not needed in order to use the library.
 */
@@ -13,24 +13,26 @@
 #define _CRT_SECURE_NO_WARNINGS
 #endif

-#include <algorithm> // std::min, std::min_element, std::sort, std::unique
-#include <atomic>    // std::atomic
-#include <chrono>    // std::chrono
-#include <cmath>     // std::abs, std::llround, std::round, std::sqrt
-#include <ctime>     // std::localtime, std::strftime, std::time_t
-#include <exception> // std::exception
-#include <fstream>   // std::ofstream
-#include <future>    // std::future
-#include <iomanip>   // std::setprecision, std::setw
-#include <ios>       // std::fixed
-#include <iostream>  // std::cout
-#include <limits>    // std::numeric_limits
-#include <random>    // std::mt19937_64, std::random_device, std::uniform_int_distribution, std::uniform_real_distribution
-#include <stdexcept> // std::runtime_error
-#include <string>    // std::string, std::to_string
-#include <thread>    // std::this_thread, std::thread
-#include <utility>   // std::pair
-#include <vector>    // std::begin, std::end, std::vector
+#include <algorithm>          // std::min, std::min_element, std::sort, std::unique
+#include <atomic>             // std::atomic
+#include <chrono>             // std::chrono
+#include <cmath>              // std::abs, std::llround, std::round, std::sqrt
+#include <condition_variable> // std::condition_variable
+#include <ctime>              // std::localtime, std::strftime, std::time_t
+#include <exception>          // std::exception
+#include <fstream>            // std::ofstream
+#include <future>             // std::future
+#include <iomanip>            // std::setprecision, std::setw
+#include <ios>                // std::fixed
+#include <iostream>           // std::cout
+#include <limits>             // std::numeric_limits
+#include <mutex>              // std::mutex, std::scoped_lock, std::unique_lock
+#include <random>             // std::mt19937_64, std::random_device, std::uniform_int_distribution, std::uniform_real_distribution
+#include <stdexcept>          // std::runtime_error
+#include <string>             // std::string, std::to_string
+#include <thread>             // std::this_thread, std::thread
+#include <utility>            // std::pair
+#include <vector>             // std::begin, std::end, std::vector

 // Include the header file for the thread pool library.
 #include "BS_thread_pool.hpp"
@@ -39,13 +41,13 @@
 // Global variables
 // ================

-// Set to false to disable output to a log file.
+// Whether to output to a log file in addition to the standard output.
 constexpr bool output_log = true;

-// Set to false to disable testing.
+// Whether to perform the tests.
 constexpr bool enable_tests = true;

-// Set to false to disable the benchmarks.
+// Whether to perform the benchmarks.
 constexpr bool enable_benchmarks = true;

 // Two global synced_streams objects. One prints to std::cout, and the other to a file.
@@ -139,34 +141,66 @@ void check(const bool condition)
    }
 }

+/**
+ * @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)
+{
+    dual_print("Expected: ", expected, ", obtained: ", obtained);
+    if (expected == obtained)
+    {
+        dual_println(" -> PASSED!");
+        ++tests_succeeded;
+    }
+    else
+    {
+        dual_println(" -> FAILED!");
+        ++tests_failed;
+    }
+}
+
 // =========================================
 // Functions to verify the number of threads
 // =========================================

 /**
- * @brief Store the ID of the current thread in memory. Waits for a short time to ensure it does not get evaluated by more than one thread.
- *
- * @param location A pointer to the location where the thread ID should be stored.
- */
-void store_ID(std::thread::id* location)
-{
-    *location = std::this_thread::get_id();
-    std::this_thread::sleep_for(std::chrono::milliseconds(10));
-}
-
-/**
- * @brief Count the number of unique threads in the thread pool to ensure that the correct number of individual threads was created. Pushes a number of tasks equal to four times the thread count into the thread pool, and count the number of unique thread IDs returned by the tasks.
+ * @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::vector<std::thread::id> thread_IDs(pool.get_thread_count() * 4);
-    BS::multi_future<void> futures;
+    const BS::concurrency_t num_tasks = pool.get_thread_count() * 2;
+    std::vector<std::thread::id> thread_IDs(num_tasks);
+    std::mutex ID_mutex, total_mutex;
+    std::condition_variable ID_cv, total_cv;
+    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)
-        futures.f.push_back(pool.submit(store_ID, &id));
-    futures.wait();
+        pool.push_task(
+            [&]
+            {
+                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, [&] { return ID_release; });
+            });
+    total_cv.wait_for(total_lock, std::chrono::milliseconds(500), [&] { 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_for(total_lock, std::chrono::milliseconds(500), [&] { return total_count == num_tasks; });
    std::sort(thread_IDs.begin(), thread_IDs.end());
-    BS::concurrency_t unique_threads = (BS::concurrency_t)(std::unique(thread_IDs.begin(), thread_IDs.end()) - thread_IDs.begin());
-    return unique_threads;
+    return static_cast<BS::concurrency_t>(std::unique(thread_IDs.begin(), thread_IDs.end()) - thread_IDs.begin());
 }

 /**
@@ -175,9 +209,9 @@ BS::concurrency_t count_unique_threads()
 void check_constructor()
 {
    dual_println("Checking that the thread pool reports a number of threads equal to the hardware concurrency...");
-    check(pool.get_thread_count() == std::thread::hardware_concurrency());
+    check(std::thread::hardware_concurrency(), pool.get_thread_count());
    dual_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());
+    check(pool.get_thread_count(), count_unique_threads());
 }

 /**
@@ -187,14 +221,14 @@ void check_reset()
 {
    pool.reset(std::thread::hardware_concurrency() / 2);
    dual_println("Checking that after reset() the thread pool reports a number of threads equal to half the hardware concurrency...");
-    check(pool.get_thread_count() == std::thread::hardware_concurrency() / 2);
+    check(std::thread::hardware_concurrency() / 2, pool.get_thread_count());
    dual_println("Checking that after reset() the manually counted number of unique thread IDs is equal to the reported number of threads...");
-    check(pool.get_thread_count() == count_unique_threads());
+    check(pool.get_thread_count(), count_unique_threads());
    pool.reset(std::thread::hardware_concurrency());
    dual_println("Checking that after a second reset() the thread pool reports a number of threads equal to the hardware concurrency...");
-    check(pool.get_thread_count() == std::thread::hardware_concurrency());
+    check(std::thread::hardware_concurrency(), pool.get_thread_count());
    dual_println("Checking that after a second reset() the manually counted number of unique thread IDs is equal to the reported number of threads...");
-    check(pool.get_thread_count() == count_unique_threads());
+    check(pool.get_thread_count(), count_unique_threads());
 }

 // =======================================
@@ -298,7 +332,7 @@ void check_submit()
 void check_wait_for_tasks()
 {
    const BS::concurrency_t n = pool.get_thread_count() * 10;
-    std::vector<std::atomic<bool>> flags(n);
+    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]
@@ -306,6 +340,7 @@ void check_wait_for_tasks()
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                flags[i] = true;
            });
+    dual_println("Waiting for tasks...");
    pool.wait_for_tasks();
    bool all_flags = true;
    for (BS::concurrency_t i = 0; i < n; ++i)
@@ -331,13 +366,13 @@ void check_parallelize_loop_no_return(const int64_t random_start, int64_t random
    dual_println("Verifying that a 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(random_start, random_end);
-    std::vector<std::atomic<bool>> flags(num_indices);
+    std::unique_ptr<std::atomic<bool>[]> flags = std::make_unique<std::atomic<bool>[]>(num_indices);
    pool.parallelize_loop(
            random_start, random_end,
            [&flags, offset](const int64_t start, const int64_t end)
            {
                for (int64_t i = start; i < end; ++i)
-                    flags[(size_t)(i - offset)] = true;
+                    flags[static_cast<size_t>(i - offset)] = true;
            },
            num_tasks)
        .wait();
@@ -373,7 +408,7 @@ void check_parallelize_loop_return(const int64_t random_start, int64_t random_en
    int64_t sum = 0;
    for (const int64_t& s : sums_vector)
        sum += s;
-    check(sum * 2 == std::abs(random_start - random_end) * (random_start + random_end - 1));
+    check(std::abs(random_start - random_end) * (random_start + random_end - 1), sum * 2);
 }

 /**
@@ -404,7 +439,7 @@ void check_task_monitoring()
    dual_println("Resetting pool to ", n, " threads.");
    pool.reset(n);
    dual_println("Submitting ", n * 3, " tasks.");
-    std::vector<std::atomic<bool>> release(n * 3);
+    std::unique_ptr<std::atomic<bool>[]> release = std::make_unique<std::atomic<bool>[]>(n * 3);
    for (BS::concurrency_t i = 0; i < n * 3; ++i)
        pool.push_task(
            [&release, i]
@@ -416,21 +451,25 @@ void check_task_monitoring()
    constexpr std::chrono::milliseconds sleep_time(300);
    std::this_thread::sleep_for(sleep_time);
    dual_println("After submission, should have: ", n * 3, " tasks total, ", n, " tasks running, ", n * 2, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == n * 3 && pool.get_tasks_running() == n && pool.get_tasks_queued() == n * 2);
    for (BS::concurrency_t i = 0; i < n; ++i)
        release[i] = true;
    std::this_thread::sleep_for(sleep_time);
    dual_println("After releasing ", n, " tasks, should have: ", n * 2, " tasks total, ", n, " tasks running, ", n, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == n * 2 && pool.get_tasks_running() == n && pool.get_tasks_queued() == n);
    for (BS::concurrency_t i = n; i < n * 2; ++i)
        release[i] = true;
    std::this_thread::sleep_for(sleep_time);
    dual_println("After releasing ", n, " more tasks, should have: ", n, " tasks total, ", n, " tasks running, ", 0, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == n && pool.get_tasks_running() == n && pool.get_tasks_queued() == 0);
    for (BS::concurrency_t i = n * 2; i < n * 3; ++i)
        release[i] = true;
    std::this_thread::sleep_for(sleep_time);
    dual_println("After releasing the final ", n, " tasks, should have: ", 0, " tasks total, ", 0, " tasks running, ", 0, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == 0 && pool.get_tasks_running() == 0 && pool.get_tasks_queued() == 0);
    dual_println("Resetting pool to ", std::thread::hardware_concurrency(), " threads.");
    pool.reset(std::thread::hardware_concurrency());
@@ -455,27 +494,33 @@ void check_pausing()
                dual_println("Task ", i, " done.");
            });
    dual_println("Immediately after submission, should have: ", n * 3, " tasks total, ", 0, " tasks running, ", n * 3, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == n * 3 && pool.get_tasks_running() == 0 && pool.get_tasks_queued() == n * 3);
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
    dual_println("300ms later, should still have: ", n * 3, " tasks total, ", 0, " tasks running, ", n * 3, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == n * 3 && pool.get_tasks_running() == 0 && pool.get_tasks_queued() == n * 3);
    dual_println("Unpausing pool.");
    pool.paused = false;
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
    dual_println("300ms later, should have: ", n * 2, " tasks total, ", n, " tasks running, ", n, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == n * 2 && pool.get_tasks_running() == n && pool.get_tasks_queued() == n);
    dual_println("Pausing pool and using wait_for_tasks() to wait for the running tasks.");
    pool.paused = true;
    pool.wait_for_tasks();
    dual_println("After waiting, should have: ", n, " tasks total, ", 0, " tasks running, ", n, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == n && pool.get_tasks_running() == 0 && pool.get_tasks_queued() == n);
    std::this_thread::sleep_for(std::chrono::milliseconds(200));
    dual_println("200ms later, should still have: ", n, " tasks total, ", 0, " tasks running, ", n, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == n && pool.get_tasks_running() == 0 && pool.get_tasks_queued() == n);
    dual_println("Unpausing pool and using wait_for_tasks() to wait for all tasks.");
    pool.paused = false;
    pool.wait_for_tasks();
    dual_println("After waiting, should have: ", 0, " tasks total, ", 0, " tasks running, ", 0, " tasks queued...");
+    dual_print("Result: ", pool.get_tasks_total(), " tasks total, ", pool.get_tasks_running(), " tasks running, ", pool.get_tasks_queued(), " tasks queued ");
    check(pool.get_tasks_total() == 0 && pool.get_tasks_running() == 0 && pool.get_tasks_queued() == 0);
    dual_println("Resetting pool to ", std::thread::hardware_concurrency(), " threads.");
    pool.reset(std::thread::hardware_concurrency());
@@ -491,7 +536,12 @@ void check_pausing()
 void check_exceptions()
 {
    bool caught = false;
-    std::future<void> my_future = pool.submit([] { throw std::runtime_error("Exception thrown!"); });
+    std::future<void> my_future = pool.submit(
+        []
+        {
+            dual_println("Throwing exception...");
+            throw std::runtime_error("Exception thrown!");
+        });
    try
    {
        my_future.get();
@@ -766,7 +816,7 @@ int main()
    if (output_log)
        log_file.open(log_filename);

-    dual_println("A C++17 Thread Pool for High-Performance Scientific Computing");
+    dual_println("BS::thread_pool: a fast, lightweight, and easy-to-use C++17 thread pool library");
    dual_println("(c) 2022 Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)");
    dual_println("GitHub: https://github.com/bshoshany/thread-pool\n");