src/ex/thread_pool.cpp

100.0% Lines (140/0/140) 100.0% List of functions (29/0/29)

[ ] prev/next file

thread_pool.cpp

f(x) Functions (29)

Function Calls Lines Blocks

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