include/boost/corosio/detail/thread_pool.hpp

89.1% Lines (41/46) 100.0% List of functions (7/7)

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

f(x) Functions (7)

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/ex/execution_context.hpp>
16
17		#include <condition_variable>
18		#include <mutex>
19		#include <stdexcept>
20		#include <thread>
21		#include <vector>
22
23		namespace boost::corosio::detail {
24
25		/** Base class for thread pool work items.
26
27		Derive from this to create work that can be posted to a
28		@ref thread_pool. Uses static function pointer dispatch,
29		consistent with the IOCP `op` pattern.
30
31		@par Example
32		@code
33		struct my_work : pool_work_item
34		{
35		int* result;
36		static void execute( pool_work_item* w ) noexcept
37		{
38		auto* self = static_cast<my_work*>( w );
39		*self->result = 42;
40		}
41		};
42
43		my_work w;
44		w.func_ = &my_work::execute;
45		w.result = &r;
46		pool.post( &w );
47		@endcode
48		*/
49		struct pool_work_item : intrusive_queue<pool_work_item>::node
50		{
51		/// Static dispatch function signature.
52		using func_type = void ()(pool_work_item) noexcept;
53
54		/// Completion handler invoked by the worker thread.
55		func_type func_ = nullptr;
56		};
57
58		/** Shared thread pool for dispatching blocking operations.
59
60		Provides a fixed pool of reusable worker threads for operations
61		that cannot be integrated with async I/O (e.g. blocking DNS
62		calls). Registered as an `execution_context::service` so it
63		is a singleton per io_context.
64
65		Threads are created eagerly in the constructor. The default
66		thread count is 1.
67
68		@par Thread Safety
69		All public member functions are thread-safe.
70
71		@par Shutdown
72		Sets a shutdown flag, notifies all threads, and joins them.
73		In-flight blocking calls complete naturally before the thread
74		exits.
75		*/
76		class thread_pool final : public capy::execution_context::service
77		{
78		std::mutex mutex_;
79		std::condition_variable cv_;
80		intrusive_queue<pool_work_item> work_queue_;
81		std::vector<std::thread> threads_;
82		bool shutdown_ = false;
83
84		void worker_loop();
85
86		public:
87		using key_type = thread_pool;
88
89		/** Construct the thread pool service.
90
91		Eagerly creates all worker threads.
92
93		@par Exception Safety
94		Strong guarantee. If thread creation fails, all
95		already-created threads are shut down and joined
96		before the exception propagates.
97
98		@param ctx Reference to the owning execution_context.
99		@param num_threads Number of worker threads. Must be
100		at least 1.
101
102		@throws std::logic_error If `num_threads` is 0.
103		*/
104	536x	explicit thread_pool(capy::execution_context& ctx, unsigned num_threads = 1)
105	536x	{
106		(void)ctx;
107	536x	if (!num_threads)
108	1x	throw std::logic_error("thread_pool requires at least 1 thread");
109	535x	threads_.reserve(num_threads);
110		try
111		{
112	1073x	for (unsigned i = 0; i < num_threads; ++i)
113	1076x	threads_.emplace_back([this] { worker_loop(); });
114		}
115	✗	catch (...)
116		{
117	✗	shutdown();
118	✗	throw;
119	✗	}
120	538x	}
121
122	1069x	~thread_pool() override = default;
123
124		thread_pool(thread_pool const&) = delete;
125		thread_pool& operator=(thread_pool const&) = delete;
126
127		/** Enqueue a work item for execution on the thread pool.
128
129		Zero-allocation: the caller owns the work item's storage.
130
131		@param w The work item to execute. Must remain valid until
132		its `func_` has been called.
133
134		@return `true` if the item was enqueued, `false` if the
135		pool has already shut down.
136		*/
137		bool post(pool_work_item* w) noexcept;
138
139		/** Shut down the thread pool.
140
141		Signals all threads to exit after draining any
142		remaining queued work, then joins them.
143		*/
144		void shutdown() override;
145		};
146
147		inline void
148	538x	thread_pool::worker_loop()
149		{
150		for (;;)
151		{
152		pool_work_item* w;
153		{
154	579x	std::unique_lock<std::mutex> lock(mutex_);
155	579x	cv_.wait(
156	1037x	lock, [this] { return shutdown_ \|\| !work_queue_.empty(); });
157
158	579x	w = work_queue_.pop();
159	579x	if (!w)
160		{
161	538x	if (shutdown_)
162	1076x	return;
163	✗	continue;
164		}
165	579x	}
166	41x	w->func_(w);
167	41x	}
168		}
169
170		inline bool
171	42x	thread_pool::post(pool_work_item* w) noexcept
172		{
173		{
174	42x	std::lock_guard<std::mutex> lock(mutex_);
175	42x	if (shutdown_)
176	1x	return false;
177	41x	work_queue_.push(w);
178	42x	}
179	41x	cv_.notify_one();
180	41x	return true;
181		}
182
183		inline void
184	539x	thread_pool::shutdown()
185		{
186		{
187	539x	std::lock_guard<std::mutex> lock(mutex_);
188	539x	shutdown_ = true;
189	539x	}
190	539x	cv_.notify_all();
191
192	1077x	for (auto& t : threads_)
193		{
194	538x	if (t.joinable())
195	538x	t.join();
196		}
197	539x	threads_.clear();
198
199		{
200	539x	std::lock_guard<std::mutex> lock(mutex_);
201	539x	while (work_queue_.pop())
202		;
203	539x	}
204	539x	}
205
206		} // namespace boost::corosio::detail
207
208		#endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
209