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	23183x	void push(continuation* c) noexcept
73		{
74	23183x	c->reserved = nullptr;
75	23183x	if(tail_)
76	3564x	tail_->reserved = c;
77		else
78	19619x	head_ = c;
79	23183x	tail_ = c;
80	23183x	}
81
82	26048x	continuation* pop() noexcept
83		{
84	26048x	if(!head_)
85	2865x	return nullptr;
86	23183x	continuation* c = head_;
87	23183x	head_ = static_cast<continuation*>(head_->reserved);
88	23183x	if(!head_)
89	19619x	tail_ = nullptr;
90	23183x	return c;
91		}
92
93	40736x	bool empty() const noexcept
94		{
95	40736x	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	1431x	~impl() = default;
111
112		bool
113	2712x	running_in_this_thread() const noexcept
114		{
115	2712x	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	2865x	drain_abandoned() noexcept
126		{
127	2865x	bool any = false;
128		for(;;)
129		{
130		continuation* c;
131		{
132	5398x	std::lock_guard<std::mutex> lock(mutex_);
133	5398x	c = pop();
134	5398x	}
135	5398x	if(!c)
136	2865x	return any;
137	2533x	any = true;
138	2533x	if(c->h && c->h != std::noop_coroutine())
139	2282x	capy::abandon(*c);
140	2533x	}
141		}
142
143	1431x	impl(std::size_t num_threads, std::string_view thread_name_prefix)
144	1431x	: num_threads_(num_threads)
145		{
146	1431x	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	1431x	auto n = thread_name_prefix.copy(thread_name_prefix_, 12);
152	1431x	thread_name_prefix_[n] = '\0';
153	1431x	}
154
155		void
156	23183x	post(continuation& c)
157		{
158	23183x	if(!started_.load(std::memory_order_acquire))
159		{
160		{
161	2578x	std::lock_guard<std::mutex> lock(mutex_);
162	2578x	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	2578x	}
170	374x	ensure_started();
171		}
172	20979x	std::lock_guard<std::mutex> lock(mutex_);
173	20979x	push(&c);
174		// Under the lock so the pool cannot drain, join, and
175		// destroy the condition variable mid-signal.
176	20979x	work_cv_.notify_one();
177	20979x	}
178
179		void
180	3589x	on_work_started() noexcept
181		{
182	3589x	outstanding_work_.fetch_add(1, std::memory_order_acq_rel);
183	3589x	}
184
185		void
186	3289x	on_work_finished() noexcept
187		{
188	3289x	auto n = outstanding_work_.load(std::memory_order_relaxed);
189	3300x	while(n > 1)
190		{
191	1956x	if(outstanding_work_.compare_exchange_weak(
192		n, n - 1, std::memory_order_acq_rel))
193	967x	return;
194		}
195	2322x	std::lock_guard<std::mutex> lock(mutex_);
196	2322x	outstanding_work_.fetch_sub(1, std::memory_order_acq_rel);
197		// Re-read: a concurrent on_work_started() may have raised it.
198	2322x	if(outstanding_work_.load(
199	2322x	std::memory_order_acquire) == 0 && joined_ && !stop_)
200		{
201	715x	stop_ = true;
202	715x	done_cv_.notify_all();
203	715x	work_cv_.notify_all();
204		}
205	2322x	}
206
207		void
208	2719x	join() noexcept
209		{
210		{
211	2719x	std::unique_lock<std::mutex> lock(mutex_);
212	2719x	if(joined_)
213	1288x	return;
214	1431x	joined_ = true;
215
216	1431x	if(outstanding_work_.load(
217	1431x	std::memory_order_acquire) == 0)
218		{
219	362x	stop_ = true;
220	362x	work_cv_.notify_all();
221		}
222		else
223		{
224	1069x	done_cv_.wait(lock, [this]{
225	2085x	return stop_;
226		});
227		}
228	2719x	}
229
230	1929x	for(auto& t : threads_)
231	498x	if(t.joinable())
232	498x	t.join();
233		}
234
235		void
236	1734x	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	1734x	std::lock_guard<std::mutex> lock(mutex_);
241	1734x	stop_ = true;
242	1734x	work_cv_.notify_all();
243	1734x	done_cv_.notify_all();
244	1734x	}
245
246		private:
247		void
248	374x	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	374x	std::lock_guard<std::mutex> lock(mutex_);
254	374x	if(started_.load(std::memory_order_relaxed) \|\| stop_)
255	✗	return;
256	374x	threads_.reserve(num_threads_);
257	872x	for(std::size_t i = 0; i < num_threads_; ++i)
258	996x	threads_.emplace_back([this, i]{ run(i); });
259	374x	started_.store(true, std::memory_order_release);
260	374x	}
261
262		void
263	498x	run(std::size_t index)
264		{
265		// Build name; set_current_thread_name truncates to platform limits.
266		char name[16];
267	498x	std::snprintf(name, sizeof(name), "%s%zu", thread_name_prefix_, index);
268	498x	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	498x	scoped_pool(impl const* p) noexcept { current_.set(p); }
275	498x	~scoped_pool() noexcept { current_.set(nullptr); }
276	498x	} guard(this);
277
278		for(;;)
279		{
280	21148x	continuation* c = nullptr;
281		{
282	21148x	std::unique_lock<std::mutex> lock(mutex_);
283	21148x	work_cv_.wait(lock, [this]{
284	60729x	return !empty() \|\|
285	60729x	stop_;
286		});
287	21148x	if(stop_)
288	996x	return;
289	20650x	c = pop();
290	21148x	}
291	20650x	if(c)
292	20650x	safe_resume(c->h);
293	20650x	}
294	498x	}
295		};
296
297		//------------------------------------------------------------------------------
298
299	1431x	thread_pool::
300		~thread_pool()
301		{
302	1431x	impl_->stop();
303	1431x	impl_->join();
304	1431x	shutdown();
305	1431x	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	1434x	destroy();
311		}
312	1434x	while(impl_->drain_abandoned());
313	1431x	delete impl_;
314	1431x	}
315
316	1431x	thread_pool::
317	1431x	thread_pool(std::size_t num_threads, std::string_view thread_name_prefix)
318	1431x	: impl_(new impl(num_threads, thread_name_prefix))
319		{
320	1431x	this->set_frame_allocator(std::allocator<void>{});
321	1431x	}
322
323		void
324	1288x	thread_pool::
325		join() noexcept
326		{
327	1288x	impl_->join();
328	1288x	}
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	14848x	thread_pool::
341		get_executor() const noexcept
342		{
343	14848x	return executor_type(
344	14848x	const_cast<thread_pool&>(*this));
345		}
346
347		void
348	3589x	thread_pool::executor_type::
349		on_work_started() const noexcept
350		{
351	3589x	pool_->impl_->on_work_started();
352	3589x	}
353
354		void
355	3289x	thread_pool::executor_type::
356		on_work_finished() const noexcept
357		{
358	3289x	pool_->impl_->on_work_finished();
359	3289x	}
360
361		void
362	20478x	thread_pool::executor_type::
363		post(continuation& c) const
364		{
365	20478x	pool_->impl_->post(c);
366	20478x	}
367
368		std::coroutine_handle<>
369	2712x	thread_pool::executor_type::
370		dispatch(continuation& c) const
371		{
372	2712x	if(pool_->impl_->running_in_this_thread())
373	7x	return c.h;
374	2705x	pool_->impl_->post(c);
375	2705x	return std::noop_coroutine();
376		}
377
378		} // capy
379		} // boost
380