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 : @code
38 : thread_pool pool(4); // 4 worker threads
39 : auto ex = pool.get_executor();
40 : run_async(ex)(some_task()); // start work; tracked so join() waits for it
41 : pool.join(); // wait for outstanding work to complete
42 : // pool destructor stops the pool, discarding any pending work
43 : @endcode
44 :
45 : @note `join()` waits only for work that holds outstanding-work
46 : counting, which `run_async` (and `make_work_guard`) provide. A bare
47 : `executor_type::post()` does not register outstanding work, so
48 : `join()` does not wait for it.
49 : */
50 : class BOOST_CAPY_DECL
51 : thread_pool
52 : : public execution_context
53 : {
54 : class impl;
55 : impl* impl_;
56 :
57 : public:
58 : class executor_type;
59 :
60 : /** Destroy the thread pool.
61 :
62 : Signals all worker threads to stop, waits for them to
63 : finish, and destroys any pending work items.
64 :
65 : @pre No thread outside this pool may post or dispatch work to it
66 : (or to a strand built on it) concurrently with, or after,
67 : destruction. Doing so is undefined behavior. Submit such work
68 : through @ref run_async or @ref run and call @ref join before
69 : the pool is destroyed, so it has completed first.
70 : */
71 : ~thread_pool();
72 :
73 : /** Construct a thread pool.
74 :
75 : Records the requested worker count; no threads are created
76 : yet. Threads start lazily on the executor's first `post()`.
77 : If `num_threads` is zero, the number of threads is set to
78 : the hardware concurrency, or one if that cannot be determined.
79 :
80 : @param num_threads The number of worker threads, or zero
81 : for automatic selection.
82 :
83 : @param thread_name_prefix The prefix for worker thread names.
84 : Thread names appear as "{prefix}0", "{prefix}1", etc.
85 : The prefix is truncated to 12 characters. Defaults to
86 : "capy-pool-".
87 : */
88 : explicit
89 : thread_pool(
90 : std::size_t num_threads = 0,
91 : std::string_view thread_name_prefix = "capy-pool-");
92 :
93 : /** Copy construction is disabled; a pool owns its worker threads.
94 :
95 : @param other The pool that would be copied.
96 : */
97 : thread_pool(thread_pool const& other) = delete;
98 :
99 : /** Copy assignment is disabled; a pool owns its worker threads.
100 :
101 : @param other The pool that would be assigned from.
102 :
103 : @return A reference to `*this`.
104 : */
105 : thread_pool& operator=(thread_pool const& other) = delete;
106 :
107 : /** Wait for all outstanding work to complete.
108 :
109 : Releases the internal work guard, then blocks the calling
110 : thread until all outstanding work tracked by
111 : @ref executor_type::on_work_started and
112 : @ref executor_type::on_work_finished completes. After all
113 : work finishes, joins the worker threads.
114 :
115 : If @ref stop is called while `join()` is blocking, the
116 : pool stops without waiting for remaining work to
117 : complete. Worker threads finish their current item and
118 : exit; `join()` still waits for all threads to be joined
119 : before returning.
120 :
121 : This function is idempotent. The first call performs the
122 : join; subsequent calls return immediately.
123 :
124 : @pre Must not be called from a thread in this pool (undefined
125 : behavior).
126 :
127 : @par Postconditions
128 : All worker threads have been joined. The pool cannot be
129 : reused.
130 :
131 : @par Thread Safety
132 : May be called from any thread not in this pool.
133 : */
134 : void
135 : join() noexcept;
136 :
137 : /** Request all worker threads to stop.
138 :
139 : Signals all threads to exit after finishing their current
140 : work item. Queued work that has not started is abandoned.
141 : Does not wait for threads to exit.
142 :
143 : If @ref join is blocking on another thread, calling
144 : `stop()` causes it to stop waiting for outstanding
145 : work. The `join()` call still waits for worker threads
146 : to finish their current item and exit before returning.
147 :
148 : @par Thread Safety
149 : May be called concurrently from any thread, including a
150 : thread in this pool.
151 : */
152 : void
153 : stop() noexcept;
154 :
155 : /** Return an executor for this thread pool.
156 :
157 : @return An executor associated with this thread pool.
158 : */
159 : executor_type
160 : get_executor() const noexcept;
161 : };
162 :
163 : /** An executor that submits work to a thread_pool.
164 :
165 : Executors are lightweight handles that can be copied and stored.
166 : All copies refer to the same underlying thread pool.
167 :
168 : @par Thread Safety
169 : Distinct objects: Safe.
170 : Shared objects: Safe.
171 : */
172 : class thread_pool::executor_type
173 : {
174 : friend class thread_pool;
175 :
176 : thread_pool* pool_ = nullptr;
177 :
178 : explicit
179 HIT 11800 : executor_type(thread_pool& pool) noexcept
180 11800 : : pool_(&pool)
181 : {
182 11800 : }
183 :
184 : public:
185 : /** Construct a default null executor.
186 :
187 : The resulting executor is not associated with any pool.
188 : `context()`, `dispatch()`, and `post()` require the
189 : executor to be associated with a pool before use.
190 : */
191 : executor_type() = default;
192 :
193 : /** Return the underlying thread pool.
194 :
195 : @return A reference to the associated pool. The behavior is
196 : undefined if the executor is not associated with a pool.
197 : */
198 : thread_pool&
199 12104 : context() const noexcept
200 : {
201 12104 : return *pool_;
202 : }
203 :
204 : /** Notify that work has started.
205 :
206 : Increments the outstanding work count. Must be paired
207 : with a subsequent call to @ref on_work_finished.
208 :
209 : @see on_work_finished, work_guard
210 : */
211 : BOOST_CAPY_DECL
212 : void
213 : on_work_started() const noexcept;
214 :
215 : /** Notify that work has finished.
216 :
217 : Decrements the outstanding work count. When the count
218 : reaches zero after @ref thread_pool::join is called,
219 : the pool's worker threads are signaled to stop.
220 :
221 : @pre A preceding call to @ref on_work_started was made.
222 :
223 : @see on_work_started, work_guard
224 : */
225 : BOOST_CAPY_DECL
226 : void
227 : on_work_finished() const noexcept;
228 :
229 : /** Dispatch a continuation for execution.
230 :
231 : If the calling thread is a worker of this pool, returns
232 : `c.h` for symmetric transfer so the caller can resume the
233 : continuation inline. Otherwise, posts the continuation to
234 : the pool for execution on a worker thread and returns
235 : `std::noop_coroutine()`.
236 :
237 : @param c The continuation to execute. On the post path,
238 : must remain at a stable address until dequeued
239 : and resumed.
240 :
241 : @return `c.h` when the calling thread is a pool worker;
242 : `std::noop_coroutine()` otherwise.
243 : */
244 : BOOST_CAPY_DECL
245 : std::coroutine_handle<>
246 : dispatch(continuation& c) const;
247 :
248 : /** Post a continuation to the thread pool.
249 :
250 : The continuation is resumed on one of the pool's
251 : worker threads. The continuation must remain at a stable
252 : address until it is dequeued and resumed.
253 :
254 : @param c The continuation to execute.
255 : */
256 : BOOST_CAPY_DECL
257 : void
258 : post(continuation& c) const;
259 :
260 : /** Return true if two executors refer to the same thread pool.
261 :
262 : @param other The executor to compare against.
263 :
264 : @return `true` if both executors refer to the same pool.
265 : */
266 : bool
267 13 : operator==(executor_type const& other) const noexcept
268 : {
269 13 : return pool_ == other.pool_;
270 : }
271 : };
272 :
273 : } // capy
274 : } // boost
275 :
276 : #endif
|
