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 : 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 HIT 15321 : executor_type(thread_pool& pool) noexcept
178 15321 : : pool_(&pool)
179 : {
180 15321 : }
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 12611 : context() const noexcept
198 : {
199 12611 : 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 15 : operator==(executor_type const& other) const noexcept
266 : {
267 15 : return pool_ == other.pool_;
268 : }
269 : };
270 :
271 : } // capy
272 : } // boost
273 :
274 : #endif
|
