. /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 - Signals all worker threads to stop, waits for them to 57 + Stops the pool, joins its worker threads, then abandons any
58 - finish, and destroys any pending work items. 58 + work that has not started:
  59 +
  60 + @li Calls @ref stop. Each worker thread finishes the item it
  61 + is running and exits; queued work is not run.
  62 + @li Joins the worker threads. Unlike @ref join, this does not
  63 + wait for outstanding work. If @ref join was already called,
  64 + the threads are already joined and this step does nothing.
  65 + @li Shuts down and destroys the pool's services, and abandons
  66 + every coroutine still queued on the pool or its strands.
  67 + Each abandoned chain is unwound, without delivering a
  68 + result, on the executor of the coroutine that started it.
  69 + That happens during this destructor when the executor is
  70 + this pool, and later when it belongs to another execution
  71 + context.
  72 + See @ref chain_owner.
  73 +
  74 + To let queued work finish first, call @ref join before the
  75 + pool is destroyed.
59   76  
60   @pre No thread outside this pool may post or dispatch work to it 77   @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, 78   (or to a strand built on it) concurrently with, or after,
62   destruction. Doing so is undefined behavior. Submit such work 79   destruction. Doing so is undefined behavior. Submit such work
63 - through @ref run_async or @ref run and call @ref join before 80 + through @ref run_async and call @ref join before the pool is
64 - the pool is destroyed, so it has completed first. 81 + destroyed, so it has completed first.
  82 +
  83 + @pre Must not be called from a thread in this pool. A worker
  84 + thread cannot join itself, so this calls `std::terminate`.
