mirror of
https://github.com/bshoshany/thread-pool.git
synced 2026-07-24 20:43:00 +04:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Barak Shoshany
|
cabb3df587 | ||
|
Barak Shoshany
|
87d415f89c | ||
|
Barak Shoshany
|
67fad04348 |
+190
@@ -0,0 +1,190 @@
|
||||
---
|
||||
Language: Cpp
|
||||
AccessModifierOffset: -4
|
||||
AlignAfterOpenBracket: Align
|
||||
AlignArrayOfStructures: None
|
||||
AlignConsecutiveMacros: None
|
||||
AlignConsecutiveAssignments: None
|
||||
AlignConsecutiveBitFields: None
|
||||
AlignConsecutiveDeclarations: None
|
||||
AlignEscapedNewlines: Right
|
||||
AlignOperands: Align
|
||||
AlignTrailingComments: true
|
||||
AllowAllArgumentsOnNextLine: true
|
||||
AllowAllParametersOfDeclarationOnNextLine: true
|
||||
AllowShortEnumsOnASingleLine: true
|
||||
AllowShortBlocksOnASingleLine: Never
|
||||
AllowShortCaseLabelsOnASingleLine: false
|
||||
AllowShortFunctionsOnASingleLine: Empty
|
||||
AllowShortLambdasOnASingleLine: All
|
||||
AllowShortIfStatementsOnASingleLine: Never
|
||||
AllowShortLoopsOnASingleLine: false
|
||||
AlwaysBreakAfterDefinitionReturnType: None
|
||||
AlwaysBreakAfterReturnType: None
|
||||
AlwaysBreakBeforeMultilineStrings: false
|
||||
AlwaysBreakTemplateDeclarations: Yes
|
||||
AttributeMacros:
|
||||
- __capability
|
||||
BinPackArguments: true
|
||||
BinPackParameters: true
|
||||
BraceWrapping:
|
||||
AfterCaseLabel: false
|
||||
AfterClass: false
|
||||
AfterControlStatement: Never
|
||||
AfterEnum: false
|
||||
AfterFunction: false
|
||||
AfterNamespace: false
|
||||
AfterObjCDeclaration: false
|
||||
AfterStruct: false
|
||||
AfterUnion: false
|
||||
AfterExternBlock: false
|
||||
BeforeCatch: false
|
||||
BeforeElse: false
|
||||
BeforeLambdaBody: false
|
||||
BeforeWhile: false
|
||||
IndentBraces: false
|
||||
SplitEmptyFunction: true
|
||||
SplitEmptyRecord: true
|
||||
SplitEmptyNamespace: true
|
||||
BreakBeforeBinaryOperators: None
|
||||
BreakBeforeConceptDeclarations: true
|
||||
BreakBeforeBraces: Allman
|
||||
BreakBeforeInheritanceComma: false
|
||||
BreakInheritanceList: BeforeColon
|
||||
BreakBeforeTernaryOperators: true
|
||||
BreakConstructorInitializersBeforeComma: false
|
||||
BreakConstructorInitializers: BeforeColon
|
||||
BreakAfterJavaFieldAnnotations: false
|
||||
BreakStringLiterals: true
|
||||
ColumnLimit: 10000
|
||||
CommentPragmas: "^ IWYU pragma:"
|
||||
QualifierAlignment: Leave
|
||||
CompactNamespaces: false
|
||||
ConstructorInitializerIndentWidth: 4
|
||||
ContinuationIndentWidth: 4
|
||||
Cpp11BracedListStyle: true
|
||||
DeriveLineEnding: true
|
||||
DerivePointerAlignment: false
|
||||
DisableFormat: false
|
||||
EmptyLineAfterAccessModifier: Never
|
||||
EmptyLineBeforeAccessModifier: LogicalBlock
|
||||
ExperimentalAutoDetectBinPacking: false
|
||||
PackConstructorInitializers: BinPack
|
||||
BasedOnStyle: ""
|
||||
ConstructorInitializerAllOnOneLineOrOnePerLine: false
|
||||
AllowAllConstructorInitializersOnNextLine: true
|
||||
FixNamespaceComments: true
|
||||
ForEachMacros:
|
||||
- foreach
|
||||
- Q_FOREACH
|
||||
- BOOST_FOREACH
|
||||
IfMacros:
|
||||
- KJ_IF_MAYBE
|
||||
IncludeBlocks: Preserve
|
||||
IncludeCategories:
|
||||
- Regex: '^"(llvm|llvm-c|clang|clang-c)/'
|
||||
Priority: 2
|
||||
SortPriority: 0
|
||||
CaseSensitive: false
|
||||
- Regex: '^(<|"(gtest|gmock|isl|json)/)'
|
||||
Priority: 3
|
||||
SortPriority: 0
|
||||
CaseSensitive: false
|
||||
- Regex: ".*"
|
||||
Priority: 1
|
||||
SortPriority: 0
|
||||
CaseSensitive: false
|
||||
IncludeIsMainRegex: "(Test)?$"
|
||||
IncludeIsMainSourceRegex: ""
|
||||
IndentAccessModifiers: false
|
||||
IndentCaseLabels: false
|
||||
IndentCaseBlocks: false
|
||||
IndentGotoLabels: true
|
||||
IndentPPDirectives: None
|
||||
IndentExternBlock: AfterExternBlock
|
||||
IndentRequires: false
|
||||
IndentWidth: 4
|
||||
IndentWrappedFunctionNames: false
|
||||
InsertTrailingCommas: None
|
||||
JavaScriptQuotes: Leave
|
||||
JavaScriptWrapImports: true
|
||||
KeepEmptyLinesAtTheStartOfBlocks: true
|
||||
LambdaBodyIndentation: Signature
|
||||
MacroBlockBegin: ""
|
||||
MacroBlockEnd: ""
|
||||
MaxEmptyLinesToKeep: 1
|
||||
NamespaceIndentation: None
|
||||
ObjCBinPackProtocolList: Auto
|
||||
ObjCBlockIndentWidth: 4
|
||||
ObjCBreakBeforeNestedBlockParam: true
|
||||
ObjCSpaceAfterProperty: false
|
||||
ObjCSpaceBeforeProtocolList: true
|
||||
PenaltyBreakAssignment: 2
|
||||
PenaltyBreakBeforeFirstCallParameter: 19
|
||||
PenaltyBreakComment: 300
|
||||
PenaltyBreakFirstLessLess: 120
|
||||
PenaltyBreakOpenParenthesis: 0
|
||||
PenaltyBreakString: 1000
|
||||
PenaltyBreakTemplateDeclaration: 10
|
||||
PenaltyExcessCharacter: 1000000
|
||||
PenaltyReturnTypeOnItsOwnLine: 60
|
||||
PenaltyIndentedWhitespace: 0
|
||||
PointerAlignment: Left
|
||||
PPIndentWidth: -1
|
||||
ReferenceAlignment: Pointer
|
||||
ReflowComments: true
|
||||
RemoveBracesLLVM: false
|
||||
SeparateDefinitionBlocks: Leave
|
||||
ShortNamespaceLines: 1
|
||||
SortIncludes: CaseSensitive
|
||||
SortJavaStaticImport: Before
|
||||
SortUsingDeclarations: true
|
||||
SpaceAfterCStyleCast: false
|
||||
SpaceAfterLogicalNot: false
|
||||
SpaceAfterTemplateKeyword: true
|
||||
SpaceBeforeAssignmentOperators: true
|
||||
SpaceBeforeCaseColon: false
|
||||
SpaceBeforeCpp11BracedList: true
|
||||
SpaceBeforeCtorInitializerColon: true
|
||||
SpaceBeforeInheritanceColon: true
|
||||
SpaceBeforeParens: ControlStatements
|
||||
SpaceBeforeParensOptions:
|
||||
AfterControlStatements: true
|
||||
AfterForeachMacros: true
|
||||
AfterFunctionDefinitionName: false
|
||||
AfterFunctionDeclarationName: false
|
||||
AfterIfMacros: true
|
||||
AfterOverloadedOperator: false
|
||||
BeforeNonEmptyParentheses: false
|
||||
SpaceAroundPointerQualifiers: Default
|
||||
SpaceBeforeRangeBasedForLoopColon: true
|
||||
SpaceInEmptyBlock: false
|
||||
SpaceInEmptyParentheses: false
|
||||
SpacesBeforeTrailingComments: 1
|
||||
SpacesInAngles: Never
|
||||
SpacesInConditionalStatement: false
|
||||
SpacesInContainerLiterals: true
|
||||
SpacesInCStyleCastParentheses: false
|
||||
SpacesInLineCommentPrefix:
|
||||
Minimum: 1
|
||||
Maximum: -1
|
||||
SpacesInParentheses: false
|
||||
SpacesInSquareBrackets: false
|
||||
SpaceBeforeSquareBrackets: false
|
||||
BitFieldColonSpacing: Both
|
||||
Standard: Latest
|
||||
StatementAttributeLikeMacros:
|
||||
- Q_EMIT
|
||||
StatementMacros:
|
||||
- Q_UNUSED
|
||||
- QT_REQUIRE_VERSION
|
||||
TabWidth: 4
|
||||
UseCRLF: false
|
||||
UseTab: Never
|
||||
WhitespaceSensitiveMacros:
|
||||
- STRINGIZE
|
||||
- PP_STRINGIZE
|
||||
- BOOST_PP_STRINGIZE
|
||||
- NS_SWIFT_NAME
|
||||
- CF_SWIFT_NAME
|
||||
---
|
||||
@@ -6,9 +6,15 @@
|
||||
|
||||
What does your pull request fix or add to the library?
|
||||
|
||||
**Style**
|
||||
|
||||
Have you formatted your code using the `.clang-format` file attached to this project?
|
||||
|
||||
**Testing**
|
||||
|
||||
Have you tested the new code using the provided automated test program and/or performed any other tests to ensure that it works correctly? If so, please provide information about the test system(s):
|
||||
Have you tested the new code using the provided automated test program `BS_thread_pool_test.cpp` (preferably with the provided multi-compiler test script `BS_thread_pool_test.ps1`) and/or performed any other tests to ensure that the new code works correctly?
|
||||
|
||||
If so, please provide information about the test system(s):
|
||||
|
||||
* CPU model, architecture, # of cores and threads:
|
||||
* Operating system:
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
build
|
||||
+78
-14
@@ -1,21 +1,14 @@
|
||||
[](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://vscode.dev/github/bshoshany/thread-pool)
|
||||
|
||||
# `BS::thread_pool`: a fast, lightweight, and easy-to-use C++17 thread pool library
|
||||
|
||||
By Barak Shoshany<br />
|
||||
Email: [baraksh@gmail.com](mailto:baraksh@gmail.com)<br />
|
||||
Website: [https://baraksh.com/](https://baraksh.com/)<br />
|
||||
GitHub: [https://github.com/bshoshany](https://github.com/bshoshany)<br />
|
||||
By Barak Shoshany\
|
||||
Email: <baraksh@gmail.com>\
|
||||
Website: <https://baraksh.com/>\
|
||||
GitHub: <https://github.com/bshoshany>
|
||||
|
||||
* [Version history](#version-history)
|
||||
* [v3.5.0 (2023-05-25)](#v350-2023-05-25)
|
||||
* [v3.4.0 (2023-05-12)](#v340-2023-05-12)
|
||||
* [v3.3.0 (2022-08-03)](#v330-2022-08-03)
|
||||
* [v3.2.0 (2022-07-28)](#v320-2022-07-28)
|
||||
* [v3.1.0 (2022-07-13)](#v310-2022-07-13)
|
||||
* [v3.0.0 (2022-05-30)](#v300-2022-05-30)
|
||||
@@ -33,6 +26,77 @@ GitHub: [https://github.com/bshoshany](https://github.com/bshoshany)<br />
|
||||
|
||||
## Version history
|
||||
|
||||
### v3.5.0 (2023-05-25)
|
||||
|
||||
* `BS_thread_pool.hpp` and `BS_thread_pool_light.hpp`:
|
||||
* Added a new member function, `purge()`, to the full (non-light) thread pool. This function purges all the tasks waiting in the queue. Tasks that are currently running will not be affected, but any tasks still waiting in the queue will be removed and will never be executed by the threads. Please note that there is no way to restore the purged tasks.
|
||||
* Fix a bug which caused `wait_for_tasks()` to only block the first thread that called it. Now it blocks every thread that calls it, which is the expected behavior. In addition, all related deadlock have now been completely resolved. This also applies to the variants `wait_for_tasks_duration()` and `wait_for_tasks_until()` in the non-light version. See [#110](https://github.com/bshoshany/thread-pool/pull/110).
|
||||
* Note: You should never call `wait_for_tasks()` from within a thread of the same thread pool, as that will cause it to wait forever! This fix is relevant for situations when `wait_for_tasks()` is called from an auxiliary `std::thread` or a separate thread pool.
|
||||
* `push_task()` and `submit()` now avoid creating unnecessary copies of the function object. This should improve performance, especially if large objects are involved. See [#90](https://github.com/bshoshany/thread-pool/pull/90).
|
||||
* Optimized the way condition variables are used by the thread pool class. Shared variables are now modified while owning the mutex, but condition variables are notified after the mutex is released, if possible. See [#84](https://github.com/bshoshany/thread-pool/pull/84).
|
||||
* Instead of a variable `tasks_total` to keep track of the total number of tasks (queued + running), the thread pool class now uses a variable `tasks_running` to keep track only of the number of running tasks, with the number of tasks in the queue obtained via `tasks.size()`. This makes more sense in terms of the internal logic of the class.
|
||||
* All atomic variables have been converted to non-atomic. They are now all governed by `tasks_mutex`, so they do not need to be atomic. This eliminates redundant locking, and may improve performance a bit.
|
||||
* `running` has been renamed to `workers_running` and `task_done_cv` has been renamed to `tasks_done_cv`.
|
||||
* The worker now only notifies thi condition variable `tasks_done_cv` if all the tasks are done, not just a single task. Checking if the tasks are done is cheaper than notifying the condition variable, so since the worker no longer notifies the condition variable every single time it finishes a task, this should improve performance a bit if `wait_for_tasks()` is used.
|
||||
* `BS_thread_pool_test.cpp`:
|
||||
* Combined the tests for the full and light versions into one program. The file `BS_thread_pool_light_test.cpp` has been removed.
|
||||
* The tests for the light version are now much more comprehensive. The only features that are not tested in the light version are those that do not exist in it.
|
||||
* Added a test for the new `purge()` member function.
|
||||
* Added a test to ensure that `push_task()` and `submit()` do not create unnecessary copies of the function object.
|
||||
* Added a test to ensure that `push_task()` and `submit()` correctly accept arguments passed by value, reference, and constant reference.
|
||||
* Added a test to ensure that `wait_for_tasks()` blocks all external threads that call it.
|
||||
* `_CRT_SECURE_NO_WARNINGS` is now set only if it has not already been defined, to prevent errors in MSVC projects which already have it set as part of the default build settings. See [#72](https://github.com/bshoshany/thread-pool/pull/72).
|
||||
* `README.md`:
|
||||
* Added documentation for the new `purge()` member function.
|
||||
* Added an explanation for how to pass arguments by reference or constant reference when submitting functions to the queue, using the wrappers `std::ref()` and `std::cref()` respectively. See [#83](https://github.com/bshoshany/thread-pool/issues/83).
|
||||
* Added a link to [my lecture notes](https://baraksh.com/CSE701/notes.php) for a course taught at McMaster University, for the benefit of beginner C++ programmers who wish to learn some of the advanced techniques and programming practices used in developing this library.
|
||||
* Removed the sample test results, since the complete log file (including the deadlock tests) is now over 500 lines long.
|
||||
* Other:
|
||||
* A `.clang-format` file with the project's formatting conventions is now included in the GitHub repository. The pull request template now asks to format any new code using this file, so that it is consistent with the rest of the library.
|
||||
* A PowerShell script, `BS_thread_pool_test.ps1`, is now provided in the GitHub repository to make running the test on multiple compilers and operating systems easier. Since it is written in PowerShell, it is fully portable and works on Windows, Linux, and macOS. The script will automatically detect if Clang, GCC, and/or MSVC are available, compile the test program using each available compiler, and then run each compiled test program 5 times and report on any errors. The pull request template now recommends using this script for testing.
|
||||
* Since the root folder has become a bit crowded, the header files `BS_thread_pool.hpp` and `BS_thread_pool_light.hpp` have been moved to the `include` subfolder, and the test file `BS_thread_pool_test.cpp` has been moved to the `tests` subfolder, which also contains the new test script `BS_thread_pool_test.ps1`.
|
||||
|
||||
### v3.4.0 (2023-05-12)
|
||||
|
||||
* `BS_thread_pool.hpp` and `BS_thread_pool_light.hpp`:
|
||||
* Resolved an issue which could have caused `tasks_total` to not be synchronized in some cases. See [#70](https://github.com/bshoshany/thread-pool/pull/70).
|
||||
* Resolved a deadlock which could rarely be caused when the pool was destructed or reset. See [#93](https://github.com/bshoshany/thread-pool/pull/93), [#100](https://github.com/bshoshany/thread-pool/pull/100), [#107](https://github.com/bshoshany/thread-pool/pull/107), and [#108](https://github.com/bshoshany/thread-pool/pull/108).
|
||||
* Resolved a deadlock which could be caused when `wait_for_tasks()` was called more than once.
|
||||
* Two new member functions have been added to the non-light version: `wait_for_tasks_duration()` and `wait_for_tasks_until()`. They allow waiting for the tasks to complete, but with a timeout. `wait_for_tasks_duration()` will stop waiting after the specified duration has passed, and `wait_for_tasks_until()` will stop waiting after the specified time point has been reached.
|
||||
* Renamed `BS_THREAD_POOL_VERSION` in `BS_thread_pool_light.hpp` to `BS_THREAD_POOL_LIGHT_VERSION` and removed the `[light]` tag. This allows including both header files in the same program in case we want to use both the light and non-light thread pools simultaneously.
|
||||
* `BS_thread_pool_test.cpp` and `BS_thread_pool_light_test.cpp`:
|
||||
* Fixed an issue that caused a compilation error when using MSVC and including `Windows.h`. See [#72](https://github.com/bshoshany/thread-pool/pull/72).
|
||||
* The number and size of the vectors in the performance test (`BS_thread_pool_test.cpp` only) are now guaranteed to be multiples of the number of threads, for optimal performance.
|
||||
* In `count_unique_threads()`, moved the condition variables and mutexes to the function scope to prevent cluttering the global scope.
|
||||
* Three new tests have been added to `BS_thread_pool_test.cpp` to check the deadlocks issue that were resolved in this release (see above). The tests rely on the new wait for tasks with timeout feature, so they are not available in the light version.
|
||||
* One test checks for deadlocks when calling `wait_for_tasks()` more than once.
|
||||
* Two tests check for deadlocks when destructing and resetting the pool respectively. They are turned off by default, since they take a long time to complete, but can be turned on by setting `enable_long_deadlock_tests` to `true`.
|
||||
* Two new tests have been added to the non-light version to check the new member functions `wait_for_tasks_duration()` and `wait_for_tasks_until()`.
|
||||
* The test programs now return the number of failed tests upon exit, instead of just 1 if any number of tests failed, which was the case in previous versions. Also, if any tests failed, `std::quick_exit()` is invoked instead of `return`, to avoid getting stuck due to any lingering tasks or deadlocks.
|
||||
* `README.md`:
|
||||
* Added documentation for the two new member functions, `wait_for_tasks_duration()` and `wait_for_tasks_until()`.
|
||||
* Fixed Markdown rendering incorrectly on Visual Studio. See [#77](https://github.com/bshoshany/thread-pool/pull/77).
|
||||
* The sample performance tests are now taken from a 40-core / 80-thread dual-CPU computing node, which is a more typical use case for high-performance scientific software.
|
||||
|
||||
### v3.3.0 (2022-08-03)
|
||||
|
||||
* `BS_thread_pool.hpp`:
|
||||
* The public member variable `paused` of `BS::thread_pool` has been made private for future-proofing (in case future versions implement a more involved pausing mechanism) and better encapsulation. It is now accessible only via the `pause()`, `unpause()`, and `is_paused()` member functions. In other words:
|
||||
* Replace `pool.paused = true` with `pool.pause()`.
|
||||
* Replace `pool.paused = false` with `pool.unpause()`.
|
||||
* Replace `if (pool.paused)` (or similar) with `if (pool.is_paused())`.
|
||||
* The public member variable `f` of `BS::multi_future` has been renamed to `futures` for clarity, and has been made private for encapsulation and simplification purposes. Instead of operating on the vector `futures` itself, you can now use the `[]` operator of the `BS::multi_future` to access the future at a specific index directly, or the `push_back()` member function to append a new future to the list. The `size()` member function tells you how many futures are currently stored in the object.
|
||||
* The explicit casts of `std::endl` and `std::flush`, added in v3.2.0 to enable flushing a `BS::synced_stream`, caused ODR (One Definition Rule) violations if `BS_thread_pool.hpp` was included in two different translation units, since they were mistakenly not defined as `inline`. To fix this, I decided to make them static members of `BS::synced_stream` instead of global variables, which also makes the code better organized in my opinion. These objects can now be accessed as `BS::synced_stream::endl` and `BS::synced_stream::flush`. I also added an example for how to use them in `README.md`. See [#64](https://github.com/bshoshany/thread-pool/issues/64).
|
||||
* `BS_thread_pool_light.hpp`:
|
||||
* This package started out as a very lightweight thread pool, but over time has expanded to include many additional features, and at the time of writing it has a total of 340 lines of code, including all the helper classes. Therefore, I have decided to bundle a light version of the thread pool in a separate and stand-alone header file, `BS_thread_pool_light.hpp`, with only 170 lines of code (half the size of the full package). This file does not contain any of the helper classes, only a new `BS::thread_pool_light` class, which is a minimal thread pool with only the 5 most basic member functions:
|
||||
* `get_thread_count()`
|
||||
* `push_loop()`
|
||||
* `push_task()`
|
||||
* `submit()`
|
||||
* `wait_for_tasks()`
|
||||
* A separate test program `BS_thread_pool_light_test.cpp` tests only the features of the lightweight `BS::thread_pool_light` class. In the spirit of minimalism, it does not generate a log file and does not do any benchmarks.
|
||||
* To be perfectly clear, each header file is 100% stand-alone. If you wish to use the full package, you only need `BS_thread_pool.hpp`, and if you wish to use the light version, you only need `BS_thread_pool_light.hpp`. Only a single header file needs to be included in your project.
|
||||
|
||||
### v3.2.0 (2022-07-28)
|
||||
|
||||
* `BS_thread_pool.hpp`:
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2022 Barak Shoshany
|
||||
Copyright (c) 2023 Barak Shoshany
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
|
||||
@@ -5,17 +5,16 @@
|
||||
|
||||
|
||||
[](https://github.com/bshoshany/thread-pool)
|
||||
[](https://twitter.com/BarakShoshany)
|
||||
[](https://vscode.dev/github/bshoshany/thread-pool)
|
||||
|
||||
# `BS::thread_pool`: a fast, lightweight, and easy-to-use C++17 thread pool library
|
||||
|
||||
By Barak Shoshany<br />
|
||||
Email: [baraksh@gmail.com](mailto:baraksh@gmail.com)<br />
|
||||
Website: [https://baraksh.com/](https://baraksh.com/)<br />
|
||||
GitHub: [https://github.com/bshoshany](https://github.com/bshoshany)<br />
|
||||
By Barak Shoshany\
|
||||
Email: <baraksh@gmail.com>\
|
||||
Website: <https://baraksh.com/>\
|
||||
GitHub: <https://github.com/bshoshany>
|
||||
|
||||
This is the complete documentation for v3.2.0 of the library, released on 2022-07-28.
|
||||
This is the complete documentation for v3.5.0 of the library, released on 2023-05-25.
|
||||
|
||||
* [Introduction](#introduction)
|
||||
* [Motivation](#motivation)
|
||||
@@ -32,38 +31,44 @@ This is the complete documentation for v3.2.0 of the library, released on 2022-0
|
||||
* [Submitting tasks to the queue with futures](#submitting-tasks-to-the-queue-with-futures)
|
||||
* [Submitting tasks to the queue without futures](#submitting-tasks-to-the-queue-without-futures)
|
||||
* [Manually waiting for all tasks to complete](#manually-waiting-for-all-tasks-to-complete)
|
||||
* [Waiting with a timeout](#waiting-with-a-timeout)
|
||||
* [Submitting class member functions to the queue](#submitting-class-member-functions-to-the-queue)
|
||||
* [Parallelizing loops](#parallelizing-loops)
|
||||
* [Passing task arguments by reference](#passing-task-arguments-by-reference)
|
||||
* [Parallelizing loops](#parallelizing-loops)
|
||||
* [Automatic parallelization of loops](#automatic-parallelization-of-loops)
|
||||
* [Loops with return values](#loops-with-return-values)
|
||||
* [Parallelizing loops without futures](#parallelizing-loops-without-futures)
|
||||
* [Helper classes](#helper-classes)
|
||||
* [Handling multiple futures at once](#handling-multiple-futures-at-once)
|
||||
* [Synchronizing printing to an output stream](#synchronizing-printing-to-an-output-stream)
|
||||
* [Handling multiple futures at once](#handling-multiple-futures-at-once)
|
||||
* [Measuring execution time](#measuring-execution-time)
|
||||
* [Other features](#other-features)
|
||||
* [Monitoring the tasks](#monitoring-the-tasks)
|
||||
* [Pausing the workers](#pausing-the-workers)
|
||||
* [Purging tasks](#purging-tasks)
|
||||
* [Exception handling](#exception-handling)
|
||||
* [Testing the package](#testing-the-package)
|
||||
* [Automated tests](#automated-tests)
|
||||
* [Performance tests](#performance-tests)
|
||||
* [The light version of the package](#the-light-version-of-the-package)
|
||||
* [About the project](#about-the-project)
|
||||
* [Issue and pull request policy](#issue-and-pull-request-policy)
|
||||
* [Acknowledgements](#acknowledgements)
|
||||
* [Starring the repository](#starring-the-repository)
|
||||
* [Copyright and citing](#copyright-and-citing)
|
||||
* [Learning more about C++](#learning-more-about-c)
|
||||
|
||||
## Introduction
|
||||
|
||||
### Motivation
|
||||
|
||||
Multithreading is essential for modern high-performance computing. Since C++11, the C++ standard library has included built-in low-level multithreading support using constructs such as `std::thread`. However, `std::thread` creates a new thread each time it is called, which can have a significant performance overhead. Furthermore, it is possible to create more threads than the hardware can handle simultaneously, potentially resulting in a substantial slowdown.
|
||||
Multithreading is essential for modern high-performance computing. Since C\+\+11, the C++ standard library has included built-in low-level multithreading support using constructs such as `std::thread`. However, `std::thread` creates a new thread each time it is called, which can have a significant performance overhead. Furthermore, it is possible to create more threads than the hardware can handle simultaneously, potentially resulting in a substantial slowdown.
|
||||
|
||||
The library presented here contains a thread pool class, `BS::thread_pool`, which avoids these issues by creating a fixed pool of threads once and for all, and then continuously reusing the same threads to perform different tasks throughout the lifetime of the program. By default, the number of threads in the pool is equal to the maximum number of threads that the hardware can run in parallel.
|
||||
The library presented here contains a C++ thread pool class, `BS::thread_pool`, which avoids these issues by creating a fixed pool of threads once and for all, and then continuously reusing the same threads to perform different tasks throughout the lifetime of the program. By default, the number of threads in the pool is equal to the maximum number of threads that the hardware can run in parallel.
|
||||
|
||||
The user submits tasks to be executed into a queue. Whenever a thread becomes available, it retrieves the next task from the queue and executes it. The pool automatically produces an `std::future` for each task, which allows the user to wait for the task to finish executing and/or obtain its eventual return value, if applicable. Threads and tasks are autonomously managed by the pool in the background, without requiring any input from the user aside from submitting the desired tasks.
|
||||
|
||||
The design of this package was guided by four important principles. First, *compactness*: the entire library consists of just one small self-contained header file, with no other components or dependencies. Second, *portability*: the package only utilizes the C++17 standard library, without relying on any compiler extensions or 3rd-party libraries, and is therefore compatible with any modern standards-conforming C++17 compiler on any platform. Third, *ease of use*: the package is extensively documented, and programmers of any level should be able to use it right out of the box.
|
||||
The design of this package was guided by four important principles. First, *compactness*: the entire library consists of just one small self-contained header file, with no other components or dependencies. Second, *portability*: the package only utilizes the C\+\+17 standard library, without relying on any compiler extensions or 3rd-party libraries, and is therefore compatible with any modern standards-conforming C\+\+17 compiler on any platform. Third, *ease of use*: the package is extensively documented, and programmers of any level should be able to use it right out of the box.
|
||||
|
||||
The fourth and final guiding principle is *performance*: each and every line of code in this library was carefully designed with maximum performance in mind, and performance was tested and verified on a variety of compilers and platforms. Indeed, the library was originally designed for use in the author's own computationally-intensive scientific computing projects, running both on high-end desktop/laptop computers and high-performance computing nodes.
|
||||
|
||||
@@ -78,47 +83,51 @@ Other, more advanced multithreading libraries may offer more features and/or hig
|
||||
* 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 ~190 lines of code, excluding comments, blank lines, and the optional helper classes.
|
||||
* Single header file: simply `#include "BS_thread_pool.hpp"` and you're all set!
|
||||
* Header-only: no need to install or build the library.
|
||||
* Self-contained: no external requirements or dependencies.
|
||||
* Portable: uses only the C++ standard library, and works with any C++17-compliant compiler.
|
||||
* Only 247 lines of code, excluding comments, blank lines, and lines containing only a single brace.
|
||||
* A stand-alone "light version" of the C++ thread pool is also available in the `BS_thread_pool_light.hpp` header file, with only 115 lines of code.
|
||||
* **Easy to use:**
|
||||
* Very simple operation, using a handful of member functions.
|
||||
* Every task submitted to the queue using the `submit()` member function automatically generates an `std::future`, which can be used to wait for the task to finish executing and/or obtain its eventual return value.
|
||||
* Optionally, tasks may also be submitted using the `push_task()` member function without generating a future, sacrificing convenience for even greater performance.
|
||||
* Loops can be automatically parallelized into any number of parallel tasks using the `parallelize_loop()` member function, which returns a `BS::multi_future` (see below) that can be used to track the execution of all parallel tasks at once.
|
||||
* If futures are not needed, tasks may be submitted using `push_task()`, and loops can be parallelized using `push_loop()` - sacrificing convenience for even greater performance.
|
||||
* The code is thoroughly documented using Doxygen comments - not only the interface, but also the implementation, in case the user would like to make modifications.
|
||||
* The included test program `BS_thread_pool_test.cpp` can be used to perform exhaustive automated tests and benchmarks, and also serves as a comprehensive example of how to properly use the package.
|
||||
* **Helper classes:**
|
||||
* Automatically parallelize a loop into any number of parallel tasks using the `parallelize_loop()` member function, and track its execution using the `BS::multi_future` helper class.
|
||||
* Track the execution of multiple futures at once using the `BS::multi_future` helper class.
|
||||
* Synchronize output to a stream from multiple threads in parallel using the `BS::synced_stream` helper class.
|
||||
* Easily measure execution time for benchmarking purposes using the `BS::timer` helper class.
|
||||
* **Additional features:**
|
||||
* Easily wait for all tasks in the queue to complete using the `wait_for_tasks()` member function.
|
||||
* Easily wait for all tasks in the queue to complete using the `wait_for_tasks()`, `wait_for_tasks_duration()`, and `wait_for_tasks_until()` member functions.
|
||||
* Change the number of threads in the pool safely and on-the-fly as needed using the `reset()` member function.
|
||||
* Monitor the number of queued and/or running tasks using the `get_tasks_queued()`, `get_tasks_running()`, and `get_tasks_total()` member functions.
|
||||
* Freely pause and resume the pool by modifying the `paused` member variable. When paused, threads do not retrieve new tasks out of the queue.
|
||||
* Catch exceptions thrown by the submitted tasks.
|
||||
* Freely pause and resume the pool using the `pause()`, `unpause()`, and `is_paused()` member functions; when paused, threads do not retrieve new tasks out of the queue.
|
||||
* Purge all tasks currently waiting in the queue with the `purge()` member function.
|
||||
* Catch exceptions thrown by tasks submitted using `submit()` or `parallelize_loop()` from the main thread through their futures.
|
||||
* Submit class member functions to the pool, either applied to a specific object or from within the object itself.
|
||||
* Pass arguments to tasks by value, reference, or constant reference.
|
||||
* Under continuous and active development. Bug reports and feature requests are welcome, and should be made via [GitHub issues](https://github.com/bshoshany/thread-pool/issues).
|
||||
|
||||
### Compiling and compatibility
|
||||
|
||||
This library should successfully compile on any C++17 standard-compliant compiler, on all operating systems and architectures for which such a compiler is available. Compatibility was verified with a 12-core / 24-thread AMD Ryzen 9 3900X CPU using the following compilers and platforms:
|
||||
This library should successfully compile on any C++17 standard-compliant compiler, on all operating systems and architectures for which such a compiler is available. Compatibility was verified with a 24-core (8P+16E) / 32-thread Intel i9-13900K CPU using the following compilers and platforms:
|
||||
|
||||
* Windows 11 build 22000.795:
|
||||
* [Clang](https://clang.llvm.org/) v14.0.6
|
||||
* [GCC](https://gcc.gnu.org/) v12.1.0 ([WinLibs build](https://winlibs.com/))
|
||||
* [MSVC](https://docs.microsoft.com/en-us/cpp/) v19.32.31332
|
||||
* Ubuntu 22.04 LTS:
|
||||
* [Clang](https://clang.llvm.org/) v14.0.0
|
||||
* [GCC](https://gcc.gnu.org/) v12.0.1
|
||||
* Windows 11 build 22621.1702:
|
||||
* [Clang](https://clang.llvm.org/) v16.0.4
|
||||
* [GCC](https://gcc.gnu.org/) v13.1.0 ([WinLibs build](https://winlibs.com/))
|
||||
* [MSVC](https://docs.microsoft.com/en-us/cpp/) v19.36.32532
|
||||
* Ubuntu 23.04:
|
||||
* [Clang](https://clang.llvm.org/) v16.0.0
|
||||
* [GCC](https://gcc.gnu.org/) v13.0.1
|
||||
|
||||
In addition, this library was tested on a [Compute Canada](https://www.computecanada.ca/) node equipped with two 20-core / 40-thread Intel Xeon Gold 6148 CPUs (for a total of 40 cores and 80 threads), running CentOS Linux 7.9.2009, using [GCC](https://gcc.gnu.org/) v12.1.1.
|
||||
In addition, this library was tested on a [Digital Research Alliance of Canada](https://alliancecan.ca/en) node equipped with two 20-core / 40-thread Intel Xeon Gold 6148 CPUs (for a total of 40 cores and 80 threads), running CentOS Linux 7.9.2009, using [GCC](https://gcc.gnu.org/) v12.2.0.
|
||||
|
||||
The test program `BS_thread_pool_test.cpp` was compiled without warnings (with the warning flags `-Wall -Wextra -Wconversion -Wsign-conversion -Wpedantic -Weffc++ -Wshadow` in GCC/Clang and `/W4` in MSVC), executed, and successfully completed all [automated tests](#testing-the-package) and benchmarks using all of the compilers and systems mentioned above.
|
||||
|
||||
As this library requires C++17 features, the code must be compiled with C++17 support:
|
||||
As this library requires C\+\+17 features, the code must be compiled with C\+\+17 support:
|
||||
|
||||
* For Clang or GCC, use the `-std=c++17` flag. On Linux, you will also need to use the `-pthread` flag to enable the POSIX threads library.
|
||||
* For MSVC, use `/std:c++17`, and preferably also `/permissive-` to ensure standards conformance.
|
||||
@@ -151,7 +160,7 @@ On Windows:
|
||||
.\vcpkg install bshoshany-thread-pool:x86-windows bshoshany-thread-pool:x64-windows
|
||||
```
|
||||
|
||||
The thread pool will then be available automatically in the build system you integrated vcpkg with (e.g. MSBuild or CMake). Simply write `#include "BS_thread_pool.hpp"` in any project to use the thread pool, without having to copy to file into the project first. I will update the vcpkg port with each new release, so it will be updated automatically when you run `vcpkg upgrade`.
|
||||
The C++ thread pool will then be available automatically in the build system you integrated vcpkg with (e.g. MSBuild or CMake). Simply write `#include "BS_thread_pool.hpp"` in any project to use the thread pool, without having to copy to file into the project first. I will update the vcpkg port with each new release, so it will be updated automatically when you run `vcpkg upgrade`.
|
||||
|
||||
Please see the [vcpkg repository](https://github.com/microsoft/vcpkg) for more information on how to use vcpkg.
|
||||
|
||||
@@ -163,7 +172,7 @@ If you are using the [Conan](https://conan.io/) C/C++ package manager, please re
|
||||
|
||||
### Including the library
|
||||
|
||||
If you are not using a C++ library manager (such as vcpkg), simply download the [latest release](https://github.com/bshoshany/thread-pool/releases) from the GitHub repository, place the single header file `BS_thread_pool.hpp` in the desired folder, and include it in your program:
|
||||
If you are not using a C++ library manager (such as vcpkg), simply download the [latest release](https://github.com/bshoshany/thread-pool/releases) from the GitHub repository, place the single header file `BS_thread_pool.hpp` from the `include` folder of the repository in the desired folder, and include it in your program:
|
||||
|
||||
```cpp
|
||||
#include "BS_thread_pool.hpp"
|
||||
@@ -210,7 +219,7 @@ std::cout << "Thread pool library version is " << BS_THREAD_POOL_VERSION << ".\n
|
||||
Sample output:
|
||||
|
||||
```none
|
||||
Thread pool library version is v3.1.0 (2022-07-13).
|
||||
Thread pool library version is v3.5.0 (2023-05-25).
|
||||
```
|
||||
|
||||
This can be used, for example, to allow the same code to work with several incompatible versions of the library.
|
||||
@@ -333,7 +342,7 @@ pool.push_task(task, arg);
|
||||
pool.push_task(task, arg1, arg2);
|
||||
```
|
||||
|
||||
**Warning!** Since `push_task()` does not return a future, there is no built-in way for the user to know when the task finishes executing. You must use either `wait_for_tasks()` (see below), or some other method such as condition variables, to ensure that the task finishes executing before trying to use anything that depends on its output. Otherwise, bad things will happen!
|
||||
**Warning:** Since `push_task()` does not return a future, there is no built-in way for the user to know when the task finishes executing. You must use either `wait_for_tasks()` (see below), or some other method such as condition variables, to ensure that the task finishes executing before trying to use anything that depends on its output. Otherwise, bad things will happen!
|
||||
|
||||
### Manually waiting for all tasks to complete
|
||||
|
||||
@@ -369,6 +378,52 @@ after the `for` loop will ensure - as efficiently as possible - that all tasks h
|
||||
|
||||
Note, however, that `wait_for_tasks()` will wait for **all** the tasks in the queue, including those that are unrelated to the `for` loop. Using [`parallelize_loop()`](#parallelizing-loops) would make much more sense in this particular case, as it will allow waiting only for the tasks related to the loop.
|
||||
|
||||
**Warning:** Never call `wait_for_tasks()` from within a thread of the same thread pool, for example `pool.push_task([] { pool.wait_for_tasks(); })`, as that will cause it to wait forever!
|
||||
|
||||
### Waiting with a timeout
|
||||
|
||||
Sometimes you may wish to wait for the tasks to complete, but only for a certain amount of time, or until a specific point in time. For example, if the tasks have not yet completed after some time, you may wish to let the user know that there is a delay. This can be achieved using two member functions:
|
||||
|
||||
* `wait_for_tasks_duration()` waits for the tasks to be completed, but stops waiting after the specified duration, given as an argument of type `std::chrono::duration`, has passed.
|
||||
* `wait_for_tasks_until()` waits for the tasks to be completed, but stops waiting after the specified time point, given as an argument of type `std::chrono::time_point`, has been reached.
|
||||
|
||||
Here is an example:
|
||||
|
||||
```cpp
|
||||
#include "BS_thread_pool.hpp"
|
||||
|
||||
int main()
|
||||
{
|
||||
BS::thread_pool pool;
|
||||
std::atomic<bool> done = false;
|
||||
pool.push_task(
|
||||
[&done]
|
||||
{
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||
done = true;
|
||||
});
|
||||
while (true)
|
||||
{
|
||||
pool.wait_for_tasks_duration(std::chrono::milliseconds(200));
|
||||
if (!done)
|
||||
std::cout << "Sorry, task is not done yet.\n";
|
||||
else
|
||||
break;
|
||||
}
|
||||
std::cout << "Task done!\n";
|
||||
}
|
||||
```
|
||||
|
||||
The output is:
|
||||
|
||||
```none
|
||||
Sorry, task is not done yet.
|
||||
Sorry, task is not done yet.
|
||||
Sorry, task is not done yet.
|
||||
Sorry, task is not done yet.
|
||||
Task done!
|
||||
```
|
||||
|
||||
### Submitting class member functions to the queue
|
||||
|
||||
Consider the following program:
|
||||
@@ -467,7 +522,42 @@ int main()
|
||||
}
|
||||
```
|
||||
|
||||
### Parallelizing loops
|
||||
### Passing task arguments by reference
|
||||
|
||||
In C++, it is often crucial to pass function arguments by reference or constant reference, instead of by value. This allows the function to access the object being passed directly, rather than creating a new copy of the object.
|
||||
|
||||
When submitting a task using `push_task()` or `submit()`, the task's arguments are always passed by value by default. To pass arguments to the task by reference, you must wrap them with `std::ref()`. Similarly, to pass arguments by constant reference, you must wrap them with `std::cref()`. Here is an example:
|
||||
|
||||
```cpp
|
||||
#include "BS_thread_pool.hpp"
|
||||
|
||||
BS::thread_pool pool;
|
||||
|
||||
void increment(int& x)
|
||||
{
|
||||
++x;
|
||||
}
|
||||
|
||||
void print(const int& x)
|
||||
{
|
||||
std::cout << x;
|
||||
}
|
||||
|
||||
int main()
|
||||
{
|
||||
int n = 0;
|
||||
pool.submit(increment, std::ref(n)).wait();
|
||||
pool.submit(print, std::cref(n)).wait();
|
||||
}
|
||||
```
|
||||
|
||||
The `increment()` function takes a **reference** to an integer, and increments that integer. Passing the argument by reference guarantees that `n` itself, in the scope of `main()`, will be incremented - rather than a copy of it in the scope of `increment()`. To pass `n` by reference, we wrapped it inside `std::ref()`. Note that the program will not compile otherwise, since `increment()` only accepts arguments by reference.
|
||||
|
||||
Similarly, the `print()` function takes a **constant reference** to an integer, and prints that integer. Passing the argument by constant reference guarantees that the variable will not be accidentally modified by the function, even though we are accessing `n` itself, rather than a copy. To pass `n` by constant reference, we wrapped it inside `std::cref()`.
|
||||
|
||||
## Parallelizing loops
|
||||
|
||||
### Automatic parallelization of loops
|
||||
|
||||
One of the most common and effective methods of parallelization is splitting a loop into smaller loops and running them in parallel. It is most effective in "embarrassingly parallel" computations, such as vector or matrix operations, where each iteration of the loop is completely independent of every other iteration. For example, if we are summing up two vectors of 1000 elements each, and we have 10 threads, we could split the summation into 10 blocks of 100 elements each, and run all the blocks in parallel, potentially increasing performance by up to a factor of 10.
|
||||
|
||||
@@ -623,51 +713,10 @@ int main()
|
||||
|
||||
As with `parallelize_loop()`, the first argument can be omitted if the start index is 0, and the last argument can be omitted if the number of blocks should be equal to the number of threads.
|
||||
|
||||
**Warning!** Since `push_loop()` does not return a `BS::multi_future`, there is no built-in way for the user to know when the loop finishes executing. You must use either [`wait_for_tasks()`](#manually-waiting-for-all-tasks-to-complete), or some other method such as condition variables, to ensure that the loop finishes executing before trying to use anything that depends on its output. Otherwise, bad things will happen!
|
||||
**Warning:** Since `push_loop()` does not return a `BS::multi_future`, there is no built-in way for the user to know when the loop finishes executing. You must use either [`wait_for_tasks()`](#manually-waiting-for-all-tasks-to-complete), or some other method such as condition variables, to ensure that the loop finishes executing before trying to use anything that depends on its output. Otherwise, bad things will happen!
|
||||
|
||||
## Helper classes
|
||||
|
||||
### Handling multiple futures at once
|
||||
|
||||
The helper class template `BS::multi_future<T>`, already introduced in the context of `parallelize_loop()`, provides a convenient way to collect and access groups of futures. The futures are stored in a public member variable `f` of type `std::vector<std::future<T>>`, so all standard `std::vector` operations are available for organizing the futures. Once the futures are stored, you can use `wait()` to wait for all of them at once or `get()` to get an `std::vector<T>` with the results from all of them. Here's a simple example:
|
||||
|
||||
```cpp
|
||||
#include "BS_thread_pool.hpp"
|
||||
|
||||
int square(const int i)
|
||||
{
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||
return i * i;
|
||||
};
|
||||
|
||||
int main()
|
||||
{
|
||||
BS::thread_pool pool;
|
||||
BS::multi_future<int> mf1;
|
||||
BS::multi_future<int> mf2;
|
||||
for (int i = 0; i < 100; ++i)
|
||||
mf1.f.push_back(pool.submit(square, i));
|
||||
for (int i = 100; i < 200; ++i)
|
||||
mf2.f.push_back(pool.submit(square, i));
|
||||
/// ...
|
||||
/// Do some stuff while the first group of tasks executes...
|
||||
/// ...
|
||||
const std::vector<int> squares1 = mf1.get();
|
||||
std::cout << "Results from the first group:" << '\n';
|
||||
for (const int s : squares1)
|
||||
std::cout << s << ' ';
|
||||
/// ...
|
||||
/// Do other stuff while the second group of tasks executes...
|
||||
/// ...
|
||||
const std::vector<int> squares2 = mf2.get();
|
||||
std::cout << '\n' << "Results from the second group:" << '\n';
|
||||
for (const int s : squares2)
|
||||
std::cout << s << ' ';
|
||||
}
|
||||
```
|
||||
|
||||
In this example, we simulate complicated tasks by having each task wait for 500ms before returning its result. We collect the futures of the tasks submitted within each loop into two separate `BS::multi_future<int>` objects. `mf1` holds the results from the first loop, and `mf2` holds the results from the second loop. Now we can wait for and/or get the results from `mf1` whenever is convenient, and separately wait for and/or get the results from `mf2` at another time.
|
||||
|
||||
### 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:
|
||||
@@ -732,6 +781,94 @@ Task no. 5 executing.
|
||||
|
||||
**Warning:** Always create the `BS::synced_stream` object **before** the `BS::thread_pool` object, as we did in this example. When the `BS::thread_pool` object goes out of scope, it waits for the remaining tasks to be executed. If the `BS::synced_stream` object goes out of scope before the `BS::thread_pool` object, then any tasks using the `BS::synced_stream` will crash. Since objects are destructed in the opposite order of construction, creating the `BS::synced_stream` object before the `BS::thread_pool` object ensures that the `BS::synced_stream` is always available to the tasks, even while the pool is destructing.
|
||||
|
||||
Most stream manipulators defined in the headers `<ios>` and `<iomanip>`, such as `std::setw` (set the character width of the next output), `std::setprecision` (set the precision of floating point numbers), and `std::fixed` (display floating point numbers with a fixed number of digits), can be passed to `print()` and `println()` just as you would pass them to a stream.
|
||||
|
||||
The only exceptions are the flushing manipulators `std::endl` and `std::flush`, which will not work because the compiler will not be able to figure out which template specializations to use. Instead, use `BS::synced_stream::endl` and `BS::synced_stream::flush`. Here is an example:
|
||||
|
||||
```cpp
|
||||
#include "BS_thread_pool.hpp"
|
||||
#include <cmath>
|
||||
#include <iomanip>
|
||||
|
||||
int main()
|
||||
{
|
||||
BS::synced_stream sync_out;
|
||||
BS::thread_pool pool;
|
||||
sync_out.print(std::setprecision(10), std::fixed);
|
||||
for (size_t i = 1; i <= 10; ++i)
|
||||
pool.push_task([i, &sync_out] { sync_out.print("The square root of ", std::setw(2), i, " is ", std::sqrt(i), ".", BS::synced_stream::endl); });
|
||||
}
|
||||
```
|
||||
|
||||
### Handling multiple futures at once
|
||||
|
||||
The helper class template `BS::multi_future<T>`, already introduced in the context of `parallelize_loop()`, provides a convenient way to collect and access groups of futures. This class works similarly to STL containers such as `std::vector`:
|
||||
|
||||
* When you create a new object, either use the default constructor to create an empty object and add futures to it later, or pass the desired number of futures to the constructor in advance.
|
||||
* Use the `[]` operator to access the future at a specific index, or the `push_back()` member function to append a new future to the list.
|
||||
* The `size()` member function tells you how many futures are currently stored in the object.
|
||||
* Once all the futures are stored, you can use `wait()` to wait for all of them at once or `get()` to get an `std::vector<T>` with the results from all of them.
|
||||
|
||||
Aside from using `BS::multi_future` to track the execution of parallelized loops, it can also be used whenever you have several different groups of tasks and you want to track the execution of each group individually. Here's a simple example:
|
||||
|
||||
```cpp
|
||||
#include "BS_thread_pool.hpp"
|
||||
#include <cmath>
|
||||
|
||||
BS::synced_stream sync_out;
|
||||
BS::thread_pool pool;
|
||||
|
||||
double power(const double i, const double j)
|
||||
{
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10 * pool.get_thread_count()));
|
||||
return std::pow(i, j);
|
||||
}
|
||||
|
||||
void print_vector(const std::vector<double>& vec)
|
||||
{
|
||||
for (const double i : vec)
|
||||
sync_out.print(i, ' ');
|
||||
sync_out.println();
|
||||
}
|
||||
|
||||
int main()
|
||||
{
|
||||
constexpr size_t n = 100;
|
||||
|
||||
// First group of tasks: calculate n squares.
|
||||
// Here we create an empty BS::multi_future object, and append futures to it via push_back().
|
||||
BS::multi_future<double> mf_squares;
|
||||
for (int i = 0; i < n; ++i)
|
||||
mf_squares.push_back(pool.submit(power, i, 2));
|
||||
|
||||
// Second group of tasks: calculate n cubes.
|
||||
// In this case, we create a BS::multi_future object of the desired size in advance, and store the futures via the [] operator. This is faster since there will be no memory reallocations, but also more prone to errors.
|
||||
BS::multi_future<double> mf_cubes(n);
|
||||
for (int i = 0; i < n; ++i)
|
||||
mf_cubes[i] = pool.submit(power, i, 3);
|
||||
|
||||
// Both groups are now queued, but it will take some time until they all execute.
|
||||
|
||||
/// ...
|
||||
/// Do some stuff while the first group of tasks executes...
|
||||
/// ...
|
||||
|
||||
// Get and print the results from the first group.
|
||||
sync_out.println("Squares:");
|
||||
print_vector(mf_squares.get());
|
||||
|
||||
/// ...
|
||||
/// Do other stuff while the second group of tasks executes...
|
||||
/// ...
|
||||
|
||||
// Get and print the results from the second group.
|
||||
sync_out.println("Cubes:");
|
||||
print_vector(mf_cubes.get());
|
||||
}
|
||||
```
|
||||
|
||||
In this example, we simulate complicated tasks by having each task wait for a bit before returning its result. We collect the futures of the tasks submitted within each group into two separate `BS::multi_future<double>` objects. `mf_squares` holds the results from the first group, and `mf_cubes` holds the results from the second group. Now we can wait for and/or get the results from `mf_squares` whenever is convenient, and separately wait for and/or get the results from `mf_cubes` at another time.
|
||||
|
||||
### Measuring execution time
|
||||
|
||||
If you are using a thread pool, then your code is most likely performance-critical. Achieving maximum performance requires performing a considerable amount of benchmarking to determine the optimal settings and algorithms. Therefore, it is important to be able to measure the execution time of various computations and operations under different conditions.
|
||||
@@ -822,9 +959,9 @@ Task 11 done.
|
||||
|
||||
### Pausing the workers
|
||||
|
||||
Sometimes you may wish to temporarily pause the execution of tasks, or perhaps you want to submit tasks to the queue in advance and only start executing them at a later time. You can do this using the public member variable `paused`.
|
||||
Sometimes you may wish to temporarily pause the execution of tasks, or perhaps you want to submit tasks to the queue in advance and only start executing them at a later time. You can do this using the member functions `pause()`, `unpause()`, and `is_paused()`.
|
||||
|
||||
When `paused` is set to `true`, the workers will temporarily stop retrieving 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 retrieving tasks, set `paused` back to its default value of `false`.
|
||||
When you call `pause()`, the workers will temporarily stop retrieving 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 retrieving tasks, call `unpause()`. To check whether the pool is currently paused, call `is_paused()`.
|
||||
|
||||
Here is an example:
|
||||
|
||||
@@ -840,14 +977,22 @@ void sleep_half_second(const size_t i)
|
||||
sync_out.println("Task ", i, " done.");
|
||||
}
|
||||
|
||||
void check_if_paused()
|
||||
{
|
||||
if (pool.is_paused())
|
||||
sync_out.println("Pool paused.");
|
||||
else
|
||||
sync_out.println("Pool unpaused.");
|
||||
}
|
||||
|
||||
int main()
|
||||
{
|
||||
for (size_t i = 0; i < 8; ++i)
|
||||
pool.push_task(sleep_half_second, i);
|
||||
sync_out.println("Submitted 8 tasks.");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(250));
|
||||
pool.paused = true;
|
||||
sync_out.println("Pool paused.");
|
||||
pool.pause();
|
||||
check_if_paused();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||
sync_out.println("Still paused...");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||
@@ -856,8 +1001,8 @@ int main()
|
||||
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.");
|
||||
pool.unpause();
|
||||
check_if_paused();
|
||||
}
|
||||
```
|
||||
|
||||
@@ -873,7 +1018,7 @@ Task 3 done.
|
||||
Still paused...
|
||||
Submitted 4 more tasks.
|
||||
Still paused...
|
||||
Pool resumed.
|
||||
Pool unpaused.
|
||||
Task 4 done.
|
||||
Task 5 done.
|
||||
Task 6 done.
|
||||
@@ -884,7 +1029,7 @@ 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.
|
||||
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 unpaused, 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:
|
||||
|
||||
@@ -900,6 +1045,14 @@ void sleep_half_second(const size_t i)
|
||||
sync_out.println("Task ", i, " done.");
|
||||
}
|
||||
|
||||
void check_if_paused()
|
||||
{
|
||||
if (pool.is_paused())
|
||||
sync_out.println("Pool paused.");
|
||||
else
|
||||
sync_out.println("Pool unpaused.");
|
||||
}
|
||||
|
||||
int main()
|
||||
{
|
||||
for (size_t i = 0; i < 8; ++i)
|
||||
@@ -910,8 +1063,9 @@ int main()
|
||||
pool.push_task(sleep_half_second, i);
|
||||
sync_out.println("Submitted 12 more tasks.");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(250));
|
||||
pool.paused = true;
|
||||
sync_out.println("Pool paused. Waiting for the ", pool.get_tasks_running(), " running tasks to complete.");
|
||||
pool.pause();
|
||||
check_if_paused();
|
||||
sync_out.println("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));
|
||||
@@ -919,9 +1073,10 @@ int main()
|
||||
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;
|
||||
pool.unpause();
|
||||
check_if_paused();
|
||||
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.");
|
||||
sync_out.println("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.");
|
||||
}
|
||||
@@ -940,7 +1095,8 @@ Task 5 done.
|
||||
Task 6 done.
|
||||
Task 7 done.
|
||||
Submitted 12 more tasks.
|
||||
Pool paused. Waiting for the 4 running tasks to complete.
|
||||
Pool paused.
|
||||
Waiting for the 4 running tasks to complete.
|
||||
Task 8 done.
|
||||
Task 9 done.
|
||||
Task 10 done.
|
||||
@@ -948,7 +1104,8 @@ 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.
|
||||
Pool unpaused.
|
||||
Waiting for the remaining 8 tasks (4 running and 4 queued) to complete.
|
||||
Task 12 done.
|
||||
Task 13 done.
|
||||
Task 14 done.
|
||||
@@ -960,9 +1117,51 @@ Task 19 done.
|
||||
All tasks completed.
|
||||
```
|
||||
|
||||
The first `wait_for_tasks()`, which was called with `paused == false`, waited for all 8 tasks, both running and queued. The second `wait_for_tasks()`, which was called with `paused == true`, only waited for the 4 running tasks, while the other 8 tasks remained queued, and were not executed since the pool was paused. Finally, the third `wait_for_tasks()`, which was called with `paused == false`, waited for the remaining 8 tasks, both running and queued.
|
||||
The first `wait_for_tasks()`, which was called while the pool was not paused, waited for all 8 tasks, both running and queued. The second `wait_for_tasks()`, which was called after pausing the pool, 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 after unpausing the pool, 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!
|
||||
**Warning:** If the thread pool is destroyed while paused, any tasks still in the queue will never be executed!
|
||||
|
||||
### Purging tasks
|
||||
|
||||
Consider a situation where the user cancels a multi-threaded operation while it is still ongoing. Perhaps the operation was split into multiple tasks, and half of the tasks are currently being executed by the pool's threads, but the other half are still waiting in the queue.
|
||||
|
||||
The thread pool cannot terminate the tasks that are already running, as the C++17 standard does not provide that functionality (and in any case, abruptly terminating a task while it's running could have extremely bad consequences, such as memory leaks and data corruption). However, the tasks that are still waiting in the queue can be purged using the `purge()` member function.
|
||||
|
||||
Once `purge()` is called, any tasks still waiting in the queue will be discarded, and will never be executed by the threads. Please note that there is no way to restore the purged tasks; they are gone forever.
|
||||
|
||||
Consider for example the following program:
|
||||
|
||||
```cpp
|
||||
#include "BS_thread_pool.hpp"
|
||||
|
||||
BS::synced_stream sync_out;
|
||||
BS::thread_pool pool(4);
|
||||
|
||||
int main()
|
||||
{
|
||||
for (size_t i = 0; i < 8; ++i)
|
||||
{
|
||||
pool.push_task(
|
||||
[i]
|
||||
{
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(100));
|
||||
sync_out.println("Task ", i, " done.");
|
||||
});
|
||||
}
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
pool.purge();
|
||||
pool.wait_for_tasks();
|
||||
}
|
||||
```
|
||||
|
||||
The program submit 8 tasks to the queue. Each task waits 100 milliseconds and then prints a message. The thread pool has 4 threads, so it will execute the first 4 tasks in parallel, and then the remaining 4. We wait 50 milliseconds, to ensure that the first 4 tasks have all started running. Then we call `purge()` to purge the remaining 4 tasks. As a result, these tasks never get executed. However, since the first 4 tasks are still running when `purge()` is called, they will finish uninterrupted; `purge()` only discards tasks that have not yet started running. The output of the program therefore only contains the messages from the first 4 tasks:
|
||||
|
||||
```none
|
||||
Task 0 done.
|
||||
Task 1 done.
|
||||
Task 2 done.
|
||||
Task 3 done.
|
||||
```
|
||||
|
||||
### Exception handling
|
||||
|
||||
@@ -1041,324 +1240,67 @@ When using `BS::multi_future` to handle multiple futures at once, exception hand
|
||||
|
||||
## Testing the package
|
||||
|
||||
The included file `BS_thread_pool_test.cpp` will perform automated tests of all aspects of the package, and perform simple benchmarks. The output will be printed both to `std::cout` and to a file named `BS_thread_pool_test-yyyy-mm-dd_hh.mm.ss.log` based on the current date and time. In addition, the code is thoroughly documented, and is meant to serve as an extensive example of how to properly use the package.
|
||||
### Automated tests
|
||||
|
||||
The file `BS_thread_pool_test.cpp` in the `tests` folder of the GitHub repository will perform automated tests of all aspects of the package. The output will be printed both to `std::cout` and to a file named `BS_thread_pool_test-yyyy-mm-dd_hh.mm.ss.log` based on the current date and time. In addition, the code is thoroughly documented, and is meant to serve as an extensive example of how to properly use the package.
|
||||
|
||||
Please make sure to:
|
||||
|
||||
1. [Compile](#compiling-and-compatibility) `BS_thread_pool_test.cpp` with optimization flags enabled (e.g. `-O3` on GCC / Clang or `/O2` on MSVC).
|
||||
2. Run the test without any other applications, especially multithreaded applications, running in parallel.
|
||||
|
||||
A PowerShell script, `BS_thread_pool_test.ps1`, is provided for your convenience in the `tests` folder to make running the test on multiple compilers and operating systems easier. Since it is written in PowerShell, it is fully portable and works on Windows, Linux, and macOS. The script will automatically detect if Clang, GCC, and/or MSVC are available, compile the test program using each available compiler, and then run each compiled test program 5 times and report on any errors.
|
||||
|
||||
If any of the tests fail, please [submit a bug report](https://github.com/bshoshany/thread-pool/issues) including the exact specifications of your system (OS, CPU, compiler, etc.) and the generated log file.
|
||||
|
||||
### Automated tests
|
||||
|
||||
A sample output of a successful run of the automated tests is as follows:
|
||||
|
||||
```none
|
||||
BS::thread_pool: a fast, lightweight, and easy-to-use C++17 thread pool library
|
||||
(c) 2022 Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
|
||||
GitHub: https://github.com/bshoshany/thread-pool
|
||||
|
||||
Thread pool library version is v3.2.0 (2022-07-28).
|
||||
Hardware concurrency is 24.
|
||||
Generating log file: BS_thread_pool_test-2022-07-28_15.29.31.log.
|
||||
|
||||
Important: Please do not run any other applications, especially multithreaded applications, in parallel with this test!
|
||||
|
||||
====================================
|
||||
Checking that the constructor works:
|
||||
====================================
|
||||
Checking that the thread pool reports a number of threads equal to the hardware concurrency...
|
||||
Expected: 24, obtained: 24 -> PASSED!
|
||||
Checking that the manually counted number of unique thread IDs is equal to the reported number of threads...
|
||||
Expected: 24, obtained: 24 -> PASSED!
|
||||
|
||||
============================
|
||||
Checking that reset() works:
|
||||
============================
|
||||
Checking that after reset() the thread pool reports a number of threads equal to half the hardware concurrency...
|
||||
Expected: 12, obtained: 12 -> PASSED!
|
||||
Checking that after reset() the manually counted number of unique thread IDs is equal to the reported number of threads...
|
||||
Expected: 12, obtained: 12 -> PASSED!
|
||||
Checking that after a second reset() the thread pool reports a number of threads equal to the hardware concurrency...
|
||||
Expected: 24, obtained: 24 -> PASSED!
|
||||
Checking that after a second reset() the manually counted number of unique thread IDs is equal to the reported number of threads...
|
||||
Expected: 24, obtained: 24 -> PASSED!
|
||||
|
||||
================================
|
||||
Checking that push_task() works:
|
||||
================================
|
||||
Checking that push_task() works for a function with no arguments or return value...
|
||||
-> PASSED!
|
||||
Checking that push_task() works for a function with one argument and no return value...
|
||||
-> PASSED!
|
||||
Checking that push_task() works for a function with two arguments and no return value...
|
||||
-> PASSED!
|
||||
|
||||
=============================
|
||||
Checking that submit() works:
|
||||
=============================
|
||||
Checking that submit() works for a function with no arguments or return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a function with one argument and no return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a function with two arguments and no return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a function with no arguments and a return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a function with one argument and a return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a function with two arguments and a return value...
|
||||
-> PASSED!
|
||||
|
||||
================================================
|
||||
Checking that submitting member functions works:
|
||||
================================================
|
||||
Checking that push_task() works for a member function with no arguments or return value...
|
||||
-> PASSED!
|
||||
Checking that push_task() works for a member function with one argument and no return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a member function with no arguments or return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a member function with one argument and no return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a member function with no arguments and a return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works for a member function with one argument and a return value...
|
||||
-> PASSED!
|
||||
|
||||
======================================================================
|
||||
Checking that submitting member functions from within an object works:
|
||||
======================================================================
|
||||
Checking that push_task() works within an object for a member function with no arguments or return value...
|
||||
-> PASSED!
|
||||
Checking that push_task() works within an object for a member function with one argument and no return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works within an object for a member function with no arguments or return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works within an object for a member function with one argument and no return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works within an object for a member function with no arguments and a return value...
|
||||
-> PASSED!
|
||||
Checking that submit() works within an object for a member function with one argument and a return value...
|
||||
-> PASSED!
|
||||
|
||||
=======================================
|
||||
Checking that wait_for_tasks() works...
|
||||
=======================================
|
||||
Waiting for tasks...
|
||||
-> PASSED!
|
||||
|
||||
======================================================
|
||||
Checking that push_loop() and parallelize_loop() work:
|
||||
======================================================
|
||||
Verifying that push_loop() from 390892 to 541943 with 20 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from -233617 to 52646 with 22 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from 409845 to 410887 with 23 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from 977764 to 726111 with 12 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from -940107 to -882673 with 18 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from -613072 to 675872 with 10 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from 998082 to -267173 with 3 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from -779960 to 518984 with 4 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from -52424 to 2576 with 3 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that push_loop() from 131005 to -557044 with 10 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from -119788 to -727738 with 14 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from -911425 to -923429 with 8 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 49601 to -772605 with 16 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from -322809 to -66366 with 15 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 880099 to -150434 with 22 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 984341 to 69159 with 11 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from -207913 to 829987 with 9 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 297749 to -332031 with 2 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 517539 to 811728 with 8 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 311585 to -81170 with 21 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 670932 to 740527 with 3 tasks correctly sums all indices...
|
||||
Expected: 98230419510, obtained: 98230419510 -> PASSED!
|
||||
Verifying that parallelize_loop() from 645122 to -546036 with 11 tasks correctly sums all indices...
|
||||
Expected: 118025890430, obtained: 118025890430 -> PASSED!
|
||||
Verifying that parallelize_loop() from 251002 to -903037 with 6 tasks correctly sums all indices...
|
||||
Expected: -752474973404, obtained: -752474973404 -> PASSED!
|
||||
Verifying that parallelize_loop() from -222340 to -791805 with 15 tasks correctly sums all indices...
|
||||
Expected: -577520651890, obtained: -577520651890 -> PASSED!
|
||||
Verifying that parallelize_loop() from 937604 to 819765 with 3 tasks correctly sums all indices...
|
||||
Expected: 207086487752, obtained: 207086487752 -> PASSED!
|
||||
Verifying that parallelize_loop() from -53900 to -339294 with 24 tasks correctly sums all indices...
|
||||
Expected: -112215493830, obtained: -112215493830 -> PASSED!
|
||||
Verifying that parallelize_loop() from 65429 to 903556 with 22 tasks correctly sums all indices...
|
||||
Expected: 812131652968, obtained: 812131652968 -> PASSED!
|
||||
Verifying that parallelize_loop() from 519293 to -332413 with 19 tasks correctly sums all indices...
|
||||
Expected: 159165965574, obtained: 159165965574 -> PASSED!
|
||||
Verifying that parallelize_loop() from 165831 to 47482 with 18 tasks correctly sums all indices...
|
||||
Expected: 25245261888, obtained: 25245261888 -> PASSED!
|
||||
Verifying that parallelize_loop() from -534648 to 475350 with 13 tasks correctly sums all indices...
|
||||
Expected: -59891871402, obtained: -59891871402 -> PASSED!
|
||||
Verifying that parallelize_loop() with identical start and end indices does nothing...
|
||||
-> PASSED!
|
||||
Trying parallelize_loop() with start and end indices of different types:
|
||||
Verifying that parallelize_loop() from -962605 to 21974 with 17 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Trying the overloads for push_loop() and parallelize_loop() for the case where the first index is equal to 0:
|
||||
Verifying that push_loop() from 0 to 482251 with 7 tasks modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 0 to 151431 with 1 task modifies all indices...
|
||||
-> PASSED!
|
||||
Verifying that parallelize_loop() from 0 to 806998 with 4 tasks correctly sums all indices...
|
||||
Expected: 651244965006, obtained: 651244965006 -> PASSED!
|
||||
|
||||
====================================
|
||||
Checking that task monitoring works:
|
||||
====================================
|
||||
Resetting pool to 4 threads.
|
||||
Submitting 12 tasks.
|
||||
After submission, should have: 12 tasks total, 4 tasks running, 8 tasks queued...
|
||||
Result: 12 tasks total, 4 tasks running, 8 tasks queued -> PASSED!
|
||||
Task 1 released.
|
||||
Task 2 released.
|
||||
Task 3 released.
|
||||
Task 0 released.
|
||||
After releasing 4 tasks, should have: 8 tasks total, 4 tasks running, 4 tasks queued...
|
||||
Result: 8 tasks total, 4 tasks running, 4 tasks queued -> PASSED!
|
||||
Task 5 released.
|
||||
Task 4 released.
|
||||
Task 7 released.
|
||||
Task 6 released.
|
||||
After releasing 4 more tasks, should have: 4 tasks total, 4 tasks running, 0 tasks queued...
|
||||
Result: 4 tasks total, 4 tasks running, 0 tasks queued -> PASSED!
|
||||
Task 9 released.
|
||||
Task 11 released.
|
||||
Task 10 released.
|
||||
Task 8 released.
|
||||
After releasing the final 4 tasks, should have: 0 tasks total, 0 tasks running, 0 tasks queued...
|
||||
Result: 0 tasks total, 0 tasks running, 0 tasks queued -> PASSED!
|
||||
Resetting pool to 24 threads.
|
||||
|
||||
============================
|
||||
Checking that pausing works:
|
||||
============================
|
||||
Resetting pool to 4 threads.
|
||||
Pausing pool.
|
||||
Submitting 12 tasks, each one waiting for 200ms.
|
||||
Immediately after submission, should have: 12 tasks total, 0 tasks running, 12 tasks queued...
|
||||
Result: 12 tasks total, 0 tasks running, 12 tasks queued -> PASSED!
|
||||
300ms later, should still have: 12 tasks total, 0 tasks running, 12 tasks queued...
|
||||
Result: 12 tasks total, 0 tasks running, 12 tasks queued -> PASSED!
|
||||
Unpausing pool.
|
||||
Task 3 done.
|
||||
Task 1 done.
|
||||
Task 2 done.
|
||||
Task 0 done.
|
||||
300ms later, should have: 8 tasks total, 4 tasks running, 4 tasks queued...
|
||||
Result: 8 tasks total, 4 tasks running, 4 tasks queued -> PASSED!
|
||||
Pausing pool and using wait_for_tasks() to wait for the running tasks.
|
||||
Task 4 done.
|
||||
Task 6 done.
|
||||
Task 7 done.
|
||||
Task 5 done.
|
||||
After waiting, should have: 4 tasks total, 0 tasks running, 4 tasks queued...
|
||||
Result: 4 tasks total, 0 tasks running, 4 tasks queued -> PASSED!
|
||||
200ms later, should still have: 4 tasks total, 0 tasks running, 4 tasks queued...
|
||||
Result: 4 tasks total, 0 tasks running, 4 tasks queued -> PASSED!
|
||||
Unpausing pool and using wait_for_tasks() to wait for all tasks.
|
||||
Task 9 done.
|
||||
Task 8 done.
|
||||
Task 10 done.
|
||||
Task 11 done.
|
||||
After waiting, should have: 0 tasks total, 0 tasks running, 0 tasks queued...
|
||||
Result: 0 tasks total, 0 tasks running, 0 tasks queued -> PASSED!
|
||||
Resetting pool to 24 threads.
|
||||
|
||||
=======================================
|
||||
Checking that exception handling works:
|
||||
=======================================
|
||||
Checking that exceptions are forwarded correctly by submit()...
|
||||
Throwing exception...
|
||||
-> PASSED!
|
||||
Checking that exceptions are forwarded correctly by BS::multi_future...
|
||||
Throwing exception...
|
||||
Throwing exception...
|
||||
-> PASSED!
|
||||
|
||||
============================================================
|
||||
Testing that vector operations produce the expected results:
|
||||
============================================================
|
||||
Adding two vectors with 767202 elements using 4 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 3575 elements using 3 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 392555 elements using 11 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 754640 elements using 16 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 516335 elements using 9 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 564723 elements using 17 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 558475 elements using 15 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 447497 elements using 21 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 121486 elements using 19 tasks...
|
||||
-> PASSED!
|
||||
Adding two vectors with 324254 elements using 24 tasks...
|
||||
-> PASSED!
|
||||
|
||||
++++++++++++++++++++++++++++++
|
||||
SUCCESS: Passed all 85 checks!
|
||||
++++++++++++++++++++++++++++++
|
||||
```
|
||||
|
||||
### Performance tests
|
||||
|
||||
If all checks passed, `BS_thread_pool_test.cpp` will perform simple benchmarks by filling a specific number of vectors of fixed size with random values. The program decides how many vectors to use by testing how many are needed to reach a target duration in the single-threaded test. This ensures that the test takes approximately the same amount of time on different systems, and is thus more consistent and portable.
|
||||
If all checks passed, `BS_thread_pool_test.cpp` performs simple benchmarks by filling a specific number of vectors of fixed size with values. The program decides how many vectors to use, and of what size, by testing how many are needed to reach a certain target duration in a single-threaded computation. This ensures that the test takes approximately the same amount of time on all systems, and is thus more consistent and portable.
|
||||
|
||||
Once the required number of vectors has been determined, the program will test the performance of several multi-threaded tests, dividing the total number of vectors into different numbers of tasks, compare them to the performance of the single-threaded test, and indicate the maximum speedup obtained.
|
||||
Once the appropriate number and size of vectors has been determined, the program allocates the vectors and fills them with values, calculated according to a fixed prescription. This operation is performed both single-threaded and multithreaded, with the multithreaded computation spread across multiple tasks submitted to the pool.
|
||||
|
||||
Please note that these benchmarks are only intended to demonstrate that the package can provide a significant speedup, and it is highly recommended to perform your own benchmarks with your specific system, compiler, and code.
|
||||
Several different multithreaded tests are performed, with the number of tasks either equal to, smaller than, or larger than the pool's thread count. Each test is repeated multiple times, with the run times averaged over all runs of the same test. The run times of the tests are compared, and the maximum speedup obtained is calculated.
|
||||
|
||||
Here we will present the results of the performance test running on a high-end desktop computer equipped with a 12-core / 24-thread AMD Ryzen 9 3900X CPU at 3.8 GHz and 32 GB of DDR4 RAM at 3600 MHz, compiled using [MSVC](https://docs.microsoft.com/en-us/cpp/) v19.32.31332 on Windows 11 build 22000.795 with the `/O2` compiler flag. The output was as follows:
|
||||
As an example, here are the results of the benchmarks from a [Digital Research Alliance of Canada](https://alliancecan.ca/en) node equipped with two 20-core / 40-thread Intel Xeon Gold 6148 CPUs (for a total of 40 cores and 80 threads), running CentOS Linux 7.9.2009. The tests were compiled using GCC v12.2.0 with the `-O3` and `-march=native` flags. The output was as follows:
|
||||
|
||||
```none
|
||||
======================
|
||||
Performing benchmarks:
|
||||
======================
|
||||
Using 24 threads.
|
||||
Using 80 threads.
|
||||
Each test will be repeated 20 times to collect reliable statistics.
|
||||
Determining the number and size of vectors to generate in order to achieve an approximate mean execution time of 50 ms with 24 tasks...
|
||||
Generating 3312 vectors with 4096 elements each:
|
||||
Single-threaded, mean execution time was 542.2 ms with standard deviation 5.8 ms.
|
||||
With 6 tasks, mean execution time was 95.2 ms with standard deviation 1.7 ms.
|
||||
With 12 tasks, mean execution time was 49.6 ms with standard deviation 0.7 ms.
|
||||
With 24 tasks, mean execution time was 29.0 ms with standard deviation 2.9 ms.
|
||||
With 48 tasks, mean execution time was 33.2 ms with standard deviation 4.3 ms.
|
||||
With 96 tasks, mean execution time was 35.5 ms with standard deviation 1.9 ms.
|
||||
Maximum speedup obtained by multithreading vs. single-threading: 18.7x, using 24 tasks.
|
||||
Determining the number and size of vectors to generate in order to achieve an approximate mean execution time of 50 ms with 80 tasks...
|
||||
Generating 3840 vectors with 5120 elements each:
|
||||
Single-threaded, mean execution time was 2122.5 ms with standard deviation 18.1 ms.
|
||||
With 20 tasks, mean execution time was 119.8 ms with standard deviation 13.3 ms.
|
||||
With 40 tasks, mean execution time was 68.9 ms with standard deviation 0.3 ms.
|
||||
With 80 tasks, mean execution time was 45.5 ms with standard deviation 6.2 ms.
|
||||
With 160 tasks, mean execution time was 39.3 ms with standard deviation 3.3 ms.
|
||||
With 320 tasks, mean execution time was 40.4 ms with standard deviation 2.4 ms.
|
||||
Maximum speedup obtained by multithreading vs. single-threading: 53.9x, using 160 tasks.
|
||||
|
||||
+++++++++++++++++++++++++++++++++++++++
|
||||
Thread pool performance test completed!
|
||||
+++++++++++++++++++++++++++++++++++++++
|
||||
```
|
||||
|
||||
This CPU has 12 physical cores, with each core providing two separate logical cores via hyperthreading, for a total of 24 threads. Without hyperthreading, we would expect a maximum theoretical speedup of 12x. With hyperthreading, one might naively expect to achieve up to a 24x speedup, but this is in fact impossible, as both logical cores share the same physical core's resources. However, generally we would expect [an estimated 30% additional speedup](https://software.intel.com/content/www/us/en/develop/articles/how-to-determine-the-effectiveness-of-hyper-threading-technology-with-an-application.html) from hyperthreading, which amounts to around 15.6x in this case. In our performance test, we see a speedup of 18.7x, saturating and even surpassing this estimated theoretical upper bound.
|
||||
These two CPUs have 40 physical cores in total, with each core providing two separate logical cores via hyperthreading, for a total of 80 threads. Without hyperthreading, we would expect a maximum theoretical speedup of 40x. With hyperthreading, one might naively expect to achieve up to an 80x speedup, but this is in fact impossible, as each pair of hyperthreaded logical cores share the same physical core's resources. However, generally we would expect at most an estimated 30% additional speedup from hyperthreading, which amounts to around 52x in this case. The speedup of 53.9x in our performance test exceeds this estimate.
|
||||
|
||||
## The light version of the package
|
||||
|
||||
This package started out as a very lightweight C++ thread pool, but over time has expanded to include many additional features and helper classes. Therefore, I have decided to bundle a light version of the thread pool in a separate and stand-alone header file, `BS_thread_pool_light.hpp`.
|
||||
|
||||
This file does not contain any of the helper classes, only a new `BS::thread_pool_light` class, which is a minimal thread pool with only the 5 most basic member functions:
|
||||
|
||||
* `get_thread_count()`
|
||||
* `push_loop()`
|
||||
* `push_task()`
|
||||
* `submit()`
|
||||
* `wait_for_tasks()`
|
||||
|
||||
Note that each header file is 100% stand-alone. If you wish to use the full package, you only need `BS_thread_pool.hpp`, and if you wish to use the light version, you only need `BS_thread_pool_light.hpp`. Only a single header file needs to be included in your project. However, if you wish to use both the light and non-light thread pool classes in the same project, you can include both header files.
|
||||
|
||||
The test program `BS_thread_pool_test.cpp` tests both the full and the light versions of the package. If needed, the current version of the light thread pool can be obtained using the macro `BS_THREAD_POOL_LIGHT_VERSION`.
|
||||
|
||||
## About the project
|
||||
|
||||
@@ -1378,9 +1320,9 @@ If you found this project useful, please consider [starring it on GitHub](https:
|
||||
|
||||
### Copyright and citing
|
||||
|
||||
Copyright (c) 2022 [Barak Shoshany](http://baraksh.com). Licensed under the [MIT license](LICENSE.txt).
|
||||
Copyright (c) 2023 [Barak Shoshany](http://baraksh.com). Licensed under the [MIT license](LICENSE.txt).
|
||||
|
||||
If you use the library in software of any kind, please provide a link to [the GitHub repository](https://github.com/bshoshany/thread-pool) in the source code and documentation.
|
||||
If you use this C++ thread pool library in software of any kind, please provide a link to [the GitHub repository](https://github.com/bshoshany/thread-pool) in the source code and documentation.
|
||||
|
||||
If you use this library in published research, please cite it as follows:
|
||||
|
||||
@@ -1405,3 +1347,7 @@ You can use the following BibTeX entry:
|
||||
```
|
||||
|
||||
Please note that the [companion paper on arXiv](https://arxiv.org/abs/2105.00613) is updated infrequently. The paper is intended to facilitate discovery of the package by scientists who may find it useful for scientific computing purposes and to allow citing the package in scientific research, but most users should read the `README.md` file on [the GitHub repository](https://github.com/bshoshany/thread-pool) instead, as it is guaranteed to always be up to date.
|
||||
|
||||
### Learning more about C++
|
||||
|
||||
Beginner C++ programmers may be interested in [my lecture notes](https://baraksh.com/CSE701/notes.php) for a course taught at McMaster University, which teach modern C and C++ from scratch, including some of the advanced techniques and programming practices used in developing this library.
|
||||
|
||||
@@ -3,16 +3,15 @@
|
||||
/**
|
||||
* @file BS_thread_pool.hpp
|
||||
* @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
|
||||
* @version 3.2.0
|
||||
* @date 2022-07-28
|
||||
* @copyright Copyright (c) 2022 Barak Shoshany. Licensed under the MIT license. If you found this project useful, please consider starring it on GitHub! If you use this library in software of any kind, please provide a link to the GitHub repository https://github.com/bshoshany/thread-pool in the source code and documentation. If you use this library in published research, please cite it as follows: Barak Shoshany, "A C++17 Thread Pool for High-Performance Scientific Computing", doi:10.5281/zenodo.4742687, arXiv:2105.00613 (May 2021)
|
||||
* @version 3.5.0
|
||||
* @date 2023-05-25
|
||||
* @copyright Copyright (c) 2023 Barak Shoshany. Licensed under the MIT license. If you found this project useful, please consider starring it on GitHub! If you use this library in software of any kind, please provide a link to the GitHub repository https://github.com/bshoshany/thread-pool in the source code and documentation. If you use this library in published research, please cite it as follows: Barak Shoshany, "A C++17 Thread Pool for High-Performance Scientific Computing", doi:10.5281/zenodo.4742687, arXiv:2105.00613 (May 2021)
|
||||
*
|
||||
* @brief BS::thread_pool: a fast, lightweight, and easy-to-use C++17 thread pool library. This header file contains the entire library, including the main BS::thread_pool class and the helper classes BS::multi_future, BS::blocks, BS:synced_stream, and BS::timer.
|
||||
*/
|
||||
|
||||
#define BS_THREAD_POOL_VERSION "v3.2.0 (2022-07-28)"
|
||||
#define BS_THREAD_POOL_VERSION "v3.5.0 (2023-05-25)"
|
||||
|
||||
#include <atomic> // std::atomic
|
||||
#include <chrono> // std::chrono
|
||||
#include <condition_variable> // std::condition_variable
|
||||
#include <exception> // std::current_exception
|
||||
@@ -34,12 +33,6 @@ namespace BS
|
||||
*/
|
||||
using concurrency_t = std::invoke_result_t<decltype(std::thread::hardware_concurrency)>;
|
||||
|
||||
/**
|
||||
* @brief Explicit casts of the flushing stream manipulators, to enable using them with synced_stream, e.g. sync_out.print(BS::flush).
|
||||
*/
|
||||
std::ostream& (&endl)(std::ostream&) = static_cast<std::ostream& (&)(std::ostream&)>(std::endl);
|
||||
std::ostream& (&flush)(std::ostream&) = static_cast<std::ostream& (&)(std::ostream&)>(std::flush);
|
||||
|
||||
// ============================================================================================= //
|
||||
// Begin class multi_future //
|
||||
|
||||
@@ -57,7 +50,7 @@ public:
|
||||
*
|
||||
* @param num_futures_ The desired number of futures to store.
|
||||
*/
|
||||
multi_future(const size_t num_futures_ = 0) : f(num_futures_) {}
|
||||
multi_future(const size_t num_futures_ = 0) : futures(num_futures_) {}
|
||||
|
||||
/**
|
||||
* @brief Get the results from all the futures stored in this multi_future object, rethrowing any stored exceptions.
|
||||
@@ -68,32 +61,64 @@ public:
|
||||
{
|
||||
if constexpr (std::is_void_v<T>)
|
||||
{
|
||||
for (size_t i = 0; i < f.size(); ++i)
|
||||
f[i].get();
|
||||
for (size_t i = 0; i < futures.size(); ++i)
|
||||
futures[i].get();
|
||||
return;
|
||||
}
|
||||
else
|
||||
{
|
||||
std::vector<T> results(f.size());
|
||||
for (size_t i = 0; i < f.size(); ++i)
|
||||
results[i] = f[i].get();
|
||||
std::vector<T> results(futures.size());
|
||||
for (size_t i = 0; i < futures.size(); ++i)
|
||||
results[i] = futures[i].get();
|
||||
return results;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Get a reference to one of the futures stored in this multi_future object.
|
||||
*
|
||||
* @param i The index of the desired future.
|
||||
* @return The future.
|
||||
*/
|
||||
[[nodiscard]] std::future<T>& operator[](const size_t i)
|
||||
{
|
||||
return futures[i];
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Append a future to this multi_future object.
|
||||
*
|
||||
* @param future The future to append.
|
||||
*/
|
||||
void push_back(std::future<T> future)
|
||||
{
|
||||
futures.push_back(std::move(future));
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Get the number of futures stored in this multi_future object.
|
||||
*
|
||||
* @return The number of futures.
|
||||
*/
|
||||
[[nodiscard]] size_t size() const
|
||||
{
|
||||
return futures.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Wait for all the futures stored in this multi_future object.
|
||||
*/
|
||||
void wait() const
|
||||
{
|
||||
for (size_t i = 0; i < f.size(); ++i)
|
||||
f[i].wait();
|
||||
for (size_t i = 0; i < futures.size(); ++i)
|
||||
futures[i].wait();
|
||||
}
|
||||
|
||||
private:
|
||||
/**
|
||||
* @brief A vector to store the futures.
|
||||
*/
|
||||
std::vector<std::future<T>> f;
|
||||
std::vector<std::future<T>> futures;
|
||||
};
|
||||
|
||||
// End class multi_future //
|
||||
@@ -229,7 +254,7 @@ public:
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Destruct the thread pool. Waits for all tasks to complete, then destroys all threads. Note that if the variable paused is set to true, then any tasks still in the queue will never be executed.
|
||||
* @brief Destruct the thread pool. Waits for all tasks to complete, then destroys all threads. Note that if the pool is paused, then any tasks still in the queue will never be executed.
|
||||
*/
|
||||
~thread_pool()
|
||||
{
|
||||
@@ -260,17 +285,18 @@ public:
|
||||
[[nodiscard]] size_t get_tasks_running() const
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
return tasks_total - tasks.size();
|
||||
return tasks_running;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Get the total number of unfinished tasks: either still in the queue, or running in a thread. Note that get_tasks_total() == get_tasks_queued() + get_tasks_running().
|
||||
* @brief Get the total number of unfinished tasks: either still waiting in the queue, or running in a thread. Note that get_tasks_total() == get_tasks_queued() + get_tasks_running().
|
||||
*
|
||||
* @return The total number of tasks.
|
||||
*/
|
||||
[[nodiscard]] size_t get_tasks_total() const
|
||||
{
|
||||
return tasks_total;
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
return tasks_running + tasks.size();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -283,6 +309,17 @@ public:
|
||||
return thread_count;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Check whether the pool is currently paused.
|
||||
*
|
||||
* @return true if the pool is paused, false if it is not paused.
|
||||
*/
|
||||
[[nodiscard]] bool is_paused() const
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
return paused;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. Returns a multi_future object that contains the futures for all of the blocks.
|
||||
*
|
||||
@@ -305,7 +342,7 @@ public:
|
||||
{
|
||||
multi_future<R> mf(blks.get_num_blocks());
|
||||
for (size_t i = 0; i < blks.get_num_blocks(); ++i)
|
||||
mf.f[i] = submit(std::forward<F>(loop), blks.start(i), blks.end(i));
|
||||
mf[i] = submit(std::forward<F>(loop), blks.start(i), blks.end(i));
|
||||
return mf;
|
||||
}
|
||||
else
|
||||
@@ -331,6 +368,25 @@ public:
|
||||
return parallelize_loop(0, index_after_last, std::forward<F>(loop), num_blocks);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Pause the pool. The workers will temporarily stop retrieving new tasks out of the queue, although any tasks already executed will keep running until they are finished.
|
||||
*/
|
||||
void pause()
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
paused = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Purge all the tasks waiting in the queue. Tasks that are currently running will not be affected, but any tasks still waiting in the queue will be discarded, and will never be executed by the threads. Please note that there is no way to restore the purged tasks.
|
||||
*/
|
||||
void purge()
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
while (!tasks.empty())
|
||||
tasks.pop();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. Does not return a multi_future, so the user must use wait_for_tasks() or some other method to ensure that the loop finishes executing, otherwise bad things will happen.
|
||||
*
|
||||
@@ -380,12 +436,10 @@ public:
|
||||
template <typename F, typename... A>
|
||||
void push_task(F&& task, A&&... args)
|
||||
{
|
||||
std::function<void()> task_function = std::bind(std::forward<F>(task), std::forward<A>(args)...);
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
tasks.push(task_function);
|
||||
tasks.push(std::bind(std::forward<F>(task), std::forward<A>(args)...)); // cppcheck-suppress ignoredReturnValue
|
||||
}
|
||||
++tasks_total;
|
||||
task_available_cv.notify_one();
|
||||
}
|
||||
|
||||
@@ -396,8 +450,10 @@ public:
|
||||
*/
|
||||
void reset(const concurrency_t thread_count_ = 0)
|
||||
{
|
||||
std::unique_lock tasks_lock(tasks_mutex);
|
||||
const bool was_paused = paused;
|
||||
paused = true;
|
||||
tasks_lock.unlock();
|
||||
wait_for_tasks();
|
||||
destroy_threads();
|
||||
thread_count = determine_thread_count(thread_count_);
|
||||
@@ -419,10 +475,9 @@ public:
|
||||
template <typename F, typename... A, typename R = std::invoke_result_t<std::decay_t<F>, std::decay_t<A>...>>
|
||||
[[nodiscard]] std::future<R> submit(F&& task, A&&... args)
|
||||
{
|
||||
std::function<R()> task_function = std::bind(std::forward<F>(task), std::forward<A>(args)...);
|
||||
std::shared_ptr<std::promise<R>> task_promise = std::make_shared<std::promise<R>>();
|
||||
push_task(
|
||||
[task_function, task_promise]
|
||||
[task_function = std::bind(std::forward<F>(task), std::forward<A>(args)...), task_promise]
|
||||
{
|
||||
try
|
||||
{
|
||||
@@ -450,25 +505,61 @@ public:
|
||||
return task_promise->get_future();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Unpause the pool. The workers will resume retrieving new tasks out of the queue.
|
||||
*/
|
||||
void unpause()
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
paused = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Wait for tasks to be completed. Normally, this function waits for all tasks, both those that are currently running in the threads and those that are still waiting in the queue. However, if the pool is paused, this function only waits for the currently running tasks (otherwise it would wait forever). Note: To wait for just one specific task, use submit() instead, and call the wait() member function of the generated future.
|
||||
*/
|
||||
void wait_for_tasks()
|
||||
{
|
||||
std::unique_lock tasks_lock(tasks_mutex);
|
||||
waiting = true;
|
||||
std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
|
||||
task_done_cv.wait(tasks_lock, [this] { return (tasks_total == (paused ? tasks.size() : 0)); });
|
||||
tasks_done_cv.wait(tasks_lock, [this] { return !tasks_running && (paused || tasks.empty()); });
|
||||
waiting = false;
|
||||
}
|
||||
|
||||
// ===========
|
||||
// Public data
|
||||
// ===========
|
||||
/**
|
||||
* @brief Wait for tasks to be completed, but stop waiting after the specified duration has passed.
|
||||
*
|
||||
* @tparam R An arithmetic type representing the number of ticks to wait.
|
||||
* @tparam P An std::ratio representing the length of each tick in seconds.
|
||||
* @param duration The time duration to wait.
|
||||
* @return true if all tasks finished running, false if the duration expired but some tasks are still running.
|
||||
*/
|
||||
template <typename R, typename P>
|
||||
bool wait_for_tasks_duration(const std::chrono::duration<R, P>& duration)
|
||||
{
|
||||
std::unique_lock tasks_lock(tasks_mutex);
|
||||
waiting = true;
|
||||
const bool status = tasks_done_cv.wait_for(tasks_lock, duration, [this] { return !tasks_running && (paused || tasks.empty()); });
|
||||
waiting = false;
|
||||
return status;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief An atomic variable indicating whether the workers should pause. When set to true, the workers temporarily stop retrieving new tasks out of the queue, although any tasks already executed will keep running until they are finished. Set to false again to resume retrieving tasks.
|
||||
* @brief Wait for tasks to be completed, but stop waiting after the specified time point has been reached.
|
||||
*
|
||||
* @tparam C The type of the clock used to measure time.
|
||||
* @tparam D An std::chrono::duration type used to indicate the time point.
|
||||
* @param timeout_time The time point at which to stop waiting.
|
||||
* @return true if all tasks finished running, false if the time point was reached but some tasks are still running.
|
||||
*/
|
||||
std::atomic<bool> paused = false;
|
||||
template <typename C, typename D>
|
||||
bool wait_for_tasks_until(const std::chrono::time_point<C, D>& timeout_time)
|
||||
{
|
||||
std::unique_lock tasks_lock(tasks_mutex);
|
||||
waiting = true;
|
||||
const bool status = tasks_done_cv.wait_until(tasks_lock, timeout_time, [this] { return !tasks_running && (paused || tasks.empty()); });
|
||||
waiting = false;
|
||||
return status;
|
||||
}
|
||||
|
||||
private:
|
||||
// ========================
|
||||
@@ -480,7 +571,10 @@ private:
|
||||
*/
|
||||
void create_threads()
|
||||
{
|
||||
running = true;
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
workers_running = true;
|
||||
}
|
||||
for (concurrency_t i = 0; i < thread_count; ++i)
|
||||
{
|
||||
threads[i] = std::thread(&thread_pool::worker, this);
|
||||
@@ -492,7 +586,10 @@ private:
|
||||
*/
|
||||
void destroy_threads()
|
||||
{
|
||||
running = false;
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
workers_running = false;
|
||||
}
|
||||
task_available_cv.notify_all();
|
||||
for (concurrency_t i = 0; i < thread_count; ++i)
|
||||
{
|
||||
@@ -506,7 +603,7 @@ private:
|
||||
* @param thread_count_ The parameter passed to the constructor or reset(). If the parameter is a positive number, then the pool will be created with this number of threads. If the parameter is non-positive, or a parameter was not supplied (in which case it will have the default value of 0), then the pool will be created with the total number of hardware threads available, as obtained from std::thread::hardware_concurrency(). If the latter returns a non-positive number for some reason, then the pool will be created with just one thread.
|
||||
* @return The number of threads to use for constructing the pool.
|
||||
*/
|
||||
[[nodiscard]] concurrency_t determine_thread_count(const concurrency_t thread_count_)
|
||||
[[nodiscard]] concurrency_t determine_thread_count(const concurrency_t thread_count_) const
|
||||
{
|
||||
if (thread_count_ > 0)
|
||||
return thread_count_;
|
||||
@@ -524,22 +621,24 @@ private:
|
||||
*/
|
||||
void worker()
|
||||
{
|
||||
while (running)
|
||||
std::function<void()> task;
|
||||
while (true)
|
||||
{
|
||||
std::function<void()> task;
|
||||
std::unique_lock<std::mutex> tasks_lock(tasks_mutex);
|
||||
task_available_cv.wait(tasks_lock, [this] { return !tasks.empty() || !running; });
|
||||
if (running && !paused)
|
||||
{
|
||||
task = std::move(tasks.front());
|
||||
tasks.pop();
|
||||
tasks_lock.unlock();
|
||||
task();
|
||||
tasks_lock.lock();
|
||||
--tasks_total;
|
||||
if (waiting)
|
||||
task_done_cv.notify_one();
|
||||
}
|
||||
std::unique_lock tasks_lock(tasks_mutex);
|
||||
task_available_cv.wait(tasks_lock, [this] { return !tasks.empty() || !workers_running; });
|
||||
if (!workers_running)
|
||||
break;
|
||||
if (paused)
|
||||
continue;
|
||||
task = std::move(tasks.front());
|
||||
tasks.pop();
|
||||
++tasks_running;
|
||||
tasks_lock.unlock();
|
||||
task();
|
||||
tasks_lock.lock();
|
||||
--tasks_running;
|
||||
if (waiting && !tasks_running && (paused || tasks.empty()))
|
||||
tasks_done_cv.notify_all();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -548,19 +647,19 @@ private:
|
||||
// ============
|
||||
|
||||
/**
|
||||
* @brief An atomic variable indicating to the workers to keep running. When set to false, the workers permanently stop working.
|
||||
* @brief A flag indicating whether the workers should pause. When set to true, the workers temporarily stop retrieving new tasks out of the queue, although any tasks already executed will keep running until they are finished. When set to false again, the workers resume retrieving tasks.
|
||||
*/
|
||||
std::atomic<bool> running = false;
|
||||
bool paused = false;
|
||||
|
||||
/**
|
||||
* @brief A condition variable used to notify worker() that a new task has become available.
|
||||
* @brief A condition variable to notify worker() that a new task has become available.
|
||||
*/
|
||||
std::condition_variable task_available_cv = {};
|
||||
|
||||
/**
|
||||
* @brief A condition variable used to notify wait_for_tasks() that a tasks is done.
|
||||
* @brief A condition variable to notify wait_for_tasks() that the tasks are done.
|
||||
*/
|
||||
std::condition_variable task_done_cv = {};
|
||||
std::condition_variable tasks_done_cv = {};
|
||||
|
||||
/**
|
||||
* @brief A queue of tasks to be executed by the threads.
|
||||
@@ -568,9 +667,9 @@ private:
|
||||
std::queue<std::function<void()>> tasks = {};
|
||||
|
||||
/**
|
||||
* @brief An atomic variable to keep track of the total number of unfinished tasks - either still in the queue, or running in a thread.
|
||||
* @brief A counter for the total number of currently running tasks.
|
||||
*/
|
||||
std::atomic<size_t> tasks_total = 0;
|
||||
size_t tasks_running = 0;
|
||||
|
||||
/**
|
||||
* @brief A mutex to synchronize access to the task queue by different threads.
|
||||
@@ -588,9 +687,14 @@ private:
|
||||
std::unique_ptr<std::thread[]> threads = nullptr;
|
||||
|
||||
/**
|
||||
* @brief An atomic variable indicating that wait_for_tasks() is active and expects to be notified whenever a task is done.
|
||||
* @brief A flag indicating that wait_for_tasks() is active and expects to be notified whenever a task is done.
|
||||
*/
|
||||
std::atomic<bool> waiting = false;
|
||||
bool waiting = false;
|
||||
|
||||
/**
|
||||
* @brief A flag indicating to the workers to keep running. When set to false, the workers terminate permanently.
|
||||
*/
|
||||
bool workers_running = false;
|
||||
};
|
||||
|
||||
// End class thread_pool //
|
||||
@@ -637,6 +741,16 @@ public:
|
||||
print(std::forward<T>(items)..., '\n');
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief A stream manipulator to pass to a synced_stream (an explicit cast of std::endl). Prints a newline character to the stream, and then flushes it. Should only be used if flushing is desired, otherwise '\n' should be used instead.
|
||||
*/
|
||||
inline static std::ostream& (&endl)(std::ostream&) = static_cast<std::ostream& (&)(std::ostream&)>(std::endl);
|
||||
|
||||
/**
|
||||
* @brief A stream manipulator to pass to a synced_stream (an explicit cast of std::flush). Used to flush the stream.
|
||||
*/
|
||||
inline static std::ostream& (&flush)(std::ostream&) = static_cast<std::ostream& (&)(std::ostream&)>(std::flush);
|
||||
|
||||
private:
|
||||
/**
|
||||
* @brief The output stream to print to.
|
||||
@@ -0,0 +1,327 @@
|
||||
#pragma once
|
||||
|
||||
/**
|
||||
* @file BS_thread_pool_light.hpp
|
||||
* @author Barak Shoshany (baraksh@gmail.com) (http://baraksh.com)
|
||||
* @version 3.5.0
|
||||
* @date 2023-05-25
|
||||
* @copyright Copyright (c) 2023 Barak Shoshany. Licensed under the MIT license. If you found this project useful, please consider starring it on GitHub! If you use this library in software of any kind, please provide a link to the GitHub repository https://github.com/bshoshany/thread-pool in the source code and documentation. If you use this library in published research, please cite it as follows: Barak Shoshany, "A C++17 Thread Pool for High-Performance Scientific Computing", doi:10.5281/zenodo.4742687, arXiv:2105.00613 (May 2021)
|
||||
*
|
||||
* @brief BS::thread_pool_light: a fast, lightweight, and easy-to-use C++17 thread pool library. This header file contains a light version of the main library, for use when advanced features are not needed.
|
||||
*/
|
||||
|
||||
#define BS_THREAD_POOL_LIGHT_VERSION "v3.5.0 (2023-05-25)"
|
||||
|
||||
#include <condition_variable> // std::condition_variable
|
||||
#include <exception> // std::current_exception
|
||||
#include <functional> // std::bind, std::function, std::invoke
|
||||
#include <future> // std::future, std::promise
|
||||
#include <memory> // std::make_shared, std::make_unique, std::shared_ptr, std::unique_ptr
|
||||
#include <mutex> // std::mutex, std::scoped_lock, std::unique_lock
|
||||
#include <queue> // std::queue
|
||||
#include <thread> // std::thread
|
||||
#include <type_traits> // std::common_type_t, std::decay_t, std::invoke_result_t, std::is_void_v
|
||||
#include <utility> // std::forward, std::move, std::swap
|
||||
|
||||
namespace BS
|
||||
{
|
||||
/**
|
||||
* @brief A convenient shorthand for the type of std::thread::hardware_concurrency(). Should evaluate to unsigned int.
|
||||
*/
|
||||
using concurrency_t = std::invoke_result_t<decltype(std::thread::hardware_concurrency)>;
|
||||
|
||||
/**
|
||||
* @brief A fast, lightweight, and easy-to-use C++17 thread pool class. This is a lighter version of the main thread pool class.
|
||||
*/
|
||||
class [[nodiscard]] thread_pool_light
|
||||
{
|
||||
public:
|
||||
// ============================
|
||||
// Constructors and destructors
|
||||
// ============================
|
||||
|
||||
/**
|
||||
* @brief Construct a new thread pool.
|
||||
*
|
||||
* @param thread_count_ The number of threads to use. The default value is the total number of hardware threads available, as reported by the implementation. This is usually determined by the number of cores in the CPU. If a core is hyperthreaded, it will count as two threads.
|
||||
*/
|
||||
thread_pool_light(const concurrency_t thread_count_ = 0) : thread_count(determine_thread_count(thread_count_)), threads(std::make_unique<std::thread[]>(determine_thread_count(thread_count_)))
|
||||
{
|
||||
create_threads();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Destruct the thread pool. Waits for all tasks to complete, then destroys all threads.
|
||||
*/
|
||||
~thread_pool_light()
|
||||
{
|
||||
wait_for_tasks();
|
||||
destroy_threads();
|
||||
}
|
||||
|
||||
// =======================
|
||||
// Public member functions
|
||||
// =======================
|
||||
|
||||
/**
|
||||
* @brief Get the number of threads in the pool.
|
||||
*
|
||||
* @return The number of threads.
|
||||
*/
|
||||
[[nodiscard]] concurrency_t get_thread_count() const
|
||||
{
|
||||
return thread_count;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. The user must use wait_for_tasks() or some other method to ensure that the loop finishes executing, otherwise bad things will happen.
|
||||
*
|
||||
* @tparam F The type of the function to loop through.
|
||||
* @tparam T1 The type of the first index in the loop. Should be a signed or unsigned integer.
|
||||
* @tparam T2 The type of the index after the last index in the loop. Should be a signed or unsigned integer. If T1 is not the same as T2, a common type will be automatically inferred.
|
||||
* @tparam T The common type of T1 and T2.
|
||||
* @param first_index The first index in the loop.
|
||||
* @param index_after_last The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = first_index; i < index_after_last; ++i)". Note that if index_after_last == first_index, no blocks will be submitted.
|
||||
* @param loop The function to loop through. Will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. loop(start, end) should typically involve a loop of the form "for (T i = start; i < end; ++i)".
|
||||
* @param num_blocks The maximum number of blocks to split the loop into. The default is to use the number of threads in the pool.
|
||||
*/
|
||||
template <typename F, typename T1, typename T2, typename T = std::common_type_t<T1, T2>>
|
||||
void push_loop(T1 first_index_, T2 index_after_last_, F&& loop, size_t num_blocks = 0)
|
||||
{
|
||||
T first_index = static_cast<T>(first_index_);
|
||||
T index_after_last = static_cast<T>(index_after_last_);
|
||||
if (num_blocks == 0)
|
||||
num_blocks = thread_count;
|
||||
if (index_after_last < first_index)
|
||||
std::swap(index_after_last, first_index);
|
||||
size_t total_size = static_cast<size_t>(index_after_last - first_index);
|
||||
size_t block_size = static_cast<size_t>(total_size / num_blocks);
|
||||
if (block_size == 0)
|
||||
{
|
||||
block_size = 1;
|
||||
num_blocks = (total_size > 1) ? total_size : 1;
|
||||
}
|
||||
if (total_size > 0)
|
||||
{
|
||||
for (size_t i = 0; i < num_blocks; ++i)
|
||||
push_task(std::forward<F>(loop), static_cast<T>(i * block_size) + first_index, (i == num_blocks - 1) ? index_after_last : (static_cast<T>((i + 1) * block_size) + first_index));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue. The user must use wait_for_tasks() or some other method to ensure that the loop finishes executing, otherwise bad things will happen. This overload is used for the special case where the first index is 0.
|
||||
*
|
||||
* @tparam F The type of the function to loop through.
|
||||
* @tparam T The type of the loop indices. Should be a signed or unsigned integer.
|
||||
* @param index_after_last The index after the last index in the loop. The loop will iterate from 0 to (index_after_last - 1) inclusive. In other words, it will be equivalent to "for (T i = 0; i < index_after_last; ++i)". Note that if index_after_last == 0, no blocks will be submitted.
|
||||
* @param loop The function to loop through. Will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. loop(start, end) should typically involve a loop of the form "for (T i = start; i < end; ++i)".
|
||||
* @param num_blocks The maximum number of blocks to split the loop into. The default is to use the number of threads in the pool.
|
||||
*/
|
||||
template <typename F, typename T>
|
||||
void push_loop(const T index_after_last, F&& loop, const size_t num_blocks = 0)
|
||||
{
|
||||
push_loop(0, index_after_last, std::forward<F>(loop), num_blocks);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Push a function with zero or more arguments, but no return value, into the task queue. Does not return a future, so the user must use wait_for_tasks() or some other method to ensure that the task finishes executing, otherwise bad things will happen.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @tparam A The types of the arguments.
|
||||
* @param task The function to push.
|
||||
* @param args The zero or more arguments to pass to the function. Note that if the task is a class member function, the first argument must be a pointer to the object, i.e. &object (or this), followed by the actual arguments.
|
||||
*/
|
||||
template <typename F, typename... A>
|
||||
void push_task(F&& task, A&&... args)
|
||||
{
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
tasks.push(std::bind(std::forward<F>(task), std::forward<A>(args)...)); // cppcheck-suppress ignoredReturnValue
|
||||
}
|
||||
task_available_cv.notify_one();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Submit a function with zero or more arguments into the task queue. If the function has a return value, get a future for the eventual returned value. If the function has no return value, get an std::future<void> which can be used to wait until the task finishes.
|
||||
*
|
||||
* @tparam F The type of the function.
|
||||
* @tparam A The types of the zero or more arguments to pass to the function.
|
||||
* @tparam R The return type of the function (can be void).
|
||||
* @param task The function to submit.
|
||||
* @param args The zero or more arguments to pass to the function. Note that if the task is a class member function, the first argument must be a pointer to the object, i.e. &object (or this), followed by the actual arguments.
|
||||
* @return A future to be used later to wait for the function to finish executing and/or obtain its returned value if it has one.
|
||||
*/
|
||||
template <typename F, typename... A, typename R = std::invoke_result_t<std::decay_t<F>, std::decay_t<A>...>>
|
||||
[[nodiscard]] std::future<R> submit(F&& task, A&&... args)
|
||||
{
|
||||
std::shared_ptr<std::promise<R>> task_promise = std::make_shared<std::promise<R>>();
|
||||
push_task(
|
||||
[task_function = std::bind(std::forward<F>(task), std::forward<A>(args)...), task_promise]
|
||||
{
|
||||
try
|
||||
{
|
||||
if constexpr (std::is_void_v<R>)
|
||||
{
|
||||
std::invoke(task_function);
|
||||
task_promise->set_value();
|
||||
}
|
||||
else
|
||||
{
|
||||
task_promise->set_value(std::invoke(task_function));
|
||||
}
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
try
|
||||
{
|
||||
task_promise->set_exception(std::current_exception());
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
}
|
||||
}
|
||||
});
|
||||
return task_promise->get_future();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Wait for tasks to be completed, both those that are currently running in the threads and those that are still waiting in the queue. Note: To wait for just one specific task, use submit() instead, and call the wait() member function of the generated future.
|
||||
*/
|
||||
void wait_for_tasks()
|
||||
{
|
||||
std::unique_lock tasks_lock(tasks_mutex);
|
||||
waiting = true;
|
||||
tasks_done_cv.wait(tasks_lock, [this] { return !tasks_running && tasks.empty(); });
|
||||
waiting = false;
|
||||
}
|
||||
|
||||
private:
|
||||
// ========================
|
||||
// Private member functions
|
||||
// ========================
|
||||
|
||||
/**
|
||||
* @brief Create the threads in the pool and assign a worker to each thread.
|
||||
*/
|
||||
void create_threads()
|
||||
{
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
workers_running = true;
|
||||
}
|
||||
for (concurrency_t i = 0; i < thread_count; ++i)
|
||||
{
|
||||
threads[i] = std::thread(&thread_pool_light::worker, this);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Destroy the threads in the pool.
|
||||
*/
|
||||
void destroy_threads()
|
||||
{
|
||||
{
|
||||
const std::scoped_lock tasks_lock(tasks_mutex);
|
||||
workers_running = false;
|
||||
}
|
||||
task_available_cv.notify_all();
|
||||
for (concurrency_t i = 0; i < thread_count; ++i)
|
||||
{
|
||||
threads[i].join();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Determine how many threads the pool should have, based on the parameter passed to the constructor.
|
||||
*
|
||||
* @param thread_count_ The parameter passed to the constructor. If the parameter is a positive number, then the pool will be created with this number of threads. If the parameter is non-positive, or a parameter was not supplied (in which case it will have the default value of 0), then the pool will be created with the total number of hardware threads available, as obtained from std::thread::hardware_concurrency(). If the latter returns a non-positive number for some reason, then the pool will be created with just one thread.
|
||||
* @return The number of threads to use for constructing the pool.
|
||||
*/
|
||||
[[nodiscard]] concurrency_t determine_thread_count(const concurrency_t thread_count_) const
|
||||
{
|
||||
if (thread_count_ > 0)
|
||||
return thread_count_;
|
||||
else
|
||||
{
|
||||
if (std::thread::hardware_concurrency() > 0)
|
||||
return std::thread::hardware_concurrency();
|
||||
else
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief A worker function to be assigned to each thread in the pool. Waits until it is notified by push_task() that a task is available, and then retrieves the task from the queue and executes it. Once the task finishes, the worker notifies wait_for_tasks() in case it is waiting.
|
||||
*/
|
||||
void worker()
|
||||
{
|
||||
std::function<void()> task;
|
||||
while (true)
|
||||
{
|
||||
std::unique_lock tasks_lock(tasks_mutex);
|
||||
task_available_cv.wait(tasks_lock, [this] { return !tasks.empty() || !workers_running; });
|
||||
if (!workers_running)
|
||||
break;
|
||||
task = std::move(tasks.front());
|
||||
tasks.pop();
|
||||
++tasks_running;
|
||||
tasks_lock.unlock();
|
||||
task();
|
||||
tasks_lock.lock();
|
||||
--tasks_running;
|
||||
if (waiting && !tasks_running && tasks.empty())
|
||||
tasks_done_cv.notify_all();
|
||||
}
|
||||
}
|
||||
|
||||
// ============
|
||||
// Private data
|
||||
// ============
|
||||
|
||||
/**
|
||||
* @brief A condition variable to notify worker() that a new task has become available.
|
||||
*/
|
||||
std::condition_variable task_available_cv = {};
|
||||
|
||||
/**
|
||||
* @brief A condition variable to notify wait_for_tasks() that the tasks are done.
|
||||
*/
|
||||
std::condition_variable tasks_done_cv = {};
|
||||
|
||||
/**
|
||||
* @brief A queue of tasks to be executed by the threads.
|
||||
*/
|
||||
std::queue<std::function<void()>> tasks = {};
|
||||
|
||||
/**
|
||||
* @brief A counter for the total number of currently running tasks.
|
||||
*/
|
||||
size_t tasks_running = 0;
|
||||
|
||||
/**
|
||||
* @brief A mutex to synchronize access to the task queue by different threads.
|
||||
*/
|
||||
mutable std::mutex tasks_mutex = {};
|
||||
|
||||
/**
|
||||
* @brief The number of threads in the pool.
|
||||
*/
|
||||
concurrency_t thread_count = 0;
|
||||
|
||||
/**
|
||||
* @brief A smart pointer to manage the memory allocated for the threads.
|
||||
*/
|
||||
std::unique_ptr<std::thread[]> threads = nullptr;
|
||||
|
||||
/**
|
||||
* @brief A flag indicating that wait_for_tasks() is active and expects to be notified whenever a task is done.
|
||||
*/
|
||||
bool waiting = false;
|
||||
|
||||
/**
|
||||
* @brief A flag indicating to the workers to keep running. When set to false, the workers terminate permanently.
|
||||
*/
|
||||
bool workers_running = false;
|
||||
};
|
||||
|
||||
} // namespace BS
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,224 @@
|
||||
#!/usr/bin/env pwsh
|
||||
$Host.UI.RawUI.WindowTitle = 'BS::thread_pool Test Script'
|
||||
|
||||
$TitleColor = 'Green'
|
||||
$TextColor = 'Yellow'
|
||||
$ErrorColor = 'Red'
|
||||
$CommandColor = 'Blue'
|
||||
|
||||
Function Write-Title([String] $Title)
|
||||
{
|
||||
If ($Title.Length -gt 0)
|
||||
{
|
||||
$Separator = '=' * $Title.Length
|
||||
Write-Host
|
||||
Write-Host $Separator -ForegroundColor $TitleColor
|
||||
Write-Host $Title -ForegroundColor $TitleColor
|
||||
Write-Host $Separator -ForegroundColor $TitleColor
|
||||
Write-Host
|
||||
}
|
||||
}
|
||||
|
||||
Function Write-Text([String] $Text)
|
||||
{
|
||||
Write-Host $Text -ForegroundColor $TextColor
|
||||
}
|
||||
|
||||
Function Write-Error([String] $Text)
|
||||
{
|
||||
Write-Host $Text -ForegroundColor $ErrorColor
|
||||
}
|
||||
|
||||
Function Write-Command([String] $Text)
|
||||
{
|
||||
Write-Host $Text -ForegroundColor $CommandColor
|
||||
}
|
||||
|
||||
Function Exit-Script([Int] $Code)
|
||||
{
|
||||
Set-Location $CurrentDir
|
||||
Exit $Code
|
||||
}
|
||||
|
||||
$CurrentDir = Get-Location
|
||||
|
||||
Write-Title '=== BS::thread_pool test script ==='
|
||||
|
||||
$SourceFile = 'BS_thread_pool_test.cpp'
|
||||
If (Test-Path -Path $SourceFile)
|
||||
{
|
||||
Write-Text "Found source file $SourceFile."
|
||||
}
|
||||
Else
|
||||
{
|
||||
Write-Error "Source file $SourceFile not found, aborting script!"
|
||||
Exit-Script 1
|
||||
}
|
||||
|
||||
$BuildFolder = Join-Path '..' 'build'
|
||||
If (Test-Path -Path $BuildFolder)
|
||||
{
|
||||
Write-Text 'Cleaning up build folder...'
|
||||
Remove-Item (Join-Path $BuildFolder '*')
|
||||
}
|
||||
Else
|
||||
{
|
||||
Write-Text 'Creating build folder...'
|
||||
$Null = New-Item $BuildFolder -ItemType 'Directory'
|
||||
}
|
||||
Write-Text 'Done.'
|
||||
|
||||
Write-Title 'Compiling...'
|
||||
|
||||
If ($IsWindows)
|
||||
{
|
||||
$Extension = '.exe'
|
||||
$PThread = ''
|
||||
}
|
||||
Else
|
||||
{
|
||||
$Extension = ''
|
||||
$PThread = ' -pthread'
|
||||
}
|
||||
|
||||
$ClangExe = 'BS_thread_pool_test_clang' + $Extension
|
||||
$GCCExe = 'BS_thread_pool_test_gcc' + $Extension
|
||||
|
||||
$MSVCName = 'BS_thread_pool_test_msvc'
|
||||
$MSVCExe = $MSVCName + $Extension
|
||||
$MSVCObj = $MSVCName + '.obj'
|
||||
|
||||
If (Get-Command 'clang++' -ErrorAction SilentlyContinue)
|
||||
{
|
||||
$Clang = $True
|
||||
$FullClangExe = Join-Path $BuildFolder $ClangExe
|
||||
Write-Text 'Compiling with Clang...'
|
||||
$ClangCommand = "clang++ $SourceFile$PThread -std=c++17 -O3 -Wall -Wextra -Wconversion -Wsign-conversion -Wpedantic -Weffc++ -Wshadow -o $FullClangExe"
|
||||
Write-Command $ClangCommand
|
||||
Invoke-Expression $ClangCommand
|
||||
If ($LASTEXITCODE)
|
||||
{
|
||||
Write-Error "Failed to compile, aborting script! (Exit code: $LASTEXITCODE)"
|
||||
Exit-Script $LASTEXITCODE
|
||||
}
|
||||
Else
|
||||
{
|
||||
Write-Text "Successfully compiled to $FullClangExe."
|
||||
}
|
||||
}
|
||||
Else
|
||||
{
|
||||
$Clang = $False
|
||||
Write-Error 'Clang not found, skipping it!'
|
||||
}
|
||||
|
||||
Write-Host
|
||||
|
||||
If (Get-Command 'g++' -ErrorAction SilentlyContinue)
|
||||
{
|
||||
$GCC = $True
|
||||
$FullGCCExe = Join-Path $BuildFolder $GCCExe
|
||||
Write-Text 'Compiling with GCC...'
|
||||
$GCCCommand = "g++ $SourceFile$PThread -std=c++17 -O3 -Wall -Wextra -Wconversion -Wsign-conversion -Wpedantic -Weffc++ -Wshadow -o $FullGCCExe"
|
||||
Write-Command $GCCCommand
|
||||
Invoke-Expression $GCCCommand
|
||||
If ($LASTEXITCODE)
|
||||
{
|
||||
Write-Error "Failed to compile, aborting script! (Exit code: $LASTEXITCODE)"
|
||||
Exit-Script $LASTEXITCODE
|
||||
}
|
||||
Else
|
||||
{
|
||||
Write-Text "Successfully compiled to $FullGCCExe."
|
||||
}
|
||||
}
|
||||
Else
|
||||
{
|
||||
$GCC = $False
|
||||
Write-Error 'GCC not found, skipping it!'
|
||||
}
|
||||
|
||||
Write-Host
|
||||
|
||||
If ($IsWindows)
|
||||
{
|
||||
$MSVCEnv = 'C:\Program Files\Microsoft Visual Studio\2022\Community\Common7\Tools\Launch-VsDevShell.ps1'
|
||||
If (Test-Path $MSVCEnv)
|
||||
{
|
||||
$MSVC = $True
|
||||
$FullMSVCExe = Join-Path $BuildFolder $MSVCExe
|
||||
$FullMSVCObj = Join-Path $BuildFolder $MSVCObj
|
||||
Write-Text 'Compiling with MSVC...'
|
||||
$Env:BS_THREAD_POOL_MSVC_COMMAND = "cl $SourceFile /std:c++17 /permissive- /O2 /W4 /EHsc /Fe:$FullMSVCExe /Fo:$FullMSVCObj"
|
||||
Write-Command $Env:BS_THREAD_POOL_MSVC_COMMAND
|
||||
pwsh -Command {
|
||||
$CurrentDir = Get-Location
|
||||
& 'C:\Program Files\Microsoft Visual Studio\2022\Community\Common7\Tools\Launch-VsDevShell.ps1'
|
||||
Set-Location $CurrentDir
|
||||
Invoke-Expression $Env:BS_THREAD_POOL_MSVC_COMMAND
|
||||
Exit $LASTEXITCODE
|
||||
}
|
||||
$Env:BS_THREAD_POOL_MSVC_COMMAND = $Null
|
||||
If ($LASTEXITCODE)
|
||||
{
|
||||
Write-Error "Failed to compile, aborting script! (Exit code: $LASTEXITCODE)"
|
||||
Exit-Script $LASTEXITCODE
|
||||
}
|
||||
Else
|
||||
{
|
||||
Remove-Item $FullMSVCObj
|
||||
Write-Text "Successfully compiled to $FullMSVCExe."
|
||||
}
|
||||
}
|
||||
Else
|
||||
{
|
||||
$MSVC = $False
|
||||
Write-Error 'MSVC not found, skipping it!'
|
||||
}
|
||||
}
|
||||
Else
|
||||
{
|
||||
$MSVC = $False
|
||||
}
|
||||
|
||||
Write-Title 'Starting tests...'
|
||||
|
||||
Set-Location $BuildFolder
|
||||
$Repeats = 5;
|
||||
|
||||
Write-Text "Each test will be repeated $Repeats times. Outputs can be found in the generated log files."
|
||||
|
||||
Function Invoke-Test([String] $Compiler, [String] $Executable)
|
||||
{
|
||||
Write-Title "Starting $Compiler tests..."
|
||||
For ($i = 1; $i -Le $Repeats; $i++)
|
||||
{
|
||||
Write-Text "Running $Compiler test #$i of $Repeats..."
|
||||
Invoke-Expression $Executable | Out-Null
|
||||
if ($LASTEXITCODE)
|
||||
{
|
||||
Write-Error "Test Failed, aborting script! (Exit code: $LASTEXITCODE)"
|
||||
Exit-Script $LASTEXITCODE
|
||||
}
|
||||
Write-Text 'Test finished successfully!'
|
||||
Write-Host
|
||||
}
|
||||
Write-Text "$Compiler tests done."
|
||||
}
|
||||
|
||||
if ($Clang)
|
||||
{
|
||||
Invoke-Test 'Clang' (Join-Path '.' $ClangExe)
|
||||
}
|
||||
if ($GCC)
|
||||
{
|
||||
Invoke-Test 'GCC' (Join-Path '.' $GCCExe)
|
||||
}
|
||||
if ($MSVC)
|
||||
{
|
||||
Invoke-Test 'MSVC' (Join-Path '.' $MSVCExe)
|
||||
}
|
||||
|
||||
Write-Title '=== All tests completed successfully! ==='
|
||||
|
||||
Exit-Script 0
|
||||
Reference in New Issue
Block a user