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