src/corosio/src/io_context.cpp

100.0% Lines (88/0/88) 100.0% List of functions (15/0/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 <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	1014x	epoll_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
57		{
58	2023x	auto& sched = ctx.make_service<detail::epoll_scheduler>(
59	1014x	static_cast<int>(concurrency_hint));
60
61	1009x	ctx.make_service<detail::epoll_tcp_service>();
62	1009x	ctx.make_service<detail::epoll_tcp_acceptor_service>();
63	1009x	ctx.make_service<detail::epoll_udp_service>();
64	1009x	ctx.make_service<detail::epoll_local_stream_service>();
65	1009x	ctx.make_service<detail::epoll_local_stream_acceptor_service>();
66	1009x	ctx.make_service<detail::epoll_local_datagram_service>();
67
68	1009x	return sched;
69		}
70		#endif
71
72		#if BOOST_COROSIO_HAS_SELECT
73		detail::scheduler&
74	788x	select_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
75		{
76	1569x	auto& sched = ctx.make_service<detail::select_scheduler>(
77	788x	static_cast<int>(concurrency_hint));
78
79	781x	ctx.make_service<detail::select_tcp_service>();
80	781x	ctx.make_service<detail::select_tcp_acceptor_service>();
81	781x	ctx.make_service<detail::select_udp_service>();
82	781x	ctx.make_service<detail::select_local_stream_service>();
83	781x	ctx.make_service<detail::select_local_stream_acceptor_service>();
84	781x	ctx.make_service<detail::select_local_datagram_service>();
85
86	781x	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		// Reject options that construct() would otherwise act on.
152		void
153	30x	check_options([[maybe_unused]] io_context_options const& opts)
154		{
155		#if BOOST_COROSIO_POSIX
156	30x	if (opts.thread_pool_size < 1)
157		throw std::invalid_argument(
158	2x	"thread_pool_size must be at least 1");
159		#endif
160	28x	}
161
162		// Create the shared pool that runs blocking file and DNS work. Runs
163		// after construct() so the pool is newer than the scheduler its work
164		// items post completions to: services shut down newest first, and the
165		// pool must join its workers while that scheduler can still drain what
166		// the last of them posted. Only the service is built here; its workers
167		// wait for a first post, so a context that hands off no blocking work
168		// carries no thread for the pool it holds.
169		//
170		// Every io_context constructor has to reach here, and reach it before
171		// anything can call thread_pool_ref::get(): that is what keeps the
172		// binding from ever constructing a pool on an initiator's thread, and
173		// make_service throws on a duplicate if get() got there first.
174		void
175	1790x	create_thread_pool(
176		capy::execution_context& ctx,
177		[[maybe_unused]] io_context_options const& opts)
178		{
179		#if BOOST_COROSIO_POSIX
180	1790x	ctx.make_service<detail::thread_pool>(opts.thread_pool_size);
181		#else
182		// thread_pool_size is a POSIX file-service option; the IOCP
183		// backend uses the pool for DNS alone.
184		ctx.make_service<detail::thread_pool>();
185		#endif
186	1790x	}
187
188		// Map the locking tier to the scheduler's threading facilities. one_thread is
189		// set only for the lockless tiers, where a single run thread is guaranteed.
190		detail::scheduler::threading_config
191	1790x	make_threading_config(io_context_options const& opts)
192		{
193	1790x	detail::scheduler::threading_config cfg;
194	1790x	cfg.scheduler_locking = opts.locking != locking_mode::unsafe;
195	1790x	cfg.reactor_io_locking = opts.locking == locking_mode::safe;
196	1790x	cfg.one_thread = opts.locking != locking_mode::safe;
197	1790x	return cfg;
198		}
199
200		// Apply runtime tuning after construction. `concurrency_hint` is the effective
201		// hint (normalized to 1 for lockless tiers). Budget heuristic: with default
202		// budgets and hint > 1, disable the inline-completion fast path so multi-thread
203		// runs post everything for cross-thread work-stealing.
204		void
205	28x	apply_scheduler_options(
206		[[maybe_unused]] detail::scheduler& sched,
207		[[maybe_unused]] io_context_options const& opts,
208		[[maybe_unused]] unsigned concurrency_hint)
209		{
210	28x	sched.configure_threading(make_threading_config(opts));
211
212		#if BOOST_COROSIO_HAS_EPOLL \|\| BOOST_COROSIO_HAS_KQUEUE \|\| BOOST_COROSIO_HAS_SELECT
213		// dynamic_cast — when io_uring is also linked, the runtime probe may
214		// have selected io_uring_scheduler instead of a reactor_scheduler.
215	28x	if (auto* reactor =
216	28x	dynamic_cast<detail::reactor_scheduler*>(&sched))
217		{
218		// Detect "user kept the defaults" by comparing all three to the
219		// io_context-options-defined struct defaults.
220	28x	io_context_options defaults;
221	28x	bool budget_at_defaults =
222	46x	opts.inline_budget_initial == defaults.inline_budget_initial &&
223	45x	opts.inline_budget_max == defaults.inline_budget_max &&
224	17x	opts.unassisted_budget == defaults.unassisted_budget;
225
226	28x	unsigned init = opts.inline_budget_initial;
227	28x	unsigned max = opts.inline_budget_max;
228	28x	unsigned ua = opts.unassisted_budget;
229
230	28x	if (budget_at_defaults && concurrency_hint > 1)
231		{
232		// Multi-thread default: disable budget (post-everything).
233	5x	init = 0;
234	5x	max = 0;
235	5x	ua = 0;
236		}
237
238	28x	reactor->configure_reactor(
239	28x	opts.max_events_per_poll,
240		init,
241		max,
242		ua);
243		}
244		#endif
245
246		#if BOOST_COROSIO_HAS_IO_URING
247		if (auto* uring_sched =
248		dynamic_cast<detail::io_uring_scheduler*>(&sched))
249		{
250		if (opts.enable_sqpoll)
251		uring_sched->configure_sqpoll(
252		true, opts.sq_thread_idle_ms, opts.sq_thread_cpu);
253		}
254		#endif
255
256	26x	}
257
258		// Bring up backend infrastructure whose setup depends on the options
259		// applied above. Runs last in every constructor: an io_context that
260		// constructs is usable, so a kernel that refuses the infrastructure is
261		// reported from the constructor and not from the first operation.
262		void
263	1788x	finish_construction([[maybe_unused]] detail::scheduler& sched)
264		{
265		#if BOOST_COROSIO_HAS_IO_URING
266		if (auto* uring_sched =
267		dynamic_cast<detail::io_uring_scheduler*>(&sched))
268		uring_sched->init_ring();
269		#endif
270	1788x	}
271
272		detail::scheduler&
273	202x	construct_default(capy::execution_context& ctx, unsigned concurrency_hint)
274		{
275		#if BOOST_COROSIO_HAS_IOCP
276		return iocp_t::construct(ctx, concurrency_hint);
277		#elif BOOST_COROSIO_HAS_EPOLL
278	202x	return epoll_t::construct(ctx, concurrency_hint);
279		#elif BOOST_COROSIO_HAS_KQUEUE
280		return kqueue_t::construct(ctx, concurrency_hint);
281		#elif BOOST_COROSIO_HAS_SELECT
282		return select_t::construct(ctx, concurrency_hint);
283		#endif
284		}
285
286		} // anonymous namespace
287
288	186x	io_context::io_context()
289	186x	: io_context(std::max(1u, std::thread::hardware_concurrency()))
290		{
291	186x	}
292
293	193x	io_context::io_context(unsigned concurrency_hint)
294		: capy::execution_context(this)
295	193x	, sched_(&construct_default(*this, concurrency_hint))
296		{
297		// Threading config only; the plain path leaves the reactor budget at its
298		// defaults (no options-ctor budget heuristic).
299	193x	apply_threading_(io_context_options{});
300	193x	}
301
302	11x	io_context::io_context(
303		io_context_options const& opts_in,
304	11x	unsigned concurrency_hint)
305		: capy::execution_context(this)
306	11x	, sched_(nullptr)
307		{
308	11x	apply_options_pre_(opts_in);
309		// Computed before construct_default so IOCP's completion port is created
310		// with the effective concurrency.
311		unsigned const eff =
312	9x	detail::effective_concurrency_hint(opts_in, concurrency_hint);
313	9x	sched_ = &construct_default(*this, eff);
314	9x	apply_options_post_(opts_in, eff);
315	11x	}
316
317		void
318	30x	io_context::apply_options_pre_(io_context_options const& opts)
319		{
320	30x	check_options(opts);
321	28x	}
322
323		void
324	28x	io_context::apply_options_post_(
325		io_context_options const& opts_in,
326		unsigned concurrency_hint)
327		{
328	28x	create_thread_pool(*this, opts_in);
329	28x	apply_scheduler_options(*sched_, opts_in, concurrency_hint);
330	26x	finish_construction(*sched_);
331	26x	}
332
333		void
334	1762x	io_context::apply_threading_(io_context_options const& opts_in)
335		{
336	1762x	create_thread_pool(*this, opts_in);
337	1762x	sched_->configure_threading(make_threading_config(opts_in));
338	1762x	finish_construction(*sched_);
339	1762x	}
340
341	1788x	io_context::~io_context()
342		{
343	1788x	shutdown();
344	1788x	destroy();
345	1788x	}
346
347		} // namespace boost::corosio
348