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 is abandoned.
136 : Does not wait for threads to exit.
137 :
138 : If @ref join is blocking on another thread, calling
139 : `stop()` causes it to stop waiting for outstanding
140 : work. The `join()` call still waits for worker threads
141 : to finish their current item and exit before returning.
142 :
143 : @par Thread Safety
144 : May be called concurrently from any thread, including a
145 : thread in this pool.
146 : */
147 : void
148 : stop() noexcept;
149 :
150 : /** Return an executor for this thread pool.
151 :
152 : @return An executor associated with this thread pool.
153 : */
154 : executor_type
155 : get_executor() const noexcept;
156 : };
157 :
158 : /** An executor that submits work to a thread_pool.
159 :
160 : Executors are lightweight handles that can be copied and stored.
161 : All copies refer to the same underlying thread pool.
162 :
163 : @par Thread Safety
164 : Distinct objects: Safe.
165 : Shared objects: Safe.
166 : */
167 : class thread_pool::executor_type
168 : {
169 : friend class thread_pool;
170 :
171 : thread_pool* pool_ = nullptr;
172 :
173 : explicit
174 HIT 11809 : executor_type(thread_pool& pool) noexcept
175 11809 : : pool_(&pool)
176 : {
177 11809 : }
178 :
179 : public:
180 : /** Construct a default null executor.
181 :
182 : The resulting executor is not associated with any pool.
183 : `context()`, `dispatch()`, and `post()` require the
184 : executor to be associated with a pool before use.
185 : */
186 : executor_type() = default;
187 :
188 : /** Return the underlying thread pool.
189 :
190 : @return A reference to the associated pool. The behavior is
191 : undefined if the executor is not associated with a pool.
192 : */
193 : thread_pool&
194 12113 : context() const noexcept
195 : {
196 12113 : return *pool_;
197 : }
198 :
199 : /** Notify that work has started.
200 :
201 : Increments the outstanding work count. Must be paired
202 : with a subsequent call to @ref on_work_finished.
203 :
204 : @see on_work_finished, work_guard
205 : */
206 : BOOST_CAPY_DECL
207 : void
208 : on_work_started() const noexcept;
209 :
210 : /** Notify that work has finished.
211 :
212 : Decrements the outstanding work count. When the count
213 : reaches zero after @ref thread_pool::join is called,
214 : the pool's worker threads are signaled to stop.
215 :
216 : @pre A preceding call to @ref on_work_started was made.
217 :
218 : @see on_work_started, work_guard
219 : */
220 : BOOST_CAPY_DECL
221 : void
222 : on_work_finished() const noexcept;
223 :
224 : /** Dispatch a continuation for execution.
225 :
226 : If the calling thread is a worker of this pool, returns
227 : `c.h` for symmetric transfer so the caller can resume the
228 : continuation inline. Otherwise, posts the continuation to
229 : the pool for execution on a worker thread and returns
230 : `std::noop_coroutine()`.
231 :
232 : @param c The continuation to execute. On the post path,
233 : must remain at a stable address until dequeued
234 : and resumed.
235 :
236 : @return `c.h` when the calling thread is a pool worker;
237 : `std::noop_coroutine()` otherwise.
238 : */
239 : BOOST_CAPY_DECL
240 : std::coroutine_handle<>
241 : dispatch(continuation& c) const;
242 :
243 : /** Post a continuation to the thread pool.
244 :
245 : The continuation is resumed on one of the pool's
246 : worker threads. The continuation must remain at a stable
247 : address until it is dequeued and resumed.
248 :
249 : @param c The continuation to execute.
250 : */
251 : BOOST_CAPY_DECL
252 : void
253 : post(continuation& c) const;
254 :
255 : /** Return true if two executors refer to the same thread pool.
256 :
257 : @param other The executor to compare against.
258 :
259 : @return `true` if both executors refer to the same pool.
260 : */
261 : bool
262 13 : operator==(executor_type const& other) const noexcept
263 : {
264 13 : return pool_ == other.pool_;
265 : }
266 : };
267 :
268 : } // capy
269 : } // boost
270 :
271 : #endif
|
