. /src /ex /thread_pool.cpp
100.00% Lines (140/140) 100.00% Functions (29/29)
[ ] 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/boostorg/capy
9   // 9   //
10   10  
11   #include <boost/capy/ex/thread_pool.hpp> 11   #include <boost/capy/ex/thread_pool.hpp>
12   #include <boost/capy/continuation.hpp> 12   #include <boost/capy/continuation.hpp>
13   #include <boost/capy/detail/thread_local_ptr.hpp> 13   #include <boost/capy/detail/thread_local_ptr.hpp>
14   #include <boost/capy/ex/frame_allocator.hpp> 14   #include <boost/capy/ex/frame_allocator.hpp>
15   #include <boost/capy/test/thread_name.hpp> 15   #include <boost/capy/test/thread_name.hpp>
16   #include <algorithm> 16   #include <algorithm>
17   #include <atomic> 17   #include <atomic>
18   #include <condition_variable> 18   #include <condition_variable>
19   #include <cstdio> 19   #include <cstdio>
20   #include <mutex> 20   #include <mutex>
21   #include <thread> 21   #include <thread>
22   #include <vector> 22   #include <vector>
23   23  
24   /* 24   /*
25   Thread pool implementation using a shared work queue. 25   Thread pool implementation using a shared work queue.
26   26  
27   Work items are continuations linked via their intrusive next pointer, 27   Work items are continuations linked via their intrusive next pointer,
28   stored in a single queue protected by a mutex. No per-post heap 28   stored in a single queue protected by a mutex. No per-post heap
29   allocation: the continuation is owned by the caller and linked 29   allocation: the continuation is owned by the caller and linked
30   directly. Worker threads wait on a condition_variable until work 30   directly. Worker threads wait on a condition_variable until work
31   is available or stop is requested. 31   is available or stop is requested.
32   32  
33   Threads are started lazily on first post() via std::call_once to avoid 33   Threads are started lazily on first post() via std::call_once to avoid
34   spawning threads for pools that are constructed but never used. Each 34   spawning threads for pools that are constructed but never used. Each
35   thread is named with a configurable prefix plus index for debugger 35   thread is named with a configurable prefix plus index for debugger
36   visibility. 36   visibility.
37   37  
38   Work tracking: on_work_started/on_work_finished maintain the atomic 38   Work tracking: on_work_started/on_work_finished maintain the atomic
39   outstanding_work_ counter. on_work_started is lock-free; the worker 39   outstanding_work_ counter. on_work_started is lock-free; the worker
40   that drives the count to zero takes mutex_ and re-reads the count 40   that drives the count to zero takes mutex_ and re-reads the count
41   before deciding to stop, so the count and the stop decision stay 41   before deciding to stop, so the count and the stop decision stay
42   consistent even if work is started in between. join() blocks until 42   consistent even if work is started in between. join() blocks until
43   this counter reaches zero, then signals workers to stop and joins 43   this counter reaches zero, then signals workers to stop and joins
44   threads. 44   threads.
45   45  
46   Two shutdown paths: 46   Two shutdown paths:
47   - join(): waits for outstanding work to drain, then stops workers. 47   - join(): waits for outstanding work to drain, then stops workers.
48   - stop(): immediately signals workers to exit; queued work is abandoned. 48   - stop(): immediately signals workers to exit; queued work is abandoned.
49   - Destructor: stop() then join() (abandon + wait for threads). 49   - Destructor: stop() then join() (abandon + wait for threads).
50   */ 50   */
51   51  
52   namespace boost { 52   namespace boost {
53   namespace capy { 53   namespace capy {
54   54  
55   //------------------------------------------------------------------------------ 55   //------------------------------------------------------------------------------
56   56  
57   class thread_pool::impl 57   class thread_pool::impl
58   { 58   {
59   // Identifies the pool owning the current worker thread, or 59   // Identifies the pool owning the current worker thread, or
60   // nullptr if the calling thread is not a pool worker. Checked 60   // nullptr if the calling thread is not a pool worker. Checked
61   // by dispatch() to decide between symmetric transfer (inline 61   // by dispatch() to decide between symmetric transfer (inline
62   // resume) and post. 62   // resume) and post.
63   static inline detail::thread_local_ptr<impl const> current_; 63   static inline detail::thread_local_ptr<impl const> current_;
64   64  
65   // Intrusive queue of continuations: the next link is stored in 65   // Intrusive queue of continuations: the next link is stored in
66   // continuation::reserved (typed continuation* round-tripped through 66   // continuation::reserved (typed continuation* round-tripped through
67   // void*). No per-post allocation: the continuation is owned by the caller. 67   // void*). No per-post allocation: the continuation is owned by the caller.
68   continuation* head_ = nullptr; 68   continuation* head_ = nullptr;
69   continuation* tail_ = nullptr; 69   continuation* tail_ = nullptr;
70   70  
HITCBC 71   20457 void push(continuation* c) noexcept 71   18077 void push(continuation* c) noexcept
72   { 72   {
HITCBC 73   20457 c->reserved = nullptr; 73   18077 c->reserved = nullptr;
HITCBC 74   20457 if(tail_) 74   18077 if(tail_)
HITCBC 75   1801 tail_->reserved = c; 75   5204 tail_->reserved = c;
76   else 76   else
HITCBC 77   18656 head_ = c; 77   12873 head_ = c;
HITCBC 78   20457 tail_ = c; 78   18077 tail_ = c;
HITCBC 79   20457 } 79   18077 }
80   80  
HITCBC 81   20730 continuation* pop() noexcept 81   18412 continuation* pop() noexcept
82   { 82   {
HITCBC 83   20730 if(!head_) 83   18412 if(!head_)
HITCBC 84   273 return nullptr; 84   335 return nullptr;
HITCBC 85   20457 continuation* c = head_; 85   18077 continuation* c = head_;
HITCBC 86   20457 head_ = static_cast<continuation*>(head_->reserved); 86   18077 head_ = static_cast<continuation*>(head_->reserved);
HITCBC 87   20457 if(!head_) 87   18077 if(!head_)
HITCBC 88   18656 tail_ = nullptr; 88   12873 tail_ = nullptr;
HITCBC 89   20457 return c; 89   18077 return c;
90   } 90   }
91   91  
HITCBC 92   40183 bool empty() const noexcept 92   31165 bool empty() const noexcept
93   { 93   {
HITCBC 94   40183 return head_ == nullptr; 94   31165 return head_ == nullptr;
95   } 95   }
96   96  
97   std::mutex mutex_; 97   std::mutex mutex_;
98   std::condition_variable work_cv_; 98   std::condition_variable work_cv_;
99   std::condition_variable done_cv_; 99   std::condition_variable done_cv_;
100   std::vector<std::thread> threads_; 100   std::vector<std::thread> threads_;
101   std::atomic<std::size_t> outstanding_work_{0}; 101   std::atomic<std::size_t> outstanding_work_{0};
102   bool stop_{false}; 102   bool stop_{false};
103   bool joined_{false}; 103   bool joined_{false};
104   std::size_t num_threads_; 104   std::size_t num_threads_;
105   char thread_name_prefix_[13]{}; // 12 chars max + null terminator 105   char thread_name_prefix_[13]{}; // 12 chars max + null terminator
106   std::once_flag start_flag_; 106   std::once_flag start_flag_;
107   107  
108   public: 108   public:
HITCBC 109   273 ~impl() = default; 109   335 ~impl() = default;
110   110  
111   bool 111   bool
HITCBC 112   456 running_in_this_thread() const noexcept 112   517 running_in_this_thread() const noexcept
113   { 113   {
HITCBC 114   456 return current_.get() == this; 114   517 return current_.get() == this;
115   } 115   }
116   116  
117   // Destroy abandoned coroutine frames. Must be called 117   // Destroy abandoned coroutine frames. Must be called
118   // before execution_context::shutdown()/destroy() so 118   // before execution_context::shutdown()/destroy() so
119   // that suspended-frame destructors touching services 119   // that suspended-frame destructors touching services
120   // (e.g. cancelling registrations) run while those 120   // (e.g. cancelling registrations) run while those
121   // services are still valid. 121   // services are still valid.
122   void 122   void
HITCBC 123   273 drain_abandoned() noexcept 123   335 drain_abandoned() noexcept
124   { 124   {
HITCBC 125   475 while(auto* c = pop()) 125   500 while(auto* c = pop())
126   { 126   {
HITCBC 127   202 auto h = c->h; 127   165 auto h = c->h;
HITCBC 128   202 if(h && h != std::noop_coroutine()) 128   165 if(h && h != std::noop_coroutine())
HITCBC 129   151 h.destroy(); 129   114 h.destroy();
HITCBC 130   202 } 130   165 }
HITCBC 131   273 } 131   335 }
132   132  
HITCBC 133   273 impl(std::size_t num_threads, std::string_view thread_name_prefix) 133   335 impl(std::size_t num_threads, std::string_view thread_name_prefix)
HITCBC 134   273 : num_threads_(num_threads) 134   335 : num_threads_(num_threads)
135   { 135   {
HITCBC 136   273 if(num_threads_ == 0) 136   335 if(num_threads_ == 0)
HITCBC 137   4 num_threads_ = std::max( 137   4 num_threads_ = std::max(
HITCBC 138   2 std::thread::hardware_concurrency(), 1u); 138   2 std::thread::hardware_concurrency(), 1u);
139   139  
140   // Truncate prefix to 12 chars, leaving room for up to 3-digit index. 140   // Truncate prefix to 12 chars, leaving room for up to 3-digit index.
HITCBC 141   273 auto n = thread_name_prefix.copy(thread_name_prefix_, 12); 141   335 auto n = thread_name_prefix.copy(thread_name_prefix_, 12);
HITCBC 142   273 thread_name_prefix_[n] = '\0'; 142   335 thread_name_prefix_[n] = '\0';
HITCBC 143   273 } 143   335 }
144   144  
145   void 145   void
HITCBC 146   20457 post(continuation& c) 146   18077 post(continuation& c)
147   { 147   {
HITCBC 148   20457 ensure_started(); 148   18077 ensure_started();
149   { 149   {
HITCBC 150   20457 std::lock_guard<std::mutex> lock(mutex_); 150   18077 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 151   20457 push(&c); 151   18077 push(&c);
  152 + // Under the lock so the pool cannot drain, join, and
  153 + // destroy the condition variable mid-signal.
HITGNC   154 + 18077 work_cv_.notify_one();
DCB 152 - 20457 work_cv_.notify_one();  
HITCBC 153   20457 } 155   18077 }
HITCBC 154   20457 } 156   18077 }
155   157  
156   void 158   void
HITCBC 157   456 on_work_started() noexcept 159   517 on_work_started() noexcept
158   { 160   {
HITCBC 159   456 outstanding_work_.fetch_add(1, std::memory_order_acq_rel); 161   517 outstanding_work_.fetch_add(1, std::memory_order_acq_rel);
HITCBC 160   456 } 162   517 }
161   163  
162   void 164   void
HITCBC 163   456 on_work_finished() noexcept 165   517 on_work_finished() noexcept
164   { 166   {
HITCBC 165   456 if(outstanding_work_.fetch_sub( 167   517 if(outstanding_work_.fetch_sub(
HITCBC 166   456 1, std::memory_order_acq_rel) == 1) 168   517 1, std::memory_order_acq_rel) == 1)
167   { 169   {
168   // fetch_sub's result can be stale: a concurrent 170   // fetch_sub's result can be stale: a concurrent
169   // on_work_started() may raise the count before we take the 171   // on_work_started() may raise the count before we take the
170   // lock, so re-read it here rather than trust the decrement. 172   // lock, so re-read it here rather than trust the decrement.
HITCBC 171   205 std::lock_guard<std::mutex> lock(mutex_); 173   265 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 172   205 if(outstanding_work_.load( 174   265 if(outstanding_work_.load(
HITCBC 173   205 std::memory_order_acquire) == 0 && joined_ && !stop_) 175   265 std::memory_order_acquire) == 0 && joined_ && !stop_)
174   { 176   {
HITCBC 175   82 stop_ = true; 177   137 stop_ = true;
HITCBC 176   82 done_cv_.notify_all(); 178   137 done_cv_.notify_all();
HITCBC 177   82 work_cv_.notify_all(); 179   137 work_cv_.notify_all();
178   } 180   }
HITCBC 179   205 } 181   265 }
HITCBC 180   456 } 182   517 }
181   183  
182   void 184   void
HITCBC 183   416 join() noexcept 185   540 join() noexcept
184   { 186   {
185   { 187   {
HITCBC 186   416 std::unique_lock<std::mutex> lock(mutex_); 188   540 std::unique_lock<std::mutex> lock(mutex_);
HITCBC 187   416 if(joined_) 189   540 if(joined_)
HITCBC 188   143 return; 190   205 return;
HITCBC 189   273 joined_ = true; 191   335 joined_ = true;
190   192  
HITCBC 191   273 if(outstanding_work_.load( 193   335 if(outstanding_work_.load(
HITCBC 192   273 std::memory_order_acquire) == 0) 194   335 std::memory_order_acquire) == 0)
193   { 195   {
HITCBC 194   136 stop_ = true; 196   147 stop_ = true;
HITCBC 195   136 work_cv_.notify_all(); 197   147 work_cv_.notify_all();
196   } 198   }
197   else 199   else
198   { 200   {
HITCBC 199   137 done_cv_.wait(lock, [this]{ 201   188 done_cv_.wait(lock, [this]{
HITCBC 200   220 return stop_; 202   326 return stop_;
201   }); 203   });
202   } 204   }
HITCBC 203   416 } 205   540 }
204   206  
HITCBC 205   580 for(auto& t : threads_) 207   704 for(auto& t : threads_)
HITCBC 206   307 if(t.joinable()) 208   369 if(t.joinable())
HITCBC 207   307 t.join(); 209   369 t.join();
208   } 210   }
209   211  
210   void 212   void
HITCBC 211   275 stop() noexcept 213   337 stop() noexcept
212   { 214   {
213   { 215   {
HITCBC 214   275 std::lock_guard<std::mutex> lock(mutex_); 216   337 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 215   275 stop_ = true; 217   337 stop_ = true;
HITCBC 216   275 } 218   337 }
HITCBC 217   275 work_cv_.notify_all(); 219   337 work_cv_.notify_all();
HITCBC 218   275 done_cv_.notify_all(); 220   337 done_cv_.notify_all();
HITCBC 219   275 } 221   337 }
220   222  
221   private: 223   private:
222   void 224   void
HITCBC 223   20457 ensure_started() 225   18077 ensure_started()
224   { 226   {
HITCBC 225   20457 std::call_once(start_flag_, [this]{ 227   18077 std::call_once(start_flag_, [this]{
HITCBC 226   226 threads_.reserve(num_threads_); 228   288 threads_.reserve(num_threads_);
HITCBC 227   533 for(std::size_t i = 0; i < num_threads_; ++i) 229   657 for(std::size_t i = 0; i < num_threads_; ++i)
HITCBC 228   614 threads_.emplace_back([this, i]{ run(i); }); 230   738 threads_.emplace_back([this, i]{ run(i); });
HITCBC 229   226 }); 231   288 });
HITCBC 230   20457 } 232   18077 }
231   233  
232   void 234   void
HITCBC 233   307 run(std::size_t index) 235   369 run(std::size_t index)
234   { 236   {
235   // Build name; set_current_thread_name truncates to platform limits. 237   // Build name; set_current_thread_name truncates to platform limits.
236   char name[16]; 238   char name[16];
HITCBC 237   307 std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index); 239   369 std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index);
HITCBC 238   307 set_current_thread_name(name); 240   369 set_current_thread_name(name);
239   241  
240   // Mark this thread as a worker of this pool so dispatch() 242   // Mark this thread as a worker of this pool so dispatch()
241   // can symmetric-transfer when called from within pool work. 243   // can symmetric-transfer when called from within pool work.
242   struct scoped_pool 244   struct scoped_pool
243   { 245   {
HITCBC 244   307 scoped_pool(impl const* p) noexcept { current_.set(p); } 246   369 scoped_pool(impl const* p) noexcept { current_.set(p); }
HITCBC 245   307 ~scoped_pool() noexcept { current_.set(nullptr); } 247   369 ~scoped_pool() noexcept { current_.set(nullptr); }
HITCBC 246   307 } guard(this); 248   369 } guard(this);
247   249  
248   for(;;) 250   for(;;)
249   { 251   {
HITCBC 250   20562 continuation* c = nullptr; 252   18281 continuation* c = nullptr;
251   { 253   {
HITCBC 252   20562 std::unique_lock<std::mutex> lock(mutex_); 254   18281 std::unique_lock<std::mutex> lock(mutex_);
HITCBC 253   20562 work_cv_.wait(lock, [this]{ 255   18281 work_cv_.wait(lock, [this]{
HITCBC 254   60010 return !empty() || 256   44333 return !empty() ||
HITCBC 255   60010 stop_; 257   44333 stop_;
256   }); 258   });
HITCBC 257   20562 if(stop_) 259   18281 if(stop_)
HITCBC 258   614 return; 260   738 return;
HITCBC 259   20255 c = pop(); 261   17912 c = pop();
HITCBC 260   20562 } 262   18281 }
HITCBC 261   20255 if(c) 263   17912 if(c)
HITCBC 262   20255 safe_resume(c->h); 264   17912 safe_resume(c->h);
HITCBC 263   20255 } 265   17912 }
HITCBC 264   307 } 266   369 }
265   }; 267   };
266   268  
267   //------------------------------------------------------------------------------ 269   //------------------------------------------------------------------------------
268   270  
HITCBC 269   273 thread_pool:: 271   335 thread_pool::
270   ~thread_pool() 272   ~thread_pool()
271   { 273   {
HITCBC 272   273 impl_->stop(); 274   335 impl_->stop();
HITCBC 273   273 impl_->join(); 275   335 impl_->join();
HITCBC 274   273 impl_->drain_abandoned(); 276   335 impl_->drain_abandoned();
HITCBC 275   273 shutdown(); 277   335 shutdown();
HITCBC 276   273 destroy(); 278   335 destroy();
HITCBC 277   273 delete impl_; 279   335 delete impl_;
HITCBC 278   273 } 280   335 }
279   281  
HITCBC 280   273 thread_pool:: 282   335 thread_pool::
HITCBC 281   273 thread_pool(std::size_t num_threads, std::string_view thread_name_prefix) 283   335 thread_pool(std::size_t num_threads, std::string_view thread_name_prefix)
HITCBC 282   273 : impl_(new impl(num_threads, thread_name_prefix)) 284   335 : impl_(new impl(num_threads, thread_name_prefix))
283   { 285   {
HITCBC 284   273 this->set_frame_allocator(std::allocator<void>{}); 286   335 this->set_frame_allocator(std::allocator<void>{});
HITCBC 285   273 } 287   335 }
286   288  
287   void 289   void
HITCBC 288   143 thread_pool:: 290   205 thread_pool::
289   join() noexcept 291   join() noexcept
290   { 292   {
HITCBC 291   143 impl_->join(); 293   205 impl_->join();
HITCBC 292   143 } 294   205 }
293   295  
294   void 296   void
HITCBC 295   2 thread_pool:: 297   2 thread_pool::
296   stop() noexcept 298   stop() noexcept
297   { 299   {
HITCBC 298   2 impl_->stop(); 300   2 impl_->stop();
HITCBC 299   2 } 301   2 }
300   302  
301   //------------------------------------------------------------------------------ 303   //------------------------------------------------------------------------------
302   304  
303   thread_pool::executor_type 305   thread_pool::executor_type
HITCBC 304   11679 thread_pool:: 306   11741 thread_pool::
305   get_executor() const noexcept 307   get_executor() const noexcept
306   { 308   {
HITCBC 307   11679 return executor_type( 309   11741 return executor_type(
HITCBC 308   11679 const_cast<thread_pool&>(*this)); 310   11741 const_cast<thread_pool&>(*this));
309   } 311   }
310   312  
311   void 313   void
HITCBC 312   456 thread_pool::executor_type:: 314   517 thread_pool::executor_type::
313   on_work_started() const noexcept 315   on_work_started() const noexcept
314   { 316   {
HITCBC 315   456 pool_->impl_->on_work_started(); 317   517 pool_->impl_->on_work_started();
HITCBC 316   456 } 318   517 }
317   319  
318   void 320   void
HITCBC 319   456 thread_pool::executor_type:: 321   517 thread_pool::executor_type::
320   on_work_finished() const noexcept 322   on_work_finished() const noexcept
321   { 323   {
HITCBC 322   456 pool_->impl_->on_work_finished(); 324   517 pool_->impl_->on_work_finished();
HITCBC 323   456 } 325   517 }
324   326  
325   void 327   void
HITCBC 326   20006 thread_pool::executor_type:: 328   17565 thread_pool::executor_type::
327   post(continuation& c) const 329   post(continuation& c) const
328   { 330   {
HITCBC 329   20006 pool_->impl_->post(c); 331   17565 pool_->impl_->post(c);
HITCBC 330   20006 } 332   17565 }
331   333  
332   std::coroutine_handle<> 334   std::coroutine_handle<>
HITCBC 333   456 thread_pool::executor_type:: 335   517 thread_pool::executor_type::
334   dispatch(continuation& c) const 336   dispatch(continuation& c) const
335   { 337   {
HITCBC 336   456 if(pool_->impl_->running_in_this_thread()) 338   517 if(pool_->impl_->running_in_this_thread())
HITCBC 337   5 return c.h; 339   5 return c.h;
HITCBC 338   451 pool_->impl_->post(c); 340   512 pool_->impl_->post(c);
HITCBC 339   451 return std::noop_coroutine(); 341   512 return std::noop_coroutine();
340   } 342   }
341   343  
342   } // capy 344   } // capy
343   } // boost 345   } // boost