include/boost/corosio/detail/thread_pool.hpp

98.7% Lines (74/0/75) 100.0% List of functions (11/0/11)

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

thread_pool.hpp

f(x) Functions (11)

Function Calls Lines Blocks

Line	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		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		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		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	3298x	explicit thread_pool(
126		[[maybe_unused]] capy::execution_context& ctx, unsigned num_threads = 1)
127	3298x	: num_threads_(num_threads)
128		{
129	3298x	if (!num_threads)
130	1x	throw std::logic_error("thread_pool requires at least 1 thread");
131	3300x	}
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	6592x	~thread_pool() override
142	3297x	{
143	3297x	if (!threads_.empty())
144	1x	shutdown();
145	6592x	}
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	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	196x	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	196x	std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
201	196x	capy::set_current_thread_name(name);
202
203		for (;;)
204		{
205		pool_work_item* w;
206		{
207	735x	std::unique_lock<std::mutex> lock(mutex_);
208	735x	cv_.wait(
209	927x	lock, [this] { return shutdown_ \|\| !work_queue_.empty(); });
210
211	735x	w = work_queue_.pop();
212	735x	if (!w)
213		{
214	196x	if (shutdown_)
215	392x	return;
216	✗	continue;
217		}
218	735x	}
219	539x	w->func_(w);
220	539x	}
221		}
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	544x	thread_pool::start_workers() noexcept
227		{
228	544x	if (!threads_.empty())
229	346x	return {};
230	198x	std::error_code ec;
231		try
232		{
233	198x	threads_.reserve(num_threads_);
234	393x	for (unsigned i = 0; i < num_threads_; ++i)
235	396x	threads_.emplace_back([this, i] { worker_loop(i + 1); });
236		}
237	5x	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	3x	ec = e.code();
244	3x	}
245	2x	catch (...)
246		{
247	2x	ec = std::make_error_code(std::errc::resource_unavailable_try_again);
248	2x	}
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	198x	if (!threads_.empty())
255	193x	return {};
256	5x	return ec;
257		}
258
259		inline std::error_code
260	555x	thread_pool::post(pool_work_item* w) noexcept
261		{
262		{
263	555x	std::lock_guard<std::mutex> lock(mutex_);
264	555x	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	544x	if (auto ec = start_workers())
270	5x	return ec;
271	539x	work_queue_.push(w);
272	555x	}
273	539x	cv_.notify_one();
274	539x	return {};
275		}
276
277		inline void
278	3307x	thread_pool::shutdown()
279		{
280		{
281	3307x	std::lock_guard<std::mutex> lock(mutex_);
282	3307x	shutdown_ = true;
283	3307x	}
284	3307x	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	3503x	for (auto& t : threads_)
291		{
292	196x	if (t.joinable())
293	196x	t.join();
294		}
295	3307x	threads_.clear();
296
297		{
298	3307x	std::lock_guard<std::mutex> lock(mutex_);
299	3307x	while (work_queue_.pop())
300		;
301	3307x	}
302	3307x	}
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		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	8287x	explicit thread_pool_ref(capy::execution_context& ctx) noexcept : ctx_(ctx)
338		{
339	8287x	}
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	531x	thread_pool& get()
363		{
364	531x	auto* p = pool_.load(std::memory_order_acquire);
365	531x	if (!p)
366		{
367	197x	p = &ctx_.use_service<thread_pool>();
368	197x	pool_.store(p, std::memory_order_release);
369		}
370	531x	return *p;
371		}
372		};
373
374		} // namespace boost::corosio::detail
375
376		#endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
377