diff --git a/CHANGELOG.md b/CHANGELOG.md index c128b1b..c6bfaf4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## Version history +* v2.0.0 (2021-08-14) + * From now on, version numbers will adhere to the [Semantic Versioning](https://semver.org/) specification in the format **major.minor.patch**. + * A file named `thread_pool_test.cpp` has been added to the package. It will perform automated tests of all aspects of the package, and benchmark some multithreaded matrix operations. Please run it on your system and [submit a bug report](https://github.com/bshoshany/thread-pool/issues) if any of the tests fail. In addition, the code is thoroughly documented, and is meant to serve as an extensive example of how to properly use the package. + * The package is now available through [vcpkg](https://github.com/microsoft/vcpkg). Instructions for how to install it have been added to `README.md`. See [this pull request](https://github.com/bshoshany/thread-pool/pull/18). + * The package now defines a macro `THREAD_POOL_VERSION`, which returns the version number and release date of the thread pool library as a string. + * `parallelize_loop()` has undergone some major changes (and is now incompatible with v1.x): + * The second argument is now the index **after** the last index, instead of the last index itself. This is more consistent with C++ conventions (e.g. standard library algorithms) where the range is always `[first, last)`. For example, for an array with `n` indices, instead of `parallelize_loop(0, n - 1, ...)` you should now write `parallelize_loop(0, n, ...)`. + * The `loop` function is now only called once per block, instead of once per index, as was the case before. This should provide a performance boost due to significantly reducing the number of function calls, and it also allows you to conserve resources by using them only once per block instead of once per index (an example can be found in the `random_matrix_generator` class in `thread_pool_test.cpp`). It also means that `loop` now takes two arguments: the first index in the block and the index after the last index in the block. Thus, `loop(start, end)` should typically involve a loop of the form `for (T i = start; i < end; i++)`. + * The first and last indices can now be of two different integer types. Previously, `parallelize_loop(0, i, ...)` did not work if `i` was not an `int`, because `0` was interpreted as `int`, and the two arguments had to be of the same type. Therefore, one had to use casting, e.g. `parallelize_loop((size_t)0, i)`, to make it work. Now this is no longer necessary; the common type is inferred automatically using `std::common_type_t`. * v1.9 (2021-07-29) * Fixed a bug in `reset()` which caused it to create the wrong number of threads. * v1.8 (2021-07-28) diff --git a/LICENSE.txt b/LICENSE.txt index 118ef11..a8684f1 100644 --- a/LICENSE.txt +++ b/LICENSE.txt @@ -18,4 +18,4 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. +SOFTWARE. diff --git a/README.md b/README.md index c8920d6..ed1f561 100644 --- a/README.md +++ b/README.md @@ -24,24 +24,28 @@ DOI: [doi:10.5281/zenodo.4742687](https://doi.org/10.5281/zenodo.4742687) * [Compiling and compatibility](#compiling-and-compatibility) * [Getting started](#getting-started) * [Including the library](#including-the-library) + * [Installing using vcpkg](#installing-using-vcpkg) * [Constructors](#constructors) * [Getting and resetting the number of threads in the pool](#getting-and-resetting-the-number-of-threads-in-the-pool) + * [Finding the version of the package](#finding-the-version-of-the-package) * [Submitting and waiting for tasks](#submitting-and-waiting-for-tasks) * [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) * [Parallelizing loops](#parallelizing-loops) -* [Other features](#other-features) +* [Helper classes](#helper-classes) * [Synchronizing printing to an output stream](#synchronizing-printing-to-an-output-stream) + * [Measuring execution time](#measuring-execution-time) +* [Other features](#other-features) * [Setting the worker function's sleep duration](#setting-the-worker-functions-sleep-duration) * [Monitoring the tasks](#monitoring-the-tasks) * [Pausing the workers](#pausing-the-workers) -* [Exception handling](#exception-handling) -* [Performance tests](#performance-tests) - * [Measuring execution time](#measuring-execution-time) - * [AMD Ryzen 9 3900X (24 threads)](#amd-ryzen-9-3900x-24-threads) + * [Exception handling](#exception-handling) +* [Testing the package](#testing-the-package) + * [Automated tests](#automated-tests) + * [Performance tests](#performance-tests) * [Dual Intel Xeon Gold 6148 (80 threads)](#dual-intel-xeon-gold-6148-80-threads) -* [Feedback](#feedback) +* [Issue and pull request policy](#issue-and-pull-request-policy) * [Copyright and citing](#copyright-and-citing) ## Abstract @@ -82,6 +86,7 @@ As demonstrated in the performance tests [below](#performance-tests), using our * Every task submitted to the queue automatically generates an `std::future`, which can be used to wait for the task to finish executing and/or obtain its eventual return value. * Optionally, tasks may also be submitted without generating a future, sacrificing convenience for greater performance. * The code is thoroughly documented using Doxygen comments - not only the interface, but also the implementation, in case the user would like to make modifications. + * The included test program `thread_pool_test.cpp` can be used to perform comprehensive automated tests and benchmarks, and also serves as an extensive example of how to properly use the package. * **Additional features:** * Automatically parallelize a loop into any number of parallel tasks. * Easily wait for all tasks in the queue to complete. @@ -92,28 +97,42 @@ As demonstrated in the performance tests [below](#performance-tests), using our * Catch exceptions thrown by the submitted tasks. * Synchronize output to a stream from multiple threads in parallel using the `synced_stream` helper class. * Easily measure execution time for benchmarking purposes using the `timer` helper class. + * Under continuous and active development. Bug reports and feature requests are welcome, and should be made via [GitHub issues](https://github.com/bshoshany/thread-pool/issues). ### Compiling and compatibility -This library should successfully compile on any C++17 standard-compliant compiler, on all operating systems for which such a compiler is available. Compatibility was verified with a 12-core / 24-thread AMD Ryzen 9 3900X CPU at 3.8 GHz 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 12-core / 24-thread AMD Ryzen 9 3900X CPU at 3.8 GHz using the following compilers and platforms: -* GCC v10.2.0 on Windows 10 build 19042.928. -* GCC v10.3.0 on Ubuntu 21.04. -* Clang v11.0.0 on Windows 10 build 19042.928 and Ubuntu 21.04. -* MSVC v14.28.29910 on Windows 10 build 19042.928. +* Windows 10 build 19043.1165: + * [GCC](https://gcc.gnu.org/) v11.2.0 ([WinLibs build](https://winlibs.com/)) + * [Clang](https://clang.llvm.org/) v12.0.1 + * [MSVC](https://docs.microsoft.com/en-us/cpp/) v19.29.30133 +* Ubuntu 21.04: + * [GCC](https://gcc.gnu.org/) v11.1.0 + * [Clang](https://clang.llvm.org/) v12.0.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 at 2.4 GHz, for a total of 40 cores and 80 threads, running CentOS Linux 7.6.1810, using the following compilers: +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 at 2.4 GHz (for a total of 40 cores and 80 threads), running CentOS Linux 7.6.1810, using the following compilers: -* GCC v9.2.0 -* Intel C++ Compiler (ICC) v19.1.3.304 +* [GCC](https://gcc.gnu.org/) v9.4.0 +* [Intel C++ Compiler (ICC)](https://software.intel.com/content/www/us/en/develop/tools/oneapi/components/dpc-compiler.html) v19.1.3.304 -As this library requires C++17 features, the code must be compiled with C++17 support. For GCC, Clang, and ICC, use the `-std=c++17` flag. For MSVC, use `/std:c++17`. On Linux, you will also need to use the `-pthread` flag with GCC, Clang, or ICC to enable the POSIX threads library. +The test program `thread_pool_test.cpp` was compiled without warnings (with the warning flags `-Wall -Wpedantic -Wextra -Wconversion -Weffc++` in GCC/Clang and `/W4` in MSVC), executed, and successfully completed all [automated tests](#testing-the-package) 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: + +* For GCC, Clang, or ICC, 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`. + +For maximum performance, it is recommended to compile with all available compiler optimizations: + +* For GCC, Clang, or ICC, use the `-O3` flag. +* For MSVC, use `/O2`. ## Getting started ### Including the library -To use the thread pool library, simply include the header file: +To use the thread pool library, simply download the [latest release](https://github.com/bshoshany/thread-pool/releases) from the GitHub repository, place the single header file `thread_pool.hpp` in the desired folder, and include it in your program: ```cpp #include "thread_pool.hpp" @@ -121,6 +140,26 @@ To use the thread pool library, simply include the header file: The thread pool will now be accessible via the `thread_pool` class. +### Installing using vcpkg + +If you are using the [vcpkg](https://github.com/microsoft/vcpkg) C/C++ library manager, you can easily download and install this package with the following commands. + +On Linux/macOS: + +```none +./vcpkg install bshoshany-thread-pool +``` + +On Windows: + +```none +.\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 "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. + ### Constructors The default constructor creates a thread pool with as many threads as the hardware can handle concurrently, as reported by the implementation via `std::thread::hardware_concurrency()`. With a hyperthreaded CPU, this will be twice the number of CPU cores. This is probably the constructor you want to use. For example: @@ -149,6 +188,20 @@ It is generally unnecessary to change the number of threads in the pool after it `reset()` will wait for all currently running tasks to be completed, but will leave the rest of the tasks in the queue. Then it will destroy the thread pool and create a new one with the desired new number of threads, as specified in the function's argument (or the hardware concurrency if no argument is given). The new thread pool will then resume executing the tasks that remained in the queue and any new submitted tasks. +### Finding the version of the package + +If desired, the version of this package may be read during compilation time from the macro `THREAD_POOL_VERSION`. The value will be a string containing the version number and release date. For example: + +```cpp +std::cout << "Thread pool library version is " << THREAD_POOL_VERSION << ".\n"; +``` + +Sample output: + +```none +Thread pool library version is v2.0.0 (2021-08-14). +``` + ## Submitting and waiting for tasks ### Submitting tasks to the queue with futures @@ -224,36 +277,34 @@ after the `for` loop will ensure - as efficiently as possible - that all tasks h Consider the following loop: ```cpp -for (T i = start; i <= end; i++) - loop(i); +for (T i = start; i < end; i++) + do_something(i); ``` where: * `T` is any signed or unsigned integer type. -* `start` is the first index to loop over (inclusive). -* `end` is the last index to loop over (inclusive). -* `loop()` is a function that takes exactly one argument, the loop index, and has no return value. +* The loop is over the range `[start, end)`, i.e. inclusive of `start` but exclusive of `end`. +* `do_something()` is an operation performed for each loop index `i`, such as modifying an array with `end - start` elements. This loop may be automatically parallelized and submitted to the thread pool's queue using the member function `parallelize_loop()` as follows: ```cpp -// Equivalent to the above loop, but will be automatically parallelized. -pool.parallelize_loop(start, end, loop); +auto loop = [](const T &a, const T &b) +{ + for (T i = a; i < b; i++) + do_something(i); +}; +pool.parallelize_loop(start, end, loop, n); ``` -The loop will be parallelized into a number of tasks equal to the number of threads in the pool, with each task executing the function `loop()` for a roughly equal number of indices. The main thread will then wait until all tasks generated by `parallelize_loop()` finish executing (and only those tasks - not any other tasks that also happen to be in the queue). +The range of indices `[start, end)` will be divided into `n` blocks of the form `[a, b)`. For example, if the range is `[0, 9)` and there are 3 blocks, then the blocks will be the ranges `[0, 3)`, `[3, 6)`, and `[6, 9)`. If possible, the blocks will be equal in size, otherwise the last block may be a bit longer. Then, a task will be submitted for each block, consisting of the function `loop()` with its two arguments being the start and end of the range `[a, b)` of each block. The main thread will then wait until all tasks generated by `parallelize_loop()` finish executing (and only those tasks - not any other tasks that also happen to be in the queue). -If desired, the number of parallel tasks may be manually specified using a fourth argument: +In the example above, the lambda function `loop` was defined separately for clarity. In practice, the lambda function will usually be defined within the argument itself, as in the example below. `loop` can also be an ordinary function (with no return value) instead of a lambda function, but that may be less useful, since typically one would like to capture some of the surrounding variables, as below. -```cpp -// Parallelize the loop into 12 parallel tasks -pool.parallelize_loop(start, end, loop, 12); -``` +If the fourth argument `n` is not specified, the number of blocks will be equal to the number of threads in the pool. For best performance, it is recommended to do your own benchmarks to find the optimal number of blocks for each loop (you can use the `timer` helper class - see [below](#measuring-execution-time)). Using less tasks than there are threads may be preferred if you are also running other tasks in parallel. Using more tasks than there are threads may improve performance in some cases. -For best performance, it is recommended to do your own benchmarks to find the optimal number of tasks for each loop (you can use the `timer` helper class - see [below](#measuring-execution-time)). Using less tasks than there are threads may be preferred if you are also running other tasks in parallel. Using more tasks than there are threads may improve performance in some cases. - -As a simple example, the following code will calculate the squares of all integers from 0 to 99. Since there are 10 threads, the loop will be divided into 10 tasks, each calculating 10 squares: +As a simple example, the following code will calculate the squares of all integers from 0 to 99. Since there are 10 threads, and we did not specify a fourth argument, the loop will be divided into 10 blocks, each calculating 10 squares: ```cpp #include "thread_pool.hpp" @@ -261,8 +312,13 @@ As a simple example, the following code will calculate the squares of all intege int main() { thread_pool pool(10); - size_t squares[100]; - pool.parallelize_loop(0, 99, [&squares](size_t i) { squares[i] = i * i; }); + uint32_t squares[100]; + pool.parallelize_loop(0, 100, + [&squares](const uint32_t &a, const uint32_t &b) + { + for (uint32_t i = a; i < b; i++) + squares[i] = i * i; + }); std::cout << "16^2 = " << squares[16] << '\n'; std::cout << "32^2 = " << squares[32] << '\n'; } @@ -275,7 +331,7 @@ The output should be: 32^2 = 1024 ``` -## Other features +## Helper classes ### Synchronizing printing to an output stream @@ -335,6 +391,35 @@ Task no. 5 executing. **Warning:** Always create the `synced_stream` object **before** the `thread_pool` object, as we did in this example. When the `thread_pool` object goes out of scope, it waits for the remaining tasks to be executed. If the `synced_stream` object goes out of scope before the `thread_pool` object, then any tasks using the `synced_stream` will crash. Since objects are destructed in the opposite order of construction, creating the `synced_stream` object before the `thread_pool` object ensures that the `synced_stream` is always available to the tasks, even while the pool is destructing. +### Measuring execution time + +If you are using a thread pool, then your code is most likely performance-critical. Achieving maximum performance requires performing a considerable amount of benchmarking to determine the optimal settings and algorithms. Therefore, it is important to be able to measure the execution time of various computations and operations under different conditions. + +For example, you may be interested in figuring out: + +* The optimal number of threads in the pool. +* The optimal number of tasks to divide a specific operation into, either using `parallelize_loop()` or manually. +* The optimal [sleep duration](#setting-the-worker-functions-sleep-duration) for the worker functions. + +The helper class `timer` provides a simple way to measure execution time. It is very straightforward to use: + +1. Create a new `timer` object. +2. Immediately before you execute the computation that you want to time, call the `start()` member function. +3. Immediately after the computation ends, call the `stop()` member function. +4. Use the member function `ms()` to obtain the elapsed time for the computation in milliseconds. + +For example: + +```cpp +timer tmr; +tmr.start(); +do_something(); +tmr.stop(); +std::cout << "The elapsed time was " << tmr.ms() << " ms.\n"; +``` + +## Other features + ### Setting the worker function's sleep duration The **worker function** is the function that controls the execution of tasks by each thread. It loops continuously, and with each iteration of the loop, checks if there are any tasks in the queue. If it finds a task, it pops it out of the queue and executes it. If it does not find a task, it will wait for a bit, by calling `std::this_thread::sleep_for()`, and then check the queue again. The public member variable `sleep_duration` controls the duration, in microseconds, that the worker function sleeps for when it cannot find a task in the queue. @@ -361,35 +446,36 @@ These functions are demonstrated in the following program: ```cpp #include "thread_pool.hpp" -void sleep_half_second(const size_t &i, synced_stream *sync_out) +synced_stream sync_out; +thread_pool pool(4); + +void sleep_half_second(const size_t &i) { std::this_thread::sleep_for(std::chrono::milliseconds(500)); - sync_out->println("Task ", i, " done."); + sync_out.println("Task ", i, " done."); } -void monitor_tasks(const thread_pool *pool, synced_stream *sync_out) +void monitor_tasks() { - sync_out->println(pool->get_tasks_total(), - " tasks total, ", - pool->get_tasks_running(), - " tasks running, ", - pool->get_tasks_queued(), - " tasks queued."); + sync_out.println(pool.get_tasks_total(), + " tasks total, ", + pool.get_tasks_running(), + " tasks running, ", + pool.get_tasks_queued(), + " tasks queued."); } int main() { - synced_stream sync_out; - thread_pool pool(4); for (size_t i = 0; i < 12; i++) - pool.push_task(sleep_half_second, i, &sync_out); - monitor_tasks(&pool, &sync_out); + pool.push_task(sleep_half_second, i); + monitor_tasks(); std::this_thread::sleep_for(std::chrono::milliseconds(750)); - monitor_tasks(&pool, &sync_out); + monitor_tasks(); std::this_thread::sleep_for(std::chrono::milliseconds(500)); - monitor_tasks(&pool, &sync_out); + monitor_tasks(); std::this_thread::sleep_for(std::chrono::milliseconds(500)); - monitor_tasks(&pool, &sync_out); + monitor_tasks(); } ``` @@ -425,18 +511,19 @@ Here is an example: ```cpp #include "thread_pool.hpp" -void sleep_half_second(const size_t &i, synced_stream *sync_out) +synced_stream sync_out; +thread_pool pool(4); + +void sleep_half_second(const size_t &i) { std::this_thread::sleep_for(std::chrono::milliseconds(500)); - sync_out->println("Task ", i, " done."); + sync_out.println("Task ", i, " done."); } int main() { - synced_stream sync_out; - thread_pool pool(4); for (size_t i = 0; i < 8; i++) - pool.push_task(sleep_half_second, i, &sync_out); + pool.push_task(sleep_half_second, i); sync_out.println("Submitted 8 tasks."); std::this_thread::sleep_for(std::chrono::milliseconds(250)); pool.paused = true; @@ -445,7 +532,7 @@ int main() sync_out.println("Still paused..."); std::this_thread::sleep_for(std::chrono::milliseconds(1000)); for (size_t i = 8; i < 12; i++) - pool.push_task(sleep_half_second, i, &sync_out); + pool.push_task(sleep_half_second, i); sync_out.println("Submitted 4 more tasks."); sync_out.println("Still paused..."); std::this_thread::sleep_for(std::chrono::milliseconds(1000)); @@ -484,22 +571,23 @@ While the workers are paused, `wait_for_tasks()` will wait for the running tasks ```cpp #include "thread_pool.hpp" -void sleep_half_second(const size_t &i, synced_stream *sync_out) +synced_stream sync_out; +thread_pool pool(4); + +void sleep_half_second(const size_t &i) { std::this_thread::sleep_for(std::chrono::milliseconds(500)); - sync_out->println("Task ", i, " done."); + sync_out.println("Task ", i, " done."); } int main() { - synced_stream sync_out; - thread_pool pool(4); for (size_t i = 0; i < 8; i++) - pool.push_task(sleep_half_second, i, &sync_out); + pool.push_task(sleep_half_second, i); sync_out.println("Submitted 8 tasks. Waiting for them to complete."); pool.wait_for_tasks(); for (size_t i = 8; i < 20; i++) - pool.push_task(sleep_half_second, i, &sync_out); + pool.push_task(sleep_half_second, i); sync_out.println("Submitted 12 more tasks."); std::this_thread::sleep_for(std::chrono::milliseconds(250)); pool.paused = true; @@ -549,13 +637,14 @@ Task 16 done. Task 17 done. Task 18 done. Task 19 done. +All tasks completed. ``` The first `wait_for_tasks()`, which was called with `paused == false`, waited for all 8 tasks, both running and queued. The second `wait_for_tasks()`, which was called with `paused == true`, only waited for the 4 running tasks, while the other 8 tasks remained queued, and were not executed since the pool was paused. Finally, the third `wait_for_tasks()`, which was called with `paused == false`, waited for the remaining 8 tasks, both running and queued. **Warning**: If the thread pool is destroyed while paused, any tasks still in the queue will never be executed. -## Exception handling +### Exception handling `submit()` catches any exceptions thrown by the submitted task and forwards them to the corresponding future. They can then be caught when invoking the `get()` member function of the future. For example: @@ -592,127 +681,339 @@ The output will be: Caught exception: Division by zero! ``` -## Performance tests +## Testing the package -### Measuring execution time +The included file `thread_pool_test.cpp` will perform automated tests of all aspects of the package, and benchmark some multithreaded matrix operations. The output will be printed both to `std::cout` and to a file named `thread_pool_test-yyyy-mm-dd_hh.mm.ss.log` based on the current date and time. In addition, the code is thoroughly documented, and is meant to serve as an extensive example of how to properly use the package. -If you are using a thread pool, then your code is most likely performance-critical. Achieving maximum performance requires performing a considerable amount of benchmarking to determine the optimal settings and algorithms. Therefore, it is important to be able to measure the execution time of various computations under different conditions. In the context of the thread pool class, you would probably be interested in finding the optimal number of threads in the pool and the optimal sleep duration for the worker functions. +Please make sure to: -The helper class `timer` provides a simple way to measure execution time. It is very straightforward to use: +1. [Compile](#compiling-and-compatibility) `thread_pool_test.cpp` with optimization flags enabled (e.g. `-O3` on GCC / Clang or `/O2` on MSVC). +2. Run the test without any other applications, especially multithreaded applications, running in parallel. -1. Create a new `timer` object. -2. Immediately before you execute the computation that you want to time, call the `start()` member function. -3. Immediately after the computation ends, call the `stop()` member function. -4. Use the member function `ms()` to obtain the elapsed time for the computation in milliseconds. +If any of the tests fail, please [submit a bug report](https://github.com/bshoshany/thread-pool/issues) including the exact specifications of your system (OS, CPU, compiler, etc.) and the generated log file. -For example: +### Automated tests -```cpp -timer tmr; -tmr.start(); -do_something(); -tmr.stop(); -std::cout << "The elapsed time was " << tmr.ms() << " ms.\n"; -``` - -To benchmark the performance of our thread pool class, we measured the execution time of various parallelized operations on large matrices, using a custom-built matrix class template. The test code makes use of version 1.3 of the `thread_pool` class and implements a generalization of its `parallelize_loop()` member function adapted specifically for parallelizing matrix operations. Execution time was measured using the `timer` helper class. - -For each matrix operation, we parallelized the computation into blocks. Each block consists of a number of atomic operations equal to the block size, and was submitted as a separate task to the thread pool's queue, such that the number of blocks equals the total number of tasks. We tested 6 different block sizes for each operation in order to compare their execution time. - -### AMD Ryzen 9 3900X (24 threads) - -The first test was performed on a computer equipped with a 12-core / 24-thread AMD Ryzen 9 3900X CPU at 3.8 GHz, compiled using GCC v10.3.0 on Ubuntu 21.04 with the `-O3` compiler flag. The thread pool consisted of 24 threads, making full use of the CPU's hyperthreading capabilities. - -The output of our test program was as follows: +A sample output of a successful run of the automated tests is as follows: ```none -Adding two 4800x4800 matrices: -With block size of 23040000 ( 1 block ), execution took 37 ms. -With block size of 3840000 ( 6 blocks), execution took 17 ms. -With block size of 1920000 (12 blocks), execution took 16 ms. -With block size of 960000 (24 blocks), execution took 17 ms. -With block size of 480000 (48 blocks), execution took 17 ms. -With block size of 240000 (96 blocks), execution took 17 ms. +A C++17 Thread Pool for High-Performance Scientific Computing +(c) 2021 Barak Shoshany (baraksh@gmail.com) (http://baraksh.com) +GitHub: https://github.com/bshoshany/thread-pool -Generating random 4800x4800 matrix: -With block size of 23040000 ( 1 block ), execution took 291 ms. -With block size of 3840000 ( 6 blocks), execution took 52 ms. -With block size of 1920000 (12 blocks), execution took 27 ms. -With block size of 960000 (24 blocks), execution took 25 ms. -With block size of 480000 (48 blocks), execution took 20 ms. -With block size of 240000 (96 blocks), execution took 17 ms. +Thread pool library version is v2.0.0 (2021-08-14). +Hardware concurrency is 24. +Generating log file: thread_pool_test-2021-08-14_23.34.25.log. -Transposing one 4800x4800 matrix: -With block size of 23040000 ( 1 block ), execution took 129 ms. -With block size of 3840000 ( 6 blocks), execution took 24 ms. -With block size of 1920000 (12 blocks), execution took 19 ms. -With block size of 960000 (24 blocks), execution took 17 ms. -With block size of 480000 (48 blocks), execution took 16 ms. -With block size of 240000 (96 blocks), execution took 15 ms. +Important: Please do not run any other applications, especially multithreaded applications, in parallel with this test! -Multiplying two 800x800 matrices: -With block size of 640000 ( 1 block ), execution took 431 ms. -With block size of 106666 ( 6 blocks), execution took 88 ms. -With block size of 53333 (12 blocks), execution took 61 ms. -With block size of 26666 (24 blocks), execution took 42 ms. -With block size of 13333 (48 blocks), execution took 37 ms. -With block size of 6666 (96 blocks), execution took 32 ms. +==================================== +Checking that the constructor works: +==================================== +Checking that the thread pool reports a number of threads equal to the hardware concurrency... +-> PASSED! +Checking that the manually counted number of unique thread IDs is equal to the reported number of threads... +-> PASSED! + +============================ +Checking that reset() works: +============================ +Checking that after reset() the thread pool reports a number of threads equal to half the hardware concurrency... +-> PASSED! +Checking that after reset() the manually counted number of unique thread IDs is equal to the reported number of threads... +-> PASSED! +Checking that after a second reset() the thread pool reports a number of threads equal to the hardware concurrency... +-> PASSED! +Checking that after a second reset() the manually counted number of unique thread IDs is equal to the reported number of threads... +-> PASSED! + +================================ +Checking that push_task() works: +================================ +Checking that push_task() works for a function with no arguments or return value... +-> PASSED! +Checking that push_task() works for a function with one argument and no return value... +-> PASSED! +Checking that push_task() works for a function with two arguments and no return value... +-> PASSED! + +============================= +Checking that submit() works: +============================= +Checking that submit() works for a function with no arguments or return value... +-> PASSED! +Checking that submit() works for a function with one argument and no return value... +-> PASSED! +Checking that submit() works for a function with two arguments and no return value... +-> PASSED! +Checking that submit() works for a function with no arguments and a return value... +-> PASSED! +Checking that submit() works for a function with one argument and a return value... +-> PASSED! +Checking that submit() works for a function with two arguments and a return value... +-> PASSED! + +======================================= +Checking that wait_for_tasks() works... +======================================= +-> PASSED! + +======================================= +Checking that parallelize_loop() works: +======================================= +Verifying that a loop from -2064 to 551 with 4 tasks modifies all indices... +-> PASSED! +Verifying that a loop from -658 to -77 with 19 tasks modifies all indices... +-> PASSED! +Verifying that a loop from 1512 to -1046 with 1 task modifies all indices... +-> PASSED! +Verifying that a loop from -2334 to -1770 with 23 tasks modifies all indices... +-> PASSED! +Verifying that a loop from 1775 to -1242 with 13 tasks modifies all indices... +-> PASSED! +Verifying that a loop from 846 to -506 with 14 tasks modifies all indices... +-> PASSED! +Verifying that a loop from -301 to -2111 with 5 tasks modifies all indices... +-> PASSED! +Verifying that a loop from 1758 to -1602 with 11 tasks modifies all indices... +-> PASSED! +Verifying that a loop from 94 to -1103 with 24 tasks modifies all indices... +-> PASSED! +Verifying that a loop from 612 to 2026 with 13 tasks modifies all indices... +-> PASSED! + +====================================================== +Checking that different values of sleep_duration work: +====================================================== +Submitting tasks with sleep_duration = 0 microseconds... +-> PASSED! +Submitting tasks with sleep_duration = 1909 microseconds... +-> PASSED! +Submitting tasks with sleep_duration = 469 microseconds... +-> PASSED! +Submitting tasks with sleep_duration = 964 microseconds... +-> PASSED! +Submitting tasks with sleep_duration = 1946 microseconds... +-> PASSED! +Submitting tasks with sleep_duration = 773 microseconds... +-> PASSED! +Resetting sleep_duration to the default value (1000 microseconds). + +==================================== +Checking that task monitoring works: +==================================== +Resetting pool to 4 threads. +Submitting 12 tasks. +After submission, should have: 12 tasks total, 4 tasks running, 8 tasks queued... +-> PASSED! +Task 1 released. +Task 3 released. +Task 0 released. +Task 2 released. +After releasing 4 tasks, should have: 8 tasks total, 4 tasks running, 4 tasks queued... +Task 5 released. +Task 4 released. +Task 7 released. +Task 6 released. +-> PASSED! +After releasing 4 more tasks, should have: 4 tasks total, 4 tasks running, 0 tasks queued... +-> PASSED! +Task 11 released. +Task 8 released. +Task 9 released. +Task 10 released. +After releasing the final 4 tasks, should have: 0 tasks total, 0 tasks running, 0 tasks queued... +-> PASSED! +Resetting pool to 24 threads. + +============================ +Checking that pausing works: +============================ +Resetting pool to 4 threads. +Pausing pool. +Submitting 12 tasks, each one waiting for 200ms. +Immediately after submission, should have: 12 tasks total, 0 tasks running, 12 tasks queued... +-> PASSED! +300ms later, should still have: 12 tasks total, 0 tasks running, 12 tasks queued... +-> PASSED! +Unpausing pool. +Task 1 done. +Task 2 done. +Task 3 done. +Task 0 done. +300ms later, should have: 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. +After waiting, should have: 4 tasks total, 0 tasks running, 4 tasks queued... +-> PASSED! +200ms later, should still have: 4 tasks total, 0 tasks running, 4 tasks queued... +-> PASSED! +Unpausing pool and using wait_for_tasks() to wait for all tasks. +Task 9 done. +Task 8 done. +Task 10 done. +Task 11 done. +After waiting, should have: 0 tasks total, 0 tasks running, 0 tasks queued... +-> PASSED! +Resetting pool to 24 threads. + +======================================= +Checking that exception handling works: +======================================= +-> PASSED! + +============================================================ +Testing that matrix operations produce the expected results: +============================================================ +Using matrices of size 240x240 with a total of 57600 elements. +Adding two matrices (single-threaded)... +Adding two matrices (multithreaded)... +Comparing the results... +-> PASSED! +Transposing a matrix (single-threaded)... +Transposing a matrix (multithreaded)... +Comparing the results... +-> PASSED! +Multiplying two matrices (single-threaded)... +Multiplying two matrices (multithreaded)... +Comparing the results... +-> PASSED! + +++++++++++++++++++++++++++++++ +SUCCESS: Passed all 46 checks! +++++++++++++++++++++++++++++++ ``` -In this test, we find a speedup by roughly a factor of 2 for addition, 9 for transposition, 13 for matrix multiplication, and 17 for random matrix generation. Here are some lessons we can learn from these results: +### Performance tests -* For simple element-wise operations such as addition, multithreading improves performance very modestly, only by a factor of 2, even when utilizing every available hardware thread. This is because compiler optimizations already parallelize simple loops fairly well on their own. Omitting the `-O3` optimization flag, we observed a factor of 9 speedup for addition. However, the user will most likely be compiling with optimizations turned on anyway. -* Matrix multiplication and random matrix generation, which are more complicated operations that cannot be automatically parallelized by compiler optimizations, gain the most out of multithreading - with a very significant speedup by a factor of 15 on average. Given that the test CPU only has 12 physical cores, and hyperthreading can generally produce no more than a 30% performance improvement, a factor of 15 speedup is about as good as can be expected. -* Transposition also enjoys a factor of 9 speedup with multithreading. Note that transposition requires reading memory is non-sequential order, jumping between the rows of the source matrix, which is why, compared to sequential operations such as addition, it is much slower when single-threaded, but benefits more from multithreading, especially when split into smaller blocks. -* Even though the test CPU only has 24 threads, there is still a small but consistent benefit to dividing the computation into 48 or even 96 parallel blocks. This is especially significant in multiplication, where we get roughly a 25% speedup with 96 blocks (4 blocks per thread) compared to 24 blocks (1 block per thread). +If all checks passed, `thread_pool_test.cpp` will perform benchmarking of multithreaded matrix operations. Here we will present the results obtained with two different systems. + +The first test was performed 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 GCC v11.2.0 on Windows 10 build 19043.1165 with the `-O3` compiler flag. The thread pool used 22 out of 24 threads, leaving 2 threads free for the operating system - which in our tests increased performance, presumably since all 22 threads could be dedicated entirely to the test. The output was as follows: + +```none +=================================== +Performing matrix performance test: +=================================== +Using 22 out of 24 threads. +Determining the optimal sleep duration........................ +Result: The optimal sleep duration is 300 microseconds. + +Adding two 4400x4400 matrices 20 times: +With 1 task, mean execution time was 39.3 ms with standard deviation 2.4 ms. +With 5 tasks, mean execution time was 21.2 ms with standard deviation 1.7 ms. +With 11 tasks, mean execution time was 20.4 ms with standard deviation 1.1 ms. +With 22 tasks, mean execution time was 18.3 ms with standard deviation 1.3 ms. +With 44 tasks, mean execution time was 17.4 ms with standard deviation 0.7 ms. +With 88 tasks, mean execution time was 18.0 ms with standard deviation 1.0 ms. +Maximum speedup obtained: 2.3x. + +Transposing one 4400x4400 matrix 20 times: +With 1 task, mean execution time was 139.8 ms with standard deviation 3.0 ms. +With 5 tasks, mean execution time was 38.2 ms with standard deviation 2.4 ms. +With 11 tasks, mean execution time was 23.3 ms with standard deviation 1.8 ms. +With 22 tasks, mean execution time was 18.9 ms with standard deviation 1.6 ms. +With 44 tasks, mean execution time was 19.5 ms with standard deviation 1.5 ms. +With 88 tasks, mean execution time was 18.1 ms with standard deviation 0.7 ms. +Maximum speedup obtained: 7.7x. + +Multiplying two 550x550 matrices 20 times: +With 1 task, mean execution time was 165.2 ms with standard deviation 2.5 ms. +With 5 tasks, mean execution time was 35.9 ms with standard deviation 1.0 ms. +With 11 tasks, mean execution time was 17.6 ms with standard deviation 0.5 ms. +With 22 tasks, mean execution time was 10.2 ms with standard deviation 0.7 ms. +With 44 tasks, mean execution time was 16.1 ms with standard deviation 1.4 ms. +With 88 tasks, mean execution time was 15.4 ms with standard deviation 0.7 ms. +Maximum speedup obtained: 16.2x. + +Generating random 4400x4400 matrix 20 times: +With 1 task, mean execution time was 244.7 ms with standard deviation 2.6 ms. +With 5 tasks, mean execution time was 51.5 ms with standard deviation 1.5 ms. +With 11 tasks, mean execution time was 25.7 ms with standard deviation 0.9 ms. +With 22 tasks, mean execution time was 19.1 ms with standard deviation 2.7 ms. +With 44 tasks, mean execution time was 17.2 ms with standard deviation 2.1 ms. +With 88 tasks, mean execution time was 15.8 ms with standard deviation 1.0 ms. +Maximum speedup obtained: 15.5x. + +Overall, multithreading provided speedups of up to 16.2x. + ++++++++++++++++++++++++++++++++++++++++ +Thread pool performance test completed! ++++++++++++++++++++++++++++++++++++++++ +``` + +Here are some lessons we can learn from these results: + +* For simple element-wise operations such as addition, multithreading improves performance very modestly, only by a factor of 2.3, even when utilizing 22 threads in parallel. This is because compiler optimizations already parallelize simple loops fairly well on their own. Omitting the `-O3` optimization flag, we observed a 6.8x speedup for addition. However, the user will most likely be compiling with optimizations turned on anyway. +* Transposition enjoys a moderate 7.7x speedup with multithreading. Note that transposition requires reading memory is non-sequential order, jumping between the rows of the source matrix, which is why, compared to sequential operations such as addition, it is much slower when single-threaded, and benefits more from multithreading. +* Matrix multiplication and random matrix generation, which are more complicated operations that cannot be automatically parallelized by compiler optimizations, gain the most out of multithreading - with a very significant speedup by a factor of around 16 on average. Given that the test CPU only has 12 physical cores, and hyperthreading can generally produce no more than a 30% performance improvement, a 16x speedup is about as good as can be expected. +* Using as many tasks as there are threads almost always provides the best performance. Although in some cases 44 or 88 tasks seem to provide a slightly lower mean execution time compared to 22 tasks, the difference is within less than 1 standard deviation in all cases. ### Dual Intel Xeon Gold 6148 (80 threads) -The second test was performed on a [Compute Canada](https://www.computecanada.ca/) node equipped with dual 20-core / 40-thread Intel Xeon Gold 6148 CPUs at 2.4 GHz, for a total of 40 cores and 80 threads, compiled using GCC v9.2.0 on CentOS Linux 7.6.1810 with the `-O3` compiler flag. The thread pool consisted of 80 threads, making full use of the hyperthreading capabilities of both CPUs. - -We adjusted the block sizes compared to the previous test, to match the larger number of threads. The output of our test program was as follows: +The second test was performed on a [Compute Canada](https://www.computecanada.ca/) node equipped with dual 20-core / 40-thread Intel Xeon Gold 6148 CPUs at 2.4 GHz (for a total of 40 cores and 80 threads) and 202 GB of RAM, compiled using GCC v9.4.0 on CentOS Linux 7.6.1810 with the `-O3` compiler flag. The thread pool consisted of 78 threads. The output was as follows: ```none -Adding two 4800x4800 matrices: -With block size of 23040000 ( 1 block ), execution took 73 ms. -With block size of 1152000 ( 20 blocks), execution took 9 ms. -With block size of 576000 ( 40 blocks), execution took 7 ms. -With block size of 288000 ( 80 blocks), execution took 7 ms. -With block size of 144000 (160 blocks), execution took 8 ms. -With block size of 72000 (320 blocks), execution took 10 ms. +=================================== +Performing matrix performance test: +=================================== +Using 78 out of 80 threads. +Determining the optimal sleep duration........................ +Result: The optimal sleep duration is 1000 microseconds. -Generating random 4800x4800 matrix: -With block size of 23040000 ( 1 block ), execution took 423 ms. -With block size of 1152000 ( 20 blocks), execution took 29 ms. -With block size of 576000 ( 40 blocks), execution took 15 ms. -With block size of 288000 ( 80 blocks), execution took 13 ms. -With block size of 144000 (160 blocks), execution took 11 ms. -With block size of 72000 (320 blocks), execution took 10 ms. +Adding two 15600x15600 matrices 20 times: +With 1 task, mean execution time was 846.1 ms with standard deviation 40.2 ms. +With 19 tasks, mean execution time was 88.1 ms with standard deviation 8.6 ms. +With 39 tasks, mean execution time was 73.5 ms with standard deviation 4.8 ms. +With 78 tasks, mean execution time was 67.3 ms with standard deviation 2.2 ms. +With 156 tasks, mean execution time was 64.9 ms with standard deviation 2.3 ms. +With 312 tasks, mean execution time was 65.8 ms with standard deviation 1.5 ms. +Maximum speedup obtained: 13.0x. -Transposing one 4800x4800 matrix: -With block size of 23040000 ( 1 block ), execution took 167 ms. -With block size of 1152000 ( 20 blocks), execution took 18 ms. -With block size of 576000 ( 40 blocks), execution took 11 ms. -With block size of 288000 ( 80 blocks), execution took 9 ms. -With block size of 144000 (160 blocks), execution took 10 ms. -With block size of 72000 (320 blocks), execution took 12 ms. +Transposing one 15600x15600 matrix 20 times: +With 1 task, mean execution time was 1689.4 ms with standard deviation 75.3 ms. +With 19 tasks, mean execution time was 155.3 ms with standard deviation 19.7 ms. +With 39 tasks, mean execution time was 115.0 ms with standard deviation 10.8 ms. +With 78 tasks, mean execution time was 99.0 ms with standard deviation 6.0 ms. +With 156 tasks, mean execution time was 96.2 ms with standard deviation 1.6 ms. +With 312 tasks, mean execution time was 97.8 ms with standard deviation 1.7 ms. +Maximum speedup obtained: 17.6x. -Multiplying two 800x800 matrices: -With block size of 640000 ( 1 block ), execution took 771 ms. -With block size of 32000 ( 20 blocks), execution took 57 ms. -With block size of 16000 ( 40 blocks), execution took 24 ms. -With block size of 8000 ( 80 blocks), execution took 21 ms. -With block size of 4000 (160 blocks), execution took 17 ms. -With block size of 2000 (320 blocks), execution took 15 ms. +Multiplying two 1950x1950 matrices 20 times: +With 1 task, mean execution time was 15415.1 ms with standard deviation 672.5 ms. +With 19 tasks, mean execution time was 1152.5 ms with standard deviation 62.8 ms. +With 39 tasks, mean execution time was 537.9 ms with standard deviation 4.1 ms. +With 78 tasks, mean execution time was 292.3 ms with standard deviation 42.5 ms. +With 156 tasks, mean execution time was 936.4 ms with standard deviation 15.8 ms. +With 312 tasks, mean execution time was 951.2 ms with standard deviation 22.3 ms. +Maximum speedup obtained: 52.7x. + +Generating random 15600x15600 matrix 20 times: +With 1 task, mean execution time was 4318.3 ms with standard deviation 6.3 ms. +With 19 tasks, mean execution time was 260.8 ms with standard deviation 15.1 ms. +With 39 tasks, mean execution time was 156.1 ms with standard deviation 1.6 ms. +With 78 tasks, mean execution time was 86.2 ms with standard deviation 1.9 ms. +With 156 tasks, mean execution time was 84.8 ms with standard deviation 0.4 ms. +With 312 tasks, mean execution time was 85.2 ms with standard deviation 1.3 ms. +Maximum speedup obtained: 51.0x. + +Overall, multithreading provided speedups of up to 52.7x. + ++++++++++++++++++++++++++++++++++++++++ +Thread pool performance test completed! ++++++++++++++++++++++++++++++++++++++++ ``` -In this test, we find a speedup by roughly a factor of 10 for addition, 19 for transposition, 42 for random matrix generation, and 51 for matrix multiplication. The last result again matches the estimation of a 30% improvement in performance due to hyperthreading, which indicates that we are once again saturating the maximum possible performance of our system. +The speedup of around 51.9x on average for matrix multiplication and random matrix generation again matches the estimation of a 30% improvement in performance over the 40 physical CPU cores due to hyperthreading, which indicates that we are once again saturating the maximum possible performance of our system. -An interesting point to notice is that for **single-threaded** calculations (1 block), the dual Xeon CPUs actually perform worse by up to a factor of 2 compared to the single Ryzen CPU. This is due to the base clock speed of the Ryzen (3.8 GHz) being considerably higher than the base clock speed of the Xeon (2.4 GHz). Since each core of the Xeon is slower than each core of the Ryzen, we need more parallelization to achieve the same overall speed. However, with full parallelization (24 threads on the Ryzen, 80 threads on the Xeon), the latter is faster by about a factor of 2. +## Issue and pull request policy -## Feedback +This package is under continuous and active development. If you encounter any bugs, or if you would like to request any additional features, please feel free to [open a new issue on GitHub](https://github.com/bshoshany/thread-pool/issues) and I will look into it as soon as I can. -If you would like a request any additional features, or if you encounter any bugs, please feel free to [open a new issue](https://github.com/bshoshany/thread-pool/issues)! +Contributions are always welcome. However, I release my projects in cumulative updates after editing them locally on my system, so my policy is not to accept any pull requests. If you open a pull request, and I decide to incorporate it into the code, I will first perform some tests to ensure that the change doesn't break anything, and then merge it into the next release of the project, possibly together with some other changes, and along with a version bump and a corresponding note in `CHANGELOG.md` with a link to the pull request. ## Copyright and citing diff --git a/thread_pool.hpp b/thread_pool.hpp index b260509..4460e54 100644 --- a/thread_pool.hpp +++ b/thread_pool.hpp @@ -3,8 +3,8 @@ /** * @file thread_pool.hpp * @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com) - * @version 1.9 - * @date 2021-07-29 + * @version 2.0.0 + * @date 2021-08-14 * @copyright Copyright (c) 2021 Barak Shoshany. Licensed under the MIT license. 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) * @@ -12,7 +12,7 @@ * @details A modern C++17-compatible thread pool implementation, built from scratch with high-performance scientific computing in mind. The thread pool is implemented as a single lightweight and self-contained class, and does not have any dependencies other than the C++17 standard library, thus allowing a great degree of portability. In particular, this implementation does not utilize OpenMP or any other high-level multithreading APIs, and thus gives the programmer precise low-level control over the details of the parallelization, which permits more robust optimizations. The thread pool was extensively tested on both AMD and Intel CPUs with up to 40 cores and 80 threads. Other features include automatic generation of futures and easy parallelization of loops. Two helper classes enable synchronizing printing to an output stream by different threads and measuring execution time for benchmarking purposes. Please visit the GitHub repository at https://github.com/bshoshany/thread-pool for documentation and updates, or to submit feature requests and bug reports. */ -#define THREAD_POOL_VERSION 1.9 +#define THREAD_POOL_VERSION "v2.0.0 (2021-08-14)" #include // std::atomic #include // std::chrono @@ -24,8 +24,8 @@ #include // std::mutex, std::scoped_lock #include // std::queue #include // std::this_thread, std::thread -#include // std::decay_t, std::enable_if_t, std::is_void_v, std::invoke_result_t -#include // std::move, std::swap +#include // std::common_type_t, std::decay_t, std::enable_if_t, std::is_void_v, std::invoke_result_t +#include // std::move // ============================================================================================= // // Begin class thread_pool // @@ -36,6 +36,7 @@ class thread_pool { typedef std::uint_fast32_t ui32; + typedef std::uint_fast64_t ui64; public: // ============================ @@ -72,7 +73,7 @@ public: * * @return The number of queued tasks. */ - size_t get_tasks_queued() const + ui64 get_tasks_queued() const { const std::scoped_lock lock(queue_mutex); return tasks.size(); @@ -109,39 +110,49 @@ public: } /** - * @brief Parallelize a loop by splitting it into blocks, submitting each block separately to the thread pool, and waiting for all blocks to finish executing. The loop will be equivalent to: for (T i = first_index; i <= last_index; i++) loop(i); + * @brief Parallelize a loop by splitting it into blocks, submitting each block separately to the thread pool, and waiting for all blocks to finish executing. The user supplies a loop function, which will be called once per block and should iterate over the block's range. * - * @tparam T The type of the loop index. Should be a signed or unsigned integer. + * @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 F The type of the function to loop through. - * @param first_index The first index in the loop (inclusive). - * @param last_index The last index in the loop (inclusive). - * @param loop The function to loop through. Should take exactly one argument, the loop index. - * @param num_tasks The maximum number of tasks to split the loop into. The default is to use the number of threads in the pool. + * @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, the function will terminate without doing anything. + * @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 - void parallelize_loop(T first_index, T last_index, const F &loop, ui32 num_tasks = 0) + template + void parallelize_loop(const T1 &first_index, const T2 &index_after_last, const F &loop, ui32 num_blocks = 0) { - if (num_tasks == 0) - num_tasks = thread_count; - if (last_index < first_index) - std::swap(last_index, first_index); - size_t total_size = last_index - first_index + 1; - size_t block_size = total_size / num_tasks; + typedef std::common_type_t T; + T the_first_index = (T)first_index; + T last_index = (T)index_after_last; + if (the_first_index == last_index) + return; + if (last_index < the_first_index) + { + T temp = last_index; + last_index = the_first_index; + the_first_index = temp; + } + last_index--; + if (num_blocks == 0) + num_blocks = thread_count; + ui64 total_size = (ui64)(last_index - the_first_index + 1); + ui64 block_size = (ui64)(total_size / num_blocks); if (block_size == 0) { block_size = 1; - num_tasks = (ui32)total_size > 1 ? (ui32)total_size : 1; + num_blocks = (ui32)total_size > 1 ? (ui32)total_size : 1; } std::atomic blocks_running = 0; - for (ui32 t = 0; t < num_tasks; t++) + for (ui32 t = 0; t < num_blocks; t++) { - T start = (T)(t * block_size + first_index); - T end = (t == num_tasks - 1) ? last_index : (T)((t + 1) * block_size + first_index - 1); + T start = ((T)(t * block_size) + the_first_index); + T end = (t == num_blocks - 1) ? last_index + 1 : ((T)((t + 1) * block_size) + the_first_index); blocks_running++; push_task([start, end, &loop, &blocks_running] { - for (T i = start; i <= end; i++) - loop(i); + loop(start, end); blocks_running--; }); } diff --git a/thread_pool_test.cpp b/thread_pool_test.cpp new file mode 100644 index 0000000..95b0157 --- /dev/null +++ b/thread_pool_test.cpp @@ -0,0 +1,1112 @@ +// Get rid of annoying MSVC warning. +#ifdef _MSC_VER +#define _CRT_SECURE_NO_WARNINGS +#endif + +#include +#include +#include +#include +#include + +#include "thread_pool.hpp" + +// Define short names for commonly-used integer types. +typedef std::int_fast32_t i32; +typedef std::uint_fast32_t ui32; +typedef std::int_fast64_t i64; +typedef std::uint_fast64_t ui64; + +// Define two global synced_streams objects: one prints to std::cout and the other to a file. +synced_stream sync_cout(std::cout); +std::ofstream log_file; +synced_stream sync_file(log_file); + +// A global thread pool object. +thread_pool pool; + +// A global random_device object used to seed some random number generators. +std::random_device rd; + +// Global variables to measure how many checks succeeded and how many failed. +ui32 tests_succeeded = 0; +ui32 tests_failed = 0; + +/** + * @brief Print any number of items into both std::cout and the log file, syncing both independently. + * + * @tparam T The types of the items. + * @param items The items to print. + */ +template +void dual_print(const T &...items) +{ + sync_cout.print(items...); + sync_file.print(items...); +} + +/** + * @brief Print any number of items into both std::cout and the log file, followed by a newline character, syncing both independently. + * + * @tparam T The types of the items. + * @param items The items to print. + */ +template +void dual_println(const T &...items) +{ + dual_print(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 = '=') +{ + dual_println(); + dual_println(std::string(text.length(), symbol)); + dual_println(text); + dual_println(std::string(text.length(), symbol)); +} + +/** + * @brief Get a string representing the current time. + * + * @return The string. + */ +std::string get_time() +{ + const std::time_t t = std::time(nullptr); + char time_string[32]; + std::strftime(time_string, sizeof(time_string), "%Y-%m-%d_%H.%M.%S", std::localtime(&t)); + return std::string(time_string); +} + +/** + * @brief Check if a condition is met, report the result, and count the number of successes and failures. + * + * @param condition The condition to check. + */ +void check(const bool condition) +{ + if (condition) + { + dual_println("-> PASSED!"); + tests_succeeded++; + } + else + { + dual_println("-> FAILED!"); + tests_failed++; + } +} + +/** + * @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. + */ +ui32 count_unique_threads() +{ + std::vector thread_IDs(pool.get_thread_count() * 4); + for (std::thread::id &id : thread_IDs) + pool.push_task(store_ID, &id); + pool.wait_for_tasks(); + std::sort(thread_IDs.begin(), thread_IDs.end()); + ui32 unique_threads = (ui32)(std::unique(thread_IDs.begin(), thread_IDs.end()) - thread_IDs.begin()); + return unique_threads; +} + +/** + * @brief Check that the constructor works. + */ +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()); + 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()); +} + +/** + * @brief Check that reset() works. + */ +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); + 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()); + 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()); + 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()); +} + +/** + * @brief Check that push_task() works. + */ +void check_push_task() +{ + dual_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); + } + dual_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); + } + dual_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() +{ + dual_println("Checking that submit() works for a function with no arguments or return value..."); + { + bool flag = false; + auto my_future = pool.submit([&flag] + { flag = true; }); + check(my_future.get() && flag); + } + dual_println("Checking that submit() works for a function with one argument and no return value..."); + { + bool flag = false; + auto my_future = pool.submit([](bool *flag) + { *flag = true; }, + &flag); + check(my_future.get() && flag); + } + dual_println("Checking that submit() works for a function with two arguments and no return value..."); + { + bool flag1 = false; + bool flag2 = false; + auto my_future = pool.submit([](bool *flag1, bool *flag2) + { *flag1 = *flag2 = true; }, + &flag1, &flag2); + check(my_future.get() && flag1 && flag2); + } + dual_println("Checking that submit() works for a function with no arguments and a return value..."); + { + bool flag = false; + auto my_future = pool.submit([&flag] + { + flag = true; + return 42; + }); + check(my_future.get() == 42 && flag); + } + dual_println("Checking that submit() works for a function with one argument and a return value..."); + { + bool flag = false; + auto my_future = pool.submit([](bool *flag) + { + *flag = true; + return 42; + }, + &flag); + check(my_future.get() == 42 && flag); + } + dual_println("Checking that submit() works for a function with two arguments and a return value..."); + { + bool flag1 = false; + bool flag2 = false; + auto my_future = pool.submit([](bool *flag1, bool *flag2) + { + *flag1 = *flag2 = true; + return 42; + }, + &flag1, &flag2); + check(my_future.get() == 42 && flag1 && flag2); + } +} + +/** + * @brief Check that wait_for_tasks() works. + */ +void check_wait_for_tasks() +{ + ui32 n = pool.get_thread_count() * 10; + std::vector> flags(n); + for (ui32 i = 0; i < n; i++) + pool.push_task([&flags, i] + { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + flags[i] = true; + }); + pool.wait_for_tasks(); + bool all_flags = true; + for (ui32 i = 0; i < n; i++) + all_flags = all_flags && flags[i]; + check(all_flags); +} + +/** + * @brief Check that parallelize_loop() works for a specific number of indices split over a specific number of tasks. + * + * @param start The first index in the loop. + * @param end The last index in the loop plus 1. + * @param num_tasks The number of tasks. + */ +void check_parallelize_loop(i32 start, i32 end, const ui32 &num_tasks) +{ + if (start == end) + end++; + dual_println("Verifying that a loop from ", start, " to ", end, " with ", num_tasks, num_tasks == 1 ? " task" : " tasks", " modifies all indices..."); + ui64 num_indices = (ui64)std::abs(end - start); + i32 offset = std::min(start, end); + std::vector> flags((ui64)num_indices); + pool.parallelize_loop( + start, end, [&flags, &offset](const i32 &start, const i32 &end) + { + for (i32 i = start; i < end; i++) + flags[(ui64)(i - offset)] = true; + }, + num_tasks); + bool all_flags = true; + for (ui64 i = 0; i < num_indices; i++) + all_flags = all_flags && flags[i]; + check(all_flags); +} + +/** + * @brief Check that parallelize_loop() works using several different random values for the range of indices and number of tasks. + */ +void check_parallelize_loop() +{ + std::mt19937_64 mt(rd()); + std::uniform_int_distribution index_dist((i32)pool.get_thread_count() * -100, (i32)pool.get_thread_count() * 100); + std::uniform_int_distribution task_dist(1, pool.get_thread_count()); + for (ui32 i = 0; i < 10; i++) + check_parallelize_loop(index_dist(mt), index_dist(mt), task_dist(mt)); +} + +/** + * @brief Check that sleep_duration works for a specific value. + * + * @param duration The value of sleep_duration. + */ +void check_sleep_duration(const ui32 &duration) +{ + dual_println("Submitting tasks with sleep_duration = ", duration, " microseconds..."); + pool.sleep_duration = duration; + ui32 n = pool.get_thread_count() * 100; + std::vector> flags(n); + for (ui32 i = 0; i < n; i++) + pool.push_task([&flags, i] + { flags[i] = true; }); + pool.wait_for_tasks(); + bool all_flags = true; + for (ui32 i = 0; i < n; i++) + all_flags = all_flags && flags[i]; + check(all_flags); +} + +/** + * @brief Check that sleep_duration works for several different random values. + */ +void check_sleep_duration() +{ + ui32 old_duration = pool.sleep_duration; + check_sleep_duration(0); + std::mt19937_64 mt(rd()); + std::uniform_int_distribution dist(1, 2000); + for (ui32 i = 0; i < 5; i++) + check_sleep_duration(dist(mt)); + dual_println("Resetting sleep_duration to the default value (", old_duration, " microseconds)."); + pool.sleep_duration = old_duration; +} + +/** + * @brief Check that task monitoring works. + */ +void check_task_monitoring() +{ + ui32 n = std::min(std::thread::hardware_concurrency(), 4); + dual_println("Resetting pool to ", n, " threads."); + pool.reset(n); + dual_println("Submitting ", n * 3, " tasks."); + std::vector> release(n * 3); + for (ui32 i = 0; i < n * 3; i++) + pool.push_task([&release, i] + { + while (!release[i]) + std::this_thread::yield(); + dual_println("Task ", i, " released."); + }); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + dual_println("After submission, should have: ", n * 3, " tasks total, ", n, " tasks running, ", n * 2, " tasks queued..."); + check(pool.get_tasks_total() == n * 3 && pool.get_tasks_running() == n && pool.get_tasks_queued() == n * 2); + for (ui32 i = 0; i < n; i++) + release[i] = true; + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + dual_println("After releasing ", n, " tasks, should have: ", n * 2, " tasks total, ", n, " tasks running, ", n, " tasks queued..."); + for (ui32 i = n; i < n * 2; i++) + release[i] = true; + check(pool.get_tasks_total() == n * 2 && pool.get_tasks_running() == n && pool.get_tasks_queued() == n); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + dual_println("After releasing ", n, " more tasks, should have: ", n, " tasks total, ", n, " tasks running, ", 0, " tasks queued..."); + check(pool.get_tasks_total() == n && pool.get_tasks_running() == n && pool.get_tasks_queued() == 0); + for (ui32 i = n * 2; i < n * 3; i++) + release[i] = true; + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + dual_println("After releasing the final ", n, " tasks, should have: ", 0, " tasks total, ", 0, " tasks running, ", 0, " 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()); +} + +/** + * @brief Check that pausing works. + */ +void check_pausing() +{ + ui32 n = std::min(std::thread::hardware_concurrency(), 4); + dual_println("Resetting pool to ", n, " threads."); + pool.reset(n); + dual_println("Pausing pool."); + pool.paused = true; + dual_println("Submitting ", n * 3, " tasks, each one waiting for 200ms."); + for (ui32 i = 0; i < n * 3; i++) + pool.push_task([i] + { + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + dual_println("Task ", i, " done."); + }); + dual_println("Immediately after submission, should have: ", n * 3, " tasks total, ", 0, " tasks running, ", n * 3, " 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..."); + 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..."); + 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..."); + 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..."); + 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..."); + 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()); +} + +/** + * @brief Check that exception handling work. + */ +void check_exceptions() +{ + bool caught = false; + auto my_future = pool.submit([] + { throw std::runtime_error("Exception thrown!"); }); + try + { + my_future.get(); + } + catch (const std::exception &e) + { + if (e.what() == std::string("Exception thrown!")) + caught = true; + } + check(caught); +} + +/** + * @brief A lightweight matrix class template for performance testing purposes. Not for general use; only contains the bare minimum functionality needed for the test. Based on https://github.com/bshoshany/multithreaded-matrix + * + * @tparam T The type to use for the matrix elements. + */ +template +class matrix +{ +public: + // ===================================== + // Constructors and assignment operators + // ===================================== + + /** + * @brief Construct an uninitialized matrix. + * + * @param _rows The number of rows. + * @param _cols The number of columns. + */ + matrix(const ui64 &_rows, const ui64 &_cols) + : rows(_rows), cols(_cols), smart_elements(new T[rows * cols]) + { + elements = smart_elements.get(); + } + + /** + * @brief Construct a new matrix by copying the elements of an existing matrix. + * + * @param m The matrix to be copied. + */ + matrix(const matrix &m) + : rows(m.rows), cols(m.cols), smart_elements(new T[rows * cols]) + { + elements = smart_elements.get(); + for (ui64 i = 0; i < rows * cols; i++) + elements[i] = m.elements[i]; + } + + /** + * @brief Construct a new matrix by moving the elements of an existing matrix. + * + * @param m The matrix to be moved. + */ + matrix(matrix &&m) + : rows(m.rows), cols(m.cols), smart_elements(std::move(m.smart_elements)) + { + elements = smart_elements.get(); + m.rows = 0; + m.cols = 0; + m.elements = nullptr; + } + + /** + * @brief Copy the elements of another matrix to this matrix. + * + * @param m The matrix to be copied. + * @return A reference to this matrix. + */ + matrix &operator=(const matrix &m) + { + rows = m.rows; + cols = m.cols; + smart_elements.reset(new T[rows * cols]); + elements = smart_elements.get(); + for (ui64 i = 0; i < rows * cols; i++) + elements[i] = m.elements[i]; + return *this; + } + + /** + * @brief Move the elements of another matrix to this matrix. + * + * @param m The matrix to be moved. + * @return A reference to this matrix. + */ + matrix &operator=(matrix &&m) + { + rows = m.rows; + cols = m.cols; + smart_elements = std::move(m.smart_elements); + elements = smart_elements.get(); + m.rows = 0; + m.cols = 0; + m.elements = nullptr; + return *this; + } + + // ==================== + // Overloaded operators + // ==================== + + /** + * @brief Read or modify a matrix element. + * + * @param row The row index (starting from zero). + * @param col The column index (starting from zero). + * @return A reference to the element. + */ + inline T &operator()(const ui64 &row, const ui64 &col) + { + return elements[(cols * row) + col]; + } + + /** + * @brief Read a matrix element. + * + * @param row The row index (starting from zero). + * @param col The column index (starting from zero). + * @return The value of the element. + */ + inline T operator()(const ui64 &row, const ui64 &col) const + { + return elements[(cols * row) + col]; + } + + /** + * @brief Read or modify an element of the underlying 1-dimensional array. + * + * @param i The element index (starting from zero). + * @return A reference to the element. + */ + inline T &operator[](const ui64 &i) + { + return elements[i]; + } + + /** + * @brief Read an element of the underlying 1-dimensional array. + * + * @param i The element index (starting from zero). + * @return The value of the element. + */ + inline T operator[](const ui64 &i) const + { + return elements[i]; + } + + /** + * @brief Compare this matrix to another matrix. + * + * @param m The matrix to compare to. + * @return Whether the matrices have the same elements. + */ + bool operator==(const matrix &m) const + { + bool compare_result = true; + for (ui64 i = 0; i < rows * cols; i++) + compare_result = compare_result && (elements[i] == m.elements[i]); + return compare_result; + } + + // ======================= + // Public member functions + // ======================= + + /** + * @brief Transpose a matrix. + * + * @param num_tasks The number of parallel tasks to use. If set to 0, no multithreading will be used. + * @return The transposed matrix. + */ + matrix transpose(const ui32 &num_tasks) const + { + matrix out(cols, rows); + if (num_tasks == 0) + { + for (ui64 i = 0; i < out.rows; i++) + for (ui64 j = 0; j < out.cols; j++) + out(i, j) = operator()(j, i); + } + else + { + pool.parallelize_loop( + 0, out.rows, [this, &out](const ui64 &start, const ui64 &end) + { + for (ui64 i = start; i < end; i++) + for (ui64 j = 0; j < out.cols; j++) + out(i, j) = operator()(j, i); + }, + num_tasks); + } + return out; + } + + // ================ + // Friend functions + // ================ + + /** + * @brief Add two matrices using the specified number of parallel tasks. + * + * @param a The first matrix to be added. + * @param b The second matrix to be added. + * @param num_tasks The number of parallel tasks to use. If set to 0, no multithreading will be used. + * @return The sum of the matrices. + */ + friend matrix add_matrices(const matrix &a, const matrix &b, const ui32 &num_tasks) + { + matrix c(a.rows, a.cols); + if (num_tasks == 0) + for (ui64 i = 0; i < a.rows * a.cols; i++) + c[i] = a[i] + b[i]; + else + pool.parallelize_loop( + 0, a.rows * a.cols, [&a, &b, &c](const ui64 &start, const ui64 &end) + { + for (ui64 i = start; i < end; i++) + c[i] = a[i] + b[i]; + }, + num_tasks); + return c; + } + + /** + * @brief Multiply two matrices using the specified number of parallel tasks. + * + * @param a The first matrix to be multiplied. + * @param b The second matrix to be multiplied. + * @param num_tasks The number of parallel tasks to use. If set to 0, no multithreading will be used. + * @return The product of the matrices. + */ + friend matrix multiply_matrices(const matrix &a, const matrix &b, const ui32 &num_tasks) + { + matrix c(a.rows, b.cols); + if (num_tasks == 0) + { + for (ui64 i = 0; i < a.rows; i++) + for (ui64 j = 0; j < b.cols; j++) + { + c(i, j) = 0; + for (ui64 k = 0; k < a.cols; k++) + c(i, j) += a(i, k) * b(k, j); + } + } + else + { + pool.parallelize_loop( + 0, a.rows, [&a, &b, &c, &a_cols = a.cols, &b_cols = b.cols](const ui64 &start, const ui64 &end) + { + for (ui64 i = start; i < end; i++) + for (ui64 j = 0; j < b_cols; j++) + { + c(i, j) = 0; + for (ui64 k = 0; k < a_cols; k++) + c(i, j) += a(i, k) * b(k, j); + } + }, + num_tasks); + } + return c; + } + +private: + // ======================== + // Private member variables + // ======================== + + /** + * @brief The number of rows. + */ + ui64 rows = 0; + + /** + * @brief The number of columns. + */ + ui64 cols = 0; + + /** + * @brief A pointer to an array storing the elements of the matrix in flattened 1-dimensional form. + */ + T *elements = nullptr; + + /** + * @brief A smart pointer to manage the memory allocated for the matrix elements. + */ + std::unique_ptr smart_elements; +}; + +/** + * @brief A class template for generating random matrices. + * + * @tparam T The type to use for the matrix elements. + * @tparam D The distribution to use, e.g. std::uniform_real_distribution. + */ +template +class random_matrix_generator +{ +public: + // ============ + // Constructors + // ============ + + /** + * @brief Construct a new random matrix generator. + * + * @tparam P The types of the parameters to pass to the constructor of the distribution. + * @param params The parameters to pass to the constructor of the distribution. The number of parameters and their types depends on the particular distribution being used. + */ + template + random_matrix_generator(const P &...params) : dist(params...), rd() {} + + // ======================= + // Public member functions + // ======================= + + /** + * @brief Generate a random matrix with the given number of rows and columns. + * + * @param rows The desired number of rows in the matrix. + * @param cols The desired number of columns in the matrix. + * @param num_tasks The number of parallel tasks to use. If set to 0, no multithreading will be used. + * @return The random matrix. + */ + matrix generate_matrix(const ui64 &rows, const ui64 &cols, const ui32 &num_tasks) + { + matrix m(rows, cols); + if (num_tasks == 0) + { + std::mt19937_64 mt(generate_seed()); + for (ui64 i = 0; i < rows * cols; i++) + m[i] = dist(mt); + } + else + pool.parallelize_loop( + 0, rows * cols, [this, &m](const ui64 &start, const ui64 &end) + { + std::mt19937_64 mt(generate_seed()); + for (ui64 i = start; i < end; i++) + m[i] = dist(mt); + }, + num_tasks); + return m; + } + +private: + // ======================== + // Private member functions + // ======================== + + /** + * @brief Generate a seed. The std::mt19937_64 in each block will be seeded using this function in order to avoid depleting the entropy of the random_device. + * + * @return A random unsigned 64-bit integer. + */ + ui64 generate_seed() + { + static std::mt19937_64 mt(rd()); + return mt(); + } + + // ======================== + // Private member variables + // ======================== + + /** + * @brief The distribution to use for generating random numbers. + */ + D dist; + + /** + * @brief The random device to be used for seeding the pseudo-random number generators. + */ + std::random_device rd; +}; + +/** + * @brief Check the matrix class template by comparing the results of adding, multiplying, and transposing matrices calculated in two ways: single-threaded and multithreaded. + */ +void check_matrix() +{ + // Initialize a random_matrix_generator object to generates matrices with integers uniformly distributed between -1000 and 1000. + random_matrix_generator> rnd(-1000, 1000); + // Define the size of the matrices to use. + const ui32 thread_count = pool.get_thread_count(); + const ui64 rows = thread_count * 10; + const ui64 cols = rows; + const ui64 total_size = rows * cols; + dual_println("Using matrices of size ", rows, "x", cols, " with a total of ", total_size, " elements."); + + matrix A = rnd.generate_matrix(rows, cols, thread_count); + matrix B = rnd.generate_matrix(rows, cols, thread_count); + + dual_println("Adding two matrices (single-threaded)..."); + matrix ApB_single = add_matrices(A, B, 0); + dual_println("Adding two matrices (multithreaded)..."); + matrix ApB_multi = add_matrices(A, B, thread_count); + dual_println("Comparing the results..."); + check(ApB_single == ApB_multi); + + dual_println("Transposing a matrix (single-threaded)..."); + matrix At_single = A.transpose(0); + dual_println("Transposing a matrix (multithreaded)..."); + matrix At_multi = A.transpose(thread_count); + dual_println("Comparing the results..."); + check(At_single == At_multi); + + dual_println("Multiplying two matrices (single-threaded)..."); + matrix AxB_single = multiply_matrices(A, B, 0); + dual_println("Multiplying two matrices (multithreaded)..."); + matrix AxB_multi = multiply_matrices(A, B, thread_count); + dual_println("Comparing the results..."); + check(AxB_single == AxB_multi); +} + +/** + * @brief Print the timing of a specific test. + * + * @param num_tasks The number of tasks. + * @param mean_sd std::pair containing the mean as the first member and standard deviation as the second member. + */ +void print_timing(const ui32 &num_tasks, const std::pair &mean_sd) +{ + if (num_tasks == 1) + dual_print("With 1 task"); + else + dual_print("With ", std::setw(3), num_tasks, " tasks"); + dual_println(", mean execution time was ", std::setw(6), mean_sd.first, " ms with standard deviation ", std::setw(4), mean_sd.second, " ms."); +} + +/** + * @brief Calculate and print the speedup obtained by multithreading. + * + * @param timings A vector of the timings corresponding to different numbers of tasks. + * @return The maximum speedup obtained. + */ +double print_speedup(const std::vector &timings) +{ + const auto [min_time, max_time] = std::minmax_element(std::begin(timings), std::end(timings)); + double max_speedup = *max_time / *min_time; + dual_println("Maximum speedup obtained: ", max_speedup, "x."); + return max_speedup; +} + +/** + * @brief Calculate the mean and standard deviation of a set of integers. + * + * @param timings The integers. + * @return std::pair containing the mean as the first member and standard deviation as the second member. + */ +std::pair analyze(const std::vector &timings) +{ + double mean = 0; + for (size_t i = 0; i < timings.size(); i++) + mean += (double)timings[i] / (double)timings.size(); + double variance = 0; + for (size_t i = 0; i < timings.size(); i++) + variance += ((double)timings[i] - mean) * ((double)timings[i] - mean) / (double)timings.size(); + double sd = std::sqrt(variance); + return std::pair(mean, sd); +} + +/** + * @brief Perform a performance test using some matrix operations. + */ +void check_performance() +{ + // Set the formatting of floating point numbers. + dual_print(std::fixed, std::setprecision(1)); + + // Initialize a random_matrix_generator object to generates matrices with real (floating-point) numbers uniformly distributed between -1000 and 1000. + random_matrix_generator> rnd(-1000, 1000); + + // Initialize a timer object to measure the execution time of various operations. + timer tmr; + + // If the CPU has more than 8 threads, we leave 2 threads for the rest of the operating system. Otherwise, performance may suffer. + const ui32 thread_count = pool.get_thread_count() <= 8 ? pool.get_thread_count() : pool.get_thread_count() - 2; + dual_println("Using ", thread_count, " out of ", pool.get_thread_count(), " threads."); + + // Define the size of the matrices to use. + const ui64 rows = thread_count * 200; + const ui64 cols = rows; + + // The number of tasks to try for each operation. + const ui32 try_tasks[] = {1, thread_count / 4, thread_count / 2, thread_count, thread_count * 2, thread_count * 4}; + + // Generate two random test matrices to be used for benchmarking addition, transposition, and random matrix generation. + matrix A = rnd.generate_matrix(rows, cols, thread_count); + matrix B = rnd.generate_matrix(rows, cols, thread_count); + + // Generate two random test matrices to be used for benchmarking multiplication. Since matrix multiplication is O(n^3), we reduce the size of the test matrices so that this operation completes within a reasonable time. + constexpr ui64 mult_factor = 8; + matrix X = rnd.generate_matrix(rows / mult_factor, cols / mult_factor, thread_count); + matrix Y = rnd.generate_matrix(cols / mult_factor, rows / mult_factor, thread_count); + + // Determine the optimal sleep duration for this system. + dual_print("Determining the optimal sleep duration..."); + i64 optimal_ms = 0; + ui64 optimal_sleep = 0; + for (ui64 sleep = 0; sleep <= 2000; sleep += 100) + { + dual_print("."); + pool.sleep_duration = (ui32)sleep; + tmr.start(); + matrix C = add_matrices(A, B, thread_count); + matrix D = A.transpose(thread_count); + matrix E = multiply_matrices(X, Y, thread_count); + matrix F = rnd.generate_matrix(rows, cols, thread_count); + tmr.stop(); + if (tmr.ms() < optimal_ms || optimal_ms == 0) + { + optimal_ms = tmr.ms(); + optimal_sleep = sleep; + } + } + if (optimal_sleep == 0) + dual_println("\nResult: Using std::this_thread::yield() instead of std::this_thread::sleep_for() is optimal."); + else + dual_println("\nResult: The optimal sleep duration is ", optimal_sleep, " microseconds."); + pool.sleep_duration = (ui32)optimal_sleep; + + // Vectors to store statistics. + std::vector different_n_timings; + std::vector same_n_timings; + std::vector speedups; + + // How many times to run each test. + constexpr ui32 repeat = 20; + + dual_println("\nAdding two ", rows, "x", cols, " matrices ", repeat, " times:"); + for (ui32 n : try_tasks) + { + for (ui32 i = 0; i < repeat; i++) + { + tmr.start(); + matrix C = add_matrices(A, B, n); + tmr.stop(); + same_n_timings.push_back(tmr.ms()); + } + auto mean_sd = analyze(same_n_timings); + print_timing(n, mean_sd); + different_n_timings.push_back(mean_sd.first); + same_n_timings.clear(); + } + speedups.push_back(print_speedup(different_n_timings)); + different_n_timings.clear(); + + dual_println("\nTransposing one ", rows, "x", cols, " matrix ", repeat, " times:"); + for (ui32 n : try_tasks) + { + for (ui32 i = 0; i < repeat; i++) + { + tmr.start(); + matrix C = A.transpose(n); + tmr.stop(); + same_n_timings.push_back(tmr.ms()); + } + auto mean_sd = analyze(same_n_timings); + print_timing(n, mean_sd); + different_n_timings.push_back(mean_sd.first); + same_n_timings.clear(); + } + speedups.push_back(print_speedup(different_n_timings)); + different_n_timings.clear(); + + dual_println("\nMultiplying two ", rows / mult_factor, "x", cols / mult_factor, " matrices ", repeat, " times:"); + for (ui32 n : try_tasks) + { + for (ui32 i = 0; i < repeat; i++) + { + tmr.start(); + matrix C = multiply_matrices(X, Y, n); + tmr.stop(); + same_n_timings.push_back(tmr.ms()); + } + auto mean_sd = analyze(same_n_timings); + print_timing(n, mean_sd); + different_n_timings.push_back(mean_sd.first); + same_n_timings.clear(); + } + speedups.push_back(print_speedup(different_n_timings)); + different_n_timings.clear(); + + dual_println("\nGenerating random ", rows, "x", cols, " matrix ", repeat, " times:"); + for (ui32 n : try_tasks) + { + for (ui32 i = 0; i < repeat; i++) + { + tmr.start(); + matrix C = rnd.generate_matrix(rows, cols, n); + tmr.stop(); + same_n_timings.push_back(tmr.ms()); + } + auto mean_sd = analyze(same_n_timings); + print_timing(n, mean_sd); + different_n_timings.push_back(mean_sd.first); + same_n_timings.clear(); + } + speedups.push_back(print_speedup(different_n_timings)); + + const double max_speedup = *std::max_element(std::begin(speedups), std::end(speedups)); + dual_println("\nOverall, multithreading provided speedups of up to ", max_speedup, "x."); +} + +int main() +{ + std::string log_filename = "thread_pool_test-" + get_time() + ".log"; + log_file.open(log_filename); + + dual_println("A C++17 Thread Pool for High-Performance Scientific Computing"); + dual_println("(c) 2021 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 ", THREAD_POOL_VERSION, "."); + dual_println("Hardware concurrency is ", std::thread::hardware_concurrency(), "."); + dual_println("Generating log file: ", log_filename, ".\n"); + + dual_println("Important: Please do not run any other applications, especially multithreaded applications, in parallel with this test!"); + + print_header("Checking that the constructor works:"); + check_constructor(); + + print_header("Checking that reset() works:"); + check_reset(); + + print_header("Checking that push_task() works:"); + check_push_task(); + + print_header("Checking that submit() works:"); + check_submit(); + + print_header("Checking that wait_for_tasks() works..."); + check_wait_for_tasks(); + + print_header("Checking that parallelize_loop() works:"); + check_parallelize_loop(); + + print_header("Checking that different values of sleep_duration work:"); + check_sleep_duration(); + + print_header("Checking that task monitoring works:"); + check_task_monitoring(); + + print_header("Checking that pausing works:"); + check_pausing(); + + print_header("Checking that exception handling works:"); + check_exceptions(); + + print_header("Testing that matrix operations produce the expected results:"); + check_matrix(); + + if (tests_failed == 0) + { + print_header("SUCCESS: Passed all " + std::to_string(tests_succeeded) + " checks!", '+'); + print_header("Performing matrix performance test:"); + check_performance(); + print_header("Thread pool performance test completed!", '+'); + } + 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 0; +}