. /include/boost/corosio /native /detail /epoll /epoll_scheduler.hpp
99.38% Lines (159/160) 100.00% Functions (12/12)
[ ] 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 <mutex> 33   #include <mutex>
34   #include <vector> 34   #include <vector>
35   35  
36   #include <errno.h> 36   #include <errno.h>
37   #include <sys/epoll.h> 37   #include <sys/epoll.h>
38   #include <sys/eventfd.h> 38   #include <sys/eventfd.h>
39   #include <sys/timerfd.h> 39   #include <sys/timerfd.h>
40   #include <unistd.h> 40   #include <unistd.h>
41   41  
42   namespace boost::corosio::detail { 42   namespace boost::corosio::detail {
43   43  
44   /** Linux scheduler using epoll for I/O multiplexing. 44   /** Linux scheduler using epoll for I/O multiplexing.
45   45  
46   This scheduler implements the scheduler interface using Linux epoll 46   This scheduler implements the scheduler interface using Linux epoll
47   for efficient I/O event notification. It uses a single reactor model 47   for efficient I/O event notification. It uses a single reactor model
48   where one thread runs epoll_wait while other threads 48   where one thread runs epoll_wait while other threads
49   wait on a condition variable for handler work. This design provides: 49   wait on a condition variable for handler work. This design provides:
50   50  
51   - Handler parallelism: N posted handlers can execute on N threads 51   - Handler parallelism: N posted handlers can execute on N threads
52   - No thundering herd: condition_variable wakes exactly one thread 52   - No thundering herd: condition_variable wakes exactly one thread
53   - IOCP parity: Behavior matches Windows I/O completion port semantics 53   - IOCP parity: Behavior matches Windows I/O completion port semantics
54   54  
55   When threads call run(), they first try to execute queued handlers. 55   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 56   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 57   the reactor and runs epoll_wait. Other threads wait on a condition
58   variable until handlers are available. 58   variable until handlers are available.
59   59  
60   @par Thread Safety 60   @par Thread Safety
61   All public member functions are thread-safe. 61   All public member functions are thread-safe.
62   */ 62   */
63   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler 63   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
64   { 64   {
65   public: 65   public:
66   /** Construct the scheduler. 66   /** Construct the scheduler.
67   67  
68   Creates an epoll instance, eventfd for reactor interruption, 68   Creates an epoll instance, eventfd for reactor interruption,
69   and timerfd for kernel-managed timer expiry. 69   and timerfd for kernel-managed timer expiry.
70   70  
71   @param ctx Reference to the owning execution_context. 71   @param ctx Reference to the owning execution_context.
72   @param concurrency_hint Hint for expected thread count (unused). 72   @param concurrency_hint Hint for expected thread count (unused).
73   */ 73   */
74   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 74   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
75   75  
76   /// Destroy the scheduler. 76   /// Destroy the scheduler.
77   ~epoll_scheduler() override; 77   ~epoll_scheduler() override;
78   78  
79   epoll_scheduler(epoll_scheduler const&) = delete; 79   epoll_scheduler(epoll_scheduler const&) = delete;
80   epoll_scheduler& operator=(epoll_scheduler const&) = delete; 80   epoll_scheduler& operator=(epoll_scheduler const&) = delete;
81   81  
82   /// Shut down the scheduler, draining pending operations. 82   /// Shut down the scheduler, draining pending operations.
83   void shutdown() override; 83   void shutdown() override;
84   84  
85   /// Apply runtime configuration, resizing the event buffer. 85   /// Apply runtime configuration, resizing the event buffer.
86   void configure_reactor( 86   void configure_reactor(
87   unsigned max_events, 87   unsigned max_events,
88   unsigned budget_init, 88   unsigned budget_init,
89   unsigned budget_max, 89   unsigned budget_max,
90   unsigned unassisted) override; 90   unsigned unassisted) override;
91   91  
92   /** Return the epoll file descriptor. 92   /** Return the epoll file descriptor.
93   93  
94   Used by socket services to register file descriptors 94   Used by socket services to register file descriptors
95   for I/O event notification. 95   for I/O event notification.
96   96  
97   @return The epoll file descriptor. 97   @return The epoll file descriptor.
98   */ 98   */
99   int epoll_fd() const noexcept 99   int epoll_fd() const noexcept
100   { 100   {
101   return epoll_fd_; 101   return epoll_fd_;
102   } 102   }
103   103  
104   /** Register a descriptor for persistent monitoring. 104   /** Register a descriptor for persistent monitoring.
105   105  
106   The fd is registered once and stays registered until explicitly 106   The fd is registered once and stays registered until explicitly
107   deregistered. Events are dispatched via reactor_descriptor_state which 107   deregistered. Events are dispatched via reactor_descriptor_state which
108   tracks pending read/write/connect operations. 108   tracks pending read/write/connect operations.
109   109  
110   @param fd The file descriptor to register. 110   @param fd The file descriptor to register.
111   @param desc Pointer to descriptor data (stored in epoll_event.data.ptr). 111   @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
112   112  
113   @return The error if registration fails, otherwise a default 113   @return The error if registration fails, otherwise a default
114   constructed error code. 114   constructed error code.
115   */ 115   */
116   std::error_code 116   std::error_code
117   register_descriptor(int fd, reactor_descriptor_state* desc) const; 117   register_descriptor(int fd, reactor_descriptor_state* desc) const;
118   118  
119   /// No-op: write readiness is watched from registration on. 119   /// No-op: write readiness is watched from registration on.
120   std::error_code 120   std::error_code
HITCBC 121   2658 ensure_write_registered(int, reactor_descriptor_state*) const noexcept 121   2626 ensure_write_registered(int, reactor_descriptor_state*) const noexcept
122   { 122   {
HITCBC 123   2658 return {}; 123   2626 return {};
124   } 124   }
125   125  
126   /** Deregister a persistently registered descriptor. 126   /** Deregister a persistently registered descriptor.
127   127  
128   @param fd The file descriptor to deregister. 128   @param fd The file descriptor to deregister.
129   */ 129   */
130   void deregister_descriptor(int fd) const; 130   void deregister_descriptor(int fd) const;
131   131  
132   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp). 132   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
HITCBC 133   76 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 133   76 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
134   { 134   {
HITCBC 135   76 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 135   76 return register_descriptor(read_fd, signal_pipe_reader_.arm());
136   } 136   }
137   137  
138   private: 138   private:
139   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; 139   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
140   void interrupt_reactor() const override; 140   void interrupt_reactor() const override;
141   void update_timerfd() const; 141   void update_timerfd() const;
142   142  
143   int epoll_fd_; 143   int epoll_fd_;
144   int event_fd_; 144   int event_fd_;
145   int timer_fd_; 145   int timer_fd_;
146   146  
147   // Watches the global signal self-pipe's read end (armed lazily by 147   // Watches the global signal self-pipe's read end (armed lazily by
148   // register_signal_reader on the first signal registration). 148   // register_signal_reader on the first signal registration).
149   reactor_signal_pipe_reader signal_pipe_reader_; 149   reactor_signal_pipe_reader signal_pipe_reader_;
150   150  
151   // Edge-triggered eventfd state 151   // Edge-triggered eventfd state
152   mutable std::atomic<bool> eventfd_armed_{false}; 152   mutable std::atomic<bool> eventfd_armed_{false};
153   153  
154   // Set when the earliest timer changes; flushed before epoll_wait 154   // Set when the earliest timer changes; flushed before epoll_wait
155   mutable std::atomic<bool> timerfd_stale_{false}; 155   mutable std::atomic<bool> timerfd_stale_{false};
156   156  
157   // Event buffer sized from max_events_per_poll_ (set at construction, 157   // Event buffer sized from max_events_per_poll_ (set at construction,
158   // resized by configure_reactor via io_context_options). 158   // resized by configure_reactor via io_context_options).
159   std::vector<epoll_event> event_buffer_; 159   std::vector<epoll_event> event_buffer_;
160   }; 160   };
161   161  
HITCBC 162   1577 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) 162   1577 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
HITCBC 163   1577 : epoll_fd_(-1) 163   1577 : epoll_fd_(-1)
HITCBC 164   1577 , event_fd_(-1) 164   1577 , event_fd_(-1)
HITCBC 165   1577 , timer_fd_(-1) 165   1577 , timer_fd_(-1)
HITCBC 166   3154 , event_buffer_(max_events_per_poll_) 166   3154 , event_buffer_(max_events_per_poll_)
167   { 167   {
HITCBC 168   1577 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC); 168   1577 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
HITCBC 169   1577 if (epoll_fd_ < 0) 169   1577 if (epoll_fd_ < 0)
HITCBC 170   1 detail::throw_system_error(make_err(errno), "epoll_create1"); 170   1 detail::throw_system_error(make_err(errno), "epoll_create1");
171   171  
HITCBC 172   1576 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); 172   1576 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
HITCBC 173   1576 if (event_fd_ < 0) 173   1576 if (event_fd_ < 0)
174   { 174   {
HITCBC 175   1 int errn = errno; 175   1 int errn = errno;
HITCBC 176   1 ::close(epoll_fd_); 176   1 ::close(epoll_fd_);
HITCBC 177   1 detail::throw_system_error(make_err(errn), "eventfd"); 177   1 detail::throw_system_error(make_err(errn), "eventfd");
178   } 178   }
179   179  
HITCBC 180   1575 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); 180   1575 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
HITCBC 181   1575 if (timer_fd_ < 0) 181   1575 if (timer_fd_ < 0)
182   { 182   {
HITCBC 183   1 int errn = errno; 183   1 int errn = errno;
HITCBC 184   1 ::close(event_fd_); 184   1 ::close(event_fd_);
HITCBC 185   1 ::close(epoll_fd_); 185   1 ::close(epoll_fd_);
HITCBC 186   1 detail::throw_system_error(make_err(errn), "timerfd_create"); 186   1 detail::throw_system_error(make_err(errn), "timerfd_create");
187   } 187   }
188   188  
HITCBC 189   1574 epoll_event ev{}; 189   1574 epoll_event ev{};
HITCBC 190   1574 ev.events = EPOLLIN | EPOLLET; 190   1574 ev.events = EPOLLIN | EPOLLET;
HITCBC 191   1574 ev.data.ptr = nullptr; 191   1574 ev.data.ptr = nullptr;
HITCBC 192   1574 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0) 192   1574 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
193   { 193   {
HITCBC 194   1 int errn = errno; 194   1 int errn = errno;
HITCBC 195   1 ::close(timer_fd_); 195   1 ::close(timer_fd_);
HITCBC 196   1 ::close(event_fd_); 196   1 ::close(event_fd_);
HITCBC 197   1 ::close(epoll_fd_); 197   1 ::close(epoll_fd_);
HITCBC 198   1 detail::throw_system_error(make_err(errn), "epoll_ctl"); 198   1 detail::throw_system_error(make_err(errn), "epoll_ctl");
199   } 199   }
200   200  
HITCBC 201   1573 epoll_event timer_ev{}; 201   1573 epoll_event timer_ev{};
HITCBC 202   1573 timer_ev.events = EPOLLIN | EPOLLERR; 202   1573 timer_ev.events = EPOLLIN | EPOLLERR;
HITCBC 203   1573 timer_ev.data.ptr = &timer_fd_; 203   1573 timer_ev.data.ptr = &timer_fd_;
HITCBC 204   1573 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0) 204   1573 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
205   { 205   {
HITCBC 206   1 int errn = errno; 206   1 int errn = errno;
HITCBC 207   1 ::close(timer_fd_); 207   1 ::close(timer_fd_);
HITCBC 208   1 ::close(event_fd_); 208   1 ::close(event_fd_);
HITCBC 209   1 ::close(epoll_fd_); 209   1 ::close(epoll_fd_);
HITCBC 210   1 detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)"); 210   1 detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
211   } 211   }
212   212  
HITCBC 213   1572 timer_svc_ = &get_timer_service(ctx, *this); 213   1572 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 214   1572 timer_svc_->set_on_earliest_changed( 214   1572 timer_svc_->set_on_earliest_changed(
HITCBC 215   6092 timer_service::callback(this, [](void* p) { 215   5976 timer_service::callback(this, [](void* p) {
HITCBC 216   4520 auto* self = static_cast<epoll_scheduler*>(p); 216   4404 auto* self = static_cast<epoll_scheduler*>(p);
HITCBC 217   4520 self->timerfd_stale_.store(true, std::memory_order_release); 217   4404 self->timerfd_stale_.store(true, std::memory_order_release);
HITCBC 218   4520 self->interrupt_reactor(); 218   4404 self->interrupt_reactor();
HITCBC 219   4520 })); 219   4404 }));
220   220  
HITCBC 221   1572 completed_ops_.push(&task_op_); 221   1572 completed_ops_.push(&task_op_);
HITCBC 222   1587 } 222   1587 }
223   223  
HITCBC 224   3144 inline epoll_scheduler::~epoll_scheduler() 224   3144 inline epoll_scheduler::~epoll_scheduler()
225   { 225   {
HITCBC 226   1572 if (timer_fd_ >= 0) 226   1572 if (timer_fd_ >= 0)
HITCBC 227   1572 ::close(timer_fd_); 227   1572 ::close(timer_fd_);
HITCBC 228   1572 if (event_fd_ >= 0) 228   1572 if (event_fd_ >= 0)
HITCBC 229   1572 ::close(event_fd_); 229   1572 ::close(event_fd_);
HITCBC 230   1572 if (epoll_fd_ >= 0) 230   1572 if (epoll_fd_ >= 0)
HITCBC 231   1572 ::close(epoll_fd_); 231   1572 ::close(epoll_fd_);
HITCBC 232   3144 } 232   3144 }
233   233  
234   inline void 234   inline void
HITCBC 235   1572 epoll_scheduler::shutdown() 235   1572 epoll_scheduler::shutdown()
236   { 236   {
HITCBC 237   1572 shutdown_drain(); 237   1572 shutdown_drain();
238   238  
HITCBC 239   1572 if (event_fd_ >= 0) 239   1572 if (event_fd_ >= 0)
HITCBC 240   1572 interrupt_reactor(); 240   1572 interrupt_reactor();
HITCBC 241   1572 } 241   1572 }
242   242  
243   inline void 243   inline void
HITCBC 244   28 epoll_scheduler::configure_reactor( 244   28 epoll_scheduler::configure_reactor(
245   unsigned max_events, 245   unsigned max_events,
246   unsigned budget_init, 246   unsigned budget_init,
247   unsigned budget_max, 247   unsigned budget_max,
248   unsigned unassisted) 248   unsigned unassisted)
249   { 249   {
HITCBC 250   28 reactor_scheduler::configure_reactor( 250   28 reactor_scheduler::configure_reactor(
251   max_events, budget_init, budget_max, unassisted); 251   max_events, budget_init, budget_max, unassisted);
HITCBC 252   26 event_buffer_.resize(max_events_per_poll_); 252   26 event_buffer_.resize(max_events_per_poll_);
HITCBC 253   26 } 253   26 }
254   254  
255   inline std::error_code 255   inline std::error_code
HITCBC 256   6003 epoll_scheduler::register_descriptor( 256   5939 epoll_scheduler::register_descriptor(
257   int fd, reactor_descriptor_state* desc) const 257   int fd, reactor_descriptor_state* desc) const
258   { 258   {
HITCBC 259   6003 epoll_event ev{}; 259   5939 epoll_event ev{};
HITCBC 260   6003 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP; 260   5939 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
HITCBC 261   6003 ev.data.ptr = desc; 261   5939 ev.data.ptr = desc;
262   262  
HITCBC 263   6003 bool unpollable = false; 263   5939 bool unpollable = false;
HITCBC 264   6003 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) 264   5939 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
265   { 265   {
266   // EPERM: a file type epoll cannot watch. Its I/O does not 266   // EPERM: a file type epoll cannot watch. Its I/O does not
267   // block, so adopt it unwatched, as asio does. 267   // block, so adopt it unwatched, as asio does.
HITCBC 268   11 if (errno != EPERM) 268   11 if (errno != EPERM)
HITCBC 269   7 return make_err(errno); 269   7 return make_err(errno);
HITCBC 270   4 unpollable = true; 270   4 unpollable = true;
271   } 271   }
272   272  
HITCBC 273   5996 desc->registered_events = unpollable ? 0 : ev.events; 273   5932 desc->registered_events = unpollable ? 0 : ev.events;
HITCBC 274   5996 desc->unpollable = unpollable; 274   5932 desc->unpollable = unpollable;
HITCBC 275   5996 desc->fd = fd; 275   5932 desc->fd = fd;
HITCBC 276   5996 desc->scheduler_ = this; 276   5932 desc->scheduler_ = this;
HITCBC 277   5996 desc->mutex.set_enabled(reactor_io_locking_); 277   5932 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 278   5996 desc->ready_events_.store(0, std::memory_order_relaxed); 278   5932 desc->ready_events_.store(0, std::memory_order_relaxed);
279   279  
HITCBC 280   5996 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 280   5932 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 281   5996 desc->impl_ref_.reset(); 281   5932 desc->impl_ref_.reset();
HITCBC 282   5996 desc->read_ready = false; 282   5932 desc->read_ready = false;
HITCBC 283   5996 desc->write_ready = false; 283   5932 desc->write_ready = false;
HITCBC 284   5996 return {}; 284   5932 return {};
HITCBC 285   5996 } 285   5932 }
286   286  
287   inline void 287   inline void
HITCBC 288   5917 epoll_scheduler::deregister_descriptor(int fd) const 288   5853 epoll_scheduler::deregister_descriptor(int fd) const
289   { 289   {
HITCBC 290   5917 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr); 290   5853 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
HITCBC 291   5917 } 291   5853 }
292   292  
293   inline void 293   inline void
HITCBC 294   8607 epoll_scheduler::interrupt_reactor() const 294   8636 epoll_scheduler::interrupt_reactor() const
295   { 295   {
HITCBC 296   8607 bool expected = false; 296   8636 bool expected = false;
HITCBC 297   8607 if (eventfd_armed_.compare_exchange_strong( 297   8636 if (eventfd_armed_.compare_exchange_strong(
298   expected, true, std::memory_order_release, 298   expected, true, std::memory_order_release,
299   std::memory_order_relaxed)) 299   std::memory_order_relaxed))
300   { 300   {
HITCBC 301   6799 std::uint64_t val = 1; 301   6887 std::uint64_t val = 1;
HITCBC 302   6799 if (::write(event_fd_, &val, sizeof(val)) < 0) 302   6887 if (::write(event_fd_, &val, sizeof(val)) < 0)
303   { 303   {
304   // The flag is what coalesces later interrupts into a byte 304   // The flag is what coalesces later interrupts into a byte
305   // already in the eventfd; a write that failed put no byte 305   // already in the eventfd; a write that failed put no byte
306   // there, so leaving it armed would swallow every interrupt 306   // there, so leaving it armed would swallow every interrupt
307   // that follows. Disarming keeps the cost to the interrupts 307   // that follows. Disarming keeps the cost to the interrupts
308   // already in flight -- the next one arms and writes again, 308   // already in flight -- the next one arms and writes again,
309   // instead of every one after this coalescing into a byte 309   // instead of every one after this coalescing into a byte
310   // that does not exist. 310   // that does not exist.
HITCBC 311   2 eventfd_armed_.store(false, std::memory_order_release); 311   2 eventfd_armed_.store(false, std::memory_order_release);
312   } 312   }
313   } 313   }
HITCBC 314   8607 } 314   8636 }
315   315  
316   inline void 316   inline void
HITCBC 317   11035 epoll_scheduler::update_timerfd() const 317   10892 epoll_scheduler::update_timerfd() const
318   { 318   {
HITCBC 319   11035 auto nearest = timer_svc_->nearest_expiry(); 319   10892 auto nearest = timer_svc_->nearest_expiry();
320   320  
HITCBC 321   11035 itimerspec ts{}; 321   10892 itimerspec ts{};
HITCBC 322   11035 int flags = 0; 322   10892 int flags = 0;
323   323  
HITCBC 324   11035 if (nearest == timer_service::time_point::max()) 324   10892 if (nearest == timer_service::time_point::max())
325   { 325   {
326   // No timers — disarm by setting to 0 (relative) 326   // No timers — disarm by setting to 0 (relative)
327   } 327   }
328   else 328   else
329   { 329   {
HITCBC 330   9810 auto now = std::chrono::steady_clock::now(); 330   9631 auto now = std::chrono::steady_clock::now();
HITCBC 331   9810 if (nearest <= now) 331   9631 if (nearest <= now)
332   { 332   {
333   // Use 1ns instead of 0 — zero disarms the timerfd 333   // Use 1ns instead of 0 — zero disarms the timerfd
HITCBC 334   955 ts.it_value.tv_nsec = 1; 334   971 ts.it_value.tv_nsec = 1;
335   } 335   }
336   else 336   else
337   { 337   {
HITCBC 338   8855 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>( 338   8660 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
HITCBC 339   8855 nearest - now) 339   8660 nearest - now)
HITCBC 340   8855 .count(); 340   8660 .count();
HITCBC 341   8855 ts.it_value.tv_sec = nsec / 1000000000; 341   8660 ts.it_value.tv_sec = nsec / 1000000000;
HITCBC 342   8855 ts.it_value.tv_nsec = nsec % 1000000000; 342   8660 ts.it_value.tv_nsec = nsec % 1000000000;
HITCBC 343   8855 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0) 343   8660 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
MISUBC 344   ✗ ts.it_value.tv_nsec = 1; 344   ✗ ts.it_value.tv_nsec = 1;
345   } 345   }
346   } 346   }
347   347  
HITCBC 348   11035 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0) 348   10892 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
HITCBC 349   1 detail::throw_system_error(make_err(errno), "timerfd_settime"); 349   1 detail::throw_system_error(make_err(errno), "timerfd_settime");
HITCBC 350   11034 } 350   10891 }
351   351  
352   inline void 352   inline void
HITCBC 353   40693 epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 353   41337 epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
354   { 354   {
355   int timeout_ms; 355   int timeout_ms;
HITCBC 356   40693 if (task_interrupted_) 356   41337 if (task_interrupted_)
HITCBC 357   28527 timeout_ms = 0; 357   29469 timeout_ms = 0;
HITCBC 358   12166 else if (timeout_us < 0) 358   11868 else if (timeout_us < 0)
HITCBC 359   11806 timeout_ms = -1; 359   11501 timeout_ms = -1;
360   else 360   else
HITCBC 361   360 timeout_ms = static_cast<int>((timeout_us + 999) / 1000); 361   367 timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
362   362  
HITCBC 363   40693 if (lock.owns_lock()) 363   41337 if (lock.owns_lock())
HITCBC 364   12168 lock.unlock(); 364   11870 lock.unlock();
365   365  
HITCBC 366   40693 task_cleanup on_exit{this, &lock, ctx}; 366   41337 task_cleanup on_exit{this, &lock, ctx};
367   367  
368   // Flush deferred timerfd programming before blocking 368   // Flush deferred timerfd programming before blocking
HITCBC 369   40693 if (timerfd_stale_.exchange(false, std::memory_order_acquire)) 369   41337 if (timerfd_stale_.exchange(false, std::memory_order_acquire))
HITCBC 370   3850 update_timerfd(); 370   3799 update_timerfd();
371   371  
HITCBC 372   40692 int nfds = ::epoll_wait( 372   41336 int nfds = ::epoll_wait(
HITCBC 373   40692 epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()), 373   41336 epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
374   timeout_ms); 374   timeout_ms);
375   375  
HITCBC 376   40692 if (nfds < 0 && errno != EINTR) 376   41336 if (nfds < 0 && errno != EINTR)
HITCBC 377   1 detail::throw_system_error(make_err(errno), "epoll_wait"); 377   1 detail::throw_system_error(make_err(errno), "epoll_wait");
378   378  
HITCBC 379   40691 bool check_timers = false; 379   41335 bool check_timers = false;
HITCBC 380   40691 ready_queue local_ops; 380   41335 ready_queue local_ops;
381   381  
HITCBC 382   88729 for (int i = 0; i < nfds; ++i) 382   89874 for (int i = 0; i < nfds; ++i)
383   { 383   {
HITCBC 384   48038 if (event_buffer_[i].data.ptr == nullptr) 384   48539 if (event_buffer_[i].data.ptr == nullptr)
385   { 385   {
386   std::uint64_t val; 386   std::uint64_t val;
387   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 387   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
HITCBC 388   5225 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val)); 388   5313 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
HITCBC 389   5225 eventfd_armed_.store(false, std::memory_order_relaxed); 389   5313 eventfd_armed_.store(false, std::memory_order_relaxed);
HITCBC 390   5225 continue; 390   5313 continue;
HITCBC 391   5225 } 391   5313 }
392   392  
HITCBC 393   42813 if (event_buffer_[i].data.ptr == &timer_fd_) 393   43226 if (event_buffer_[i].data.ptr == &timer_fd_)
394   { 394   {
395   std::uint64_t expirations; 395   std::uint64_t expirations;
396   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 396   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
397   [[maybe_unused]] auto r = 397   [[maybe_unused]] auto r =
HITCBC 398   7185 ::read(timer_fd_, &expirations, sizeof(expirations)); 398   7093 ::read(timer_fd_, &expirations, sizeof(expirations));
HITCBC 399   7185 check_timers = true; 399   7093 check_timers = true;
HITCBC 400   7185 continue; 400   7093 continue;
HITCBC 401   7185 } 401   7093 }
402   402  
403   auto* desc = 403   auto* desc =
HITCBC 404   35628 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr); 404   36133 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
405   405  
406   // EPOLLHUP maps to no reactor_event_* bit, so a HUP the widening 406   // EPOLLHUP maps to no reactor_event_* bit, so a HUP the widening
407   // below did not translate would take no branch in 407   // below did not translate would take no branch in
408   // invoke_deferred_io() and, the registration being 408   // invoke_deferred_io() and, the registration being
409   // edge-triggered, never get another chance. 409   // edge-triggered, never get another chance.
410   // 410   //
411   // HUP without OUT means a non-socket: a pipe or tty whose peer 411   // HUP without OUT means a non-socket: a pipe or tty whose peer
412   // closed, with or without data still unread (HUP alone, or 412   // closed, with or without data still unread (HUP alone, or
413   // IN|HUP). Treat it as readable, writable and faulted, the way 413   // IN|HUP). Treat it as readable, writable and faulted, the way
414   // poll() and io_uring report it, so a parked error wait 414   // poll() and io_uring report it, so a parked error wait
415   // completes too. Sockets carry at least OUT with HUP (a fresh 415   // completes too. Sockets carry at least OUT with HUP (a fresh
416   // unconnected stream socket reports HUP|OUT), so they take the 416   // unconnected stream socket reports HUP|OUT), so they take the
417   // plain widening, where forcing IN costs at most one spurious 417   // plain widening, where forcing IN costs at most one spurious
418   // EAGAIN. 418   // EAGAIN.
HITCBC 419   35628 std::uint32_t ev = event_buffer_[i].events; 419   36133 std::uint32_t ev = event_buffer_[i].events;
HITCBC 420   35628 if ((ev & (EPOLLHUP | EPOLLOUT)) == EPOLLHUP) 420   36133 if ((ev & (EPOLLHUP | EPOLLOUT)) == EPOLLHUP)
HITCBC 421   8 ev |= EPOLLIN | EPOLLOUT | EPOLLERR; 421   8 ev |= EPOLLIN | EPOLLOUT | EPOLLERR;
HITCBC 422   35620 else if (ev & EPOLLHUP) 422   36125 else if (ev & EPOLLHUP)
HITCBC 423   235 ev |= EPOLLIN | EPOLLOUT; 423   234 ev |= EPOLLIN | EPOLLOUT;
HITCBC 424   35628 desc->add_ready_events(ev); 424   36133 desc->add_ready_events(ev);
425   425  
HITCBC 426   35628 bool expected = false; 426   36133 bool expected = false;
HITCBC 427   35628 if (desc->is_enqueued_.compare_exchange_strong( 427   36133 if (desc->is_enqueued_.compare_exchange_strong(
428   expected, true, std::memory_order_release, 428   expected, true, std::memory_order_release,
429   std::memory_order_relaxed)) 429   std::memory_order_relaxed))
430   { 430   {
HITCBC 431   35628 local_ops.push(desc); 431   36133 local_ops.push(desc);
432   } 432   }
433   } 433   }
434   434  
HITCBC 435   40691 if (check_timers) 435   41335 if (check_timers)
436   { 436   {
HITCBC 437   7185 timer_svc_->process_expired(); 437   7093 timer_svc_->process_expired();
HITCBC 438   7185 update_timerfd(); 438   7093 update_timerfd();
439   } 439   }
440   440  
HITCBC 441   40691 lock.lock(); 441   41335 lock.lock();
442   442  
HITCBC 443   40691 completed_ops_.splice(local_ops); 443   41335 completed_ops_.splice(local_ops);
HITCBC 444   40693 } 444   41337 }
445   445  
446   } // namespace boost::corosio::detail 446   } // namespace boost::corosio::detail
447   447  
448   #endif // BOOST_COROSIO_HAS_EPOLL 448   #endif // BOOST_COROSIO_HAS_EPOLL
449   449  
450   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 450   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP