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