. /include/boost/corosio /native /detail /epoll /epoll_scheduler.hpp
85.53% Lines (130/152) 100.00% Functions (11/11)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2026 Steve Gerbino 2   // Copyright (c) 2026 Steve Gerbino
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
4   // 4   //
5   // Distributed under the Boost Software License, Version 1.0. (See accompanying 5   // Distributed under the Boost Software License, Version 1.0. (See accompanying
6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) 6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7   // 7   //
8   // Official repository: https://github.com/cppalliance/corosio 8   // Official repository: https://github.com/cppalliance/corosio
9   // 9   //
10   10  
11   #ifndef BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 11   #ifndef BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
12   #define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 12   #define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
13   13  
14   #include <boost/corosio/detail/platform.hpp> 14   #include <boost/corosio/detail/platform.hpp>
15   15  
16   #if BOOST_COROSIO_HAS_EPOLL 16   #if BOOST_COROSIO_HAS_EPOLL
17   17  
18   #include <boost/corosio/detail/config.hpp> 18   #include <boost/corosio/detail/config.hpp>
19   #include <boost/capy/ex/execution_context.hpp> 19   #include <boost/capy/ex/execution_context.hpp>
20   20  
21   #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp> 21   #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
22   #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp> 22   #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp>
23   23  
24   #include <boost/corosio/native/detail/epoll/epoll_traits.hpp> 24   #include <boost/corosio/native/detail/epoll/epoll_traits.hpp>
25   #include <boost/corosio/detail/timer_service.hpp> 25   #include <boost/corosio/detail/timer_service.hpp>
26   #include <boost/corosio/native/detail/make_err.hpp> 26   #include <boost/corosio/native/detail/make_err.hpp>
27   #include <boost/corosio/native/detail/posix/posix_resolver_service.hpp> 27   #include <boost/corosio/native/detail/posix/posix_resolver_service.hpp>
28   #include <boost/corosio/native/detail/posix/posix_signal_service.hpp> 28   #include <boost/corosio/native/detail/posix/posix_signal_service.hpp>
29   #include <boost/corosio/native/detail/posix/posix_stream_file_service.hpp> 29   #include <boost/corosio/native/detail/posix/posix_stream_file_service.hpp>
30   #include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp> 30   #include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp>
31   31  
32   #include <boost/corosio/detail/except.hpp> 32   #include <boost/corosio/detail/except.hpp>
33   33  
34   #include <atomic> 34   #include <atomic>
35   #include <chrono> 35   #include <chrono>
36   #include <cstdint> 36   #include <cstdint>
37   #include <mutex> 37   #include <mutex>
38   #include <vector> 38   #include <vector>
39   39  
40   #include <errno.h> 40   #include <errno.h>
41   #include <sys/epoll.h> 41   #include <sys/epoll.h>
42   #include <sys/eventfd.h> 42   #include <sys/eventfd.h>
43   #include <sys/timerfd.h> 43   #include <sys/timerfd.h>
44   #include <unistd.h> 44   #include <unistd.h>
45   45  
46   namespace boost::corosio::detail { 46   namespace boost::corosio::detail {
47   47  
48   /** Linux scheduler using epoll for I/O multiplexing. 48   /** Linux scheduler using epoll for I/O multiplexing.
49   49  
50   This scheduler implements the scheduler interface using Linux epoll 50   This scheduler implements the scheduler interface using Linux epoll
51   for efficient I/O event notification. It uses a single reactor model 51   for efficient I/O event notification. It uses a single reactor model
52   where one thread runs epoll_wait while other threads 52   where one thread runs epoll_wait while other threads
53   wait on a condition variable for handler work. This design provides: 53   wait on a condition variable for handler work. This design provides:
54   54  
55   - Handler parallelism: N posted handlers can execute on N threads 55   - Handler parallelism: N posted handlers can execute on N threads
56   - No thundering herd: condition_variable wakes exactly one thread 56   - No thundering herd: condition_variable wakes exactly one thread
57   - IOCP parity: Behavior matches Windows I/O completion port semantics 57   - IOCP parity: Behavior matches Windows I/O completion port semantics
58   58  
59   When threads call run(), they first try to execute queued handlers. 59   When threads call run(), they first try to execute queued handlers.
60   If the queue is empty and no reactor is running, one thread becomes 60   If the queue is empty and no reactor is running, one thread becomes
61   the reactor and runs epoll_wait. Other threads wait on a condition 61   the reactor and runs epoll_wait. Other threads wait on a condition
62   variable until handlers are available. 62   variable until handlers are available.
63   63  
64   @par Thread Safety 64   @par Thread Safety
65   All public member functions are thread-safe. 65   All public member functions are thread-safe.
66   */ 66   */
67   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler 67   class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
68   { 68   {
69   public: 69   public:
70   /** Construct the scheduler. 70   /** Construct the scheduler.
71   71  
72   Creates an epoll instance, eventfd for reactor interruption, 72   Creates an epoll instance, eventfd for reactor interruption,
73   and timerfd for kernel-managed timer expiry. 73   and timerfd for kernel-managed timer expiry.
74   74  
75   @param ctx Reference to the owning execution_context. 75   @param ctx Reference to the owning execution_context.
76   @param concurrency_hint Hint for expected thread count (unused). 76   @param concurrency_hint Hint for expected thread count (unused).
77   */ 77   */
78   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 78   epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
79   79  
80   /// Destroy the scheduler. 80   /// Destroy the scheduler.
81   ~epoll_scheduler() override; 81   ~epoll_scheduler() override;
82   82  
83   epoll_scheduler(epoll_scheduler const&) = delete; 83   epoll_scheduler(epoll_scheduler const&) = delete;
84   epoll_scheduler& operator=(epoll_scheduler const&) = delete; 84   epoll_scheduler& operator=(epoll_scheduler const&) = delete;
85   85  
86   /// Shut down the scheduler, draining pending operations. 86   /// Shut down the scheduler, draining pending operations.
87   void shutdown() override; 87   void shutdown() override;
88   88  
89   /// Apply runtime configuration, resizing the event buffer. 89   /// Apply runtime configuration, resizing the event buffer.
90   void configure_reactor( 90   void configure_reactor(
91   unsigned max_events, 91   unsigned max_events,
92   unsigned budget_init, 92   unsigned budget_init,
93   unsigned budget_max, 93   unsigned budget_max,
94   unsigned unassisted) override; 94   unsigned unassisted) override;
95   95  
96   /** Return the epoll file descriptor. 96   /** Return the epoll file descriptor.
97   97  
98   Used by socket services to register file descriptors 98   Used by socket services to register file descriptors
99   for I/O event notification. 99   for I/O event notification.
100   100  
101   @return The epoll file descriptor. 101   @return The epoll file descriptor.
102   */ 102   */
103   int epoll_fd() const noexcept 103   int epoll_fd() const noexcept
104   { 104   {
105   return epoll_fd_; 105   return epoll_fd_;
106   } 106   }
107   107  
108   /** Register a descriptor for persistent monitoring. 108   /** Register a descriptor for persistent monitoring.
109   109  
110   The fd is registered once and stays registered until explicitly 110   The fd is registered once and stays registered until explicitly
111   deregistered. Events are dispatched via reactor_descriptor_state which 111   deregistered. Events are dispatched via reactor_descriptor_state which
112   tracks pending read/write/connect operations. 112   tracks pending read/write/connect operations.
113   113  
114   @param fd The file descriptor to register. 114   @param fd The file descriptor to register.
115   @param desc Pointer to descriptor data (stored in epoll_event.data.ptr). 115   @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
116   */ 116   */
117   void register_descriptor(int fd, reactor_descriptor_state* desc) const; 117   void register_descriptor(int fd, reactor_descriptor_state* desc) const;
118   118  
119   /** Deregister a persistently registered descriptor. 119   /** Deregister a persistently registered descriptor.
120   120  
121   @param fd The file descriptor to deregister. 121   @param fd The file descriptor to deregister.
122   */ 122   */
123   void deregister_descriptor(int fd) const; 123   void deregister_descriptor(int fd) const;
124   124  
125   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp). 125   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
HITCBC 126   41 void register_signal_reader(int read_fd) override 126   51 void register_signal_reader(int read_fd) override
127   { 127   {
HITCBC 128   41 register_descriptor(read_fd, signal_pipe_reader_.arm()); 128   51 register_descriptor(read_fd, signal_pipe_reader_.arm());
HITCBC 129   41 } 129   51 }
130   130  
131   private: 131   private:
132   void 132   void
133   run_task(lock_type& lock, context_type* ctx, 133   run_task(lock_type& lock, context_type* ctx,
134   long timeout_us) override; 134   long timeout_us) override;
135   void interrupt_reactor() const override; 135   void interrupt_reactor() const override;
136   void update_timerfd() const; 136   void update_timerfd() const;
137   137  
138   int epoll_fd_; 138   int epoll_fd_;
139   int event_fd_; 139   int event_fd_;
140   int timer_fd_; 140   int timer_fd_;
141   141  
142   // Watches the global signal self-pipe's read end (armed lazily by 142   // Watches the global signal self-pipe's read end (armed lazily by
143   // register_signal_reader on the first signal registration). 143   // register_signal_reader on the first signal registration).
144   reactor_signal_pipe_reader signal_pipe_reader_; 144   reactor_signal_pipe_reader signal_pipe_reader_;
145   145  
146   // Edge-triggered eventfd state 146   // Edge-triggered eventfd state
147   mutable std::atomic<bool> eventfd_armed_{false}; 147   mutable std::atomic<bool> eventfd_armed_{false};
148   148  
149   // Set when the earliest timer changes; flushed before epoll_wait 149   // Set when the earliest timer changes; flushed before epoll_wait
150   mutable std::atomic<bool> timerfd_stale_{false}; 150   mutable std::atomic<bool> timerfd_stale_{false};
151   151  
152   // Event buffer sized from max_events_per_poll_ (set at construction, 152   // Event buffer sized from max_events_per_poll_ (set at construction,
153   // resized by configure_reactor via io_context_options). 153   // resized by configure_reactor via io_context_options).
154   std::vector<epoll_event> event_buffer_; 154   std::vector<epoll_event> event_buffer_;
155   }; 155   };
156   156  
HITCBC 157   642 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) 157   785 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
HITCBC 158   642 : epoll_fd_(-1) 158   785 : epoll_fd_(-1)
HITCBC 159   642 , event_fd_(-1) 159   785 , event_fd_(-1)
HITCBC 160   642 , timer_fd_(-1) 160   785 , timer_fd_(-1)
HITCBC 161   1284 , event_buffer_(max_events_per_poll_) 161   1570 , event_buffer_(max_events_per_poll_)
162   { 162   {
HITCBC 163   642 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC); 163   785 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
HITCBC 164   642 if (epoll_fd_ < 0) 164   785 if (epoll_fd_ < 0)
MISUBC 165   ✗ detail::throw_system_error(make_err(errno), "epoll_create1"); 165   ✗ detail::throw_system_error(make_err(errno), "epoll_create1");
166   166  
HITCBC 167   642 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); 167   785 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
HITCBC 168   642 if (event_fd_ < 0) 168   785 if (event_fd_ < 0)
169   { 169   {
MISUBC 170   ✗ int errn = errno; 170   ✗ int errn = errno;
MISUBC 171   ✗ ::close(epoll_fd_); 171   ✗ ::close(epoll_fd_);
MISUBC 172   ✗ detail::throw_system_error(make_err(errn), "eventfd"); 172   ✗ detail::throw_system_error(make_err(errn), "eventfd");
173   } 173   }
174   174  
HITCBC 175   642 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); 175   785 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
HITCBC 176   642 if (timer_fd_ < 0) 176   785 if (timer_fd_ < 0)
177   { 177   {
MISUBC 178   ✗ int errn = errno; 178   ✗ int errn = errno;
MISUBC 179   ✗ ::close(event_fd_); 179   ✗ ::close(event_fd_);
MISUBC 180   ✗ ::close(epoll_fd_); 180   ✗ ::close(epoll_fd_);
MISUBC 181   ✗ detail::throw_system_error(make_err(errn), "timerfd_create"); 181   ✗ detail::throw_system_error(make_err(errn), "timerfd_create");
182   } 182   }
183   183  
HITCBC 184   642 epoll_event ev{}; 184   785 epoll_event ev{};
HITCBC 185   642 ev.events = EPOLLIN | EPOLLET; 185   785 ev.events = EPOLLIN | EPOLLET;
HITCBC 186   642 ev.data.ptr = nullptr; 186   785 ev.data.ptr = nullptr;
HITCBC 187   642 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0) 187   785 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
188   { 188   {
MISUBC 189   ✗ int errn = errno; 189   ✗ int errn = errno;
MISUBC 190   ✗ ::close(timer_fd_); 190   ✗ ::close(timer_fd_);
MISUBC 191   ✗ ::close(event_fd_); 191   ✗ ::close(event_fd_);
MISUBC 192   ✗ ::close(epoll_fd_); 192   ✗ ::close(epoll_fd_);
MISUBC 193   ✗ detail::throw_system_error(make_err(errn), "epoll_ctl"); 193   ✗ detail::throw_system_error(make_err(errn), "epoll_ctl");
194   } 194   }
195   195  
HITCBC 196   642 epoll_event timer_ev{}; 196   785 epoll_event timer_ev{};
HITCBC 197   642 timer_ev.events = EPOLLIN | EPOLLERR; 197   785 timer_ev.events = EPOLLIN | EPOLLERR;
HITCBC 198   642 timer_ev.data.ptr = &timer_fd_; 198   785 timer_ev.data.ptr = &timer_fd_;
HITCBC 199   642 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0) 199   785 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
200   { 200   {
MISUBC 201   ✗ int errn = errno; 201   ✗ int errn = errno;
MISUBC 202   ✗ ::close(timer_fd_); 202   ✗ ::close(timer_fd_);
MISUBC 203   ✗ ::close(event_fd_); 203   ✗ ::close(event_fd_);
MISUBC 204   ✗ ::close(epoll_fd_); 204   ✗ ::close(epoll_fd_);
MISUBC 205   ✗ detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)"); 205   ✗ detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
206   } 206   }
207   207  
HITCBC 208   642 timer_svc_ = &get_timer_service(ctx, *this); 208   785 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 209   642 timer_svc_->set_on_earliest_changed( 209   785 timer_svc_->set_on_earliest_changed(
HITCBC 210   5817 timer_service::callback(this, [](void* p) { 210   3885 timer_service::callback(this, [](void* p) {
HITCBC 211   5175 auto* self = static_cast<epoll_scheduler*>(p); 211   3100 auto* self = static_cast<epoll_scheduler*>(p);
HITCBC 212   5175 self->timerfd_stale_.store(true, std::memory_order_release); 212   3100 self->timerfd_stale_.store(true, std::memory_order_release);
HITCBC 213   5175 self->interrupt_reactor(); 213   3100 self->interrupt_reactor();
HITCBC 214   5175 })); 214   3100 }));
215   215  
HITCBC 216   642 get_resolver_service(ctx, *this); 216   785 get_resolver_service(ctx, *this);
HITCBC 217   642 get_signal_service(ctx, *this); 217   785 get_signal_service(ctx, *this);
HITCBC 218   642 get_stream_file_service(ctx, *this); 218   785 get_stream_file_service(ctx, *this);
HITCBC 219   642 get_random_access_file_service(ctx, *this); 219   785 get_random_access_file_service(ctx, *this);
220   220  
HITCBC 221   642 completed_ops_.push(&task_op_); 221   785 completed_ops_.push(&task_op_);
HITCBC 222   642 } 222   785 }
223   223  
HITCBC 224   1284 inline epoll_scheduler::~epoll_scheduler() 224   1570 inline epoll_scheduler::~epoll_scheduler()
225   { 225   {
HITCBC 226   642 if (timer_fd_ >= 0) 226   785 if (timer_fd_ >= 0)
HITCBC 227   642 ::close(timer_fd_); 227   785 ::close(timer_fd_);
HITCBC 228   642 if (event_fd_ >= 0) 228   785 if (event_fd_ >= 0)
HITCBC 229   642 ::close(event_fd_); 229   785 ::close(event_fd_);
HITCBC 230   642 if (epoll_fd_ >= 0) 230   785 if (epoll_fd_ >= 0)
HITCBC 231   642 ::close(epoll_fd_); 231   785 ::close(epoll_fd_);
HITCBC 232   1284 } 232   1570 }
233   233  
234   inline void 234   inline void
HITCBC 235   642 epoll_scheduler::shutdown() 235   785 epoll_scheduler::shutdown()
236   { 236   {
HITCBC 237   642 shutdown_drain(); 237   785 shutdown_drain();
238   238  
HITCBC 239   642 if (event_fd_ >= 0) 239   785 if (event_fd_ >= 0)
HITCBC 240   642 interrupt_reactor(); 240   785 interrupt_reactor();
HITCBC 241   642 } 241   785 }
242   242  
243   inline void 243   inline void
HITCBC 244   16 epoll_scheduler::configure_reactor( 244   19 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   16 reactor_scheduler::configure_reactor( 250   19 reactor_scheduler::configure_reactor(
251   max_events, budget_init, budget_max, unassisted); 251   max_events, budget_init, budget_max, unassisted);
HITCBC 252   15 event_buffer_.resize(max_events_per_poll_); 252   18 event_buffer_.resize(max_events_per_poll_);
HITCBC 253   15 } 253   18 }
254   254  
255   inline void 255   inline void
HITCBC 256   9131 epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const 256   5185 epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const
257   { 257   {
HITCBC 258   9131 epoll_event ev{}; 258   5185 epoll_event ev{};
HITCBC 259   9131 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP; 259   5185 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
HITCBC 260   9131 ev.data.ptr = desc; 260   5185 ev.data.ptr = desc;
261   261  
HITCBC 262   9131 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) 262   5185 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
MISUBC 263   ✗ detail::throw_system_error(make_err(errno), "epoll_ctl (register)"); 263   ✗ detail::throw_system_error(make_err(errno), "epoll_ctl (register)");
264   264  
HITCBC 265   9131 desc->registered_events = ev.events; 265   5185 desc->registered_events = ev.events;
HITCBC 266   9131 desc->fd = fd; 266   5185 desc->fd = fd;
HITCBC 267   9131 desc->scheduler_ = this; 267   5185 desc->scheduler_ = this;
HITCBC 268   9131 desc->mutex.set_enabled(reactor_io_locking_); 268   5185 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 269   9131 desc->ready_events_.store(0, std::memory_order_relaxed); 269   5185 desc->ready_events_.store(0, std::memory_order_relaxed);
270   270  
HITCBC 271   9131 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 271   5185 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 272   9131 desc->impl_ref_.reset(); 272   5185 desc->impl_ref_.reset();
HITCBC 273   9131 desc->read_ready = false; 273   5185 desc->read_ready = false;
HITCBC 274   9131 desc->write_ready = false; 274   5185 desc->write_ready = false;
HITCBC 275   9131 } 275   5185 }
276   276  
277   inline void 277   inline void
HITCBC 278   9090 epoll_scheduler::deregister_descriptor(int fd) const 278   5134 epoll_scheduler::deregister_descriptor(int fd) const
279   { 279   {
HITCBC 280   9090 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr); 280   5134 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
HITCBC 281   9090 } 281   5134 }
282   282  
283   inline void 283   inline void
HITCBC 284   6337 epoll_scheduler::interrupt_reactor() const 284   4583 epoll_scheduler::interrupt_reactor() const
285   { 285   {
HITCBC 286   6337 bool expected = false; 286   4583 bool expected = false;
HITCBC 287   6337 if (eventfd_armed_.compare_exchange_strong( 287   4583 if (eventfd_armed_.compare_exchange_strong(
288   expected, true, std::memory_order_release, 288   expected, true, std::memory_order_release,
289   std::memory_order_relaxed)) 289   std::memory_order_relaxed))
290   { 290   {
HITCBC 291   5365 std::uint64_t val = 1; 291   3472 std::uint64_t val = 1;
HITCBC 292   5365 [[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val)); 292   3472 [[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val));
293   } 293   }
HITCBC 294   6337 } 294   4583 }
295   295  
296   inline void 296   inline void
HITCBC 297   9139 epoll_scheduler::update_timerfd() const 297   4939 epoll_scheduler::update_timerfd() const
298   { 298   {
HITCBC 299   9139 auto nearest = timer_svc_->nearest_expiry(); 299   4939 auto nearest = timer_svc_->nearest_expiry();
300   300  
HITCBC 301   9139 itimerspec ts{}; 301   4939 itimerspec ts{};
HITCBC 302   9139 int flags = 0; 302   4939 int flags = 0;
303   303  
HITCBC 304   9139 if (nearest == timer_service::time_point::max()) 304   4939 if (nearest == timer_service::time_point::max())
305   { 305   {
306   // No timers — disarm by setting to 0 (relative) 306   // No timers — disarm by setting to 0 (relative)
307   } 307   }
308   else 308   else
309   { 309   {
HITCBC 310   9001 auto now = std::chrono::steady_clock::now(); 310   4789 auto now = std::chrono::steady_clock::now();
HITCBC 311   9001 if (nearest <= now) 311   4789 if (nearest <= now)
312   { 312   {
313   // Use 1ns instead of 0 — zero disarms the timerfd 313   // Use 1ns instead of 0 — zero disarms the timerfd
HITCBC 314   1297 ts.it_value.tv_nsec = 1; 314   428 ts.it_value.tv_nsec = 1;
315   } 315   }
316   else 316   else
317   { 317   {
HITCBC 318   7704 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>( 318   4361 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
HITCBC 319   7704 nearest - now) 319   4361 nearest - now)
HITCBC 320   7704 .count(); 320   4361 .count();
HITCBC 321   7704 ts.it_value.tv_sec = nsec / 1000000000; 321   4361 ts.it_value.tv_sec = nsec / 1000000000;
HITCBC 322   7704 ts.it_value.tv_nsec = nsec % 1000000000; 322   4361 ts.it_value.tv_nsec = nsec % 1000000000;
HITCBC 323   7704 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0) 323   4361 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
MISUBC 324   ✗ ts.it_value.tv_nsec = 1; 324   ✗ ts.it_value.tv_nsec = 1;
325   } 325   }
326   } 326   }
327   327  
HITCBC 328   9139 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0) 328   4939 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
MISUBC 329   ✗ detail::throw_system_error(make_err(errno), "timerfd_settime"); 329   ✗ detail::throw_system_error(make_err(errno), "timerfd_settime");
HITCBC 330   9139 } 330   4939 }
331   331  
332   inline void 332   inline void
HITCBC 333   38874 epoll_scheduler::run_task( 333   30282 epoll_scheduler::run_task(
334   lock_type& lock, context_type* ctx, long timeout_us) 334   lock_type& lock, context_type* ctx, long timeout_us)
335   { 335   {
336   int timeout_ms; 336   int timeout_ms;
HITCBC 337   38874 if (task_interrupted_) 337   30282 if (task_interrupted_)
HITCBC 338   27475 timeout_ms = 0; 338   23891 timeout_ms = 0;
HITCBC 339   11399 else if (timeout_us < 0) 339   6391 else if (timeout_us < 0)
HITCBC 340   11395 timeout_ms = -1; 340   6388 timeout_ms = -1;
341   else 341   else
HITCBC 342   4 timeout_ms = static_cast<int>((timeout_us + 999) / 1000); 342   3 timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
343   343  
HITCBC 344   38874 if (lock.owns_lock()) 344   30282 if (lock.owns_lock())
HITCBC 345   11401 lock.unlock(); 345   6393 lock.unlock();
346   346  
HITCBC 347   38874 task_cleanup on_exit{this, &lock, ctx}; 347   30282 task_cleanup on_exit{this, &lock, ctx};
348   348  
349   // Flush deferred timerfd programming before blocking 349   // Flush deferred timerfd programming before blocking
HITCBC 350   38874 if (timerfd_stale_.exchange(false, std::memory_order_acquire)) 350   30282 if (timerfd_stale_.exchange(false, std::memory_order_acquire))
HITCBC 351   4575 update_timerfd(); 351   2476 update_timerfd();
352   352  
HITCBC 353   38874 int nfds = ::epoll_wait( 353   30282 int nfds = ::epoll_wait(
354   epoll_fd_, event_buffer_.data(), 354   epoll_fd_, event_buffer_.data(),
HITCBC 355   38874 static_cast<int>(event_buffer_.size()), timeout_ms); 355   30282 static_cast<int>(event_buffer_.size()), timeout_ms);
356   356  
HITCBC 357   38874 if (nfds < 0 && errno != EINTR) 357   30282 if (nfds < 0 && errno != EINTR)
MISUBC 358   ✗ detail::throw_system_error(make_err(errno), "epoll_wait"); 358   ✗ detail::throw_system_error(make_err(errno), "epoll_wait");
359   359  
HITCBC 360   38874 bool check_timers = false; 360   30282 bool check_timers = false;
HITCBC 361   38874 ready_queue local_ops; 361   30282 ready_queue local_ops;
362   362  
HITCBC 363   89536 for (int i = 0; i < nfds; ++i) 363   68321 for (int i = 0; i < nfds; ++i)
364   { 364   {
HITCBC 365   50662 if (event_buffer_[i].data.ptr == nullptr) 365   38039 if (event_buffer_[i].data.ptr == nullptr)
366   { 366   {
367   std::uint64_t val; 367   std::uint64_t val;
368   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 368   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
HITCBC 369   4723 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val)); 369   2687 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
HITCBC 370   4723 eventfd_armed_.store(false, std::memory_order_relaxed); 370   2687 eventfd_armed_.store(false, std::memory_order_relaxed);
HITCBC 371   4723 continue; 371   2687 continue;
HITCBC 372   4723 } 372   2687 }
373   373  
HITCBC 374   45939 if (event_buffer_[i].data.ptr == &timer_fd_) 374   35352 if (event_buffer_[i].data.ptr == &timer_fd_)
375   { 375   {
376   std::uint64_t expirations; 376   std::uint64_t expirations;
377   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 377   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
378   [[maybe_unused]] auto r = 378   [[maybe_unused]] auto r =
HITCBC 379   4564 ::read(timer_fd_, &expirations, sizeof(expirations)); 379   2463 ::read(timer_fd_, &expirations, sizeof(expirations));
HITCBC 380   4564 check_timers = true; 380   2463 check_timers = true;
HITCBC 381   4564 continue; 381   2463 continue;
HITCBC 382   4564 } 382   2463 }
383   383  
384   auto* desc = 384   auto* desc =
HITCBC 385   41375 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr); 385   32889 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
HITCBC 386   41375 desc->add_ready_events(event_buffer_[i].events); 386   32889 desc->add_ready_events(event_buffer_[i].events);
387   387  
HITCBC 388   41375 bool expected = false; 388   32889 bool expected = false;
HITCBC 389   41375 if (desc->is_enqueued_.compare_exchange_strong( 389   32889 if (desc->is_enqueued_.compare_exchange_strong(
390   expected, true, std::memory_order_release, 390   expected, true, std::memory_order_release,
391   std::memory_order_relaxed)) 391   std::memory_order_relaxed))
392   { 392   {
HITCBC 393   41375 local_ops.push(desc); 393   32889 local_ops.push(desc);
394   } 394   }
395   } 395   }
396   396  
HITCBC 397   38874 if (check_timers) 397   30282 if (check_timers)
398   { 398   {
HITCBC 399   4564 timer_svc_->process_expired(); 399   2463 timer_svc_->process_expired();
HITCBC 400   4564 update_timerfd(); 400   2463 update_timerfd();
401   } 401   }
402   402  
HITCBC 403   38874 lock.lock(); 403   30282 lock.lock();
404   404  
HITCBC 405   38874 completed_ops_.splice(local_ops); 405   30282 completed_ops_.splice(local_ops);
HITCBC 406   38874 } 406   30282 }
407   407  
408   } // namespace boost::corosio::detail 408   } // namespace boost::corosio::detail
409   409  
410   #endif // BOOST_COROSIO_HAS_EPOLL 410   #endif // BOOST_COROSIO_HAS_EPOLL
411   411  
412   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 412   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP