. /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 <limits>  
33   #include <cstdint> 32   #include <cstdint>
34   #include <mutex> 33   #include <mutex>
35   #include <vector> 34   #include <vector>
36   35  
37   #include <errno.h> 36   #include <errno.h>
38   #include <sys/epoll.h> 37   #include <sys/epoll.h>
39   #include <sys/eventfd.h> 38   #include <sys/eventfd.h>
40   #include <sys/timerfd.h> 39   #include <sys/timerfd.h>
41   #include <unistd.h> 40   #include <unistd.h>
42   41  
43   namespace boost::corosio::detail { 42   namespace boost::corosio::detail {
44   43  
45   /** Linux scheduler using epoll for I/O multiplexing. 44   /** Linux scheduler using epoll for I/O multiplexing.
46   45  
47   This scheduler implements the scheduler interface using Linux epoll 46   This scheduler implements the scheduler interface using Linux epoll
48   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
49   where one thread runs epoll_wait while other threads 48   where one thread runs epoll_wait while other threads
50   wait on a condition variable for handler work. This design provides: 49   wait on a condition variable for handler work. This design provides:
51   50  
52   - Handler parallelism: N posted handlers can execute on N threads 51   - Handler parallelism: N posted handlers can execute on N threads
53   - No thundering herd: condition_variable wakes exactly one thread 52   - No thundering herd: condition_variable wakes exactly one thread
54   - IOCP parity: Behavior matches Windows I/O completion port semantics 53   - IOCP parity: Behavior matches Windows I/O completion port semantics
55   54  
56   When threads call run(), they first try to execute queued handlers. 55   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 56   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 57   the reactor and runs epoll_wait. Other threads wait on a condition
59   variable until handlers are available. 58   variable until handlers are available.
60   59  
61   @par Thread Safety 60   @par Thread Safety
62   All public member functions are thread-safe. 61   All public member functions are thread-safe.
63   */ 62   */
64   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler 63   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
65   { 64   {
66   public: 65   public:
67   /** Construct the scheduler. 66   /** Construct the scheduler.
68   67  
69 - Creates an epoll instance and an eventfd for reactor 68 + Creates an epoll instance, eventfd for reactor interruption,
70 - interruption. The timerfd for timer expiry is created with the 69 + and timerfd for kernel-managed timer expiry.
71 - context's first timer.  
72   70  
73   @param ctx Reference to the owning execution_context. 71   @param ctx Reference to the owning execution_context.
74   @param concurrency_hint Hint for expected thread count (unused). 72   @param concurrency_hint Hint for expected thread count (unused).
75   */ 73   */
76   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 74   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
77   75  
78   /// Destroy the scheduler. 76   /// Destroy the scheduler.
79   ~epoll_scheduler() override; 77   ~epoll_scheduler() override;
80   78  
81   epoll_scheduler(epoll_scheduler const&) = delete; 79   epoll_scheduler(epoll_scheduler const&) = delete;
82   epoll_scheduler& operator=(epoll_scheduler const&) = delete; 80   epoll_scheduler& operator=(epoll_scheduler const&) = delete;
83   81  
84   /// Shut down the scheduler, draining pending operations. 82   /// Shut down the scheduler, draining pending operations.
85   void shutdown() override; 83   void shutdown() override;
86   84  
87   /// Apply runtime configuration, resizing the event buffer. 85   /// Apply runtime configuration, resizing the event buffer.
88   void configure_reactor( 86   void configure_reactor(
89   unsigned max_events, 87   unsigned max_events,
90   unsigned budget_init, 88   unsigned budget_init,
91   unsigned budget_max, 89   unsigned budget_max,
92   unsigned unassisted) override; 90   unsigned unassisted) override;
93   91  
94   /** Return the epoll file descriptor. 92   /** Return the epoll file descriptor.
95   93  
96   Used by socket services to register file descriptors 94   Used by socket services to register file descriptors
97   for I/O event notification. 95   for I/O event notification.
98   96  
99   @return The epoll file descriptor. 97   @return The epoll file descriptor.
100   */ 98   */
101   int epoll_fd() const noexcept 99   int epoll_fd() const noexcept
102   { 100   {
103   return epoll_fd_; 101   return epoll_fd_;
104   } 102   }
105   103  
106   /** Register a descriptor for persistent monitoring. 104   /** Register a descriptor for persistent monitoring.
107   105  
108   The fd is registered once and stays registered until explicitly 106   The fd is registered once and stays registered until explicitly
109   deregistered. Events are dispatched via reactor_descriptor_state which 107   deregistered. Events are dispatched via reactor_descriptor_state which
110   tracks pending read/write/connect operations. 108   tracks pending read/write/connect operations.
111   109  
112   @param fd The file descriptor to register. 110   @param fd The file descriptor to register.
113   @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).
114   112  
115   @return The error if registration fails, otherwise a default 113   @return The error if registration fails, otherwise a default
116   constructed error code. 114   constructed error code.
117   */ 115   */
118   std::error_code 116   std::error_code
119   register_descriptor(int fd, reactor_descriptor_state* desc) const; 117   register_descriptor(int fd, reactor_descriptor_state* desc) const;
120   118  
121   /// No-op: write readiness is watched from registration on. 119   /// No-op: write readiness is watched from registration on.
122   std::error_code 120   std::error_code
HITCBC 123   5637 ensure_write_registered(int, reactor_descriptor_state*) const noexcept 121   5188 ensure_write_registered(int, reactor_descriptor_state*) const noexcept
124   { 122   {
HITCBC 125   5637 return {}; 123   5188 return {};
126   } 124   }
127   125  
128   /** Deregister a persistently registered descriptor. 126   /** Deregister a persistently registered descriptor.
129   127  
130   @param fd The file descriptor to deregister. 128   @param fd The file descriptor to deregister.
131   */ 129   */
132   void deregister_descriptor(int fd) const; 130   void deregister_descriptor(int fd) const;
133   131  
134   /// 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 135   286 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 133   285 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
136   { 134   {
HITCBC 137   286 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 135   285 return register_descriptor(read_fd, signal_pipe_reader_.arm());
138   } 136   }
139   137  
140   private: 138   private:
141   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;
142   void interrupt_reactor() const override; 140   void interrupt_reactor() const override;
143 - static int timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept;  
144 - void on_timer_service_attached(timer_service& svc) override;  
145   void update_timerfd() const; 141   void update_timerfd() const;
146   142  
147   int epoll_fd_; 143   int epoll_fd_;
148   int event_fd_; 144   int event_fd_;
149   int timer_fd_; 145   int timer_fd_;
150   146  
151   // 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
152   // register_signal_reader on the first signal registration). 148   // register_signal_reader on the first signal registration).
153   reactor_signal_pipe_reader signal_pipe_reader_; 149   reactor_signal_pipe_reader signal_pipe_reader_;
154   150  
155   // Edge-triggered eventfd state 151   // Edge-triggered eventfd state
156   mutable std::atomic<bool> eventfd_armed_{false}; 152   mutable std::atomic<bool> eventfd_armed_{false};
157   153  
158   // Set when the earliest timer changes; flushed before epoll_wait 154   // Set when the earliest timer changes; flushed before epoll_wait
159   mutable std::atomic<bool> timerfd_stale_{false}; 155   mutable std::atomic<bool> timerfd_stale_{false};
160   156  
161   // Event buffer sized from max_events_per_poll_ (set at construction, 157   // Event buffer sized from max_events_per_poll_ (set at construction,
162   // resized by configure_reactor via io_context_options). 158   // resized by configure_reactor via io_context_options).
163   std::vector<epoll_event> event_buffer_; 159   std::vector<epoll_event> event_buffer_;
164   }; 160   };
165   161  
HITCBC 166 - 2484 inline epoll_scheduler::epoll_scheduler(capy::execution_context&, int) 162 + 2474 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
HITCBC 167   2484 : epoll_fd_(-1) 163   2474 : epoll_fd_(-1)
HITCBC 168   2484 , event_fd_(-1) 164   2474 , event_fd_(-1)
HITCBC 169   2484 , timer_fd_(-1) 165   2474 , timer_fd_(-1)
HITCBC 170   4968 , event_buffer_(max_events_per_poll_) 166   4948 , event_buffer_(max_events_per_poll_)
171   { 167   {
HITCBC 172   2484 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC); 168   2474 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
HITCBC 173   2484 if (epoll_fd_ < 0) 169   2474 if (epoll_fd_ < 0)
HITCBC 174   3 detail::throw_system_error(make_err(errno), "epoll_create1"); 170   1 detail::throw_system_error(make_err(errno), "epoll_create1");
175   171  
HITCBC 176   2481 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); 172   2473 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
HITCBC 177   2481 if (event_fd_ < 0) 173   2473 if (event_fd_ < 0)
178   { 174   {
HITCBC 179   3 int errn = errno; 175   1 int errn = errno;
HITCBC 180   3 ::close(epoll_fd_); 176   1 ::close(epoll_fd_);
HITCBC 181   3 detail::throw_system_error(make_err(errn), "eventfd"); 177   1 detail::throw_system_error(make_err(errn), "eventfd");
182   } 178   }
183   179  
HITGNC   180 + 2472 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
HITGNC   181 + 2472 if (timer_fd_ < 0)
  182 + {
HITGNC   183 + 1 int errn = errno;
HITGNC   184 + 1 ::close(event_fd_);
HITGNC   185 + 1 ::close(epoll_fd_);
HITGNC   186 + 1 detail::throw_system_error(make_err(errn), "timerfd_create");
  187 + }
  188 +
HITCBC 184   2478 epoll_event ev{}; 189   2471 epoll_event ev{};
HITCBC 185   2478 ev.events = EPOLLIN | EPOLLET; 190   2471 ev.events = EPOLLIN | EPOLLET;
HITCBC 186   2478 ev.data.ptr = nullptr; 191   2471 ev.data.ptr = nullptr;
HITCBC 187   2478 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)
188   { 193   {
HITCBC 189   3 int errn = errno; 194   1 int errn = errno;
HITGNC   195 + 1 ::close(timer_fd_);
HITCBC 190   3 ::close(event_fd_); 196   1 ::close(event_fd_);
HITCBC 191   3 ::close(epoll_fd_); 197   1 ::close(epoll_fd_);
HITCBC 192   3 detail::throw_system_error(make_err(errn), "epoll_ctl"); 198   1 detail::throw_system_error(make_err(errn), "epoll_ctl");
193   } 199   }
194   200  
HITCBC 195 - 2475 completed_ops_.push(&task_op_); 201 + 2470 epoll_event timer_ev{};
HITCBC 196 - 2502 } 202 + 2470 timer_ev.events = EPOLLIN | EPOLLERR;
HITGIC 197 - 203 + 2470 timer_ev.data.ptr = &timer_fd_;
HITGIC 198 - inline void 204 + 2470 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
DCB 199 - 533 epoll_scheduler::on_timer_service_attached(timer_service& svc)  
200 - {  
201 - // Without a timerfd the epoll_wait timeout drives the timers: the  
202 - // first timer is often made where a throw can only terminate.  
DCB 203 - 533 int fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);  
DCB 204 - 533 if (fd >= 0)  
205   { 205   {
HITGIC 206 - // Set before registration, which orders it for every thread 206 + 1 int errn = errno;
HITGIC 207 - // the timerfd's events can wake. 207 + 1 ::close(timer_fd_);
HITCBC 208 - 532 timer_fd_ = fd; 208 + 1 ::close(event_fd_);
HITCBC 209 - 532 epoll_event timer_ev{}; 209 + 1 ::close(epoll_fd_);
HITCBC 210 - 532 timer_ev.events = EPOLLIN | EPOLLERR; 210 + 1 detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
DCB 211 - 532 timer_ev.data.ptr = &timer_fd_;  
DCB 212 - 532 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &timer_ev) < 0)  
213 - {  
DCB 214 - 1 ::close(fd);  
DCB 215 - 1 timer_fd_ = -1;  
216 - }  
217   } 211   }
ECB 218 - 533 svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) { 212 +
HITCBC 219 - 4881 auto* self = static_cast<epoll_scheduler*>(p); 213 + 2469 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 220 - 4881 self->timerfd_stale_.store(true, std::memory_order_release); 214 + 2469 timer_svc_->set_on_earliest_changed(
HITCBC 221 - 4881 self->interrupt_reactor(); 215 + 7372 timer_service::callback(this, [](void* p) {
HITCBC 222 - 4881 })); 216 + 4903 auto* self = static_cast<epoll_scheduler*>(p);
HITGNC   217 + 4903 self->timerfd_stale_.store(true, std::memory_order_release);
HITGNC   218 + 4903 self->interrupt_reactor();
HITGNC   219 + 4903 }));
  220 +
