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