include/boost/corosio/detail/thread_pool.hpp

100.0% Lines (88 / 88) 82.6% Branches (38 / 46) 100.0% Functions (12 / 12, 1 excl)

[ ] 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		1932x	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		1932x	func_type func_ = nullptr;
61
62			/** Teardown for an item the pool shuts down before running.
63
64			An operation that owns a suspended coroutine sets this, so the
65			coroutine is abandoned at shutdown instead of the pool making
66			the blocking call nobody is left to wait for. When null, the
67			item is run by `func_` instead.
68			*/
69		1932x	func_type destroy_ = nullptr;
70			};
71
72			/** Shared thread pool for dispatching blocking operations.
73
74			Provides a fixed pool of reusable worker threads for operations
75			that cannot be integrated with async I/O (e.g. blocking DNS
76			calls). Registered as an `execution_context::service` so it
77			is a singleton per io_context.
78
79			The service is created with its context, but the workers start on
80			the first `post()`: a context that never opens a file and never
81			resolves a name never pays for a thread. The default thread count
82			is 1.
83
84			@par Thread Safety
85			All public member functions are thread-safe.
86
87			@par Shutdown
88			Sets a shutdown flag, notifies all threads, and joins them.
89			In-flight blocking calls complete naturally before the thread
90			exits. Items still queued are not run: each is handed to its
91			`destroy_`, or run by `func_` on the shutting-down thread when it
92			has none.
93
94			@note Create this service after the scheduler its work items post
95			completions to. Services shut down newest first, so a pool created
96			earlier joins its workers only after the scheduler has drained its
97			completion queue, and the completion the last worker posts is then
98			destroyed only when the scheduler is deleted.
99
100			@note The type is symbol-visible because services are keyed by type
101			identity: with RTTI, hidden behind a shared library boundary, a
102			module that asks for the pool would look up, and create, one of its
103			own (the no-RTTI key is a template static whose visibility follows
104			the template it is instantiated from).
105			*/
106			class BOOST_COROSIO_SYMBOL_VISIBLE thread_pool final
107			: public capy::execution_context::service
108			{
109			std::mutex mutex_;
110			std::condition_variable cv_;
111			intrusive_queue<pool_work_item> work_queue_;
112			std::vector<std::thread> threads_;
113			unsigned num_threads_;
114		4632x	bool shutdown_ = false;
115
116			void worker_loop(unsigned index);
117			std::error_code start_workers() noexcept;
118
119			public:
120			using key_type = thread_pool;
121
122			/** Construct the thread pool service.
123
124			Records the worker count. The workers themselves start on the
125			first `post()`.
126
127			@par Exception Safety
128			Strong guarantee.
129
130			@param ctx Reference to the owning execution_context.
131			@param num_threads Number of worker threads. Must be
132			at least 1.
133
134			@throws std::logic_error If `num_threads` is 0.
135			*/
136		13896x	explicit thread_pool(
137			[[maybe_unused]] capy::execution_context& ctx, unsigned num_threads = 1)
138		4632x	: num_threads_(num_threads)
139		9264x	{
140	2/2 ✓ Branch 0 taken 4631 times. ✓ Branch 1 taken 1 time.	4632x	if (!num_threads)
141	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");
142		9264x	}
143
144			/** Destroy the pool, joining any worker `shutdown()` did not.
145
146			A context shuts down the pool it owns before destroying it, or
147			at creation if the context's shutdown already began. A pool
148			constructed directly gets no such call, so this shuts it down
149			rather than leave a joinable worker to terminate the process.
150			*/
151		13893x	~thread_pool() override
152		9261x	{
153	2/2 ✓ Branch 0 taken 4631 times. ✓ Branch 1 taken 1 time.	4632x	if (!threads_.empty())
154	1/2 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken.	1x	shutdown();
155		13893x	}
156
157			thread_pool(thread_pool const&) = delete;
158			thread_pool& operator=(thread_pool const&) = delete;
159
160			/** Enqueue a work item for execution on the thread pool.
161
162			The first item posted starts the workers. Zero-allocation:
163			the caller owns the work item's storage.
164
165			A refusal answers with the code the caller reports for the
166			operation it was starting, so that a system that will not give
167			the pool a thread is not mistaken for a cancellation.
168
169			@par Thread Safety
170			Safe. Racing first posts start the workers once.
171
172			@param w The work item to execute. Must remain valid until
173			its `func_` or `destroy_` has been called.
174
175			@return An empty code if the item was enqueued;
176			`capy::error::canceled` if the pool has already shut
177			down; otherwise the code of the thread the system
178			refused, which left the pool with no worker at all.
179			*/
180			[[nodiscard]] std::error_code post(pool_work_item* w) noexcept;
181
182			/** Return the number of workers the pool has started.
183
184			Zero until the first `post()`, and zero again once
185			`shutdown()` has joined them.
186
187			@par Thread Safety
188			Safe.
189			*/
190		6x	unsigned worker_count() noexcept
191			{
192	1/2 ✓ Branch 0 taken 6 times. ✗ Branch 1 not taken.	6x	std::lock_guard<std::mutex> lock(mutex_);
193		6x	return static_cast<unsigned>(threads_.size());
194		6x	}
195
196			/** Shut down the thread pool.
197
198			Signals all threads to exit once their current item is done,
199			joins them, then disposes of every item still queued without
200			running its blocking work where the item says how.
201			*/
202			void shutdown() override;
203			};
204
205			inline void
206		682x	thread_pool::worker_loop(unsigned index)
207			{
208			// Name format chosen to fit Linux's 15-char pthread limit:
209			// "tpool-svc-" (10) + up to 4 digit index leaves "tpool-svc-9999".
210			char name[16];
211		682x	std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
212		682x	capy::set_current_thread_name(name);
213
214		1873x	for (;;)
215			{
216			pool_work_item* w;
217			{
218		1873x	std::unique_lock<std::mutex> lock(mutex_);
219	2/4 ✓ Branch 0 taken 1873 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 1873 times. ✗ Branch 3 not taken.	3746x	cv_.wait(
220	2/2 ✓ Branch 0 taken 682 times. ✓ Branch 1 taken 2110 times.	4665x	lock, [this] { return shutdown_ \|\| !work_queue_.empty(); });
221
222			// What is still queued is left to shutdown(), which does not
223			// make blocking calls whose results nobody will read.
224	2/2 ✓ Branch 0 taken 682 times. ✓ Branch 1 taken 1191 times.	1873x	if (shutdown_)
225		682x	return;
226		1191x	w = work_queue_.pop();
227		1873x	}
228		1191x	w->func_(w);
229			}
230		682x	}
231
232			// Called with mutex_ held, so the workers are started once however
233			// many threads race the first post.
234			inline std::error_code
235		1242x	thread_pool::start_workers() noexcept
236			{
237	2/2 ✓ Branch 0 taken 558 times. ✓ Branch 1 taken 684 times.	1242x	if (!threads_.empty())
238		558x	return {};
239		684x	std::error_code ec;
240			try
241			{
242	2/2 ✓ Branch 0 taken 683 times. ✓ Branch 1 taken 1 time.	684x	threads_.reserve(num_threads_);
243	2/2 ✓ Branch 0 taken 686 times. ✓ Branch 1 taken 679 times.	1365x	for (unsigned i = 0; i < num_threads_; ++i)
244	2/2 ✓ Branch 0 taken 682 times. ✓ Branch 1 taken 4 times.	1368x	threads_.emplace_back([this, i] { worker_loop(i + 1); });
245		684x	}
246			catch (std::system_error const& e)
247			{
248			// The refusal is carried out, not swallowed: a thread the
249			// system will not give is a real error and the operation that
250			// asked for it says so, rather than reporting the cancellation
251			// that belongs to a stop token.
252		1x	ec = e.code();
253		1x	}
254			catch (...)
255			{
256		4x	ec = std::make_error_code(std::errc::resource_unavailable_try_again);
257		5x	}
258			// A pool short of workers still runs everything posted to it, only
259			// less of it at once, so a partial start is a start. What it does
260			// not do is come back for the rest: the size is a tuning knob, and
261			// topping it up would put a thread creation on the initiator's
262			// path for every operation after a refusal.
263	2/2 ✓ Branch 0 taken 5 times. ✓ Branch 1 taken 679 times.	684x	if (!threads_.empty())
264		679x	return {};
265		5x	return ec;
266		1247x	}
267
268			inline std::error_code
269		1264x	thread_pool::post(pool_work_item* w) noexcept
270			{
271			{
272	1/2 ✓ Branch 0 taken 1264 times. ✗ Branch 1 not taken.	1264x	std::lock_guard<std::mutex> lock(mutex_);
273	2/2 ✓ Branch 0 taken 22 times. ✓ Branch 1 taken 1242 times.	1264x	if (shutdown_)
274		22x	return capy::error::canceled;
275			// The system can refuse a thread, and an initiator has no way
276			// to throw; a refused post is the failure the callers already
277			// report through the operation they were starting.
278	2/2 ✓ Branch 0 taken 5 times. ✓ Branch 1 taken 1237 times.	1242x	if (auto ec = start_workers())
279		5x	return ec;
280		1237x	work_queue_.push(w);
281		1264x	}
282		1237x	cv_.notify_one();
283		1237x	return {};
284		1264x	}
285
286			inline void
287		4650x	thread_pool::shutdown()
288			{
289			{
290		4650x	std::lock_guard<std::mutex> lock(mutex_);
291		4650x	shutdown_ = true;
292		4650x	}
293		4650x	cv_.notify_all();
294
295			// Unlocked, though a post may add to threads_: the flag above is
296			// published under the same mutex, so a post that has not taken it
297			// yet will find it set and start nothing, and one already inside
298			// released the mutex before this thread acquired it.
299	2/2 ✓ Branch 0 taken 682 times. ✓ Branch 1 taken 4650 times.	5332x	for (auto& t : threads_)
300			{
301	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 682 times.	682x	if (t.joinable())
302		682x	t.join();
303			}
304		4650x	threads_.clear();
305
306		4650x	intrusive_queue<pool_work_item> leftover;
307			{
308		4650x	std::lock_guard<std::mutex> lock(mutex_);
309		4650x	leftover.splice(work_queue_);
310		4650x	}
311			// Unlocked: an abandoned coroutine's unwind can reach post()
312			// through the objects its frame owned.
313	2/2 ✓ Branch 0 taken 46 times. ✓ Branch 1 taken 4650 times.	4696x	while (auto* w = leftover.pop())
314			{
315	2/2 ✓ Branch 0 taken 29 times. ✓ Branch 1 taken 17 times.	46x	if (w->destroy_)
316		29x	w->destroy_(w);
317			else
318		17x	w->func_(w);
319			}
320		4650x	}
321
322			/** A reference to the context's shared thread pool, bound on first use.
323
324			Services that hand blocking work to the pool hold one of these
325			instead of a reference bound at construction. A service made before
326			the pool, as one made by the scheduler's constructor would be, would
327			otherwise create the pool itself, older than the scheduler and so
328			joined too late; binding on first use puts the pool after it.
329
330			The owning `io_context` creates the pool service during
331			construction, so by the time any operation can run the binding only
332			ever finds it. That is what keeps `get()` from constructing
333			anything on an initiator's thread, and so from throwing where an
334			initiator may not: the throwing spelling exists for a scheduler
335			driven without an `io_context`. What the service defers is its
336			workers, and those are started by `post()`, which reports a refusal
337			rather than throwing it.
338
339			@par Thread Safety
340			Distinct objects: Safe.
341			Shared objects: Safe.
342
343			@see thread_pool
344			*/
345			class thread_pool_ref
346			{
347			capy::execution_context& ctx_;
348		850x	std::atomic<thread_pool*> pool_{nullptr};
349
350			public:
351			/** Construct a reference into the given context.
352
353			@param ctx The context whose pool is used.
354			*/
355		1700x	explicit thread_pool_ref(capy::execution_context& ctx) noexcept : ctx_(ctx)
356		850x	{
357		1700x	}
358
359			thread_pool_ref(thread_pool_ref const&) = delete;
360			thread_pool_ref& operator=(thread_pool_ref const&) = delete;
361
362			/** Return the pool, creating it if this is the first use.
363
364			@pre For the throwing clauses below to be unreachable, the owning
365			context must already hold the pool service. Every `io_context`
366			constructor installs it — what waits for a first post is the
367			service's workers, not the service — so the creating branch is
368			reached only by a scheduler driven without one.
369
370			@par Exception Safety
371			Strong guarantee.
372
373			@throws std::bad_alloc If the service cannot be allocated.
374
375			@throws std::logic_error If the pool is asked for zero threads.
376
377			@return The context's shared thread pool.
378			*/
379		1190x	thread_pool& get()
380			{
381		1190x	auto* p = pool_.load(std::memory_order_acquire);
382	2/2 ✓ Branch 0 taken 506 times. ✓ Branch 1 taken 684 times.	1190x	if (!p)
383			{
384		684x	p = &ctx_.use_service<thread_pool>();
385		684x	pool_.store(p, std::memory_order_release);
386		684x	}
387		1190x	return *p;
388			}
389			};
390
391			} // namespace boost::corosio::detail
392
393			#endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
394