. /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   51 register_signal_reader(int read_fd) override 131   61 register_signal_reader(int read_fd) override
132   { 132   {
HITCBC 133   51 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 133   61 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   902 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) 162   1014 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
HITCBC 163   902 : epoll_fd_(-1) 163   1014 : epoll_fd_(-1)
HITCBC 164   902 , event_fd_(-1) 164   1014 , event_fd_(-1)
HITCBC 165   902 , timer_fd_(-1) 165   1014 , timer_fd_(-1)
HITCBC 166   1804 , event_buffer_(max_events_per_poll_) 166   2028 , event_buffer_(max_events_per_poll_)
167   { 167   {
HITCBC 168   902 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC); 168   1014 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
HITCBC 169   902 if (epoll_fd_ < 0) 169   1014 if (epoll_fd_ < 0)
HITGBC 170   ✗ detail::throw_system_error(make_err(errno), "epoll_create1"); 170   1 detail::throw_system_error(make_err(errno), "epoll_create1");
171   171  
HITCBC 172   902 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); 172   1013 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
HITCBC 173   902 if (event_fd_ < 0) 173   1013 if (event_fd_ < 0)
174   { 174   {
HITGBC 175   ✗ int errn = errno; 175   1 int errn = errno;
HITGBC 176   ✗ ::close(epoll_fd_); 176   1 ::close(epoll_fd_);
HITGBC 177   ✗ detail::throw_system_error(make_err(errn), "eventfd"); 177   1 detail::throw_system_error(make_err(errn), "eventfd");
178   } 178   }
179   179  
HITCBC 180   902 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); 180   1012 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
HITCBC 181   902 if (timer_fd_ < 0) 181   1012 if (timer_fd_ < 0)
182   { 182   {
HITGBC 183   ✗ int errn = errno; 183   1 int errn = errno;
HITGBC 184   ✗ ::close(event_fd_); 184   1 ::close(event_fd_);
HITGBC 185   ✗ ::close(epoll_fd_); 185   1 ::close(epoll_fd_);
HITGBC 186   ✗ 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   902 epoll_event ev{}; 189   1011 epoll_event ev{};
HITCBC 190   902 ev.events = EPOLLIN | EPOLLET; 190   1011 ev.events = EPOLLIN | EPOLLET;
HITCBC 191   902 ev.data.ptr = nullptr; 191   1011 ev.data.ptr = nullptr;
HITCBC 192   902 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0) 192   1011 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
193   { 193   {
HITGBC 194   ✗ int errn = errno; 194   1 int errn = errno;
HITGBC 195   ✗ ::close(timer_fd_); 195   1 ::close(timer_fd_);
HITGBC 196   ✗ ::close(event_fd_); 196   1 ::close(event_fd_);
HITGBC 197   ✗ ::close(epoll_fd_); 197   1 ::close(epoll_fd_);
HITGBC 198   ✗ 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   902 epoll_event timer_ev{}; 201   1010 epoll_event timer_ev{};
HITCBC 202   902 timer_ev.events = EPOLLIN | EPOLLERR; 202   1010 timer_ev.events = EPOLLIN | EPOLLERR;
HITCBC 203   902 timer_ev.data.ptr = &timer_fd_; 203   1010 timer_ev.data.ptr = &timer_fd_;
HITCBC 204   902 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0) 204   1010 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
205   { 205   {
HITGBC 206   ✗ int errn = errno; 206   1 int errn = errno;
HITGBC 207   ✗ ::close(timer_fd_); 207   1 ::close(timer_fd_);
HITGBC 208   ✗ ::close(event_fd_); 208   1 ::close(event_fd_);
HITGBC 209   ✗ ::close(epoll_fd_); 209   1 ::close(epoll_fd_);
HITGBC 210   ✗ 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   902 timer_svc_ = &get_timer_service(ctx, *this); 213   1009 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 214   902 timer_svc_->set_on_earliest_changed( 214   1009 timer_svc_->set_on_earliest_changed(
HITCBC 215   3924 timer_service::callback(this, [](void* p) { 215   5026 timer_service::callback(this, [](void* p) {
HITCBC 216   3022 auto* self = static_cast<epoll_scheduler*>(p); 216   4017 auto* self = static_cast<epoll_scheduler*>(p);
HITCBC 217   3022 self->timerfd_stale_.store(true, std::memory_order_release); 217   4017 self->timerfd_stale_.store(true, std::memory_order_release);
HITCBC 218   3022 self->interrupt_reactor(); 218   4017 self->interrupt_reactor();
HITCBC 219   3022 })); 219   4017 }));
220   220  
HITCBC 221   902 get_resolver_service(ctx, *this); 221   1009 get_resolver_service(ctx, *this);
HITCBC 222   902 get_signal_service(ctx, *this); 222   1009 get_signal_service(ctx, *this);
HITCBC 223   902 get_stream_file_service(ctx, *this); 223   1009 get_stream_file_service(ctx, *this);
HITCBC 224   902 get_random_access_file_service(ctx, *this); 224   1009 get_random_access_file_service(ctx, *this);
225   225  
HITCBC 226   902 completed_ops_.push(&task_op_); 226   1009 completed_ops_.push(&task_op_);
HITCBC 227   902 } 227   1024 }
228   228  
HITCBC 229   1804 inline epoll_scheduler::~epoll_scheduler() 229   2018 inline epoll_scheduler::~epoll_scheduler()
230   { 230   {
HITCBC 231   902 if (timer_fd_ >= 0) 231   1009 if (timer_fd_ >= 0)
HITCBC 232   902 ::close(timer_fd_); 232   1009 ::close(timer_fd_);
HITCBC 233   902 if (event_fd_ >= 0) 233   1009 if (event_fd_ >= 0)
HITCBC 234   902 ::close(event_fd_); 234   1009 ::close(event_fd_);
HITCBC 235   902 if (epoll_fd_ >= 0) 235   1009 if (epoll_fd_ >= 0)
HITCBC 236   902 ::close(epoll_fd_); 236   1009 ::close(epoll_fd_);
HITCBC 237   1804 } 237   2018 }
238   238  
239   inline void 239   inline void
HITCBC 240   902 epoll_scheduler::shutdown() 240   1009 epoll_scheduler::shutdown()
241   { 241   {
HITCBC 242   902 shutdown_drain(); 242   1009 shutdown_drain();
243   243  
HITCBC 244   902 if (event_fd_ >= 0) 244   1009 if (event_fd_ >= 0)
HITCBC 245   902 interrupt_reactor(); 245   1009 interrupt_reactor();
HITCBC 246   902 } 246   1009 }
247   247  
248   inline void 248   inline void
HITCBC 249   19 epoll_scheduler::configure_reactor( 249   19 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   19 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   18 event_buffer_.resize(max_events_per_poll_);
HITCBC 258   18 } 258   18 }
259   259  
260   inline std::error_code 260   inline std::error_code
HITCBC 261   5162 epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const 261   7099 epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const
262   { 262   {
HITCBC 263   5162 epoll_event ev{}; 263   7099 epoll_event ev{};
HITCBC 264   5162 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP; 264   7099 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
HITCBC 265   5162 ev.data.ptr = desc; 265   7099 ev.data.ptr = desc;
266   266  
HITCBC 267   5162 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) 267   7099 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
HITCBC 268   1 return make_err(errno); 268   7 return make_err(errno);
269   269  
HITCBC 270   5161 desc->registered_events = ev.events; 270   7092 desc->registered_events = ev.events;
HITCBC 271   5161 desc->fd = fd; 271   7092 desc->fd = fd;
HITCBC 272   5161 desc->scheduler_ = this; 272   7092 desc->scheduler_ = this;
HITCBC 273   5161 desc->mutex.set_enabled(reactor_io_locking_); 273   7092 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 274   5161 desc->ready_events_.store(0, std::memory_order_relaxed); 274   7092 desc->ready_events_.store(0, std::memory_order_relaxed);
275   275  
HITCBC 276   5161 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 276   7092 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 277   5161 desc->impl_ref_.reset(); 277   7092 desc->impl_ref_.reset();
HITCBC 278   5161 desc->read_ready = false; 278   7092 desc->read_ready = false;
HITCBC 279   5161 desc->write_ready = false; 279   7092 desc->write_ready = false;
HITCBC 280   5161 return {}; 280   7092 return {};
HITCBC 281   5161 } 281   7092 }
282   282  
283   inline void 283   inline void
HITCBC 284   5110 epoll_scheduler::deregister_descriptor(int fd) const 284   7032 epoll_scheduler::deregister_descriptor(int fd) const
285   { 285   {
HITCBC 286   5110 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr); 286   7032 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
HITCBC 287   5110 } 287   7032 }
288   288  
289   inline void 289   inline void
HITCBC 290   4755 epoll_scheduler::interrupt_reactor() const 290   5925 epoll_scheduler::interrupt_reactor() const
291   { 291   {
HITCBC 292   4755 bool expected = false; 292   5925 bool expected = false;
HITCBC 293   4755 if (eventfd_armed_.compare_exchange_strong( 293   5925 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 - 3590 std::uint64_t val = 1; 297 + 4637 std::uint64_t val = 1;
HITCBC 298 - 3590 [[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val)); 298 + 4637 if (::write(event_fd_, &val, sizeof(val)) < 0)
  299 + {
  300 + // The flag is what coalesces later interrupts into a byte
  301 + // already in the eventfd; a write that failed put no byte
  302 + // there, so leaving it armed would swallow every interrupt
  303 + // that follows. Disarming keeps the cost to the interrupts
  304 + // already in flight -- the next one arms and writes again,
  305 + // instead of every one after this coalescing into a byte
  306 + // that does not exist.
HITGNC   307 + 2 eventfd_armed_.store(false, std::memory_order_release);
  308 + }
299   } 309   }
HITCBC 300   4755 } 310   5925 }
301   311  
302   inline void 312   inline void
HITCBC 303   4873 epoll_scheduler::update_timerfd() const 313   6666 epoll_scheduler::update_timerfd() const
304   { 314   {
HITCBC 305   4873 auto nearest = timer_svc_->nearest_expiry(); 315   6666 auto nearest = timer_svc_->nearest_expiry();
306   316  
HITCBC 307   4873 itimerspec ts{}; 317   6666 itimerspec ts{};
HITCBC 308   4873 int flags = 0; 318   6666 int flags = 0;
309   319  
HITCBC 310   4873 if (nearest == timer_service::time_point::max()) 320   6666 if (nearest == timer_service::time_point::max())
311   { 321   {
312   // No timers — disarm by setting to 0 (relative) 322   // No timers — disarm by setting to 0 (relative)
313   } 323   }
314   else 324   else
315   { 325   {
HITCBC 316   4694 auto now = std::chrono::steady_clock::now(); 326   6484 auto now = std::chrono::steady_clock::now();
HITCBC 317   4694 if (nearest <= now) 327   6484 if (nearest <= now)
318   { 328   {
319   // Use 1ns instead of 0 — zero disarms the timerfd 329   // Use 1ns instead of 0 — zero disarms the timerfd
HITCBC 320   354 ts.it_value.tv_nsec = 1; 330   363 ts.it_value.tv_nsec = 1;
321   } 331   }
322   else 332   else
323   { 333   {
HITCBC 324   4340 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>( 334   6121 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
HITCBC 325   4340 nearest - now) 335   6121 nearest - now)
HITCBC 326   4340 .count(); 336   6121 .count();
HITCBC 327   4340 ts.it_value.tv_sec = nsec / 1000000000; 337   6121 ts.it_value.tv_sec = nsec / 1000000000;
HITCBC 328   4340 ts.it_value.tv_nsec = nsec % 1000000000; 338   6121 ts.it_value.tv_nsec = nsec % 1000000000;
HITCBC 329   4340 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0) 339   6121 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
MISUBC 330   ✗ ts.it_value.tv_nsec = 1; 340   ✗ ts.it_value.tv_nsec = 1;
331   } 341   }
332   } 342   }
333   343  
HITCBC 334   4873 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0) 344   6666 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
HITGBC 335   ✗ detail::throw_system_error(make_err(errno), "timerfd_settime"); 345   1 detail::throw_system_error(make_err(errno), "timerfd_settime");
HITCBC 336   4873 } 346   6665 }
337   347  
338   inline void 348   inline void
HITCBC 339   32432 epoll_scheduler::run_task( 349   35780 epoll_scheduler::run_task(
340   lock_type& lock, context_type* ctx, long timeout_us) 350   lock_type& lock, context_type* ctx, long timeout_us)
341   { 351   {
342   int timeout_ms; 352   int timeout_ms;
HITCBC 343   32432 if (task_interrupted_) 353   35780 if (task_interrupted_)
HITCBC 344   25592 timeout_ms = 0; 354   26200 timeout_ms = 0;
HITCBC 345   6840 else if (timeout_us < 0) 355   9580 else if (timeout_us < 0)
HITCBC 346   6836 timeout_ms = -1; 356   9576 timeout_ms = -1;
347   else 357   else
HITCBC 348   4 timeout_ms = static_cast<int>((timeout_us + 999) / 1000); 358   4 timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
349   359  
HITCBC 350   32432 if (lock.owns_lock()) 360   35780 if (lock.owns_lock())
HITCBC 351   6842 lock.unlock(); 361   9582 lock.unlock();
352   362  
HITCBC 353   32432 task_cleanup on_exit{this, &lock, ctx}; 363   35780 task_cleanup on_exit{this, &lock, ctx};
354   364  
355   // Flush deferred timerfd programming before blocking 365   // Flush deferred timerfd programming before blocking
HITCBC 356   32432 if (timerfd_stale_.exchange(false, std::memory_order_acquire)) 366   35780 if (timerfd_stale_.exchange(false, std::memory_order_acquire))
HITCBC 357   2444 update_timerfd(); 367   3342 update_timerfd();
358   368  
HITCBC 359   32432 int nfds = ::epoll_wait( 369   35779 int nfds = ::epoll_wait(
360   epoll_fd_, event_buffer_.data(), 370   epoll_fd_, event_buffer_.data(),
HITCBC 361   32432 static_cast<int>(event_buffer_.size()), timeout_ms); 371   35779 static_cast<int>(event_buffer_.size()), timeout_ms);
362   372  
HITCBC 363   32432 if (nfds < 0 && errno != EINTR) 373   35779 if (nfds < 0 && errno != EINTR)
HITGBC 364   ✗ detail::throw_system_error(make_err(errno), "epoll_wait"); 374   1 detail::throw_system_error(make_err(errno), "epoll_wait");
365   375  
HITCBC 366   32432 bool check_timers = false; 376   35778 bool check_timers = false;
HITCBC 367   32432 ready_queue local_ops; 377   35778 ready_queue local_ops;
368   378  
HITCBC 369   74644 for (int i = 0; i < nfds; ++i) 379   80571 for (int i = 0; i < nfds; ++i)
370   { 380   {
HITCBC 371   42212 if (event_buffer_[i].data.ptr == nullptr) 381   44793 if (event_buffer_[i].data.ptr == nullptr)
372   { 382   {
373   std::uint64_t val; 383   std::uint64_t val;
374   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 384   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
HITCBC 375   2688 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val)); 385   3626 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
HITCBC 376   2688 eventfd_armed_.store(false, std::memory_order_relaxed); 386   3626 eventfd_armed_.store(false, std::memory_order_relaxed);
HITCBC 377   2688 continue; 387   3626 continue;
HITCBC 378   2688 } 388   3626 }
379   389  
HITCBC 380   39524 if (event_buffer_[i].data.ptr == &timer_fd_) 390   41167 if (event_buffer_[i].data.ptr == &timer_fd_)
381   { 391   {
382   std::uint64_t expirations; 392   std::uint64_t expirations;
383   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 393   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
384   [[maybe_unused]] auto r = 394   [[maybe_unused]] auto r =
HITCBC 385   2429 ::read(timer_fd_, &expirations, sizeof(expirations)); 395   3324 ::read(timer_fd_, &expirations, sizeof(expirations));
HITCBC 386   2429 check_timers = true; 396   3324 check_timers = true;
HITCBC 387   2429 continue; 397   3324 continue;
HITCBC 388   2429 } 398   3324 }
389   399  
390   auto* desc = 400   auto* desc =
HITCBC 391   37095 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr); 401   37843 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
HITCBC 392   37095 desc->add_ready_events(event_buffer_[i].events); 402   37843 desc->add_ready_events(event_buffer_[i].events);
393   403  
HITCBC 394   37095 bool expected = false; 404   37843 bool expected = false;
HITCBC 395   37095 if (desc->is_enqueued_.compare_exchange_strong( 405   37843 if (desc->is_enqueued_.compare_exchange_strong(
396   expected, true, std::memory_order_release, 406   expected, true, std::memory_order_release,
397   std::memory_order_relaxed)) 407   std::memory_order_relaxed))
398   { 408   {
HITCBC 399   37095 local_ops.push(desc); 409   37843 local_ops.push(desc);
400   } 410   }
401   } 411   }
402   412  
HITCBC 403   32432 if (check_timers) 413   35778 if (check_timers)
404   { 414   {
HITCBC 405   2429 timer_svc_->process_expired(); 415   3324 timer_svc_->process_expired();
HITCBC 406   2429 update_timerfd(); 416   3324 update_timerfd();
407   } 417   }
408   418  
HITCBC 409   32432 lock.lock(); 419   35778 lock.lock();
410   420  
HITCBC 411   32432 completed_ops_.splice(local_ops); 421   35778 completed_ops_.splice(local_ops);
HITCBC 412   32432 } 422   35780 }
413   423  
414   } // namespace boost::corosio::detail 424   } // namespace boost::corosio::detail
415   425  
416   #endif // BOOST_COROSIO_HAS_EPOLL 426   #endif // BOOST_COROSIO_HAS_EPOLL
417   427  
418   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 428   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP