. /include/boost/corosio /native /detail /epoll /epoll_scheduler.hpp
85.71% Lines (132/154) 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 +
  117 + @return The error if registration fails, otherwise a default
  118 + constructed error code.
116   */ 119   */
117 - void register_descriptor(int fd, reactor_descriptor_state* desc) const; 120 + std::error_code
  121 + register_descriptor(int fd, reactor_descriptor_state* desc) const;
118   122  
119   /** Deregister a persistently registered descriptor. 123   /** Deregister a persistently registered descriptor.
120   124  
121   @param fd The file descriptor to deregister. 125   @param fd The file descriptor to deregister.
122   */ 126   */
123   void deregister_descriptor(int fd) const; 127   void deregister_descriptor(int fd) const;
124   128  
125   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp). 129   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
HITCBC 126   51 void register_signal_reader(int read_fd) override 130   51 void register_signal_reader(int read_fd) override
127   { 131   {
HITCBC 128 - 51 register_descriptor(read_fd, signal_pipe_reader_.arm()); 132 + 51 if (auto ec = register_descriptor(read_fd, signal_pipe_reader_.arm()))
MISUNC   133 + ✗ detail::throw_system_error(ec, "epoll_ctl (register)");
HITCBC 129   51 } 134   51 }
130   135  
131   private: 136   private:
132   void 137   void
133   run_task(lock_type& lock, context_type* ctx, 138   run_task(lock_type& lock, context_type* ctx,
134   long timeout_us) override; 139   long timeout_us) override;
135   void interrupt_reactor() const override; 140   void interrupt_reactor() const override;
136   void update_timerfd() const; 141   void update_timerfd() const;
137   142  
138   int epoll_fd_; 143   int epoll_fd_;
139   int event_fd_; 144   int event_fd_;
140   int timer_fd_; 145   int timer_fd_;
141   146  
142   // 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
143   // register_signal_reader on the first signal registration). 148   // register_signal_reader on the first signal registration).
144   reactor_signal_pipe_reader signal_pipe_reader_; 149   reactor_signal_pipe_reader signal_pipe_reader_;
145   150  
146   // Edge-triggered eventfd state 151   // Edge-triggered eventfd state
147   mutable std::atomic<bool> eventfd_armed_{false}; 152   mutable std::atomic<bool> eventfd_armed_{false};
148   153  
149   // Set when the earliest timer changes; flushed before epoll_wait 154   // Set when the earliest timer changes; flushed before epoll_wait
150   mutable std::atomic<bool> timerfd_stale_{false}; 155   mutable std::atomic<bool> timerfd_stale_{false};
151   156  
152   // Event buffer sized from max_events_per_poll_ (set at construction, 157   // Event buffer sized from max_events_per_poll_ (set at construction,
153   // resized by configure_reactor via io_context_options). 158   // resized by configure_reactor via io_context_options).
154   std::vector<epoll_event> event_buffer_; 159   std::vector<epoll_event> event_buffer_;
155   }; 160   };
156   161  
HITCBC 157   825 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int) 162   867 inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
HITCBC 158   825 : epoll_fd_(-1) 163   867 : epoll_fd_(-1)
HITCBC 159   825 , event_fd_(-1) 164   867 , event_fd_(-1)
HITCBC 160   825 , timer_fd_(-1) 165   867 , timer_fd_(-1)
HITCBC 161   1650 , event_buffer_(max_events_per_poll_) 166   1734 , event_buffer_(max_events_per_poll_)
162   { 167   {
HITCBC 163   825 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC); 168   867 epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
HITCBC 164   825 if (epoll_fd_ < 0) 169   867 if (epoll_fd_ < 0)
MISUBC 165   ✗ detail::throw_system_error(make_err(errno), "epoll_create1"); 170   ✗ detail::throw_system_error(make_err(errno), "epoll_create1");
166   171  
HITCBC 167   825 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); 172   867 event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
HITCBC 168   825 if (event_fd_ < 0) 173   867 if (event_fd_ < 0)
169   { 174   {
MISUBC 170   ✗ int errn = errno; 175   ✗ int errn = errno;
MISUBC 171   ✗ ::close(epoll_fd_); 176   ✗ ::close(epoll_fd_);
MISUBC 172   ✗ detail::throw_system_error(make_err(errn), "eventfd"); 177   ✗ detail::throw_system_error(make_err(errn), "eventfd");
173   } 178   }
174   179  
HITCBC 175   825 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); 180   867 timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
HITCBC 176   825 if (timer_fd_ < 0) 181   867 if (timer_fd_ < 0)
177   { 182   {
MISUBC 178   ✗ int errn = errno; 183   ✗ int errn = errno;
MISUBC 179   ✗ ::close(event_fd_); 184   ✗ ::close(event_fd_);
MISUBC 180   ✗ ::close(epoll_fd_); 185   ✗ ::close(epoll_fd_);
MISUBC 181   ✗ detail::throw_system_error(make_err(errn), "timerfd_create"); 186   ✗ detail::throw_system_error(make_err(errn), "timerfd_create");
182   } 187   }
183   188  
HITCBC 184   825 epoll_event ev{}; 189   867 epoll_event ev{};
HITCBC 185   825 ev.events = EPOLLIN | EPOLLET; 190   867 ev.events = EPOLLIN | EPOLLET;
HITCBC 186   825 ev.data.ptr = nullptr; 191   867 ev.data.ptr = nullptr;
HITCBC 187   825 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0) 192   867 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
188   { 193   {
MISUBC 189   ✗ int errn = errno; 194   ✗ int errn = errno;
MISUBC 190   ✗ ::close(timer_fd_); 195   ✗ ::close(timer_fd_);
MISUBC 191   ✗ ::close(event_fd_); 196   ✗ ::close(event_fd_);
MISUBC 192   ✗ ::close(epoll_fd_); 197   ✗ ::close(epoll_fd_);
MISUBC 193   ✗ detail::throw_system_error(make_err(errn), "epoll_ctl"); 198   ✗ detail::throw_system_error(make_err(errn), "epoll_ctl");
194   } 199   }
195   200  
HITCBC 196   825 epoll_event timer_ev{}; 201   867 epoll_event timer_ev{};
HITCBC 197   825 timer_ev.events = EPOLLIN | EPOLLERR; 202   867 timer_ev.events = EPOLLIN | EPOLLERR;
HITCBC 198   825 timer_ev.data.ptr = &timer_fd_; 203   867 timer_ev.data.ptr = &timer_fd_;
HITCBC 199   825 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0) 204   867 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
200   { 205   {
MISUBC 201   ✗ int errn = errno; 206   ✗ int errn = errno;
MISUBC 202   ✗ ::close(timer_fd_); 207   ✗ ::close(timer_fd_);
MISUBC 203   ✗ ::close(event_fd_); 208   ✗ ::close(event_fd_);
MISUBC 204   ✗ ::close(epoll_fd_); 209   ✗ ::close(epoll_fd_);
MISUBC 205   ✗ detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)"); 210   ✗ detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
206   } 211   }
207   212  
HITCBC 208   825 timer_svc_ = &get_timer_service(ctx, *this); 213   867 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 209   825 timer_svc_->set_on_earliest_changed( 214   867 timer_svc_->set_on_earliest_changed(
HITCBC 210   3936 timer_service::callback(this, [](void* p) { 215   3963 timer_service::callback(this, [](void* p) {
HITCBC 211   3111 auto* self = static_cast<epoll_scheduler*>(p); 216   3096 auto* self = static_cast<epoll_scheduler*>(p);
HITCBC 212   3111 self->timerfd_stale_.store(true, std::memory_order_release); 217   3096 self->timerfd_stale_.store(true, std::memory_order_release);
HITCBC 213   3111 self->interrupt_reactor(); 218   3096 self->interrupt_reactor();
HITCBC 214   3111 })); 219   3096 }));
215   220  
HITCBC 216   825 get_resolver_service(ctx, *this); 221   867 get_resolver_service(ctx, *this);
HITCBC 217   825 get_signal_service(ctx, *this); 222   867 get_signal_service(ctx, *this);
HITCBC 218   825 get_stream_file_service(ctx, *this); 223   867 get_stream_file_service(ctx, *this);
HITCBC 219   825 get_random_access_file_service(ctx, *this); 224   867 get_random_access_file_service(ctx, *this);
220   225  
HITCBC 221   825 completed_ops_.push(&task_op_); 226   867 completed_ops_.push(&task_op_);
HITCBC 222   825 } 227   867 }
223   228  
HITCBC 224   1650 inline epoll_scheduler::~epoll_scheduler() 229   1734 inline epoll_scheduler::~epoll_scheduler()
225   { 230   {
HITCBC 226   825 if (timer_fd_ >= 0) 231   867 if (timer_fd_ >= 0)
HITCBC 227   825 ::close(timer_fd_); 232   867 ::close(timer_fd_);
HITCBC 228   825 if (event_fd_ >= 0) 233   867 if (event_fd_ >= 0)
HITCBC 229   825 ::close(event_fd_); 234   867 ::close(event_fd_);
HITCBC 230   825 if (epoll_fd_ >= 0) 235   867 if (epoll_fd_ >= 0)
HITCBC 231   825 ::close(epoll_fd_); 236   867 ::close(epoll_fd_);
HITCBC 232   1650 } 237   1734 }
233   238  
234   inline void 239   inline void
HITCBC 235   825 epoll_scheduler::shutdown() 240   867 epoll_scheduler::shutdown()
236   { 241   {
HITCBC 237   825 shutdown_drain(); 242   867 shutdown_drain();
238   243  
HITCBC 239   825 if (event_fd_ >= 0) 244   867 if (event_fd_ >= 0)
HITCBC 240   825 interrupt_reactor(); 245   867 interrupt_reactor();
HITCBC 241   825 } 246   867 }
242   247  
243   inline void 248   inline void
HITCBC 244   19 epoll_scheduler::configure_reactor( 249   19 epoll_scheduler::configure_reactor(
245   unsigned max_events, 250   unsigned max_events,
246   unsigned budget_init, 251   unsigned budget_init,
247   unsigned budget_max, 252   unsigned budget_max,
248   unsigned unassisted) 253   unsigned unassisted)
249   { 254   {
HITCBC 250   19 reactor_scheduler::configure_reactor( 255   19 reactor_scheduler::configure_reactor(
251   max_events, budget_init, budget_max, unassisted); 256   max_events, budget_init, budget_max, unassisted);
HITCBC 252   18 event_buffer_.resize(max_events_per_poll_); 257   18 event_buffer_.resize(max_events_per_poll_);
HITCBC 253   18 } 258   18 }
254   259  
255 - inline void 260 + inline std::error_code
HITCBC 256   5065 epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const 261   5108 epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const
257   { 262   {
HITCBC 258   5065 epoll_event ev{}; 263   5108 epoll_event ev{};
HITCBC 259   5065 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP; 264   5108 ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
HITCBC 260   5065 ev.data.ptr = desc; 265   5108 ev.data.ptr = desc;
261   266  
HITCBC 262   5065 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) 267   5108 if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
HITGBC 263 - ✗ detail::throw_system_error(make_err(errno), "epoll_ctl (register)"); 268 + 1 return make_err(errno);
264   269  
HITCBC 265   5065 desc->registered_events = ev.events; 270   5107 desc->registered_events = ev.events;
HITCBC 266   5065 desc->fd = fd; 271   5107 desc->fd = fd;
HITCBC 267   5065 desc->scheduler_ = this; 272   5107 desc->scheduler_ = this;
HITCBC 268   5065 desc->mutex.set_enabled(reactor_io_locking_); 273   5107 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 269   5065 desc->ready_events_.store(0, std::memory_order_relaxed); 274   5107 desc->ready_events_.store(0, std::memory_order_relaxed);
270   275  
HITCBC 271   5065 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 276   5107 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 272   5065 desc->impl_ref_.reset(); 277   5107 desc->impl_ref_.reset();
HITCBC 273   5065 desc->read_ready = false; 278   5107 desc->read_ready = false;
HITCBC 274   5065 desc->write_ready = false; 279   5107 desc->write_ready = false;
HITGNC   280 + 5107 return {};
HITCBC 275   5065 } 281   5107 }
276   282  
277   inline void 283   inline void
HITCBC 278   5014 epoll_scheduler::deregister_descriptor(int fd) const 284   5056 epoll_scheduler::deregister_descriptor(int fd) const
279   { 285   {
HITCBC 280   5014 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr); 286   5056 ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
HITCBC 281   5014 } 287   5056 }
282   288  
283   inline void 289   inline void
HITCBC 284   4681 epoll_scheduler::interrupt_reactor() const 290   4757 epoll_scheduler::interrupt_reactor() const
285   { 291   {
HITCBC 286   4681 bool expected = false; 292   4757 bool expected = false;
HITCBC 287   4681 if (eventfd_armed_.compare_exchange_strong( 293   4757 if (eventfd_armed_.compare_exchange_strong(
288   expected, true, std::memory_order_release, 294   expected, true, std::memory_order_release,
289   std::memory_order_relaxed)) 295   std::memory_order_relaxed))
290   { 296   {
HITCBC 291   3485 std::uint64_t val = 1; 297   3530 std::uint64_t val = 1;
HITCBC 292   3485 [[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val)); 298   3530 [[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val));
293   } 299   }
HITCBC 294   4681 } 300   4757 }
295   301  
296   inline void 302   inline void
HITCBC 297   4866 epoll_scheduler::update_timerfd() const 303   4828 epoll_scheduler::update_timerfd() const
298   { 304   {
HITCBC 299   4866 auto nearest = timer_svc_->nearest_expiry(); 305   4828 auto nearest = timer_svc_->nearest_expiry();
300   306  
HITCBC 301   4866 itimerspec ts{}; 307   4828 itimerspec ts{};
HITCBC 302   4866 int flags = 0; 308   4828 int flags = 0;
303   309  
HITCBC 304   4866 if (nearest == timer_service::time_point::max()) 310   4828 if (nearest == timer_service::time_point::max())
305   { 311   {
306   // No timers — disarm by setting to 0 (relative) 312   // No timers — disarm by setting to 0 (relative)
307   } 313   }
308   else 314   else
309   { 315   {
HITCBC 310   4694 auto now = std::chrono::steady_clock::now(); 316   4649 auto now = std::chrono::steady_clock::now();
HITCBC 311   4694 if (nearest <= now) 317   4649 if (nearest <= now)
312   { 318   {
313   // Use 1ns instead of 0 — zero disarms the timerfd 319   // Use 1ns instead of 0 — zero disarms the timerfd
HITCBC 314   325 ts.it_value.tv_nsec = 1; 320   438 ts.it_value.tv_nsec = 1;
315   } 321   }
316   else 322   else
317   { 323   {
HITCBC 318   4369 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>( 324   4211 auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
HITCBC 319   4369 nearest - now) 325   4211 nearest - now)
HITCBC 320   4369 .count(); 326   4211 .count();
HITCBC 321   4369 ts.it_value.tv_sec = nsec / 1000000000; 327   4211 ts.it_value.tv_sec = nsec / 1000000000;
HITCBC 322   4369 ts.it_value.tv_nsec = nsec % 1000000000; 328   4211 ts.it_value.tv_nsec = nsec % 1000000000;
HITCBC 323   4369 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0) 329   4211 if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
MISUBC 324   ✗ ts.it_value.tv_nsec = 1; 330   ✗ ts.it_value.tv_nsec = 1;
325   } 331   }
326   } 332   }
327   333  
HITCBC 328   4866 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0) 334   4828 if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
MISUBC 329   ✗ detail::throw_system_error(make_err(errno), "timerfd_settime"); 335   ✗ detail::throw_system_error(make_err(errno), "timerfd_settime");
HITCBC 330   4866 } 336   4828 }
331   337  
332   inline void 338   inline void
HITCBC 333   31284 epoll_scheduler::run_task( 339   29696 epoll_scheduler::run_task(
334   lock_type& lock, context_type* ctx, long timeout_us) 340   lock_type& lock, context_type* ctx, long timeout_us)
335   { 341   {
336   int timeout_ms; 342   int timeout_ms;
HITCBC 337   31284 if (task_interrupted_) 343   29696 if (task_interrupted_)
HITCBC 338   24496 timeout_ms = 0; 344   23347 timeout_ms = 0;
HITCBC 339   6788 else if (timeout_us < 0) 345   6349 else if (timeout_us < 0)
HITCBC 340   6784 timeout_ms = -1; 346   6345 timeout_ms = -1;
341   else 347   else
HITCBC 342   4 timeout_ms = static_cast<int>((timeout_us + 999) / 1000); 348   4 timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
343   349  
HITCBC 344   31284 if (lock.owns_lock()) 350   29696 if (lock.owns_lock())
HITCBC 345   6790 lock.unlock(); 351   6351 lock.unlock();
346   352  
HITCBC 347   31284 task_cleanup on_exit{this, &lock, ctx}; 353   29696 task_cleanup on_exit{this, &lock, ctx};
348   354  
349   // Flush deferred timerfd programming before blocking 355   // Flush deferred timerfd programming before blocking
HITCBC 350   31284 if (timerfd_stale_.exchange(false, std::memory_order_acquire)) 356   29696 if (timerfd_stale_.exchange(false, std::memory_order_acquire))
HITCBC 351   2440 update_timerfd(); 357   2422 update_timerfd();
352   358  
HITCBC 353   31284 int nfds = ::epoll_wait( 359   29696 int nfds = ::epoll_wait(
354   epoll_fd_, event_buffer_.data(), 360   epoll_fd_, event_buffer_.data(),
HITCBC 355   31284 static_cast<int>(event_buffer_.size()), timeout_ms); 361   29696 static_cast<int>(event_buffer_.size()), timeout_ms);
356   362  
HITCBC 357   31284 if (nfds < 0 && errno != EINTR) 363   29696 if (nfds < 0 && errno != EINTR)
MISUBC 358   ✗ detail::throw_system_error(make_err(errno), "epoll_wait"); 364   ✗ detail::throw_system_error(make_err(errno), "epoll_wait");
359   365  
HITCBC 360   31284 bool check_timers = false; 366   29696 bool check_timers = false;
HITCBC 361   31284 ready_queue local_ops; 367   29696 ready_queue local_ops;
362   368  
HITCBC 363   70920 for (int i = 0; i < nfds; ++i) 369   69126 for (int i = 0; i < nfds; ++i)
364   { 370   {
HITCBC 365   39636 if (event_buffer_[i].data.ptr == nullptr) 371   39430 if (event_buffer_[i].data.ptr == nullptr)
366   { 372   {
367   std::uint64_t val; 373   std::uint64_t val;
368   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 374   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
HITCBC 369   2660 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val)); 375   2663 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
HITCBC 370   2660 eventfd_armed_.store(false, std::memory_order_relaxed); 376   2663 eventfd_armed_.store(false, std::memory_order_relaxed);
HITCBC 371   2660 continue; 377   2663 continue;
HITCBC 372   2660 } 378   2663 }
373   379  
HITCBC 374   36976 if (event_buffer_[i].data.ptr == &timer_fd_) 380   36767 if (event_buffer_[i].data.ptr == &timer_fd_)
375   { 381   {
376   std::uint64_t expirations; 382   std::uint64_t expirations;
377   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection) 383   // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
378   [[maybe_unused]] auto r = 384   [[maybe_unused]] auto r =
HITCBC 379   2426 ::read(timer_fd_, &expirations, sizeof(expirations)); 385   2406 ::read(timer_fd_, &expirations, sizeof(expirations));
HITCBC 380   2426 check_timers = true; 386   2406 check_timers = true;
HITCBC 381   2426 continue; 387   2406 continue;
HITCBC 382   2426 } 388   2406 }
383   389  
384   auto* desc = 390   auto* desc =
HITCBC 385   34550 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr); 391   34361 static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
HITCBC 386   34550 desc->add_ready_events(event_buffer_[i].events); 392   34361 desc->add_ready_events(event_buffer_[i].events);
387   393  
HITCBC 388   34550 bool expected = false; 394   34361 bool expected = false;
HITCBC 389   34550 if (desc->is_enqueued_.compare_exchange_strong( 395   34361 if (desc->is_enqueued_.compare_exchange_strong(
390   expected, true, std::memory_order_release, 396   expected, true, std::memory_order_release,
391   std::memory_order_relaxed)) 397   std::memory_order_relaxed))
392   { 398   {
HITCBC 393   34550 local_ops.push(desc); 399   34361 local_ops.push(desc);
394   } 400   }
395   } 401   }
396   402  
HITCBC 397   31284 if (check_timers) 403   29696 if (check_timers)
398   { 404   {
HITCBC 399   2426 timer_svc_->process_expired(); 405   2406 timer_svc_->process_expired();
HITCBC 400   2426 update_timerfd(); 406   2406 update_timerfd();
401   } 407   }
402   408  
HITCBC 403   31284 lock.lock(); 409   29696 lock.lock();
404   410  
HITCBC 405   31284 completed_ops_.splice(local_ops); 411   29696 completed_ops_.splice(local_ops);
HITCBC 406   31284 } 412   29696 }
407   413  
408   } // namespace boost::corosio::detail 414   } // namespace boost::corosio::detail
409   415  
410   #endif // BOOST_COROSIO_HAS_EPOLL 416   #endif // BOOST_COROSIO_HAS_EPOLL
411   417  
412   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP 418   #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP