include/boost/corosio/detail/thread_pool.hpp

100.0% Lines (79 / 79) 100.0% Functions (11 / 11)

[ ] prev/next file

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		/** 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		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		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	4419x	explicit thread_pool(
137		[[maybe_unused]] capy::execution_context& ctx, unsigned num_threads = 1)
138	4419x	: num_threads_(num_threads)
139		{
140	4419x	if (!num_threads)
141	1x	throw std::logic_error("thread_pool requires at least 1 thread");
142	4421x	}
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	8833x	~thread_pool() override
152	4418x	{
153	4418x	if (!threads_.empty())
154	1x	shutdown();
155	8833x	}
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	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	735x	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	735x	std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
212	735x	capy::set_current_thread_name(name);
213
214		for (;;)
215		{
216		pool_work_item* w;
217		{
218	1997x	std::unique_lock<std::mutex> lock(mutex_);
219	1997x	cv_.wait(
220	2981x	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	1997x	if (shutdown_)
225	1470x	return;
226	1262x	w = work_queue_.pop();
227	1997x	}
228	1262x	w->func_(w);
229	1262x	}
230		}
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	1314x	thread_pool::start_workers() noexcept
236		{
237	1314x	if (!threads_.empty())
238	578x	return {};
239	736x	std::error_code ec;
240		try
241		{
242	736x	threads_.reserve(num_threads_);
243	1470x	for (unsigned i = 0; i < num_threads_; ++i)
244	1473x	threads_.emplace_back([this, i] { worker_loop(i + 1); });
245		}
246	4x	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	2x	ec = e.code();
253	2x	}
254	2x	catch (...)
255		{
256	2x	ec = std::make_error_code(std::errc::resource_unavailable_try_again);
257	2x	}
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	736x	if (!threads_.empty())
264	732x	return {};
265	4x	return ec;
266		}
267
268		inline std::error_code
269	1336x	thread_pool::post(pool_work_item* w) noexcept
270		{
271		{
272	1336x	std::lock_guard<std::mutex> lock(mutex_);
273	1336x	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	1314x	if (auto ec = start_workers())
279	4x	return ec;
280	1310x	work_queue_.push(w);
281	1336x	}
282	1310x	cv_.notify_one();
283	1310x	return {};
284		}
285
286		inline void
287	4436x	thread_pool::shutdown()
288		{
289		{
290	4436x	std::lock_guard<std::mutex> lock(mutex_);
291	4436x	shutdown_ = true;
292	4436x	}
293	4436x	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	5171x	for (auto& t : threads_)
300		{
301	735x	if (t.joinable())
302	735x	t.join();
303		}
304	4436x	threads_.clear();
305
306	4436x	intrusive_queue<pool_work_item> leftover;
307		{
308	4436x	std::lock_guard<std::mutex> lock(mutex_);
309	4436x	leftover.splice(work_queue_);
310	4436x	}
311		// Unlocked: an abandoned coroutine's unwind can reach post()
312		// through the objects its frame owned.
313	4484x	while (auto* w = leftover.pop())
314		{
315	48x	if (w->destroy_)
316	28x	w->destroy_(w);
317		else
318	20x	w->func_(w);
319	48x	}
320	4436x	}
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		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	907x	explicit thread_pool_ref(capy::execution_context& ctx) noexcept : ctx_(ctx)
356		{
357	907x	}
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	1263x	thread_pool& get()
380		{
381	1263x	auto* p = pool_.load(std::memory_order_acquire);
382	1263x	if (!p)
383		{
384	735x	p = &ctx_.use_service<thread_pool>();
385	735x	pool_.store(p, std::memory_order_release);
386		}
387	1263x	return *p;
388		}
389		};
390
391		} // namespace boost::corosio::detail
392
393		#endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
394