Diff coverage report for

//

//

// Copyright (c) 2026 Steve Gerbino

// Copyright (c) 2026 Steve Gerbino

//

//

// Distributed under the Boost Software License, Version 1.0. (See accompanying

// Distributed under the Boost Software License, Version 1.0. (See accompanying

// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

//

//

// Official repository: https://github.com/cppalliance/corosio

// Official repository: https://github.com/cppalliance/corosio

//

//

#ifndef BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP

#ifndef BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP

#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP

#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP

#include <boost/corosio/detail/platform.hpp>

#include <boost/corosio/detail/platform.hpp>

#if BOOST_COROSIO_HAS_EPOLL

#if BOOST_COROSIO_HAS_EPOLL

#include <boost/corosio/detail/config.hpp>

#include <boost/corosio/detail/config.hpp>

#include <boost/capy/ex/execution_context.hpp>

#include <boost/capy/ex/execution_context.hpp>

#include <boost/corosio/native/native_scheduler.hpp>

#include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>

#include <boost/corosio/detail/scheduler_op.hpp>

#include <boost/corosio/native/detail/epoll/epoll_op.hpp>

#include <boost/corosio/native/detail/epoll/epoll_op.hpp>

#include <boost/corosio/detail/timer_service.hpp>

#include <boost/corosio/detail/timer_service.hpp>

#include <boost/corosio/native/detail/make_err.hpp>

#include <boost/corosio/native/detail/make_err.hpp>

#include <boost/corosio/native/detail/posix/posix_resolver_service.hpp>

#include <boost/corosio/native/detail/posix/posix_resolver_service.hpp>

#include <boost/corosio/native/detail/posix/posix_signal_service.hpp>

#include <boost/corosio/native/detail/posix/posix_signal_service.hpp>

#include <boost/corosio/detail/thread_local_ptr.hpp>

#include <boost/corosio/detail/except.hpp>

#include <boost/corosio/detail/except.hpp>

#include <atomic>

#include <atomic>

#include <condition_variable>

#include <cstddef>

#include <chrono>

#include <chrono>

#include <limits>

#include <cstdint>

#include <cstdint>

#include <utility>

#include <mutex>

#include <mutex>

#include <fcntl.h>

#include <errno.h>

#include <errno.h>

#include <sys/epoll.h>

#include <sys/epoll.h>

#include <sys/socket.h>

#include <sys/eventfd.h>

#include <sys/eventfd.h>

#include <sys/timerfd.h>

#include <sys/timerfd.h>

#include <unistd.h>

#include <unistd.h>

namespace boost::corosio::detail {

namespace boost::corosio::detail {

struct epoll_op;

struct epoll_op;

namespace epoll {

struct BOOST_COROSIO_SYMBOL_VISIBLE scheduler_context;

} // namespace epoll

struct descriptor_state;

struct descriptor_state;

/** Linux scheduler using epoll for I/O multiplexing.

/** Linux scheduler using epoll for I/O multiplexing.

    This scheduler implements the scheduler interface using Linux epoll

    This scheduler implements the scheduler interface using Linux epoll

    for efficient I/O event notification. It uses a single reactor model

    for efficient I/O event notification. It uses a single reactor model

    where one thread runs epoll_wait while other threads

    where one thread runs epoll_wait while other threads

    wait on a condition variable for handler work. This design provides:

    wait on a condition variable for handler work. This design provides:

    - Handler parallelism: N posted handlers can execute on N threads

    - Handler parallelism: N posted handlers can execute on N threads

    - No thundering herd: condition_variable wakes exactly one thread

    - No thundering herd: condition_variable wakes exactly one thread

    - IOCP parity: Behavior matches Windows I/O completion port semantics

    - IOCP parity: Behavior matches Windows I/O completion port semantics

    When threads call run(), they first try to execute queued handlers.

    When threads call run(), they first try to execute queued handlers.

    If the queue is empty and no reactor is running, one thread becomes

    If the queue is empty and no reactor is running, one thread becomes

    the reactor and runs epoll_wait. Other threads wait on a condition

    the reactor and runs epoll_wait. Other threads wait on a condition

    variable until handlers are available.

    variable until handlers are available.

    @par Thread Safety

    @par Thread Safety

    All public member functions are thread-safe.

    All public member functions are thread-safe.

*/

*/

class BOOST_COROSIO_DECL epoll_scheduler final

class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler_base

    : public native_scheduler

