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_op.hpp>
22   
#include <boost/corosio/native/detail/epoll/epoll_op.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  
27  
28   
#include <boost/corosio/detail/except.hpp>
28   
#include <boost/corosio/detail/except.hpp>
29  
29  
30   
#include <atomic>
30   
#include <atomic>
31   
#include <chrono>
31   
#include <chrono>
32   
#include <cstdint>
32   
#include <cstdint>
33   
#include <mutex>
33   
#include <mutex>
34  
34  
35   
#include <errno.h>
35   
#include <errno.h>
36   
#include <sys/epoll.h>
36   
#include <sys/epoll.h>
37   
#include <sys/eventfd.h>
37   
#include <sys/eventfd.h>
38   
#include <sys/timerfd.h>
38   
#include <sys/timerfd.h>
39   
#include <unistd.h>
39   
#include <unistd.h>
40  
40  
41   
namespace boost::corosio::detail {
41   
namespace boost::corosio::detail {
42  
42  
43   
struct epoll_op;
43   
struct epoll_op;
44   
struct descriptor_state;
44   
struct descriptor_state;
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_base
65   
class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler_base
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   
    /** Return the epoll file descriptor.
87   
    /** Return the epoll file descriptor.
88  
88  
89   
        Used by socket services to register file descriptors
89   
        Used by socket services to register file descriptors
90   
        for I/O event notification.
90   
        for I/O event notification.
91  
91  
92   
        @return The epoll file descriptor.
92   
        @return The epoll file descriptor.
93   
    */
93   
    */
94   
    int epoll_fd() const noexcept
94   
    int epoll_fd() const noexcept
95   
    {
95   
    {
96   
        return epoll_fd_;
96   
        return epoll_fd_;
97   
    }
97   
    }
98  
98  
99   
    /** Register a descriptor for persistent monitoring.
99   
    /** Register a descriptor for persistent monitoring.
100  
100  
101   
        The fd is registered once and stays registered until explicitly
101   
        The fd is registered once and stays registered until explicitly
102   
        deregistered. Events are dispatched via descriptor_state which
102   
        deregistered. Events are dispatched via descriptor_state which
103   
        tracks pending read/write/connect operations.
103   
        tracks pending read/write/connect operations.
104  
104  
105   
        @param fd The file descriptor to register.
105   
        @param fd The file descriptor to register.
106   
        @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
106   
        @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
107   
    */
107   
    */
108   
    void register_descriptor(int fd, descriptor_state* desc) const;
108   
    void register_descriptor(int fd, descriptor_state* desc) const;
109  
109  
110   
    /** Deregister a persistently registered descriptor.
110   
    /** Deregister a persistently registered descriptor.
111  
111  
112   
        @param fd The file descriptor to deregister.
112   
        @param fd The file descriptor to deregister.
113   
    */
113   
    */
114   
    void deregister_descriptor(int fd) const;
114   
    void deregister_descriptor(int fd) const;
115  
115  
116   
private:
116   
private:
117   
    void
117   
    void
118   
    run_task(std::unique_lock<std::mutex>& lock, context_type* ctx) override;
118   
    run_task(std::unique_lock<std::mutex>& lock, context_type* ctx) override;
119   
    void interrupt_reactor() const override;
119   
    void interrupt_reactor() const override;
120   
    void update_timerfd() const;
120   
    void update_timerfd() const;
121  
121  
122   
    int epoll_fd_;
122   
    int epoll_fd_;
123   
    int event_fd_;
123   
    int event_fd_;
124   
    int timer_fd_;
124   
    int timer_fd_;
125  
125  
126   
    // Edge-triggered eventfd state
126   
    // Edge-triggered eventfd state
127   
    mutable std::atomic<bool> eventfd_armed_{false};
127   
    mutable std::atomic<bool> eventfd_armed_{false};
128  
128  
129   
    // Set when the earliest timer changes; flushed before epoll_wait
129   
    // Set when the earliest timer changes; flushed before epoll_wait
130   
    mutable std::atomic<bool> timerfd_stale_{false};
130   
    mutable std::atomic<bool> timerfd_stale_{false};
131   
};
131   
};
132  
132  
133   
inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
133   
inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
134   
    : epoll_fd_(-1)