HITGNC   221 + 2469 completed_ops_.push(&task_op_);
HITCBC 223   533 } 222   2484 }
224   223  
HITCBC 225   4950 inline epoll_scheduler::~epoll_scheduler() 224   4938 inline epoll_scheduler::~epoll_scheduler()
226   { 225   {
HITCBC 227   2475 if (timer_fd_ >= 0) 226   2469 if (timer_fd_ >= 0)
HITCBC 228   531 ::close(timer_fd_); 227   2469 ::close(timer_fd_);
HITCBC 229   2475 if (event_fd_ >= 0) 228   2469 if (event_fd_ >= 0)
HITCBC 230   2475 ::close(event_fd_); 229   2469 ::close(event_fd_);
HITCBC 231   2475 if (epoll_fd_ >= 0) 230   2469 if (epoll_fd_ >= 0)
HITCBC 232   2475 ::close(epoll_fd_); 231   2469 ::close(epoll_fd_);
HITCBC 233   4950 } 232   4938 }
234   233  
235   inline void 234   inline void
HITCBC 236   2475 epoll_scheduler::shutdown() 235   2469 epoll_scheduler::shutdown()
237   { 236   {
HITCBC 238   2475 shutdown_drain(); 237   2469 shutdown_drain();
239   238  
HITCBC 240   2475 if (event_fd_ >= 0) 239   2469 if (event_fd_ >= 0)
HITCBC 241   2475 interrupt_reactor(); 240   2469 interrupt_reactor();
HITCBC 242   2475 } 241   2469 }
243   242  
244   inline void 243   inline void
HITCBC 245   32 epoll_scheduler::configure_reactor( 244   32 epoll_scheduler::configure_reactor(
246   unsigned max_events, 245   unsigned max_events,
247   unsigned budget_init, 246   unsigned budget_init,
248   unsigned budget_max, 247   unsigned budget_max,
249   unsigned unassisted) 248   unsigned unassisted)
250   { 249   {
HITCBC 251   32 reactor_scheduler::configure_reactor( 250   32 reactor_scheduler::configure_reactor(
252   max_events, budget_init, budget_max, unassisted); 251   max_events, budget_init, budget_max, unassisted);
HITCBC 253   30 event_buffer_.resize(max_events_per_poll_); 252   30 event_buffer_.resize(max_events_per_poll_);
HITCBC 254   30 } 253   30 }
255   254  
256   inline std::error_code 255   inline std::error_code
HITCBC 257   13439 epoll_scheduler::register_descriptor( 256   12936 epoll_scheduler::register_descriptor(
258   int fd, reactor_descriptor_state* desc) const 257   int fd, reactor_descriptor_state* desc) const
259   { 258   {
HITCBC 260   13439 epoll_event ev{}; 259   12936 epoll_event ev{};
HITCBC 261   13439 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP; 260   12936 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
HITCBC 262   13439 ev.data.ptr = desc; 261   12936 ev.data.ptr = desc;
263   262  
264   // Initialized before the add: another run thread's reactor can 263   // Initialized before the add: another run thread's reactor can
265   // 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.
266   // A completion queued by an earlier registration of the same state 265   // A completion queued by an earlier registration of the same state
267   // 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
268   // mutex, and the mutex mode is only written when it differs. 267   // mutex, and the mutex mode is only written when it differs.
HITCBC 269   13439 if (desc->mutex.enabled() != reactor_io_locking_) 268   12936 if (desc->mutex.enabled() != reactor_io_locking_)
HITCBC 270   3 desc->mutex.set_enabled(reactor_io_locking_); 269   3 desc->mutex.set_enabled(reactor_io_locking_);
271   { 270   {
HITCBC 272   13439 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 271   12936 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 273   13439 desc->registered_events = ev.events; 272   12936 desc->registered_events = ev.events;
HITCBC 274   13439 desc->unpollable = false; 273   12936 desc->unpollable = false;
HITCBC 275   13439 desc->fd = fd; 274   12936 desc->fd = fd;
HITCBC 276   13439 desc->scheduler_ = this; 275   12936 desc->scheduler_ = this;
HITCBC 277   13439 desc->ready_events_.store(0, std::memory_order_release); 276   12936 desc->ready_events_.store(0, std::memory_order_release);
278   // object_ref_ is not touched: a queued invocation from an 277   // object_ref_ is not touched: a queued invocation from an
279   // earlier registration still owns the reference in it. 278   // earlier registration still owns the reference in it.
HITCBC 280   13439 desc->read_ready = false; 279   12936 desc->read_ready = false;
HITCBC 281   13439 desc->write_ready = false; 280   12936 desc->write_ready = false;
HITCBC 282   13439 } 281   12936 }
283   282  
HITCBC 284   13439 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) 283   12936 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
285   { 284   {
HITCBC 286   12 int const err = errno; 285   12 int const err = errno;
HITCBC 287   12 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 286   12 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 288   12 desc->registered_events = 0; 287   12 desc->registered_events = 0;
289   // 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
290   // block, so adopt it unwatched, as asio does. 289   // block, so adopt it unwatched, as asio does.
HITCBC 291   12 if (err != EPERM) 290   12 if (err != EPERM)
HITCBC 292   7 return make_err(err); 291   7 return make_err(err);
HITCBC 293   5 desc->unpollable = true; 292   5 desc->unpollable = true;
HITCBC 294   12 } 293   12 }
HITCBC 295   13432 return {}; 294   12929 return {};
296   } 295   }
297   296  
298   inline void 297   inline void
HITCBC 299   13142 epoll_scheduler::deregister_descriptor(int fd) const 298   12640 epoll_scheduler::deregister_descriptor(int fd) const
300   { 299   {
HITCBC 301   13142 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr); 300   12640 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
HITCBC 302   13142 } 301   12640 }
303   302  
304   inline void 303   inline void
HITCBC 305   13313 epoll_scheduler::interrupt_reactor() const 304   12937 epoll_scheduler::interrupt_reactor() const
306   { 305   {
HITCBC 307   13313 bool expected = false; 306   12937 bool expected = false;
HITCBC 308   13313 if (eventfd_armed_.compare_exchange_strong( 307   12937 if (eventfd_armed_.compare_exchange_strong(
309   expected, true, std::memory_order_release, 308   expected, true, std::memory_order_release,
310   std::memory_order_relaxed)) 309   std::memory_order_relaxed))
311   { 310   {
HITCBC 312   10811 std::uint64_t val = 1; 311   10512 std::uint64_t val = 1;
HITCBC 313   10811 if (::write(event_fd_, &val, sizeof(val)) < 0) 312   10512 if (::write(event_fd_, &val, sizeof(val)) < 0)
314   { 313   {
315   // The flag is what coalesces later interrupts into a byte 314   // The flag is what coalesces later interrupts into a byte
316   // already in the eventfd; a write that failed put no byte 315   // already in the eventfd; a write that failed put no byte
317   // there, so leaving it armed would swallow every interrupt 316   // there, so leaving it armed would swallow every interrupt
318   // that follows. Disarming keeps the cost to the interrupts 317   // that follows. Disarming keeps the cost to the interrupts
319   // already in flight -- the next one arms and writes again, 318   // already in flight -- the next one arms and writes again,
320   // instead of every one after this coalescing into a byte 319   // instead of every one after this coalescing into a byte
321   // that does not exist. 320   // that does not exist.
HITCBC 322   2 eventfd_armed_.store(false, std::memory_order_release); 321   2 eventfd_armed_.store(false, std::memory_order_release);
323   } 322   }
324   } 323   }
HITCBC 325   13313 } 324   12937 }
326   325  
327   inline void 326   inline void
HITCBC 328   11244 epoll_scheduler::update_timerfd() const 327   12130 epoll_scheduler::update_timerfd() const
329   { 328   {
HITCBC 330 - 11244 auto* tsvc = timers(); 329 + 12130 auto nearest = timer_svc_->nearest_expiry();
DCB 331 - 11244 if (!tsvc || timer_fd_ < 0)  
DCB 332 - 6 return;  
DCB 333 - 11238 auto nearest = tsvc->nearest_expiry();  
334   330  
HITCBC 335   11238 itimerspec ts{}; 331   12130 itimerspec ts{};
HITCBC 336   11238 int flags = 0; 332   12130 int flags = 0;
337   333  
HITCBC 338   11238 if (nearest == timer_service::time_point::max()) 334   12130 if (nearest == timer_service::time_point::max())
339   { 335   {
340   // No timers — disarm by setting to 0 (relative) 336   // No timers — disarm by setting to 0 (relative)
341   } 337   }
342   else 338   else
343   { 339   {
HITCBC 344   9889 auto now = std::chrono::steady_clock::now(); 340   10824 auto now = std::chrono::steady_clock::now();
HITCBC 345   9889 if (nearest <= now) 341   10824 if (nearest <= now)
346   { 342   {
347   // Use 1ns instead of 0 — zero disarms the timerfd 343   // Use 1ns instead of 0 — zero disarms the timerfd
HITCBC 348   774 ts.it_value.tv_nsec = 1; 344   1111 ts.it_value.tv_nsec = 1;
349   } 345   }
350   else 346   else
351   { 347   {
HITCBC 352   9115 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>( 348   9713 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
HITCBC 353   9115 nearest - now) 349   9713 nearest - now)
HITCBC 354   9115 .count(); 350   9713 .count();
HITCBC 355   9115 ts.it_value.tv_sec = nsec / 1000000000; 351   9713 ts.it_value.tv_sec = nsec / 1000000000;
HITCBC 356   9115 ts.it_value.tv_nsec = nsec % 1000000000; 352   9713 ts.it_value.tv_nsec = nsec % 1000000000;
HITCBC 357   9115 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0) 353   9713 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
MISUBC 358   ✗ ts.it_value.tv_nsec = 1; 354   ✗ ts.it_value.tv_nsec = 1;
359   } 355   }
360   } 356   }
361   357  
HITCBC 362   11238 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0) 358   12130 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
HITCBC 363   1 detail::throw_system_error(make_err(errno), "timerfd_settime"); 359   1 detail::throw_system_error(make_err(errno), "timerfd_settime");
HITGIC 364   } 360   12129 }
365 - // Rounded up, so the wait never wakes before the timer is due.  
366 - inline int  
367 - epoll_scheduler::timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept  
DCB 368 - 4 {  
369 - auto nearest = svc.nearest_expiry();  
DCB 370 - 4 if (nearest == timer_service::time_point::max())  
DCB 371 - 4 return timeout_ms;  
DUB 372 - ✗ auto now = std::chrono::steady_clock::now();  
DCB 373 - 4 if (nearest <= now)  
DCB 374 - 4 return 0;  
DUB 375 - ✗ auto ms =  
376 - std::chrono::ceil<std::chrono::milliseconds>(nearest - now).count();  
DCB 377 - 4 ms = (std::min)(ms,  
DCB 378 - 8 static_cast<decltype(ms)>(  
DCB 379 - 4 (std::numeric_limits<int>::max)()));  
DCB 380 - 4 if (timeout_ms >= 0 && timeout_ms < ms)  
DCB 381 - 4 return timeout_ms;  
DUB 382 - ✗ return static_cast<int>(ms);  
DCB 383 - 4 }  
384 -  
385   361  
386   inline void 362   inline void
HITCBC 387   145871 epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 363   154777 epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
388   { 364   {
389   int timeout_ms; 365   int timeout_ms;
HITCBC 390   145871 if (task_interrupted_) 366   154777 if (task_interrupted_)
HITCBC 391   126183 timeout_ms = 0; 367   136115 timeout_ms = 0;
HITCBC 392   19688 else if (timeout_us < 0) 368   18662 else if (timeout_us < 0)
HITCBC 393   19233 timeout_ms = -1; 369   18146 timeout_ms = -1;
394   else 370   else
HITCBC 395   455 timeout_ms = static_cast<int>((timeout_us + 999) / 1000); 371   516 timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
396   372  
HITCBC 397   145871 if (lock.owns_lock()) 373   154777 if (lock.owns_lock())
HITCBC 398   19692 lock.unlock(); 374   18666 lock.unlock();
399   375  
HITCBC 400   145871 task_cleanup on_exit{this, &lock, ctx}; 376   154777 task_cleanup on_exit{this, &lock, ctx};
401   377  
402   // Flush deferred timerfd programming before blocking 378   // Flush deferred timerfd programming before blocking
HITCBC 403   145871 if (timerfd_stale_.exchange(false, std::memory_order_acquire)) 379   154777 if (timerfd_stale_.exchange(false, std::memory_order_acquire))
HITCBC 404   3891 update_timerfd(); 380   3996 update_timerfd();
405 - // timer_fd_ is read only once the service is seen, after which it  
406 - // no longer changes.  
407 - auto* tsvc = timers();  
DCB 408 - 145870 bool const poll_timers = tsvc && timer_fd_ < 0;  
DCB 409 - 145870 if (poll_timers && timeout_ms != 0)  
DCB 410 - 145870 timeout_ms = timer_timeout_ms(*tsvc, timeout_ms);  
DCB 411 - 4  
412   381  
HITCBC 413   145870 int nfds = ::epoll_wait( 382   154776 int nfds = ::epoll_wait(
HITCBC 414   145870 epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()), 383   154776 epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
415   timeout_ms); 384   timeout_ms);
416   385  
HITCBC 417   145870 if (nfds < 0 && errno != EINTR) 386   154776 if (nfds < 0 && errno != EINTR)
HITCBC 418   1 detail::throw_system_error(make_err(errno), "epoll_wait"); 387   1 detail::throw_system_error(make_err(errno), "epoll_wait");
419   388  
HITCBC 420 - 145869 bool check_timers = poll_timers; 389 + 154775 bool check_timers = false;
HITCBC 421   145869 ready_queue local_ops; 390   154775 ready_queue local_ops;
422   391  
HITCBC 423   204152 for (int i = 0; i < nfds; ++i) 392   218413 for (int i = 0; i < nfds; ++i)
424   { 393   {
HITCBC 425   58283 if (event_buffer_[i].data.ptr == nullptr) 394   63638 if (event_buffer_[i].data.ptr == nullptr)
426   { 395   {
427   std::uint64_t val; 396   std::uint64_t val;
428   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 397   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
HITCBC 429   8334 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val)); 398   8041 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
HITCBC 430   8334 eventfd_armed_.store(false, std::memory_order_relaxed); 399   8041 eventfd_armed_.store(false, std::memory_order_relaxed);
HITCBC 431   8334 continue; 400   8041 continue;
HITCBC 432   8334 } 401   8041 }
433   402  
HITCBC 434   49949 if (event_buffer_[i].data.ptr == &timer_fd_) 403   55597 if (event_buffer_[i].data.ptr == &timer_fd_)
435   { 404   {
436   std::uint64_t expirations; 405   std::uint64_t expirations;
437   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 406   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
438   [[maybe_unused]] auto r = 407   [[maybe_unused]] auto r =
HITCBC 439   7349 ::read(timer_fd_, &expirations, sizeof(expirations)); 408   8134 ::read(timer_fd_, &expirations, sizeof(expirations));
HITCBC 440   7349 check_timers = true; 409   8134 check_timers = true;
HITCBC 441   7349 continue; 410   8134 continue;
HITCBC 442   7349 } 411   8134 }
443   412  
444   auto* desc = 413   auto* desc =
HITCBC 445   42600 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr); 414   47463 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
446   415  
447   // 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
448   // below did not translate would take no branch in 417   // below did not translate would take no branch in
449   // invoke_deferred_io() and, the registration being 418   // invoke_deferred_io() and, the registration being
450   // edge-triggered, never get another chance. 419   // edge-triggered, never get another chance.
451   // 420   //
452   // 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
453   // closed, with or without data still unread (HUP alone, or 422   // closed, with or without data still unread (HUP alone, or
454   // IN|HUP). Treat it as readable, writable and faulted, the way 423   // IN|HUP). Treat it as readable, writable and faulted, the way
455   // poll() and io_uring report it, so a parked error wait 424   // poll() and io_uring report it, so a parked error wait
456   // completes too. Sockets carry at least OUT with HUP (a fresh 425   // completes too. Sockets carry at least OUT with HUP (a fresh
457   // unconnected stream socket reports HUP|OUT), so they take the 426   // unconnected stream socket reports HUP|OUT), so they take the
458   // plain widening, where forcing IN costs at most one spurious 427   // plain widening, where forcing IN costs at most one spurious
459   // EAGAIN. 428   // EAGAIN.
HITCBC 460   42600 std::uint32_t ev = event_buffer_[i].events; 429   47463 std::uint32_t ev = event_buffer_[i].events;
HITCBC 461   42600 if ((ev & (EPOLLHUP | EPOLLOUT)) == EPOLLHUP) 430   47463 if ((ev & (EPOLLHUP | EPOLLOUT)) == EPOLLHUP)
HITCBC 462   8 ev |= EPOLLIN | EPOLLOUT | EPOLLERR; 431   8 ev |= EPOLLIN | EPOLLOUT | EPOLLERR;
HITCBC 463   42592 else if (ev & EPOLLHUP) 432   47455 else if (ev & EPOLLHUP)
HITCBC 464   3130 ev |= EPOLLIN | EPOLLOUT; 433   2520 ev |= EPOLLIN | EPOLLOUT;
HITCBC 465   42600 desc->add_ready_events(ev); 434   47463 desc->add_ready_events(ev);
466   435  
HITCBC 467   42600 bool expected = false; 436   47463 bool expected = false;
HITCBC 468   42600 if (desc->is_enqueued_.compare_exchange_strong( 437   47463 if (desc->is_enqueued_.compare_exchange_strong(
469   expected, true, std::memory_order_release, 438   expected, true, std::memory_order_release,
470   std::memory_order_relaxed)) 439   std::memory_order_relaxed))
471   { 440   {
HITCBC 472   42597 local_ops.push(desc); 441   47458 local_ops.push(desc);
473   } 442   }
474   } 443   }
475   444  
HITCBC 476   145869 if (check_timers) 445   154775 if (check_timers)
477   { 446   {
HITCBC 478 - 7353 if (auto* svc = timers()) 447 + 8134 timer_svc_->process_expired();
DCB 479 - 7353 svc->process_expired();  
HITCBC 480   7353 update_timerfd(); 448   8134 update_timerfd();
481   } 449   }
482   450  
HITCBC 483   145869 lock.lock(); 451   154775 lock.lock();
484   452  
HITCBC 485   145869 completed_ops_.splice(local_ops); 453   154775 completed_ops_.splice(local_ops);
HITCBC 486   145871 } 454   154777 }
487   455  
488   } // namespace boost::corosio::detail 456   } // namespace boost::corosio::detail
489   457  
490   #endif // BOOST_COROSIO_HAS_EPOLL 458   #endif // BOOST_COROSIO_HAS_EPOLL
491   459  
492   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 460   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP