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