include/boost/corosio/detail/scheduler.hpp

93.9% Lines (31 / 33, 1 excl) 90.9% Functions (10 / 11)

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

scheduler.hpp

f(x) Functions (11)

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		class timer_service;
32
33		/** Define the abstract interface for the event loop scheduler.
34
35		Concrete backends (epoll, IOCP, kqueue, select) derive from
36		this to implement the reactor/proactor event loop. The
37		@ref io_context delegates all scheduling operations here.
38
39		The scheduler is a registry service keyed under this abstract
40		type, so services created on first use can locate it without
41		naming a concrete backend.
42
43		@see io_context
44		*/
45		class BOOST_COROSIO_DECL scheduler
46		: public capy::execution_context::service
47		{
48		public:
49		using key_type = scheduler;
50
51	4442x	~scheduler() override = default;
52
53		/// Post a coroutine handle for deferred execution.
54		virtual void post(std::coroutine_handle<>) const = 0;
55
56		/// Post a scheduler operation for deferred execution.
57		virtual void post(scheduler_op*) const = 0;
58
59		/// Post a continuation for deferred execution (zero-allocation).
60		virtual void post(capy::continuation&) const = 0;
61
62		/// Increment the outstanding work count.
63		virtual void work_started() noexcept = 0;
64
65		/// Decrement the outstanding work count.
66		virtual void work_finished() noexcept = 0;
67
68		/// Check if the calling thread is running the event loop.
69		virtual bool running_in_this_thread() const noexcept = 0;
70
71		/// Signal the event loop to stop.
72		virtual void stop() = 0;
73
74		/// Check if the event loop has been stopped.
75		virtual bool stopped() const noexcept = 0;
76
77		/// Reset the stopped state so `run()` can be called again.
78		virtual void restart() = 0;
79
80		/// Run the event loop, blocking until all work completes.
81		virtual std::size_t run() = 0;
82
83		/// Run one handler, blocking until one completes.
84		virtual std::size_t run_one() = 0;
85
86		/** Run one handler, blocking up to @p usec microseconds.
87
88		@param usec Maximum wait time in microseconds.
89
90		@return The number of handlers executed (0 or 1).
91		*/
92		virtual std::size_t wait_one(long usec) = 0;
93
94		/// Run all ready handlers without blocking.
95		virtual std::size_t poll() = 0;
96
97		/// Run at most one ready handler without blocking.
98		virtual std::size_t poll_one() = 0;
99
100		/** Register the read end of the POSIX signal self-pipe.
101
102		Called once (by the first signal_set to register a signal) so the
103		backend's event loop watches @p read_fd for readability. When the
104		pipe becomes readable the backend drains it and calls
105		`posix_signal_service::deliver_signal` for each pending signal, in
106		normal thread context. This keeps the C signal handler
107		async-signal-safe: it only writes the signal number to the pipe.
108
109		POSIX backends override this; the default is a no-op (Windows/IOCP
110		uses synchronous C-runtime signal handling instead).
111
112		@param read_fd The read end of the global signal self-pipe.
113
114		@return The error code, empty on success.
115		*/
116		[[nodiscard]] virtual std::error_code
117	✗	register_signal_reader([[maybe_unused]] int read_fd)
118		{
119	−	return {}; // LCOV_EXCL_LINE POSIX overrides; the only caller never runs on IOCP
120		}
121
122		/// Decomposed threading configuration applied via @ref configure_threading.
123		struct threading_config
124		{
125		/// Scheduler mutex/condvar enabled. Off only in the `unsafe` tier.
126		bool scheduler_locking = true;
127		/// Per-descriptor (reactor) or ring (uring) I/O lock enabled.
128		/// Off in the `unsafe_io` and `unsafe` tiers.
129		bool reactor_io_locking = true;
130		/// A single run thread is guaranteed (a lockless tier): elide
131		/// inter-run-thread wakeups.
132		bool one_thread = false;
133		};
134
135		/// True in the fully-lockless (`unsafe`) tier. The resolver and POSIX
136		/// file services gate their `operation_not_supported` result on this.
137		virtual bool scheduler_locking_disabled() const noexcept = 0;
138
139		/// Apply @ref threading_config.
140		virtual void configure_threading(threading_config) noexcept = 0;
141
142		/** Note a canceller that claimed an operation and has yet to post it.
143
144		Called no later than the claim becomes visible: under the lock
145		that makes it, or before a lock-free claim. A service's
146		shutdown then either finds the operation or finds it counted
147		here.
148		*/
149	983x	void cancel_posting_begin() noexcept
150		{
151	983x	cancels_posting_.fetch_add(1, std::memory_order_relaxed);
152	983x	}
153
154		/// End a @ref cancel_posting_begin once its post is made.
155	983x	void cancel_posting_end() noexcept
156		{
157	983x	cancels_posting_.fetch_sub(1, std::memory_order_release);
158	983x	}
159
160		/** Attach the context's timer service.
161
162		The timer service is created with the context's first timer,
163		possibly on a thread other than the one running the loop; until
164		it is attached the loop has no timers to wait on.
165
166		@par Preconditions
167		No timer service is attached.
168
169		@par Exception Safety
170		Strong guarantee: on a throw, nothing is attached.
171
172		@param svc The service to attach.
173		*/
174	927x	void attach_timer_service(timer_service& svc)
175		{
176	927x	on_timer_service_attached(svc);
177	927x	timer_svc_.store(&svc, std::memory_order_release);
178	927x	}
179
180		/// Detach the timer service as it is destroyed.
181	927x	void detach_timer_service() noexcept
182		{
183	927x	timer_svc_.store(nullptr, std::memory_order_release);
184	927x	}
185
186		protected:
187		/// Return the attached timer service, or null before the first timer.
188	329963x	timer_service* timers() const noexcept
189		{
190	329963x	return timer_svc_.load(std::memory_order_acquire);
191		}
192
193		/** Prepare to wait on a timer service about to be attached.
194
195		Called before the service is published to the loop, so the loop
196		never sees a service it cannot yet wait on. A scheduler that
197		cannot get a kernel timer falls back to bounding its wait by the
198		nearest expiry rather than throwing: the first timer is often
199		made inside an `await_suspend`, where a throw could only
200		terminate.
201
202		@par Exception Safety
203		Strong guarantee: a throw leaves the scheduler as it was, and
204		the next timer calls this again.
205
206		@param svc The service being attached.
207		*/
208		virtual void on_timer_service_attached(timer_service& svc) = 0;
209
210		/** Wait out the cancellers counted by @ref cancel_posting_begin.
211
212		Called before the shutdown drain, so it finds their posts; the
213		services that own those operations shut down earlier and missed
214		them.
215		*/
216	4426x	void wait_cancels_posting() const noexcept
217		{
218	12856x	while (cancels_posting_.load(std::memory_order_acquire) != 0)
219	2002x	std::this_thread::yield();
220	4426x	}
221
222		private:
223		BOOST_COROSIO_MSVC_WARNING_PUSH
224		BOOST_COROSIO_MSVC_WARNING_DISABLE(4251)
225		std::atomic<std::size_t> cancels_posting_{0};
226		std::atomic<timer_service*> timer_svc_{nullptr};
227		BOOST_COROSIO_MSVC_WARNING_POP
228		};
229
230		/** Return the scheduler registered with the context.
231
232		@throws std::logic_error If the context has no backend installed.
233		*/
234		/** The innermost scheduler running on this thread, or null.
235
236		Each backend's run guard sets it and restores the previous value,
237		so a completion can ask whether its awaiter's context is running
238		here with one compare. Nested runs of other schedulers are still
239		answered by the scheduler's own frame stack.
240		*/
241		inline thread_local_ptr<scheduler const> running_scheduler;
242
243		/** RAII guard marking a scheduler as innermost on this thread.
244
245		Restores the previous innermost scheduler on destruction.
246		*/
247		class running_scheduler_guard
248		{
249		scheduler const* prev_;
250
251		public:
252		/// Construct the guard, making @p s the innermost scheduler.
253	13011x	explicit running_scheduler_guard(scheduler const* s) noexcept
254	13011x	: prev_(running_scheduler.get())
255		{
256	13011x	running_scheduler.set(s);
257	13011x	}
258
259		/// Destroy the guard, restoring the previous innermost scheduler.
260	13011x	~running_scheduler_guard()
261		{
262	13011x	running_scheduler.set(prev_);
263	13011x	}
264
265		running_scheduler_guard(running_scheduler_guard const&) = delete;
266		running_scheduler_guard& operator=(running_scheduler_guard const&) = delete;
267		};
268
269		inline scheduler&
270	1496x	get_scheduler(capy::execution_context& ctx)
271		{
272	1496x	auto* sched = ctx.find_service<scheduler>();
273	1496x	if (!sched)
274	✗	throw_logic_error("no scheduler installed");
275	1496x	return *sched;
276		}
277
278		} // namespace boost::corosio::detail
279
280		#endif
281