. /include/boost/corosio /native /detail /epoll /epoll_scheduler.hpp
97.74% Lines (173/177) 100.00% Functions (14/14)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2026 Steve Gerbino 2   // Copyright (c) 2026 Steve Gerbino
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
4   // 4   //
5   // Distributed under the Boost Software License, Version 1.0. (See accompanying 5   // Distributed under the Boost Software License, Version 1.0. (See accompanying
6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) 6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7   // 7   //
8   // Official repository: https://github.com/cppalliance/corosio 8   // Official repository: https://github.com/cppalliance/corosio
9   // 9   //
10   10  
11   #ifndef BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 11   #ifndef BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
12   #define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 12   #define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
13   13  
14   #include <boost/corosio/detail/platform.hpp> 14   #include <boost/corosio/detail/platform.hpp>
15   15  
16   #if BOOST_COROSIO_HAS_EPOLL 16   #if BOOST_COROSIO_HAS_EPOLL
17   17  
18   #include <boost/corosio/detail/config.hpp> 18   #include <boost/corosio/detail/config.hpp>
19   #include <boost/capy/ex/execution_context.hpp> 19   #include <boost/capy/ex/execution_context.hpp>
20   20  
21   #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp> 21   #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
22   #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp> 22   #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp>
23   23  
24   #include <boost/corosio/native/detail/epoll/epoll_traits.hpp> 24   #include <boost/corosio/native/detail/epoll/epoll_traits.hpp>
25   #include <boost/corosio/detail/timer_service.hpp> 25   #include <boost/corosio/detail/timer_service.hpp>
26   #include <boost/corosio/native/detail/make_err.hpp> 26   #include <boost/corosio/native/detail/make_err.hpp>
27   27  
28   #include <boost/corosio/detail/except.hpp> 28   #include <boost/corosio/detail/except.hpp>
29   29  
30   #include <atomic> 30   #include <atomic>
31   #include <chrono> 31   #include <chrono>
32   #include <cstdint> 32   #include <cstdint>
  33 + #include <limits>
