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