    , public capy::execution_context::service

{

{

    using key_type = scheduler;

public:

public:

    /** Construct the scheduler.

    /** Construct the scheduler.

        Creates an epoll instance, eventfd for reactor interruption,

        Creates an epoll instance, eventfd for reactor interruption,

        and timerfd for kernel-managed timer expiry.

        and timerfd for kernel-managed timer expiry.

        @param ctx Reference to the owning execution_context.

        @param ctx Reference to the owning execution_context.

        @param concurrency_hint Hint for expected thread count (unused).

        @param concurrency_hint Hint for expected thread count (unused).

    */

    */

    epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);

    epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);

    /// Destroy the scheduler.

    /// Destroy the scheduler.

    ~epoll_scheduler() override;

    ~epoll_scheduler() override;

    epoll_scheduler(epoll_scheduler const&)            = delete;

    epoll_scheduler(epoll_scheduler const&)            = delete;

    epoll_scheduler& operator=(epoll_scheduler const&) = delete;

    epoll_scheduler& operator=(epoll_scheduler const&) = delete;

    /// Shut down the scheduler, draining pending operations.

    void post(std::coroutine_handle<> h) const override;

    void post(scheduler_op* h) const override;

    bool running_in_this_thread() const noexcept override;

    void stop() override;

    bool stopped() const noexcept override;

    void restart() override;

    std::size_t run() override;

    std::size_t run_one() override;

    std::size_t wait_one(long usec) override;

    std::size_t poll() override;

    std::size_t poll_one() override;

    void shutdown() override;

    void shutdown() override;

    /** Return the epoll file descriptor.

    /** Return the epoll file descriptor.

        Used by socket services to register file descriptors

        Used by socket services to register file descriptors

        for I/O event notification.

        for I/O event notification.

        @return The epoll file descriptor.

        @return The epoll file descriptor.

    */

    */

    int epoll_fd() const noexcept

    int epoll_fd() const noexcept

    {

    {

        return epoll_fd_;

        return epoll_fd_;

    }

    }

    /** Reset the thread's inline completion budget.

        Called at the start of each posted completion handler to

        grant a fresh budget for speculative inline completions.

    */

    void reset_inline_budget() const noexcept;

    /** Consume one unit of inline budget if available.

        @return True if budget was available and consumed.

    */

    bool try_consume_inline_budget() const noexcept;

    /** Register a descriptor for persistent monitoring.

    /** Register a descriptor for persistent monitoring.

        The fd is registered once and stays registered until explicitly

        The fd is registered once and stays registered until explicitly

        deregistered. Events are dispatched via descriptor_state which

        deregistered. Events are dispatched via descriptor_state which

        tracks pending read/write/connect operations.

        tracks pending read/write/connect operations.

        @param fd The file descriptor to register.

        @param fd The file descriptor to register.

        @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).

        @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).

    */

    */

    void register_descriptor(int fd, descriptor_state* desc) const;

    void register_descriptor(int fd, descriptor_state* desc) const;

    /** Deregister a persistently registered descriptor.

    /** Deregister a persistently registered descriptor.

        @param fd The file descriptor to deregister.

        @param fd The file descriptor to deregister.

    */

    */

    void deregister_descriptor(int fd) const;

    void deregister_descriptor(int fd) const;

    void work_started() noexcept override;

    void work_finished() noexcept override;

    /** Offset a forthcoming work_finished from work_cleanup.

        Called by descriptor_state when all I/O returned EAGAIN and no

        handler will be executed. Must be called from a scheduler thread.

    */

    void compensating_work_started() const noexcept;

    /** Drain work from thread context's private queue to global queue.

        Called by thread_context_guard destructor when a thread exits run().

        Transfers pending work to the global queue under mutex protection.

        @param queue The private queue to drain.

        @param count Item count for wakeup decisions (wakes other threads if positive).

    */

    void drain_thread_queue(op_queue& queue, long count) const;

    /** Post completed operations for deferred invocation.

        If called from a thread running this scheduler, operations go to

        the thread's private queue (fast path). Otherwise, operations are

        added to the global queue under mutex and a waiter is signaled.

        @par Preconditions

        work_started() must have been called for each operation.

        @param ops Queue of operations to post.

    */

    void post_deferred_completions(op_queue& ops) const;

    struct work_cleanup

    {

        epoll_scheduler* scheduler;

        std::unique_lock<std::mutex>* lock;

        epoll::scheduler_context* ctx;

        ~work_cleanup();

    };

    struct task_cleanup

    {

        epoll_scheduler const* scheduler;

        std::unique_lock<std::mutex>* lock;

        epoll::scheduler_context* ctx;

        ~task_cleanup();

    };

    std::size_t do_one(

        std::unique_lock<std::mutex>& lock,

        long timeout_us,

        epoll::scheduler_context* ctx);

private:

private:

