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