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