33   #include <mutex> 34   #include <mutex>
34   #include <vector> 35   #include <vector>
35   36  
36   #include <errno.h> 37   #include <errno.h>
37   #include <sys/epoll.h> 38   #include <sys/epoll.h>
38   #include <sys/eventfd.h> 39   #include <sys/eventfd.h>
39   #include <sys/timerfd.h> 40   #include <sys/timerfd.h>
40   #include <unistd.h> 41   #include <unistd.h>
41   42  
42   namespace boost::corosio::detail { 43   namespace boost::corosio::detail {
43   44  
44   /** Linux scheduler using epoll for I/O multiplexing. 45   /** Linux scheduler using epoll for I/O multiplexing.
45   46  
46   This scheduler implements the scheduler interface using Linux epoll 47   This scheduler implements the scheduler interface using Linux epoll
47   for efficient I/O event notification. It uses a single reactor model 48   for efficient I/O event notification. It uses a single reactor model
48   where one thread runs epoll_wait while other threads 49   where one thread runs epoll_wait while other threads
49   wait on a condition variable for handler work. This design provides: 50   wait on a condition variable for handler work. This design provides:
50   51  
51   - Handler parallelism: N posted handlers can execute on N threads 52   - Handler parallelism: N posted handlers can execute on N threads
52   - No thundering herd: condition_variable wakes exactly one thread 53   - No thundering herd: condition_variable wakes exactly one thread
53   - IOCP parity: Behavior matches Windows I/O completion port semantics 54   - IOCP parity: Behavior matches Windows I/O completion port semantics
54   55  
55   When threads call run(), they first try to execute queued handlers. 56   When threads call run(), they first try to execute queued handlers.
56   If the queue is empty and no reactor is running, one thread becomes 57   If the queue is empty and no reactor is running, one thread becomes
57   the reactor and runs epoll_wait. Other threads wait on a condition 58   the reactor and runs epoll_wait. Other threads wait on a condition
58   variable until handlers are available. 59   variable until handlers are available.
59   60  
60   @par Thread Safety 61   @par Thread Safety
61   All public member functions are thread-safe. 62   All public member functions are thread-safe.
62   */ 63   */
63   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler 64   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
64   { 65   {
65   public: 66   public:
66   /** Construct the scheduler. 67   /** Construct the scheduler.
67   68  
68 - Creates an epoll instance, eventfd for reactor interruption, 69 + Creates an epoll instance and an eventfd for reactor
69 - and timerfd for kernel-managed timer expiry. 70 + interruption. The timerfd for timer expiry is created with the
  71 + context's first timer.
70   72  
71   @param ctx Reference to the owning execution_context. 73   @param ctx Reference to the owning execution_context.
72   @param concurrency_hint Hint for expected thread count (unused). 74   @param concurrency_hint Hint for expected thread count (unused).
73   */ 75   */
74   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 76   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
75   77  
76   /// Destroy the scheduler. 78   /// Destroy the scheduler.
77   ~epoll_scheduler() override; 79   ~epoll_scheduler() override;
78   80  
79   epoll_scheduler(epoll_scheduler const&) = delete; 81   epoll_scheduler(epoll_scheduler const&) = delete;
80   epoll_scheduler& operator=(epoll_scheduler const&) = delete; 82   epoll_scheduler& operator=(epoll_scheduler const&) = delete;
81   83  
82   /// Shut down the scheduler, draining pending operations. 84   /// Shut down the scheduler, draining pending operations.
83   void shutdown() override; 85   void shutdown() override;
84   86  
85   /// Apply runtime configuration, resizing the event buffer. 87   /// Apply runtime configuration, resizing the event buffer.
86   void configure_reactor( 88   void configure_reactor(
87   unsigned max_events, 89   unsigned max_events,
88   unsigned budget_init, 90   unsigned budget_init,
89   unsigned budget_max, 91   unsigned budget_max,
90   unsigned unassisted) override; 92   unsigned unassisted) override;
91   93  
92   /** Return the epoll file descriptor. 94   /** Return the epoll file descriptor.
93   95  
94   Used by socket services to register file descriptors 96   Used by socket services to register file descriptors
95   for I/O event notification. 97   for I/O event notification.
96   98  
97   @return The epoll file descriptor. 99   @return The epoll file descriptor.
98   */ 100   */
99   int epoll_fd() const noexcept 101   int epoll_fd() const noexcept
100   { 102   {
101   return epoll_fd_; 103   return epoll_fd_;
102   } 104   }
103   105  
104   /** Register a descriptor for persistent monitoring. 106   /** Register a descriptor for persistent monitoring.
105   107  
106   The fd is registered once and stays registered until explicitly 108   The fd is registered once and stays registered until explicitly
107   deregistered. Events are dispatched via reactor_descriptor_state which 109   deregistered. Events are dispatched via reactor_descriptor_state which
108   tracks pending read/write/connect operations. 110   tracks pending read/write/connect operations.
109   111  
110   @param fd The file descriptor to register. 112   @param fd The file descriptor to register.
111   @param desc Pointer to descriptor data (stored in epoll_event.data.ptr). 113   @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
112   114  
113   @return The error if registration fails, otherwise a default 115   @return The error if registration fails, otherwise a default
114   constructed error code. 116   constructed error code.
115   */ 117   */
116   std::error_code 118   std::error_code
117   register_descriptor(int fd, reactor_descriptor_state* desc) const; 119   register_descriptor(int fd, reactor_descriptor_state* desc) const;
118   120  
119   /// No-op: write readiness is watched from registration on. 121   /// No-op: write readiness is watched from registration on.
120   std::error_code 122   std::error_code
HITCBC 121   5602 ensure_write_registered(int, reactor_descriptor_state*) const noexcept 123   5401 ensure_write_registered(int, reactor_descriptor_state*) const noexcept
122   { 124   {
HITCBC 123   5602 return {}; 125   5401 return {};
124   } 126   }
125   127  
126   /** Deregister a persistently registered descriptor. 128   /** Deregister a persistently registered descriptor.
127   129  
128   @param fd The file descriptor to deregister. 130   @param fd The file descriptor to deregister.
129   */ 131   */
130   void deregister_descriptor(int fd) const; 132   void deregister_descriptor(int fd) const;
131   133  
132   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp). 134   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
HITCBC 133   286 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 135   286 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
134   { 136   {
HITCBC 135   286 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 137   286 return register_descriptor(read_fd, signal_pipe_reader_.arm());
136   } 138   }
137   139  
138   private: 140   private:
139   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; 141   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
140   void interrupt_reactor() const override; 142   void interrupt_reactor() const override;
141   void update_timerfd() const; 143   void update_timerfd() const;
  144 + static int timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept;
  145 + void on_timer_service_attached(timer_service& svc) override;
142   146  
143   int epoll_fd_; 147   int epoll_fd_;
144   int event_fd_; 148   int event_fd_;
145   int timer_fd_; 149   int timer_fd_;
146   150  
147   // Watches the global signal self-pipe's read end (armed lazily by 151   // Watches the global signal self-pipe's read end (armed lazily by
148   // register_signal_reader on the first signal registration). 152   // register_signal_reader on the first signal registration).
149   reactor_signal_pipe_reader signal_pipe_reader_; 153   reactor_signal_pipe_reader signal_pipe_reader_;
150   154  
151   // Edge-triggered eventfd state 155   // Edge-triggered eventfd state
152   mutable std::atomic<bool> eventfd_armed_{false}; 156   mutable std::atomic<bool> eventfd_armed_{false};
153   157  
154   // Set when the earliest timer changes; flushed before epoll_wait 158   // Set when the earliest timer changes; flushed before epoll_wait
155   mutable std::atomic<bool> timerfd_stale_{false}; 159   mutable std::atomic<bool> timerfd_stale_{false};
156   160  
157   // Event buffer sized from max_events_per_poll_ (set at construction, 161   // Event buffer sized from max_events_per_poll_ (set at construction,
158   // resized by configure_reactor via io_context_options). 162   // resized by configure_reactor via io_context_options).
159   std::vector<epoll_event> event_buffer_; 163   std::vector<epoll_event> event_buffer_;
160   }; 164   };
161   165  
HITCBC 162 - 2475 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) 166 + 2484 inline epoll_scheduler::epoll_scheduler(capy::execution_context&, int)
HITCBC 163   2475 : epoll_fd_(-1) 167   2484 : epoll_fd_(-1)
HITCBC 164   2475 , event_fd_(-1) 168   2484 , event_fd_(-1)
HITCBC 165   2475 , timer_fd_(-1) 169   2484 , timer_fd_(-1)
HITCBC 166   4950 , event_buffer_(max_events_per_poll_) 170   4968 , event_buffer_(max_events_per_poll_)
167   { 171   {
HITCBC 168   2475 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC); 172   2484 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
HITCBC 169   2475 if (epoll_fd_ < 0) 173   2484 if (epoll_fd_ < 0)
HITCBC 170   1 detail::throw_system_error(make_err(errno), "epoll_create1"); 174   3 detail::throw_system_error(make_err(errno), "epoll_create1");
171   175  
HITCBC 172   2474 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); 176   2481 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
HITCBC 173   2474 if (event_fd_ < 0) 177   2481 if (event_fd_ < 0)
174   { 178   {
HITCBC 175   1 int errn = errno; 179   3 int errn = errno;
HITCBC 176   1 ::close(epoll_fd_); 180   3 ::close(epoll_fd_);
HITCBC 177   1 detail::throw_system_error(make_err(errn), "eventfd"); 181   3 detail::throw_system_error(make_err(errn), "eventfd");
178   } 182   }
179 - timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);  
DCB 180 - 2473 if (timer_fd_ < 0)  
DCB 181 - 2473 {  
182 - int errn = errno;  
DCB 183 - 1 ::close(event_fd_);  
DCB 184 - 1 ::close(epoll_fd_);  
DCB 185 - 1 detail::throw_system_error(make_err(errn), "timerfd_create");  
DCB 186 - 1 }  
187 -  
188   183  
HITCBC 189   2472 epoll_event ev{}; 184   2478 epoll_event ev{};
HITCBC 190   2472 ev.events = EPOLLIN | EPOLLET; 185   2478 ev.events = EPOLLIN | EPOLLET;
HITCBC 191   2472 ev.data.ptr = nullptr; 186   2478 ev.data.ptr = nullptr;
HITCBC 192   2472 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0) 187   2478 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
193   { 188   {
DCB 194 - 1 ::close(timer_fd_);  
HITCBC 195   1 int errn = errno; 189   3 int errn = errno;
HITCBC 196   1 ::close(event_fd_); 190   3 ::close(event_fd_);
HITCBC 197   1 ::close(epoll_fd_); 191   3 ::close(epoll_fd_);
HITCBC 198   1 detail::throw_system_error(make_err(errn), "epoll_ctl"); 192   3 detail::throw_system_error(make_err(errn), "epoll_ctl");
199   } 193   }
200   194  
HITCBC 201 - 2471 epoll_event timer_ev{}; 195 + 2475 completed_ops_.push(&task_op_);
HITCBC 202 - 2471 timer_ev.events = EPOLLIN | EPOLLERR; 196 + 2502 }
ECB 203 - 2471 timer_ev.data.ptr = &timer_fd_; 197 +
ECB 204 - 2471 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0) 198 + inline void
HITGNC   199 + 534 epoll_scheduler::on_timer_service_attached(timer_service& svc)
  200 + {
  201 + // Without a timerfd the epoll_wait timeout drives the timers: the
  202 + // first timer is often made where a throw can only terminate.
HITGNC   203 + 534 int fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
HITGNC   204 + 534 if (fd >= 0)
205   { 205   {
ECB 206 - 1 int errn = errno; 206 + // Set before registration, which orders it for every thread
ECB 207 - 1 ::close(timer_fd_); 207 + // the timerfd's events can wake.
HITCBC 208 - 1 ::close(event_fd_); 208 + 533 timer_fd_ = fd;
HITCBC 209 - 1 ::close(epoll_fd_); 209 + 533 epoll_event timer_ev{};
HITCBC 210 - 1 detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)"); 210 + 533 timer_ev.events = EPOLLIN | EPOLLERR;
HITGNC   211 + 533 timer_ev.data.ptr = &timer_fd_;
HITGNC   212 + 533 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &timer_ev) < 0)
  213 + {
HITGNC   214 + 1 ::close(fd);
HITGNC   215 + 1 timer_fd_ = -1;
  216 + }