    void

    void

    run_task(std::unique_lock<std::mutex>& lock, epoll::scheduler_context* ctx);

    run_task(std::unique_lock<std::mutex>& lock, context_type* ctx) override;

    void wake_one_thread_and_unlock(std::unique_lock<std::mutex>& lock) const;

    void interrupt_reactor() const override;

    void interrupt_reactor() const;

    void update_timerfd() const;

    void update_timerfd() const;

    /** Set the signaled state and wake all waiting threads.

        @par Preconditions

        Mutex must be held.

        @param lock The held mutex lock.

    */

    void signal_all(std::unique_lock<std::mutex>& lock) const;

    /** Set the signaled state and wake one waiter if any exist.

        Only unlocks and signals if at least one thread is waiting.

        Use this when the caller needs to perform a fallback action

        (such as interrupting the reactor) when no waiters exist.

        @par Preconditions

        Mutex must be held.

        @param lock The held mutex lock.

        @return `true` if unlocked and signaled, `false` if lock still held.

    */

    bool maybe_unlock_and_signal_one(std::unique_lock<std::mutex>& lock) const;

    /** Set the signaled state, unlock, and wake one waiter if any exist.

        Always unlocks the mutex. Use this when the caller will release

        the lock regardless of whether a waiter exists.

        @par Preconditions

        Mutex must be held.

        @param lock The held mutex lock.

        @return `true` if a waiter was signaled, `false` otherwise.

    */

    bool unlock_and_signal_one(std::unique_lock<std::mutex>& lock) const;

    /** Clear the signaled state before waiting.

        @par Preconditions

        Mutex must be held.

    */

    void clear_signal() const;

    /** Block until the signaled state is set.

        Returns immediately if already signaled (fast-path). Otherwise

        increments the waiter count, waits on the condition variable,

        and decrements the waiter count upon waking.

        @par Preconditions

        Mutex must be held.

        @param lock The held mutex lock.

    */

    void wait_for_signal(std::unique_lock<std::mutex>& lock) const;

    /** Block until signaled or timeout expires.

        @par Preconditions

        Mutex must be held.

        @param lock The held mutex lock.

        @param timeout_us Maximum time to wait in microseconds.

    */

    void wait_for_signal_for(

        std::unique_lock<std::mutex>& lock, long timeout_us) const;

    int epoll_fd_;

    int epoll_fd_;

    int event_fd_; // for interrupting reactor

    int event_fd_;

    int timer_fd_; // timerfd for kernel-managed timer expiry

    int timer_fd_;

    mutable std::mutex mutex_;

    mutable std::condition_variable cond_;

    mutable op_queue completed_ops_;

    mutable std::atomic<long> outstanding_work_;

    bool stopped_;

    // True while a thread is blocked in epoll_wait. Used by

    // wake_one_thread_and_unlock and work_finished to know when

    // an eventfd interrupt is needed instead of a condvar signal.

    mutable std::atomic<bool> task_running_{false};

    // True when the reactor has been told to do a non-blocking poll

    // (more handlers queued or poll mode). Prevents redundant eventfd

    // writes and controls the epoll_wait timeout.

    mutable bool task_interrupted_ = false;

    // Signaling state: bit 0 = signaled, upper bits = waiter count (incremented by 2)

    mutable std::size_t state_ = 0;

    // Edge-triggered eventfd state

    // Edge-triggered eventfd state

    mutable std::atomic<bool> eventfd_armed_{false};

    mutable std::atomic<bool> eventfd_armed_{false};

    // blocks. Avoids timerfd_settime syscalls for timers that are

    // scheduled then cancelled without being waited on.

    // Set when the earliest timer changes; flushed before epoll_wait

    // Set when the earliest timer changes; flushed before epoll_wait

    // Sentinel operation for interleaving reactor runs with handler execution.

    // Ensures the reactor runs periodically even when handlers are continuously

    // posted, preventing starvation of I/O events, timers, and signals.

    struct task_op final : scheduler_op

    {

        void operator()() override {}

    task_op task_op_;

    mutable std::atomic<bool> timerfd_stale_{false};

