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