211   } 217   }
HITGIC 212 - 218 + 534 svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) {
HITCBC 213 - 2470 timer_svc_ = &get_timer_service(ctx, *this); 219 + 4650 auto* self = static_cast<epoll_scheduler*>(p);
HITCBC 214 - 2470 timer_svc_->set_on_earliest_changed( 220 + 4650 self->timerfd_stale_.store(true, std::memory_order_release);
HITCBC 215 - 7432 timer_service::callback(this, [](void* p) { 221 + 4650 self->interrupt_reactor();
HITCBC 216 - 4962 auto* self = static_cast<epoll_scheduler*>(p); 222 + 4650 }));
DCB 217 - 4962 self->timerfd_stale_.store(true, std::memory_order_release);  
DCB 218 - 4962 self->interrupt_reactor();  
DCB 219 - 4962 }));  
220 -  
DCB 221 - 2470 completed_ops_.push(&task_op_);  
HITCBC 222   2485 } 223   534 }
223   224  
HITCBC 224   4940 inline epoll_scheduler::~epoll_scheduler() 225   4950 inline epoll_scheduler::~epoll_scheduler()
225   { 226   {
HITCBC 226   2470 if (timer_fd_ >= 0) 227   2475 if (timer_fd_ >= 0)
HITCBC 227   2470 ::close(timer_fd_); 228   532 ::close(timer_fd_);
HITCBC 228   2470 if (event_fd_ >= 0) 229   2475 if (event_fd_ >= 0)
HITCBC 229   2470 ::close(event_fd_); 230   2475 ::close(event_fd_);
HITCBC 230   2470 if (epoll_fd_ >= 0) 231   2475 if (epoll_fd_ >= 0)
HITCBC 231   2470 ::close(epoll_fd_); 232   2475 ::close(epoll_fd_);
HITCBC 232   4940 } 233   4950 }
233   234  
234   inline void 235   inline void
HITCBC 235   2470 epoll_scheduler::shutdown() 236   2475 epoll_scheduler::shutdown()
236   { 237   {
HITCBC 237   2470 shutdown_drain(); 238   2475 shutdown_drain();
238   239  
HITCBC 239   2470 if (event_fd_ >= 0) 240   2475 if (event_fd_ >= 0)
HITCBC 240   2470 interrupt_reactor(); 241   2475 interrupt_reactor();
HITCBC 241   2470 } 242   2475 }
242   243  
243   inline void 244   inline void
HITCBC 244   32 epoll_scheduler::configure_reactor( 245   32 epoll_scheduler::configure_reactor(
245   unsigned max_events, 246   unsigned max_events,
246   unsigned budget_init, 247   unsigned budget_init,
247   unsigned budget_max, 248   unsigned budget_max,
248   unsigned unassisted) 249   unsigned unassisted)
249   { 250   {
HITCBC 250   32 reactor_scheduler::configure_reactor( 251   32 reactor_scheduler::configure_reactor(
251   max_events, budget_init, budget_max, unassisted); 252   max_events, budget_init, budget_max, unassisted);
HITCBC 252   30 event_buffer_.resize(max_events_per_poll_); 253   30 event_buffer_.resize(max_events_per_poll_);
HITCBC 253   30 } 254   30 }
254   255  
255   inline std::error_code 256   inline std::error_code
HITCBC 256   13426 epoll_scheduler::register_descriptor( 257   13287 epoll_scheduler::register_descriptor(
257   int fd, reactor_descriptor_state* desc) const 258   int fd, reactor_descriptor_state* desc) const
258   { 259   {
HITCBC 259   13426 epoll_event ev{}; 260   13287 epoll_event ev{};
HITCBC 260   13426 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP; 261   13287 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
HITCBC 261   13426 ev.data.ptr = desc; 262   13287 ev.data.ptr = desc;
262   263  
263   // Initialized before the add: another run thread's reactor can 264   // Initialized before the add: another run thread's reactor can
264   // dispatch an event for the descriptor as soon as it is added. 265   // dispatch an event for the descriptor as soon as it is added.
265   // A completion queued by an earlier registration of the same state 266   // A completion queued by an earlier registration of the same state
266   // may still be running, so the fields it reads change under its 267   // may still be running, so the fields it reads change under its
267   // mutex, and the mutex mode is only written when it differs. 268   // mutex, and the mutex mode is only written when it differs.
HITCBC 268   13426 if (desc->mutex.enabled() != reactor_io_locking_) 269   13287 if (desc->mutex.enabled() != reactor_io_locking_)
HITCBC 269   3 desc->mutex.set_enabled(reactor_io_locking_); 270   3 desc->mutex.set_enabled(reactor_io_locking_);
270   { 271   {
HITCBC 271   13426 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 272   13287 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 272   13426 desc->registered_events = ev.events; 273   13287 desc->registered_events = ev.events;
HITCBC 273   13426 desc->unpollable = false; 274   13287 desc->unpollable = false;
HITCBC 274   13426 desc->fd = fd; 275   13287 desc->fd = fd;
HITCBC 275   13426 desc->scheduler_ = this; 276   13287 desc->scheduler_ = this;
HITCBC 276   13426 desc->ready_events_.store(0, std::memory_order_release); 277   13287 desc->ready_events_.store(0, std::memory_order_release);
277   // object_ref_ is not touched: a queued invocation from an 278   // object_ref_ is not touched: a queued invocation from an
278   // earlier registration still owns the reference in it. 279   // earlier registration still owns the reference in it.
HITCBC 279   13426 desc->read_ready = false; 280   13287 desc->read_ready = false;
HITCBC 280   13426 desc->write_ready = false; 281   13287 desc->write_ready = false;
HITCBC 281   13426 } 282   13287 }
282   283  
HITCBC 283   13426 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) 284   13287 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
284   { 285   {
HITCBC 285   12 int const err = errno; 286   12 int const err = errno;
HITCBC 286   12 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 287   12 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 287   12 desc->registered_events = 0; 288   12 desc->registered_events = 0;
288   // EPERM: a file type epoll cannot watch. Its I/O does not 289   // EPERM: a file type epoll cannot watch. Its I/O does not
289   // block, so adopt it unwatched, as asio does. 290   // block, so adopt it unwatched, as asio does.
HITCBC 290   12 if (err != EPERM) 291   12 if (err != EPERM)
HITCBC 291   7 return make_err(err); 292   7 return make_err(err);
HITCBC 292   5 desc->unpollable = true; 293   5 desc->unpollable = true;
HITCBC 293   12 } 294   12 }
HITCBC 294   13419 return {}; 295   13280 return {};
295   } 296   }
296   297  
297   inline void 298   inline void
HITCBC 298   13129 epoll_scheduler::deregister_descriptor(int fd) const 299   12990 epoll_scheduler::deregister_descriptor(int fd) const
299   { 300   {
HITCBC 300   13129 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr); 301   12990 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
HITCBC 301   13129 } 302   12990 }
302   303  
303   inline void 304   inline void
HITCBC 304   13357 epoll_scheduler::interrupt_reactor() const 305   12879 epoll_scheduler::interrupt_reactor() const
305   { 306   {
HITCBC 306   13357 bool expected = false; 307   12879 bool expected = false;
HITCBC 307   13357 if (eventfd_armed_.compare_exchange_strong( 308   12879 if (eventfd_armed_.compare_exchange_strong(
308   expected, true, std::memory_order_release, 309   expected, true, std::memory_order_release,
309   std::memory_order_relaxed)) 310   std::memory_order_relaxed))
310   { 311   {
HITCBC 311   10972 std::uint64_t val = 1; 312   10523 std::uint64_t val = 1;
HITCBC 312   10972 if (::write(event_fd_, &val, sizeof(val)) < 0) 313   10523 if (::write(event_fd_, &val, sizeof(val)) < 0)
313   { 314   {
314   // The flag is what coalesces later interrupts into a byte 315   // The flag is what coalesces later interrupts into a byte
315   // already in the eventfd; a write that failed put no byte 316   // already in the eventfd; a write that failed put no byte
316   // there, so leaving it armed would swallow every interrupt 317   // there, so leaving it armed would swallow every interrupt
317   // that follows. Disarming keeps the cost to the interrupts 318   // that follows. Disarming keeps the cost to the interrupts
318   // already in flight -- the next one arms and writes again, 319   // already in flight -- the next one arms and writes again,
319   // instead of every one after this coalescing into a byte 320   // instead of every one after this coalescing into a byte
320   // that does not exist. 321   // that does not exist.
HITCBC 321   2 eventfd_armed_.store(false, std::memory_order_release); 322   2 eventfd_armed_.store(false, std::memory_order_release);
322   } 323   }
323   } 324   }
HITCBC 324   13357 } 325   12879 }
325   326  
326   inline void 327   inline void
HITCBC 327   12076 epoll_scheduler::update_timerfd() const 328   11568 epoll_scheduler::update_timerfd() const
328   { 329   {
HITCBC 329 - 12076 auto nearest = timer_svc_->nearest_expiry(); 330 + 11568 auto* tsvc = timers();
HITGNC   331 + 11568 if (!tsvc || timer_fd_ < 0)
HITGNC   332 + 6 return;
HITGNC   333 + 11562 auto nearest = tsvc->nearest_expiry();
330   334  
HITCBC 331   12076 itimerspec ts{}; 335   11562 itimerspec ts{};
HITCBC 332   12076 int flags = 0; 336   11562 int flags = 0;
333   337  
HITCBC 334   12076 if (nearest == timer_service::time_point::max()) 338   11562 if (nearest == timer_service::time_point::max())
335   { 339   {
336   // No timers — disarm by setting to 0 (relative) 340   // No timers — disarm by setting to 0 (relative)
337   } 341   }
338   else 342   else
339   { 343   {
HITCBC 340   10704 auto now = std::chrono::steady_clock::now(); 344   10281 auto now = std::chrono::steady_clock::now();
HITCBC 341   10704 if (nearest <= now) 345   10281 if (nearest <= now)
342   { 346   {
343   // Use 1ns instead of 0 — zero disarms the timerfd 347   // Use 1ns instead of 0 — zero disarms the timerfd
HITCBC 344   1044 ts.it_value.tv_nsec = 1; 348   1109 ts.it_value.tv_nsec = 1;
345   } 349   }
346   else 350   else
347   { 351   {
HITCBC 348   9660 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>( 352   9172 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
HITCBC 349   9660 nearest - now) 353   9172 nearest - now)
HITCBC 350   9660 .count(); 354   9172 .count();
HITCBC 351   9660 ts.it_value.tv_sec = nsec / 1000000000; 355   9172 ts.it_value.tv_sec = nsec / 1000000000;
HITCBC 352   9660 ts.it_value.tv_nsec = nsec % 1000000000; 356   9172 ts.it_value.tv_nsec = nsec % 1000000000;
HITCBC 353   9660 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0) 357   9172 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
MISUBC 354   ✗ ts.it_value.tv_nsec = 1; 358   ✗ ts.it_value.tv_nsec = 1;
355   } 359   }
356   } 360   }
357   361  
HITCBC 358   12076 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0) 362   11562 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
HITCBC 359   1 detail::throw_system_error(make_err(errno), "timerfd_settime"); 363   1 detail::throw_system_error(make_err(errno), "timerfd_settime");
ECB 360   12075 } 364   }
361   365  
  366 + // Rounded up, so the wait never wakes before the timer is due.
  367 + inline int
