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