mirror of
https://github.com/bshoshany/thread-pool.git
synced 2026-07-22 11:33:00 +04:00
Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Barak Shoshany
|
f7284c56db | ||
|
Barak Shoshany
|
291ad92175 | ||
|
Barak Shoshany
|
910f7cd95b | ||
|
Barak Shoshany
|
bc66b8a222 | ||
|
Barak Shoshany
|
31efea058c | ||
|
Barak Shoshany
|
41637a8d1e |
@@ -0,0 +1,47 @@
|
||||
# A C++17 Thread Pool for High-Performance Scientific Computing
|
||||
|
||||
## Version history
|
||||
|
||||
* 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)
|
||||
* The version history has become too long to be included in `README.md`, so I moved it to a separate file, `CHANGELOG.md`.
|
||||
* A button to open this repository directly in Visual Studio Code has been added to the badges in `README.md`.
|
||||
* An internal variable named `promise` has been renamed to `task_promise` to avoid any potential errors in case the user invokes `using namespace std`.
|
||||
* `submit()` now catches exceptions thrown by the submitted task and forwards them to the future. See [this issue](https://github.com/bshoshany/thread-pool/issues/14).
|
||||
* Eliminated compiler warnings that appeared when using the `-Weffc++` flag in GCC. See [this pull request](https://github.com/bshoshany/thread-pool/pull/17).
|
||||
* v1.7 (2021-06-02)
|
||||
* Fixed a bug in `parallelize_loop()` which prevented it from actually running loops in parallel, see [this issue](https://github.com/bshoshany/thread-pool/issues/11).
|
||||
* v1.6 (2021-05-26)
|
||||
* Since MSVC does not interpret `and` as `&&` by default, the previous release did not compile with MSVC unless the `/permissive-` or `/Za` compiler flags were used. This has been fixed in this version, and the code now successfully compiles with GCC, Clang, and MSVC. See [this pull request](https://github.com/bshoshany/thread-pool/pull/10).
|
||||
* v1.5 (2021-05-07)
|
||||
* This library now has a DOI for citation purposes. Information on how to cite it in publications has been added to the source code and to `README.md`.
|
||||
* Added GitHub badges to `README.md`.
|
||||
* v1.4 (2021-05-05)
|
||||
* Added three new public member functions to monitor the tasks submitted to the pool:
|
||||
* `get_tasks_queued()` gets the number of tasks currently waiting in the queue to be executed by the threads.
|
||||
* `get_tasks_running()` gets the number of tasks currently being executed by the threads.
|
||||
* `get_tasks_total()` gets the total number of unfinished tasks - either still in the queue, or running in a thread.
|
||||
* Note that `get_tasks_running() == get_tasks_total() - get_tasks_queued()`.
|
||||
* Renamed the private member variable `tasks_waiting` to `tasks_total` to make its purpose clearer.
|
||||
* Added an option to temporarily pause the workers:
|
||||
* When public member variable `paused` is set to `true`, the workers temporarily stop popping new tasks out of the queue, although any tasks already executed will keep running until they are done. Set to `false` again to resume popping tasks.
|
||||
* While the workers are paused, `wait_for_tasks()` will wait for the running tasks instead of all tasks (otherwise it would wait forever).
|
||||
* By utilizing the new pausing mechanism, `reset()` can now change the number of threads on-the-fly while there are still tasks waiting in the queue. The new thread pool will resume executing tasks from the queue once it is created.
|
||||
* `parallelize_loop()` and `wait_for_tasks()` now have the same behavior as the worker function with regards to waiting for tasks to complete. If the relevant tasks are not yet complete, then before checking again, they will sleep for `sleep_duration` microseconds, unless that variable is set to zero, in which case they will call `std::this_thread::yield()`. This should improve performance and reduce CPU usage.
|
||||
* Merged [this commit](https://github.com/bshoshany/thread-pool/pull/8): Fixed weird error when using MSVC and including `windows.h`.
|
||||
* The `README.md` file has been reorganized and expanded.
|
||||
* v1.3 (2021-05-03)
|
||||
* Fixed [this issue](https://github.com/bshoshany/thread-pool/issues/3): Removed `std::move` from the `return` statement in `push_task()`. This previously generated a `-Wpessimizing-move` warning in Clang. The assembly code generated by the compiler seems to be the same before and after this change, presumably because the compiler eliminates the `std::move` automatically, but this change gets rid of the Clang warning.
|
||||
* Fixed [this issue](https://github.com/bshoshany/thread-pool/issues/5): Removed a debugging message printed to `std::cout`, which was left in the code by mistake.
|
||||
* Fixed [this issue](https://github.com/bshoshany/thread-pool/issues/6): `parallelize_loop()` no longer sends references for the variables `start` and `stop` when calling `push_task()`, which may lead to undefined behavior.
|
||||
* A companion paper is now published at <a href="https://arxiv.org/abs/2105.00613">arXiv:2105.00613</a>, including additional information such as performance tests on systems with up to 80 hardware threads. The `README.md` has been updated, and it is now roughly identical in content to the paper.
|
||||
* v1.2 (2021-04-29)
|
||||
* The worker function, which controls the execution of tasks by each thread, now sleeps by default instead of yielding. Previously, when the worker could not find any tasks in the queue, it called `std::this_thread::yield()` and then tried again. However, this caused the workers to have high CPU usage when idle, [as reported by some users](https://github.com/bshoshany/thread-pool/issues/1). Now, when the worker function cannot find a task to run, it instead sleeps for a duration given by the public member variable `sleep_duration` (in microseconds) before checking the queue again. The default value is `1000` microseconds, which I found to be optimal in terms of both CPU usage and performance, but your own optimal value may be different.
|
||||
* If the constructor is called with an argument of zero for the number of threads, then the default value, `std::thread::hardware_concurrency()`, is used instead.
|
||||
* Added a simple helper class, `timer`, which can be used to measure execution time for benchmarking purposes.
|
||||
* Improved and expanded the documentation.
|
||||
* v1.1 (2021-04-24)
|
||||
* Cosmetic changes only. Fixed a typo in the Doxygen comments and added a link to the GitHub repository.
|
||||
* v1.0 (2021-01-15)
|
||||
* Initial release.
|
||||
+21
-21
@@ -1,21 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2021 Barak Shoshany
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
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.
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2021 Barak Shoshany
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
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.
|
||||
|
||||
@@ -1,49 +1,55 @@
|
||||
[](https://doi.org/10.5281/zenodo.4742687)
|
||||
[](https://arxiv.org/abs/2105.00613)
|
||||
[](https://github.com/bshoshany/thread-pool/blob/master/LICENSE.txt)
|
||||
|
||||
|
||||
|
||||
[](https://github.com/bshoshany/thread-pool)
|
||||
[](https://twitter.com/BarakShoshany)
|
||||
[](https://open.vscode.dev/bshoshany/thread-pool)
|
||||
|
||||
# A C++17 Thread Pool for High-Performance Scientific Computing
|
||||
|
||||
**Barak Shoshany**\
|
||||
Department of Physics, Brock University,\
|
||||
1812 Sir Isaac Brock Way, St. Catharines, Ontario, L2S 3A1, Canada\
|
||||
[baraksh@gmail.com](mailto:baraksh@gmail.com) | [https://baraksh.com/](https://baraksh.com/)\
|
||||
Companion paper: [arXiv:2105.00613](https://arxiv.org/abs/2105.00613)
|
||||
[bshoshany@brocku.ca](mailto:bshoshany@brocku.ca) | [https://baraksh.com/](https://baraksh.com/)\
|
||||
Companion paper: [arXiv:2105.00613](https://arxiv.org/abs/2105.00613)\
|
||||
DOI: [doi:10.5281/zenodo.4742687](https://doi.org/10.5281/zenodo.4742687)
|
||||
|
||||
<!-- TOC depthFrom:2 -->
|
||||
* [Abstract](#abstract)
|
||||
* [Introduction](#introduction)
|
||||
* [Motivation](#motivation)
|
||||
* [Overview of features](#overview-of-features)
|
||||
* [Compiling and compatibility](#compiling-and-compatibility)
|
||||
* [Getting started](#getting-started)
|
||||
* [Including the library](#including-the-library)
|
||||
* [Constructors](#constructors)
|
||||
* [Getting and resetting the number of threads in the pool](#getting-and-resetting-the-number-of-threads-in-the-pool)
|
||||
* [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)
|
||||
* [Synchronizing printing to an output stream](#synchronizing-printing-to-an-output-stream)
|
||||
* [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)
|
||||
* [Dual Intel Xeon Gold 6148 (80 threads)](#dual-intel-xeon-gold-6148-80-threads)
|
||||
* [Feedback](#feedback)
|
||||
* [Copyright and citing](#copyright-and-citing)
|
||||
|
||||
- [Abstract](#abstract)
|
||||
- [Introduction](#introduction)
|
||||
- [Motivation](#motivation)
|
||||
- [Overview of features](#overview-of-features)
|
||||
- [Basic usage](#basic-usage)
|
||||
- [Including the library](#including-the-library)
|
||||
- [Constructors](#constructors)
|
||||
- [Submitting tasks to the queue](#submitting-tasks-to-the-queue)
|
||||
- [Parallelizing loops](#parallelizing-loops)
|
||||
- [Compiling and compatibility](#compiling-and-compatibility)
|
||||
- [Advanced usage](#advanced-usage)
|
||||
- [Getting and resetting the number of threads in the pool](#getting-and-resetting-the-number-of-threads-in-the-pool)
|
||||
- [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)
|
||||
- [Setting the worker function's sleep duration](#setting-the-worker-functions-sleep-duration)
|
||||
- [Optional helper classes](#optional-helper-classes)
|
||||
- [Synchronizing printing to an output stream](#synchronizing-printing-to-an-output-stream)
|
||||
- [Measuring execution time](#measuring-execution-time)
|
||||
- [Performance tests](#performance-tests)
|
||||
- [AMD Ryzen 9 3900X (24 threads)](#amd-ryzen-9-3900x-24-threads)
|
||||
- [Dual Intel Xeon Gold 6148 (80 threads)](#dual-intel-xeon-gold-6148-80-threads)
|
||||
- [Version history](#version-history)
|
||||
- [Feedback](#feedback)
|
||||
- [Author and copyright](#author-and-copyright)
|
||||
|
||||
<!-- /TOC -->
|
||||
|
||||
<a id="markdown-abstract" name="abstract"></a>
|
||||
## Abstract
|
||||
|
||||
We present 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, our 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. This paper provides motivation, detailed usage instructions, and performance tests. The code is freely available in the [GitHub repository](https://github.com/bshoshany/thread-pool). This `README.md` file contains roughly the same content as the [companion paper](https://arxiv.org/abs/2105.00613).
|
||||
|
||||
<a id="markdown-introduction" name="introduction"></a>
|
||||
## Introduction
|
||||
|
||||
<a id="markdown-motivation" name="motivation"></a>
|
||||
### Motivation
|
||||
|
||||
Multithreading is essential for modern high-performance computing. Since C++11, the C++ standard library has included built-in low-level multithreading support using constructs such as `std::thread`. However, `std::thread` creates a new thread each time it is called, which can have a significant performance overhead. Furthermore, it is possible to create more threads than the hardware can handle simultaneously, potentially resulting in a substantial slowdown.
|
||||
@@ -58,17 +64,16 @@ High-level multithreading APIs, such as OpenMP, allow simple one-line automatic
|
||||
|
||||
As demonstrated in the performance tests [below](#performance-tests), using our thread pool class we were able to saturate the upper bound of expected speedup for matrix multiplication and generation of random matrices. These performance tests were performed on 12-core / 24-thread and 40-core / 80-thread systems using GCC on Linux.
|
||||
|
||||
<a id="markdown-overview-of-features" name="overview-of-features"></a>
|
||||
### Overview of features
|
||||
|
||||
* **Fast:**
|
||||
* Built from scratch with maximum performance in mind.
|
||||
* Suitable for use in high-performance computing clusters with a very large number of CPU cores.
|
||||
* Suitable for use in high-performance computing nodes with a very large number of CPU cores.
|
||||
* Compact code, to reduce both compilation time and binary size.
|
||||
* Reusing threads avoids the overhead of creating and destroying them for individual tasks.
|
||||
* A task queue ensures that there are never more threads running in parallel than allowed by the hardware.
|
||||
* **Lightweight:**
|
||||
* Only ~160 lines of code, excluding comments, blank lines, and the two optional helper classes.
|
||||
* Only ~180 lines of code, excluding comments, blank lines, and the two optional helper classes.
|
||||
* Single header file: simply `#include "thread_pool.hpp"`.
|
||||
* Header-only: no need to install or build the library.
|
||||
* Self-contained: no external requirements or dependencies. Does not require OpenMP or any other multithreading APIs. Only uses the C++ standard library, and works with any C++17-compliant compiler.
|
||||
@@ -80,15 +85,32 @@ As demonstrated in the performance tests [below](#performance-tests), using our
|
||||
* **Additional features:**
|
||||
* Automatically parallelize a loop into any number of parallel tasks.
|
||||
* Easily wait for all tasks in the queue to complete.
|
||||
* Change the number of threads in the pool safely as needed.
|
||||
* Change the number of threads in the pool safely and on-the-fly as needed.
|
||||
* Fine-tune the sleep duration of each thread's worker function for optimal performance.
|
||||
* Monitor the number of queued and/or running tasks.
|
||||
* Pause and resume popping new tasks out of the queue.
|
||||
* 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.
|
||||
|
||||
<a id="markdown-basic-usage" name="basic-usage"></a>
|
||||
## Basic usage
|
||||
### 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:
|
||||
|
||||
* 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.
|
||||
|
||||
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
|
||||
|
||||
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.
|
||||
|
||||
## Getting started
|
||||
|
||||
<a id="markdown-including-the-library" name="including-the-library"></a>
|
||||
### Including the library
|
||||
|
||||
To use the thread pool library, simply include the header file:
|
||||
@@ -99,7 +121,6 @@ To use the thread pool library, simply include the header file:
|
||||
|
||||
The thread pool will now be accessible via the `thread_pool` class.
|
||||
|
||||
<a id="markdown-constructors" name="constructors"></a>
|
||||
### 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:
|
||||
@@ -120,8 +141,17 @@ If your program's main thread only submits tasks to the thread pool and waits fo
|
||||
|
||||
However, if your main thread does perform computationally intensive tasks on its own, then it is recommended to use the value `std::thread::hardware_concurrency() - 1` for the number of threads. In this case, the main thread plus the thread pool will together take up exactly all the threads available in the hardware.
|
||||
|
||||
<a id="markdown-submitting-tasks-to-the-queue" name="submitting-tasks-to-the-queue"></a>
|
||||
### Submitting tasks to the queue
|
||||
### Getting and resetting the number of threads in the pool
|
||||
|
||||
The member function `get_thread_count()` returns the number of threads in the pool. This will be equal to `std::thread::hardware_concurrency()` if the default constructor was used.
|
||||
|
||||
It is generally unnecessary to change the number of threads in the pool after it has been created, since the whole point of a thread pool is that you only create the threads once. However, if needed, this can be done, safely and on-the-fly, using the `reset()` member function.
|
||||
|
||||
`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.
|
||||
|
||||
## Submitting and waiting for tasks
|
||||
|
||||
### Submitting tasks to the queue with futures
|
||||
|
||||
A task can be any function, with zero or more arguments, and with or without a return value. Once a task has been submitted to the queue, it will be executed as soon as a thread becomes available. Tasks are executed in the order that they were submitted (first-in, first-out).
|
||||
|
||||
@@ -148,12 +178,47 @@ To wait until the future's value becomes available, use the member function `wai
|
||||
auto my_future = pool.submit(task);
|
||||
// Do some other stuff while the task is executing.
|
||||
do_stuff();
|
||||
// Get the task's return value from the future,
|
||||
// waiting for it to finish running if needed.
|
||||
// Get the task's return value from the future, waiting for it to finish running if needed.
|
||||
auto my_return_value = my_future.get();
|
||||
```
|
||||
|
||||
<a id="markdown-parallelizing-loops" name="parallelizing-loops"></a>
|
||||
### Submitting tasks to the queue without futures
|
||||
|
||||
Usually, it is best to submit a task to the queue using `submit()`. This allows you to wait for the task to finish and/or get its return value later. However, sometimes a future is not needed, for example when you just want to "set and forget" a certain task, or if the task already communicates with the main thread or with other tasks without using futures, such as via references or pointers. In such cases, you may wish to avoid the overhead involved in assigning a future to the task in order to increase performance.
|
||||
|
||||
The member function `push_task()` allows you to submit a task to the queue without generating a future for it. The task can have any number of arguments, but it cannot have a return value. For example:
|
||||
|
||||
```cpp
|
||||
// Submit a task without arguments or return value to the queue.
|
||||
pool.push_task(task);
|
||||
// Submit a task with one argument and no return value to the queue.
|
||||
pool.push_task(task, arg);
|
||||
// Submit a task with two arguments and no return value to the queue.
|
||||
pool.push_task(task, arg1, arg2);
|
||||
```
|
||||
|
||||
### Manually waiting for all tasks to complete
|
||||
|
||||
To wait for a **single** submitted task to complete, use `submit()` and then use the `wait()` or `get()` member functions of the obtained future. However, in cases where you need to wait until **all** submitted tasks finish their execution, or if the tasks have been submitted without futures using `push_task()`, you can use the member function `wait_for_tasks()`.
|
||||
|
||||
Consider, for example, the following code:
|
||||
|
||||
```cpp
|
||||
thread_pool pool;
|
||||
size_t a[100];
|
||||
for (size_t i = 0; i < 100; i++)
|
||||
pool.push_task([&a, i] { a[i] = i * i; });
|
||||
std::cout << a[50];
|
||||
```
|
||||
|
||||
The output will most likely be garbage, since the task that modifies `a[50]` has not yet finished executing by the time we try to access that element (in fact, that task is probably still waiting in the queue). One solution would be to use `submit()` instead of `push_task()`, but perhaps we don't want the overhead of generating 100 different futures. Instead, simply adding the line
|
||||
|
||||
```cpp
|
||||
pool.wait_for_tasks();
|
||||
```
|
||||
|
||||
after the `for` loop will ensure - as efficiently as possible - that all tasks have finished running before we attempt to access any elements of the array `a`, and the code will print out the value `2500` as expected. (Note, however, that `wait_for_tasks()` will wait for **all** the tasks in the queue, including those that are unrelated to the `for` loop. Using `parallelize_loop()` would make much more sense in this particular case, as it will wait only for the tasks related to the loop.)
|
||||
|
||||
### Parallelizing loops
|
||||
|
||||
Consider the following loop:
|
||||
@@ -188,89 +253,30 @@ pool.parallelize_loop(start, end, loop, 12);
|
||||
|
||||
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.
|
||||
|
||||
<a id="markdown-compiling-and-compatibility" name="compiling-and-compatibility"></a>
|
||||
### 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:
|
||||
|
||||
* 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.
|
||||
|
||||
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
|
||||
|
||||
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.
|
||||
|
||||
<a id="markdown-advanced-usage" name="advanced-usage"></a>
|
||||
## Advanced usage
|
||||
|
||||
<a id="markdown-getting-and-resetting-the-number-of-threads-in-the-pool" name="getting-and-resetting-the-number-of-threads-in-the-pool"></a>
|
||||
### Getting and resetting the number of threads in the pool
|
||||
|
||||
The member function `get_thread_count()` returns the number of threads in the pool. This will be equal to `std::thread::hardware_concurrency()` if the default constructor was used.
|
||||
|
||||
It is generally unnecessary to change the number of threads in the pool after it has been created, since the whole point of a thread pool is that you only create the threads once. However, if needed, this can be done - safely - using the `reset()` member function, which waits for all submitted tasks to be completed, then destroys all of the threads and creates a new thread pool with the desired new number of threads, as specified in the function's argument. If no argument is given to `reset()`, the new number of threads will be the hardware concurrency.
|
||||
|
||||
<a id="markdown-submitting-tasks-to-the-queue-without-futures" name="submitting-tasks-to-the-queue-without-futures"></a>
|
||||
### Submitting tasks to the queue without futures
|
||||
|
||||
Usually, it is best to submit a task to the queue using `submit()`. This allows you to wait for the task to finish and/or get its return value later. However, sometimes a future is not needed, for example when you just want to "set and forget" a certain task, or if the task already communicates with the main thread or with other tasks without using futures, such as via references or pointers. In such cases, you may wish to avoid the overhead involved in assigning a future to the task in order to increase performance.
|
||||
|
||||
The member function `push_task()` allows you to submit a task to the queue without generating a future for it. The task can have any number of arguments, but it cannot have a return value. For example:
|
||||
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:
|
||||
|
||||
```cpp
|
||||
// Submit a task without arguments or return value to the queue.
|
||||
pool.push_task(task);
|
||||
// Submit a task with one argument and no return value to the queue.
|
||||
pool.push_task(task, arg);
|
||||
// Submit a task with two arguments and no return value to the queue.
|
||||
pool.push_task(task, arg1, arg2);
|
||||
#include "thread_pool.hpp"
|
||||
|
||||
int main()
|
||||
{
|
||||
thread_pool pool(10);
|
||||
size_t squares[100];
|
||||
pool.parallelize_loop(0, 99, [&squares](size_t i) { squares[i] = i * i; });
|
||||
std::cout << "16^2 = " << squares[16] << '\n';
|
||||
std::cout << "32^2 = " << squares[32] << '\n';
|
||||
}
|
||||
```
|
||||
|
||||
<a id="markdown-manually-waiting-for-all-tasks-to-complete" name="manually-waiting-for-all-tasks-to-complete"></a>
|
||||
### Manually waiting for all tasks to complete
|
||||
The output should be:
|
||||
|
||||
To wait for a **single** submitted task to complete, use `submit()` and then use the `wait()` or `get()` member functions of the obtained future. However, in cases where you need to wait until **all** submitted tasks finish their execution, or if the tasks have been submitted without futures using `push_task()`, you can use the member function `wait_for_tasks()`.
|
||||
|
||||
Consider, for example, the following code:
|
||||
|
||||
```cpp
|
||||
thread_pool pool;
|
||||
size_t a[100];
|
||||
for (size_t i = 0; i < 100; i++)
|
||||
pool.push_task([&a, i] { a[i] = i * i; });
|
||||
std::cout << a[50];
|
||||
```none
|
||||
16^2 = 256
|
||||
32^2 = 1024
|
||||
```
|
||||
|
||||
The output will most likely be garbage, since the task that modifies `a[50]` has not yet finished executing by the time we try to access that element (in fact, that task is probably still waiting in the queue). One solution would be to use `submit()` instead of `push_task()`, but perhaps we don't want the overhead of generating 100 different futures. Instead, simply adding the line
|
||||
## Other features
|
||||
|
||||
```cpp
|
||||
pool.wait_for_tasks();
|
||||
```
|
||||
|
||||
after the `for` loop will ensure - as efficiently as possible - that all tasks have finished running before we attempt to access any elements of the array `a`, and the code will print out the value `2500` as expected. (Note, however, that `wait_for_tasks()` will wait for **all** the tasks in the queue, including those that are unrelated to the `for` loop. Using `parallelize_loop()` would make much more sense in this particular case, as it will wait only for the tasks related to the loop.)
|
||||
|
||||
<a id="markdown-setting-the-worker-functions-sleep-duration" name="setting-the-worker-functions-sleep-duration"></a>
|
||||
### 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.
|
||||
|
||||
The default value of `sleep_duration` is `1000` microseconds, or `1` millisecond. In our benchmarks, lower values resulted in high CPU usage when the workers were idle. The value of `1000` microseconds was roughly the minimum value needed to reduce the idle CPU usage to a negligible amount.
|
||||
|
||||
In addition, in our benchmarks this value resulted in moderately improved performance compared to lower values, since the workers check the queue - which is a costly process - less frequently. On the other hand, increasing the value even more could potentially cause the workers to spend too much time sleeping and not pick up tasks from the queue quickly enough, so `1000` is the "sweet spot".
|
||||
|
||||
However, please note that this value is likely unique to the particular system our benchmarks were performed on, and your own optimal value would depend on factors such as your OS and C++ implementation, the type, complexity, and average duration of the tasks submitted to the pool, and whether there are any other programs running at the same time. Therefore, it is strongly recommended to do your own benchmarks and find the value that works best for you.
|
||||
|
||||
If `sleep_duration` is set to `0`, then the worker function will execute `std::this_thread::yield()` instead of sleeping if there are no tasks in the queue. This will suggest to the OS that it should put this thread on hold and allow other threads to run instead. However, this also causes the worker functions to have high CPU usage when idle. On the other hand, for some applications this setting may provide better performance than sleeping - again, do your own benchmarks and find what works best for you.
|
||||
|
||||
<a id="markdown-optional-helper-classes" name="optional-helper-classes"></a>
|
||||
## Optional helper classes
|
||||
|
||||
<a id="markdown-synchronizing-printing-to-an-output-stream" name="synchronizing-printing-to-an-output-stream"></a>
|
||||
### Synchronizing printing to an output stream
|
||||
|
||||
When printing to an output stream from multiple threads in parallel, the output may become garbled. For example, consider this code:
|
||||
@@ -327,7 +333,267 @@ Task no. 4 executing.
|
||||
Task no. 5 executing.
|
||||
```
|
||||
|
||||
<a id="markdown-measuring-execution-time" name="measuring-execution-time"></a>
|
||||
**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.
|
||||
|
||||
### 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.
|
||||
|
||||
The default value of `sleep_duration` is `1000` microseconds, or `1` millisecond. In our benchmarks, lower values resulted in high CPU usage when the workers were idle. The value of `1000` microseconds was roughly the minimum value needed to reduce the idle CPU usage to a negligible amount.
|
||||
|
||||
In addition, in our benchmarks this value resulted in moderately improved performance compared to lower values, since the workers check the queue - which is a costly process - less frequently. On the other hand, increasing the value even more could potentially cause the workers to spend too much time sleeping and not pick up tasks from the queue quickly enough, so `1000` is the "sweet spot".
|
||||
|
||||
However, please note that this value is likely unique to the particular system our benchmarks were performed on, and your own optimal value would depend on factors such as your OS and C++ implementation, the type, complexity, and average duration of the tasks submitted to the pool, and whether there are any other programs running at the same time. Therefore, it is strongly recommended to do your own benchmarks and find the value that works best for you.
|
||||
|
||||
If `sleep_duration` is set to `0`, then the worker function will execute `std::this_thread::yield()` instead of sleeping if there are no tasks in the queue. This will suggest to the OS that it should put this thread on hold and allow other threads to run instead. However, this also causes the worker functions to have high CPU usage when idle. On the other hand, for some applications this setting may provide better performance than sleeping - again, do your own benchmarks and find what works best for you.
|
||||
|
||||
### Monitoring the tasks
|
||||
|
||||
Sometimes you may wish to monitor what is happening with the tasks you submitted to the pool. This may be done using three member functions:
|
||||
|
||||
* `get_tasks_queued()` gets the number of tasks currently waiting in the queue to be executed by the threads.
|
||||
* `get_tasks_running()` gets the number of tasks currently being executed by the threads.
|
||||
* `get_tasks_total()` gets the total number of unfinished tasks - either still in the queue, or running in a thread.
|
||||
* Note that `get_tasks_running() == get_tasks_total() - get_tasks_queued()`.
|
||||
|
||||
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)
|
||||
{
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||
sync_out->println("Task ", i, " done.");
|
||||
}
|
||||
|
||||
void monitor_tasks(const thread_pool *pool, synced_stream *sync_out)
|
||||
{
|
||||
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);
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(750));
|
||||
monitor_tasks(&pool, &sync_out);
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||
monitor_tasks(&pool, &sync_out);
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||
monitor_tasks(&pool, &sync_out);
|
||||
}
|
||||
```
|
||||
|
||||
Assuming you have at least 4 hardware threads (so that 4 tasks can run concurrently), the output will be similar to:
|
||||
|
||||
```none
|
||||
12 tasks total, 0 tasks running, 12 tasks queued.
|
||||
Task 0 done.
|
||||
Task 1 done.
|
||||
Task 2 done.
|
||||
Task 3 done.
|
||||
8 tasks total, 4 tasks running, 4 tasks queued.
|
||||
Task 4 done.
|
||||
Task 5 done.
|
||||
Task 6 done.
|
||||
Task 7 done.
|
||||
4 tasks total, 4 tasks running, 0 tasks queued.
|
||||
Task 8 done.
|
||||
Task 9 done.
|
||||
Task 10 done.
|
||||
Task 11 done.
|
||||
0 tasks total, 0 tasks running, 0 tasks queued.
|
||||
```
|
||||
|
||||
### Pausing the workers
|
||||
|
||||
Sometimes you may wish to temporarily pause the execution of tasks, or perhaps you want to submit tasks to the queue but only start executing them at a later time. You can do this using the public member variable `paused`.
|
||||
|
||||
When `paused` is set to `true`, the workers will temporarily stop popping new tasks out of the queue. However, any tasks already executed will keep running until they are done, since the thread pool has no control over the internal code of your tasks. If you need to pause a task in the middle of its execution, you must do that manually by programming your own pause mechanism into the task itself. To resume popping tasks, set `paused` back to its default value of `false`.
|
||||
|
||||
Here is an example:
|
||||
|
||||
```cpp
|
||||
#include "thread_pool.hpp"
|
||||
|
||||
void sleep_half_second(const size_t &i, synced_stream *sync_out)
|
||||
{
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||
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);
|
||||
sync_out.println("Submitted 8 tasks.");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(250));
|
||||
pool.paused = true;
|
||||
sync_out.println("Pool paused.");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||
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);
|
||||
sync_out.println("Submitted 4 more tasks.");
|
||||
sync_out.println("Still paused...");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||
pool.paused = false;
|
||||
sync_out.println("Pool resumed.");
|
||||
}
|
||||
```
|
||||
|
||||
Assuming you have at least 4 hardware threads, the output will be similar to:
|
||||
|
||||
```none
|
||||
Submitted 8 tasks.
|
||||
Pool paused.
|
||||
Task 0 done.
|
||||
Task 1 done.
|
||||
Task 2 done.
|
||||
Task 3 done.
|
||||
Still paused...
|
||||
Submitted 4 more tasks.
|
||||
Still paused...
|
||||
Pool resumed.
|
||||
Task 4 done.
|
||||
Task 5 done.
|
||||
Task 6 done.
|
||||
Task 7 done.
|
||||
Task 8 done.
|
||||
Task 9 done.
|
||||
Task 10 done.
|
||||
Task 11 done.
|
||||
```
|
||||
|
||||
Here is what happened. We initially submitted a total of 8 tasks to the queue. Since we waited for 250ms before pausing, the first 4 tasks have already started running, so they kept running until they finished. While the pool was paused, we submitted 4 more tasks to the queue, but they just waited at the end of the queue. When we resumed, the remaining 4 initial tasks were executed, followed by the 4 new tasks.
|
||||
|
||||
While the workers are paused, `wait_for_tasks()` will wait for the running tasks instead of all tasks (otherwise it would wait forever). This is demonstrated by the following program:
|
||||
|
||||
```cpp
|
||||
#include "thread_pool.hpp"
|
||||
|
||||
void sleep_half_second(const size_t &i, synced_stream *sync_out)
|
||||
{
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||
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);
|
||||
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);
|
||||
sync_out.println("Submitted 12 more tasks.");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(250));
|
||||
pool.paused = true;
|
||||
sync_out.println("Pool paused. Waiting for the ", pool.get_tasks_running(), " running tasks to complete.");
|
||||
pool.wait_for_tasks();
|
||||
sync_out.println("All running tasks completed. ", pool.get_tasks_queued(), " tasks still queued.");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||
sync_out.println("Still paused...");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||
sync_out.println("Still paused...");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||
pool.paused = false;
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(250));
|
||||
sync_out.println("Pool resumed. Waiting for the remaining ", pool.get_tasks_total(), " tasks (", pool.get_tasks_running(), " running and ", pool.get_tasks_queued(), " queued) to complete.");
|
||||
pool.wait_for_tasks();
|
||||
sync_out.println("All tasks completed.");
|
||||
}
|
||||
```
|
||||
|
||||
The output should be similar to:
|
||||
|
||||
```none
|
||||
Submitted 8 tasks. Waiting for them to complete.
|
||||
Task 0 done.
|
||||
Task 1 done.
|
||||
Task 2 done.
|
||||
Task 3 done.
|
||||
Task 4 done.
|
||||
Task 5 done.
|
||||
Task 6 done.
|
||||
Task 7 done.
|
||||
Submitted 12 more tasks.
|
||||
Pool paused. Waiting for the 4 running tasks to complete.
|
||||
Task 8 done.
|
||||
Task 9 done.
|
||||
Task 10 done.
|
||||
Task 11 done.
|
||||
All running tasks completed. 8 tasks still queued.
|
||||
Still paused...
|
||||
Still paused...
|
||||
Pool resumed. Waiting for the remaining 8 tasks (4 running and 4 queued) to complete.
|
||||
Task 12 done.
|
||||
Task 13 done.
|
||||
Task 14 done.
|
||||
Task 15 done.
|
||||
Task 16 done.
|
||||
Task 17 done.
|
||||
Task 18 done.
|
||||
Task 19 done.
|
||||
```
|
||||
|
||||
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
|
||||
|
||||
`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:
|
||||
|
||||
```cpp
|
||||
#include "thread_pool.hpp"
|
||||
|
||||
double inverse(const double &x)
|
||||
{
|
||||
if (x == 0)
|
||||
throw std::runtime_error("Division by zero!");
|
||||
else
|
||||
return 1 / x;
|
||||
}
|
||||
|
||||
int main()
|
||||
{
|
||||
thread_pool pool;
|
||||
auto my_future = pool.submit(inverse, 0);
|
||||
try
|
||||
{
|
||||
double result = my_future.get();
|
||||
std::cout << "The result is: " << result << '\n';
|
||||
}
|
||||
catch (const std::exception &e)
|
||||
{
|
||||
std::cout << "Caught exception: " << e.what() << '\n';
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The output will be:
|
||||
|
||||
```none
|
||||
Caught exception: Division by zero!
|
||||
```
|
||||
|
||||
## Performance tests
|
||||
|
||||
### 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 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.
|
||||
@@ -349,14 +615,10 @@ tmr.stop();
|
||||
std::cout << "The elapsed time was " << tmr.ms() << " ms.\n";
|
||||
```
|
||||
|
||||
<a id="markdown-performance-tests" name="performance-tests"></a>
|
||||
## Performance tests
|
||||
|
||||
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 (which is the most recent version at the time of writing) 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.
|
||||
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.
|
||||
|
||||
<a id="markdown-amd-ryzen-9-3900x-24-threads" name="amd-ryzen-9-3900x-24-threads"></a>
|
||||
### 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.
|
||||
@@ -404,7 +666,6 @@ In this test, we find a speedup by roughly a factor of 2 for addition, 9 for tra
|
||||
* 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).
|
||||
|
||||
<a id="markdown-dual-intel-xeon-gold-6148-80-threads" name="dual-intel-xeon-gold-6148-80-threads"></a>
|
||||
### 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.
|
||||
@@ -449,32 +710,32 @@ In this test, we find a speedup by roughly a factor of 10 for addition, 19 for t
|
||||
|
||||
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.
|
||||
|
||||
<a id="markdown-version-history" name="version-history"></a>
|
||||
## Version history
|
||||
|
||||
* Version 1.3 (2021-05-03)
|
||||
* Fixed [this issue](https://github.com/bshoshany/thread-pool/issues/3): Removed `std::move` from the `return` statement in `push_task()`. This previously generated a `-Wpessimizing-move` warning in Clang. The assembly code generated by the compiler seems to be the same before and after this change, presumably because the compiler eliminates the `std::move` automatically, but this change gets rid of the Clang warning.
|
||||
* Fixed [this issue](https://github.com/bshoshany/thread-pool/issues/5): Removed a debugging message printed to `std::cout`, which was left in the code by mistake.
|
||||
* Fixed [this issue](https://github.com/bshoshany/thread-pool/issues/6): `parallelize_loop()` no longer sends references for the variables `start` and `stop` when calling `push_task()`, which may lead to undefined behavior.
|
||||
* A companion paper is now published at <a href="https://arxiv.org/abs/2105.00613">arXiv:2105.00613</a>, including additional information such as performance tests on systems with up to 80 hardware threads. The `README.md` has been updated, and it is now roughly identical in content to the paper.
|
||||
* Version 1.2 (2021-04-29)
|
||||
* The worker function, which controls the execution of tasks by each thread, now sleeps by default instead of yielding. Previously, when the worker could not find any tasks in the queue, it called `std::this_thread::yield()` and then tried again. However, this caused the workers to have high CPU usage when idle, [as reported by some users](https://github.com/bshoshany/thread-pool/issues/1). Now, when the worker function cannot find a task to run, it instead sleeps for a duration given by the public member variable `sleep_duration` (in microseconds) before checking the queue again. The default value is `1000` microseconds, which I found to be optimal in terms of both CPU usage and performance, but your own optimal value may be different.
|
||||
* If the constructor is called with an argument of zero for the number of threads, then the default value, `std::thread::hardware_concurrency()`, is used instead.
|
||||
* Added a simple helper class, `timer`, which can be used to measure execution time for benchmarking purposes.
|
||||
* Improved and expanded the documentation.
|
||||
* Version 1.1 (2021-04-24)
|
||||
* Cosmetic changes only. Fixed a typo in the Doxygen comments and added a link to the GitHub repository.
|
||||
* Version 1.0 (2021-01-15)
|
||||
* Initial release.
|
||||
|
||||
<a id="markdown-feedback" name="feedback"></a>
|
||||
## Feedback
|
||||
|
||||
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)!
|
||||
|
||||
<a id="markdown-author-and-copyright" name="author-and-copyright"></a>
|
||||
## Author and copyright
|
||||
## Copyright and citing
|
||||
|
||||
Copyright (c) 2021 [Barak Shoshany](http://baraksh.com). Licensed under the [MIT license](LICENSE.txt). If you found this code useful, please consider providing a link to the [GitHub repository](https://github.com/bshoshany/thread-pool) and/or citing the [companion paper](https://arxiv.org/abs/2105.00613):
|
||||
Copyright (c) 2021 [Barak Shoshany](http://baraksh.com). Licensed under the [MIT license](LICENSE.txt).
|
||||
|
||||
* Barak Shoshany, "A C++17 Thread Pool for High-Performance Scientific Computing", [arXiv:2105.00613](https://arxiv.org/abs/2105.00613) (May 2021)
|
||||
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](https://doi.org/10.5281/zenodo.4742687), [arXiv:2105.00613](https://arxiv.org/abs/2105.00613) (May 2021)
|
||||
|
||||
You can use the following BibTeX entry:
|
||||
|
||||
```none
|
||||
@article{Shoshany2021_ThreadPool,
|
||||
archiveprefix = {arXiv},
|
||||
author = {Barak Shoshany},
|
||||
doi = {10.5281/zenodo.4742687},
|
||||
eid = {arXiv:2105.00613},
|
||||
eprint = {2105.00613},
|
||||
journal = {arXiv e-prints},
|
||||
keywords = {Computer Science - Distributed, Parallel, and Cluster Computing, D.1.3, D.1.5},
|
||||
month = {May},
|
||||
primaryclass = {cs.DC},
|
||||
title = {{A C++17 Thread Pool for High-Performance Scientific Computing}},
|
||||
year = {2021}
|
||||
}
|
||||
```
|
||||
|
||||
+533
-439
@@ -1,439 +1,533 @@
|
||||
#pragma once
|
||||
|
||||
/**
|
||||
* @file thread_pool.hpp
|
||||
* @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
|
||||
* @version 1.3
|
||||
* @date 2021-05-03
|
||||
* @copyright Copyright (c) 2021 Barak Shoshany. Licensed under the MIT license. If you found this code useful, please consider providing a link to the GitHub repository: https://github.com/bshoshany/thread-pool and/or citing the companion paper: https://arxiv.org/abs/2105.00613
|
||||
*
|
||||
* @brief A C++17 thread pool for high-performance scientific computing.
|
||||
* @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 for documentation and updates, or to submit feature requests and bug reports.
|
||||
*/
|
||||
|
||||
#include <algorithm> // std::max
|
||||
#include <atomic> // std::atomic
|
||||
#include <chrono> // std::chrono
|
||||
#include <cstdint> // std::int_fast64_t, std::uint_fast32_t
|
||||
#include <functional> // std::function
|
||||
#include <future> // std::future, std::promise
|
||||
#include <iostream> // std::cout, std::ostream
|
||||
#include <memory> // std::shared_ptr, std::unique_ptr
|
||||
#include <mutex> // std::mutex, std::scoped_lock
|
||||
#include <queue> // std::queue
|
||||
#include <thread> // std::this_thread, std::thread
|
||||
#include <type_traits> // std::decay_t, std::enable_if_t, std::is_void_v, std::invoke_result_t
|
||||
#include <utility> // std::move, std::swap
|
||||
|
||||
// ================================== Begin class thread_pool ================================== //
|
||||
|
||||
/**
|
||||
* @brief A simple but powerful thread pool class. The user submits tasks to be executed into a queue. Whenever a thread becomes available, it pops a task from the queue and executes it. Each task is automatically assigned a future, which can be used to wait for the task to finish executing and/or obtain its eventual return value.
|
||||
*/
|
||||
class thread_pool
|
||||
{
|
||||
typedef std::uint_fast32_t ui32;
|
||||
|
||||
public:
|
||||
// ============================
|
||||
// Constructors and destructors
|
||||
// ============================
|
||||
|
||||
/**
|
||||
* @brief Construct a new thread pool.
|
||||
*
|
||||
* @param _thread_count The number of threads to use. The default value is the total number of hardware threads available, as reported by the implementation. With a hyperthreaded CPU, this will be twice the number of CPU cores. If the argument is zero, the default value will be used instead.
|
||||
*/
|
||||
thread_pool(const ui32 &_thread_count = std::thread::hardware_concurrency())
|
||||
: thread_count(_thread_count ? _thread_count : std::thread::hardware_concurrency()), threads(new std::thread[_thread_count ? _thread_count : std::thread::hardware_concurrency()])
|
||||
{
|
||||
create_threads();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Destruct the thread pool. Waits for all submitted tasks to be completed, then destroys all threads.
|
||||
*/
|
||||
~thread_pool()
|
||||
{
|
||||
wait_for_tasks();
|
||||
running = false;
|
||||
destroy_threads();
|
||||
}
|
||||
|
||||
// =======================
|
||||
// Public member functions
|
||||
// =======================
|
||||
|
||||
/**
|
||||
* @brief Get the number of threads in the pool.
|
||||
*
|
||||
* @return The number of threads.
|
||||
*/
|
||||
ui32 get_thread_count() const
|
||||
{
|
||||
return thread_count;
|
||||
}
|
||||
|
||||
/**
|
||||
* @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);
|
||||
*
|
||||
* @tparam T The type of the loop index. Should be a signed or unsigned integer.
|
||||
* @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.
|
||||
*/
|
||||
template <typename T, typename F>
|
||||
void parallelize_loop(T first_index, T last_index, const F &loop, ui32 num_tasks = 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;
|
||||
if (block_size == 0)
|
||||
{
|
||||
block_size = 1;
|
||||
num_tasks = std::max((ui32)1, (ui32)total_size);
|
||||
}
|
||||
std::atomic<ui32> blocks_running = 0;
|
||||
for (ui32 t = 0; t < num_tasks; 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);
|
||||
blocks_running++;
|
||||
push_task([start, end, &loop, &blocks_running]
|
||||
{
|
||||
for (T i = start; i <= end; i++)
|
||||
loop(i);
|
||||
blocks_running--;
|
||||
});
|
||||
while (blocks_running != 0)
|
||||
{
|
||||
std::this_thread::yield();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Push a function with no arguments or return value into the task queue.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @param task The function to push.
|
||||
*/
|
||||
template <typename F>
|
||||
void push_task(const F &task)
|
||||
{
|
||||
tasks_waiting++;
|
||||
{
|
||||
const std::scoped_lock lock(queue_mutex);
|
||||
tasks.push(std::function<void()>(task));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Push a function with arguments, but no return value, into the task queue.
|
||||
* @details The function is wrapped inside a lambda in order to hide the arguments, as the tasks in the queue must be of type std::function<void()>, so they cannot have any arguments or return value. If no arguments are provided, the other overload will be used, in order to avoid the (slight) overhead of using a lambda.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @tparam A The types of the arguments.
|
||||
* @param task The function to push.
|
||||
* @param args The arguments to pass to the function.
|
||||
*/
|
||||
template <typename F, typename... A>
|
||||
void push_task(const F &task, const A &...args)
|
||||
{
|
||||
push_task([task, args...]
|
||||
{ task(args...); });
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Reset the number of threads in the pool. Waits for all submitted tasks to be completed, then destroys all threads and creates a new thread pool with the new number of threads.
|
||||
*
|
||||
* @param _thread_count The number of threads to use. The default value is the total number of hardware threads available, as reported by the implementation. With a hyperthreaded CPU, this will be twice the number of CPU cores. If the argument is zero, the default value will be used instead.
|
||||
*/
|
||||
void reset(const ui32 &_thread_count = std::thread::hardware_concurrency())
|
||||
{
|
||||
wait_for_tasks();
|
||||
running = false;
|
||||
destroy_threads();
|
||||
thread_count = _thread_count ? _thread_count : std::thread::hardware_concurrency();
|
||||
threads.reset(new std::thread[_thread_count ? _thread_count : std::thread::hardware_concurrency()]);
|
||||
running = true;
|
||||
create_threads();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Submit a function with zero or more arguments and no return value into the task queue, and get an std::future<bool> that will be set to true upon completion of the task.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @tparam A The types of the zero or more arguments to pass to the function.
|
||||
* @param task The function to submit.
|
||||
* @param args The zero or more arguments to pass to the function.
|
||||
* @return A future to be used later to check if the function has finished its execution.
|
||||
*/
|
||||
template <typename F, typename... A, typename = std::enable_if_t<std::is_void_v<std::invoke_result_t<std::decay_t<F>, std::decay_t<A>...>>>>
|
||||
std::future<bool> submit(const F &task, const A &...args)
|
||||
{
|
||||
std::shared_ptr<std::promise<bool>> promise(new std::promise<bool>);
|
||||
std::future<bool> future = promise->get_future();
|
||||
push_task([task, args..., promise]
|
||||
{
|
||||
task(args...);
|
||||
promise->set_value(true);
|
||||
});
|
||||
return future;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Submit a function with zero or more arguments and a return value into the task queue, and get a future for its eventual returned value.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @tparam A The types of the zero or more arguments to pass to the function.
|
||||
* @tparam R The return type of the function.
|
||||
* @param task The function to submit.
|
||||
* @param args The zero or more arguments to pass to the function.
|
||||
* @return A future to be used later to obtain the function's returned value, waiting for it to finish its execution if needed.
|
||||
*/
|
||||
template <typename F, typename... A, typename R = std::invoke_result_t<std::decay_t<F>, std::decay_t<A>...>, typename = std::enable_if_t<!std::is_void_v<R>>>
|
||||
std::future<R> submit(const F &task, const A &...args)
|
||||
{
|
||||
std::shared_ptr<std::promise<R>> promise(new std::promise<R>);
|
||||
std::future<R> future = promise->get_future();
|
||||
push_task([task, args..., promise]
|
||||
{ promise->set_value(task(args...)); });
|
||||
return future;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Wait for all submitted tasks to be completed - both those that are currently being executed by threads, and those that are still waiting in the queue. To wait for a specific task, use submit() instead, and call the wait() member function of the generated future.
|
||||
*/
|
||||
void wait_for_tasks()
|
||||
{
|
||||
while (tasks_waiting != 0)
|
||||
{
|
||||
std::this_thread::yield();
|
||||
}
|
||||
}
|
||||
|
||||
// ===========
|
||||
// Public data
|
||||
// ===========
|
||||
|
||||
/**
|
||||
* @brief The duration, in microseconds, that the worker function should sleep for when it cannot find any tasks in the queue. If set to 0, then instead of sleeping, the worker function will execute std::this_thread::yield() if there are no tasks in the queue. The default value is 1000.
|
||||
*/
|
||||
ui32 sleep_duration = 1000;
|
||||
|
||||
private:
|
||||
// ========================
|
||||
// Private member functions
|
||||
// ========================
|
||||
|
||||
/**
|
||||
* @brief Create the threads in the pool and assign a worker to each thread.
|
||||
*/
|
||||
void create_threads()
|
||||
{
|
||||
for (ui32 i = 0; i < thread_count; i++)
|
||||
{
|
||||
threads[i] = std::thread(&thread_pool::worker, this);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Destroy the threads in the pool by joining them.
|
||||
*/
|
||||
void destroy_threads()
|
||||
{
|
||||
for (ui32 i = 0; i < thread_count; i++)
|
||||
{
|
||||
threads[i].join();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Try to pop a new task out of the queue.
|
||||
*
|
||||
* @param task A reference to the task. Will be populated with a function if the queue is not empty.
|
||||
* @return true if a task was found, false if the queue is empty.
|
||||
*/
|
||||
bool pop_task(std::function<void()> &task)
|
||||
{
|
||||
const std::scoped_lock lock(queue_mutex);
|
||||
if (tasks.empty())
|
||||
return false;
|
||||
else
|
||||
{
|
||||
task = std::move(tasks.front());
|
||||
tasks.pop();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief A worker function to be assigned to each thread in the pool. Continuously pops tasks out of the queue and executes them, as long as the atomic variable running is set to true.
|
||||
*/
|
||||
void worker()
|
||||
{
|
||||
while (running)
|
||||
{
|
||||
std::function<void()> task;
|
||||
if (pop_task(task))
|
||||
{
|
||||
task();
|
||||
tasks_waiting--;
|
||||
}
|
||||
else
|
||||
{
|
||||
if (sleep_duration)
|
||||
std::this_thread::sleep_for(std::chrono::microseconds(sleep_duration));
|
||||
else
|
||||
std::this_thread::yield();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ============
|
||||
// Private data
|
||||
// ============
|
||||
|
||||
/**
|
||||
* @brief An atomic variable indicating to the workers to keep running.
|
||||
*/
|
||||
std::atomic<bool> running = true;
|
||||
|
||||
/**
|
||||
* @brief An atomic variable to keep track of how many tasks are currently waiting to finish - either still in the queue, or running in a thread.
|
||||
*/
|
||||
std::atomic<ui32> tasks_waiting = 0;
|
||||
|
||||
/**
|
||||
* @brief A mutex to synchronize access to the task queue by different threads.
|
||||
*/
|
||||
mutable std::mutex queue_mutex;
|
||||
|
||||
/**
|
||||
* @brief A queue of tasks to be executed by the threads.
|
||||
*/
|
||||
std::queue<std::function<void()>> tasks;
|
||||
|
||||
/**
|
||||
* @brief The number of threads in the pool.
|
||||
*/
|
||||
ui32 thread_count;
|
||||
|
||||
/**
|
||||
* @brief A smart pointer to manage the memory allocated for the threads.
|
||||
*/
|
||||
std::unique_ptr<std::thread[]> threads;
|
||||
};
|
||||
|
||||
// =================================== End class thread_pool =================================== //
|
||||
|
||||
// ================================= Begin class synced_stream ================================= //
|
||||
|
||||
/**
|
||||
* @brief A helper class to synchronize printing to an output stream by different threads.
|
||||
*/
|
||||
class synced_stream
|
||||
{
|
||||
public:
|
||||
/**
|
||||
* @brief Construct a new synced stream.
|
||||
*
|
||||
* @param _out_stream The output stream to print to. The default value is std::cout.
|
||||
*/
|
||||
synced_stream(std::ostream &_out_stream = std::cout)
|
||||
: out_stream(_out_stream){};
|
||||
|
||||
/**
|
||||
* @brief Print any number of items into the output stream. Ensures that no other threads print to this stream simultaneously, as long as they all exclusively use this synced_stream object to print.
|
||||
*
|
||||
* @tparam T The types of the items
|
||||
* @param items The items to print.
|
||||
*/
|
||||
template <typename... T>
|
||||
void print(const T &...items)
|
||||
{
|
||||
const std::scoped_lock lock(stream_mutex);
|
||||
(out_stream << ... << items);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Print any number of items into the output stream, followed by a newline character. Ensures that no other threads print to this stream simultaneously, as long as they all exclusively use this synced_stream object to print.
|
||||
*
|
||||
* @tparam T The types of the items
|
||||
* @param items The items to print.
|
||||
*/
|
||||
template <typename... T>
|
||||
void println(const T &...items)
|
||||
{
|
||||
print(items..., '\n');
|
||||
}
|
||||
|
||||
private:
|
||||
/**
|
||||
* @brief A mutex to synchronize printing.
|
||||
*/
|
||||
mutable std::mutex stream_mutex;
|
||||
|
||||
/**
|
||||
* @brief The output stream to print to.
|
||||
*/
|
||||
std::ostream &out_stream;
|
||||
};
|
||||
|
||||
// ================================== End class synced_stream ================================== //
|
||||
|
||||
// ===================================== Begin class timer ===================================== //
|
||||
|
||||
/**
|
||||
* @brief A helper class to measure execution time for benchmarking purposes.
|
||||
*/
|
||||
class timer
|
||||
{
|
||||
typedef std::int_fast64_t i64;
|
||||
|
||||
public:
|
||||
/**
|
||||
* @brief Start (or restart) measuring time.
|
||||
*/
|
||||
void start()
|
||||
{
|
||||
start_time = std::chrono::steady_clock::now();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Stop measuring time and store the elapsed time since start().
|
||||
*/
|
||||
void stop()
|
||||
{
|
||||
elapsed_time = std::chrono::steady_clock::now() - start_time;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Get the number of milliseconds that have elapsed between start() and stop().
|
||||
*
|
||||
* @return The number of milliseconds.
|
||||
*/
|
||||
i64 ms() const
|
||||
{
|
||||
return (std::chrono::duration_cast<std::chrono::milliseconds>(elapsed_time)).count();
|
||||
}
|
||||
|
||||
private:
|
||||
/**
|
||||
* @brief The time point when measuring started.
|
||||
*/
|
||||
std::chrono::time_point<std::chrono::steady_clock> start_time = std::chrono::steady_clock::now();
|
||||
|
||||
/**
|
||||
* @brief The duration that has elapsed between start() and stop().
|
||||
*/
|
||||
std::chrono::duration<double> elapsed_time = std::chrono::duration<double>::zero();
|
||||
};
|
||||
|
||||
// ====================================== End class timer ====================================== //
|
||||
#pragma once
|
||||
|
||||
/**
|
||||
* @file thread_pool.hpp
|
||||
* @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
|
||||
* @version 1.9
|
||||
* @date 2021-07-29
|
||||
* @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)
|
||||
*
|
||||
* @brief A C++17 thread pool for high-performance scientific computing.
|
||||
* @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
|
||||
|
||||
#include <atomic> // std::atomic
|
||||
#include <chrono> // std::chrono
|
||||
#include <cstdint> // std::int_fast64_t, std::uint_fast32_t
|
||||
#include <functional> // std::function
|
||||
#include <future> // std::future, std::promise
|
||||
#include <iostream> // std::cout, std::ostream
|
||||
#include <memory> // std::shared_ptr, std::unique_ptr
|
||||
#include <mutex> // std::mutex, std::scoped_lock
|
||||
#include <queue> // std::queue
|
||||
#include <thread> // std::this_thread, std::thread
|
||||
#include <type_traits> // std::decay_t, std::enable_if_t, std::is_void_v, std::invoke_result_t
|
||||
#include <utility> // std::move, std::swap
|
||||
|
||||
// ============================================================================================= //
|
||||
// Begin class thread_pool //
|
||||
|
||||
/**
|
||||
* @brief A C++17 thread pool class. The user submits tasks to be executed into a queue. Whenever a thread becomes available, it pops a task from the queue and executes it. Each task is automatically assigned a future, which can be used to wait for the task to finish executing and/or obtain its eventual return value.
|
||||
*/
|
||||
class thread_pool
|
||||
{
|
||||
typedef std::uint_fast32_t ui32;
|
||||
|
||||
public:
|
||||
// ============================
|
||||
// Constructors and destructors
|
||||
// ============================
|
||||
|
||||
/**
|
||||
* @brief Construct a new thread pool.
|
||||
*
|
||||
* @param _thread_count The number of threads to use. The default value is the total number of hardware threads available, as reported by the implementation. With a hyperthreaded CPU, this will be twice the number of CPU cores. If the argument is zero, the default value will be used instead.
|
||||
*/
|
||||
thread_pool(const ui32 &_thread_count = std::thread::hardware_concurrency())
|
||||
: thread_count(_thread_count ? _thread_count : std::thread::hardware_concurrency()), threads(new std::thread[_thread_count ? _thread_count : std::thread::hardware_concurrency()])
|
||||
{
|
||||
create_threads();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Destruct the thread pool. Waits for all tasks to complete, then destroys all threads. Note that if the variable paused is set to true, then any tasks still in the queue will never be executed.
|
||||
*/
|
||||
~thread_pool()
|
||||
{
|
||||
wait_for_tasks();
|
||||
running = false;
|
||||
destroy_threads();
|
||||
}
|
||||
|
||||
// =======================
|
||||
// Public member functions
|
||||
// =======================
|
||||
|
||||
/**
|
||||
* @brief Get the number of tasks currently waiting in the queue to be executed by the threads.
|
||||
*
|
||||
* @return The number of queued tasks.
|
||||
*/
|
||||
size_t get_tasks_queued() const
|
||||
{
|
||||
const std::scoped_lock lock(queue_mutex);
|
||||
return tasks.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Get the number of tasks currently being executed by the threads.
|
||||
*
|
||||
* @return The number of running tasks.
|
||||
*/
|
||||
ui32 get_tasks_running() const
|
||||
{
|
||||
return tasks_total - (ui32)get_tasks_queued();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Get the total number of unfinished tasks - either still in the queue, or running in a thread.
|
||||
*
|
||||
* @return The total number of tasks.
|
||||
*/
|
||||
ui32 get_tasks_total() const
|
||||
{
|
||||
return tasks_total;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Get the number of threads in the pool.
|
||||
*
|
||||
* @return The number of threads.
|
||||
*/
|
||||
ui32 get_thread_count() const
|
||||
{
|
||||
return thread_count;
|
||||
}
|
||||
|
||||
/**
|
||||
* @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);
|
||||
*
|
||||
* @tparam T The type of the loop index. Should be a signed or unsigned integer.
|
||||
* @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.
|
||||
*/
|
||||
template <typename T, typename F>
|
||||
void parallelize_loop(T first_index, T last_index, const F &loop, ui32 num_tasks = 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;
|
||||
if (block_size == 0)
|
||||
{
|
||||
block_size = 1;
|
||||
num_tasks = (ui32)total_size > 1 ? (ui32)total_size : 1;
|
||||
}
|
||||
std::atomic<ui32> blocks_running = 0;
|
||||
for (ui32 t = 0; t < num_tasks; 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);
|
||||
blocks_running++;
|
||||
push_task([start, end, &loop, &blocks_running]
|
||||
{
|
||||
for (T i = start; i <= end; i++)
|
||||
loop(i);
|
||||
blocks_running--;
|
||||
});
|
||||
}
|
||||
while (blocks_running != 0)
|
||||
{
|
||||
sleep_or_yield();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Push a function with no arguments or return value into the task queue.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @param task The function to push.
|
||||
*/
|
||||
template <typename F>
|
||||
void push_task(const F &task)
|
||||
{
|
||||
tasks_total++;
|
||||
{
|
||||
const std::scoped_lock lock(queue_mutex);
|
||||
tasks.push(std::function<void()>(task));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Push a function with arguments, but no return value, into the task queue.
|
||||
* @details The function is wrapped inside a lambda in order to hide the arguments, as the tasks in the queue must be of type std::function<void()>, so they cannot have any arguments or return value. If no arguments are provided, the other overload will be used, in order to avoid the (slight) overhead of using a lambda.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @tparam A The types of the arguments.
|
||||
* @param task The function to push.
|
||||
* @param args The arguments to pass to the function.
|
||||
*/
|
||||
template <typename F, typename... A>
|
||||
void push_task(const F &task, const A &...args)
|
||||
{
|
||||
push_task([task, args...]
|
||||
{ task(args...); });
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Reset the number of threads in the pool. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.
|
||||
*
|
||||
* @param _thread_count The number of threads to use. The default value is the total number of hardware threads available, as reported by the implementation. With a hyperthreaded CPU, this will be twice the number of CPU cores. If the argument is zero, the default value will be used instead.
|
||||
*/
|
||||
void reset(const ui32 &_thread_count = std::thread::hardware_concurrency())
|
||||
{
|
||||
bool was_paused = paused;
|
||||
paused = true;
|
||||
wait_for_tasks();
|
||||
running = false;
|
||||
destroy_threads();
|
||||
thread_count = _thread_count ? _thread_count : std::thread::hardware_concurrency();
|
||||
threads.reset(new std::thread[thread_count]);
|
||||
paused = was_paused;
|
||||
running = true;
|
||||
create_threads();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Submit a function with zero or more arguments and no return value into the task queue, and get an std::future<bool> that will be set to true upon completion of the task.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @tparam A The types of the zero or more arguments to pass to the function.
|
||||
* @param task The function to submit.
|
||||
* @param args The zero or more arguments to pass to the function.
|
||||
* @return A future to be used later to check if the function has finished its execution.
|
||||
*/
|
||||
template <typename F, typename... A, typename = std::enable_if_t<std::is_void_v<std::invoke_result_t<std::decay_t<F>, std::decay_t<A>...>>>>
|
||||
std::future<bool> submit(const F &task, const A &...args)
|
||||
{
|
||||
std::shared_ptr<std::promise<bool>> task_promise(new std::promise<bool>);
|
||||
std::future<bool> future = task_promise->get_future();
|
||||
push_task([task, args..., task_promise]
|
||||
{
|
||||
try
|
||||
{
|
||||
task(args...);
|
||||
task_promise->set_value(true);
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
try
|
||||
{
|
||||
task_promise->set_exception(std::current_exception());
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
}
|
||||
}
|
||||
});
|
||||
return future;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Submit a function with zero or more arguments and a return value into the task queue, and get a future for its eventual returned value.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @tparam A The types of the zero or more arguments to pass to the function.
|
||||
* @tparam R The return type of the function.
|
||||
* @param task The function to submit.
|
||||
* @param args The zero or more arguments to pass to the function.
|
||||
* @return A future to be used later to obtain the function's returned value, waiting for it to finish its execution if needed.
|
||||
*/
|
||||
template <typename F, typename... A, typename R = std::invoke_result_t<std::decay_t<F>, std::decay_t<A>...>, typename = std::enable_if_t<!std::is_void_v<R>>>
|
||||
std::future<R> submit(const F &task, const A &...args)
|
||||
{
|
||||
std::shared_ptr<std::promise<R>> task_promise(new std::promise<R>);
|
||||
std::future<R> future = task_promise->get_future();
|
||||
push_task([task, args..., task_promise]
|
||||
{
|
||||
try
|
||||
{
|
||||
task_promise->set_value(task(args...));
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
try
|
||||
{
|
||||
task_promise->set_exception(std::current_exception());
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
}
|
||||
}
|
||||
});
|
||||
return future;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Wait for tasks to be completed. Normally, this function waits for all tasks, both those that are currently running in the threads and those that are still waiting in the queue. However, if the variable paused is set to true, this function only waits for the currently running tasks (otherwise it would wait forever). To wait for a specific task, use submit() instead, and call the wait() member function of the generated future.
|
||||
*/
|
||||
void wait_for_tasks()
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
if (!paused)
|
||||
{
|
||||
if (tasks_total == 0)
|
||||
break;
|
||||
}
|
||||
else
|
||||
{
|
||||
if (get_tasks_running() == 0)
|
||||
break;
|
||||
}
|
||||
sleep_or_yield();
|
||||
}
|
||||
}
|
||||
|
||||
// ===========
|
||||
// Public data
|
||||
// ===========
|
||||
|
||||
/**
|
||||
* @brief An atomic variable indicating to the workers to pause. When set to true, the workers temporarily stop popping new tasks out of the queue, although any tasks already executed will keep running until they are done. Set to false again to resume popping tasks.
|
||||
*/
|
||||
std::atomic<bool> paused = false;
|
||||
|
||||
/**
|
||||
* @brief The duration, in microseconds, that the worker function should sleep for when it cannot find any tasks in the queue. If set to 0, then instead of sleeping, the worker function will execute std::this_thread::yield() if there are no tasks in the queue. The default value is 1000.
|
||||
*/
|
||||
ui32 sleep_duration = 1000;
|
||||
|
||||
private:
|
||||
// ========================
|
||||
// Private member functions
|
||||
// ========================
|
||||
|
||||
/**
|
||||
* @brief Create the threads in the pool and assign a worker to each thread.
|
||||
*/
|
||||
void create_threads()
|
||||
{
|
||||
for (ui32 i = 0; i < thread_count; i++)
|
||||
{
|
||||
threads[i] = std::thread(&thread_pool::worker, this);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Destroy the threads in the pool by joining them.
|
||||
*/
|
||||
void destroy_threads()
|
||||
{
|
||||
for (ui32 i = 0; i < thread_count; i++)
|
||||
{
|
||||
threads[i].join();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Try to pop a new task out of the queue.
|
||||
*
|
||||
* @param task A reference to the task. Will be populated with a function if the queue is not empty.
|
||||
* @return true if a task was found, false if the queue is empty.
|
||||
*/
|
||||
bool pop_task(std::function<void()> &task)
|
||||
{
|
||||
const std::scoped_lock lock(queue_mutex);
|
||||
if (tasks.empty())
|
||||
return false;
|
||||
else
|
||||
{
|
||||
task = std::move(tasks.front());
|
||||
tasks.pop();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Sleep for sleep_duration microseconds. If that variable is set to zero, yield instead.
|
||||
*
|
||||
*/
|
||||
void sleep_or_yield()
|
||||
{
|
||||
if (sleep_duration)
|
||||
std::this_thread::sleep_for(std::chrono::microseconds(sleep_duration));
|
||||
else
|
||||
std::this_thread::yield();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief A worker function to be assigned to each thread in the pool. Continuously pops tasks out of the queue and executes them, as long as the atomic variable running is set to true.
|
||||
*/
|
||||
void worker()
|
||||
{
|
||||
while (running)
|
||||
{
|
||||
std::function<void()> task;
|
||||
if (!paused && pop_task(task))
|
||||
{
|
||||
task();
|
||||
tasks_total--;
|
||||
}
|
||||
else
|
||||
{
|
||||
sleep_or_yield();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ============
|
||||
// Private data
|
||||
// ============
|
||||
|
||||
/**
|
||||
* @brief A mutex to synchronize access to the task queue by different threads.
|
||||
*/
|
||||
mutable std::mutex queue_mutex = {};
|
||||
|
||||
/**
|
||||
* @brief An atomic variable indicating to the workers to keep running. When set to false, the workers permanently stop working.
|
||||
*/
|
||||
std::atomic<bool> running = true;
|
||||
|
||||
/**
|
||||
* @brief A queue of tasks to be executed by the threads.
|
||||
*/
|
||||
std::queue<std::function<void()>> tasks = {};
|
||||
|
||||
/**
|
||||
* @brief The number of threads in the pool.
|
||||
*/
|
||||
ui32 thread_count;
|
||||
|
||||
/**
|
||||
* @brief A smart pointer to manage the memory allocated for the threads.
|
||||
*/
|
||||
std::unique_ptr<std::thread[]> threads;
|
||||
|
||||
/**
|
||||
* @brief An atomic variable to keep track of the total number of unfinished tasks - either still in the queue, or running in a thread.
|
||||
*/
|
||||
std::atomic<ui32> tasks_total = 0;
|
||||
};
|
||||
|
||||
// End class thread_pool //
|
||||
// ============================================================================================= //
|
||||
|
||||
// ============================================================================================= //
|
||||
// Begin class synced_stream //
|
||||
|
||||
/**
|
||||
* @brief A helper class to synchronize printing to an output stream by different threads.
|
||||
*/
|
||||
class synced_stream
|
||||
{
|
||||
public:
|
||||
/**
|
||||
* @brief Construct a new synced stream.
|
||||
*
|
||||
* @param _out_stream The output stream to print to. The default value is std::cout.
|
||||
*/
|
||||
synced_stream(std::ostream &_out_stream = std::cout)
|
||||
: out_stream(_out_stream){};
|
||||
|
||||
/**
|
||||
* @brief Print any number of items into the output stream. Ensures that no other threads print to this stream simultaneously, as long as they all exclusively use this synced_stream object to print.
|
||||
*
|
||||
* @tparam T The types of the items
|
||||
* @param items The items to print.
|
||||
*/
|
||||
template <typename... T>
|
||||
void print(const T &...items)
|
||||
{
|
||||
const std::scoped_lock lock(stream_mutex);
|
||||
(out_stream << ... << items);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Print any number of items into the output stream, followed by a newline character. Ensures that no other threads print to this stream simultaneously, as long as they all exclusively use this synced_stream object to print.
|
||||
*
|
||||
* @tparam T The types of the items
|
||||
* @param items The items to print.
|
||||
*/
|
||||
template <typename... T>
|
||||
void println(const T &...items)
|
||||
{
|
||||
print(items..., '\n');
|
||||
}
|
||||
|
||||
private:
|
||||
/**
|
||||
* @brief A mutex to synchronize printing.
|
||||
*/
|
||||
mutable std::mutex stream_mutex = {};
|
||||
|
||||
/**
|
||||
* @brief The output stream to print to.
|
||||
*/
|
||||
std::ostream &out_stream;
|
||||
};
|
||||
|
||||
// End class synced_stream //
|
||||
// ============================================================================================= //
|
||||
|
||||
// ============================================================================================= //
|
||||
// Begin class timer //
|
||||
|
||||
/**
|
||||
* @brief A helper class to measure execution time for benchmarking purposes.
|
||||
*/
|
||||
class timer
|
||||
{
|
||||
typedef std::int_fast64_t i64;
|
||||
|
||||
public:
|
||||
/**
|
||||
* @brief Start (or restart) measuring time.
|
||||
*/
|
||||
void start()
|
||||
{
|
||||
start_time = std::chrono::steady_clock::now();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Stop measuring time and store the elapsed time since start().
|
||||
*/
|
||||
void stop()
|
||||
{
|
||||
elapsed_time = std::chrono::steady_clock::now() - start_time;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Get the number of milliseconds that have elapsed between start() and stop().
|
||||
*
|
||||
* @return The number of milliseconds.
|
||||
*/
|
||||
i64 ms() const
|
||||
{
|
||||
return (std::chrono::duration_cast<std::chrono::milliseconds>(elapsed_time)).count();
|
||||
}
|
||||
|
||||
private:
|
||||
/**
|
||||
* @brief The time point when measuring started.
|
||||
*/
|
||||
std::chrono::time_point<std::chrono::steady_clock> start_time = std::chrono::steady_clock::now();
|
||||
|
||||
/**
|
||||
* @brief The duration that has elapsed between start() and stop().
|
||||
*/
|
||||
std::chrono::duration<double> elapsed_time = std::chrono::duration<double>::zero();
|
||||
};
|
||||
|
||||
// End class timer //
|
||||
// ============================================================================================= //
|
||||
|
||||
Reference in New Issue
Block a user