. /include/boost/corosio /native /detail /epoll /epoll_scheduler.hpp
99.35% Lines (152/153) 100.00% Functions (11/11)
[ ] 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   #include <boost/corosio/native/detail/posix/posix_resolver_service.hpp> 27   #include <boost/corosio/native/detail/posix/posix_resolver_service.hpp>
28   #include <boost/corosio/native/detail/posix/posix_signal_service.hpp> 28   #include <boost/corosio/native/detail/posix/posix_signal_service.hpp>
29   #include <boost/corosio/native/detail/posix/posix_stream_file_service.hpp> 29   #include <boost/corosio/native/detail/posix/posix_stream_file_service.hpp>
30   #include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp> 30   #include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp>
31   31  
32   #include <boost/corosio/detail/except.hpp> 32   #include <boost/corosio/detail/except.hpp>
33   33  
34   #include <atomic> 34   #include <atomic>
35   #include <chrono> 35   #include <chrono>
36   #include <cstdint> 36   #include <cstdint>
37   #include <mutex> 37   #include <mutex>
38   #include <vector> 38   #include <vector>
39   39  
40   #include <errno.h> 40   #include <errno.h>
41   #include <sys/epoll.h> 41   #include <sys/epoll.h>
42   #include <sys/eventfd.h> 42   #include <sys/eventfd.h>
43   #include <sys/timerfd.h> 43   #include <sys/timerfd.h>
44   #include <unistd.h> 44   #include <unistd.h>
45   45  
46   namespace boost::corosio::detail { 46   namespace boost::corosio::detail {
47   47  
48   /** Linux scheduler using epoll for I/O multiplexing. 48   /** Linux scheduler using epoll for I/O multiplexing.
49   49  
50   This scheduler implements the scheduler interface using Linux epoll 50   This scheduler implements the scheduler interface using Linux epoll
51   for efficient I/O event notification. It uses a single reactor model 51   for efficient I/O event notification. It uses a single reactor model
52   where one thread runs epoll_wait while other threads 52   where one thread runs epoll_wait while other threads
53   wait on a condition variable for handler work. This design provides: 53   wait on a condition variable for handler work. This design provides:
54   54  
55   - Handler parallelism: N posted handlers can execute on N threads 55   - Handler parallelism: N posted handlers can execute on N threads
56   - No thundering herd: condition_variable wakes exactly one thread 56   - No thundering herd: condition_variable wakes exactly one thread
57   - IOCP parity: Behavior matches Windows I/O completion port semantics 57   - IOCP parity: Behavior matches Windows I/O completion port semantics
58   58  
59   When threads call run(), they first try to execute queued handlers. 59   When threads call run(), they first try to execute queued handlers.
60   If the queue is empty and no reactor is running, one thread becomes 60   If the queue is empty and no reactor is running, one thread becomes
61   the reactor and runs epoll_wait. Other threads wait on a condition 61   the reactor and runs epoll_wait. Other threads wait on a condition
62   variable until handlers are available. 62   variable until handlers are available.
63   63  
64   @par Thread Safety 64   @par Thread Safety
65   All public member functions are thread-safe. 65   All public member functions are thread-safe.
66   */ 66   */
67   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler 67   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
68   { 68   {
69   public: 69   public:
70   /** Construct the scheduler. 70   /** Construct the scheduler.
71   71  
72   Creates an epoll instance, eventfd for reactor interruption, 72   Creates an epoll instance, eventfd for reactor interruption,
73   and timerfd for kernel-managed timer expiry. 73   and timerfd for kernel-managed timer expiry.
74   74  
75   @param ctx Reference to the owning execution_context. 75   @param ctx Reference to the owning execution_context.
76   @param concurrency_hint Hint for expected thread count (unused). 76   @param concurrency_hint Hint for expected thread count (unused).
77   */ 77   */
78   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 78   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
79   79  
80   /// Destroy the scheduler. 80   /// Destroy the scheduler.
81   ~epoll_scheduler() override; 81   ~epoll_scheduler() override;
82   82  
83   epoll_scheduler(epoll_scheduler const&) = delete; 83   epoll_scheduler(epoll_scheduler const&) = delete;
84   epoll_scheduler& operator=(epoll_scheduler const&) = delete; 84   epoll_scheduler& operator=(epoll_scheduler const&) = delete;
85   85  
86   /// Shut down the scheduler, draining pending operations. 86   /// Shut down the scheduler, draining pending operations.
87   void shutdown() override; 87   void shutdown() override;
88   88  
89   /// Apply runtime configuration, resizing the event buffer. 89   /// Apply runtime configuration, resizing the event buffer.
90   void configure_reactor( 90   void configure_reactor(
91   unsigned max_events, 91   unsigned max_events,
92   unsigned budget_init, 92   unsigned budget_init,
93   unsigned budget_max, 93   unsigned budget_max,
94   unsigned unassisted) override; 94   unsigned unassisted) override;
95   95  
96   /** Return the epoll file descriptor. 96   /** Return the epoll file descriptor.
97   97  
98   Used by socket services to register file descriptors 98   Used by socket services to register file descriptors
99   for I/O event notification. 99   for I/O event notification.
100   100  
101   @return The epoll file descriptor. 101   @return The epoll file descriptor.
102   */ 102   */
103   int epoll_fd() const noexcept 103   int epoll_fd() const noexcept
104   { 104   {
105   return epoll_fd_; 105   return epoll_fd_;
106   } 106   }
107   107  
108   /** Register a descriptor for persistent monitoring. 108   /** Register a descriptor for persistent monitoring.
109   109  
110   The fd is registered once and stays registered until explicitly 110   The fd is registered once and stays registered until explicitly
111   deregistered. Events are dispatched via reactor_descriptor_state which 111   deregistered. Events are dispatched via reactor_descriptor_state which
112   tracks pending read/write/connect operations. 112   tracks pending read/write/connect operations.
113   113  
114   @param fd The file descriptor to register. 114   @param fd The file descriptor to register.
115   @param desc Pointer to descriptor data (stored in epoll_event.data.ptr). 115   @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
116   116  
117   @return The error if registration fails, otherwise a default 117   @return The error if registration fails, otherwise a default
118   constructed error code. 118   constructed error code.
119   */ 119   */
120   std::error_code 120   std::error_code
121   register_descriptor(int fd, reactor_descriptor_state* desc) const; 121   register_descriptor(int fd, reactor_descriptor_state* desc) const;
122   122  
123   /** Deregister a persistently registered descriptor. 123   /** Deregister a persistently registered descriptor.
124   124  
125   @param fd The file descriptor to deregister. 125   @param fd The file descriptor to deregister.
126   */ 126   */
127   void deregister_descriptor(int fd) const; 127   void deregister_descriptor(int fd) const;
128   128  
129   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp). 129   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
HITCBC 130   69 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 130   69 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
131   { 131   {
HITCBC 132   69 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 132   69 return register_descriptor(read_fd, signal_pipe_reader_.arm());
133   } 133   }
134   134  
135   private: 135   private:
136   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; 136   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
137   void interrupt_reactor() const override; 137   void interrupt_reactor() const override;
138   void update_timerfd() const; 138   void update_timerfd() const;
139   139  
140   int epoll_fd_; 140   int epoll_fd_;
141   int event_fd_; 141   int event_fd_;
142   int timer_fd_; 142   int timer_fd_;
143   143  
144   // Watches the global signal self-pipe's read end (armed lazily by 144   // Watches the global signal self-pipe's read end (armed lazily by
145   // register_signal_reader on the first signal registration). 145   // register_signal_reader on the first signal registration).
146   reactor_signal_pipe_reader signal_pipe_reader_; 146   reactor_signal_pipe_reader signal_pipe_reader_;
147   147  
148   // Edge-triggered eventfd state 148   // Edge-triggered eventfd state
149   mutable std::atomic<bool> eventfd_armed_{false}; 149   mutable std::atomic<bool> eventfd_armed_{false};
150   150  
151   // Set when the earliest timer changes; flushed before epoll_wait 151   // Set when the earliest timer changes; flushed before epoll_wait
152   mutable std::atomic<bool> timerfd_stale_{false}; 152   mutable std::atomic<bool> timerfd_stale_{false};
153   153  
154   // Event buffer sized from max_events_per_poll_ (set at construction, 154   // Event buffer sized from max_events_per_poll_ (set at construction,
155   // resized by configure_reactor via io_context_options). 155   // resized by configure_reactor via io_context_options).
156   std::vector<epoll_event> event_buffer_; 156   std::vector<epoll_event> event_buffer_;
157   }; 157   };
158   158  
HITCBC 159   1232 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) 159   1232 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
HITCBC 160   1232 : epoll_fd_(-1) 160   1232 : epoll_fd_(-1)
HITCBC 161   1232 , event_fd_(-1) 161   1232 , event_fd_(-1)
HITCBC 162   1232 , timer_fd_(-1) 162   1232 , timer_fd_(-1)
HITCBC 163   2464 , event_buffer_(max_events_per_poll_) 163   2464 , event_buffer_(max_events_per_poll_)
164   { 164   {
HITCBC 165   1232 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC); 165   1232 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
HITCBC 166   1232 if (epoll_fd_ < 0) 166   1232 if (epoll_fd_ < 0)
HITCBC 167   1 detail::throw_system_error(make_err(errno), "epoll_create1"); 167   1 detail::throw_system_error(make_err(errno), "epoll_create1");
168   168  
HITCBC 169   1231 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); 169   1231 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
HITCBC 170   1231 if (event_fd_ < 0) 170   1231 if (event_fd_ < 0)
171   { 171   {
HITCBC 172   1 int errn = errno; 172   1 int errn = errno;
HITCBC 173   1 ::close(epoll_fd_); 173   1 ::close(epoll_fd_);
HITCBC 174   1 detail::throw_system_error(make_err(errn), "eventfd"); 174   1 detail::throw_system_error(make_err(errn), "eventfd");
175   } 175   }
176   176  
HITCBC 177   1230 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); 177   1230 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
HITCBC 178   1230 if (timer_fd_ < 0) 178   1230 if (timer_fd_ < 0)
179   { 179   {
HITCBC 180   1 int errn = errno; 180   1 int errn = errno;
HITCBC 181   1 ::close(event_fd_); 181   1 ::close(event_fd_);
HITCBC 182   1 ::close(epoll_fd_); 182   1 ::close(epoll_fd_);
HITCBC 183   1 detail::throw_system_error(make_err(errn), "timerfd_create"); 183   1 detail::throw_system_error(make_err(errn), "timerfd_create");
184   } 184   }
185   185  
HITCBC 186   1229 epoll_event ev{}; 186   1229 epoll_event ev{};
HITCBC 187   1229 ev.events = EPOLLIN | EPOLLET; 187   1229 ev.events = EPOLLIN | EPOLLET;
HITCBC 188   1229 ev.data.ptr = nullptr; 188   1229 ev.data.ptr = nullptr;
HITCBC 189   1229 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0) 189   1229 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
190   { 190   {
HITCBC 191   1 int errn = errno; 191   1 int errn = errno;
HITCBC 192   1 ::close(timer_fd_); 192   1 ::close(timer_fd_);
HITCBC 193   1 ::close(event_fd_); 193   1 ::close(event_fd_);
HITCBC 194   1 ::close(epoll_fd_); 194   1 ::close(epoll_fd_);
HITCBC 195   1 detail::throw_system_error(make_err(errn), "epoll_ctl"); 195   1 detail::throw_system_error(make_err(errn), "epoll_ctl");
196   } 196   }
197   197  
HITCBC 198   1228 epoll_event timer_ev{}; 198   1228 epoll_event timer_ev{};
HITCBC 199   1228 timer_ev.events = EPOLLIN | EPOLLERR; 199   1228 timer_ev.events = EPOLLIN | EPOLLERR;
HITCBC 200   1228 timer_ev.data.ptr = &timer_fd_; 200   1228 timer_ev.data.ptr = &timer_fd_;
HITCBC 201   1228 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0) 201   1228 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
202   { 202   {
HITCBC 203   1 int errn = errno; 203   1 int errn = errno;
HITCBC 204   1 ::close(timer_fd_); 204   1 ::close(timer_fd_);
HITCBC 205   1 ::close(event_fd_); 205   1 ::close(event_fd_);
HITCBC 206   1 ::close(epoll_fd_); 206   1 ::close(epoll_fd_);
HITCBC 207   1 detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)"); 207   1 detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
208   } 208   }
209   209  
HITCBC 210   1227 timer_svc_ = &get_timer_service(ctx, *this); 210   1227 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 211   1227 timer_svc_->set_on_earliest_changed( 211   1227 timer_svc_->set_on_earliest_changed(
HITCBC 212   5371 timer_service::callback(this, [](void* p) { 212   5476 timer_service::callback(this, [](void* p) {
HITCBC 213   4144 auto* self = static_cast<epoll_scheduler*>(p); 213   4249 auto* self = static_cast<epoll_scheduler*>(p);
HITCBC 214   4144 self->timerfd_stale_.store(true, std::memory_order_release); 214   4249 self->timerfd_stale_.store(true, std::memory_order_release);
HITCBC 215   4144 self->interrupt_reactor(); 215   4249 self->interrupt_reactor();
HITCBC 216   4144 })); 216   4249 }));
217   217  
HITCBC 218   1227 get_resolver_service(ctx, *this); 218   1227 get_resolver_service(ctx, *this);
HITCBC 219   1227 get_signal_service(ctx, *this); 219   1227 get_signal_service(ctx, *this);
HITCBC 220   1227 get_stream_file_service(ctx, *this); 220   1227 get_stream_file_service(ctx, *this);
HITCBC 221   1227 get_random_access_file_service(ctx, *this); 221   1227 get_random_access_file_service(ctx, *this);
222   222  
HITCBC 223   1227 completed_ops_.push(&task_op_); 223   1227 completed_ops_.push(&task_op_);
HITCBC 224   1242 } 224   1242 }
225   225  
HITCBC 226   2454 inline epoll_scheduler::~epoll_scheduler() 226   2454 inline epoll_scheduler::~epoll_scheduler()
227   { 227   {
HITCBC 228   1227 if (timer_fd_ >= 0) 228   1227 if (timer_fd_ >= 0)
HITCBC 229   1227 ::close(timer_fd_); 229   1227 ::close(timer_fd_);
HITCBC 230   1227 if (event_fd_ >= 0) 230   1227 if (event_fd_ >= 0)
HITCBC 231   1227 ::close(event_fd_); 231   1227 ::close(event_fd_);
HITCBC 232   1227 if (epoll_fd_ >= 0) 232   1227 if (epoll_fd_ >= 0)
HITCBC 233   1227 ::close(epoll_fd_); 233   1227 ::close(epoll_fd_);
HITCBC 234   2454 } 234   2454 }
235   235  
236   inline void 236   inline void
HITCBC 237   1227 epoll_scheduler::shutdown() 237   1227 epoll_scheduler::shutdown()
238   { 238   {
HITCBC 239   1227 shutdown_drain(); 239   1227 shutdown_drain();
240   240  
HITCBC 241   1227 if (event_fd_ >= 0) 241   1227 if (event_fd_ >= 0)
HITCBC 242   1227 interrupt_reactor(); 242   1227 interrupt_reactor();
HITCBC 243   1227 } 243   1227 }
244   244  
245   inline void 245   inline void
HITCBC 246   23 epoll_scheduler::configure_reactor( 246   23 epoll_scheduler::configure_reactor(
247   unsigned max_events, 247   unsigned max_events,
248   unsigned budget_init, 248   unsigned budget_init,
249   unsigned budget_max, 249   unsigned budget_max,
250   unsigned unassisted) 250   unsigned unassisted)
251   { 251   {
HITCBC 252   23 reactor_scheduler::configure_reactor( 252   23 reactor_scheduler::configure_reactor(
253   max_events, budget_init, budget_max, unassisted); 253   max_events, budget_init, budget_max, unassisted);
HITCBC 254   21 event_buffer_.resize(max_events_per_poll_); 254   21 event_buffer_.resize(max_events_per_poll_);
HITCBC 255   21 } 255   21 }
256   256  
257   inline std::error_code 257   inline std::error_code
HITCBC 258   5637 epoll_scheduler::register_descriptor( 258   5667 epoll_scheduler::register_descriptor(
259   int fd, reactor_descriptor_state* desc) const 259   int fd, reactor_descriptor_state* desc) const
260   { 260   {
HITCBC 261   5637 epoll_event ev{}; 261   5667 epoll_event ev{};
HITCBC 262   5637 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP; 262   5667 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
HITCBC 263   5637 ev.data.ptr = desc; 263   5667 ev.data.ptr = desc;
264   264  
HITCBC 265   5637 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) 265   5667 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
HITCBC 266   7 return make_err(errno); 266   7 return make_err(errno);
267   267  
HITCBC 268   5630 desc->registered_events = ev.events; 268   5660 desc->registered_events = ev.events;
HITCBC 269   5630 desc->fd = fd; 269   5660 desc->fd = fd;
HITCBC 270   5630 desc->scheduler_ = this; 270   5660 desc->scheduler_ = this;
HITCBC 271   5630 desc->mutex.set_enabled(reactor_io_locking_); 271   5660 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 272   5630 desc->ready_events_.store(0, std::memory_order_relaxed); 272   5660 desc->ready_events_.store(0, std::memory_order_relaxed);
273   273  
HITCBC 274   5630 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 274   5660 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 275   5630 desc->impl_ref_.reset(); 275   5660 desc->impl_ref_.reset();
HITCBC 276   5630 desc->read_ready = false; 276   5660 desc->read_ready = false;
HITCBC 277   5630 desc->write_ready = false; 277   5660 desc->write_ready = false;
HITCBC 278   5630 return {}; 278   5660 return {};
HITCBC 279   5630 } 279   5660 }
280   280  
281   inline void 281   inline void
HITCBC 282   5562 epoll_scheduler::deregister_descriptor(int fd) const 282   5592 epoll_scheduler::deregister_descriptor(int fd) const
283   { 283   {
HITCBC 284   5562 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr); 284   5592 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
HITCBC 285   5562 } 285   5592 }
286   286  
287   inline void 287   inline void
HITCBC 288   6557 epoll_scheduler::interrupt_reactor() const 288   6644 epoll_scheduler::interrupt_reactor() const
289   { 289   {
HITCBC 290   6557 bool expected = false; 290   6644 bool expected = false;
HITCBC 291   6557 if (eventfd_armed_.compare_exchange_strong( 291   6644 if (eventfd_armed_.compare_exchange_strong(
292   expected, true, std::memory_order_release, 292   expected, true, std::memory_order_release,
293   std::memory_order_relaxed)) 293   std::memory_order_relaxed))
294   { 294   {
HITCBC 295   5234 std::uint64_t val = 1; 295   5175 std::uint64_t val = 1;
HITCBC 296   5234 if (::write(event_fd_, &val, sizeof(val)) < 0) 296   5175 if (::write(event_fd_, &val, sizeof(val)) < 0)
297   { 297   {
298   // The flag is what coalesces later interrupts into a byte 298   // The flag is what coalesces later interrupts into a byte
299   // already in the eventfd; a write that failed put no byte 299   // already in the eventfd; a write that failed put no byte
300   // there, so leaving it armed would swallow every interrupt 300   // there, so leaving it armed would swallow every interrupt
301   // that follows. Disarming keeps the cost to the interrupts 301   // that follows. Disarming keeps the cost to the interrupts
302   // already in flight -- the next one arms and writes again, 302   // already in flight -- the next one arms and writes again,
303   // instead of every one after this coalescing into a byte 303   // instead of every one after this coalescing into a byte
304   // that does not exist. 304   // that does not exist.
HITCBC 305   2 eventfd_armed_.store(false, std::memory_order_release); 305   2 eventfd_armed_.store(false, std::memory_order_release);
306   } 306   }
307   } 307   }
HITCBC 308   6557 } 308   6644 }
309   309  
310   inline void 310   inline void
HITCBC 311   10794 epoll_scheduler::update_timerfd() const 311   10847 epoll_scheduler::update_timerfd() const
312   { 312   {
HITCBC 313   10794 auto nearest = timer_svc_->nearest_expiry(); 313   10847 auto nearest = timer_svc_->nearest_expiry();
314   314  
HITCBC 315   10794 itimerspec ts{}; 315   10847 itimerspec ts{};
HITCBC 316   10794 int flags = 0; 316   10847 int flags = 0;
317   317  
HITCBC 318   10794 if (nearest == timer_service::time_point::max()) 318   10847 if (nearest == timer_service::time_point::max())
319   { 319   {
320   // No timers — disarm by setting to 0 (relative) 320   // No timers — disarm by setting to 0 (relative)
321   } 321   }
322   else 322   else
323   { 323   {
HITCBC 324   9598 auto now = std::chrono::steady_clock::now(); 324   9707 auto now = std::chrono::steady_clock::now();
HITCBC 325   9598 if (nearest <= now) 325   9707 if (nearest <= now)
326   { 326   {
327   // Use 1ns instead of 0 — zero disarms the timerfd 327   // Use 1ns instead of 0 — zero disarms the timerfd
HITCBC 328   1170 ts.it_value.tv_nsec = 1; 328   1117 ts.it_value.tv_nsec = 1;
329   } 329   }
330   else 330   else
331   { 331   {
HITCBC 332   8428 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>( 332   8590 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
HITCBC 333   8428 nearest - now) 333   8590 nearest - now)
HITCBC 334   8428 .count(); 334   8590 .count();
HITCBC 335   8428 ts.it_value.tv_sec = nsec / 1000000000; 335   8590 ts.it_value.tv_sec = nsec / 1000000000;
HITCBC 336   8428 ts.it_value.tv_nsec = nsec % 1000000000; 336   8590 ts.it_value.tv_nsec = nsec % 1000000000;
HITCBC 337   8428 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0) 337   8590 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
MISUBC 338   ✗ ts.it_value.tv_nsec = 1; 338   ✗ ts.it_value.tv_nsec = 1;
339   } 339   }
340   } 340   }
341   341  
HITCBC 342   10794 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0) 342   10847 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
HITCBC 343   1 detail::throw_system_error(make_err(errno), "timerfd_settime"); 343   1 detail::throw_system_error(make_err(errno), "timerfd_settime");
HITCBC 344   10793 } 344   10846 }
345   345  
346   inline void 346   inline void
HITCBC 347   38473 epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 347   41216 epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
348   { 348   {
349   int timeout_ms; 349   int timeout_ms;
HITCBC 350   38473 if (task_interrupted_) 350   41216 if (task_interrupted_)
HITCBC 351   27469 timeout_ms = 0; 351   30377 timeout_ms = 0;
HITCBC 352   11004 else if (timeout_us < 0) 352   10839 else if (timeout_us < 0)
HITCBC 353   10988 timeout_ms = -1; 353   10824 timeout_ms = -1;
354   else 354   else
HITCBC 355   16 timeout_ms = static_cast<int>((timeout_us + 999) / 1000); 355   15 timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
356   356  
HITCBC 357   38473 if (lock.owns_lock()) 357   41216 if (lock.owns_lock())
HITCBC 358   11006 lock.unlock(); 358   10841 lock.unlock();
359   359  
HITCBC 360   38473 task_cleanup on_exit{this, &lock, ctx}; 360   41216 task_cleanup on_exit{this, &lock, ctx};
361   361  
362   // Flush deferred timerfd programming before blocking 362   // Flush deferred timerfd programming before blocking
HITCBC 363   38473 if (timerfd_stale_.exchange(false, std::memory_order_acquire)) 363   41216 if (timerfd_stale_.exchange(false, std::memory_order_acquire))
HITCBC 364   3642 update_timerfd(); 364   3599 update_timerfd();
365   365  
HITCBC 366   38472 int nfds = ::epoll_wait( 366   41215 int nfds = ::epoll_wait(
HITCBC 367   38472 epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()), 367   41215 epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
368   timeout_ms); 368   timeout_ms);
369   369  
HITCBC 370   38472 if (nfds < 0 && errno != EINTR) 370   41215 if (nfds < 0 && errno != EINTR)
HITCBC 371   1 detail::throw_system_error(make_err(errno), "epoll_wait"); 371   1 detail::throw_system_error(make_err(errno), "epoll_wait");
372   372  
HITCBC 373   38471 bool check_timers = false; 373   41214 bool check_timers = false;
HITCBC 374   38471 ready_queue local_ops; 374   41214 ready_queue local_ops;
375   375  
HITCBC 376   84209 for (int i = 0; i < nfds; ++i) 376   91237 for (int i = 0; i < nfds; ++i)
377   { 377   {
HITCBC 378   45738 if (event_buffer_[i].data.ptr == nullptr) 378   50023 if (event_buffer_[i].data.ptr == nullptr)
379   { 379   {
380   std::uint64_t val; 380   std::uint64_t val;
381   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 381   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
HITCBC 382   4005 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val)); 382   3946 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
HITCBC 383   4005 eventfd_armed_.store(false, std::memory_order_relaxed); 383   3946 eventfd_armed_.store(false, std::memory_order_relaxed);
HITCBC 384   4005 continue; 384   3946 continue;
HITCBC 385   4005 } 385   3946 }
386   386  
HITCBC 387   41733 if (event_buffer_[i].data.ptr == &timer_fd_) 387   46077 if (event_buffer_[i].data.ptr == &timer_fd_)
388   { 388   {
389   std::uint64_t expirations; 389   std::uint64_t expirations;
390   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 390   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
391   [[maybe_unused]] auto r = 391   [[maybe_unused]] auto r =
HITCBC 392   7152 ::read(timer_fd_, &expirations, sizeof(expirations)); 392   7248 ::read(timer_fd_, &expirations, sizeof(expirations));
HITCBC 393   7152 check_timers = true; 393   7248 check_timers = true;
HITCBC 394   7152 continue; 394   7248 continue;
HITCBC 395   7152 } 395   7248 }
396   396  
397   auto* desc = 397   auto* desc =
HITCBC 398   34581 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr); 398   38829 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
HITCBC 399   34581 desc->add_ready_events(event_buffer_[i].events); 399   38829 desc->add_ready_events(event_buffer_[i].events);
400   400  
HITCBC 401   34581 bool expected = false; 401   38829 bool expected = false;
HITCBC 402   34581 if (desc->is_enqueued_.compare_exchange_strong( 402   38829 if (desc->is_enqueued_.compare_exchange_strong(
403   expected, true, std::memory_order_release, 403   expected, true, std::memory_order_release,
404   std::memory_order_relaxed)) 404   std::memory_order_relaxed))
405   { 405   {
HITCBC 406   34581 local_ops.push(desc); 406   38829 local_ops.push(desc);
407   } 407   }
408   } 408   }
409   409  
HITCBC 410   38471 if (check_timers) 410   41214 if (check_timers)
411   { 411   {
HITCBC 412   7152 timer_svc_->process_expired(); 412   7248 timer_svc_->process_expired();
HITCBC 413   7152 update_timerfd(); 413   7248 update_timerfd();
414   } 414   }
415   415  
HITCBC 416   38471 lock.lock(); 416   41214 lock.lock();
417   417  
HITCBC 418   38471 completed_ops_.splice(local_ops); 418   41214 completed_ops_.splice(local_ops);
HITCBC 419   38473 } 419   41216 }
420   420  
421   } // namespace boost::corosio::detail 421   } // namespace boost::corosio::detail
422   422  
423   #endif // BOOST_COROSIO_HAS_EPOLL 423   #endif // BOOST_COROSIO_HAS_EPOLL
424   424  
425   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 425   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP