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