src/corosio/src/io_context.cpp

100.0% Lines (74 / 74) 100.0% Functions (15 / 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_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	1305x	epoll_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
44		{
45	2605x	auto& sched = ctx.make_service<detail::epoll_scheduler>(
46	1305x	static_cast<int>(concurrency_hint));
47	1300x	return sched;
48		}
49		#endif
50
51		#if BOOST_COROSIO_HAS_SELECT
52		detail::scheduler&
53	950x	select_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
54		{
55	1893x	auto& sched = ctx.make_service<detail::select_scheduler>(
56	950x	static_cast<int>(concurrency_hint));
57	943x	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		uring_t::construct(capy::execution_context& ctx, unsigned concurrency_hint)
84		{
85		auto& sched = ctx.make_service<detail::uring_scheduler>(
86		static_cast<int>(concurrency_hint));
87		return sched;
88		}
89		#endif
90
91		namespace {
92
93		// Reject options that construct() would otherwise act on.
94		void
95	46x	check_options([[maybe_unused]] io_context_options const& opts)
96		{
97		#if BOOST_COROSIO_POSIX
98	46x	if (opts.thread_pool_size < 1)
99	2x	throw std::invalid_argument("thread_pool_size must be at least 1");
100		#endif
101	44x	}
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	2243x	create_thread_pool(
117		capy::execution_context& ctx,
118		[[maybe_unused]] io_context_options const& opts)
119		{
120		#if BOOST_COROSIO_POSIX
121	2243x	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	2243x	}
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	2243x	make_threading_config(io_context_options const& opts)
133		{
134	2243x	detail::scheduler::threading_config cfg;
135	2243x	cfg.scheduler_locking = opts.locking != locking_mode::unsafe;
136	2243x	cfg.reactor_io_locking = opts.locking == locking_mode::safe;
137	2243x	cfg.one_thread = opts.locking != locking_mode::safe;
138	2243x	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	44x	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	44x	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	44x	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	44x	io_context_options defaults;
162	44x	bool budget_at_defaults =
163	78x	opts.inline_budget_initial == defaults.inline_budget_initial &&
164	61x	opts.inline_budget_max == defaults.inline_budget_max &&
165	17x	opts.unassisted_budget == defaults.unassisted_budget;
166
167	44x	unsigned init = opts.inline_budget_initial;
168	44x	unsigned max = opts.inline_budget_max;
169	44x	unsigned ua = opts.unassisted_budget;
170
171	44x	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	44x	reactor->configure_reactor(opts.max_events_per_poll, init, max, ua);
180		}
181		#endif
182
183		#if BOOST_COROSIO_HAS_URING
184		if (auto* uring_sched = dynamic_cast<detail::uring_scheduler*>(&sched))
185		{
186		if (opts.enable_sqpoll)
187		uring_sched->configure_sqpoll(
188		true, opts.sq_thread_idle_ms, opts.sq_thread_cpu);
189		}
190		#endif
191	40x	}
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	2239x	finish_construction([[maybe_unused]] detail::scheduler& sched)
199		{
200		#if BOOST_COROSIO_HAS_URING
201		if (auto* uring_sched = dynamic_cast<detail::uring_scheduler*>(&sched))
202		uring_sched->init_ring();
203		#endif
204	2239x	}
205
206		detail::scheduler&
207	367x	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	367x	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	351x	io_context::io_context()
223	351x	: io_context(std::max(1u, std::thread::hardware_concurrency()))
224		{
225	351x	}
226
227	358x	io_context::io_context(unsigned concurrency_hint)
228		: capy::execution_context(this)
229	358x	, 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	358x	apply_threading_(io_context_options{});
234	358x	}
235
236	11x	io_context::io_context(
237	11x	io_context_options const& opts_in, unsigned concurrency_hint)
238		: capy::execution_context(this)
239	11x	, sched_(nullptr)
240		{
241	11x	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	9x	detail::effective_concurrency_hint(opts_in, concurrency_hint);
246	9x	sched_ = &construct_default(*this, eff);
247	9x	apply_options_post_(opts_in, eff);
248	11x	}
249
250		void
251	46x	io_context::apply_options_pre_(io_context_options const& opts)
252		{
253	46x	check_options(opts);
254	44x	}
255
256		void
257	44x	io_context::apply_options_post_(
258		io_context_options const& opts_in, unsigned concurrency_hint)
259		{
260	44x	create_thread_pool(*this, opts_in);
261	44x	apply_scheduler_options(*sched_, opts_in, concurrency_hint);
262	40x	finish_construction(*sched_);
263	40x	}
264
265		void
266	2199x	io_context::apply_threading_(io_context_options const& opts_in)
267		{
268	2199x	create_thread_pool(*this, opts_in);
269	2199x	sched_->configure_threading(make_threading_config(opts_in));
270	2199x	finish_construction(*sched_);
271	2199x	}
272
273	2239x	io_context::~io_context()
274		{
275	2239x	shutdown();
276	2239x	destroy();
277	2239x	}
278
279		} // namespace boost::corosio
280