src/corosio/src/io_context.cpp

97.6% Lines (82 / 84) 100.0% Functions (16 / 16)

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

io_context.cpp

f(x) Functions (16)

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