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