include/boost/corosio/detail/thread_pool.hpp

88.0% Lines (66/75) 100.0% List of functions (11/11) 83.3% Branches (35/42)

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

thread_pool.hpp

f(x) Functions (11)

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			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		1250x	explicit thread_pool(
126			[[maybe_unused]] capy::execution_context& ctx, unsigned num_threads = 1)
127		1250x	: num_threads_(num_threads)
128			{
129	2/2 ✓ Branch 7 → 8 taken 1 time. ✓ Branch 7 → 11 taken 1249 times.	1250x	if (!num_threads)
130	1/1 ✓ Branch 9 → 10 taken 1 time.	1x	throw std::logic_error("thread_pool requires at least 1 thread");
131		1253x	}
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		2496x	~thread_pool() override
142		1249x	{
143	2/2 ✓ Branch 3 → 4 taken 1 time. ✓ Branch 3 → 5 taken 1248 times.	1249x	if (!threads_.empty())
144		1x	shutdown();
145		2496x	}
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		22x	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	1/1 ✓ Branch 2 → 3 taken 22 times.	22x	std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
201		22x	capy::set_current_thread_name(name);
202
203			for (;;)
204			{
205			pool_work_item* w;
206			{
207	1/1 ✓ Branch 4 → 5 taken 56 times.	56x	std::unique_lock<std::mutex> lock(mutex_);
208	1/1 ✓ Branch 5 → 6 taken 56 times.	56x	cv_.wait(
209	4/4 ✓ Branch 2 → 3 taken 34 times. ✓ Branch 2 → 5 taken 39 times. ✓ Branch 4 → 5 taken 17 times. ✓ Branch 4 → 6 taken 17 times.	73x	lock, [this] { return shutdown_ \|\| !work_queue_.empty(); });
210
211		56x	w = work_queue_.pop();
212	2/2 ✓ Branch 7 → 8 taken 22 times. ✓ Branch 7 → 11 taken 34 times.	56x	if (!w)
213			{
214	1/2 ✓ Branch 8 → 9 taken 22 times. ✗ Branch 8 → 10 not taken.	22x	if (shutdown_)
215		44x	return;
216		✗	continue;
217			}
218		56x	}
219		34x	w->func_(w);
220		34x	}
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		34x	thread_pool::start_workers() noexcept
227			{
228	2/2 ✓ Branch 3 → 4 taken 15 times. ✓ Branch 3 → 5 taken 19 times.	34x	if (!threads_.empty())
229		15x	return {};
230		19x	std::error_code ec;
231			try
232			{
233	1/1 ✓ Branch 6 → 7 taken 19 times.	19x	threads_.reserve(num_threads_);
234	2/2 ✓ Branch 10 → 8 taken 22 times. ✓ Branch 10 → 11 taken 19 times.	41x	for (unsigned i = 0; i < num_threads_; ++i)
235	1/1 ✓ Branch 8 → 9 taken 22 times.	44x	threads_.emplace_back([this, i] { worker_loop(i + 1); });
236			}
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		✗	ec = e.code();
244		✗	}
245		✗	catch (...)
246			{
247		✗	ec = std::make_error_code(std::errc::resource_unavailable_try_again);
248		✗	}
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	1/2 ✓ Branch 12 → 13 taken 19 times. ✗ Branch 12 → 14 not taken.	19x	if (!threads_.empty())
255		19x	return {};
256		✗	return ec;
257			}
258
259			inline std::error_code
260		35x	thread_pool::post(pool_work_item* w) noexcept
261			{
262			{
263		35x	std::lock_guard<std::mutex> lock(mutex_);
264	2/2 ✓ Branch 3 → 4 taken 1 time. ✓ Branch 3 → 6 taken 34 times.	35x	if (shutdown_)
265		1x	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	1/2 ✗ Branch 8 → 9 not taken. ✓ Branch 8 → 10 taken 34 times.	34x	if (auto ec = start_workers())
270		✗	return ec;
271		34x	work_queue_.push(w);
272		35x	}
273		34x	cv_.notify_one();
274		34x	return {};
275			}
276
277			inline void
278		1254x	thread_pool::shutdown()
279			{
280			{
281	1/1 ✓ Branch 2 → 3 taken 1254 times.	1254x	std::lock_guard<std::mutex> lock(mutex_);
282		1254x	shutdown_ = true;
283		1254x	}
284		1254x	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 13 → 7 taken 22 times. ✓ Branch 13 → 14 taken 1254 times.	1276x	for (auto& t : threads_)
291			{
292	1/2 ✓ Branch 9 → 10 taken 22 times. ✗ Branch 9 → 11 not taken.	22x	if (t.joinable())
293	1/1 ✓ Branch 10 → 11 taken 22 times.	22x	t.join();
294			}
295		1254x	threads_.clear();
296
297			{
298	1/1 ✓ Branch 15 → 16 taken 1254 times.	1254x	std::lock_guard<std::mutex> lock(mutex_);
299	2/3 ✗ Branch 18 → 17 not taken. ✓ Branch 18 → 19 taken 1123 times. ✓ Branch 18 → 20 taken 131 times.	1254x	while (work_queue_.pop())
300			;
301		1254x	}
302		1254x	}
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		1383x	explicit thread_pool_ref(capy::execution_context& ctx) noexcept : ctx_(ctx)
338			{
339		1383x	}
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		16x	thread_pool& get()
363			{
364		16x	auto* p = pool_.load(std::memory_order_acquire);
365	2/2 ✓ Branch 3 → 4 taken 14 times. ✓ Branch 3 → 6 taken 2 times.	16x	if (!p)
366			{
367		14x	p = &ctx_.use_service<thread_pool>();
368		14x	pool_.store(p, std::memory_order_release);
369			}
370		16x	return *p;
371			}
372			};
373
374			} // namespace boost::corosio::detail
375
376			#endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
377