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