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