TLA Line data 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 HIT 4442 : ~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 MIS 0 : 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 HIT 983 : void cancel_posting_begin() noexcept
150 : {
151 983 : cancels_posting_.fetch_add(1, std::memory_order_relaxed);
152 983 : }
153 :
154 : /// End a @ref cancel_posting_begin once its post is made.
155 983 : void cancel_posting_end() noexcept
156 : {
157 983 : cancels_posting_.fetch_sub(1, std::memory_order_release);
158 983 : }
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 927 : void attach_timer_service(timer_service& svc)
175 : {
176 927 : on_timer_service_attached(svc);
177 927 : timer_svc_.store(&svc, std::memory_order_release);
178 927 : }
179 :
180 : /// Detach the timer service as it is destroyed.
181 927 : void detach_timer_service() noexcept
182 : {
183 927 : timer_svc_.store(nullptr, std::memory_order_release);
184 927 : }
185 :
186 : protected:
187 : /// Return the attached timer service, or null before the first timer.
188 329963 : timer_service* timers() const noexcept
189 : {
190 329963 : 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 4426 : void wait_cancels_posting() const noexcept
217 : {
218 12856 : while (cancels_posting_.load(std::memory_order_acquire) != 0)
219 2002 : std::this_thread::yield();
220 4426 : }
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 13011 : explicit running_scheduler_guard(scheduler const* s) noexcept
254 13011 : : prev_(running_scheduler.get())
255 : {
256 13011 : running_scheduler.set(s);
257 13011 : }
258 :
259 : /// Destroy the guard, restoring the previous innermost scheduler.
260 13011 : ~running_scheduler_guard()
261 : {
262 13011 : running_scheduler.set(prev_);
263 13011 : }
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 1496 : get_scheduler(capy::execution_context& ctx)
271 : {
272 1496 : auto* sched = ctx.find_service<scheduler>();
273 1496 : if (!sched)
274 MIS 0 : throw_logic_error("no scheduler installed");
275 HIT 1496 : return *sched;
276 : }
277 :
278 : } // namespace boost::corosio::detail
279 :
280 : #endif
|
