. /src /ex /thread_pool.cpp
100.00% Lines (139/139) 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 an atomic 38   Work tracking: on_work_started/on_work_finished maintain an atomic
39   outstanding_work_ counter. join() blocks until this counter reaches 39   outstanding_work_ counter. join() blocks until this counter reaches
40   zero, then signals workers to stop and joins threads. 40   zero, then signals workers to stop and joins threads.
41   41  
42   Two shutdown paths: 42   Two shutdown paths:
43   - join(): waits for outstanding work to drain, then stops workers. 43   - join(): waits for outstanding work to drain, then stops workers.
44   - stop(): immediately signals workers to exit; queued work is abandoned. 44   - stop(): immediately signals workers to exit; queued work is abandoned.
45   - Destructor: stop() then join() (abandon + wait for threads). 45   - Destructor: stop() then join() (abandon + wait for threads).
46   */ 46   */
47   47  
48   namespace boost { 48   namespace boost {
49   namespace capy { 49   namespace capy {
50   50  
51   //------------------------------------------------------------------------------ 51   //------------------------------------------------------------------------------
52   52  
53   class thread_pool::impl 53   class thread_pool::impl
54   { 54   {
55   // Identifies the pool owning the current worker thread, or 55   // Identifies the pool owning the current worker thread, or
56   // nullptr if the calling thread is not a pool worker. Checked 56   // nullptr if the calling thread is not a pool worker. Checked
57   // by dispatch() to decide between symmetric transfer (inline 57   // by dispatch() to decide between symmetric transfer (inline
58   // resume) and post. 58   // resume) and post.
59   static inline detail::thread_local_ptr<impl const> current_; 59   static inline detail::thread_local_ptr<impl const> current_;
60   60  
61   // Intrusive queue of continuations via continuation::next. 61   // Intrusive queue of continuations via continuation::next.
62   // No per-post allocation: the continuation is owned by the caller. 62   // No per-post allocation: the continuation is owned by the caller.
63   continuation* head_ = nullptr; 63   continuation* head_ = nullptr;
64   continuation* tail_ = nullptr; 64   continuation* tail_ = nullptr;
65   65  
HITCBC 66   18425 void push(continuation* c) noexcept 66   19206 void push(continuation* c) noexcept
67   { 67   {
HITCBC 68   18425 c->next = nullptr; 68   19206 c->next = nullptr;
HITCBC 69   18425 if(tail_) 69   19206 if(tail_)
HITCBC 70   6205 tail_->next = c; 70   3178 tail_->next = c;
71   else 71   else
HITCBC 72   12220 head_ = c; 72   16028 head_ = c;
HITCBC 73   18425 tail_ = c; 73   19206 tail_ = c;
HITCBC 74   18425 } 74   19206 }
75   75  
HITCBC 76   18591 continuation* pop() noexcept 76   19373 continuation* pop() noexcept
77   { 77   {
HITCBC 78   18591 if(!head_) 78   19373 if(!head_)
HITCBC 79   166 return nullptr; 79   167 return nullptr;
HITCBC 80   18425 continuation* c = head_; 80   19206 continuation* c = head_;
HITCBC 81   18425 head_ = head_->next; 81   19206 head_ = head_->next;
HITCBC 82   18425 if(!head_) 82   19206 if(!head_)
HITCBC 83   12220 tail_ = nullptr; 83   16028 tail_ = nullptr;
HITCBC 84   18425 return c; 84   19206 return c;
85   } 85   }
86   86  
HITCBC 87   34379 bool empty() const noexcept 87   36106 bool empty() const noexcept
88   { 88   {
HITCBC 89   34379 return head_ == nullptr; 89   36106 return head_ == nullptr;
90   } 90   }
91   91  
92   std::mutex mutex_; 92   std::mutex mutex_;
93   std::condition_variable work_cv_; 93   std::condition_variable work_cv_;
94   std::condition_variable done_cv_; 94   std::condition_variable done_cv_;
95   std::vector<std::thread> threads_; 95   std::vector<std::thread> threads_;
96   std::atomic<std::size_t> outstanding_work_{0}; 96   std::atomic<std::size_t> outstanding_work_{0};
97   bool stop_{false}; 97   bool stop_{false};
98   bool joined_{false}; 98   bool joined_{false};
99   std::size_t num_threads_; 99   std::size_t num_threads_;
100   char thread_name_prefix_[13]{}; // 12 chars max + null terminator 100   char thread_name_prefix_[13]{}; // 12 chars max + null terminator
101   std::once_flag start_flag_; 101   std::once_flag start_flag_;
102   102  
103   public: 103   public:
HITCBC 104   166 ~impl() = default; 104   167 ~impl() = default;
105   105  
106   bool 106   bool
HITCBC 107   355 running_in_this_thread() const noexcept 107   355 running_in_this_thread() const noexcept
108   { 108   {
HITCBC 109   355 return current_.get() == this; 109   355 return current_.get() == this;
110   } 110   }
111   111  
112   // Destroy abandoned coroutine frames. Must be called 112   // Destroy abandoned coroutine frames. Must be called
113   // before execution_context::shutdown()/destroy() so 113   // before execution_context::shutdown()/destroy() so
114   // that suspended-frame destructors (e.g. delay_awaitable 114   // that suspended-frame destructors (e.g. delay_awaitable
115   // calling timer_service::cancel()) run while services 115   // calling timer_service::cancel()) run while services
116   // are still valid. 116   // are still valid.
117   void 117   void
HITCBC 118   166 drain_abandoned() noexcept 118   167 drain_abandoned() noexcept
119   { 119   {
HITCBC 120   361 while(auto* c = pop()) 120   364 while(auto* c = pop())
121   { 121   {
HITCBC 122   195 auto h = c->h; 122   197 auto h = c->h;
HITCBC 123   195 if(h && h != std::noop_coroutine()) 123   197 if(h && h != std::noop_coroutine())
HITCBC 124   144 h.destroy(); 124   146 h.destroy();
HITCBC 125   195 } 125   197 }
HITCBC 126   166 } 126   167 }
127   127  
HITCBC 128   166 impl(std::size_t num_threads, std::string_view thread_name_prefix) 128   167 impl(std::size_t num_threads, std::string_view thread_name_prefix)
HITCBC 129   166 : num_threads_(num_threads) 129   167 : num_threads_(num_threads)
130   { 130   {
HITCBC 131   166 if(num_threads_ == 0) 131   167 if(num_threads_ == 0)
HITCBC 132   4 num_threads_ = std::max( 132   6 num_threads_ = std::max(
HITCBC 133   2 std::thread::hardware_concurrency(), 1u); 133   3 std::thread::hardware_concurrency(), 1u);
134   134  
135   // Truncate prefix to 12 chars, leaving room for up to 3-digit index. 135   // Truncate prefix to 12 chars, leaving room for up to 3-digit index.
HITCBC 136   166 auto n = thread_name_prefix.copy(thread_name_prefix_, 12); 136   167 auto n = thread_name_prefix.copy(thread_name_prefix_, 12);
HITCBC 137   166 thread_name_prefix_[n] = '\0'; 137   167 thread_name_prefix_[n] = '\0';
HITCBC 138   166 } 138   167 }
139   139  
140   void 140   void
HITCBC 141   18425 post(continuation& c) 141   19206 post(continuation& c)
142   { 142   {
HITCBC 143   18425 ensure_started(); 143   19206 ensure_started();
144   { 144   {
HITCBC 145   18425 std::lock_guard<std::mutex> lock(mutex_); 145   19206 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 146   18425 push(&c); 146   19206 push(&c);
HITCBC 147   18425 } 147   19206 }
HITCBC 148   18425 work_cv_.notify_one(); 148   19206 work_cv_.notify_one();
HITCBC 149   18425 } 149   19206 }
150   150  
151   void 151   void
HITCBC 152   347 on_work_started() noexcept 152   347 on_work_started() noexcept
153   { 153   {
HITCBC 154   347 outstanding_work_.fetch_add(1, std::memory_order_acq_rel); 154   347 outstanding_work_.fetch_add(1, std::memory_order_acq_rel);
HITCBC 155   347 } 155   347 }
156   156  
157   void 157   void
HITCBC 158   347 on_work_finished() noexcept 158   347 on_work_finished() noexcept
159   { 159   {
HITCBC 160   347 if(outstanding_work_.fetch_sub( 160   347 if(outstanding_work_.fetch_sub(
HITCBC 161   347 1, std::memory_order_acq_rel) == 1) 161   347 1, std::memory_order_acq_rel) == 1)
162   { 162   {
HITCBC 163   87 std::lock_guard<std::mutex> lock(mutex_); 163   87 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 164   87 if(joined_ && !stop_) 164   87 if(joined_ && !stop_)
HITCBC 165   4 stop_ = true; 165   4 stop_ = true;
HITCBC 166   87 done_cv_.notify_all(); 166   87 done_cv_.notify_all();
HITCBC 167   87 work_cv_.notify_all(); 167   87 work_cv_.notify_all();
HITCBC 168   87 } 168   87 }
HITCBC 169   347 } 169   347 }
170   170  
171   void 171   void
HITCBC 172   178 join() noexcept 172   180 join() noexcept
173   { 173   {
174   { 174   {
HITCBC 175   178 std::unique_lock<std::mutex> lock(mutex_); 175   180 std::unique_lock<std::mutex> lock(mutex_);
HITCBC 176   178 if(joined_) 176   180 if(joined_)
HITCBC 177   12 return; 177   13 return;
HITCBC 178   166 joined_ = true; 178   167 joined_ = true;
179   179  
HITCBC 180   166 if(outstanding_work_.load( 180   167 if(outstanding_work_.load(
HITCBC 181   166 std::memory_order_acquire) == 0) 181   167 std::memory_order_acquire) == 0)
182   { 182   {
HITCBC 183   111 stop_ = true; 183   110 stop_ = true;
HITCBC 184   111 work_cv_.notify_all(); 184   110 work_cv_.notify_all();
185   } 185   }
186   else 186   else
187   { 187   {
HITCBC 188   55 done_cv_.wait(lock, [this]{ 188   57 done_cv_.wait(lock, [this]{
HITCBC 189   60 return stop_; 189   62 return stop_;
190   }); 190   });
191   } 191   }
HITCBC 192   178 } 192   180 }
193   193  
HITCBC 194   359 for(auto& t : threads_) 194   364 for(auto& t : threads_)
HITCBC 195   193 if(t.joinable()) 195   197 if(t.joinable())
HITCBC 196   193 t.join(); 196   197 t.join();
197   } 197   }
198   198  
199   void 199   void
HITCBC 200   168 stop() noexcept 200   169 stop() noexcept
201   { 201   {
202   { 202   {
HITCBC 203   168 std::lock_guard<std::mutex> lock(mutex_); 203   169 std::lock_guard<std::mutex> lock(mutex_);
HITCBC 204   168 stop_ = true; 204   169 stop_ = true;
HITCBC 205   168 } 205   169 }
HITCBC 206   168 work_cv_.notify_all(); 206   169 work_cv_.notify_all();
HITCBC 207   168 done_cv_.notify_all(); 207   169 done_cv_.notify_all();
HITCBC 208   168 } 208   169 }
209   209  
210   private: 210   private:
211   void 211   void
HITCBC 212   18425 ensure_started() 212   19206 ensure_started()
213   { 213   {
HITCBC 214   18425 std::call_once(start_flag_, [this]{ 214   19206 std::call_once(start_flag_, [this]{
HITCBC 215   109 threads_.reserve(num_threads_); 215   110 threads_.reserve(num_threads_);
HITCBC 216   302 for(std::size_t i = 0; i < num_threads_; ++i) 216   307 for(std::size_t i = 0; i < num_threads_; ++i)
HITCBC 217   386 threads_.emplace_back([this, i]{ run(i); }); 217   394 threads_.emplace_back([this, i]{ run(i); });
HITCBC 218   109 }); 218   110 });
HITCBC 219   18425 } 219   19206 }
220   220  
221   void 221   void
HITCBC 222   193 run(std::size_t index) 222   197 run(std::size_t index)
223   { 223   {
224   // Build name; set_current_thread_name truncates to platform limits. 224   // Build name; set_current_thread_name truncates to platform limits.
225   char name[16]; 225   char name[16];
HITCBC 226   193 std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index); 226   197 std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index);
HITCBC 227   193 set_current_thread_name(name); 227   197 set_current_thread_name(name);
228   228  
229   // Mark this thread as a worker of this pool so dispatch() 229   // Mark this thread as a worker of this pool so dispatch()
230   // can symmetric-transfer when called from within pool work. 230   // can symmetric-transfer when called from within pool work.
231   struct scoped_pool 231   struct scoped_pool
232   { 232   {
HITCBC 233   193 scoped_pool(impl const* p) noexcept { current_.set(p); } 233   197 scoped_pool(impl const* p) noexcept { current_.set(p); }
HITCBC 234   193 ~scoped_pool() noexcept { current_.set(nullptr); } 234   197 ~scoped_pool() noexcept { current_.set(nullptr); }
HITCBC 235   193 } guard(this); 235   197 } guard(this);
236   236  
237   for(;;) 237   for(;;)
238   { 238   {
HITCBC 239   18423 continuation* c = nullptr; 239   19206 continuation* c = nullptr;
240   { 240   {
HITCBC 241   18423 std::unique_lock<std::mutex> lock(mutex_); 241   19206 std::unique_lock<std::mutex> lock(mutex_);
HITCBC 242   18423 work_cv_.wait(lock, [this]{ 242   19206 work_cv_.wait(lock, [this]{
HITCBC 243   50433 return !empty() || 243   53108 return !empty() ||
HITCBC 244   50433 stop_; 244   53108 stop_;
245   }); 245   });
HITCBC 246   18423 if(stop_) 246   19206 if(stop_)
HITCBC 247   386 return; 247   394 return;
HITCBC 248   18230 c = pop(); 248   19009 c = pop();
HITCBC 249   18423 } 249   19206 }
HITCBC 250   18230 if(c) 250   19009 if(c)
HITCBC 251   18230 safe_resume(c->h); 251   19009 safe_resume(c->h);
HITCBC 252   18230 } 252   19009 }
HITCBC 253   193 } 253   197 }
254   }; 254   };
255   255  
256   //------------------------------------------------------------------------------ 256   //------------------------------------------------------------------------------
257   257  
HITCBC 258   166 thread_pool:: 258   167 thread_pool::
259   ~thread_pool() 259   ~thread_pool()
260   { 260   {
HITCBC 261   166 impl_->stop(); 261   167 impl_->stop();
HITCBC 262   166 impl_->join(); 262   167 impl_->join();
HITCBC 263   166 impl_->drain_abandoned(); 263   167 impl_->drain_abandoned();
HITCBC 264   166 shutdown(); 264   167 shutdown();
HITCBC 265   166 destroy(); 265   167 destroy();
HITCBC 266   166 delete impl_; 266   167 delete impl_;
HITCBC 267   166 } 267   167 }
268   268  
HITCBC 269   166 thread_pool:: 269   167 thread_pool::
HITCBC 270   166 thread_pool(std::size_t num_threads, std::string_view thread_name_prefix) 270   167 thread_pool(std::size_t num_threads, std::string_view thread_name_prefix)
HITCBC 271   166 : impl_(new impl(num_threads, thread_name_prefix)) 271   167 : impl_(new impl(num_threads, thread_name_prefix))
272   { 272   {
HITCBC 273   166 this->set_frame_allocator(std::allocator<void>{}); 273   167 this->set_frame_allocator(std::allocator<void>{});
HITCBC 274   166 } 274   167 }
275   275  
276   void 276   void
HITCBC 277   12 thread_pool:: 277   13 thread_pool::
278   join() noexcept 278   join() noexcept
279   { 279   {
HITCBC 280   12 impl_->join(); 280   13 impl_->join();
HITCBC 281   12 } 281   13 }
282   282  
283   void 283   void
HITCBC 284   2 thread_pool:: 284   2 thread_pool::
285   stop() noexcept 285   stop() noexcept
286   { 286   {
HITCBC 287   2 impl_->stop(); 287   2 impl_->stop();
HITCBC 288   2 } 288   2 }
289   289  
290   //------------------------------------------------------------------------------ 290   //------------------------------------------------------------------------------
291   291  
292   thread_pool::executor_type 292   thread_pool::executor_type
HITCBC 293   11580 thread_pool:: 293   11582 thread_pool::
294   get_executor() const noexcept 294   get_executor() const noexcept
295   { 295   {
HITCBC 296   11580 return executor_type( 296   11582 return executor_type(
HITCBC 297   11580 const_cast<thread_pool&>(*this)); 297   11582 const_cast<thread_pool&>(*this));
298   } 298   }
299   299  
300   void 300   void
HITCBC 301   347 thread_pool::executor_type:: 301   347 thread_pool::executor_type::
302   on_work_started() const noexcept 302   on_work_started() const noexcept
303   { 303   {
HITCBC 304   347 pool_->impl_->on_work_started(); 304   347 pool_->impl_->on_work_started();
HITCBC 305   347 } 305   347 }
306   306  
307   void 307   void
HITCBC 308   347 thread_pool::executor_type:: 308   347 thread_pool::executor_type::
309   on_work_finished() const noexcept 309   on_work_finished() const noexcept
310   { 310   {
HITCBC 311   347 pool_->impl_->on_work_finished(); 311   347 pool_->impl_->on_work_finished();
HITCBC 312   347 } 312   347 }
313   313  
314   void 314   void
HITCBC 315   18083 thread_pool::executor_type:: 315   18864 thread_pool::executor_type::
316   post(continuation& c) const 316   post(continuation& c) const
317   { 317   {
HITCBC 318   18083 pool_->impl_->post(c); 318   18864 pool_->impl_->post(c);
HITCBC 319   18083 } 319   18864 }
320   320  
321   std::coroutine_handle<> 321   std::coroutine_handle<>
HITCBC 322   355 thread_pool::executor_type:: 322   355 thread_pool::executor_type::
323   dispatch(continuation& c) const 323   dispatch(continuation& c) const
324   { 324   {
HITCBC 325   355 if(pool_->impl_->running_in_this_thread()) 325   355 if(pool_->impl_->running_in_this_thread())
HITCBC 326   13 return c.h; 326   13 return c.h;
HITCBC 327   342 pool_->impl_->post(c); 327   342 pool_->impl_->post(c);
HITCBC 328   342 return std::noop_coroutine(); 328   342 return std::noop_coroutine();
329   } 329   }
330   330  
331   } // capy 331   } // capy
332   } // boost 332   } // boost