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&) :199 15351x 100.0% 100.0% boost::capy::thread_pool::executor_type::context() const :219 12643x 100.0% 100.0% boost::capy::thread_pool::executor_type::operator==(boost::capy::thread_pool::executor_type const&) const :308 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 Stops the pool, joins its worker threads, then abandons any
58 work that has not started:
59
60 @li Calls @ref stop. Each worker thread finishes the item it
61 is running and exits; queued work is not run.
62 @li Joins the worker threads. Unlike @ref join, this does not
63 wait for outstanding work. If @ref join was already called,
64 the threads are already joined and this step does nothing.
65 @li Shuts down the pool's services, abandons every coroutine
66 still queued on the pool or its strands, then destroys the
67 services.
68 Each abandoned chain is unwound, without delivering a
69 result, on the executor of the coroutine that started it.
70 That happens during this destructor when the executor is
71 this pool, and later when it belongs to another execution
72 context.
73 See @ref chain_owner.
74
75 To let queued work finish first, call @ref join before the
76 pool is destroyed.
77
78 @pre No thread outside this pool may post or dispatch work to it
79 (or to a strand built on it) concurrently with, or after,
80 destruction. Doing so is undefined behavior. Submit such work
81 through @ref run_async and call @ref join, without calling
82 @ref stop, before the pool is destroyed, so it has completed
83 first. A thread owned by one of this pool's services may post
84 or dispatch until that service's `shutdown()` returns.
85
86 @pre Must not be called from a thread in this pool. A worker
87 thread cannot join itself, so this calls `std::terminate`.
88 */
89 ~thread_pool();
90
91 /** Construct a thread pool.
92
93 Records the requested worker count; no threads are created
94 yet. Threads start lazily on the executor's first `post()`.
95 If `num_threads` is zero, the number of threads is set to
96 the hardware concurrency, or one if that cannot be determined.
97
98 @param num_threads The number of worker threads, or zero
99 for automatic selection.
100
101 @param thread_name_prefix The prefix for worker thread names.
102 Thread names appear as "{prefix}0", "{prefix}1", etc.
103 The prefix is truncated to 12 characters. Defaults to
104 "capy-pool-".
105 */
106 explicit
107 thread_pool(
108 std::size_t num_threads = 0,
109 std::string_view thread_name_prefix = "capy-pool-");
110
111 /** Copy construction is disabled; a pool owns its worker threads.
112
113 @param other The pool that would be copied.
114 */
115 thread_pool(thread_pool const& other) = delete;
116
117 /** Copy assignment is disabled; a pool owns its worker threads.
118
119 @param other The pool that would be assigned from.
120
121 @return A reference to `*this`.
122 */
123 thread_pool& operator=(thread_pool const& other) = delete;
124
125 /** Wait for all outstanding work to complete.
126
127 Blocks the calling thread until all outstanding work
128 tracked by
129 @ref executor_type::on_work_started and
130 @ref executor_type::on_work_finished completes. After all
131 work finishes, joins the worker threads.
132
133 If @ref stop is called while `join()` is blocking, the
134 pool stops without waiting for remaining work to
135 complete. Worker threads finish their current item and
136 exit; `join()` still waits for all threads to be joined
137 before returning.
138
139 Only the first call performs the join. A later call returns
140 immediately, even while the first call is still blocking on
141 another thread.
142
143 @pre Must not be called from a thread in this pool (undefined
144 behavior).
145
146 @par Postconditions
147 If this call performed the join, all worker threads have
148 been joined. The pool cannot be reused.
149
150 @par Thread Safety
151 May be called from any thread not in this pool.
152 */
153 void
154 join() noexcept;
155
156 /** Request all worker threads to stop.
157
158 Signals all threads to exit after finishing their current
159 work item. Queued work that has not started stays queued
160 until the pool is destroyed, which abandons it. Does not
161 wait for threads to exit.
162
163 If @ref join is blocking on another thread, calling
164 `stop()` causes it to stop waiting for outstanding
165 work. The `join()` call still waits for worker threads
166 to finish their current item and exit before returning.
167
168 @par Thread Safety
169 May be called concurrently from any thread, including a
170 thread in this pool.
171 */
172 void
173 stop() noexcept;
174
175 /** Return an executor for this thread pool.
176
177 @return An executor associated with this thread pool.
178 */
179 executor_type
180 get_executor() const noexcept;
181 };
182
183 /** Runs continuations on the worker threads of a thread_pool.
184
185 Executors are lightweight handles that can be copied and stored.
186 All copies refer to the same underlying thread pool.
187
188 @par Thread Safety
189 Distinct objects: Safe.
190 Shared objects: Safe.
191 */
192 class thread_pool::executor_type
193 {
194 friend class thread_pool;
195
196 thread_pool* pool_ = nullptr;
197
198 explicit
199 15351x executor_type(thread_pool& pool) noexcept
200 15351x : pool_(&pool)
201 {
202 15351x }
203
204 public:
205 /** Construct a default null executor.
206
207 The resulting executor is not associated with any pool.
208 `context()`, `dispatch()`, and `post()` require the
209 executor to be associated with a pool before use.
210 */
211 executor_type() = default;
212
213 /** Return the underlying thread pool.
214
215 @return A reference to the associated pool. The behavior is
216 undefined if the executor is not associated with a pool.
217 */
218 thread_pool&
219 12643x context() const noexcept
220 {
221 12643x return *pool_;
222 }
223
224 /** Notify that work has started.
225
226 Increments the outstanding work count. Must be paired
227 with a subsequent call to @ref on_work_finished.
228
229 @see on_work_finished, work_guard
230 */
231 BOOST_CAPY_DECL
232 void
233 on_work_started() const noexcept;
234
235 /** Notify that work has finished.
236
237 Decrements the outstanding work count. When the count
238 reaches zero after @ref thread_pool::join is called,
239 the pool's worker threads are signaled to stop.
240
241 @pre A preceding call to @ref on_work_started was made.
242
243 @see on_work_started, work_guard
244 */
245 BOOST_CAPY_DECL
246 void
247 on_work_finished() const noexcept;
248
249 /** Dispatch a continuation for execution.
250
251 If the calling thread is a worker of this pool, returns
252 `c.h` for symmetric transfer so the caller can resume the
253 continuation inline. Otherwise, posts the continuation as
254 @ref post does and returns `std::noop_coroutine()`. On that
255 path, the continuation is not run if the pool stops first,
256 and a call that races with @ref thread_pool::stop can go
257 either way. See @ref post.
258
259 @param c The continuation to execute. On the post path,
260 must remain at a stable address until resumed or
261 abandoned.
262
263 @return `c.h` when the calling thread is a pool worker;
264 `std::noop_coroutine()` otherwise.
265
266 @throws std::system_error If this call posts and must start
267 the pool's worker threads, and a thread cannot be started.
268 `c` is not queued.
269 */
270 BOOST_CAPY_DECL
271 std::coroutine_handle<>
272 dispatch(continuation& c) const;
273
274 /** Post a continuation to the thread pool.
275
276 The continuation is resumed on one of the pool's worker
277 threads, unless the pool stops first. The pool stops when
278 @ref thread_pool::stop is called, or when
279 @ref thread_pool::join has finished waiting for outstanding
280 work. A continuation still queued when the pool stops, or
281 posted after it stops, is not run. It stays queued until
282 the pool is destroyed, which abandons it.
283
284 A call that races with @ref thread_pool::stop on another
285 thread can therefore go either way, and nothing in the call
286 tells the caller which.
287
288 The continuation must remain at a stable address until it
289 is resumed or abandoned.
290
291 @param c The continuation to execute.
292
293 @throws std::system_error If this call must start the pool's
294 worker threads, and a thread cannot be started. `c` is not
295 queued.
296 */
297 BOOST_CAPY_DECL
298 void
299 post(continuation& c) const;
300
301 /** Return true if two executors refer to the same thread pool.
302
303 @param other The executor to compare against.
304
305 @return `true` if both executors refer to the same pool.
306 */
307 bool
308 15x operator==(executor_type const& other) const noexcept
309 {
310 15x return pool_ == other.pool_;
311 }
312 };
313
314 } // capy
315 } // boost
316
317 #endif
318