134   
    : epoll_fd_(-1)
135   
    , event_fd_(-1)
135   
    , event_fd_(-1)
136   
    , timer_fd_(-1)
136   
    , timer_fd_(-1)
137   
{
137   
{
138   
    epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
138   
    epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
139   
    if (epoll_fd_ < 0)
139   
    if (epoll_fd_ < 0)
140   
        detail::throw_system_error(make_err(errno), "epoll_create1");
140   
        detail::throw_system_error(make_err(errno), "epoll_create1");
141  
141  
142   
    event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
142   
    event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
143   
    if (event_fd_ < 0)
143   
    if (event_fd_ < 0)
144   
    {
144   
    {
145   
        int errn = errno;
145   
        int errn = errno;
146   
        ::close(epoll_fd_);
146   
        ::close(epoll_fd_);
147   
        detail::throw_system_error(make_err(errn), "eventfd");
147   
        detail::throw_system_error(make_err(errn), "eventfd");
148   
    }
148   
    }
149  
149  
150   
    timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
150   
    timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
151   
    if (timer_fd_ < 0)
151   
    if (timer_fd_ < 0)
152   
    {
152   
    {
153   
        int errn = errno;
153   
        int errn = errno;
154   
        ::close(event_fd_);
154   
        ::close(event_fd_);
155   
        ::close(epoll_fd_);
155   
        ::close(epoll_fd_);
156   
        detail::throw_system_error(make_err(errn), "timerfd_create");
156   
        detail::throw_system_error(make_err(errn), "timerfd_create");
157   
    }
157   
    }
158  
158  
159   
    epoll_event ev{};
159   
    epoll_event ev{};
160   
    ev.events   = EPOLLIN | EPOLLET;
160   
    ev.events   = EPOLLIN | EPOLLET;
161   
    ev.data.ptr = nullptr;
161   
    ev.data.ptr = nullptr;
162   
    if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
162   
    if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
163   
    {
163   
    {
164   
        int errn = errno;
164   
        int errn = errno;
165   
        ::close(timer_fd_);
165   
        ::close(timer_fd_);
166   
        ::close(event_fd_);
166   
        ::close(event_fd_);
167   
        ::close(epoll_fd_);
167   
        ::close(epoll_fd_);
168   
        detail::throw_system_error(make_err(errn), "epoll_ctl");
168   
        detail::throw_system_error(make_err(errn), "epoll_ctl");
169   
    }
169   
    }
170  
170  
171   
    epoll_event timer_ev{};
171   
    epoll_event timer_ev{};
172   
    timer_ev.events   = EPOLLIN | EPOLLERR;
172   
    timer_ev.events   = EPOLLIN | EPOLLERR;
173   
    timer_ev.data.ptr = &timer_fd_;
173   
    timer_ev.data.ptr = &timer_fd_;
174   
    if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
174   
    if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
175   
    {
175   
    {
176   
        int errn = errno;
176   
        int errn = errno;
177   
        ::close(timer_fd_);
177   
        ::close(timer_fd_);
178   
        ::close(event_fd_);
178   
        ::close(event_fd_);
179   
        ::close(epoll_fd_);
179   
        ::close(epoll_fd_);
180   
        detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
180   
        detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
181   
    }
181   
    }
182  
182  
183   
    timer_svc_ = &get_timer_service(ctx, *this);
183   
    timer_svc_ = &get_timer_service(ctx, *this);
184   
    timer_svc_->set_on_earliest_changed(
184   
    timer_svc_->set_on_earliest_changed(
185   
        timer_service::callback(this, [](void* p) {
185   
        timer_service::callback(this, [](void* p) {
186   
            auto* self = static_cast<epoll_scheduler*>(p);
186   
            auto* self = static_cast<epoll_scheduler*>(p);
187   
            self->timerfd_stale_.store(true, std::memory_order_release);
187   
            self->timerfd_stale_.store(true, std::memory_order_release);
188   
            self->interrupt_reactor();
188   
            self->interrupt_reactor();
189   
        }));
189   
        }));
190  
190  
191   
    get_resolver_service(ctx, *this);
191   
    get_resolver_service(ctx, *this);
192   
    get_signal_service(ctx, *this);
192   
    get_signal_service(ctx, *this);
193  
193  
194   
    completed_ops_.push(&task_op_);
194   
    completed_ops_.push(&task_op_);
195   
}
195   
}
196  
196  
197   
inline epoll_scheduler::~epoll_scheduler()
197   
inline epoll_scheduler::~epoll_scheduler()
198   
{
198   
{
199   
    if (timer_fd_ >= 0)
199   
    if (timer_fd_ >= 0)
200   
        ::close(timer_fd_);
200   
        ::close(timer_fd_);
201   
    if (event_fd_ >= 0)
201   
    if (event_fd_ >= 0)
202   
        ::close(event_fd_);
202   
        ::close(event_fd_);
203   
    if (epoll_fd_ >= 0)
203   
    if (epoll_fd_ >= 0)
204   
        ::close(epoll_fd_);
204   
        ::close(epoll_fd_);
205   
}
205   
}
206  
206  
207   
inline void
207   
inline void
208   
epoll_scheduler::shutdown()
208   
epoll_scheduler::shutdown()
209   
{
209   
{
210   
    shutdown_drain();
210   
    shutdown_drain();
211  
211  
212   
    if (event_fd_ >= 0)
212   
    if (event_fd_ >= 0)
213   
        interrupt_reactor();
213   
        interrupt_reactor();
214   
}
214   
}
215  
215  
216   
inline void
216   
inline void
217   
epoll_scheduler::register_descriptor(int fd, descriptor_state* desc) const
217   
epoll_scheduler::register_descriptor(int fd, descriptor_state* desc) const
218   
{
218   
{
219   
    epoll_event ev{};
219   
    epoll_event ev{};
220   
    ev.events   = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
220   
    ev.events   = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
221   
    ev.data.ptr = desc;
221   
    ev.data.ptr = desc;
222  
222  
223   
    if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
223   
    if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
224   
        detail::throw_system_error(make_err(errno), "epoll_ctl (register)");
224   
        detail::throw_system_error(make_err(errno), "epoll_ctl (register)");
225  
225  
226   
    desc->registered_events = ev.events;
226   
    desc->registered_events = ev.events;
227   
    desc->fd                = fd;
227   
    desc->fd                = fd;
228   
    desc->scheduler_        = this;
228   
    desc->scheduler_        = this;
229   
    desc->ready_events_.store(0, std::memory_order_relaxed);
229   
    desc->ready_events_.store(0, std::memory_order_relaxed);
230  
230  
231   
    std::lock_guard lock(desc->mutex);
231   
    std::lock_guard lock(desc->mutex);
232   
    desc->impl_ref_.reset();
232   
    desc->impl_ref_.reset();
233   
    desc->read_ready  = false;
233   
    desc->read_ready  = false;
234   
    desc->write_ready = false;
234   
    desc->write_ready = false;
235   
}
235   
}
236  
236  
237   
inline void
237   
inline void
238   
epoll_scheduler::deregister_descriptor(int fd) const
238   
epoll_scheduler::deregister_descriptor(int fd) const
239   
{
239   
{
240   
    ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
240   
    ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
241   
}
241   
}
242  
242  
243   
inline void
243   
inline void
244   
epoll_scheduler::interrupt_reactor() const
244   
epoll_scheduler::interrupt_reactor() const
245   
{
245   
{
246   
    bool expected = false;
246   
    bool expected = false;
247   
    if (eventfd_armed_.compare_exchange_strong(
247   
    if (eventfd_armed_.compare_exchange_strong(
248   
            expected, true, std::memory_order_release,
248   
            expected, true, std::memory_order_release,
249   
            std::memory_order_relaxed))
249   
            std::memory_order_relaxed))
250   
    {
250   
    {
251   
        std::uint64_t val       = 1;
251   
        std::uint64_t val       = 1;
252   
        [[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val));
252   
        [[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val));
253   
    }
253   
    }
254   
}
254   
}
255  
255  
256   
inline void
256   
inline void
257   
epoll_scheduler::update_timerfd() const
257   
epoll_scheduler::update_timerfd() const
258   
{
258   
{
259   
    auto nearest = timer_svc_->nearest_expiry();
259   
    auto nearest = timer_svc_->nearest_expiry();
260  
260  
261   
    itimerspec ts{};
261   
    itimerspec ts{};
262   
    int flags = 0;
262   
    int flags = 0;
263  
263  
264   
    if (nearest == timer_service::time_point::max())
264   
    if (nearest == timer_service::time_point::max())
265   
    {
265   
    {
266   
        // No timers — disarm by setting to 0 (relative)
266   
        // No timers — disarm by setting to 0 (relative)
267   
    }
267   
    }
268   
    else
268   
    else
269   
    {
269   
    {
270   
        auto now = std::chrono::steady_clock::now();
270   
        auto now = std::chrono::steady_clock::now();
271   
        if (nearest <= now)
271   
        if (nearest <= now)
272   
        {
272   
        {
273   
            // Use 1ns instead of 0 — zero disarms the timerfd
273   
            // Use 1ns instead of 0 — zero disarms the timerfd
274   
            ts.it_value.tv_nsec = 1;
274   
            ts.it_value.tv_nsec = 1;
275   
        }
275   
        }
276   
        else
276   
        else
277   
        {
277   
        {
278   
            auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
278   
            auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
279   
                            nearest - now)
279   
                            nearest - now)
280   
                            .count();
280   
                            .count();
281   
            ts.it_value.tv_sec  = nsec / 1000000000;
281   
            ts.it_value.tv_sec  = nsec / 1000000000;
282   
            ts.it_value.tv_nsec = nsec % 1000000000;
282   
            ts.it_value.tv_nsec = nsec % 1000000000;
283   
            if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
283   
            if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
284   
                ts.it_value.tv_nsec = 1;
284   
                ts.it_value.tv_nsec = 1;
285   
        }
285   
        }
286   
    }
286   
    }