HITGNC   368 + 4 epoll_scheduler::timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept
  369 + {
HITGNC   370 + 4 auto nearest = svc.nearest_expiry();
HITGNC   371 + 4 if (nearest == timer_service::time_point::max())
MISUNC   372 + ✗ return timeout_ms;
HITGNC   373 + 4 auto now = std::chrono::steady_clock::now();
HITGNC   374 + 4 if (nearest <= now)
MISUNC   375 + ✗ return 0;
  376 + auto ms =
HITGNC   377 + 4 std::chrono::ceil<std::chrono::milliseconds>(nearest - now).count();
HITGNC   378 + 8 ms = (std::min)(ms,
HITGNC   379 + 4 static_cast<decltype(ms)>(
HITGNC   380 + 4 (std::numeric_limits<int>::max)()));
HITGNC   381 + 4 if (timeout_ms >= 0 && timeout_ms < ms)
MISUNC   382 + ✗ return timeout_ms;
HITGNC   383 + 4 return static_cast<int>(ms);
  384 + }
  385 +
362   inline void 386   inline void
HITCBC 363   156032 epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 387   148711 epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
364   { 388   {
365   int timeout_ms; 389   int timeout_ms;
HITCBC 366   156032 if (task_interrupted_) 390   148711 if (task_interrupted_)
HITCBC 367   137251 timeout_ms = 0; 391   128721 timeout_ms = 0;
HITCBC 368   18781 else if (timeout_us < 0) 392   19990 else if (timeout_us < 0)
HITCBC 369   18347 timeout_ms = -1; 393   19613 timeout_ms = -1;
370   else 394   else
HITCBC 371   434 timeout_ms = static_cast<int>((timeout_us + 999) / 1000); 395   377 timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
372   396  
HITCBC 373   156032 if (lock.owns_lock()) 397   148711 if (lock.owns_lock())
HITCBC 374   18785 lock.unlock(); 398   19994 lock.unlock();
375   399  
HITCBC 376   156032 task_cleanup on_exit{this, &lock, ctx}; 400   148711 task_cleanup on_exit{this, &lock, ctx};
377   401  
378   // Flush deferred timerfd programming before blocking 402   // Flush deferred timerfd programming before blocking
HITCBC 379   156032 if (timerfd_stale_.exchange(false, std::memory_order_acquire)) 403   148711 if (timerfd_stale_.exchange(false, std::memory_order_acquire))
HITCBC 380   4094 update_timerfd(); 404   3797 update_timerfd();
381   405  
  406 + // timer_fd_ is read only once the service is seen, after which it
  407 + // no longer changes.
HITGNC   408 + 148710 auto* tsvc = timers();
HITGNC   409 + 148710 bool const poll_timers = tsvc && timer_fd_ < 0;
HITGNC   410 + 148710 if (poll_timers && timeout_ms != 0)
HITGNC   411 + 4 timeout_ms = timer_timeout_ms(*tsvc, timeout_ms);
  412 +
HITCBC 382   156031 int nfds = ::epoll_wait( 413   148710 int nfds = ::epoll_wait(
HITCBC 383   156031 epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()), 414   148710 epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
384   timeout_ms); 415   timeout_ms);
385   416  
HITCBC 386   156031 if (nfds < 0 && errno != EINTR) 417   148710 if (nfds < 0 && errno != EINTR)
HITCBC 387   1 detail::throw_system_error(make_err(errno), "epoll_wait"); 418   1 detail::throw_system_error(make_err(errno), "epoll_wait");
388   419  
HITCBC 389 - 156030 bool check_timers = false; 420 + 148709 bool check_timers = poll_timers;
HITCBC 390   156030 ready_queue local_ops; 421   148709 ready_queue local_ops;
391   422  
HITCBC 392   221083 for (int i = 0; i < nfds; ++i) 423   210020 for (int i = 0; i < nfds; ++i)
393   { 424   {
HITCBC 394   65053 if (event_buffer_[i].data.ptr == nullptr) 425   61311 if (event_buffer_[i].data.ptr == nullptr)
395   { 426   {
396   std::uint64_t val; 427   std::uint64_t val;
397   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 428   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
HITCBC 398   8500 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val)); 429   8046 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
HITCBC 399   8500 eventfd_armed_.store(false, std::memory_order_relaxed); 430   8046 eventfd_armed_.store(false, std::memory_order_relaxed);
HITCBC 400   8500 continue; 431   8046 continue;
HITCBC 401   8500 } 432   8046 }
402   433  
HITCBC 403   56553 if (event_buffer_[i].data.ptr == &timer_fd_) 434   53265 if (event_buffer_[i].data.ptr == &timer_fd_)
404   { 435   {
405   std::uint64_t expirations; 436   std::uint64_t expirations;
406   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 437   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
407   [[maybe_unused]] auto r = 438   [[maybe_unused]] auto r =
HITCBC 408   7982 ::read(timer_fd_, &expirations, sizeof(expirations)); 439   7767 ::read(timer_fd_, &expirations, sizeof(expirations));
HITCBC 409   7982 check_timers = true; 440   7767 check_timers = true;
HITCBC 410   7982 continue; 441   7767 continue;
HITCBC 411   7982 } 442   7767 }
412   443  
413   auto* desc = 444   auto* desc =
HITCBC 414   48571 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr); 445   45498 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
415   446  
416   // EPOLLHUP maps to no reactor_event_* bit, so a HUP the widening 447   // EPOLLHUP maps to no reactor_event_* bit, so a HUP the widening
417   // below did not translate would take no branch in 448   // below did not translate would take no branch in
418   // invoke_deferred_io() and, the registration being 449   // invoke_deferred_io() and, the registration being
419   // edge-triggered, never get another chance. 450   // edge-triggered, never get another chance.
420   // 451   //
421   // HUP without OUT means a non-socket: a pipe or tty whose peer 452   // HUP without OUT means a non-socket: a pipe or tty whose peer
422   // closed, with or without data still unread (HUP alone, or 453   // closed, with or without data still unread (HUP alone, or
423   // IN|HUP). Treat it as readable, writable and faulted, the way 454   // IN|HUP). Treat it as readable, writable and faulted, the way
424   // poll() and io_uring report it, so a parked error wait 455   // poll() and io_uring report it, so a parked error wait
425   // completes too. Sockets carry at least OUT with HUP (a fresh 456   // completes too. Sockets carry at least OUT with HUP (a fresh
426   // unconnected stream socket reports HUP|OUT), so they take the 457   // unconnected stream socket reports HUP|OUT), so they take the
427   // plain widening, where forcing IN costs at most one spurious 458   // plain widening, where forcing IN costs at most one spurious
428   // EAGAIN. 459   // EAGAIN.
HITCBC 429   48571 std::uint32_t ev = event_buffer_[i].events; 460   45498 std::uint32_t ev = event_buffer_[i].events;
HITCBC 430   48571 if ((ev & (EPOLLHUP | EPOLLOUT)) == EPOLLHUP) 461   45498 if ((ev & (EPOLLHUP | EPOLLOUT)) == EPOLLHUP)
HITCBC 431   8 ev |= EPOLLIN | EPOLLOUT | EPOLLERR; 462   8 ev |= EPOLLIN | EPOLLOUT | EPOLLERR;
HITCBC 432   48563 else if (ev & EPOLLHUP) 463   45490 else if (ev & EPOLLHUP)
HITCBC 433   3072 ev |= EPOLLIN | EPOLLOUT; 464   2963 ev |= EPOLLIN | EPOLLOUT;
HITCBC 434   48571 desc->add_ready_events(ev); 465   45498 desc->add_ready_events(ev);
435   466  
HITCBC 436   48571 bool expected = false; 467   45498 bool expected = false;
HITCBC 437   48571 if (desc->is_enqueued_.compare_exchange_strong( 468   45498 if (desc->is_enqueued_.compare_exchange_strong(
438   expected, true, std::memory_order_release, 469   expected, true, std::memory_order_release,
439   std::memory_order_relaxed)) 470   std::memory_order_relaxed))
440   { 471   {
HITCBC 441   48567 local_ops.push(desc); 472   45476 local_ops.push(desc);
442   } 473   }
443   } 474   }
444   475  
HITCBC 445   156030 if (check_timers) 476   148709 if (check_timers)
446   { 477   {
HITCBC 447 - 7982 timer_svc_->process_expired(); 478 + 7771 if (auto* svc = timers())
HITGNC   479 + 7771 svc->process_expired();
HITCBC 448   7982 update_timerfd(); 480   7771 update_timerfd();
449   } 481   }
450   482  
HITCBC 451   156030 lock.lock(); 483   148709 lock.lock();
452   484  
HITCBC 453   156030 completed_ops_.splice(local_ops); 485   148709 completed_ops_.splice(local_ops);
HITCBC 454   156032 } 486   148711 }
455   487  
456   } // namespace boost::corosio::detail 488   } // namespace boost::corosio::detail
457   489  
458   #endif // BOOST_COROSIO_HAS_EPOLL 490   #endif // BOOST_COROSIO_HAS_EPOLL
459   491  
460   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 492   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP