TLA Line data 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 HIT 15344 : executor_type(thread_pool& pool) noexcept
200 15344 : : pool_(&pool)
201 : {
202 15344 : }
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 12632 : context() const noexcept
220 : {
221 12632 : 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 @ref on_work_finished, @ref 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 @ref on_work_started, @ref 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 : Threads that did start keep running, and a later call
270 : starts only the missing ones.
271 : */
272 : BOOST_CAPY_DECL
273 : std::coroutine_handle<>
274 : dispatch(continuation& c) const;
275 :
276 : /** Post a continuation to the thread pool.
277 :
278 : The continuation is resumed on one of the pool's worker
279 : threads, unless the pool stops first. The pool stops when
280 : @ref thread_pool::stop is called, or when
281 : @ref thread_pool::join has finished waiting for outstanding
282 : work. A continuation still queued when the pool stops, or
283 : posted after it stops, is not run. It stays queued until
284 : the pool is destroyed, which abandons it.
285 :
286 : A call that races with @ref thread_pool::stop on another
287 : thread can therefore go either way, and nothing in the call
288 : tells the caller which.
289 :
290 : The continuation must remain at a stable address until it
291 : is resumed or abandoned.
292 :
293 : @param c The continuation to execute.
294 :
295 : @throws std::system_error If this call must start the pool's
296 : worker threads, and a thread cannot be started. `c` is not
297 : queued.
298 : Threads that did start keep running, and a later call
299 : starts only the missing ones.
300 : */
301 : BOOST_CAPY_DECL
302 : void
303 : post(continuation& c) const;
304 :
305 : /** Return true if two executors refer to the same thread pool.
306 :
307 : @param other The executor to compare against.
308 :
309 : @return `true` if both executors refer to the same pool.
310 : */
311 : bool
312 15 : operator==(executor_type const& other) const noexcept
313 : {
314 15 : return pool_ == other.pool_;
315 : }
316 : };
317 :
318 : } // capy
319 : } // boost
320 :
321 : #endif
|