287  
287  
288   
    if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
288   
    if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
289   
        detail::throw_system_error(make_err(errno), "timerfd_settime");
289   
        detail::throw_system_error(make_err(errno), "timerfd_settime");
290   
}
290   
}
291  
291  
292   
inline void
292   
inline void
293   
epoll_scheduler::run_task(std::unique_lock<std::mutex>& lock, context_type* ctx)
293   
epoll_scheduler::run_task(std::unique_lock<std::mutex>& lock, context_type* ctx)
294   
{
294   
{
295   
    int timeout_ms = task_interrupted_ ? 0 : -1;
295   
    int timeout_ms = task_interrupted_ ? 0 : -1;
296  
296  
297   
    if (lock.owns_lock())
297   
    if (lock.owns_lock())
298   
        lock.unlock();
298   
        lock.unlock();
299  
299  
300   
    task_cleanup on_exit{this, &lock, ctx};
300   
    task_cleanup on_exit{this, &lock, ctx};
301  
301  
302   
    // Flush deferred timerfd programming before blocking
302   
    // Flush deferred timerfd programming before blocking
303   
    if (timerfd_stale_.exchange(false, std::memory_order_acquire))
303   
    if (timerfd_stale_.exchange(false, std::memory_order_acquire))
304   
        update_timerfd();
304   
        update_timerfd();
305  
305  
306   
    epoll_event events[128];
306   
    epoll_event events[128];
307   
    int nfds = ::epoll_wait(epoll_fd_, events, 128, timeout_ms);
307   
    int nfds = ::epoll_wait(epoll_fd_, events, 128, timeout_ms);
308  
308  
309   
    if (nfds < 0 && errno != EINTR)
309   
    if (nfds < 0 && errno != EINTR)
310   
        detail::throw_system_error(make_err(errno), "epoll_wait");
310   
        detail::throw_system_error(make_err(errno), "epoll_wait");
311  
311  
312   
    bool check_timers = false;
312   
    bool check_timers = false;
313   
    op_queue local_ops;
313   
    op_queue local_ops;
314  
314  
315   
    for (int i = 0; i < nfds; ++i)
315   
    for (int i = 0; i < nfds; ++i)
316   
    {
316   
    {
317   
        if (events[i].data.ptr == nullptr)
317   
        if (events[i].data.ptr == nullptr)
318   
        {
318   
        {
319   
            std::uint64_t val;
319   
            std::uint64_t val;
320   
            // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
320   
            // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
321   
            [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
321   
            [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
322   
            eventfd_armed_.store(false, std::memory_order_relaxed);
322   
            eventfd_armed_.store(false, std::memory_order_relaxed);
323   
            continue;
323   
            continue;
324   
        }
324   
        }
325  
325  
326   
        if (events[i].data.ptr == &timer_fd_)
326   
        if (events[i].data.ptr == &timer_fd_)
327   
        {
327   
        {
328   
            std::uint64_t expirations;
328   
            std::uint64_t expirations;
329   
            // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
329   
            // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
330   
            [[maybe_unused]] auto r =
330   
            [[maybe_unused]] auto r =
331   
                ::read(timer_fd_, &expirations, sizeof(expirations));
331   
                ::read(timer_fd_, &expirations, sizeof(expirations));
332   
            check_timers = true;
332   
            check_timers = true;
333   
            continue;
333   
            continue;
334   
        }
334   
        }
335  
335  
336   
        auto* desc = static_cast<descriptor_state*>(events[i].data.ptr);
336   
        auto* desc = static_cast<descriptor_state*>(events[i].data.ptr);
337   
        desc->add_ready_events(events[i].events);
337   
        desc->add_ready_events(events[i].events);
338  
338  
339   
        bool expected = false;
339   
        bool expected = false;
340   
        if (desc->is_enqueued_.compare_exchange_strong(
340   
        if (desc->is_enqueued_.compare_exchange_strong(
341   
                expected, true, std::memory_order_release,
341   
                expected, true, std::memory_order_release,
342   
                std::memory_order_relaxed))
342   
                std::memory_order_relaxed))
343   
        {
343   
        {
344   
            local_ops.push(desc);
344   
            local_ops.push(desc);
345   
        }
345   
        }
346   
    }
346   
    }
347  
347  
348   
    if (check_timers)
348   
    if (check_timers)
349   
    {
349   
    {
350   
        timer_svc_->process_expired();
350   
        timer_svc_->process_expired();
351   
        update_timerfd();
351   
        update_timerfd();
352   
    }
352   
    }
353  
353  
354   
    lock.lock();
354   
    lock.lock();
355  
355  
356   
    if (!local_ops.empty())
356   
    if (!local_ops.empty())
357   
        completed_ops_.splice(local_ops);
357   
        completed_ops_.splice(local_ops);
358   
}
358   
}
359  
359  
360   
} // namespace boost::corosio::detail
360   
} // namespace boost::corosio::detail
361  
361  
362   
#endif // BOOST_COROSIO_HAS_EPOLL
362   
#endif // BOOST_COROSIO_HAS_EPOLL
363  
363  
364   
#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
364   
#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP