include/boost/corosio/detail/thread_pool.hpp

98.8% Lines (83/84) 100.0% List of functions (12/13) 78.3% Branches (36/46)

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

thread_pool.hpp

f(x) Functions (13)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Steve Gerbino
3			//
4			// Distributed under the Boost Software License, Version 1.0. (See accompanying
5			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6			//
7			// Official repository: https://github.com/cppalliance/corosio
8			//
9
10			#ifndef BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
11			#define BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
12
13			#include <boost/corosio/detail/config.hpp>
14			#include <boost/corosio/detail/intrusive.hpp>
15			#include <boost/capy/error.hpp>
16			#include <boost/capy/ex/execution_context.hpp>
17			#include <boost/capy/test/thread_name.hpp>
18
19			#include <atomic>
20			#include <condition_variable>
21			#include <cstdio>
22			#include <mutex>
23			#include <stdexcept>
24			#include <system_error>
25			#include <thread>
26			#include <vector>
27
28			namespace boost::corosio::detail {
29
30			/** Base class for thread pool work items.
31
32			Derive from this to create work that can be posted to a
33			@ref thread_pool. Uses static function pointer dispatch,
34			consistent with the IOCP `op` pattern.
35
36			@par Example
37			@code
38			struct my_work : pool_work_item
39			{
40			int* result;
41			static void execute( pool_work_item* w ) noexcept
42			{
43			auto* self = static_cast<my_work*>( w );
44			*self->result = 42;
45			}
46			};
47
48			my_work w;
49			w.func_ = &my_work::execute;
50			w.result = &r;
51			auto ec = pool.post( &w );
52			@endcode
53			*/
54		712x	struct pool_work_item : intrusive_queue<pool_work_item>::node
55			{
56			/// Static dispatch function signature.
57			using func_type = void ()(pool_work_item) noexcept;
58
59			/// Completion handler invoked by the worker thread.
60		712x	func_type func_ = nullptr;
61			};
62
63			/** Shared thread pool for dispatching blocking operations.
64
65			Provides a fixed pool of reusable worker threads for operations
66			that cannot be integrated with async I/O (e.g. blocking DNS
67			calls). Registered as an `execution_context::service` so it
68			is a singleton per io_context.
69
70			The service is created with its context, but the workers start on
71			the first `post()`: a context that never opens a file and never
72			resolves a name never pays for a thread. The default thread count
73			is 1.
74
75			@par Thread Safety
76			All public member functions are thread-safe.
77
78			@par Shutdown
79			Sets a shutdown flag, notifies all threads, and joins them.
80			In-flight blocking calls complete naturally before the thread
81			exits.
82
83			@note Create this service after the scheduler its work items post
84			completions to. Services shut down newest first, so a pool created
85			earlier joins its workers only after the scheduler has drained its
86			completion queue, and the completion the last worker posts is then
87			neither run nor destroyed.
88
89			@note The type is symbol-visible because services are keyed by type
90			identity: with RTTI, hidden behind a shared library boundary, a
91			module that asks for the pool would look up, and create, one of its
92			own (the no-RTTI key is a template static whose visibility follows
93			the template it is instantiated from).
94			*/
95			class BOOST_COROSIO_SYMBOL_VISIBLE thread_pool final
96			: public capy::execution_context::service
97			{
98			std::mutex mutex_;
99			std::condition_variable cv_;
100			intrusive_queue<pool_work_item> work_queue_;
101			std::vector<std::thread> threads_;
102			unsigned num_threads_;
103		2294x	bool shutdown_ = false;
104
105			void worker_loop(unsigned index);
106			std::error_code start_workers() noexcept;
107
108			public:
109			using key_type = thread_pool;
110
111			/** Construct the thread pool service.
112
113			Records the worker count. The workers themselves start on the
114			first `post()`.
115
116			@par Exception Safety
117			Strong guarantee.
118
119			@param ctx Reference to the owning execution_context.
120			@param num_threads Number of worker threads. Must be
121			at least 1.
122
123			@throws std::logic_error If `num_threads` is 0.
124			*/
125		6882x	explicit thread_pool(
126			[[maybe_unused]] capy::execution_context& ctx, unsigned num_threads = 1)
127		2294x	: num_threads_(num_threads)
128		4588x	{
129	2/2 ✓ Branch 0 taken 2293 times. ✓ Branch 1 taken 1 time.	2294x	if (!num_threads)
130	2/4 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 1 time.	1x	throw std::logic_error("thread_pool requires at least 1 thread");
131		4588x	}
132
133			/** Destroy the pool, joining any worker `shutdown()` never reached.
134
135			The context's shutdown walk is the normal path; this only
136			catches a pool created after that walk, whose `shutdown()` is
137			therefore never called and whose joinable threads would
138			otherwise terminate the process. A pool that was never posted
139			to holds no thread and needs neither.
140			*/
141		6877x	~thread_pool() override
142		4584x	{
143	2/2 ✓ Branch 0 taken 2292 times. ✓ Branch 1 taken 1 time.	2293x	if (!threads_.empty())
144	1/2 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken.	1x	shutdown();
145		6877x	}
146
147			thread_pool(thread_pool const&) = delete;
148			thread_pool& operator=(thread_pool const&) = delete;
149
150			/** Enqueue a work item for execution on the thread pool.
151
152			The first item posted starts the workers. Zero-allocation:
153			the caller owns the work item's storage.
154
155			A refusal answers with the code the caller reports for the
156			operation it was starting, so that a system that will not give
157			the pool a thread is not mistaken for a cancellation.
158
159			@par Thread Safety
160			Safe. Racing first posts start the workers once.
161
162			@param w The work item to execute. Must remain valid until
163			its `func_` has been called.
164
165			@return An empty code if the item was enqueued;
166			`capy::error::canceled` if the pool has already shut
167			down; otherwise the code of the thread the system
168			refused, which left the pool with no worker at all.
169			*/
170			[[nodiscard]] std::error_code post(pool_work_item* w) noexcept;
171
172			/** Return the number of workers the pool has started.
173
174			Zero until the first `post()`, and zero again once
175			`shutdown()` has joined them.
176
177			@par Thread Safety
178			Safe.
179			*/
180		6x	unsigned worker_count() noexcept
181			{
182	1/2 ✓ Branch 0 taken 6 times. ✗ Branch 1 not taken.	6x	std::lock_guard<std::mutex> lock(mutex_);
183		6x	return static_cast<unsigned>(threads_.size());
184		6x	}
185
186			/** Shut down the thread pool.
187
188			Signals all threads to exit after draining any
189			remaining queued work, then joins them.
190			*/
191			void shutdown() override;
192			};
193
194			inline void
195		144x	thread_pool::worker_loop(unsigned index)
196			{
197			// Name format chosen to fit Linux's 15-char pthread limit:
198			// "tpool-svc-" (10) + up to 4 digit index leaves "tpool-svc-9999".
199			char name[16];
200		144x	std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
201		144x	capy::set_current_thread_name(name);
202
203		594x	for (;;)
204			{
205			pool_work_item* w;
206			{
207		594x	std::unique_lock<std::mutex> lock(mutex_);
208	2/4 ✓ Branch 0 taken 594 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 594 times. ✗ Branch 3 not taken.	1188x	cv_.wait(
209	2/2 ✓ Branch 0 taken 168 times. ✓ Branch 1 taken 606 times.	1368x	lock, [this] { return shutdown_ \|\| !work_queue_.empty(); });
210
211		594x	w = work_queue_.pop();
212	2/2 ✓ Branch 0 taken 450 times. ✓ Branch 1 taken 144 times.	594x	if (!w)
213			{
214	1/2 ✓ Branch 0 taken 144 times. ✗ Branch 1 not taken.	144x	if (shutdown_)
215		144x	return;
216		✗	continue;
217			}
218		594x	}
219		450x	w->func_(w);
220			}
221		144x	}
222
223			// Called with mutex_ held, so the workers are started once however
224			// many threads race the first post.
225			inline std::error_code
226		455x	thread_pool::start_workers() noexcept
227			{
228	2/2 ✓ Branch 0 taken 309 times. ✓ Branch 1 taken 146 times.	455x	if (!threads_.empty())
229		309x	return {};
230		146x	std::error_code ec;
231			try
232			{
233	2/2 ✓ Branch 0 taken 145 times. ✓ Branch 1 taken 1 time.	146x	threads_.reserve(num_threads_);
234	2/2 ✓ Branch 0 taken 148 times. ✓ Branch 1 taken 141 times.	289x	for (unsigned i = 0; i < num_threads_; ++i)
235	2/2 ✓ Branch 0 taken 144 times. ✓ Branch 1 taken 4 times.	292x	threads_.emplace_back([this, i] { worker_loop(i + 1); });
236		146x	}
237			catch (std::system_error const& e)
238			{
239			// The refusal is carried out, not swallowed: a thread the
240			// system will not give is a real error and the operation that
241			// asked for it says so, rather than reporting the cancellation
242			// that belongs to a stop token.
243		1x	ec = e.code();
244		1x	}
245			catch (...)
246			{
247		4x	ec = std::make_error_code(std::errc::resource_unavailable_try_again);
248		5x	}
249			// A pool short of workers still runs everything posted to it, only
250			// less of it at once, so a partial start is a start. What it does
251			// not do is come back for the rest: the size is a tuning knob, and
252			// topping it up would put a thread creation on the initiator's
253			// path for every operation after a refusal.
254	2/2 ✓ Branch 0 taken 5 times. ✓ Branch 1 taken 141 times.	146x	if (!threads_.empty())
255		141x	return {};
256		5x	return ec;
257		460x	}
258
259			inline std::error_code
260		466x	thread_pool::post(pool_work_item* w) noexcept
261			{
262			{
263	1/2 ✓ Branch 0 taken 466 times. ✗ Branch 1 not taken.	466x	std::lock_guard<std::mutex> lock(mutex_);
264	2/2 ✓ Branch 0 taken 11 times. ✓ Branch 1 taken 455 times.	466x	if (shutdown_)
265		11x	return capy::error::canceled;
266			// The system can refuse a thread, and an initiator has no way
267			// to throw; a refused post is the failure the callers already
268			// report through the operation they were starting.
269	2/2 ✓ Branch 0 taken 5 times. ✓ Branch 1 taken 450 times.	455x	if (auto ec = start_workers())
270		5x	return ec;
271		450x	work_queue_.push(w);
272		466x	}
273		450x	cv_.notify_one();
274		450x	return {};
275		466x	}
276
277			inline void
278		2303x	thread_pool::shutdown()
279			{
280			{
281		2303x	std::lock_guard<std::mutex> lock(mutex_);
282		2303x	shutdown_ = true;
283		2303x	}
284		2303x	cv_.notify_all();
285
286			// Unlocked, though a post may add to threads_: the flag above is
287			// published under the same mutex, so a post that has not taken it
288			// yet will find it set and start nothing, and one already inside
289			// released the mutex before this thread acquired it.
290	2/2 ✓ Branch 0 taken 144 times. ✓ Branch 1 taken 2303 times.	2447x	for (auto& t : threads_)
291			{
292	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 144 times.	144x	if (t.joinable())
293		144x	t.join();
294			}
295		2303x	threads_.clear();
296
297			{
298		2303x	std::lock_guard<std::mutex> lock(mutex_);
299	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 2303 times.	2303x	while (work_queue_.pop())
300			;
301		2303x	}
302		2303x	}
303
304			/** A reference to the context's shared thread pool, bound on first use.
305
306			Services that hand blocking work to the pool hold one of these
307			instead of a reference bound at construction. They are constructed
308			from the scheduler's constructor, where the pool they created would
309			be older than the scheduler and would join too late; binding on
310			first use puts the pool after it instead.
311
312			The owning `io_context` creates the pool service during
313			construction, so by the time any operation can run the binding only
314			ever finds it. That is what keeps `get()` from constructing
315			anything on an initiator's thread, and so from throwing where an
316			initiator may not: the throwing spelling exists for a scheduler
317			driven without an `io_context`. What the service defers is its
318			workers, and those are started by `post()`, which reports a refusal
319			rather than throwing it.
320
321			@par Thread Safety
322			Distinct objects: Safe.
323			Shared objects: Safe.
324
325			@see thread_pool
326			*/
327			class thread_pool_ref
328			{
329			capy::execution_context& ctx_;
330		6873x	std::atomic<thread_pool*> pool_{nullptr};
331
332			public:
333			/** Construct a reference into the given context.
334
335			@param ctx The context whose pool is used.
336			*/
337		13746x	explicit thread_pool_ref(capy::execution_context& ctx) noexcept : ctx_(ctx)
338		6873x	{
339		13746x	}
340
341			thread_pool_ref(thread_pool_ref const&) = delete;
342			thread_pool_ref& operator=(thread_pool_ref const&) = delete;
343
344			/** Return the pool, creating it if this is the first use.
345
346			@par Preconditions
347			For the throwing clauses below to be unreachable, the owning
348			context must already hold the pool service. Every `io_context`
349			constructor installs it — what waits for a first post is the
350			service's workers, not the service — so the creating branch is
351			reached only by a scheduler driven without one.
352
353			@par Exception Safety
354			Strong guarantee.
355
356			@throws std::bad_alloc If the service cannot be allocated.
357
358			@throws std::logic_error If the pool is asked for zero threads.
359
360			@return The context's shared thread pool.
361			*/
362		442x	thread_pool& get()
363			{
364		442x	auto* p = pool_.load(std::memory_order_acquire);
365	2/2 ✓ Branch 0 taken 295 times. ✓ Branch 1 taken 147 times.	442x	if (!p)
366			{
367		147x	p = &ctx_.use_service<thread_pool>();
368		147x	pool_.store(p, std::memory_order_release);
369		147x	}
370		442x	return *p;
371			}
372			};
373
374			} // namespace boost::corosio::detail
375
376			#endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
377