src/corosio/src/io_context.cpp

96.9% Lines (95/98) 100.0% List of functions (14/14)

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

io_context.cpp

f(x) Functions (14)

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_types.hpp>
21		#endif
22
23		#if BOOST_COROSIO_HAS_SELECT
24		#include <boost/corosio/native/detail/select/select_types.hpp>
25		#endif
26
27		#if BOOST_COROSIO_HAS_KQUEUE
28		#include <boost/corosio/native/detail/kqueue/kqueue_types.hpp>
29		#endif
30
31		#if BOOST_COROSIO_HAS_IO_URING
32		#include <boost/corosio/native/detail/io_uring/io_uring_acceptor_ops.hpp>
33		#include <boost/corosio/native/detail/io_uring/io_uring_buffer.hpp>
34		#include <boost/corosio/native/detail/io_uring/io_uring_dgram_ops.hpp>
35		#include <boost/corosio/native/detail/io_uring/io_uring_multishot_acceptor.hpp>
36		#include <boost/corosio/native/detail/io_uring/io_uring_random_access_file.hpp>
37		#include <boost/corosio/native/detail/io_uring/io_uring_scheduler.hpp>
38		#include <boost/corosio/native/detail/io_uring/io_uring_stream_file.hpp>
39		#include <boost/corosio/native/detail/io_uring/io_uring_types.hpp>
40		#endif
41
42		#if BOOST_COROSIO_HAS_IOCP
43		#include <boost/corosio/native/detail/iocp/win_scheduler.hpp>
44		#include <boost/corosio/native/detail/iocp/win_tcp_acceptor_service.hpp>
45		#include <boost/corosio/native/detail/iocp/win_udp_service.hpp>
46		#include <boost/corosio/native/detail/iocp/win_local_stream_acceptor_service.hpp>
47		#include <boost/corosio/native/detail/iocp/win_signals.hpp>
48		#include <boost/corosio/native/detail/iocp/win_file_service.hpp>
49		#include <boost/corosio/native/detail/iocp/win_random_access_file_service.hpp>
50		#endif
51
52		namespace boost::corosio {
53
54		#if BOOST_COROSIO_HAS_EPOLL
55		detail::scheduler&
56	749x	epoll_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
57		{
58	1498x	auto& sched = ctx.make_service<detail::epoll_scheduler>(
59	749x	static_cast<int>(concurrency_hint));
60
61	749x	ctx.make_service<detail::epoll_tcp_service>();
62	749x	ctx.make_service<detail::epoll_tcp_acceptor_service>();
63	749x	ctx.make_service<detail::epoll_udp_service>();
64	749x	ctx.make_service<detail::epoll_local_stream_service>();
65	749x	ctx.make_service<detail::epoll_local_stream_acceptor_service>();
66	749x	ctx.make_service<detail::epoll_local_datagram_service>();
67
68	749x	return sched;
69		}
70		#endif
71
72		#if BOOST_COROSIO_HAS_SELECT
73		detail::scheduler&
74	585x	select_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
75		{
76	1170x	auto& sched = ctx.make_service<detail::select_scheduler>(
77	585x	static_cast<int>(concurrency_hint));
78
79	585x	ctx.make_service<detail::select_tcp_service>();
80	585x	ctx.make_service<detail::select_tcp_acceptor_service>();
81	585x	ctx.make_service<detail::select_udp_service>();
82	585x	ctx.make_service<detail::select_local_stream_service>();
83	585x	ctx.make_service<detail::select_local_stream_acceptor_service>();
84	585x	ctx.make_service<detail::select_local_datagram_service>();
85
86	585x	return sched;
87		}
88		#endif
89
90		#if BOOST_COROSIO_HAS_KQUEUE
91		detail::scheduler&
92		kqueue_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
93		{
94		auto& sched = ctx.make_service<detail::kqueue_scheduler>(
95		static_cast<int>(concurrency_hint));
96
97		ctx.make_service<detail::kqueue_tcp_service>();
98		ctx.make_service<detail::kqueue_tcp_acceptor_service>();
99		ctx.make_service<detail::kqueue_udp_service>();
100		ctx.make_service<detail::kqueue_local_stream_service>();
101		ctx.make_service<detail::kqueue_local_stream_acceptor_service>();
102		ctx.make_service<detail::kqueue_local_datagram_service>();
103
104		return sched;
105		}
106		#endif
107
108		#if BOOST_COROSIO_HAS_IOCP
109		detail::scheduler&
110		iocp_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
111		{
112		auto& sched = ctx.make_service<detail::win_scheduler>(
113		static_cast<int>(concurrency_hint));
114
115		auto& tcp_svc = ctx.make_service<detail::win_tcp_service>();
116		ctx.make_service<detail::win_tcp_acceptor_service>(tcp_svc);
117		ctx.make_service<detail::win_udp_service>();
118		auto& local_svc =
119		ctx.make_service<detail::win_local_stream_service>(tcp_svc);
120		ctx.make_service<detail::win_local_stream_acceptor_service>(local_svc);
121		ctx.make_service<detail::win_signals>();
122		ctx.make_service<detail::win_file_service>();
123		ctx.make_service<detail::win_random_access_file_service>();
124
125		return sched;
126		}
127		#endif
128
129		#if BOOST_COROSIO_HAS_IO_URING
130		detail::scheduler&
131	542x	io_uring_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
132		{
133	1084x	auto& sched = ctx.make_service<detail::io_uring_scheduler>(
134	542x	static_cast<int>(concurrency_hint));
135
136	542x	ctx.make_service<detail::io_uring_tcp_service>();
137	542x	ctx.make_service<detail::io_uring_tcp_acceptor_service>();
138	542x	ctx.make_service<detail::io_uring_local_stream_service>();
139	542x	ctx.make_service<detail::io_uring_local_stream_acceptor_service>();
140	542x	ctx.make_service<detail::io_uring_udp_service>();
141	542x	ctx.make_service<detail::io_uring_local_datagram_service>();
142	542x	ctx.make_service<detail::io_uring_stream_file_service>(sched);
143	542x	ctx.make_service<detail::io_uring_random_access_file_service>(sched);
144
145	542x	return sched;
146		}
147		#endif
148
149		namespace {
150
151		// Pre-create services that must exist before construct() runs.
152		void
153	32x	pre_create_services(
154		capy::execution_context& ctx,
155		io_context_options const& opts)
156		{
157		#if BOOST_COROSIO_POSIX
158	32x	if (opts.thread_pool_size < 1)
159		throw std::invalid_argument(
160	✗	"thread_pool_size must be at least 1");
161		// Pre-create the shared thread pool with the configured size.
162		// This must happen before construct() because the scheduler
163		// constructor creates file and resolver services that call
164		// get_or_create_pool(), which would create a 1-thread pool.
165	32x	if (opts.thread_pool_size != 1)
166	3x	ctx.make_service<detail::thread_pool>(opts.thread_pool_size);
167		#endif
168
169		(void)ctx;
170		(void)opts;
171	32x	}
172
173		// Map the locking tier to the scheduler's threading facilities. one_thread is
174		// set only for the lockless tiers, where a single run thread is guaranteed.
175		detail::scheduler::threading_config
176	1876x	make_threading_config(io_context_options const& opts)
177		{
178	1876x	detail::scheduler::threading_config cfg;
179	1876x	cfg.scheduler_locking = opts.locking != locking_mode::unsafe;
180	1876x	cfg.reactor_io_locking = opts.locking == locking_mode::safe;
181	1876x	cfg.one_thread = opts.locking != locking_mode::safe;
182	1876x	return cfg;
183		}
184
185		// Apply runtime tuning after construction. `concurrency_hint` is the effective
186		// hint (normalized to 1 for lockless tiers). Budget heuristic: with default
187		// budgets and hint > 1, disable the inline-completion fast path so multi-thread
188		// runs post everything for cross-thread work-stealing.
189		void
190	32x	apply_scheduler_options(
191		detail::scheduler& sched,
192		io_context_options const& opts,
193		unsigned concurrency_hint)
194		{
195	32x	sched.configure_threading(make_threading_config(opts));
196
197		#if BOOST_COROSIO_HAS_EPOLL \|\| BOOST_COROSIO_HAS_KQUEUE \|\| BOOST_COROSIO_HAS_SELECT
198		// dynamic_cast — when io_uring is also linked, the runtime probe may
199		// have selected io_uring_scheduler instead of a reactor_scheduler.
200	32x	if (auto* reactor =
201	32x	dynamic_cast<detail::reactor_scheduler*>(&sched))
202		{
203		// Detect "user kept the defaults" by comparing all three to the
204		// io_context-options-defined struct defaults.
205	27x	io_context_options defaults;
206	27x	bool budget_at_defaults =
207	42x	opts.inline_budget_initial == defaults.inline_budget_initial &&
208	42x	opts.inline_budget_max == defaults.inline_budget_max &&
209	15x	opts.unassisted_budget == defaults.unassisted_budget;
210
211	27x	unsigned init = opts.inline_budget_initial;
212	27x	unsigned max = opts.inline_budget_max;
213	27x	unsigned ua = opts.unassisted_budget;
214
215	27x	if (budget_at_defaults && concurrency_hint > 1)
216		{
217		// Multi-thread default: disable budget (post-everything).
218	4x	init = 0;
219	4x	max = 0;
220	4x	ua = 0;
221		}
222
223	27x	reactor->configure_reactor(
224	27x	opts.max_events_per_poll,
225		init,
226		max,
227		ua);
228		}
229		#endif
230
231		#if BOOST_COROSIO_HAS_IO_URING
232	30x	if (auto* uring_sched =
233	30x	dynamic_cast<detail::io_uring_scheduler*>(&sched))
234		{
235	5x	if (opts.enable_sqpoll)
236	✗	uring_sched->configure_sqpoll(
237	✗	true, opts.sq_thread_idle_ms, opts.sq_thread_cpu);
238		}
239		#endif
240
241		(void)sched;
242		(void)opts;
243		(void)concurrency_hint;
244	30x	}
245
246		detail::scheduler&
247	164x	construct_default(capy::execution_context& ctx, unsigned concurrency_hint)
248		{
249		#if BOOST_COROSIO_HAS_IOCP
250		return iocp_t::construct(ctx, concurrency_hint);
251		#elif BOOST_COROSIO_HAS_EPOLL
252	164x	return epoll_t::construct(ctx, concurrency_hint);
253		#elif BOOST_COROSIO_HAS_KQUEUE
254		return kqueue_t::construct(ctx, concurrency_hint);
255		#elif BOOST_COROSIO_HAS_SELECT
256		return select_t::construct(ctx, concurrency_hint);
257		#endif
258		}
259
260		} // anonymous namespace
261
262	152x	io_context::io_context()
263	152x	: io_context(std::max(1u, std::thread::hardware_concurrency()))
264		{
265	152x	}
266
267	155x	io_context::io_context(unsigned concurrency_hint)
268		: capy::execution_context(this)
269	155x	, sched_(&construct_default(*this, concurrency_hint))
270		{
271		// Threading config only; the plain path leaves the reactor budget at its
272		// defaults (no options-ctor budget heuristic).
273	155x	apply_threading_(io_context_options{});
274	155x	}
275
276	9x	io_context::io_context(
277		io_context_options const& opts_in,
278	9x	unsigned concurrency_hint)
279		: capy::execution_context(this)
280	9x	, sched_(nullptr)
281		{
282	9x	pre_create_services(*this, opts_in);
283		// Computed before construct_default so IOCP's completion port is created
284		// with the effective concurrency.
285		unsigned const eff =
286	9x	detail::effective_concurrency_hint(opts_in, concurrency_hint);
287	9x	sched_ = &construct_default(*this, eff);
288	9x	apply_scheduler_options(*sched_, opts_in, eff);
289	9x	}
290
291		void
292	23x	io_context::apply_options_pre_(io_context_options const& opts)
293		{
294	23x	pre_create_services(*this, opts);
295	23x	}
296
297		void
298	23x	io_context::apply_options_post_(
299		io_context_options const& opts_in,
300		unsigned concurrency_hint)
301		{
302	23x	apply_scheduler_options(*sched_, opts_in, concurrency_hint);
303	21x	}
304
305		void
306	1844x	io_context::apply_threading_(io_context_options const& opts_in)
307		{
308	1844x	sched_->configure_threading(make_threading_config(opts_in));
309	1844x	}
310
311	1874x	io_context::~io_context()
312		{
313	1874x	shutdown();
314	1874x	destroy();
315	1874x	}
316
317		} // namespace boost::corosio
318