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	4417x	explicit thread_pool(
137		[[maybe_unused]] capy::execution_context& ctx, unsigned num_threads = 1)
138	4417x	: num_threads_(num_threads)
139		{
140	4417x	if (!num_threads)
141	1x	throw std::logic_error("thread_pool requires at least 1 thread");
142	4419x	}
143
144		/** Destroy the pool, joining any worker `shutdown()` never reached.
145
146		The context's shutdown walk is the normal path; this only
147		catches a pool created after that walk, whose `shutdown()` is
148		therefore never called and whose joinable threads would
149		otherwise terminate the process. A pool that was never posted
150		to holds no thread and needs neither.
151		*/
152	8829x	~thread_pool() override
153	4416x	{
154	4416x	if (!threads_.empty())
155	1x	shutdown();
156	8829x	}
157
158		thread_pool(thread_pool const&) = delete;
159		thread_pool& operator=(thread_pool const&) = delete;
160
161		/** Enqueue a work item for execution on the thread pool.
162
163		The first item posted starts the workers. Zero-allocation:
164		the caller owns the work item's storage.
165
166		A refusal answers with the code the caller reports for the
167		operation it was starting, so that a system that will not give
168		the pool a thread is not mistaken for a cancellation.
169
170		@par Thread Safety
171		Safe. Racing first posts start the workers once.
172
173		@param w The work item to execute. Must remain valid until
174		its `func_` or `destroy_` has been called.
175
176		@return An empty code if the item was enqueued;
177		`capy::error::canceled` if the pool has already shut
178		down; otherwise the code of the thread the system
179		refused, which left the pool with no worker at all.
180		*/
181		[[nodiscard]] std::error_code post(pool_work_item* w) noexcept;
182
183		/** Return the number of workers the pool has started.
184
185		Zero until the first `post()`, and zero again once
186		`shutdown()` has joined them.
187
188		@par Thread Safety
189		Safe.
190		*/
191	6x	unsigned worker_count() noexcept
192		{
193	6x	std::lock_guard<std::mutex> lock(mutex_);
194	6x	return static_cast<unsigned>(threads_.size());
195	6x	}
196
197		/** Shut down the thread pool.
198
199		Signals all threads to exit once their current item is done,
200		joins them, then disposes of every item still queued without
201		running its blocking work where the item says how.
202		*/
203		void shutdown() override;
204		};
205
206		inline void
207	733x	thread_pool::worker_loop(unsigned index)
208		{
209		// Name format chosen to fit Linux's 15-char pthread limit:
210		// "tpool-svc-" (10) + up to 4 digit index leaves "tpool-svc-9999".
211		char name[16];
212	733x	std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
213	733x	capy::set_current_thread_name(name);
214
215		for (;;)
216		{
217		pool_work_item* w;
218		{
219	1978x	std::unique_lock<std::mutex> lock(mutex_);
220	1978x	cv_.wait(
221	2953x	lock, [this] { return shutdown_ \|\| !work_queue_.empty(); });
222
223		// What is still queued is left to shutdown(), which does not
224		// make blocking calls whose results nobody will read.
225	1978x	if (shutdown_)
226	1466x	return;
227	1245x	w = work_queue_.pop();
228	1978x	}
229	1245x	w->func_(w);
230	1245x	}
231		}
232
233		// Called with mutex_ held, so the workers are started once however
234		// many threads race the first post.
235		inline std::error_code
236	4728x	thread_pool::start_workers() noexcept
237		{
238	4728x	if (!threads_.empty())
239	3994x	return {};
240	734x	std::error_code ec;
241		try
242		{
243	734x	threads_.reserve(num_threads_);
244	1466x	for (unsigned i = 0; i < num_threads_; ++i)
245	1469x	threads_.emplace_back([this, i] { worker_loop(i + 1); });
246		}
247	4x	catch (std::system_error const& e)
248		{
249		// The refusal is carried out, not swallowed: a thread the
250		// system will not give is a real error and the operation that
251		// asked for it says so, rather than reporting the cancellation
252		// that belongs to a stop token.
253	2x	ec = e.code();
254	2x	}
255	2x	catch (...)
256		{
257	2x	ec = std::make_error_code(std::errc::resource_unavailable_try_again);
258	2x	}
259		// A pool short of workers still runs everything posted to it, only
260		// less of it at once, so a partial start is a start. What it does
261		// not do is come back for the rest: the size is a tuning knob, and
262		// topping it up would put a thread creation on the initiator's
263		// path for every operation after a refusal.
264	734x	if (!threads_.empty())
265	730x	return {};
266	4x	return ec;
267		}
268
269		inline std::error_code
270	4748x	thread_pool::post(pool_work_item* w) noexcept
271		{
272		{
273	4748x	std::lock_guard<std::mutex> lock(mutex_);
274	4748x	if (shutdown_)
275	20x	return capy::error::canceled;
276		// The system can refuse a thread, and an initiator has no way
277		// to throw; a refused post is the failure the callers already
278		// report through the operation they were starting.
279	4728x	if (auto ec = start_workers())
280	4x	return ec;
281	4724x	work_queue_.push(w);
282	4748x	}
283	4724x	cv_.notify_one();
284	4724x	return {};
285		}
286
287		inline void
288	4434x	thread_pool::shutdown()
289		{
290		{
291	4434x	std::lock_guard<std::mutex> lock(mutex_);
292	4434x	shutdown_ = true;
293	4434x	}
294	4434x	cv_.notify_all();
295
296		// Unlocked, though a post may add to threads_: the flag above is
297		// published under the same mutex, so a post that has not taken it
298		// yet will find it set and start nothing, and one already inside
299		// released the mutex before this thread acquired it.
300	5167x	for (auto& t : threads_)
301		{
302	733x	if (t.joinable())
303	733x	t.join();
304		}
305	4434x	threads_.clear();
306
307	4434x	intrusive_queue<pool_work_item> leftover;
308		{
309	4434x	std::lock_guard<std::mutex> lock(mutex_);
310	4434x	leftover.splice(work_queue_);
311	4434x	}
312		// Unlocked: an abandoned coroutine's unwind can reach post()
313		// through the objects its frame owned.
314	7913x	while (auto* w = leftover.pop())
315		{
316	3479x	if (w->destroy_)
317	3457x	w->destroy_(w);
318		else
319	22x	w->func_(w);
320	3479x	}
321	4434x	}
322
323		/** A reference to the context's shared thread pool, bound on first use.
324
325		Services that hand blocking work to the pool hold one of these
326		instead of a reference bound at construction. A service made before
327		the pool, as one made by the scheduler's constructor would be, would
328		otherwise create the pool itself, older than the scheduler and so
329		joined too late; binding on first use puts the pool after it.
330
331		The owning `io_context` creates the pool service during
332		construction, so by the time any operation can run the binding only
333		ever finds it. That is what keeps `get()` from constructing
334		anything on an initiator's thread, and so from throwing where an
335		initiator may not: the throwing spelling exists for a scheduler
336		driven without an `io_context`. What the service defers is its
337		workers, and those are started by `post()`, which reports a refusal
338		rather than throwing it.
339
340		@par Thread Safety
341		Distinct objects: Safe.
342		Shared objects: Safe.
343
344		@see thread_pool
345		*/
346		class thread_pool_ref
347		{
348		capy::execution_context& ctx_;
349		std::atomic<thread_pool*> pool_{nullptr};
350
351		public:
352		/** Construct a reference into the given context.
353
354		@param ctx The context whose pool is used.
355		*/
356	905x	explicit thread_pool_ref(capy::execution_context& ctx) noexcept : ctx_(ctx)
357		{
358	905x	}
359
360		thread_pool_ref(thread_pool_ref const&) = delete;
361		thread_pool_ref& operator=(thread_pool_ref const&) = delete;
362
363		/** Return the pool, creating it if this is the first use.
364
365		@pre For the throwing clauses below to be unreachable, the owning
366		context must already hold the pool service. Every `io_context`
367		constructor installs it — what waits for a first post is the
368		service's workers, not the service — so the creating branch is
369		reached only by a scheduler driven without one.
370
371		@par Exception Safety
372		Strong guarantee.
373
374		@throws std::bad_alloc If the service cannot be allocated.
375
376		@throws std::logic_error If the pool is asked for zero threads.
377
378		@return The context's shared thread pool.
379		*/
380	1248x	thread_pool& get()
381		{
382	1248x	auto* p = pool_.load(std::memory_order_acquire);
383	1248x	if (!p)
384		{
385	733x	p = &ctx_.use_service<thread_pool>();
386	733x	pool_.store(p, std::memory_order_release);
387		}
388	1248x	return *p;
389		}
390		};
391
392		} // namespace boost::corosio::detail
393
394		#endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
395