src/ex/thread_pool.cpp

99.3% Lines (152 / 153) 100.0% Functions (28 / 28)

[ ] prev/next file n p prev/next uncovered line

thread_pool.cpp

f(x) Functions (28)

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/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	24103x	void push(continuation* c) noexcept
73		{
74	24103x	c->reserved = nullptr;
75	24103x	if(tail_)
76	3988x	tail_->reserved = c;
77		else
78	20115x	head_ = c;
79	24103x	tail_ = c;
80	24103x	}
81
82	26954x	continuation* pop() noexcept
83		{
84	26954x	if(!head_)
85	2851x	return nullptr;
86	24103x	continuation* c = head_;
87	24103x	head_ = static_cast<continuation*>(head_->reserved);
88	24103x	if(!head_)
89	20115x	tail_ = nullptr;
90	24103x	return c;
91		}
92
93	42747x	bool empty() const noexcept
94		{
95	42747x	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	1424x	~impl() = default;
111
112		bool
113	3207x	running_in_this_thread() const noexcept
114		{
115	3207x	return current_.get() == this;
116		}
117
118		// Abandon queued coroutines. Called after the services shut down,
119		// since their shutdown can post more, and again after they are
120		// destroyed, for anything their destructors post. Each is taken
121		// under the lock, as other threads may still post, and abandoned
122		// outside it, as abandoning can post. Returns whether there was
123		// anything to abandon.
124		bool
125	2851x	drain_abandoned() noexcept
126		{
127	2851x	bool any = false;
128		for(;;)
129		{
130		continuation* c;
131		{
132	5382x	std::lock_guard<std::mutex> lock(mutex_);
133	5382x	c = pop();
134	5382x	}
135	5382x	if(!c)
136	2851x	return any;
137	2531x	any = true;
138	2531x	if(c->h && c->h != std::noop_coroutine())
139	2280x	capy::abandon(*c);
140	2531x	}
141		}
142
143	1424x	impl(std::size_t num_threads, std::string_view thread_name_prefix)
144	1424x	: num_threads_(num_threads)
145		{
146	1424x	if(num_threads_ == 0)
147	8x	num_threads_ = std::max(
148	4x	std::thread::hardware_concurrency(), 1u);
149
150		// Truncate prefix to 12 chars, leaving room for up to 3-digit index.
151	1424x	auto n = thread_name_prefix.copy(thread_name_prefix_, 12);
152	1424x	thread_name_prefix_[n] = '\0';
153	1424x	}
154
155		void
156	24103x	post(continuation& c)
157		{
158	24103x	if(!started_.load(std::memory_order_acquire))
159		{
160		{
161	2571x	std::lock_guard<std::mutex> lock(mutex_);
162	2571x	if(stop_)
163		{
164		// Stopped before any thread started: nothing will
165		// run it, and the destructor's drain abandons it.
166	2204x	push(&c);
167	2204x	return;
168		}
169	2571x	}
170	367x	ensure_started();
171		}
172	21899x	std::lock_guard<std::mutex> lock(mutex_);
173	21899x	push(&c);
174		// Under the lock so the pool cannot drain, join, and
175		// destroy the condition variable mid-signal.
176	21899x	work_cv_.notify_one();
177	21899x	}
178
179		void
180	4084x	on_work_started() noexcept
181		{
182	4084x	outstanding_work_.fetch_add(1, std::memory_order_acq_rel);
183	4084x	}
184
185		void
186	3784x	on_work_finished() noexcept
187		{
188	3784x	auto n = outstanding_work_.load(std::memory_order_relaxed);
189	3794x	while(n > 1)
190		{
191	2802x	if(outstanding_work_.compare_exchange_weak(
192		n, n - 1, std::memory_order_acq_rel))
193	1391x	return;
194		}
195	2393x	std::lock_guard<std::mutex> lock(mutex_);
196	2393x	outstanding_work_.fetch_sub(1, std::memory_order_acq_rel);
197		// Re-read: a concurrent on_work_started() may have raised it.
198	2393x	if(outstanding_work_.load(
199	2393x	std::memory_order_acquire) == 0 && joined_ && !stop_)
200		{
201	690x	stop_ = true;
202	690x	done_cv_.notify_all();
203	690x	work_cv_.notify_all();
204		}
205	2393x	}
206
207		void
208	2705x	join() noexcept
209		{
210		{
211	2705x	std::unique_lock<std::mutex> lock(mutex_);
212	2705x	if(joined_)
213	1281x	return;
214	1424x	joined_ = true;
215
216	1424x	if(outstanding_work_.load(
217	1424x	std::memory_order_acquire) == 0)
218		{
219	378x	stop_ = true;
220	378x	work_cv_.notify_all();
221		}
222		else
223		{
224	1046x	done_cv_.wait(lock, [this]{
225	2037x	return stop_;
226		});
227		}
228	2705x	}
229
230	1916x	for(auto& t : threads_)
231	492x	if(t.joinable())
232	492x	t.join();
233		}
234
235		void
236	1727x	stop() noexcept
237		{
238		// Under the lock: the join() this releases may return and let
239		// the pool be destroyed as soon as the lock is free.
240	1727x	std::lock_guard<std::mutex> lock(mutex_);
241	1727x	stop_ = true;
242	1727x	work_cv_.notify_all();
243	1727x	done_cv_.notify_all();
244	1727x	}
245
246		private:
247		void
248	367x	ensure_started()
249		{
250		// Under the lock, so join() either sees every thread started
251		// here or makes this start none. A joined pool still starts them:
252		// join() may be waiting on work that needs them.
253	367x	std::lock_guard<std::mutex> lock(mutex_);
254	367x	if(started_.load(std::memory_order_relaxed) \|\| stop_)
255	✗	return;
256	367x	threads_.reserve(num_threads_);
257	859x	for(std::size_t i = 0; i < num_threads_; ++i)
258	984x	threads_.emplace_back([this, i]{ run(i); });
259	367x	started_.store(true, std::memory_order_release);
260	367x	}
261
262		void
263	492x	run(std::size_t index)
264		{
265		// Build name; set_current_thread_name truncates to platform limits.
266		char name[16];
267	492x	std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index);
268	492x	set_current_thread_name(name);
269
270		// Mark this thread as a worker of this pool so dispatch()
271		// can symmetric-transfer when called from within pool work.
272		struct scoped_pool
273		{
274	492x	scoped_pool(impl const* p) noexcept { current_.set(p); }
275	492x	~scoped_pool() noexcept { current_.set(nullptr); }
276	492x	} guard(this);
277
278		for(;;)
279		{
280	22064x	continuation* c = nullptr;
281		{
282	22064x	std::unique_lock<std::mutex> lock(mutex_);
283	22064x	work_cv_.wait(lock, [this]{
284	63825x	return !empty() \|\|
285	63825x	stop_;
286		});
287	22064x	if(stop_)
288	984x	return;
289	21572x	c = pop();
290	22064x	}
291	21572x	if(c)
292	21572x	safe_resume(c->h);
293	21572x	}
294	492x	}
295		};
296
297		//------------------------------------------------------------------------------
298
299	1424x	thread_pool::
300		~thread_pool()
301		{
302	1424x	impl_->stop();
303	1424x	impl_->join();
304	1424x	shutdown();
305	1424x	impl_->drain_abandoned();
306		// A frame the drain unwinds can create a service, whose destructor
307		// can post again: repeat until nothing is left of either.
308		do
309		{
310	1427x	destroy();
311		}
312	1427x	while(impl_->drain_abandoned());
313	1424x	delete impl_;
314	1424x	}
315
316	1424x	thread_pool::
317	1424x	thread_pool(std::size_t num_threads, std::string_view thread_name_prefix)
318	1424x	: impl_(new impl(num_threads, thread_name_prefix))
319		{
320	1424x	this->set_frame_allocator(std::allocator<void>{});
321	1424x	}
322
323		void
324	1281x	thread_pool::
325		join() noexcept
326		{
327	1281x	impl_->join();
328	1281x	}
329
330		void
331	303x	thread_pool::
332		stop() noexcept
333		{
334	303x	impl_->stop();
335	303x	}
336
337		//------------------------------------------------------------------------------
338
339		thread_pool::executor_type
340	15341x	thread_pool::
341		get_executor() const noexcept
342		{
343	15341x	return executor_type(
344	15341x	const_cast<thread_pool&>(*this));
345		}
346
347		void
348	4084x	thread_pool::executor_type::
349		on_work_started() const noexcept
350		{
351	4084x	pool_->impl_->on_work_started();
352	4084x	}
353
354		void
355	3784x	thread_pool::executor_type::
356		on_work_finished() const noexcept
357		{
358	3784x	pool_->impl_->on_work_finished();
359	3784x	}
360
361		void
362	20903x	thread_pool::executor_type::
363		post(continuation& c) const
364		{
365	20903x	pool_->impl_->post(c);
366	20903x	}
367
368		std::coroutine_handle<>
369	3207x	thread_pool::executor_type::
370		dispatch(continuation& c) const
371		{
372	3207x	if(pool_->impl_->running_in_this_thread())
373	7x	return c.h;
374	3200x	pool_->impl_->post(c);
375	3200x	return std::noop_coroutine();
376		}
377
378		} // capy
379		} // boost
380