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_OP_HPP

#ifndef BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_OP_HPP

#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_OP_HPP

#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_OP_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/corosio/io/io_object.hpp>

#include <boost/corosio/io/io_object.hpp>

#include <boost/corosio/endpoint.hpp>

#include <boost/corosio/endpoint.hpp>

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

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

#include <coroutine>

#include <coroutine>

#include <boost/capy/error.hpp>

#include <boost/capy/error.hpp>

#include <system_error>

#include <system_error>

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

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

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

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

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

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

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

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

#include <unistd.h>

#include <unistd.h>

#include <errno.h>

#include <errno.h>

#include <atomic>

#include <atomic>

#include <cstddef>

#include <cstddef>

#include <memory>

#include <memory>

#include <mutex>

#include <mutex>

#include <optional>

#include <optional>

#include <stop_token>

#include <stop_token>

#include <netinet/in.h>

#include <netinet/in.h>

#include <poll.h>

#include <sys/socket.h>

#include <sys/socket.h>

#include <sys/uio.h>

#include <sys/uio.h>

/*

/*

    epoll Operation State

    epoll Operation State

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

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

    Each async I/O operation has a corresponding epoll_op-derived struct that

    Each async I/O operation has a corresponding epoll_op-derived struct that

    holds the operation's state while it's in flight. The socket impl owns

    holds the operation's state while it's in flight. The socket impl owns

    fixed slots for each operation type (conn_, rd_, wr_), so only one

    fixed slots for each operation type (conn_, rd_, wr_), so only one

    operation of each type can be pending per socket at a time.

    operation of each type can be pending per socket at a time.

    Persistent Registration

    Persistent Registration

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

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

    File descriptors are registered with epoll once (via descriptor_state) and

    File descriptors are registered with epoll once (via descriptor_state) and

    stay registered until closed. The descriptor_state tracks which operations

    stay registered until closed. The descriptor_state tracks which operations

    are pending (read_op, write_op, connect_op). When an event arrives, the

    are pending (read_op, write_op, connect_op). When an event arrives, the

    reactor dispatches to the appropriate pending operation.

    reactor dispatches to the appropriate pending operation.

    Impl Lifetime Management

    Impl Lifetime Management

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

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

    When cancel() posts an op to the scheduler's ready queue, the socket impl

    When cancel() posts an op to the scheduler's ready queue, the socket impl

    might be destroyed before the scheduler processes the op. The `impl_ptr`

    might be destroyed before the scheduler processes the op. The `impl_ptr`

    member holds a shared_ptr to the impl, keeping it alive until the op

    member holds a shared_ptr to the impl, keeping it alive until the op

    completes. This is set by cancel() and cleared in operator() after the

    completes. This is set by cancel() and cleared in operator() after the

    coroutine is resumed.

    coroutine is resumed.

    EOF Detection

    EOF Detection

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

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

    For reads, 0 bytes with no error means EOF. But an empty user buffer also

    For reads, 0 bytes with no error means EOF. But an empty user buffer also

    returns 0 bytes. The `empty_buffer_read` flag distinguishes these cases.

    returns 0 bytes. The `empty_buffer_read` flag distinguishes these cases.

    SIGPIPE Prevention

    SIGPIPE Prevention

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

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

    Writes use sendmsg() with MSG_NOSIGNAL instead of writev() to prevent

    Writes use sendmsg() with MSG_NOSIGNAL instead of writev() to prevent

    SIGPIPE when the peer has closed.

    SIGPIPE when the peer has closed.

*/

*/

