. /include/boost/capy /ex /thread_pool.hpp
100.00% Lines (7/7) 100.00% Functions (3/3)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com) 2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
4   // 4   //
5   // Distributed under the Boost Software License, Version 1.0. (See accompanying 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) 6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7   // 7   //
8   // Official repository: https://github.com/cppalliance/capy 8   // Official repository: https://github.com/cppalliance/capy
9   // 9   //
10   10  
11   #ifndef BOOST_CAPY_EX_THREAD_POOL_HPP 11   #ifndef BOOST_CAPY_EX_THREAD_POOL_HPP
12   #define BOOST_CAPY_EX_THREAD_POOL_HPP 12   #define BOOST_CAPY_EX_THREAD_POOL_HPP
13   13  
14   #include <boost/capy/detail/config.hpp> 14   #include <boost/capy/detail/config.hpp>
15   #include <boost/capy/continuation.hpp> 15   #include <boost/capy/continuation.hpp>
16   #include <coroutine> 16   #include <coroutine>
17   #include <boost/capy/ex/execution_context.hpp> 17   #include <boost/capy/ex/execution_context.hpp>
18   #include <cstddef> 18   #include <cstddef>
19   #include <string_view> 19   #include <string_view>
20   20  
21   namespace boost { 21   namespace boost {
22   namespace capy { 22   namespace capy {
23   23  
24   /** Distributes posted work across a fixed group of worker threads via a shared queue. 24   /** Distributes posted work across a fixed group of worker threads via a shared queue.
25   25  
26   Use this when you need to run coroutines on multiple threads 26   Use this when you need to run coroutines on multiple threads
27   without the overhead of creating and destroying threads for 27   without the overhead of creating and destroying threads for
28   each task. Work items are distributed across the pool using 28   each task. Work items are distributed across the pool using
29   a shared queue. 29   a shared queue.
30   30  
31   @par Thread Safety 31   @par Thread Safety
32   Distinct objects: Safe. 32   Distinct objects: Safe.
33   Shared objects: Safe for @ref get_executor, @ref join, and 33   Shared objects: Safe for @ref get_executor, @ref join, and
34   @ref stop. Unsafe for construction and destruction. 34   @ref stop. Unsafe for construction and destruction.
35   35  
36   @par Example 36   @par Example
37   @par !example example 37   @par !example example
38   38  
39   39  
40   @note `join()` waits only for work that holds outstanding-work 40   @note `join()` waits only for work that holds outstanding-work
41   counting, which `run_async` (and `make_work_guard`) provide. A bare 41   counting, which `run_async` (and `make_work_guard`) provide. A bare
42   `executor_type::post()` does not register outstanding work, so 42   `executor_type::post()` does not register outstanding work, so
43   `join()` does not wait for it. 43   `join()` does not wait for it.
44   */ 44   */
45   class BOOST_CAPY_DECL 45   class BOOST_CAPY_DECL
46   thread_pool 46   thread_pool
47   : public execution_context 47   : public execution_context
48   { 48   {
49   class impl; 49   class impl;
50   impl* impl_; 50   impl* impl_;
51   51  
52   public: 52   public:
53   class executor_type; 53   class executor_type;
54   54  
55   /** Destroy the thread pool. 55   /** Destroy the thread pool.
56   56  
57   Stops the pool, joins its worker threads, then abandons any 57   Stops the pool, joins its worker threads, then abandons any
58   work that has not started: 58   work that has not started:
59   59  
60   @li Calls @ref stop. Each worker thread finishes the item it 60   @li Calls @ref stop. Each worker thread finishes the item it
61   is running and exits; queued work is not run. 61   is running and exits; queued work is not run.
62   @li Joins the worker threads. Unlike @ref join, this does not 62   @li Joins the worker threads. Unlike @ref join, this does not
63   wait for outstanding work. If @ref join was already called, 63   wait for outstanding work. If @ref join was already called,
64   the threads are already joined and this step does nothing. 64   the threads are already joined and this step does nothing.
65   @li Shuts down the pool's services, abandons every coroutine 65   @li Shuts down the pool's services, abandons every coroutine
66   still queued on the pool or its strands, then destroys the 66   still queued on the pool or its strands, then destroys the
67   services. 67   services.
68   Each abandoned chain is unwound, without delivering a 68   Each abandoned chain is unwound, without delivering a
69   result, on the executor of the coroutine that started it. 69   result, on the executor of the coroutine that started it.
70   That happens during this destructor when the executor is 70   That happens during this destructor when the executor is
71   this pool, and later when it belongs to another execution 71   this pool, and later when it belongs to another execution
72   context. 72   context.
73   See @ref chain_owner. 73   See @ref chain_owner.
74   74  
75   To let queued work finish first, call @ref join before the 75   To let queued work finish first, call @ref join before the
76   pool is destroyed. 76   pool is destroyed.
77   77  
78   @pre No thread outside this pool may post or dispatch work to it 78   @pre No thread outside this pool may post or dispatch work to it
79   (or to a strand built on it) concurrently with, or after, 79   (or to a strand built on it) concurrently with, or after,
80   destruction. Doing so is undefined behavior. Submit such work 80   destruction. Doing so is undefined behavior. Submit such work
81   through @ref run_async and call @ref join, without calling 81   through @ref run_async and call @ref join, without calling
82   @ref stop, before the pool is destroyed, so it has completed 82   @ref stop, before the pool is destroyed, so it has completed
83   first. A thread owned by one of this pool's services may post 83   first. A thread owned by one of this pool's services may post
84   or dispatch until that service's `shutdown()` returns. 84   or dispatch until that service's `shutdown()` returns.
85   85  
86   @pre Must not be called from a thread in this pool. A worker 86   @pre Must not be called from a thread in this pool. A worker
87   thread cannot join itself, so this calls `std::terminate`. 87   thread cannot join itself, so this calls `std::terminate`.
88   */ 88   */
89   ~thread_pool(); 89   ~thread_pool();
90   90  
91   /** Construct a thread pool. 91   /** Construct a thread pool.
92   92  
93   Records the requested worker count; no threads are created 93   Records the requested worker count; no threads are created
94   yet. Threads start lazily on the executor's first `post()`. 94   yet. Threads start lazily on the executor's first `post()`.
95   If `num_threads` is zero, the number of threads is set to 95   If `num_threads` is zero, the number of threads is set to
96   the hardware concurrency, or one if that cannot be determined. 96   the hardware concurrency, or one if that cannot be determined.
97   97  
98   @param num_threads The number of worker threads, or zero 98   @param num_threads The number of worker threads, or zero
99   for automatic selection. 99   for automatic selection.
100   100  
101   @param thread_name_prefix The prefix for worker thread names. 101   @param thread_name_prefix The prefix for worker thread names.
102   Thread names appear as "{prefix}0", "{prefix}1", etc. 102   Thread names appear as "{prefix}0", "{prefix}1", etc.
103   The prefix is truncated to 12 characters. Defaults to 103   The prefix is truncated to 12 characters. Defaults to
104   "capy-pool-". 104   "capy-pool-".
105   */ 105   */
106   explicit 106   explicit
107   thread_pool( 107   thread_pool(
108   std::size_t num_threads = 0, 108   std::size_t num_threads = 0,
109   std::string_view thread_name_prefix = "capy-pool-"); 109   std::string_view thread_name_prefix = "capy-pool-");
110   110  
111   /** Copy construction is disabled; a pool owns its worker threads. 111   /** Copy construction is disabled; a pool owns its worker threads.
112   112  
113   @param other The pool that would be copied. 113   @param other The pool that would be copied.
114   */ 114   */
115   thread_pool(thread_pool const& other) = delete; 115   thread_pool(thread_pool const& other) = delete;
116   116  
117   /** Copy assignment is disabled; a pool owns its worker threads. 117   /** Copy assignment is disabled; a pool owns its worker threads.
118   118  
119   @param other The pool that would be assigned from. 119   @param other The pool that would be assigned from.
120   120  
121   @return A reference to `*this`. 121   @return A reference to `*this`.
122   */ 122   */
123   thread_pool& operator=(thread_pool const& other) = delete; 123   thread_pool& operator=(thread_pool const& other) = delete;
124   124  
125   /** Wait for all outstanding work to complete. 125   /** Wait for all outstanding work to complete.
126   126  
127   Blocks the calling thread until all outstanding work 127   Blocks the calling thread until all outstanding work
128   tracked by 128   tracked by
129   @ref executor_type::on_work_started and 129   @ref executor_type::on_work_started and
130   @ref executor_type::on_work_finished completes. After all 130   @ref executor_type::on_work_finished completes. After all
131   work finishes, joins the worker threads. 131   work finishes, joins the worker threads.
132   132  
133   If @ref stop is called while `join()` is blocking, the 133   If @ref stop is called while `join()` is blocking, the
134   pool stops without waiting for remaining work to 134   pool stops without waiting for remaining work to
135   complete. Worker threads finish their current item and 135   complete. Worker threads finish their current item and
136   exit; `join()` still waits for all threads to be joined 136   exit; `join()` still waits for all threads to be joined
137   before returning. 137   before returning.
138   138  
139   Only the first call performs the join. A later call returns 139   Only the first call performs the join. A later call returns
140   immediately, even while the first call is still blocking on 140   immediately, even while the first call is still blocking on
141   another thread. 141   another thread.
142   142  
143   @pre Must not be called from a thread in this pool (undefined 143   @pre Must not be called from a thread in this pool (undefined
144   behavior). 144   behavior).
145   145  
146   @par Postconditions 146   @par Postconditions
147   If this call performed the join, all worker threads have 147   If this call performed the join, all worker threads have
148   been joined. The pool cannot be reused. 148   been joined. The pool cannot be reused.
149   149  
150   @par Thread Safety 150   @par Thread Safety
151   May be called from any thread not in this pool. 151   May be called from any thread not in this pool.
152   */ 152   */
153   void 153   void
154   join() noexcept; 154   join() noexcept;
155   155  
156   /** Request all worker threads to stop. 156   /** Request all worker threads to stop.
157   157  
158   Signals all threads to exit after finishing their current 158   Signals all threads to exit after finishing their current
159   work item. Queued work that has not started stays queued 159   work item. Queued work that has not started stays queued
160   until the pool is destroyed, which abandons it. Does not 160   until the pool is destroyed, which abandons it. Does not
161   wait for threads to exit. 161   wait for threads to exit.
162   162  
163   If @ref join is blocking on another thread, calling 163   If @ref join is blocking on another thread, calling
164   `stop()` causes it to stop waiting for outstanding 164   `stop()` causes it to stop waiting for outstanding
165   work. The `join()` call still waits for worker threads 165   work. The `join()` call still waits for worker threads
166   to finish their current item and exit before returning. 166   to finish their current item and exit before returning.
167   167  
168   @par Thread Safety 168   @par Thread Safety
169   May be called concurrently from any thread, including a 169   May be called concurrently from any thread, including a
170   thread in this pool. 170   thread in this pool.
171   */ 171   */
172   void 172   void
173   stop() noexcept; 173   stop() noexcept;
174   174  
175   /** Return an executor for this thread pool. 175   /** Return an executor for this thread pool.
176   176  
177   @return An executor associated with this thread pool. 177   @return An executor associated with this thread pool.
178   */ 178   */
179   executor_type 179   executor_type
180   get_executor() const noexcept; 180   get_executor() const noexcept;
181   }; 181   };
182   182  
183   /** Runs continuations on the worker threads of a thread_pool. 183   /** Runs continuations on the worker threads of a thread_pool.
184   184  
185   Executors are lightweight handles that can be copied and stored. 185   Executors are lightweight handles that can be copied and stored.
186   All copies refer to the same underlying thread pool. 186   All copies refer to the same underlying thread pool.
187   187  
188   @par Thread Safety 188   @par Thread Safety
189   Distinct objects: Safe. 189   Distinct objects: Safe.
190   Shared objects: Safe. 190   Shared objects: Safe.
191   */ 191   */
192   class thread_pool::executor_type 192   class thread_pool::executor_type
193   { 193   {
194   friend class thread_pool; 194   friend class thread_pool;
195   195  
196   thread_pool* pool_ = nullptr; 196   thread_pool* pool_ = nullptr;
197   197  
198   explicit 198   explicit
HITCBC 199   15346 executor_type(thread_pool& pool) noexcept 199   15344 executor_type(thread_pool& pool) noexcept
HITCBC 200   15346 : pool_(&pool) 200   15344 : pool_(&pool)
201   { 201   {
HITCBC 202   15346 } 202   15344 }
203   203  
204   public: 204   public:
205   /** Construct a default null executor. 205   /** Construct a default null executor.
206   206  
207   The resulting executor is not associated with any pool. 207   The resulting executor is not associated with any pool.
208   `context()`, `dispatch()`, and `post()` require the 208   `context()`, `dispatch()`, and `post()` require the
209   executor to be associated with a pool before use. 209   executor to be associated with a pool before use.
210   */ 210   */
211   executor_type() = default; 211   executor_type() = default;
212   212  
213   /** Return the underlying thread pool. 213   /** Return the underlying thread pool.
214   214  
215   @return A reference to the associated pool. The behavior is 215   @return A reference to the associated pool. The behavior is
216   undefined if the executor is not associated with a pool. 216   undefined if the executor is not associated with a pool.
217   */ 217   */
218   thread_pool& 218   thread_pool&
HITCBC 219   12638 context() const noexcept 219   12632 context() const noexcept
220   { 220   {
HITCBC 221   12638 return *pool_; 221   12632 return *pool_;
222   } 222   }
223   223  
224   /** Notify that work has started. 224   /** Notify that work has started.
225   225  
226   Increments the outstanding work count. Must be paired 226   Increments the outstanding work count. Must be paired
227   with a subsequent call to @ref on_work_finished. 227   with a subsequent call to @ref on_work_finished.
228   228  
229   @see @ref on_work_finished, @ref work_guard 229   @see @ref on_work_finished, @ref work_guard
230   */ 230   */
231   BOOST_CAPY_DECL 231   BOOST_CAPY_DECL
232   void 232   void
233   on_work_started() const noexcept; 233   on_work_started() const noexcept;
234   234  
235   /** Notify that work has finished. 235   /** Notify that work has finished.
236   236  
237   Decrements the outstanding work count. When the count 237   Decrements the outstanding work count. When the count
238   reaches zero after @ref thread_pool::join is called, 238   reaches zero after @ref thread_pool::join is called,
239   the pool's worker threads are signaled to stop. 239   the pool's worker threads are signaled to stop.
240   240  
241   @pre A preceding call to @ref on_work_started was made. 241   @pre A preceding call to @ref on_work_started was made.
242   242  
243   @see @ref on_work_started, @ref work_guard 243   @see @ref on_work_started, @ref work_guard
244   */ 244   */
245   BOOST_CAPY_DECL 245   BOOST_CAPY_DECL
246   void 246   void
247   on_work_finished() const noexcept; 247   on_work_finished() const noexcept;
248   248  
249   /** Dispatch a continuation for execution. 249   /** Dispatch a continuation for execution.
250   250  
251   If the calling thread is a worker of this pool, returns 251   If the calling thread is a worker of this pool, returns
252   `c.h` for symmetric transfer so the caller can resume the 252   `c.h` for symmetric transfer so the caller can resume the
253   continuation inline. Otherwise, posts the continuation as 253   continuation inline. Otherwise, posts the continuation as
254   @ref post does and returns `std::noop_coroutine()`. On that 254   @ref post does and returns `std::noop_coroutine()`. On that
255   path, the continuation is not run if the pool stops first, 255   path, the continuation is not run if the pool stops first,
256   and a call that races with @ref thread_pool::stop can go 256   and a call that races with @ref thread_pool::stop can go
257   either way. See @ref post. 257   either way. See @ref post.
258   258  
259   @param c The continuation to execute. On the post path, 259   @param c The continuation to execute. On the post path,
260   must remain at a stable address until resumed or 260   must remain at a stable address until resumed or
261   abandoned. 261   abandoned.
262   262  
263   @return `c.h` when the calling thread is a pool worker; 263   @return `c.h` when the calling thread is a pool worker;
264   `std::noop_coroutine()` otherwise. 264   `std::noop_coroutine()` otherwise.
265   265  
266   @throws std::system_error If this call posts and must start 266   @throws std::system_error If this call posts and must start
267   the pool's worker threads, and a thread cannot be started. 267   the pool's worker threads, and a thread cannot be started.
268   `c` is not queued. 268   `c` is not queued.
  269 + Threads that did start keep running, and a later call
  270 + starts only the missing ones.
269   */ 271   */
270   BOOST_CAPY_DECL 272   BOOST_CAPY_DECL
271   std::coroutine_handle<> 273   std::coroutine_handle<>
272   dispatch(continuation& c) const; 274   dispatch(continuation& c) const;
273   275  
274   /** Post a continuation to the thread pool. 276   /** Post a continuation to the thread pool.
275   277  
276   The continuation is resumed on one of the pool's worker 278   The continuation is resumed on one of the pool's worker
277   threads, unless the pool stops first. The pool stops when 279   threads, unless the pool stops first. The pool stops when
278   @ref thread_pool::stop is called, or when 280   @ref thread_pool::stop is called, or when
279   @ref thread_pool::join has finished waiting for outstanding 281   @ref thread_pool::join has finished waiting for outstanding
280   work. A continuation still queued when the pool stops, or 282   work. A continuation still queued when the pool stops, or
281   posted after it stops, is not run. It stays queued until 283   posted after it stops, is not run. It stays queued until
282   the pool is destroyed, which abandons it. 284   the pool is destroyed, which abandons it.
283   285  
284   A call that races with @ref thread_pool::stop on another 286   A call that races with @ref thread_pool::stop on another
285   thread can therefore go either way, and nothing in the call 287   thread can therefore go either way, and nothing in the call
286   tells the caller which. 288   tells the caller which.
287   289  
288   The continuation must remain at a stable address until it 290   The continuation must remain at a stable address until it
289   is resumed or abandoned. 291   is resumed or abandoned.
290   292  
291   @param c The continuation to execute. 293   @param c The continuation to execute.
292   294  
293   @throws std::system_error If this call must start the pool's 295   @throws std::system_error If this call must start the pool's
294   worker threads, and a thread cannot be started. `c` is not 296   worker threads, and a thread cannot be started. `c` is not
295   queued. 297   queued.
  298 + Threads that did start keep running, and a later call
  299 + starts only the missing ones.
296   */ 300   */
297   BOOST_CAPY_DECL 301   BOOST_CAPY_DECL
298   void 302   void
299   post(continuation& c) const; 303   post(continuation& c) const;
300   304  
301   /** Return true if two executors refer to the same thread pool. 305   /** Return true if two executors refer to the same thread pool.
302   306  
303   @param other The executor to compare against. 307   @param other The executor to compare against.
304   308  
305   @return `true` if both executors refer to the same pool. 309   @return `true` if both executors refer to the same pool.
306   */ 310   */
307   bool 311   bool
HITCBC 308   15 operator==(executor_type const& other) const noexcept 312   15 operator==(executor_type const& other) const noexcept
309   { 313   {
HITCBC 310   15 return pool_ == other.pool_; 314   15 return pool_ == other.pool_;
311   } 315   }
312   }; 316   };
313   317  
314   } // capy 318   } // capy
315   } // boost 319   } // boost
316   320  
317   #endif 321   #endif