. /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   20808 void push(continuation* c) noexcept 71   21048 void push(continuation* c) noexcept
72   { 72   {
HITCBC 73   20808 c->reserved = nullptr; 73   21048 c->reserved = nullptr;
HITCBC 74   20808 if(tail_) 74   21048 if(tail_)
HITCBC 75   2161 tail_->reserved = c; 75   2711 tail_->reserved = c;
76   else 76   else
HITCBC 77   18647 head_ = c; 77   18337 head_ = c;
HITCBC 78   20808 tail_ = c; 78   21048 tail_ = c;
HITCBC 79   20808 } 79   21048 }
80   80  
HITCBC 81   21092 continuation* pop() noexcept 81   21321 continuation* pop() noexcept
82   { 82   {
HITCBC 83   21092 if(!head_) 83   21321 if(!head_)
HITCBC 84   284 return nullptr; 84   273 return nullptr;
HITCBC 85   20808 continuation* c = head_; 85   21048 continuation* c = head_;
HITCBC 86   20808 head_ = static_cast<continuation*>(head_->reserved); 86   21048 head_ = static_cast<continuation*>(head_->reserved);
HITCBC 87   20808 if(!head_) 87   21048 if(!head_)
HITCBC 88   18647 tail_ = nullptr; 88   18337 tail_ = nullptr;
HITCBC 89   20808 return c; 89   21048 return c;
90   } 90   }
91   91  
HITCBC 92   40840 bool empty() const noexcept 92   41318 bool empty() const noexcept
93   { 93   {
HITCBC 94   40840 return head_ == nullptr; 94   41318 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   284 ~impl() = default; 109   273 ~impl() = default;
110   110  
111   bool 111   bool
HITCBC 112   467 running_in_this_thread() const noexcept 112   456 running_in_this_thread() const noexcept
113   { 113   {
HITCBC 114   467 return current_.get() == this; 114   456 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   284 drain_abandoned() noexcept 123   273 drain_abandoned() noexcept
124   { 124   {
HITCBC 125   478 while(auto* c = pop()) 125   476 while(auto* c = pop())
126   { 126   {
HITCBC 127   194 auto h = c->h; 127   203 auto h = c->h;
HITCBC 128   194 if(h && h != std::noop_coroutine()) 128   203 if(h && h != std::noop_coroutine())
HITCBC 129   143 h.destroy(); 129   152 h.destroy();
HITCBC 130   194 } 130   203 }
HITCBC 131   284 } 131   273 }
132   132  
HITCBC 133   284 impl(std::size_t num_threads, std::string_view thread_name_prefix) 133   273 impl(std::size_t num_threads, std::string_view thread_name_prefix)
HITCBC 134   284 : num_threads_(num_threads) 134   273 : num_threads_(num_threads)
135   { 135   {
HITCBC 136   284 if(num_threads_ == 0) 136   273 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   284 auto n = thread_name_prefix.copy(thread_name_prefix_, 12); 141   273 auto n = thread_name_prefix.copy(thread_name_prefix_, 12);
HITCBC 142   284 thread_name_prefix_[n] = '\0'; 142   273 thread_name_prefix_[n] = '\0';
HITCBC 143   284 } 143   273 }
144   144  
145   void 145   void
HITCBC 146   20808 post(continuation& c) 146   21048 post(continuation& c)
147   { 147   {
HITCBC 148   20808 ensure_started(); 148   21048 ensure_started();
149   { 149   {
HITCBC 150   20808 std::lock_guard<std::mutex> lock(mutex_); 150   21048 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 151   20808 push(&c); 151   21048 push(&c);
HITCBC 152   20808 } 152   21048 }
HITCBC 153   20808 work_cv_.notify_one(); 153   21048 work_cv_.notify_one();
HITCBC 154   20808 } 154   21048 }
155   155  
156   void 156   void
HITCBC 157   467 on_work_started() noexcept 157   456 on_work_started() noexcept
158   { 158   {
HITCBC 159   467 outstanding_work_.fetch_add(1, std::memory_order_acq_rel); 159   456 outstanding_work_.fetch_add(1, std::memory_order_acq_rel);
HITCBC 160   467 } 160   456 }
161   161  
162   void 162   void
HITCBC 163   467 on_work_finished() noexcept 163   456 on_work_finished() noexcept
164   { 164   {
HITCBC 165   467 if(outstanding_work_.fetch_sub( 165   456 if(outstanding_work_.fetch_sub(
HITCBC 166   467 1, std::memory_order_acq_rel) == 1) 166   456 1, std::memory_order_acq_rel) == 1)
167   { 167   {
168   // fetch_sub's result can be stale: a concurrent 168   // fetch_sub's result can be stale: a concurrent
169   // on_work_started() may raise the count before we take the 169   // on_work_started() may raise the count before we take the
170   // lock, so re-read it here rather than trust the decrement. 170   // lock, so re-read it here rather than trust the decrement.
HITCBC 171   215 std::lock_guard<std::mutex> lock(mutex_); 171   204 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 172   215 if(outstanding_work_.load( 172   204 if(outstanding_work_.load(
HITCBC 173   215 std::memory_order_acquire) == 0 && joined_ && !stop_) 173   204 std::memory_order_acquire) == 0 && joined_ && !stop_)
174   { 174   {
HITCBC 175   70 stop_ = true; 175   73 stop_ = true;
HITCBC 176   70 done_cv_.notify_all(); 176   73 done_cv_.notify_all();
HITCBC 177   70 work_cv_.notify_all(); 177   73 work_cv_.notify_all();
178   } 178   }
HITCBC 179   215 } 179   204 }
HITCBC 180   467 } 180   456 }
181   181  
182   void 182   void
HITCBC 183   438 join() noexcept 183   416 join() noexcept
184   { 184   {
185   { 185   {
HITCBC 186   438 std::unique_lock<std::mutex> lock(mutex_); 186   416 std::unique_lock<std::mutex> lock(mutex_);
HITCBC 187   438 if(joined_) 187   416 if(joined_)
HITCBC 188   154 return; 188   143 return;
HITCBC 189   284 joined_ = true; 189   273 joined_ = true;
190   190  
HITCBC 191   284 if(outstanding_work_.load( 191   273 if(outstanding_work_.load(
HITCBC 192   284 std::memory_order_acquire) == 0) 192   273 std::memory_order_acquire) == 0)
193   { 193   {
HITCBC 194   162 stop_ = true; 194   145 stop_ = true;
HITCBC 195   162 work_cv_.notify_all(); 195   145 work_cv_.notify_all();
196   } 196   }
197   else 197   else
198   { 198   {
HITCBC 199   122 done_cv_.wait(lock, [this]{ 199   128 done_cv_.wait(lock, [this]{
HITCBC 200   193 return stop_; 200   202 return stop_;
201   }); 201   });
202   } 202   }
HITCBC 203   438 } 203   416 }
204   204  
HITCBC 205   602 for(auto& t : threads_) 205   580 for(auto& t : threads_)
HITCBC 206   318 if(t.joinable()) 206   307 if(t.joinable())
HITCBC 207   318 t.join(); 207   307 t.join();
208   } 208   }
209   209  
210   void 210   void
HITCBC 211   286 stop() noexcept 211   275 stop() noexcept
212   { 212   {
213   { 213   {
HITCBC 214   286 std::lock_guard<std::mutex> lock(mutex_); 214   275 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 215   286 stop_ = true; 215   275 stop_ = true;
HITCBC 216   286 } 216   275 }
HITCBC 217   286 work_cv_.notify_all(); 217   275 work_cv_.notify_all();
HITCBC 218   286 done_cv_.notify_all(); 218   275 done_cv_.notify_all();
HITCBC 219   286 } 219   275 }
220   220  
221   private: 221   private:
222   void 222   void
HITCBC 223   20808 ensure_started() 223   21048 ensure_started()
224   { 224   {
HITCBC 225   20808 std::call_once(start_flag_, [this]{ 225   21048 std::call_once(start_flag_, [this]{
HITCBC 226   237 threads_.reserve(num_threads_); 226   226 threads_.reserve(num_threads_);
HITCBC 227   555 for(std::size_t i = 0; i < num_threads_; ++i) 227   533 for(std::size_t i = 0; i < num_threads_; ++i)
HITCBC 228   636 threads_.emplace_back([this, i]{ run(i); }); 228   614 threads_.emplace_back([this, i]{ run(i); });
HITCBC 229   237 }); 229   226 });
HITCBC 230   20808 } 230   21048 }
231   231  
232   void 232   void
HITCBC 233   318 run(std::size_t index) 233   307 run(std::size_t index)
234   { 234   {
235   // Build name; set_current_thread_name truncates to platform limits. 235   // Build name; set_current_thread_name truncates to platform limits.
236   char name[16]; 236   char name[16];
HITCBC 237   318 std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index); 237   307 std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index);
HITCBC 238   318 set_current_thread_name(name); 238   307 set_current_thread_name(name);
239   239  
240   // Mark this thread as a worker of this pool so dispatch() 240   // Mark this thread as a worker of this pool so dispatch()
241   // can symmetric-transfer when called from within pool work. 241   // can symmetric-transfer when called from within pool work.
242   struct scoped_pool 242   struct scoped_pool
243   { 243   {
HITCBC 244   318 scoped_pool(impl const* p) noexcept { current_.set(p); } 244   307 scoped_pool(impl const* p) noexcept { current_.set(p); }
HITCBC 245   318 ~scoped_pool() noexcept { current_.set(nullptr); } 245   307 ~scoped_pool() noexcept { current_.set(nullptr); }
HITCBC 246   318 } guard(this); 246   307 } guard(this);
247   247  
248   for(;;) 248   for(;;)
249   { 249   {
HITCBC 250   20932 continuation* c = nullptr; 250   21152 continuation* c = nullptr;
251   { 251   {
HITCBC 252   20932 std::unique_lock<std::mutex> lock(mutex_); 252   21152 std::unique_lock<std::mutex> lock(mutex_);
HITCBC 253   20932 work_cv_.wait(lock, [this]{ 253   21152 work_cv_.wait(lock, [this]{
HITCBC 254   60971 return !empty() || 254   61692 return !empty() ||
HITCBC 255   60971 stop_; 255   61692 stop_;
256   }); 256   });
HITCBC 257   20932 if(stop_) 257   21152 if(stop_)
HITCBC 258   636 return; 258   614 return;
HITCBC 259   20614 c = pop(); 259   20845 c = pop();
HITCBC 260   20932 } 260   21152 }
HITCBC 261   20614 if(c) 261   20845 if(c)
HITCBC 262   20614 safe_resume(c->h); 262   20845 safe_resume(c->h);
HITCBC 263   20614 } 263   20845 }
HITCBC 264   318 } 264   307 }
265   }; 265   };
266   266  
267   //------------------------------------------------------------------------------ 267   //------------------------------------------------------------------------------
268   268  
HITCBC 269   284 thread_pool:: 269   273 thread_pool::
270   ~thread_pool() 270   ~thread_pool()
271   { 271   {
HITCBC 272   284 impl_->stop(); 272   273 impl_->stop();
HITCBC 273   284 impl_->join(); 273   273 impl_->join();
HITCBC 274   284 impl_->drain_abandoned(); 274   273 impl_->drain_abandoned();
HITCBC 275   284 shutdown(); 275   273 shutdown();
HITCBC 276   284 destroy(); 276   273 destroy();
HITCBC 277   284 delete impl_; 277   273 delete impl_;
HITCBC 278   284 } 278   273 }
279   279  
HITCBC 280   284 thread_pool:: 280   273 thread_pool::
HITCBC 281   284 thread_pool(std::size_t num_threads, std::string_view thread_name_prefix) 281   273 thread_pool(std::size_t num_threads, std::string_view thread_name_prefix)
HITCBC 282   284 : impl_(new impl(num_threads, thread_name_prefix)) 282   273 : impl_(new impl(num_threads, thread_name_prefix))
283   { 283   {
HITCBC 284   284 this->set_frame_allocator(std::allocator<void>{}); 284   273 this->set_frame_allocator(std::allocator<void>{});
HITCBC 285   284 } 285   273 }
286   286  
287   void 287   void
HITCBC 288   154 thread_pool:: 288   143 thread_pool::
289   join() noexcept 289   join() noexcept
290   { 290   {
HITCBC 291   154 impl_->join(); 291   143 impl_->join();
HITCBC 292   154 } 292   143 }
293   293  
294   void 294   void
HITCBC 295   2 thread_pool:: 295   2 thread_pool::
296   stop() noexcept 296   stop() noexcept
297   { 297   {
HITCBC 298   2 impl_->stop(); 298   2 impl_->stop();
HITCBC 299   2 } 299   2 }
300   300  
301   //------------------------------------------------------------------------------ 301   //------------------------------------------------------------------------------
302   302  
303   thread_pool::executor_type 303   thread_pool::executor_type
HITCBC 304   11690 thread_pool:: 304   11679 thread_pool::
305   get_executor() const noexcept 305   get_executor() const noexcept
306   { 306   {
HITCBC 307   11690 return executor_type( 307   11679 return executor_type(
HITCBC 308   11690 const_cast<thread_pool&>(*this)); 308   11679 const_cast<thread_pool&>(*this));
309   } 309   }
310   310  
311   void 311   void
HITCBC 312   467 thread_pool::executor_type:: 312   456 thread_pool::executor_type::
313   on_work_started() const noexcept 313   on_work_started() const noexcept
314   { 314   {
HITCBC 315   467 pool_->impl_->on_work_started(); 315   456 pool_->impl_->on_work_started();
HITCBC 316   467 } 316   456 }
317   317  
318   void 318   void
HITCBC 319   467 thread_pool::executor_type:: 319   456 thread_pool::executor_type::
320   on_work_finished() const noexcept 320   on_work_finished() const noexcept
321   { 321   {
HITCBC 322   467 pool_->impl_->on_work_finished(); 322   456 pool_->impl_->on_work_finished();
HITCBC 323   467 } 323   456 }
324   324  
325   void 325   void
HITCBC 326   20346 thread_pool::executor_type:: 326   20597 thread_pool::executor_type::
327   post(continuation& c) const 327   post(continuation& c) const
328   { 328   {
HITCBC 329   20346 pool_->impl_->post(c); 329   20597 pool_->impl_->post(c);
HITCBC 330   20346 } 330   20597 }
331   331  
332   std::coroutine_handle<> 332   std::coroutine_handle<>
HITCBC 333   467 thread_pool::executor_type:: 333   456 thread_pool::executor_type::
334   dispatch(continuation& c) const 334   dispatch(continuation& c) const
335   { 335   {
HITCBC 336   467 if(pool_->impl_->running_in_this_thread()) 336   456 if(pool_->impl_->running_in_this_thread())
HITCBC 337   5 return c.h; 337   5 return c.h;
HITCBC 338   462 pool_->impl_->post(c); 338   451 pool_->impl_->post(c);
HITCBC 339   462 return std::noop_coroutine(); 339   451 return std::noop_coroutine();
340   } 340   }
341   341  
342   } // capy 342   } // capy
343   } // boost 343   } // boost