include/boost/corosio/detail/scheduler.hpp

91.7% Lines (22 / 24, 1 excl) 87.5% Functions (7 / 8)

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

scheduler.hpp

f(x) Functions (8)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3		// Copyright (c) 2026 Steve Gerbino
4		// Copyright (c) 2026 Michael Vandeberg
5		//
6		// Distributed under the Boost Software License, Version 1.0. (See accompanying
7		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
8		//
9		// Official repository: https://github.com/cppalliance/corosio
10		//
11
12		#ifndef BOOST_COROSIO_DETAIL_SCHEDULER_HPP
13		#define BOOST_COROSIO_DETAIL_SCHEDULER_HPP
14
15		#include <boost/corosio/detail/config.hpp>
16		#include <boost/corosio/detail/except.hpp>
17		#include <boost/corosio/detail/thread_local_ptr.hpp>
18
19		#include <boost/capy/continuation.hpp>
20		#include <boost/capy/ex/execution_context.hpp>
21
22		#include <atomic>
23		#include <coroutine>
24		#include <cstddef>
25		#include <system_error>
26		#include <thread>
27
28		namespace boost::corosio::detail {
29
30		class scheduler_op;
31
32		/** Define the abstract interface for the event loop scheduler.
33
34		Concrete backends (epoll, IOCP, kqueue, select) derive from
35		this to implement the reactor/proactor event loop. The
36		@ref io_context delegates all scheduling operations here.
37
38		The scheduler is a registry service keyed under this abstract
39		type, so services created on first use can locate it without
40		naming a concrete backend.
41
42		@see io_context
43		*/
44		class BOOST_COROSIO_DECL scheduler
45		: public capy::execution_context::service
46		{
47		public:
48		using key_type = scheduler;
49
50	4425x	~scheduler() override = default;
51
52		/// Post a coroutine handle for deferred execution.
53		virtual void post(std::coroutine_handle<>) const = 0;
54
55		/// Post a scheduler operation for deferred execution.
56		virtual void post(scheduler_op*) const = 0;
57
58		/// Post a continuation for deferred execution (zero-allocation).
59		virtual void post(capy::continuation&) const = 0;
60
61		/// Increment the outstanding work count.
62		virtual void work_started() noexcept = 0;
63
64		/// Decrement the outstanding work count.
65		virtual void work_finished() noexcept = 0;
66
67		/// Check if the calling thread is running the event loop.
68		virtual bool running_in_this_thread() const noexcept = 0;
69
70		/// Signal the event loop to stop.
71		virtual void stop() = 0;
72
73		/// Check if the event loop has been stopped.
74		virtual bool stopped() const noexcept = 0;
75
76		/// Reset the stopped state so `run()` can be called again.
77		virtual void restart() = 0;
78
79		/// Run the event loop, blocking until all work completes.
80		virtual std::size_t run() = 0;
81
82		/// Run one handler, blocking until one completes.
83		virtual std::size_t run_one() = 0;
84
85		/** Run one handler, blocking up to @p usec microseconds.
86
87		@param usec Maximum wait time in microseconds.
88
89		@return The number of handlers executed (0 or 1).
90		*/
91		virtual std::size_t wait_one(long usec) = 0;
92
93		/// Run all ready handlers without blocking.
94		virtual std::size_t poll() = 0;
95
96		/// Run at most one ready handler without blocking.
97		virtual std::size_t poll_one() = 0;
98
99		/** Register the read end of the POSIX signal self-pipe.
100
101		Called once (by the first signal_set to register a signal) so the
102		backend's event loop watches @p read_fd for readability. When the
103		pipe becomes readable the backend drains it and calls
104		`posix_signal_service::deliver_signal` for each pending signal, in
105		normal thread context. This keeps the C signal handler
106		async-signal-safe: it only writes the signal number to the pipe.
107
108		POSIX backends override this; the default is a no-op (Windows/IOCP
109		uses synchronous C-runtime signal handling instead).
110
111		@param read_fd The read end of the global signal self-pipe.
112
113		@return The error code, empty on success.
114		*/
115		[[nodiscard]] virtual std::error_code
116	✗	register_signal_reader([[maybe_unused]] int read_fd)
117		{
118	−	return {}; // LCOV_EXCL_LINE POSIX overrides; the only caller never runs on IOCP
119		}
120
121		/// Decomposed threading configuration applied via @ref configure_threading.
122		struct threading_config
123		{
124		/// Scheduler mutex/condvar enabled. Off only in the `unsafe` tier.
125		bool scheduler_locking = true;
126		/// Per-descriptor (reactor) or ring (uring) I/O lock enabled.
127		/// Off in the `unsafe_io` and `unsafe` tiers.
128		bool reactor_io_locking = true;
129		/// A single run thread is guaranteed (a lockless tier): elide
130		/// inter-run-thread wakeups.
131		bool one_thread = false;
132		};
133
134		/// True in the fully-lockless (`unsafe`) tier. The resolver and POSIX
135		/// file services gate their `operation_not_supported` result on this.
136		virtual bool scheduler_locking_disabled() const noexcept = 0;
137
138		/// Apply @ref threading_config.
139		virtual void configure_threading(threading_config) noexcept = 0;
140
141		/** Note a canceller that claimed an operation and has yet to post it.
142
143		Called no later than the claim becomes visible: under the lock
144		that makes it, or before a lock-free claim. A service's
145		shutdown then either finds the operation or finds it counted
146		here.
147		*/
148	1165x	void cancel_posting_begin() noexcept
149		{
150	1165x	cancels_posting_.fetch_add(1, std::memory_order_relaxed);
151	1165x	}
152
153		/// End a @ref cancel_posting_begin once its post is made.
154	1165x	void cancel_posting_end() noexcept
155		{
156	1165x	cancels_posting_.fetch_sub(1, std::memory_order_release);
157	1165x	}
158
159		protected:
160		/** Wait out the cancellers counted by @ref cancel_posting_begin.
161
162		Called before the shutdown drain, so it finds their posts; the
163		services that own those operations shut down earlier and missed
164		them.
165		*/
166	4413x	void wait_cancels_posting() const noexcept
167		{
168	17450x	while (cancels_posting_.load(std::memory_order_acquire) != 0)
169	4312x	std::this_thread::yield();
170	4413x	}
171
172		private:
173		BOOST_COROSIO_MSVC_WARNING_PUSH
174		BOOST_COROSIO_MSVC_WARNING_DISABLE(4251)
175		std::atomic<std::size_t> cancels_posting_{0};
176		BOOST_COROSIO_MSVC_WARNING_POP
177		};
178
179		/** Return the scheduler registered with the context.
180
181		@throws std::logic_error If the context has no backend installed.
182		*/
183		/** The innermost scheduler running on this thread, or null.
184
185		Each backend's run guard sets it and restores the previous value,
186		so a completion can ask whether its awaiter's context is running
187		here with one compare. Nested runs of other schedulers are still
188		answered by the scheduler's own frame stack.
189		*/
190		inline thread_local_ptr<scheduler const> running_scheduler;
191
192		/** RAII guard marking a scheduler as innermost on this thread.
193
194		Restores the previous innermost scheduler on destruction.
195		*/
196		class running_scheduler_guard
197		{
198		scheduler const* prev_;
199
200		public:
201		/// Construct the guard, making @p s the innermost scheduler.
202	13155x	explicit running_scheduler_guard(scheduler const* s) noexcept
203	13155x	: prev_(running_scheduler.get())
204		{
205	13155x	running_scheduler.set(s);
206	13155x	}
207
208		/// Destroy the guard, restoring the previous innermost scheduler.
209	13155x	~running_scheduler_guard()
210		{
211	13155x	running_scheduler.set(prev_);
212	13155x	}
213
214		running_scheduler_guard(running_scheduler_guard const&) = delete;
215		running_scheduler_guard& operator=(running_scheduler_guard const&) = delete;
216		};
217
218		inline scheduler&
219	1490x	get_scheduler(capy::execution_context& ctx)
220		{
221	1490x	auto* sched = ctx.find_service<scheduler>();
222	1490x	if (!sched)
223	✗	throw_logic_error("no scheduler installed");
224	1490x	return *sched;
225		}
226
227		} // namespace boost::corosio::detail
228
229		#endif
230