potok/thread-pool
1
0
Fork 0
mirror of https://github.com/bshoshany/thread-pool.git synced 2026-07-21 19:13:00 +04:00
Code Packages Releases Wiki Activity

Updated to v3.4.0

Browse Source
This commit is contained in:
Barak Shoshany
2023-05-12 14:45:07 -04:00
committed by GitHub
parent 67fad04348
commit 87d415f89c
7 changed files with 2190 additions and 1868 deletions
Download Patch File Download Diff File Expand all files Collapse all files
BS_thread_pool.hpp
+60 -10
View File
@@ -3,14 +3,14 @@
/**
* @file BS_thread_pool.hpp
* @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
* @version 3.3.0
* @date 2022-08-03
* @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::blocks, BS:synced_stream, and BS::timer.
*/
#define BS_THREAD_POOL_VERSION "v3.3.0 (2022-08-03)"
#define BS_THREAD_POOL_VERSION "v3.4.0 (2023-05-12)"
#include <atomic> // std::atomic
#include <chrono> // std::chrono
@@ -428,8 +428,8 @@ public:
{
const std::scoped_lock tasks_lock(tasks_mutex);
tasks.push(task_function);
++tasks_total;
}
++tasks_total;
task_available_cv.notify_one();
}
@@ -507,10 +507,57 @@ public:
*/
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;
}
}
/**
* @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 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.
*/
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:
@@ -536,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();
BS_thread_pool_light.hpp
+16 -10
View File
@@ -3,14 +3,14 @@
/**
* @file BS_thread_pool_light.hpp
* @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
* @version 3.3.0
* @date 2022-08-03
* @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_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_VERSION "v3.3.0 (2022-08-03) [light]"
#define BS_THREAD_POOL_LIGHT_VERSION "v3.4.0 (2023-05-12)"
#include <atomic> // std::atomic
#include <condition_variable> // std::condition_variable
@@ -139,8 +139,8 @@ public:
{
const std::scoped_lock tasks_lock(tasks_mutex);
tasks.push(task_function);
++tasks_total;
}
++tasks_total;
task_available_cv.notify_one();
}
@@ -193,10 +193,13 @@ public:
*/
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 == 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 == 0); });
waiting = false;
}
}
private:
@@ -222,7 +225,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();
BS_thread_pool_light_test.cpp
+49 -49
View File
@@ -1,26 +1,22 @@
/**
* @file BS_thread_pool_light_test.cpp
* @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
* @version 3.3.0
* @date 2022-08-03
* @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_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.
*/
// Get rid of annoying MSVC warning.
#ifdef _MSC_VER
#define _CRT_SECURE_NO_WARNINGS
#endif
#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, std::endl
#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
@@ -43,8 +39,10 @@ BS::thread_pool_light pool;
// A global random_device object to be used to seed some random number generators.
std::random_device rd;
// Global variables to measure how many checks succeeded and how many failed.
// 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;
// ================
@@ -52,7 +50,7 @@ size_t tests_failed = 0;
// ================
/**
* @brief Print any number of items into both std::cout and the log file, syncing both independently.
* @brief Print any number of items into std::cout.
*
* @tparam T The types of the items.
* @param items The items to print.
@@ -64,7 +62,7 @@ void print(T&&... items)
}
/**
* @brief Print any number of items into both std::cout and the log file, syncing both independently. Also prints a newline character, and flushes the stream.
* @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.
@@ -72,7 +70,7 @@ void print(T&&... items)
template <typename... T>
void println(T&&... items)
{
print(std::forward<T>(items)..., static_cast<std::ostream& (&)(std::ostream&)>(std::endl));
print(std::forward<T>(items)..., '\n');
}
/**
@@ -136,39 +134,41 @@ void check(const T1 expected, const T2 obtained)
/**
* @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.
*/
std::condition_variable ID_cv, total_cv;
std::mutex ID_mutex, total_mutex;
BS::concurrency_t count_unique_threads()
{
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 = 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(); });
std::condition_variable ID_cv, total_cv;
std::mutex ID_mutex, total_mutex;
{
const std::scoped_lock ID_lock(ID_mutex);
ID_release = true;
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());
}
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());
}
/**
@@ -471,7 +471,7 @@ void check_push_loop_no_return(const int64_t random_start, T random_end, const B
++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(random_start, static_cast<int64_t>(random_end));
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)
{
@@ -492,7 +492,7 @@ void check_push_loop_no_return(const int64_t random_start, T random_end, const B
/**
* @brief Check that push_loop() works using several different random values for the range of indices and number of tasks.
*/
void check_parallelize_loop()
void check_push_loop()
{
std::mt19937_64 mt(rd());
std::uniform_int_distribution<int64_t> index_dist(-1000000, 1000000);
@@ -621,7 +621,7 @@ void do_tests()
check_wait_for_tasks();
print_header("Checking that push_loop() works:");
check_parallelize_loop();
check_push_loop();
print_header("Checking that exception handling works:");
check_exceptions();
@@ -633,10 +633,10 @@ void do_tests()
int main()
{
println("BS::thread_pool_light: a fast, lightweight, and easy-to-use C++17 thread pool library");
println("(c) 2022 Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)");
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_VERSION, ".");
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!");
@@ -646,12 +646,12 @@ int main()
if (tests_failed == 0)
{
print_header("SUCCESS: Passed all " + std::to_string(tests_succeeded) + " checks!", '+');
return EXIT_SUCCESS;
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.");
return EXIT_FAILURE;
std::quick_exit(static_cast<int>(tests_failed));
}
}
BS_thread_pool_test.cpp
+223 -44
View File
@@ -1,9 +1,9 @@
/**
* @file BS_thread_pool_test.cpp
* @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
* @version 3.3.0
* @date 2022-08-03
* @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 program tests all aspects of the library, but is not needed in order to use the library.
*/
@@ -18,6 +18,7 @@
#include <chrono> // std::chrono
#include <cmath> // std::abs, std::cos, std::exp, std::llround, std::log, std::round, std::sin, std::sqrt
#include <condition_variable> // std::condition_variable
#include <cstdlib> // std::quick_exit
#include <ctime> // std::localtime, std::strftime, std::time, std::time_t
#include <exception> // std::exception
#include <fstream> // std::ofstream
@@ -25,14 +26,13 @@
#include <iomanip> // std::setprecision, std::setw
#include <ios> // std::fixed
#include <iostream> // std::cout
#include <limits> // std::numeric_limits
#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::pair
#include <utility> // std::forward, std::pair
#include <vector> // std::begin, std::end, std::vector
// Include the header file for the thread pool library.
@@ -51,9 +51,16 @@ constexpr bool enable_tests = true;
// 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.
// Whether to perform the long deadlock tests. Defaults to false since they can take much longer than the other tests.
constexpr bool enable_long_deadlock_tests = false;
// A global synced_stream object which prints to std::cout.
BS::synced_stream sync_cout(std::cout);
// A global stream object used to access the log file.
std::ofstream log_file;
// A global synced_stream object which prints to the log file.
BS::synced_stream sync_file(log_file);
// A global thread pool object to be used throughout the test.
@@ -62,8 +69,10 @@ BS::thread_pool pool;
// A global random_device object to be used to seed some random number generators.
std::random_device rd;
// Global variables to measure how many checks succeeded and how many failed.
// 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;
// ================
@@ -170,39 +179,41 @@ void check(const T1 expected, const T2 obtained)
/**
* @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.
*/
std::condition_variable ID_cv, total_cv;
std::mutex ID_mutex, total_mutex;
BS::concurrency_t count_unique_threads()
{
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 = 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(); });
std::condition_variable ID_cv, total_cv;
std::mutex ID_mutex, total_mutex;
{
const std::scoped_lock ID_lock(ID_mutex);
ID_release = true;
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());
}
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());
}
/**
@@ -233,6 +244,75 @@ void check_reset()
check(pool.get_thread_count(), count_unique_threads());
}
// ================================
// Functions to check for deadlocks
// ================================
// An auxiliary thread pool used by check_deadlock().
BS::thread_pool check_deadlock_pool;
/**
* @brief Check that the specified function does not create deadlocks. The function will be run many times to increase the probability of encountering a deadlock as a result of subtle timing issues. Uses an auxiliary pool so the whole test doesn't get stuck if a deadlock is encountered.
*
* @tparam F The type of the function.
* @param task The function to try.
*/
template <typename F>
void check_deadlock(const F&& task)
{
constexpr uint32_t tries = 10000;
uint32_t i = 0;
check_deadlock_pool.push_task(
[&i, &task]
{
do
task();
while (++i < tries);
});
bool passed = false;
while (true)
{
uint32_t old_i = i;
check_deadlock_pool.wait_for_tasks_duration(std::chrono::milliseconds(500));
if (i == tries)
{
dual_println("Successfully finished all tries!");
passed = true;
break;
}
else if (i > old_i)
{
dual_println("Finished ", i, " tries out of ", tries, "...");
}
else
{
dual_println("Error: deadlock detected!");
passed = false;
break;
}
}
check(passed);
}
/**
* @brief Check that the destructor does not create deadlocks.
*/
void check_destructor_deadlock()
{
dual_println("Checking for destruction deadlocks...");
check_deadlock([] { BS::thread_pool temp_pool; });
}
/**
* @brief Check that reset() does not create deadlocks.
*/
void check_reset_deadlock()
{
dual_println("Checking for reset deadlocks...");
BS::thread_pool temp_pool;
check_deadlock([&temp_pool] { temp_pool.reset(); });
}
// =======================================
// Functions to verify submission of tasks
// =======================================
@@ -504,6 +584,95 @@ void check_wait_for_tasks()
check(all_flags);
}
/**
* @brief Check that calling wait_for_tasks() more than once doesn't create a deadlock. Uses a mechanism similar to check_deadlock().
*/
void check_wait_for_tasks_deadlock()
{
dual_println("Checking for deadlocks when waiting for tasks...");
BS::thread_pool temp_pool(1);
temp_pool.push_task([] { std::this_thread::sleep_for(std::chrono::milliseconds(500)); });
constexpr uint32_t n_waiting_tasks = 1000;
std::atomic<uint32_t> count = 0;
for (uint32_t j = 0; j < n_waiting_tasks; ++j)
{
check_deadlock_pool.push_task(
[&temp_pool, &count]
{
temp_pool.wait_for_tasks();
++count;
});
}
bool passed = false;
while (true)
{
uint32_t old_count = count;
check_deadlock_pool.wait_for_tasks_duration(std::chrono::milliseconds(1000));
if (count == n_waiting_tasks)
{
dual_println("All waiting tasks successfully finished!");
passed = true;
break;
}
else if (count > old_count)
{
dual_println(count, " tasks out of ", n_waiting_tasks, " finished waiting...");
}
else
{
dual_println("Error: deadlock detected!");
passed = false;
break;
}
}
check(passed);
}
/**
* @brief Check that wait_for_tasks_duration() works.
*/
void check_wait_for_tasks_duration()
{
dual_println("Checking that wait_for_tasks_duration() works...");
pool.reset();
std::atomic<bool> done = false;
pool.push_task(
[&done]
{
std::this_thread::sleep_for(std::chrono::milliseconds(250));
done = true;
});
dual_println("Task submitted. Waiting for 10ms...");
pool.wait_for_tasks_duration(std::chrono::milliseconds(10));
check(!done);
dual_println("Waiting for 500ms...");
pool.wait_for_tasks_duration(std::chrono::milliseconds(500));
check(done);
}
/**
* @brief Check that wait_for_tasks_until() works.
*/
void check_wait_for_tasks_until()
{
dual_println("Checking that wait_for_tasks_until() works...");
pool.reset();
std::atomic<bool> done = false;
pool.push_task(
[&done]
{
std::this_thread::sleep_for(std::chrono::milliseconds(250));
done = true;
});
std::chrono::system_clock::time_point now = std::chrono::system_clock::now();
dual_println("Task submitted. Waiting until 10ms from submission time...");
pool.wait_for_tasks_until(now + std::chrono::milliseconds(10));
check(!done);
dual_println("Waiting until 500ms from submission time...");
pool.wait_for_tasks_until(now + std::chrono::milliseconds(500));
check(done);
}
// ========================================
// Functions to verify loop parallelization
// ========================================
@@ -523,7 +692,7 @@ void check_parallelize_loop_no_return(const int64_t random_start, T random_end,
++random_end;
dual_println("Verifying that ", use_push ? "push_loop()" : "parallelize_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, static_cast<int64_t>(random_end));
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)
{
@@ -580,7 +749,7 @@ void check_parallelize_loop_return(const int64_t random_start, int64_t random_en
/**
* @brief Check that push_loop() and parallelize_loop() work using several different random values for the range of indices and number of tasks.
*/
void check_parallelize_loop()
void check_push_loop()
{
std::mt19937_64 mt(rd());
std::uniform_int_distribution<int64_t> index_dist(-1000000, 1000000);
@@ -850,9 +1019,12 @@ void do_tests()
print_header("Checking that wait_for_tasks() works...");
check_wait_for_tasks();
check_wait_for_tasks_deadlock();
check_wait_for_tasks_duration();
check_wait_for_tasks_until();
print_header("Checking that push_loop() and parallelize_loop() work:");
check_parallelize_loop();
check_push_loop();
print_header("Checking that task monitoring works:");
check_task_monitoring();
@@ -961,8 +1133,8 @@ void check_performance()
// Test how many vectors we need to generate, and of what size, to roughly achieve the target execution time.
dual_println("Determining the number and size of vectors to generate in order to achieve an approximate mean execution time of ", target_ms, " ms with ", thread_count, " tasks...");
size_t num_vectors = 64;
size_t vector_size = 64;
size_t num_vectors = thread_count;
size_t vector_size = thread_count;
std::vector<std::vector<double>> vectors;
auto loop = [&vectors, &vector_size](const size_t start, const size_t end)
{
@@ -1029,7 +1201,7 @@ int main()
log_file.open(log_filename);
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("(c) 2023 Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)");
dual_println("GitHub: https://github.com/bshoshany/thread-pool\n");
dual_println("Thread pool library version is ", BS_THREAD_POOL_VERSION, ".");
@@ -1042,6 +1214,13 @@ int main()
if (enable_tests)
do_tests();
if (enable_long_deadlock_tests)
{
print_header("Checking for deadlocks:");
check_destructor_deadlock();
check_reset_deadlock();
}
if (tests_failed == 0)
{
if (enable_tests)
@@ -1052,12 +1231,12 @@ int main()
check_performance();
print_header("Thread pool performance test completed!", '+');
}
return EXIT_SUCCESS;
return 0;
}
else
{
print_header("FAILURE: Passed " + std::to_string(tests_succeeded) + " checks, but failed " + std::to_string(tests_failed) + "!", '+');
dual_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.");
return EXIT_FAILURE;
std::quick_exit(static_cast<int>(tests_failed));
}
}
CHANGELOG.md
+27 -4
View File
@@ -10,12 +10,13 @@
# `BS::thread_pool`: a fast, lightweight, and easy-to-use C++17 thread pool library
By Barak Shoshany<br />
Email: [baraksh@gmail.com](mailto:baraksh@gmail.com)<br />
Website: [https://baraksh.com/](https://baraksh.com/)<br />
GitHub: [https://github.com/bshoshany](https://github.com/bshoshany)<br />
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)
@@ -34,6 +35,28 @@ GitHub: [https://github.com/bshoshany](https://github.com/bshoshany)<br />
## 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`:
LICENSE.txt
+1 -1
View File
@@ -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
README.md
+187 -123
View File
@@ -10,12 +10,12 @@
# `BS::thread_pool`: a fast, lightweight, and easy-to-use C++17 thread pool library
By Barak Shoshany<br />
Email: [baraksh@gmail.com](mailto:baraksh@gmail.com)<br />
Website: [https://baraksh.com/](https://baraksh.com/)<br />
GitHub: [https://github.com/bshoshany](https://github.com/bshoshany)<br />
By Barak Shoshany\
Email: <baraksh@gmail.com>\
Website: <https://baraksh.com/>\
GitHub: <https://github.com/bshoshany>
This is the complete documentation for v3.3.0 of the library, released on 2022-08-03.
This is the complete documentation for v3.4.0 of the library, released on 2023-05-12.
* [Introduction](#introduction)
* [Motivation](#motivation)
@@ -32,8 +32,10 @@ This is the complete documentation for v3.3.0 of the library, released on 2022-0
* [Submitting tasks to the queue with futures](#submitting-tasks-to-the-queue-with-futures)
* [Submitting tasks to the queue without futures](#submitting-tasks-to-the-queue-without-futures)
* [Manually waiting for all tasks to complete](#manually-waiting-for-all-tasks-to-complete)
* [Waiting with a timeout](#waiting-with-a-timeout)
* [Submitting class member functions to the queue](#submitting-class-member-functions-to-the-queue)
* [Parallelizing loops](#parallelizing-loops)
* [Parallelizing loops](#parallelizing-loops)
* [Automatic parallelization of loops](#automatic-parallelization-of-loops)
* [Loops with return values](#loops-with-return-values)
* [Parallelizing loops without futures](#parallelizing-loops-without-futures)
* [Helper classes](#helper-classes)
@@ -58,13 +60,13 @@ This is the complete documentation for v3.3.0 of the library, released on 2022-0
### Motivation
Multithreading is essential for modern high-performance computing. Since C++11, the C++ standard library has included built-in low-level multithreading support using constructs such as `std::thread`. However, `std::thread` creates a new thread each time it is called, which can have a significant performance overhead. Furthermore, it is possible to create more threads than the hardware can handle simultaneously, potentially resulting in a substantial slowdown.
Multithreading is essential for modern high-performance computing. Since C\+\+11, the C++ standard library has included built-in low-level multithreading support using constructs such as `std::thread`. However, `std::thread` creates a new thread each time it is called, which can have a significant performance overhead. Furthermore, it is possible to create more threads than the hardware can handle simultaneously, potentially resulting in a substantial slowdown.
The library presented here contains a thread pool class, `BS::thread_pool`, which avoids these issues by creating a fixed pool of threads once and for all, and then continuously reusing the same threads to perform different tasks throughout the lifetime of the program. By default, the number of threads in the pool is equal to the maximum number of threads that the hardware can run in parallel.
The library presented here contains a C++ thread pool class, `BS::thread_pool`, which avoids these issues by creating a fixed pool of threads once and for all, and then continuously reusing the same threads to perform different tasks throughout the lifetime of the program. By default, the number of threads in the pool is equal to the maximum number of threads that the hardware can run in parallel.
The user submits tasks to be executed into a queue. Whenever a thread becomes available, it retrieves the next task from the queue and executes it. The pool automatically produces an `std::future` for each task, which allows the user to wait for the task to finish executing and/or obtain its eventual return value, if applicable. Threads and tasks are autonomously managed by the pool in the background, without requiring any input from the user aside from submitting the desired tasks.
The design of this package was guided by four important principles. First, *compactness*: the entire library consists of just one small self-contained header file, with no other components or dependencies. Second, *portability*: the package only utilizes the C++17 standard library, without relying on any compiler extensions or 3rd-party libraries, and is therefore compatible with any modern standards-conforming C++17 compiler on any platform. Third, *ease of use*: the package is extensively documented, and programmers of any level should be able to use it right out of the box.
The design of this package was guided by four important principles. First, *compactness*: the entire library consists of just one small self-contained header file, with no other components or dependencies. Second, *portability*: the package only utilizes the C\+\+17 standard library, without relying on any compiler extensions or 3rd-party libraries, and is therefore compatible with any modern standards-conforming C\+\+17 compiler on any platform. Third, *ease of use*: the package is extensively documented, and programmers of any level should be able to use it right out of the box.
The fourth and final guiding principle is *performance*: each and every line of code in this library was carefully designed with maximum performance in mind, and performance was tested and verified on a variety of compilers and platforms. Indeed, the library was originally designed for use in the author's own computationally-intensive scientific computing projects, running both on high-end desktop/laptop computers and high-performance computing nodes.
@@ -84,7 +86,7 @@ Other, more advanced multithreading libraries may offer more features and/or hig
* Self-contained: no external requirements or dependencies.
* Portable: uses only the C++ standard library, and works with any C++17-compliant compiler.
* Only ~340 lines of code, excluding comments and blank lines.
* A stand-alone "light version" of the thread pool is also available in the `BS_thread_pool_light.hpp` header file, with only ~170 lines of code.
* A stand-alone "light version" of the C++ thread pool is also available in the `BS_thread_pool_light.hpp` header file, with only ~170 lines of code.
* **Easy to use:**
* Very simple operation, using a handful of member functions.
* Every task submitted to the queue using the `submit()` member function automatically generates an `std::future`, which can be used to wait for the task to finish executing and/or obtain its eventual return value.
@@ -106,21 +108,21 @@ Other, more advanced multithreading libraries may offer more features and/or hig
### Compiling and compatibility
This library should successfully compile on any C++17 standard-compliant compiler, on all operating systems and architectures for which such a compiler is available. Compatibility was verified with a 12-core / 24-thread AMD Ryzen 9 3900X CPU using the following compilers and platforms:
This library should successfully compile on any C++17 standard-compliant compiler, on all operating systems and architectures for which such a compiler is available. Compatibility was verified with a 24-core (8P+16E) / 32-thread Intel i9-13900K CPU using the following compilers and platforms:
* Windows 11 build 22000.795:
* [Clang](https://clang.llvm.org/) v14.0.6
* [GCC](https://gcc.gnu.org/) v12.1.0 ([WinLibs build](https://winlibs.com/))
* [MSVC](https://docs.microsoft.com/en-us/cpp/) v19.32.31332
* Ubuntu 22.04 LTS:
* [Clang](https://clang.llvm.org/) v14.0.0
* [GCC](https://gcc.gnu.org/) v12.0.1
* Windows 11 build 22621.1702:
* [Clang](https://clang.llvm.org/) v16.0.3
* [GCC](https://gcc.gnu.org/) v13.1.0 ([WinLibs build](https://winlibs.com/))
* [MSVC](https://docs.microsoft.com/en-us/cpp/) v19.35.32217.1
* Ubuntu 22.10:
* [Clang](https://clang.llvm.org/) v15.0.7
* [GCC](https://gcc.gnu.org/) v12.2.0
In addition, this library was tested on a [Compute Canada](https://www.computecanada.ca/) node equipped with two 20-core / 40-thread Intel Xeon Gold 6148 CPUs (for a total of 40 cores and 80 threads), running CentOS Linux 7.9.2009, using [GCC](https://gcc.gnu.org/) v12.1.1.
In addition, this library was tested on a [Digital Research Alliance of Canada](https://alliancecan.ca/en) node equipped with two 20-core / 40-thread Intel Xeon Gold 6148 CPUs (for a total of 40 cores and 80 threads), running CentOS Linux 7.9.2009, using [GCC](https://gcc.gnu.org/) v12.2.0.
The test program `BS_thread_pool_test.cpp` was compiled without warnings (with the warning flags `-Wall -Wextra -Wconversion -Wsign-conversion -Wpedantic -Weffc++ -Wshadow` in GCC/Clang and `/W4` in MSVC), executed, and successfully completed all [automated tests](#testing-the-package) and benchmarks using all of the compilers and systems mentioned above.
As this library requires C++17 features, the code must be compiled with C++17 support:
As this library requires C\+\+17 features, the code must be compiled with C\+\+17 support:
* For Clang or GCC, use the `-std=c++17` flag. On Linux, you will also need to use the `-pthread` flag to enable the POSIX threads library.
* For MSVC, use `/std:c++17`, and preferably also `/permissive-` to ensure standards conformance.
@@ -153,7 +155,7 @@ On Windows:
.\vcpkg install bshoshany-thread-pool:x86-windows bshoshany-thread-pool:x64-windows
```
The thread pool will then be available automatically in the build system you integrated vcpkg with (e.g. MSBuild or CMake). Simply write `#include "BS_thread_pool.hpp"` in any project to use the thread pool, without having to copy to file into the project first. I will update the vcpkg port with each new release, so it will be updated automatically when you run `vcpkg upgrade`.
The C++ thread pool will then be available automatically in the build system you integrated vcpkg with (e.g. MSBuild or CMake). Simply write `#include "BS_thread_pool.hpp"` in any project to use the thread pool, without having to copy to file into the project first. I will update the vcpkg port with each new release, so it will be updated automatically when you run `vcpkg upgrade`.
Please see the [vcpkg repository](https://github.com/microsoft/vcpkg) for more information on how to use vcpkg.
@@ -212,7 +214,7 @@ std::cout << "Thread pool library version is " << BS_THREAD_POOL_VERSION << ".\n
Sample output:
```none
Thread pool library version is v3.1.0 (2022-07-13).
Thread pool library version is v3.4.0 (2023-05-12).
```
This can be used, for example, to allow the same code to work with several incompatible versions of the library.
@@ -371,6 +373,51 @@ after the `for` loop will ensure - as efficiently as possible - that all tasks h
Note, however, that `wait_for_tasks()` will wait for **all** the tasks in the queue, including those that are unrelated to the `for` loop. Using [`parallelize_loop()`](#parallelizing-loops) would make much more sense in this particular case, as it will allow waiting only for the tasks related to the loop.
### Waiting with a timeout
Sometimes you may wish to wait for the tasks to complete, but only for a certain amount of time, or until a specific point in time. For example, if the tasks have not yet completed after some time, you may wish to let the user know that there is a delay. This can be achieved using two member functions:
* `wait_for_tasks_duration()` waits for the tasks to be completed, but stops waiting after the specified duration, given as an argument of type `std::chrono::duration`, has passed.
* `wait_for_tasks_until()` waits for the tasks to be completed, but stops waiting after the specified time point, given as an argument of type `std::chrono::time_point`, has been reached.
Here is an example:
```cpp
#include "BS_thread_pool.hpp"
int main()
{
BS::synced_stream sync_out;
BS::thread_pool pool;
std::atomic<bool> done = false;
pool.push_task(
[&done]
{
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
done = true;
});
while (true)
{
pool.wait_for_tasks_duration(std::chrono::milliseconds(200));
if (!done)
sync_out.println("Sorry, task is not done yet.");
else
break;
}
sync_out.println("Task done!");
}
```
The output is:
```none
Sorry, task is not done yet.
Sorry, task is not done yet.
Sorry, task is not done yet.
Sorry, task is not done yet.
Task done!
```
### Submitting class member functions to the queue
Consider the following program:
@@ -469,7 +516,9 @@ int main()
}
```
### Parallelizing loops
## Parallelizing loops
### Automatic parallelization of loops
One of the most common and effective methods of parallelization is splitting a loop into smaller loops and running them in parallel. It is most effective in "embarrassingly parallel" computations, such as vector or matrix operations, where each iteration of the loop is completely independent of every other iteration. For example, if we are summing up two vectors of 1000 elements each, and we have 10 threads, we could split the summation into 10 blocks of 100 elements each, and run all the blocks in parallel, potentially increasing performance by up to a factor of 10.
@@ -1125,12 +1174,12 @@ A sample output of a successful run of the automated tests is as follows:
```none
BS::thread_pool: a fast, lightweight, and easy-to-use C++17 thread pool library
(c) 2022 Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
(c) 2023 Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
GitHub: https://github.com/bshoshany/thread-pool
Thread pool library version is v3.3.0 (2022-08-03).
Hardware concurrency is 24.
Generating log file: BS_thread_pool_test-2022-08-03_12.32.04.log.
Thread pool library version is v3.4.0 (2023-05-12).
Hardware concurrency is 32.
Generating log file: BS_thread_pool_test-2023-05-12_12.48.13.log.
Important: Please do not run any other applications, especially multithreaded applications, in parallel with this test!
@@ -1138,21 +1187,21 @@ Important: Please do not run any other applications, especially multithreaded ap
Checking that the constructor works:
====================================
Checking that the thread pool reports a number of threads equal to the hardware concurrency...
Expected: 24, obtained: 24 -> PASSED!
Expected: 32, obtained: 32 -> PASSED!
Checking that the manually counted number of unique thread IDs is equal to the reported number of threads...
Expected: 24, obtained: 24 -> PASSED!
Expected: 32, obtained: 32 -> PASSED!
============================
Checking that reset() works:
============================
Checking that after reset() the thread pool reports a number of threads equal to half the hardware concurrency...
Expected: 12, obtained: 12 -> PASSED!
Expected: 16, obtained: 16 -> PASSED!
Checking that after reset() the manually counted number of unique thread IDs is equal to the reported number of threads...
Expected: 12, obtained: 12 -> PASSED!
Expected: 16, obtained: 16 -> PASSED!
Checking that after a second reset() the thread pool reports a number of threads equal to the hardware concurrency...
Expected: 24, obtained: 24 -> PASSED!
Expected: 32, obtained: 32 -> PASSED!
Checking that after a second reset() the manually counted number of unique thread IDs is equal to the reported number of threads...
Expected: 24, obtained: 24 -> PASSED!
Expected: 32, obtained: 32 -> PASSED!
================================
Checking that push_task() works:
@@ -1217,82 +1266,95 @@ Checking that wait_for_tasks() works...
=======================================
Waiting for tasks...
-> PASSED!
Checking for deadlocks when waiting for tasks...
All waiting tasks successfully finished!
-> PASSED!
Checking that wait_for_tasks_duration() works...
Task submitted. Waiting for 10ms...
-> PASSED!
Waiting for 500ms...
-> PASSED!
Checking that wait_for_tasks_until() works...
Task submitted. Waiting until 10ms from submission time...
-> PASSED!
Waiting until 500ms from submission time...
-> PASSED!
======================================================
Checking that push_loop() and parallelize_loop() work:
======================================================
Verifying that push_loop() from 917499 to 884861 with 19 tasks modifies all indices...
Verifying that push_loop() from 117855 to 168463 with 25 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from 235488 to 296304 with 11 tasks modifies all indices...
Verifying that push_loop() from -788069 to -860364 with 21 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from -790296 to -152228 with 21 tasks modifies all indices...
Verifying that push_loop() from 545486 to 553538 with 15 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from -937055 to -135942 with 10 tasks modifies all indices...
Verifying that push_loop() from 987439 to 166022 with 29 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from 372276 to 486867 with 3 tasks modifies all indices...
Verifying that push_loop() from 843125 to 395220 with 19 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from 890415 to -163491 with 5 tasks modifies all indices...
Verifying that push_loop() from 75069 to 552964 with 3 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from 637645 to -894687 with 7 tasks modifies all indices...
Verifying that push_loop() from 986466 to -642521 with 20 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from 308032 to -254915 with 20 tasks modifies all indices...
Verifying that push_loop() from 994906 to -386703 with 7 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from 499518 to 104936 with 17 tasks modifies all indices...
Verifying that push_loop() from -574578 to 327232 with 22 tasks modifies all indices...
-> PASSED!
Verifying that push_loop() from 19080 to -378567 with 5 tasks modifies all indices...
Verifying that push_loop() from 632264 to 644863 with 12 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from -298981 to -724834 with 11 tasks modifies all indices...
Verifying that parallelize_loop() from 450897 to -789636 with 16 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from 232695 to 767243 with 3 tasks modifies all indices...
Verifying that parallelize_loop() from -986029 to -900579 with 24 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from 177768 to 966097 with 10 tasks modifies all indices...
Verifying that parallelize_loop() from -405930 to 299022 with 24 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from 474617 to -155690 with 15 tasks modifies all indices...
Verifying that parallelize_loop() from -545956 to 219212 with 13 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from -733576 to 547977 with 9 tasks modifies all indices...
Verifying that parallelize_loop() from 462224 to 865745 with 26 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from -723922 to 992233 with 1 task modifies all indices...
Verifying that parallelize_loop() from -311718 to 644762 with 28 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from 957397 to 364478 with 5 tasks modifies all indices...
Verifying that parallelize_loop() from -615396 to -267130 with 2 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from 776948 to 895847 with 3 tasks modifies all indices...
Verifying that parallelize_loop() from -510089 to 363393 with 26 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from 696779 to 400637 with 17 tasks modifies all indices...
Verifying that parallelize_loop() from -846318 to -18573 with 11 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from -5265 to 746418 with 23 tasks modifies all indices...
Verifying that parallelize_loop() from -680422 to -342474 with 23 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from -229724 to -883103 with 9 tasks correctly sums all indices...
Expected: -727098445812, obtained: -727098445812 -> PASSED!
Verifying that parallelize_loop() from 730130 to -370499 with 3 tasks correctly sums all indices...
Expected: 395819207270, obtained: 395819207270 -> PASSED!
Verifying that parallelize_loop() from -493633 to 957239 with 14 tasks correctly sums all indices...
Expected: 672631513560, obtained: 672631513560 -> PASSED!
Verifying that parallelize_loop() from -455240 to -60850 with 14 tasks correctly sums all indices...
Expected: -203541129490, obtained: -203541129490 -> PASSED!
Verifying that parallelize_loop() from -287333 to 298991 with 9 tasks correctly sums all indices...
Expected: 6834778868, obtained: 6834778868 -> PASSED!
Verifying that parallelize_loop() from 62326 to -392718 with 18 tasks correctly sums all indices...
Expected: -150343352292, obtained: -150343352292 -> PASSED!
Verifying that parallelize_loop() from 663186 to 380865 with 23 tasks correctly sums all indices...
Expected: 294757240050, obtained: 294757240050 -> PASSED!
Verifying that parallelize_loop() from 609125 to -43020 with 1 task correctly sums all indices...
Expected: 369181893080, obtained: 369181893080 -> PASSED!
Verifying that parallelize_loop() from 465469 to 112037 with 4 tasks correctly sums all indices...
Expected: 204108747160, obtained: 204108747160 -> PASSED!
Verifying that parallelize_loop() from 690574 to 113023 with 15 tasks correctly sums all indices...
Expected: 464117673396, obtained: 464117673396 -> PASSED!
Verifying that parallelize_loop() from 383147 to -130987 with 10 tasks correctly sums all indices...
Expected: 129643515306, obtained: 129643515306 -> PASSED!
Verifying that parallelize_loop() from 827793 to 219417 with 18 tasks correctly sums all indices...
Expected: 637096822584, obtained: 637096822584 -> PASSED!
Verifying that parallelize_loop() from -441740 to -531834 with 12 tasks correctly sums all indices...
Expected: -87713266050, obtained: -87713266050 -> PASSED!
Verifying that parallelize_loop() from 980942 to 18492 with 2 tasks correctly sums all indices...
Expected: 961904290850, obtained: 961904290850 -> PASSED!
Verifying that parallelize_loop() from 351318 to 433401 with 8 tasks correctly sums all indices...
Expected: 64412007594, obtained: 64412007594 -> PASSED!
Verifying that parallelize_loop() from 184385 to 382813 with 30 tasks correctly sums all indices...
Expected: 112547766316, obtained: 112547766316 -> PASSED!
Verifying that parallelize_loop() from -498480 to 323830 with 10 tasks correctly sums all indices...
Expected: -143617263810, obtained: -143617263810 -> PASSED!
Verifying that parallelize_loop() from -119493 to 109088 with 25 tasks correctly sums all indices...
Expected: -2378613886, obtained: -2378613886 -> PASSED!
Verifying that parallelize_loop() from 776258 to 340877 with 6 tasks correctly sums all indices...
Expected: 486378918054, obtained: 486378918054 -> PASSED!
Verifying that parallelize_loop() from 160863 to 750589 with 3 tasks correctly sums all indices...
Expected: 537506352426, obtained: 537506352426 -> PASSED!
Verifying that parallelize_loop() with identical start and end indices does nothing...
-> PASSED!
Trying parallelize_loop() with start and end indices of different types:
Verifying that parallelize_loop() from 894645 to 908567 with 9 tasks modifies all indices...
Verifying that parallelize_loop() from 838162 to 345683 with 13 tasks modifies all indices...
-> PASSED!
Trying the overloads for push_loop() and parallelize_loop() for the case where the first index is equal to 0:
Verifying that push_loop() from 0 to 949967 with 10 tasks modifies all indices...
Verifying that push_loop() from 0 to 15216 with 30 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from 0 to 241018 with 3 tasks modifies all indices...
Verifying that parallelize_loop() from 0 to 239084 with 5 tasks modifies all indices...
-> PASSED!
Verifying that parallelize_loop() from 0 to 574984 with 19 tasks correctly sums all indices...
Expected: 330606025272, obtained: 330606025272 -> PASSED!
Verifying that parallelize_loop() from 0 to -824186 with 18 tasks correctly sums all indices...
Expected: -679283386782, obtained: -679283386782 -> PASSED!
====================================
Checking that task monitoring works:
@@ -1301,25 +1363,25 @@ Resetting pool to 4 threads.
Submitting 12 tasks.
After submission, should have: 12 tasks total, 4 tasks running, 8 tasks queued...
Result: 12 tasks total, 4 tasks running, 8 tasks queued -> PASSED!
Task 3 released.
Task 1 released.
Task 2 released.
Task 0 released.
Task 1 released.
Task 3 released.
After releasing 4 tasks, should have: 8 tasks total, 4 tasks running, 4 tasks queued...
Result: 8 tasks total, 4 tasks running, 4 tasks queued -> PASSED!
Task 4 released.
Task 7 released.
Task 6 released.
Task 5 released.
Task 7 released.
After releasing 4 more tasks, should have: 4 tasks total, 4 tasks running, 0 tasks queued...
Result: 4 tasks total, 4 tasks running, 0 tasks queued -> PASSED!
Task 11 released.
Task 10 released.
Task 9 released.
Task 8 released.
Task 11 released.
Task 9 released.
Task 10 released.
After releasing the final 4 tasks, should have: 0 tasks total, 0 tasks running, 0 tasks queued...
Result: 0 tasks total, 0 tasks running, 0 tasks queued -> PASSED!
Resetting pool to 24 threads.
Resetting pool to 32 threads.
============================
Checking that pausing works:
@@ -1338,29 +1400,29 @@ Result: 12 tasks total, 0 tasks running, 12 tasks queued -> PASSED!
Unpausing pool.
Checking that the pool correctly reports that it is not paused.
-> PASSED!
Task 1 done.
Task 2 done.
Task 0 done.
Task 3 done.
Task 2 done.
Task 1 done.
Task 0 done.
300ms later, should have: 8 tasks total, 4 tasks running, 4 tasks queued...
Result: 8 tasks total, 4 tasks running, 4 tasks queued -> PASSED!
Pausing pool and using wait_for_tasks() to wait for the running tasks.
Task 7 done.
Task 5 done.
Task 6 done.
Task 4 done.
Task 7 done.
Task 5 done.
After waiting, should have: 4 tasks total, 0 tasks running, 4 tasks queued...
Result: 4 tasks total, 0 tasks running, 4 tasks queued -> PASSED!
200ms later, should still have: 4 tasks total, 0 tasks running, 4 tasks queued...
Result: 4 tasks total, 0 tasks running, 4 tasks queued -> PASSED!
Unpausing pool and using wait_for_tasks() to wait for all tasks.
Task 8 done.
Task 11 done.
Task 9 done.
Task 10 done.
Task 11 done.
Task 8 done.
After waiting, should have: 0 tasks total, 0 tasks running, 0 tasks queued...
Result: 0 tasks total, 0 tasks running, 0 tasks queued -> PASSED!
Resetting pool to 24 threads.
Resetting pool to 32 threads.
=======================================
Checking that exception handling works:
@@ -1376,68 +1438,68 @@ Throwing exception...
============================================================
Testing that vector operations produce the expected results:
============================================================
Adding two vectors with 77579 elements using 9 tasks...
Adding two vectors with 365392 elements using 9 tasks...
-> PASSED!
Adding two vectors with 925926 elements using 2 tasks...
Adding two vectors with 797060 elements using 7 tasks...
-> PASSED!
Adding two vectors with 367682 elements using 22 tasks...
Adding two vectors with 159148 elements using 19 tasks...
-> PASSED!
Adding two vectors with 28482 elements using 2 tasks...
Adding two vectors with 432461 elements using 3 tasks...
-> PASSED!
Adding two vectors with 486607 elements using 19 tasks...
Adding two vectors with 907909 elements using 30 tasks...
-> PASSED!
Adding two vectors with 688249 elements using 10 tasks...
Adding two vectors with 854259 elements using 3 tasks...
-> PASSED!
Adding two vectors with 473738 elements using 18 tasks...
Adding two vectors with 238088 elements using 2 tasks...
-> PASSED!
Adding two vectors with 743021 elements using 12 tasks...
Adding two vectors with 559647 elements using 32 tasks...
-> PASSED!
Adding two vectors with 209804 elements using 11 tasks...
Adding two vectors with 473570 elements using 25 tasks...
-> PASSED!
Adding two vectors with 635671 elements using 23 tasks...
Adding two vectors with 124722 elements using 9 tasks...
-> PASSED!
++++++++++++++++++++++++++++++
SUCCESS: Passed all 88 checks!
SUCCESS: Passed all 93 checks!
++++++++++++++++++++++++++++++
```
### Performance tests
If all checks passed, `BS_thread_pool_test.cpp` will perform simple benchmarks by filling a specific number of vectors of fixed size with random values. The program decides how many vectors to use by testing how many are needed to reach a target duration in the single-threaded test. This ensures that the test takes approximately the same amount of time on different systems, and is thus more consistent and portable.
If all checks passed, `BS_thread_pool_test.cpp` will perform simple benchmarks by filling a specific number of vectors of fixed size with values. The program decides how many vectors to use by testing how many are needed to reach a target duration in the single-threaded test. This ensures that the test takes approximately the same amount of time on different systems, and is thus more consistent and portable.
Once the required number of vectors has been determined, the program will test the performance of several multi-threaded tests, dividing the total number of vectors into different numbers of tasks, compare them to the performance of the single-threaded test, and indicate the maximum speedup obtained.
Please note that these benchmarks are only intended to demonstrate that the package can provide a significant speedup, and it is highly recommended to perform your own benchmarks with your specific system, compiler, and code.
Here we will present the results of the performance test running on a high-end desktop computer equipped with a 12-core / 24-thread AMD Ryzen 9 3900X CPU at 3.8 GHz and 32 GB of DDR4 RAM at 3600 MHz, compiled using [MSVC](https://docs.microsoft.com/en-us/cpp/) v19.32.31332 on Windows 11 build 22000.795 with the `/O2` compiler flag. The output was as follows:
As an example, here are the results of the benchmarks from a [Digital Research Alliance of Canada](https://alliancecan.ca/en) node equipped with two 20-core / 40-thread Intel Xeon Gold 6148 CPUs (for a total of 40 cores and 80 threads), running CentOS Linux 7.9.2009. The tests were compiled using GCC v12.2.0 with the `-O3` and `-march=native` flags. The output was as follows:
```none
======================
Performing benchmarks:
======================
Using 24 threads.
Using 80 threads.
Each test will be repeated 20 times to collect reliable statistics.
Determining the number and size of vectors to generate in order to achieve an approximate mean execution time of 50 ms with 24 tasks...
Generating 3312 vectors with 4096 elements each:
Single-threaded, mean execution time was 542.2 ms with standard deviation 5.8 ms.
With 6 tasks, mean execution time was 95.2 ms with standard deviation 1.7 ms.
With 12 tasks, mean execution time was 49.6 ms with standard deviation 0.7 ms.
With 24 tasks, mean execution time was 29.0 ms with standard deviation 2.9 ms.
With 48 tasks, mean execution time was 33.2 ms with standard deviation 4.3 ms.
With 96 tasks, mean execution time was 35.5 ms with standard deviation 1.9 ms.
Maximum speedup obtained by multithreading vs. single-threading: 18.7x, using 24 tasks.
Determining the number and size of vectors to generate in order to achieve an approximate mean execution time of 50 ms with 80 tasks...
Generating 4000 vectors with 5120 elements each:
Single-threaded, mean execution time was 2211.9 ms with standard deviation 39.1 ms.
With 20 tasks, mean execution time was 128.8 ms with standard deviation 12.8 ms.
With 40 tasks, mean execution time was 71.6 ms with standard deviation 1.1 ms.
With 80 tasks, mean execution time was 44.4 ms with standard deviation 5.0 ms.
With 160 tasks, mean execution time was 47.0 ms with standard deviation 6.4 ms.
With 320 tasks, mean execution time was 132.6 ms with standard deviation 2.2 ms.
Maximum speedup obtained by multithreading vs. single-threading: 49.8x, using 80 tasks.
+++++++++++++++++++++++++++++++++++++++
Thread pool performance test completed!
+++++++++++++++++++++++++++++++++++++++
```
This CPU has 12 physical cores, with each core providing two separate logical cores via hyperthreading, for a total of 24 threads. Without hyperthreading, we would expect a maximum theoretical speedup of 12x. With hyperthreading, one might naively expect to achieve up to a 24x speedup, but this is in fact impossible, as both logical cores share the same physical core's resources. However, generally we would expect [an estimated 30% additional speedup](https://software.intel.com/content/www/us/en/develop/articles/how-to-determine-the-effectiveness-of-hyper-threading-technology-with-an-application.html) from hyperthreading, which amounts to around 15.6x in this case. In our performance test, we see a speedup of 18.7x, saturating and even surpassing this estimated theoretical upper bound.
These two CPUs have 40 physical cores in total, with each core providing two separate logical cores via hyperthreading, for a total of 80 threads. Without hyperthreading, we would expect a maximum theoretical speedup of 40x. With hyperthreading, one might naively expect to achieve up to an 80x speedup, but this is in fact impossible, as each pair of hyperthreaded logical cores share the same physical core's resources. However, generally we would expect at most an estimated 30% additional speedup from hyperthreading, which amounts to around 52x in this case. The speedup of 49.8x in our performance test is very close to this estimate.
In addition, this test demonstrates that splitting the loop into a number of tasks to be equal to the number of hardware threads usually yields optimal results, since all the tasks can be run in parallel. When we used less than 80 tasks, not all of the CPU cores were taken advantage of, and when we used more than 80 tasks, they had to run in more than one batch instead of all at once, introducing additional overhead.
## The light version of the package
This package started out as a very lightweight thread pool, but over time has expanded to include many additional features and 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`, which is about half the size of the full package.
This package started out as a very lightweight C++ thread pool, but over time has expanded to include many additional features and 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`, which is about 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:
@@ -1449,7 +1511,9 @@ This file does not contain any of the helper classes, only a new `BS::thread_poo
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.
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. However, if you wish to use both the light and non-light thread pool classes in the same project, you can include both header files.
If needed, the current version of the light thread pool can be obtained using the macro `BS_THREAD_POOL_LIGHT_VERSION`.
## About the project
@@ -1469,9 +1533,9 @@ If you found this project useful, please consider [starring it on GitHub](https:
### Copyright and citing
Copyright (c) 2022 [Barak Shoshany](http://baraksh.com). Licensed under the [MIT license](LICENSE.txt).
Copyright (c) 2023 [Barak Shoshany](http://baraksh.com). Licensed under the [MIT license](LICENSE.txt).
If you use the 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 C++ thread pool 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: