src/corosio/src/io_context.cpp

100.0% Lines (80 / 80) 100.0% Functions (15 / 15)

[ ] prev/next file

io_context.cpp

f(x) Functions (15)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
3		// Copyright (c) 2026 Michael Vandeberg
4		//
5		// Distributed under the Boost Software License, Version 1.0. (See accompanying
6		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7		//
8		// Official repository: https://github.com/cppalliance/corosio
9		//
10
11		#include <boost/corosio/io_context.hpp>
12		#include <boost/corosio/backend.hpp>
13		#include <boost/corosio/detail/thread_pool.hpp>
14
15		#include <algorithm>
16		#include <limits>
17		#include <stdexcept>
18		#include <thread>
19
20		#if BOOST_COROSIO_HAS_EPOLL
21		#include <boost/corosio/native/detail/epoll/epoll_scheduler.hpp>
22		#include <sys/epoll.h>
23		#endif
24
25		#if BOOST_COROSIO_HAS_SELECT
26		#include <boost/corosio/native/detail/select/select_scheduler.hpp>
27		#endif
28
29		#if BOOST_COROSIO_HAS_KQUEUE
30		#include <boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp>
31		#endif
32
33		#if BOOST_COROSIO_HAS_URING
34		#include <boost/corosio/native/detail/uring/uring_scheduler.hpp>
35		#endif
36
37		#if BOOST_COROSIO_HAS_IOCP
38		#include <boost/corosio/native/detail/iocp/win_scheduler.hpp>
39		#endif
40
41		namespace boost::corosio {
42
43		#if BOOST_COROSIO_HAS_EPOLL
44		detail::scheduler&
45	2525x	epoll_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
46		{
47	5041x	auto& sched = ctx.make_service<detail::epoll_scheduler>(
48	2525x	static_cast<int>(concurrency_hint));
49	2516x	return sched;
50		}
51		#endif
52
53		#if BOOST_COROSIO_HAS_SELECT
54		detail::scheduler&
55	1989x	select_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
56		{
57	3971x	auto& sched = ctx.make_service<detail::select_scheduler>(
58	1989x	static_cast<int>(concurrency_hint));
59	1982x	return sched;
60		}
61		#endif
62
63		#if BOOST_COROSIO_HAS_KQUEUE
64		detail::scheduler&
65		kqueue_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
66		{
67		auto& sched = ctx.make_service<detail::kqueue_scheduler>(
68		static_cast<int>(concurrency_hint));
69		return sched;
70		}
71		#endif
72
73		#if BOOST_COROSIO_HAS_IOCP
74		detail::scheduler&
75		iocp_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
76		{
77		auto& sched = ctx.make_service<detail::win_scheduler>(
78		static_cast<int>(concurrency_hint));
79		return sched;
80		}
81		#endif
82
83		#if BOOST_COROSIO_HAS_URING
84		detail::scheduler&
85		uring_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
86		{
87		auto& sched = ctx.make_service<detail::uring_scheduler>(
88		static_cast<int>(concurrency_hint));
89		return sched;
90		}
91		#endif
92
93		namespace {
94
95		// Reject options that construct() would otherwise act on.
96		void
97	64x	check_options([[maybe_unused]] io_context_options const& opts)
98		{
99		#if BOOST_COROSIO_POSIX
100	64x	if (opts.thread_pool_size < 1)
101	2x	throw std::invalid_argument("thread_pool_size must be at least 1");
102		#endif
103		#if BOOST_COROSIO_HAS_EPOLL \|\| BOOST_COROSIO_HAS_KQUEUE \|\| \
104		BOOST_COROSIO_HAS_SELECT
105		// The reactor narrows these to int.
106	62x	constexpr auto int_max =
107		static_cast<unsigned>((std::numeric_limits<int>::max)());
108		#if BOOST_COROSIO_HAS_EPOLL
109		// Larger than epoll_wait accepts, which would fail every wait.
110	62x	constexpr auto max_events =
111		static_cast<unsigned>(int_max / sizeof(epoll_event));
112	62x	if (opts.max_events_per_poll < 1 \|\| opts.max_events_per_poll > max_events)
113		throw std::invalid_argument(
114		"max_events_per_poll must be in "
115	8x	"[1, INT_MAX / sizeof(epoll_event)]");
116		#else
117		if (opts.max_events_per_poll < 1 \|\| opts.max_events_per_poll > int_max)
118		throw std::invalid_argument(
119		"max_events_per_poll must be in [1, INT_MAX]");
120		#endif
121	54x	if (opts.inline_budget_max > int_max)
122		throw std::invalid_argument(
123	4x	"inline_budget_max must be in [0, INT_MAX]");
124		#endif
125	50x	}
126
127		// Create the shared pool that runs blocking file and DNS work. Runs
128		// after construct() so the pool is newer than the scheduler its work
129		// items post completions to: services shut down newest first, and the
130		// pool must join its workers while that scheduler can still drain what
131		// the last of them posted. Only the service is built here; its workers
132		// wait for a first post, so a context that hands off no blocking work
133		// carries no thread for the pool it holds.
134		//
135		// Every io_context constructor has to reach here, and reach it before
136		// anything can call thread_pool_ref::get(): that is what keeps the
137		// binding from ever constructing a pool on an initiator's thread, and
138		// make_service throws on a duplicate if get() got there first.
139		void
140	4498x	create_thread_pool(
141		capy::execution_context& ctx,
142		[[maybe_unused]] io_context_options const& opts)
143		{
144		#if BOOST_COROSIO_POSIX
145	4498x	ctx.make_service<detail::thread_pool>(opts.thread_pool_size);
146		#else
147		// thread_pool_size is a POSIX file-service option; the IOCP
148		// backend uses the pool for DNS alone.
149		ctx.make_service<detail::thread_pool>();
150		#endif
151	4498x	}
152
153		// Map the locking tier to the scheduler's threading facilities. one_thread is
154		// set only for the lockless tiers, where a single run thread is guaranteed.
155		detail::scheduler::threading_config
156	4498x	make_threading_config(io_context_options const& opts)
157		{
158	4498x	detail::scheduler::threading_config cfg;
159	4498x	cfg.scheduler_locking = opts.locking != locking_mode::unsafe;
160	4498x	cfg.reactor_io_locking = opts.locking == locking_mode::safe;
161	4498x	cfg.one_thread = opts.locking != locking_mode::safe;
162	4498x	return cfg;
163		}
164
165		// Apply runtime tuning after construction. `concurrency_hint` is the effective
166		// hint (normalized to 1 for lockless tiers). Budget heuristic: with default
167		// budgets and hint > 1, disable the inline-completion fast path so multi-thread
168		// runs post everything for cross-thread work-stealing.
169		void
170	50x	apply_scheduler_options(
171		[[maybe_unused]] detail::scheduler& sched,
172		[[maybe_unused]] io_context_options const& opts,
173		[[maybe_unused]] unsigned concurrency_hint)
174		{
175	50x	sched.configure_threading(make_threading_config(opts));
176
177		#if BOOST_COROSIO_HAS_EPOLL \|\| BOOST_COROSIO_HAS_KQUEUE \|\| \
178		BOOST_COROSIO_HAS_SELECT
179		// dynamic_cast — when io_uring is also linked, the runtime probe may
180		// have selected uring_scheduler instead of a reactor_scheduler.
181	50x	if (auto* reactor = dynamic_cast<detail::reactor_scheduler*>(&sched))
182		{
183		// Detect "user kept the defaults" by comparing all three to the
184		// io_context-options-defined struct defaults.
185	50x	io_context_options defaults;
186	50x	bool budget_at_defaults =
187	90x	opts.inline_budget_initial == defaults.inline_budget_initial &&
188	71x	opts.inline_budget_max == defaults.inline_budget_max &&
189	21x	opts.unassisted_budget == defaults.unassisted_budget;
190
191	50x	unsigned init = opts.inline_budget_initial;
192	50x	unsigned max = opts.inline_budget_max;
193	50x	unsigned ua = opts.unassisted_budget;
194
195	50x	if (budget_at_defaults && concurrency_hint > 1)
196		{
197		// Multi-thread default: disable budget (post-everything).
198	3x	init = 0;
199	3x	max = 0;
200	3x	ua = 0;
201		}
202
203	50x	reactor->configure_reactor(opts.max_events_per_poll, init, max, ua);
204		}
205		#endif
206
207		#if BOOST_COROSIO_HAS_URING
208		if (auto* uring_sched = dynamic_cast<detail::uring_scheduler*>(&sched))
209		{
210		if (opts.enable_sqpoll)
211		uring_sched->configure_sqpoll(
212		true, opts.sq_thread_idle_ms, opts.sq_thread_cpu);
213		}
214		#endif
215	50x	}
216
217		// Bring up backend infrastructure whose setup depends on the options
218		// applied above. Runs last in every constructor: an io_context that
219		// constructs is usable, so a kernel that refuses the infrastructure is
220		// reported from the constructor and not from the first operation.
221		void
222	4498x	finish_construction([[maybe_unused]] detail::scheduler& sched)
223		{
224		#if BOOST_COROSIO_HAS_URING
225		if (auto* uring_sched = dynamic_cast<detail::uring_scheduler*>(&sched))
226		uring_sched->init_ring();
227		#endif
228	4498x	}
229
230		detail::scheduler&
231	539x	construct_default(capy::execution_context& ctx, unsigned concurrency_hint)
232		{
233		#if BOOST_COROSIO_HAS_IOCP
234		return iocp_t::construct(ctx, concurrency_hint);
235		#elif BOOST_COROSIO_HAS_EPOLL
236	539x	return epoll_t::construct(ctx, concurrency_hint);
237		#elif BOOST_COROSIO_HAS_KQUEUE
238		return kqueue_t::construct(ctx, concurrency_hint);
239		#elif BOOST_COROSIO_HAS_SELECT
240		return select_t::construct(ctx, concurrency_hint);
241		#endif
242		}
243
244		} // anonymous namespace
245
246	523x	io_context::io_context()
247	523x	: io_context(std::max(1u, std::thread::hardware_concurrency()))
248		{
249	523x	}
250
251	530x	io_context::io_context(unsigned concurrency_hint)
252		: capy::execution_context(this)
253	530x	, sched_(&construct_default(*this, concurrency_hint))
254		{
255		// Threading config only; the plain path leaves the reactor budget at its
256		// defaults (no options-ctor budget heuristic).
257	530x	apply_threading_(io_context_options{});
258	530x	}
259
260	11x	io_context::io_context(
261	11x	io_context_options const& opts_in, unsigned concurrency_hint)
262		: capy::execution_context(this)
263	11x	, sched_(nullptr)
264		{
265	11x	apply_options_pre_(opts_in);
266		// Computed before construct_default so IOCP's completion port is created
267		// with the effective concurrency.
268		unsigned const eff =
269	9x	detail::effective_concurrency_hint(opts_in, concurrency_hint);
270	9x	sched_ = &construct_default(*this, eff);
271	9x	apply_options_post_(opts_in, eff);
272	11x	}
273
274		void
275	64x	io_context::apply_options_pre_(io_context_options const& opts)
276		{
277	64x	check_options(opts);
278	50x	}
279
280		void
281	50x	io_context::apply_options_post_(
282		io_context_options const& opts_in, unsigned concurrency_hint)
283		{
284	50x	create_thread_pool(*this, opts_in);
285	50x	apply_scheduler_options(*sched_, opts_in, concurrency_hint);
286	50x	finish_construction(*sched_);
287	50x	}
288
289		void
290	4448x	io_context::apply_threading_(io_context_options const& opts_in)
291		{
292	4448x	create_thread_pool(*this, opts_in);
293	4448x	sched_->configure_threading(make_threading_config(opts_in));
294	4448x	finish_construction(*sched_);
295	4448x	}
296
297	4498x	io_context::~io_context()
298		{
299	4498x	shutdown();
300	4498x	destroy();
301	4498x	}
302
303		} // namespace boost::corosio
304