src/corosio/src/io_context.cpp

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