namespace boost::corosio::detail {

namespace boost::corosio::detail {

// Forward declarations

// Forward declarations

class epoll_socket;

class epoll_socket;

class epoll_acceptor;

class epoll_acceptor;

struct epoll_op;

struct epoll_op;

// Forward declaration

// Forward declaration

class epoll_scheduler;

class epoll_scheduler;

/** Per-descriptor state for persistent epoll registration.

/** Per-descriptor state for persistent epoll registration.

    Tracks pending operations for a file descriptor. The fd is registered

    Tracks pending operations for a file descriptor. The fd is registered

    once with epoll and stays registered until closed.

    once with epoll and stays registered until closed.

    This struct extends scheduler_op to support deferred I/O processing.

    This struct extends scheduler_op to support deferred I/O processing.

    When epoll events arrive, the reactor sets ready_events and queues

    When epoll events arrive, the reactor sets ready_events and queues

    this descriptor for processing. When popped from the scheduler queue,

    this descriptor for processing. When popped from the scheduler queue,

    operator() performs the actual I/O and queues completion handlers.

    operator() performs the actual I/O and queues completion handlers.

    @par Deferred I/O Model

    @par Deferred I/O Model

    The reactor no longer performs I/O directly. Instead:

    The reactor no longer performs I/O directly. Instead:

    1. Reactor sets ready_events and queues descriptor_state

    1. Reactor sets ready_events and queues descriptor_state

    2. Scheduler pops descriptor_state and calls operator()

    2. Scheduler pops descriptor_state and calls operator()

    3. operator() performs I/O under mutex and queues completions

    3. operator() performs I/O under mutex and queues completions

    This eliminates per-descriptor mutex locking from the reactor hot path.

    This eliminates per-descriptor mutex locking from the reactor hot path.

    @par Thread Safety

    @par Thread Safety

    The mutex protects operation pointers and ready flags during I/O.

    The mutex protects operation pointers and ready flags during I/O.

    ready_events_ and is_enqueued_ are atomic for lock-free reactor access.

    ready_events_ and is_enqueued_ are atomic for lock-free reactor access.

*/

*/

struct descriptor_state final : scheduler_op

struct descriptor_state final : scheduler_op

{

{

    std::mutex mutex;

    std::mutex mutex;

    // Protected by mutex

    // Protected by mutex

    epoll_op* read_op    = nullptr;

    epoll_op* read_op    = nullptr;

    epoll_op* write_op   = nullptr;

    epoll_op* write_op   = nullptr;

    epoll_op* connect_op = nullptr;

    epoll_op* connect_op = nullptr;

    // Caches edge events that arrived before an op was registered

    // Caches edge events that arrived before an op was registered

    bool read_ready  = false;

    bool read_ready  = false;

    bool write_ready = false;

    bool write_ready = false;

    // Deferred cancellation: set by cancel() when the target op is not

    // Deferred cancellation: set by cancel() when the target op is not

    // parked (e.g. completing inline via speculative I/O). Checked when

    // parked (e.g. completing inline via speculative I/O). Checked when

    // the next op parks; if set, the op is immediately self-cancelled.

    // the next op parks; if set, the op is immediately self-cancelled.

    // This matches IOCP semantics where CancelIoEx always succeeds.

    // This matches IOCP semantics where CancelIoEx always succeeds.

    bool read_cancel_pending    = false;

    bool read_cancel_pending    = false;

    bool write_cancel_pending   = false;

    bool write_cancel_pending   = false;

    bool connect_cancel_pending = false;

    bool connect_cancel_pending = false;

    // Set during registration only (no mutex needed)

    // Protected by mutex (written by register_descriptor and ensure_write_events)

    std::uint32_t registered_events = 0;

    std::uint32_t registered_events = 0;

    int fd                          = -1;

    int fd                          = -1;

    // For deferred I/O - set by reactor, read by scheduler

    // For deferred I/O - set by reactor, read by scheduler

    std::atomic<std::uint32_t> ready_events_{0};

    std::atomic<std::uint32_t> ready_events_{0};

    std::atomic<bool> is_enqueued_{false};

    std::atomic<bool> is_enqueued_{false};

    epoll_scheduler const* scheduler_ = nullptr;

    epoll_scheduler const* scheduler_ = nullptr;

    // Prevents impl destruction while this descriptor_state is queued.

    // Prevents impl destruction while this descriptor_state is queued.

    // Set by close_socket() when is_enqueued_ is true, cleared by operator().

    // Set by close_socket() when is_enqueued_ is true, cleared by operator().

    std::shared_ptr<void> impl_ref_;

    std::shared_ptr<void> impl_ref_;

    /// Add ready events atomically.

    /// Add ready events atomically.

    {

    {

    /// Perform deferred I/O and queue completions.

    /// Perform deferred I/O and queue completions.

    void operator()() override;

    void operator()() override;

    /// Destroy without invoking.

    /// Destroy without invoking.

    /// Called during scheduler::shutdown() drain. Clear impl_ref_ to break

    /// Called during scheduler::shutdown() drain. Clear impl_ref_ to break

    /// the self-referential cycle set by close_socket().

    /// the self-referential cycle set by close_socket().

    {

    {

};

};

struct epoll_op : scheduler_op

struct epoll_op : scheduler_op

{

{

    struct canceller

    struct canceller

    {

    {

        epoll_op* op;

        epoll_op* op;

        void operator()() const noexcept;

        void operator()() const noexcept;

    };

    };

    std::coroutine_handle<> h;

    std::coroutine_handle<> h;

    capy::executor_ref ex;

    capy::executor_ref ex;

    std::error_code* ec_out = nullptr;

    std::error_code* ec_out = nullptr;

    std::size_t* bytes_out  = nullptr;

    std::size_t* bytes_out  = nullptr;

    int fd                        = -1;

    int fd                        = -1;

    int errn                      = 0;

    int errn                      = 0;

    std::size_t bytes_transferred = 0;

    std::size_t bytes_transferred = 0;

    std::atomic<bool> cancelled{false};

    std::atomic<bool> cancelled{false};

    std::optional<std::stop_callback<canceller>> stop_cb;

    std::optional<std::stop_callback<canceller>> stop_cb;

    // Prevents use-after-free when socket is closed with pending ops.

    // Prevents use-after-free when socket is closed with pending ops.

    // See "Impl Lifetime Management" in file header.

    // See "Impl Lifetime Management" in file header.

    std::shared_ptr<void> impl_ptr;

    std::shared_ptr<void> impl_ptr;

    // For stop_token cancellation - pointer to owning socket/acceptor impl.

    // For stop_token cancellation - pointer to owning socket/acceptor impl.

    // When stop is requested, we call back to the impl to perform actual I/O cancellation.

    // When stop is requested, we call back to the impl to perform actual I/O cancellation.

    epoll_socket* socket_impl_     = nullptr;

    epoll_socket* socket_impl_     = nullptr;

    epoll_acceptor* acceptor_impl_ = nullptr;

    epoll_acceptor* acceptor_impl_ = nullptr;

    {

    {

    // Defined in sockets.cpp where epoll_socket is complete

    // Defined in sockets.cpp where epoll_socket is complete

    void operator()() override;

    void operator()() override;

    {

    {

    }

    }

    virtual void cancel() noexcept = 0;

    virtual void cancel() noexcept = 0;

    {

    {

    {

    {

    {

    {

    {

    {

    {

    {

};

};

struct epoll_connect_op final : epoll_op

struct epoll_connect_op final : epoll_op

{

{

    endpoint target_endpoint;

    endpoint target_endpoint;

    {

    {

    {

    {

        // connect() completion status is retrieved via SO_ERROR, not return value

        // Guard against spurious write-ready events: a zero-timeout

        // poll confirms the fd is actually writable (connect finished).

        {

        }

    // Defined in sockets.cpp where epoll_socket is complete

    // Defined in sockets.cpp where epoll_socket is complete

    void operator()() override;

    void operator()() override;

    void cancel() noexcept override;

    void cancel() noexcept override;

};

};

struct epoll_read_op final : epoll_op

struct epoll_read_op final : epoll_op

{

{

    static constexpr std::size_t max_buffers = 16;

    static constexpr std::size_t max_buffers = 16;

    iovec iovecs[max_buffers];

    iovec iovecs[max_buffers];

    int iovec_count        = 0;

    int iovec_count        = 0;

    bool empty_buffer_read = false;

    bool empty_buffer_read = false;

    {

    {

    }

    }

    {

    {

    {

    {

        ssize_t n;

        ssize_t n;

        do

        do

        {

        {

        }

        }

        else

        else

    void cancel() noexcept override;

    void cancel() noexcept override;

};

};

struct epoll_write_op final : epoll_op

struct epoll_write_op final : epoll_op

{

{

    static constexpr std::size_t max_buffers = 16;

    static constexpr std::size_t max_buffers = 16;

    iovec iovecs[max_buffers];

    iovec iovecs[max_buffers];

    int iovec_count = 0;

    int iovec_count = 0;

    {

    {

    {

    {

        ssize_t n;

        ssize_t n;

        do

        do

        {

        {

        }

        }

        else

        else

    void cancel() noexcept override;

    void cancel() noexcept override;

};

};

struct epoll_accept_op final : epoll_op

struct epoll_accept_op final : epoll_op

{

{

    int accepted_fd                      = -1;

    int accepted_fd                      = -1;

    io_object::implementation** impl_out = nullptr;

    io_object::implementation** impl_out = nullptr;

    sockaddr_storage peer_storage{};

    sockaddr_storage peer_storage{};

    {

    {

    {

    {

        int new_fd;

        int new_fd;

        do

        do

        {

        {

                SOCK_NONBLOCK | SOCK_CLOEXEC);

                SOCK_NONBLOCK | SOCK_CLOEXEC);

        }

        }

        {

        {

        }

        }

        else

        else

        {

        {

        }

        }

    // Defined in acceptors.cpp where epoll_acceptor is complete

    // Defined in acceptors.cpp where epoll_acceptor is complete

    void operator()() override;

    void operator()() override;

    void cancel() noexcept override;

    void cancel() noexcept override;

};

};

} // namespace boost::corosio::detail

} // namespace boost::corosio::detail

#endif // BOOST_COROSIO_HAS_EPOLL

#endif // BOOST_COROSIO_HAS_EPOLL

#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_OP_HPP

#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_OP_HPP