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