. /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).
130   [[nodiscard]] std::error_code 130   [[nodiscard]] std::error_code
HITCBC 131   61 register_signal_reader(int read_fd) override 131   69 register_signal_reader(int read_fd) override
132   { 132   {
HITCBC 133   61 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 133   69 return register_descriptor(read_fd, signal_pipe_reader_.arm());
134   } 134   }
135   135  
136   private: 136   private:
137   void 137   void
138 - run_task(lock_type& lock, context_type* ctx, 138 + run_task(lock_type& lock, context_type& ctx,
139   long timeout_us) override; 139   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   1014 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) 162   1226 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
HITCBC 163   1014 : epoll_fd_(-1) 163   1226 : epoll_fd_(-1)
HITCBC 164   1014 , event_fd_(-1) 164   1226 , event_fd_(-1)
HITCBC 165   1014 , timer_fd_(-1) 165   1226 , timer_fd_(-1)
HITCBC 166   2028 , event_buffer_(max_events_per_poll_) 166   2452 , event_buffer_(max_events_per_poll_)
167   { 167   {
HITCBC 168   1014 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC); 168   1226 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
HITCBC 169   1014 if (epoll_fd_ < 0) 169   1226 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   1013 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); 172   1225 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
HITCBC 173   1013 if (event_fd_ < 0) 173   1225 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   1012 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); 180   1224 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
HITCBC 181   1012 if (timer_fd_ < 0) 181   1224 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   1011 epoll_event ev{}; 189   1223 epoll_event ev{};
HITCBC 190   1011 ev.events = EPOLLIN | EPOLLET; 190   1223 ev.events = EPOLLIN | EPOLLET;
HITCBC 191   1011 ev.data.ptr = nullptr; 191   1223 ev.data.ptr = nullptr;
HITCBC 192   1011 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0) 192   1223 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   1010 epoll_event timer_ev{}; 201   1222 epoll_event timer_ev{};
HITCBC 202   1010 timer_ev.events = EPOLLIN | EPOLLERR; 202   1222 timer_ev.events = EPOLLIN | EPOLLERR;
HITCBC 203   1010 timer_ev.data.ptr = &timer_fd_; 203   1222 timer_ev.data.ptr = &timer_fd_;
HITCBC 204   1010 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0) 204   1222 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   1009 timer_svc_ = &get_timer_service(ctx, *this); 213   1221 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 214   1009 timer_svc_->set_on_earliest_changed( 214   1221 timer_svc_->set_on_earliest_changed(
HITCBC 215   4776 timer_service::callback(this, [](void* p) { 215   7293 timer_service::callback(this, [](void* p) {
HITCBC 216   3767 auto* self = static_cast<epoll_scheduler*>(p); 216   6072 auto* self = static_cast<epoll_scheduler*>(p);
HITCBC 217   3767 self->timerfd_stale_.store(true, std::memory_order_release); 217   6072 self->timerfd_stale_.store(true, std::memory_order_release);
HITCBC 218   3767 self->interrupt_reactor(); 218   6072 self->interrupt_reactor();
HITCBC 219   3767 })); 219   6072 }));
220   220  
HITCBC 221   1009 get_resolver_service(ctx, *this); 221   1221 get_resolver_service(ctx, *this);
HITCBC 222   1009 get_signal_service(ctx, *this); 222   1221 get_signal_service(ctx, *this);
HITCBC 223   1009 get_stream_file_service(ctx, *this); 223   1221 get_stream_file_service(ctx, *this);
HITCBC 224   1009 get_random_access_file_service(ctx, *this); 224   1221 get_random_access_file_service(ctx, *this);
225   225  
HITCBC 226   1009 completed_ops_.push(&task_op_); 226   1221 completed_ops_.push(&task_op_);
HITCBC 227   1024 } 227   1236 }
228   228  
HITCBC 229   2018 inline epoll_scheduler::~epoll_scheduler() 229   2442 inline epoll_scheduler::~epoll_scheduler()
230   { 230   {
HITCBC 231   1009 if (timer_fd_ >= 0) 231   1221 if (timer_fd_ >= 0)
HITCBC 232   1009 ::close(timer_fd_); 232   1221 ::close(timer_fd_);
HITCBC 233   1009 if (event_fd_ >= 0) 233   1221 if (event_fd_ >= 0)
HITCBC 234   1009 ::close(event_fd_); 234   1221 ::close(event_fd_);
HITCBC 235   1009 if (epoll_fd_ >= 0) 235   1221 if (epoll_fd_ >= 0)
HITCBC 236   1009 ::close(epoll_fd_); 236   1221 ::close(epoll_fd_);
HITCBC 237   2018 } 237   2442 }
238   238  
239   inline void 239   inline void
HITCBC 240   1009 epoll_scheduler::shutdown() 240   1221 epoll_scheduler::shutdown()
241   { 241   {
HITCBC 242   1009 shutdown_drain(); 242   1221 shutdown_drain();
243   243  
HITCBC 244   1009 if (event_fd_ >= 0) 244   1221 if (event_fd_ >= 0)
HITCBC 245   1009 interrupt_reactor(); 245   1221 interrupt_reactor();
HITCBC 246   1009 } 246   1221 }
247   247  
248   inline void 248   inline void
HITCBC 249   19 epoll_scheduler::configure_reactor( 249   23 epoll_scheduler::configure_reactor(
250   unsigned max_events, 250   unsigned max_events,
251   unsigned budget_init, 251   unsigned budget_init,
252   unsigned budget_max, 252   unsigned budget_max,
253   unsigned unassisted) 253   unsigned unassisted)
254   { 254   {
HITCBC 255   19 reactor_scheduler::configure_reactor( 255   23 reactor_scheduler::configure_reactor(
256   max_events, budget_init, budget_max, unassisted); 256   max_events, budget_init, budget_max, unassisted);
HITCBC 257   18 event_buffer_.resize(max_events_per_poll_); 257   21 event_buffer_.resize(max_events_per_poll_);
HITCBC 258   18 } 258   21 }
259   259  
260   inline std::error_code 260   inline std::error_code
HITCBC 261   6935 epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const 261   9487 epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const
262   { 262   {
HITCBC 263   6935 epoll_event ev{}; 263   9487 epoll_event ev{};
HITCBC 264   6935 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP; 264   9487 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
HITCBC 265   6935 ev.data.ptr = desc; 265   9487 ev.data.ptr = desc;
266   266  
HITCBC 267   6935 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) 267   9487 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
HITCBC 268   7 return make_err(errno); 268   7 return make_err(errno);
269   269  
HITCBC 270   6928 desc->registered_events = ev.events; 270   9480 desc->registered_events = ev.events;
HITCBC 271   6928 desc->fd = fd; 271   9480 desc->fd = fd;
HITCBC 272   6928 desc->scheduler_ = this; 272   9480 desc->scheduler_ = this;
HITCBC 273   6928 desc->mutex.set_enabled(reactor_io_locking_); 273   9480 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 274   6928 desc->ready_events_.store(0, std::memory_order_relaxed); 274   9480 desc->ready_events_.store(0, std::memory_order_relaxed);
275   275  
HITCBC 276   6928 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 276   9480 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 277   6928 desc->impl_ref_.reset(); 277   9480 desc->impl_ref_.reset();
HITCBC 278   6928 desc->read_ready = false; 278   9480 desc->read_ready = false;
HITCBC 279   6928 desc->write_ready = false; 279   9480 desc->write_ready = false;
HITCBC 280   6928 return {}; 280   9480 return {};
HITCBC 281   6928 } 281   9480 }
282   282  
283   inline void 283   inline void
HITCBC 284   6868 epoll_scheduler::deregister_descriptor(int fd) const 284   9412 epoll_scheduler::deregister_descriptor(int fd) const
285   { 285   {
HITCBC 286   6868 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr); 286   9412 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
HITCBC 287   6868 } 287   9412 }
288   288  
289   inline void 289   inline void
HITCBC 290   5685 epoll_scheduler::interrupt_reactor() const 290   8464 epoll_scheduler::interrupt_reactor() const
291   { 291   {
HITCBC 292   5685 bool expected = false; 292   8464 bool expected = false;
HITCBC 293   5685 if (eventfd_armed_.compare_exchange_strong( 293   8464 if (eventfd_armed_.compare_exchange_strong(
294   expected, true, std::memory_order_release, 294   expected, true, std::memory_order_release,
295   std::memory_order_relaxed)) 295   std::memory_order_relaxed))
296   { 296   {
HITCBC 297   4577 std::uint64_t val = 1; 297   7034 std::uint64_t val = 1;
HITCBC 298   4577 if (::write(event_fd_, &val, sizeof(val)) < 0) 298   7034 if (::write(event_fd_, &val, sizeof(val)) < 0)
299   { 299   {
300   // The flag is what coalesces later interrupts into a byte 300   // The flag is what coalesces later interrupts into a byte
301   // already in the eventfd; a write that failed put no byte 301   // already in the eventfd; a write that failed put no byte
302   // there, so leaving it armed would swallow every interrupt 302   // there, so leaving it armed would swallow every interrupt
303   // that follows. Disarming keeps the cost to the interrupts 303   // that follows. Disarming keeps the cost to the interrupts
304   // already in flight -- the next one arms and writes again, 304   // already in flight -- the next one arms and writes again,
305   // instead of every one after this coalescing into a byte 305   // instead of every one after this coalescing into a byte
306   // that does not exist. 306   // that does not exist.
HITCBC 307   2 eventfd_armed_.store(false, std::memory_order_release); 307   2 eventfd_armed_.store(false, std::memory_order_release);
308   } 308   }
309   } 309   }
HITCBC 310   5685 } 310   8464 }
311   311  
312   inline void 312   inline void
HITCBC 313   6530 epoll_scheduler::update_timerfd() const 313   14962 epoll_scheduler::update_timerfd() const
314   { 314   {
HITCBC 315   6530 auto nearest = timer_svc_->nearest_expiry(); 315   14962 auto nearest = timer_svc_->nearest_expiry();
316   316  
HITCBC 317   6530 itimerspec ts{}; 317   14962 itimerspec ts{};
HITCBC 318   6530 int flags = 0; 318   14962 int flags = 0;
319   319  
HITCBC 320   6530 if (nearest == timer_service::time_point::max()) 320   14962 if (nearest == timer_service::time_point::max())
321   { 321   {
322   // No timers — disarm by setting to 0 (relative) 322   // No timers — disarm by setting to 0 (relative)
323   } 323   }
324   else 324   else
325   { 325   {
HITCBC 326   6333 auto now = std::chrono::steady_clock::now(); 326   13819 auto now = std::chrono::steady_clock::now();
HITCBC 327   6333 if (nearest <= now) 327   13819 if (nearest <= now)
328   { 328   {
329   // Use 1ns instead of 0 — zero disarms the timerfd 329   // Use 1ns instead of 0 — zero disarms the timerfd
HITCBC 330   636 ts.it_value.tv_nsec = 1; 330   1246 ts.it_value.tv_nsec = 1;
331   } 331   }
332   else 332   else
333   { 333   {
HITCBC 334   5697 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>( 334   12573 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
HITCBC 335   5697 nearest - now) 335   12573 nearest - now)
HITCBC 336   5697 .count(); 336   12573 .count();
HITCBC 337   5697 ts.it_value.tv_sec = nsec / 1000000000; 337   12573 ts.it_value.tv_sec = nsec / 1000000000;
HITCBC 338   5697 ts.it_value.tv_nsec = nsec % 1000000000; 338   12573 ts.it_value.tv_nsec = nsec % 1000000000;
HITCBC 339   5697 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0) 339   12573 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
MISUBC 340   ✗ ts.it_value.tv_nsec = 1; 340   ✗ ts.it_value.tv_nsec = 1;
341   } 341   }
342   } 342   }
343   343  
HITCBC 344   6530 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0) 344   14962 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
HITCBC 345   1 detail::throw_system_error(make_err(errno), "timerfd_settime"); 345   1 detail::throw_system_error(make_err(errno), "timerfd_settime");
HITCBC 346   6529 } 346   14961 }
347   347  
348   inline void 348   inline void
HITCBC 349   34164 epoll_scheduler::run_task( 349   47003 epoll_scheduler::run_task(
350 - lock_type& lock, context_type* ctx, long timeout_us) 350 + lock_type& lock, context_type& ctx, long timeout_us)
351   { 351   {
352   int timeout_ms; 352   int timeout_ms;
HITCBC 353   34164 if (task_interrupted_) 353   47003 if (task_interrupted_)
HITCBC 354   25154 timeout_ms = 0; 354   30232 timeout_ms = 0;
HITCBC 355   9010 else if (timeout_us < 0) 355   16771 else if (timeout_us < 0)
HITCBC 356   9006 timeout_ms = -1; 356   16756 timeout_ms = -1;
357   else 357   else
HITCBC 358   4 timeout_ms = static_cast<int>((timeout_us + 999) / 1000); 358   15 timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
359   359  
HITCBC 360   34164 if (lock.owns_lock()) 360   47003 if (lock.owns_lock())
HITCBC 361   9012 lock.unlock(); 361   16773 lock.unlock();
362   362  
HITCBC 363   34164 task_cleanup on_exit{this, &lock, ctx}; 363   47003 task_cleanup on_exit{this, &lock, ctx};
364   364  
365   // Flush deferred timerfd programming before blocking 365   // Flush deferred timerfd programming before blocking
HITCBC 366   34164 if (timerfd_stale_.exchange(false, std::memory_order_acquire)) 366   47003 if (timerfd_stale_.exchange(false, std::memory_order_acquire))
HITCBC 367   3274 update_timerfd(); 367   5457 update_timerfd();
368   368  
HITCBC 369   34163 int nfds = ::epoll_wait( 369   47002 int nfds = ::epoll_wait(
370   epoll_fd_, event_buffer_.data(), 370   epoll_fd_, event_buffer_.data(),
HITCBC 371   34163 static_cast<int>(event_buffer_.size()), timeout_ms); 371   47002 static_cast<int>(event_buffer_.size()), timeout_ms);
372   372  
HITCBC 373   34163 if (nfds < 0 && errno != EINTR) 373   47002 if (nfds < 0 && errno != EINTR)
HITCBC 374   1 detail::throw_system_error(make_err(errno), "epoll_wait"); 374   1 detail::throw_system_error(make_err(errno), "epoll_wait");
375   375  
HITCBC 376   34162 bool check_timers = false; 376   47001 bool check_timers = false;
HITCBC 377   34162 ready_queue local_ops; 377   47001 ready_queue local_ops;
378   378  
HITCBC 379   78174 for (int i = 0; i < nfds; ++i) 379   100534 for (int i = 0; i < nfds; ++i)
380   { 380   {
HITCBC 381   44012 if (event_buffer_[i].data.ptr == nullptr) 381   53533 if (event_buffer_[i].data.ptr == nullptr)
382   { 382   {
383   std::uint64_t val; 383   std::uint64_t val;
384   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 384   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
HITCBC 385   3566 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val)); 385   5811 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
HITCBC 386   3566 eventfd_armed_.store(false, std::memory_order_relaxed); 386   5811 eventfd_armed_.store(false, std::memory_order_relaxed);
HITCBC 387   3566 continue; 387   5811 continue;
HITCBC 388   3566 } 388   5811 }
389   389  
HITCBC 390   40446 if (event_buffer_[i].data.ptr == &timer_fd_) 390   47722 if (event_buffer_[i].data.ptr == &timer_fd_)
391   { 391   {
392   std::uint64_t expirations; 392   std::uint64_t expirations;
393   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 393   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
394   [[maybe_unused]] auto r = 394   [[maybe_unused]] auto r =
HITCBC 395   3256 ::read(timer_fd_, &expirations, sizeof(expirations)); 395   9505 ::read(timer_fd_, &expirations, sizeof(expirations));
HITCBC 396   3256 check_timers = true; 396   9505 check_timers = true;
HITCBC 397   3256 continue; 397   9505 continue;
HITCBC 398   3256 } 398   9505 }
399   399  
400   auto* desc = 400   auto* desc =
HITCBC 401   37190 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr); 401   38217 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
HITCBC 402   37190 desc->add_ready_events(event_buffer_[i].events); 402   38217 desc->add_ready_events(event_buffer_[i].events);
403   403  
HITCBC 404   37190 bool expected = false; 404   38217 bool expected = false;
HITCBC 405   37190 if (desc->is_enqueued_.compare_exchange_strong( 405   38217 if (desc->is_enqueued_.compare_exchange_strong(
406   expected, true, std::memory_order_release, 406   expected, true, std::memory_order_release,
407   std::memory_order_relaxed)) 407   std::memory_order_relaxed))
408   { 408   {
HITCBC 409   37190 local_ops.push(desc); 409   38217 local_ops.push(desc);
410   } 410   }
411   } 411   }
412   412  
HITCBC 413   34162 if (check_timers) 413   47001 if (check_timers)
414   { 414   {
HITCBC 415   3256 timer_svc_->process_expired(); 415   9505 timer_svc_->process_expired();
HITCBC 416   3256 update_timerfd(); 416   9505 update_timerfd();
417   } 417   }
418   418  
HITCBC 419   34162 lock.lock(); 419   47001 lock.lock();
420   420  
HITCBC 421   34162 completed_ops_.splice(local_ops); 421   47001 completed_ops_.splice(local_ops);
HITCBC 422   34164 } 422   47003 }
423   423  
424   } // namespace boost::corosio::detail 424   } // namespace boost::corosio::detail
425   425  
426   #endif // BOOST_COROSIO_HAS_EPOLL 426   #endif // BOOST_COROSIO_HAS_EPOLL
427   427  
428   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 428   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP