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