65   */ 85   */
66   ~thread_pool(); 86   ~thread_pool();
67   87  
68   /** Construct a thread pool. 88   /** Construct a thread pool.
69   89  
70   Records the requested worker count; no threads are created 90   Records the requested worker count; no threads are created
71   yet. Threads start lazily on the executor's first `post()`. 91   yet. Threads start lazily on the executor's first `post()`.
72   If `num_threads` is zero, the number of threads is set to 92   If `num_threads` is zero, the number of threads is set to
73   the hardware concurrency, or one if that cannot be determined. 93   the hardware concurrency, or one if that cannot be determined.
74   94  
75   @param num_threads The number of worker threads, or zero 95   @param num_threads The number of worker threads, or zero
76   for automatic selection. 96   for automatic selection.
77   97  
78   @param thread_name_prefix The prefix for worker thread names. 98   @param thread_name_prefix The prefix for worker thread names.
79   Thread names appear as "{prefix}0", "{prefix}1", etc. 99   Thread names appear as "{prefix}0", "{prefix}1", etc.
80   The prefix is truncated to 12 characters. Defaults to 100   The prefix is truncated to 12 characters. Defaults to
81   "capy-pool-". 101   "capy-pool-".
82   */ 102   */
83   explicit 103   explicit
84   thread_pool( 104   thread_pool(
85   std::size_t num_threads = 0, 105   std::size_t num_threads = 0,
86   std::string_view thread_name_prefix = "capy-pool-"); 106   std::string_view thread_name_prefix = "capy-pool-");
87   107  
88   /** Copy construction is disabled; a pool owns its worker threads. 108   /** Copy construction is disabled; a pool owns its worker threads.
89   109  
90   @param other The pool that would be copied. 110   @param other The pool that would be copied.
91   */ 111   */
92   thread_pool(thread_pool const& other) = delete; 112   thread_pool(thread_pool const& other) = delete;
93   113  
94   /** Copy assignment is disabled; a pool owns its worker threads. 114   /** Copy assignment is disabled; a pool owns its worker threads.
95   115  
96   @param other The pool that would be assigned from. 116   @param other The pool that would be assigned from.
97   117  
98   @return A reference to `*this`. 118   @return A reference to `*this`.
99   */ 119   */
100   thread_pool& operator=(thread_pool const& other) = delete; 120   thread_pool& operator=(thread_pool const& other) = delete;
101   121  
102   /** Wait for all outstanding work to complete. 122   /** Wait for all outstanding work to complete.
103   123  
104 - Releases the internal work guard, then blocks the calling 124 + Blocks the calling thread until all outstanding work
105 - thread until all outstanding work tracked by 125 + tracked by
106   @ref executor_type::on_work_started and 126   @ref executor_type::on_work_started and
107   @ref executor_type::on_work_finished completes. After all 127   @ref executor_type::on_work_finished completes. After all
108   work finishes, joins the worker threads. 128   work finishes, joins the worker threads.
109   129  
110   If @ref stop is called while `join()` is blocking, the 130   If @ref stop is called while `join()` is blocking, the
111   pool stops without waiting for remaining work to 131   pool stops without waiting for remaining work to
112   complete. Worker threads finish their current item and 132   complete. Worker threads finish their current item and
113   exit; `join()` still waits for all threads to be joined 133   exit; `join()` still waits for all threads to be joined
114   before returning. 134   before returning.
115   135  
116 - This function is idempotent. The first call performs the 136 + Only the first call performs the join. A later call returns
117 - join; subsequent calls return immediately. 137 + immediately, even while the first call is still blocking on
  138 + another thread.
118   139  
119   @pre Must not be called from a thread in this pool (undefined 140   @pre Must not be called from a thread in this pool (undefined
120   behavior). 141   behavior).
121   142  
122   @par Postconditions 143   @par Postconditions
123 - All worker threads have been joined. The pool cannot be 144 + If this call performed the join, all worker threads have
124 - reused. 145 + been joined. The pool cannot be reused.
125   146  
126   @par Thread Safety 147   @par Thread Safety
127   May be called from any thread not in this pool. 148   May be called from any thread not in this pool.
128   */ 149   */
129   void 150   void
130   join() noexcept; 151   join() noexcept;
131   152  
132   /** Request all worker threads to stop. 153   /** Request all worker threads to stop.
133   154  
134   Signals all threads to exit after finishing their current 155   Signals all threads to exit after finishing their current
135   work item. Queued work that has not started stays queued 156   work item. Queued work that has not started stays queued
136   until the pool is destroyed, which abandons it. Does not 157   until the pool is destroyed, which abandons it. Does not
137   wait for threads to exit. 158   wait for threads to exit.
138   159  
139   If @ref join is blocking on another thread, calling 160   If @ref join is blocking on another thread, calling
140   `stop()` causes it to stop waiting for outstanding 161   `stop()` causes it to stop waiting for outstanding
141   work. The `join()` call still waits for worker threads 162   work. The `join()` call still waits for worker threads
142   to finish their current item and exit before returning. 163   to finish their current item and exit before returning.
143   164  
144   @par Thread Safety 165   @par Thread Safety
145   May be called concurrently from any thread, including a 166   May be called concurrently from any thread, including a
146   thread in this pool. 167   thread in this pool.
147   */ 168   */
148   void 169   void
149   stop() noexcept; 170   stop() noexcept;
150   171  
151   /** Return an executor for this thread pool. 172   /** Return an executor for this thread pool.
152   173  
153   @return An executor associated with this thread pool. 174   @return An executor associated with this thread pool.
154   */ 175   */
155   executor_type 176   executor_type
156   get_executor() const noexcept; 177   get_executor() const noexcept;
157   }; 178   };
158   179  
159 - /** An executor that submits work to a thread_pool. 180 + /** Runs continuations on the worker threads of a thread_pool.
160   181  
161   Executors are lightweight handles that can be copied and stored. 182   Executors are lightweight handles that can be copied and stored.
162   All copies refer to the same underlying thread pool. 183   All copies refer to the same underlying thread pool.
163   184  
164   @par Thread Safety 185   @par Thread Safety
165   Distinct objects: Safe. 186   Distinct objects: Safe.
166   Shared objects: Safe. 187   Shared objects: Safe.
167   */ 188   */
168   class thread_pool::executor_type 189   class thread_pool::executor_type
169   { 190   {
170   friend class thread_pool; 191   friend class thread_pool;
171   192  
172   thread_pool* pool_ = nullptr; 193   thread_pool* pool_ = nullptr;
173   194  
174   explicit 195   explicit
HITCBC 175   14844 executor_type(thread_pool& pool) noexcept 196   14850 executor_type(thread_pool& pool) noexcept
HITCBC 176   14844 : pool_(&pool) 197   14850 : pool_(&pool)
177   { 198   {
HITCBC 178   14844 } 199   14850 }
179   200  
180   public: 201   public:
181   /** Construct a default null executor. 202   /** Construct a default null executor.
182   203  
183   The resulting executor is not associated with any pool. 204   The resulting executor is not associated with any pool.
184   `context()`, `dispatch()`, and `post()` require the 205   `context()`, `dispatch()`, and `post()` require the
185   executor to be associated with a pool before use. 206   executor to be associated with a pool before use.
186   */ 207   */
187   executor_type() = default; 208   executor_type() = default;
188   209  
189   /** Return the underlying thread pool. 210   /** Return the underlying thread pool.
190   211  
191   @return A reference to the associated pool. The behavior is 212   @return A reference to the associated pool. The behavior is
192   undefined if the executor is not associated with a pool. 213   undefined if the executor is not associated with a pool.
193   */ 214   */
194   thread_pool& 215   thread_pool&
HITCBC 195   12134 context() const noexcept 216   12140 context() const noexcept
196   { 217   {
HITCBC 197   12134 return *pool_; 218   12140 return *pool_;
198   } 219   }
199   220  
200   /** Notify that work has started. 221   /** Notify that work has started.
201   222  
202   Increments the outstanding work count. Must be paired 223   Increments the outstanding work count. Must be paired
203   with a subsequent call to @ref on_work_finished. 224   with a subsequent call to @ref on_work_finished.
204   225  
205   @see on_work_finished, work_guard 226   @see on_work_finished, work_guard
206   */ 227   */
207   BOOST_CAPY_DECL 228   BOOST_CAPY_DECL
208   void 229   void
209   on_work_started() const noexcept; 230   on_work_started() const noexcept;
210   231  
211   /** Notify that work has finished. 232   /** Notify that work has finished.
212   233  
213   Decrements the outstanding work count. When the count 234   Decrements the outstanding work count. When the count
214   reaches zero after @ref thread_pool::join is called, 235   reaches zero after @ref thread_pool::join is called,
215   the pool's worker threads are signaled to stop. 236   the pool's worker threads are signaled to stop.
216   237  
217   @pre A preceding call to @ref on_work_started was made. 238   @pre A preceding call to @ref on_work_started was made.
218   239  
219   @see on_work_started, work_guard 240   @see on_work_started, work_guard
220   */ 241   */
221   BOOST_CAPY_DECL 242   BOOST_CAPY_DECL
222   void 243   void
223   on_work_finished() const noexcept; 244   on_work_finished() const noexcept;
224   245  
225   /** Dispatch a continuation for execution. 246   /** Dispatch a continuation for execution.
226   247  
227   If the calling thread is a worker of this pool, returns 248   If the calling thread is a worker of this pool, returns
228   `c.h` for symmetric transfer so the caller can resume the 249   `c.h` for symmetric transfer so the caller can resume the
229 - continuation inline. Otherwise, posts the continuation to 250 + continuation inline. Otherwise, posts the continuation as
230 - the pool for execution on a worker thread and returns 251 + @ref post does and returns `std::noop_coroutine()`. On that
231 - `std::noop_coroutine()`. 252 + path, the continuation is not run if the pool stops first,
  253 + and a call that races with @ref thread_pool::stop can go
  254 + either way. See @ref post.
232   255  
233   @param c The continuation to execute. On the post path, 256   @param c The continuation to execute. On the post path,
234 - must remain at a stable address until dequeued 257 + must remain at a stable address until resumed or
235 - and resumed. 258 + abandoned.
236   259  
237   @return `c.h` when the calling thread is a pool worker; 260   @return `c.h` when the calling thread is a pool worker;
238   `std::noop_coroutine()` otherwise. 261   `std::noop_coroutine()` otherwise.
  262 +
  263 + @throws std::system_error If this call posts and must start
  264 + the pool's worker threads, and a thread cannot be started.
  265 + `c` is not queued.
239   */ 266   */
240   BOOST_CAPY_DECL 267   BOOST_CAPY_DECL
241   std::coroutine_handle<> 268   std::coroutine_handle<>
242   dispatch(continuation& c) const; 269   dispatch(continuation& c) const;
243   270  
244   /** Post a continuation to the thread pool. 271   /** Post a continuation to the thread pool.
245   272  
246 - The continuation is resumed on one of the pool's 273 + The continuation is resumed on one of the pool's worker
247 - worker threads. The continuation must remain at a stable 274 + threads, unless the pool stops first. The pool stops when
248 - address until it is dequeued and resumed. 275 + @ref thread_pool::stop is called, or when
  276 + @ref thread_pool::join has finished waiting for outstanding
  277 + work. A continuation still queued when the pool stops, or
  278 + posted after it stops, is not run. It stays queued until
  279 + the pool is destroyed, which abandons it.
  280 +
  281 + A call that races with @ref thread_pool::stop on another
  282 + thread can therefore go either way, and nothing in the call
  283 + tells the caller which.
  284 +
  285 + The continuation must remain at a stable address until it
  286 + is resumed or abandoned.
249   287  
250   @param c The continuation to execute. 288   @param c The continuation to execute.
  289 +
  290 + @throws std::system_error If this call must start the pool's
  291 + worker threads, and a thread cannot be started. `c` is not
  292 + queued.
251   */ 293   */
252   BOOST_CAPY_DECL 294   BOOST_CAPY_DECL
253   void 295   void
254   post(continuation& c) const; 296   post(continuation& c) const;
255   297  
256   /** Return true if two executors refer to the same thread pool. 298   /** Return true if two executors refer to the same thread pool.
257   299  
258   @param other The executor to compare against. 300   @param other The executor to compare against.
259   301  
260   @return `true` if both executors refer to the same pool. 302   @return `true` if both executors refer to the same pool.
261   */ 303   */
262   bool 304   bool
HITCBC 263   15 operator==(executor_type const& other) const noexcept 305   15 operator==(executor_type const& other) const noexcept
264   { 306   {
HITCBC 265   15 return pool_ == other.pool_; 307   15 return pool_ == other.pool_;
266   } 308   }
267   }; 309   };
268   310  
269   } // capy 311   } // capy
270   } // boost 312   } // boost
271   313  
272   #endif 314   #endif