Index/include / boost / capy / ex / thread_pool.hpp

include/boost/capy/ex/thread_pool.hpp

100.0% Lines (7 / 7) 100.0% Functions (3 / 3)
[ ] prev/next file
thread_pool.hpp
f(x) Functions (3)
Function Calls Lines Blocks
boost::capy::thread_pool::executor_type::executor_type(boost::capy::thread_pool&) :177 15321x 100.0% 100.0% boost::capy::thread_pool::executor_type::context() const :197 12611x 100.0% 100.0% boost::capy::thread_pool::executor_type::operator==(boost::capy::thread_pool::executor_type const&) const :265 15x 100.0% 100.0%
Line TLA Hits Source Code
1 //
2 // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3 // Copyright (c) 2026 Michael Vandeberg
4 //
5 // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 //
8 // Official repository: https://github.com/cppalliance/capy
9 //
10
11 #ifndef BOOST_CAPY_EX_THREAD_POOL_HPP
12 #define BOOST_CAPY_EX_THREAD_POOL_HPP
13
14 #include <boost/capy/detail/config.hpp>
15 #include <boost/capy/continuation.hpp>
16 #include <coroutine>
17 #include <boost/capy/ex/execution_context.hpp>
18 #include <cstddef>
19 #include <string_view>
20
21 namespace boost {
22 namespace capy {
23
24 /** Distributes posted work across a fixed group of worker threads via a shared queue.
25
26 Use this when you need to run coroutines on multiple threads
27 without the overhead of creating and destroying threads for
28 each task. Work items are distributed across the pool using
29 a shared queue.
30
31 @par Thread Safety
32 Distinct objects: Safe.
33 Shared objects: Safe for @ref get_executor, @ref join, and
34 @ref stop. Unsafe for construction and destruction.
35
36 @par Example
37 @par !example example
38
39
40 @note `join()` waits only for work that holds outstanding-work
41 counting, which `run_async` (and `make_work_guard`) provide. A bare
42 `executor_type::post()` does not register outstanding work, so
43 `join()` does not wait for it.
44 */
45 class BOOST_CAPY_DECL
46 thread_pool
47 : public execution_context
48 {
49 class impl;
50 impl* impl_;
51
52 public:
53 class executor_type;
54
55 /** Destroy the thread pool.
56
57 Signals all worker threads to stop, waits for them to
58 finish, and destroys any pending work items.
59
60 @pre No thread outside this pool may post or dispatch work to it
61 (or to a strand built on it) concurrently with, or after,
62 destruction. Doing so is undefined behavior. Submit such work
63 through @ref run_async or @ref run and call @ref join before
64 the pool is destroyed, so it has completed first. A thread
65 owned by one of this pool's services may post or dispatch
66 until that service's `shutdown()` returns.
67 */
68 ~thread_pool();
69
70 /** Construct a thread pool.
71
72 Records the requested worker count; no threads are created
73 yet. Threads start lazily on the executor's first `post()`.
74 If `num_threads` is zero, the number of threads is set to
75 the hardware concurrency, or one if that cannot be determined.
76
77 @param num_threads The number of worker threads, or zero
78 for automatic selection.
79
80 @param thread_name_prefix The prefix for worker thread names.
81 Thread names appear as "{prefix}0", "{prefix}1", etc.
82 The prefix is truncated to 12 characters. Defaults to
83 "capy-pool-".
84 */
85 explicit
86 thread_pool(
87 std::size_t num_threads = 0,
88 std::string_view thread_name_prefix = "capy-pool-");
89
90 /** Copy construction is disabled; a pool owns its worker threads.
91
92 @param other The pool that would be copied.
93 */
94 thread_pool(thread_pool const& other) = delete;
95
96 /** Copy assignment is disabled; a pool owns its worker threads.
97
98 @param other The pool that would be assigned from.
99
100 @return A reference to `*this`.
101 */
102 thread_pool& operator=(thread_pool const& other) = delete;
103
104 /** Wait for all outstanding work to complete.
105
106 Releases the internal work guard, then blocks the calling
107 thread until all outstanding work tracked by
108 @ref executor_type::on_work_started and
109 @ref executor_type::on_work_finished completes. After all
110 work finishes, joins the worker threads.
111
112 If @ref stop is called while `join()` is blocking, the
113 pool stops without waiting for remaining work to
114 complete. Worker threads finish their current item and
115 exit; `join()` still waits for all threads to be joined
116 before returning.
117
118 This function is idempotent. The first call performs the
119 join; subsequent calls return immediately.
120
121 @pre Must not be called from a thread in this pool (undefined
122 behavior).
123
124 @par Postconditions
125 All worker threads have been joined. The pool cannot be
126 reused.
127
128 @par Thread Safety
129 May be called from any thread not in this pool.
130 */
131 void
132 join() noexcept;
133
134 /** Request all worker threads to stop.
135
136 Signals all threads to exit after finishing their current
137 work item. Queued work that has not started stays queued
138 until the pool is destroyed, which abandons it. Does not
139 wait for threads to exit.
140
141 If @ref join is blocking on another thread, calling
142 `stop()` causes it to stop waiting for outstanding
143 work. The `join()` call still waits for worker threads
144 to finish their current item and exit before returning.
145
146 @par Thread Safety
147 May be called concurrently from any thread, including a
148 thread in this pool.
149 */
150 void
151 stop() noexcept;
152
153 /** Return an executor for this thread pool.
154
155 @return An executor associated with this thread pool.
156 */
157 executor_type
158 get_executor() const noexcept;
159 };
160
161 /** An executor that submits work to a thread_pool.
162
163 Executors are lightweight handles that can be copied and stored.
164 All copies refer to the same underlying thread pool.
165
166 @par Thread Safety
167 Distinct objects: Safe.
168 Shared objects: Safe.
169 */
170 class thread_pool::executor_type
171 {
172 friend class thread_pool;
173
174 thread_pool* pool_ = nullptr;
175
176 explicit
177 15321x executor_type(thread_pool& pool) noexcept
178 15321x : pool_(&pool)
179 {
180 15321x }
181
182 public:
183 /** Construct a default null executor.
184
185 The resulting executor is not associated with any pool.
186 `context()`, `dispatch()`, and `post()` require the
187 executor to be associated with a pool before use.
188 */
189 executor_type() = default;
190
191 /** Return the underlying thread pool.
192
193 @return A reference to the associated pool. The behavior is
194 undefined if the executor is not associated with a pool.
195 */
196 thread_pool&
197 12611x context() const noexcept
198 {
199 12611x return *pool_;
200 }
201
202 /** Notify that work has started.
203
204 Increments the outstanding work count. Must be paired
205 with a subsequent call to @ref on_work_finished.
206
207 @see @ref on_work_finished, @ref work_guard
208 */
209 BOOST_CAPY_DECL
210 void
211 on_work_started() const noexcept;
212
213 /** Notify that work has finished.
214
215 Decrements the outstanding work count. When the count
216 reaches zero after @ref thread_pool::join is called,
217 the pool's worker threads are signaled to stop.
218
219 @pre A preceding call to @ref on_work_started was made.
220
221 @see @ref on_work_started, @ref work_guard
222 */
223 BOOST_CAPY_DECL
224 void
225 on_work_finished() const noexcept;
226
227 /** Dispatch a continuation for execution.
228
229 If the calling thread is a worker of this pool, returns
230 `c.h` for symmetric transfer so the caller can resume the
231 continuation inline. Otherwise, posts the continuation to
232 the pool for execution on a worker thread and returns
233 `std::noop_coroutine()`.
234
235 @param c The continuation to execute. On the post path,
236 must remain at a stable address until dequeued
237 and resumed.
238
239 @return `c.h` when the calling thread is a pool worker;
240 `std::noop_coroutine()` otherwise.
241 */
242 BOOST_CAPY_DECL
243 std::coroutine_handle<>
244 dispatch(continuation& c) const;
245
246 /** Post a continuation to the thread pool.
247
248 The continuation is resumed on one of the pool's
249 worker threads. The continuation must remain at a stable
250 address until it is dequeued and resumed.
251
252 @param c The continuation to execute.
253 */
254 BOOST_CAPY_DECL
255 void
256 post(continuation& c) const;
257
258 /** Return true if two executors refer to the same thread pool.
259
260 @param other The executor to compare against.
261
262 @return `true` if both executors refer to the same pool.
263 */
264 bool
265 15x operator==(executor_type const& other) const noexcept
266 {
267 15x return pool_ == other.pool_;
268 }
269 };
270
271 } // capy
272 } // boost
273
274 #endif
275