    mutable std::atomic<bool> timerfd_stale_{false};

};

};

//--------------------------------------------------------------------------

//

// Implementation

//

//--------------------------------------------------------------------------

/*

    epoll Scheduler - Single Reactor Model

    ======================================

    This scheduler uses a thread coordination strategy to provide handler

    parallelism and avoid the thundering herd problem.

    Instead of all threads blocking on epoll_wait(), one thread becomes the

    "reactor" while others wait on a condition variable for handler work.

    Thread Model

    ------------

    - ONE thread runs epoll_wait() at a time (the reactor thread)

    - OTHER threads wait on cond_ (condition variable) for handlers

    - When work is posted, exactly one waiting thread wakes via notify_one()

    - This matches Windows IOCP semantics where N posted items wake N threads

    Event Loop Structure (do_one)

    -----------------------------

    1. Lock mutex, try to pop handler from queue

    2. If got handler: execute it (unlocked), return

    3. If queue empty and no reactor running: become reactor

       - Run epoll_wait (unlocked), queue I/O completions, loop back

    4. If queue empty and reactor running: wait on condvar for work

    The task_running_ flag ensures only one thread owns epoll_wait().

    After the reactor queues I/O completions, it loops back to try getting

    a handler, giving priority to handler execution over more I/O polling.

    Signaling State (state_)

    ------------------------

    The state_ variable encodes two pieces of information:

    - Bit 0: signaled flag (1 = signaled, persists until cleared)

    - Upper bits: waiter count (each waiter adds 2 before blocking)

    This allows efficient coordination:

    - Signalers only call notify when waiters exist (state_ > 1)

    - Waiters check if already signaled before blocking (fast-path)

    Wake Coordination (wake_one_thread_and_unlock)

    ----------------------------------------------

    When posting work:

    - If waiters exist (state_ > 1): signal and notify_one()

    - Else if reactor running: interrupt via eventfd write

    - Else: no-op (thread will find work when it checks queue)

    This avoids waking threads unnecessarily. With cascading wakes,

    each handler execution wakes at most one additional thread if

    more work exists in the queue.

    Work Counting

    -------------

    outstanding_work_ tracks pending operations. When it hits zero, run()

    returns. Each operation increments on start, decrements on completion.

    Timer Integration

    -----------------

    Timers are handled by timer_service. The reactor adjusts epoll_wait

    timeout to wake for the nearest timer expiry. When a new timer is

    scheduled earlier than current, timer_service calls interrupt_reactor()

    to re-evaluate the timeout.

*/

namespace epoll {

struct BOOST_COROSIO_SYMBOL_VISIBLE scheduler_context

{

    epoll_scheduler const* key;

    scheduler_context* next;

    op_queue private_queue;

    long private_outstanding_work;

    int inline_budget;

    int inline_budget_max;

    bool unassisted;

    scheduler_context(epoll_scheduler const* k, scheduler_context* n)

    }

inline thread_local_ptr<scheduler_context> context_stack;

struct thread_context_guard

{

    scheduler_context frame_;

    explicit thread_context_guard(epoll_scheduler const* ctx) noexcept

        context_stack.set(&frame_);

    ~thread_context_guard() noexcept

        if (!frame_.private_queue.empty())

inline scheduler_context*

find_context(epoll_scheduler const* self) noexcept

    for (auto* c = context_stack.get(); c != nullptr; c = c->next)

} // namespace epoll

inline void

epoll_scheduler::reset_inline_budget() const noexcept

    if (auto* ctx = epoll::find_context(this))

        // Cap when no other thread absorbed queued work. A moderate

        // cap (4) amortizes scheduling for small buffers while avoiding

        // bursty I/O that fills socket buffers and stalls large transfers.

        if (ctx->unassisted)

            ctx->inline_budget_max = 4;

        // Ramp up when previous cycle fully consumed budget.

        // Reset on partial consumption (EAGAIN hit or peer got scheduled).

        if (ctx->inline_budget == 0)

}

inline bool

epoll_scheduler::try_consume_inline_budget() const noexcept

    if (auto* ctx = epoll::find_context(this))

        if (ctx->inline_budget > 0)

            --ctx->inline_budget;

    }

    return false;

inline void

descriptor_state::operator()()

    is_enqueued_.store(false, std::memory_order_relaxed);

    // Take ownership of impl ref set by close_socket() to prevent

    // the owning impl from being freed while we're executing

    auto prevent_impl_destruction = std::move(impl_ref_);

    std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);

        scheduler_->compensating_work_started();

    op_queue local_ops;

    int err = 0;

        socklen_t len = sizeof(err);

    {

        std::lock_guard lock(mutex);

            if (read_op)

                auto* rd = read_op;

                    rd->perform_io();

                if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)

                    rd->errn = 0;

                else

                {

                    read_op = nullptr;

            }

            else

            {

                read_ready = true;

        }

        if (ev & EPOLLOUT)

            bool had_write_op = (connect_op || write_op);

                auto* cn = connect_op;

                    cn->perform_io();

            if (write_op)

                auto* wr = write_op;

                    wr->perform_io();

                if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)