TLA Line data Source code
1 : //
2 : // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3 : // Copyright (c) 2026 Michael Vandeberg
4 : //
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)
7 : //
8 : // Official repository: https://github.com/boostorg/capy
9 : //
10 :
11 : #include <boost/capy/ex/thread_pool.hpp>
12 : #include <boost/capy/continuation.hpp>
13 : #include <boost/capy/detail/thread_local_ptr.hpp>
14 : #include <boost/capy/ex/abandon.hpp>
15 : #include <boost/capy/ex/frame_allocator.hpp>
16 : #include <boost/capy/test/thread_name.hpp>
17 : #include <algorithm>
18 : #include <atomic>
19 : #include <condition_variable>
20 : #include <cstdio>
21 : #include <mutex>
22 : #include <thread>
23 : #include <vector>
24 :
25 : /*
26 : Thread pool implementation using a shared work queue.
27 :
28 : Work items are continuations linked via their intrusive next pointer,
29 : stored in a single queue protected by a mutex. No per-post heap
30 : allocation: the continuation is owned by the caller and linked
31 : directly. Worker threads wait on a condition_variable until work
32 : is available or stop is requested.
33 :
34 : Threads are started lazily on first post(), under mutex_, to avoid
35 : spawning threads for pools that are constructed but never used. A
36 : stopped pool starts none. Each thread is named
37 : with a configurable prefix plus index for debugger visibility.
38 :
39 : Work tracking: on_work_started/on_work_finished maintain the atomic
40 : outstanding_work_ counter. on_work_started is lock-free, and so is
41 : on_work_finished while the count stays above one. The release that
42 : may take it to zero is made under mutex_, so join() can't see zero,
43 : return and let the pool be destroyed while that call still uses it.
44 : join() blocks until this counter reaches zero, then signals workers
45 : to stop and joins threads.
46 :
47 : Two shutdown paths:
48 : - join(): waits for outstanding work to drain, then stops workers.
49 : - stop(): immediately signals workers to exit; queued work stays queued.
50 : - Destructor: stop() then join(), then queued work is abandoned.
51 : */
52 :
53 : namespace boost {
54 : namespace capy {
55 :
56 : //------------------------------------------------------------------------------
57 :
58 : class thread_pool::impl
59 : {
60 : // Identifies the pool owning the current worker thread, or
61 : // nullptr if the calling thread is not a pool worker. Checked
62 : // by dispatch() to decide between symmetric transfer (inline
63 : // resume) and post.
64 : static inline detail::thread_local_ptr<impl const> current_;
65 :
66 : // Intrusive queue of continuations: the next link is stored in
67 : // continuation::reserved (typed continuation* round-tripped through
68 : // void*). No per-post allocation: the continuation is owned by the caller.
69 : continuation* head_ = nullptr;
70 : continuation* tail_ = nullptr;
71 :
72 HIT 46808 : void push(continuation* c) noexcept
73 : {
74 46808 : c->reserved = nullptr;
75 46808 : if(tail_)
76 3252 : tail_->reserved = c;
77 : else
78 43556 : head_ = c;
79 46808 : tail_ = c;
80 46808 : }
81 :
82 49680 : continuation* pop() noexcept
83 : {
84 49680 : if(!head_)
85 2872 : return nullptr;
86 46808 : continuation* c = head_;
87 46808 : head_ = static_cast<continuation*>(head_->reserved);
88 46808 : if(!head_)
89 43556 : tail_ = nullptr;
90 46808 : return c;
91 : }
92 :
93 90343 : bool empty() const noexcept
94 : {
95 90343 : return head_ == nullptr;
96 : }
97 :
98 : std::mutex mutex_;
99 : std::condition_variable work_cv_;
100 : std::condition_variable done_cv_;
101 : std::vector<std::thread> threads_;
102 : std::atomic<std::size_t> outstanding_work_{0};
103 : bool stop_{false};
104 : bool joined_{false};
105 : std::size_t num_threads_;
106 : char thread_name_prefix_[13]{}; // 12 chars max + null terminator
107 : std::atomic<bool> started_{false};
108 :
109 : public:
110 1436 : ~impl() = default;
111 :
112 : bool
113 13213 : running_in_this_thread() const noexcept
114 : {
115 13213 : return current_.get() == this;
116 : }
117 :
118 : // Abandon queued coroutines. Each is taken under the lock, as other
119 : // threads may still post, and abandoned outside it, as abandoning
120 : // can post.
121 : void
122 2872 : drain_abandoned() noexcept
123 : {
124 : for(;;)
125 : {
126 : continuation* c;
127 : {
128 3409 : std::lock_guard<std::mutex> lock(mutex_);
129 3409 : c = pop();
130 3409 : }
131 3409 : if(!c)
132 2872 : return;
133 537 : if(c->h && c->h != std::noop_coroutine())
134 284 : capy::abandon(*c);
135 537 : }
136 : }
137 :
138 1436 : impl(std::size_t num_threads, std::string_view thread_name_prefix)
139 1436 : : num_threads_(num_threads)
140 : {
141 1436 : if(num_threads_ == 0)
142 8 : num_threads_ = std::max(
143 4 : std::thread::hardware_concurrency(), 1u);
144 :
145 : // Truncate prefix to 12 chars, leaving room for up to 3-digit index.
146 1436 : auto n = thread_name_prefix.copy(thread_name_prefix_, 12);
147 1436 : thread_name_prefix_[n] = '\0';
148 1436 : }
149 :
150 : void
151 46810 : post(continuation& c)
152 : {
153 46810 : if(!started_.load(std::memory_order_acquire))
154 : {
155 : {
156 601 : std::lock_guard<std::mutex> lock(mutex_);
157 601 : if(stop_)
158 : {
159 : // Stopped before any thread started: nothing will
160 : // run it, and the destructor's drain abandons it.
161 220 : push(&c);
162 220 : return;
163 : }
164 601 : }
165 381 : ensure_started();
166 : }
167 46588 : std::lock_guard<std::mutex> lock(mutex_);
168 46588 : push(&c);
169 : // Under the lock so the pool cannot drain, join, and
170 : // destroy the condition variable mid-signal.
171 46588 : work_cv_.notify_one();
172 46588 : }
173 :
174 : void
175 14096 : on_work_started() noexcept
176 : {
177 14096 : outstanding_work_.fetch_add(1, std::memory_order_acq_rel);
178 14096 : }
179 :
180 : void
181 13796 : on_work_finished() noexcept
182 : {
183 13796 : auto n = outstanding_work_.load(std::memory_order_relaxed);
184 13806 : while(n > 1)
185 : {
186 2014 : if(outstanding_work_.compare_exchange_weak(
187 : n, n - 1, std::memory_order_acq_rel))
188 997 : return;
189 : }
190 12799 : std::lock_guard<std::mutex> lock(mutex_);
191 12799 : outstanding_work_.fetch_sub(1, std::memory_order_acq_rel);
192 : // Re-read: a concurrent on_work_started() may have raised it.
193 12799 : if(outstanding_work_.load(
194 12799 : std::memory_order_acquire) == 0 && joined_ && !stop_)
195 : {
196 692 : stop_ = true;
197 692 : done_cv_.notify_all();
198 692 : work_cv_.notify_all();
199 : }
200 12799 : }
201 :
202 : void
203 2722 : join() noexcept
204 : {
205 : {
206 2722 : std::unique_lock<std::mutex> lock(mutex_);
207 2722 : if(joined_)
208 1286 : return;
209 1436 : joined_ = true;
210 :
211 1436 : if(outstanding_work_.load(
212 1436 : std::memory_order_acquire) == 0)
213 : {
214 387 : stop_ = true;
215 387 : work_cv_.notify_all();
216 : }
217 : else
218 : {
219 1049 : done_cv_.wait(lock, [this]{
220 2042 : return stop_;
221 : });
222 : }
223 2722 : }
224 :
225 1946 : for(auto& t : threads_)
226 510 : if(t.joinable())
227 510 : t.join();
228 : }
229 :
230 : void
231 1739 : stop() noexcept
232 : {
233 : // Under the lock: the join() this releases may return and let
234 : // the pool be destroyed as soon as the lock is free.
235 1739 : std::lock_guard<std::mutex> lock(mutex_);
236 1739 : stop_ = true;
237 1739 : work_cv_.notify_all();
238 1739 : done_cv_.notify_all();
239 1739 : }
240 :
241 : private:
242 : void
243 381 : ensure_started()
244 : {
245 : // Under the lock, so join() either sees every thread started
246 : // here or makes this start none. A joined pool still starts them:
247 : // join() may be waiting on work that needs them.
248 381 : std::lock_guard<std::mutex> lock(mutex_);
249 381 : if(started_.load(std::memory_order_relaxed) || stop_)
250 MIS 0 : return;
251 : // Start at the threads already running: if an earlier call
252 : // threw partway, this one adds only the missing ones.
253 HIT 381 : threads_.reserve(num_threads_);
254 891 : for(std::size_t i = threads_.size(); i < num_threads_; ++i)
255 1022 : threads_.emplace_back([this, i]{ run(i); });
256 379 : started_.store(true, std::memory_order_release);
257 381 : }
258 :
259 : void
260 510 : run(std::size_t index)
261 : {
262 : // Build name; set_current_thread_name truncates to platform limits.
263 : char name[16];
264 510 : std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index);
265 510 : set_current_thread_name(name);
266 :
267 : // Mark this thread as a worker of this pool so dispatch()
268 : // can symmetric-transfer when called from within pool work.
269 : struct scoped_pool
270 : {
271 510 : scoped_pool(impl const* p) noexcept { current_.set(p); }
272 510 : ~scoped_pool() noexcept { current_.set(nullptr); }
273 510 : } guard(this);
274 :
275 : for(;;)
276 : {
277 46781 : continuation* c = nullptr;
278 : {
279 46781 : std::unique_lock<std::mutex> lock(mutex_);
280 46781 : work_cv_.wait(lock, [this]{
281 134320 : return !empty() ||
282 134320 : stop_;
283 : });
284 46781 : if(stop_)
285 1020 : return;
286 46271 : c = pop();
287 46781 : }
288 46271 : if(c)
289 46271 : safe_resume(c->h);
290 46271 : }
291 510 : }
292 : };
293 :
294 : //------------------------------------------------------------------------------
295 :
296 1436 : thread_pool::
297 : ~thread_pool()
298 : {
299 1436 : impl_->stop();
300 1436 : impl_->join();
301 1436 : shutdown();
302 : // Takes what the unwinds post too, and the work of any service they
303 : // create, which is shut down as it is created.
304 1436 : impl_->drain_abandoned();
305 1436 : destroy();
306 : // Services give up their work in shutdown(), but anything a
307 : // destructor posts anyway is still destroyed.
308 1436 : impl_->drain_abandoned();
309 1436 : delete impl_;
310 1436 : }
311 :
312 1436 : thread_pool::
313 1436 : thread_pool(std::size_t num_threads, std::string_view thread_name_prefix)
314 1436 : : impl_(new impl(num_threads, thread_name_prefix))
315 : {
316 1436 : this->set_frame_allocator(std::allocator<void>{});
317 1436 : }
318 :
319 : void
320 1286 : thread_pool::
321 : join() noexcept
322 : {
323 1286 : impl_->join();
324 1286 : }
325 :
326 : void
327 303 : thread_pool::
328 : stop() noexcept
329 : {
330 303 : impl_->stop();
331 303 : }
332 :
333 : //------------------------------------------------------------------------------
334 :
335 : thread_pool::executor_type
336 25360 : thread_pool::
337 : get_executor() const noexcept
338 : {
339 25360 : return executor_type(
340 25360 : const_cast<thread_pool&>(*this));
341 : }
342 :
343 : void
344 14096 : thread_pool::executor_type::
345 : on_work_started() const noexcept
346 : {
347 14096 : pool_->impl_->on_work_started();
348 14096 : }
349 :
350 : void
351 13796 : thread_pool::executor_type::
352 : on_work_finished() const noexcept
353 : {
354 13796 : pool_->impl_->on_work_finished();
355 13796 : }
356 :
357 : void
358 33604 : thread_pool::executor_type::
359 : post(continuation& c) const
360 : {
361 33604 : pool_->impl_->post(c);
362 33602 : }
363 :
364 : std::coroutine_handle<>
365 13213 : thread_pool::executor_type::
366 : dispatch(continuation& c) const
367 : {
368 13213 : if(pool_->impl_->running_in_this_thread())
369 7 : return c.h;
370 13206 : pool_->impl_->post(c);
371 13206 : return std::noop_coroutine();
372 : }
373 :
374 : } // capy
375 : } // boost
|
