src/corosio/src/io_context.cpp

100.0% Lines (81/0/81) 100.0% List of functions (13/0/13)

[ ] prev/next file

io_context.cpp

f(x) Functions (13)

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	902x	epoll_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
57		{
58	1804x	auto& sched = ctx.make_service<detail::epoll_scheduler>(
59	902x	static_cast<int>(concurrency_hint));
60
61	902x	ctx.make_service<detail::epoll_tcp_service>();
62	902x	ctx.make_service<detail::epoll_tcp_acceptor_service>();
63	902x	ctx.make_service<detail::epoll_udp_service>();
64	902x	ctx.make_service<detail::epoll_local_stream_service>();
65	902x	ctx.make_service<detail::epoll_local_stream_acceptor_service>();
66	902x	ctx.make_service<detail::epoll_local_datagram_service>();
67
68	902x	return sched;
69		}
70		#endif
71
72		#if BOOST_COROSIO_HAS_SELECT
73		detail::scheduler&
74	701x	select_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
75		{
76	1402x	auto& sched = ctx.make_service<detail::select_scheduler>(
77	701x	static_cast<int>(concurrency_hint));
78
79	701x	ctx.make_service<detail::select_tcp_service>();
80	701x	ctx.make_service<detail::select_tcp_acceptor_service>();
81	701x	ctx.make_service<detail::select_udp_service>();
82	701x	ctx.make_service<detail::select_local_stream_service>();
83	701x	ctx.make_service<detail::select_local_stream_acceptor_service>();
84	701x	ctx.make_service<detail::select_local_datagram_service>();
85
86	701x	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		io_uring_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
132		{
133		auto& sched = ctx.make_service<detail::io_uring_scheduler>(
134		static_cast<int>(concurrency_hint));
135
136		ctx.make_service<detail::io_uring_tcp_service>();
137		ctx.make_service<detail::io_uring_tcp_acceptor_service>();
138		ctx.make_service<detail::io_uring_local_stream_service>();
139		ctx.make_service<detail::io_uring_local_stream_acceptor_service>();
140		ctx.make_service<detail::io_uring_udp_service>();
141		ctx.make_service<detail::io_uring_local_datagram_service>();
142		ctx.make_service<detail::io_uring_stream_file_service>(sched);
143		ctx.make_service<detail::io_uring_random_access_file_service>(sched);
144
145		return sched;
146		}
147		#endif
148
149		namespace {
150
151		// Pre-create services that must exist before construct() runs.
152		void
153	30x	pre_create_services(
154		[[maybe_unused]] capy::execution_context& ctx,
155		[[maybe_unused]] io_context_options const& opts)
156		{
157		#if BOOST_COROSIO_POSIX
158	30x	if (opts.thread_pool_size < 1)
159		throw std::invalid_argument(
160	2x	"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	28x	if (opts.thread_pool_size != 1)
166	2x	ctx.make_service<detail::thread_pool>(opts.thread_pool_size);
167		#endif
168
169	28x	}
170
171		// Map the locking tier to the scheduler's threading facilities. one_thread is
172		// set only for the lockless tiers, where a single run thread is guaranteed.
173		detail::scheduler::threading_config
174	1603x	make_threading_config(io_context_options const& opts)
175		{
176	1603x	detail::scheduler::threading_config cfg;
177	1603x	cfg.scheduler_locking = opts.locking != locking_mode::unsafe;
178	1603x	cfg.reactor_io_locking = opts.locking == locking_mode::safe;
179	1603x	cfg.one_thread = opts.locking != locking_mode::safe;
180	1603x	return cfg;
181		}
182
183		// Apply runtime tuning after construction. `concurrency_hint` is the effective
184		// hint (normalized to 1 for lockless tiers). Budget heuristic: with default
185		// budgets and hint > 1, disable the inline-completion fast path so multi-thread
186		// runs post everything for cross-thread work-stealing.
187		void
188	28x	apply_scheduler_options(
189		[[maybe_unused]] detail::scheduler& sched,
190		[[maybe_unused]] io_context_options const& opts,
191		[[maybe_unused]] unsigned concurrency_hint)
192		{
193	28x	sched.configure_threading(make_threading_config(opts));
194
195		#if BOOST_COROSIO_HAS_EPOLL \|\| BOOST_COROSIO_HAS_KQUEUE \|\| BOOST_COROSIO_HAS_SELECT
196		// dynamic_cast — when io_uring is also linked, the runtime probe may
197		// have selected io_uring_scheduler instead of a reactor_scheduler.
198	28x	if (auto* reactor =
199	28x	dynamic_cast<detail::reactor_scheduler*>(&sched))
200		{
201		// Detect "user kept the defaults" by comparing all three to the
202		// io_context-options-defined struct defaults.
203	28x	io_context_options defaults;
204	28x	bool budget_at_defaults =
205	46x	opts.inline_budget_initial == defaults.inline_budget_initial &&
206	45x	opts.inline_budget_max == defaults.inline_budget_max &&
207	17x	opts.unassisted_budget == defaults.unassisted_budget;
208
209	28x	unsigned init = opts.inline_budget_initial;
210	28x	unsigned max = opts.inline_budget_max;
211	28x	unsigned ua = opts.unassisted_budget;
212
213	28x	if (budget_at_defaults && concurrency_hint > 1)
214		{
215		// Multi-thread default: disable budget (post-everything).
216	5x	init = 0;
217	5x	max = 0;
218	5x	ua = 0;
219		}
220
221	28x	reactor->configure_reactor(
222	28x	opts.max_events_per_poll,
223		init,
224		max,
225		ua);
226		}
227		#endif
228
229		#if BOOST_COROSIO_HAS_IO_URING
230		if (auto* uring_sched =
231		dynamic_cast<detail::io_uring_scheduler*>(&sched))
232		{
233		if (opts.enable_sqpoll)
234		uring_sched->configure_sqpoll(
235		true, opts.sq_thread_idle_ms, opts.sq_thread_cpu);
236		}
237		#endif
238
239	26x	}
240
241		detail::scheduler&
242	198x	construct_default(capy::execution_context& ctx, unsigned concurrency_hint)
243		{
244		#if BOOST_COROSIO_HAS_IOCP
245		return iocp_t::construct(ctx, concurrency_hint);
246		#elif BOOST_COROSIO_HAS_EPOLL
247	198x	return epoll_t::construct(ctx, concurrency_hint);
248		#elif BOOST_COROSIO_HAS_KQUEUE
249		return kqueue_t::construct(ctx, concurrency_hint);
250		#elif BOOST_COROSIO_HAS_SELECT
251		return select_t::construct(ctx, concurrency_hint);
252		#endif
253		}
254
255		} // anonymous namespace
256
257	182x	io_context::io_context()
258	182x	: io_context(std::max(1u, std::thread::hardware_concurrency()))
259		{
260	182x	}
261
262	189x	io_context::io_context(unsigned concurrency_hint)
263		: capy::execution_context(this)
264	189x	, sched_(&construct_default(*this, concurrency_hint))
265		{
266		// Threading config only; the plain path leaves the reactor budget at its
267		// defaults (no options-ctor budget heuristic).
268	189x	apply_threading_(io_context_options{});
269	189x	}
270
271	11x	io_context::io_context(
272		io_context_options const& opts_in,
273	11x	unsigned concurrency_hint)
274		: capy::execution_context(this)
275	11x	, sched_(nullptr)
276		{
277	11x	pre_create_services(*this, opts_in);
278		// Computed before construct_default so IOCP's completion port is created
279		// with the effective concurrency.
280		unsigned const eff =
281	9x	detail::effective_concurrency_hint(opts_in, concurrency_hint);
282	9x	sched_ = &construct_default(*this, eff);
283	9x	apply_scheduler_options(*sched_, opts_in, eff);
284	11x	}
285
286		void
287	19x	io_context::apply_options_pre_(io_context_options const& opts)
288		{
289	19x	pre_create_services(*this, opts);
290	19x	}
291
292		void
293	19x	io_context::apply_options_post_(
294		io_context_options const& opts_in,
295		unsigned concurrency_hint)
296		{
297	19x	apply_scheduler_options(*sched_, opts_in, concurrency_hint);
298	17x	}
299
300		void
301	1575x	io_context::apply_threading_(io_context_options const& opts_in)
302		{
303	1575x	sched_->configure_threading(make_threading_config(opts_in));
304	1575x	}
305
306	1601x	io_context::~io_context()
307		{
308	1601x	shutdown();
309	1601x	destroy();
310	1601x	}
311
312		} // namespace boost::corosio
313