epoll.c

Go to the documentation of this file.

/* -*- C -*-
 * Serene programming language
 * Copyright (C) 2019-2026 Sameer Rahmani <[email protected]>
 *
 * This library is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library.  If not, see <https://www.gnu.org/licenses/>.
 */
 
/** @file
## The epoll backend. Linux only.
 
It arms a descriptor, and when epoll reports it ready it performs the transfer
itself and posts the finished result. A self-wake `eventfd`, registered in the
epoll set, lets a submission on any thread break the reactor thread out of
`epoll_wait`.
 
## Notes:
- Operations (`epoll_op_t`) live in a fixed pool sized from the reactor ring
magnitude, with an intrusive free list. No need for growth as we're limited by
the magnitude of the reactor.
 
- Several operations can share one descriptor (two fibers reading and writing
the same socket, many fibers accepting on one listener). They are grouped in a
per-fd list (`epoll_op_list_t`) and the fd is registered once with the union of
their interests. `epoll_event.data.ptr` carries the list, so a readiness event
resolves to the fd's ops with no search. The wait loop then walks the list and
completes those whose direction is ready. The wake fd is distinguished by a NULL
data pointer (a real list pointer is never NULL).
 
- The fd -> list mapping is a hashmap (check the serene/rt/impl/hashmap.h). The
fd integer is stuffed straight into the key's `data` field (it fits in a
pointer), so the map needs no separate key storage; a custom control hashes and
compares `data` by value instead of dereferencing it.
*/
 
#include <assert.h>
 
#include "serene/rt/engine.h"
#include "serene/rt/errors.h"
 
#ifdef __linux__
 
#  include <errno.h>
#  include <stdint.h>
#  include <sys/epoll.h>
#  include <sys/eventfd.h>
#  include <sys/socket.h>
#  include <unistd.h>
 
#  include "../backend.h"
#  include "../internal.h"
#  include "serene/rt/impl/hashmap.h"
#  include "serene/utils.h"
 
// Since use fd value as (void *) for the map keys. We don't use them as
// pointers so we're safe really.
// NOLINTBEGIN(performance-no-int-to-ptr)
typedef struct epoll_op_t {
  /// Whether this slot holds a live operation. The discriminator between the
  /// slot's two lives, read op fields only when true, `next_free` only when
  /// false.
  bool active;
  /// While this slot is free, the index of the next free slot (SIZE_MAX ends
  /// the chain). Meaningless while active.
  size_t next_free;
 
  int fd;
  /// Next op on the same fd, or NULL. The per-fd list is threaded through here.
  struct epoll_op_t *fd_next;
 
  /// The kind of transfer. The epoll interest (EPOLLIN/EPOLLOUT) is derived
  /// from it by `op_events`, so it is not stored twice.
  srn_reactor_io_op_t op;
  /// Where the completion is posted back.
  size_t channel;
 
  void *fiber_data;
  void *buf;
  size_t len;
  int64_t offset;
 
  /// Socket address buffer, the peer to use for CONNECT/SENDTO, the peer to
  /// fill in for ACCEPT/RECVFROM, or NULL. A `struct sockaddr *` held as `void
  /// *`.
  void *addr;
  /// Length of `addr`, the exact length for CONNECT/SENDTO, the buffer capacity
  /// for ACCEPT/RECVFROM.
  uint32_t addrlen;
  /// Operation flags, the `MSG_*` flags for the datagram and message ops, the
  /// `accept4` flags for ACCEPT, and the requested readiness for POLL.
  int flags;
} epoll_op_t;
 
typedef struct epoll_op_list_t {
  /// The map key for this fd. `key.data` holds the fd value itself (see the
  /// custom control), so it doubles as the fd's stable storage. The map keeps a
  /// pointer to it, so it must live as long as the entry, it does, inside this
  /// arena-allocated list.
  hmap_key_t key;
  /// Tombstone. The map is never deleted from (that keeps allocation bounded by
  /// distinct fd integers). Instead, when the last op on an fd finishes, the fd
  /// is removed from epoll and the list is marked dormant. A later op on the
  /// same fd integer revives this same list. A dormant list (`deleted == true`)
  /// is not registered in epoll and has no ops. And since POSIX specifies that
  /// when allocating a FD, the smallest available FD should be chosen. So
  /// We will end up recycling the deleted slots quite often.
  bool deleted;
  /// Head of the ops waiting on this fd
  epoll_op_t *ops;
} epoll_op_list_t;
 
typedef struct epoll_state_t {
  /// The main epoll descriptor.
  int epfd;
  /// Self-wake eventfd, registered in `epfd` with EPOLLIN. The reactor thread
  /// sleeps in `epoll_wait`, which the kernel only breaks for IO readiness --
  /// but two non-IO events must also break it, a worker submitting a new
  /// request (so the reactor drains the SQ and arms the new fd) and shutdown
  /// (so the thread observes `running == false` and exits).
  /// `epoll_backend_wake` writes this fd from any thread, making it readable so
  /// `epoll_wait` returns at once. The wait loop recognises it, drains the
  /// counter (`drain_wake`) to silence the level-triggered fd, and treats it as
  /// a no-op event. This is the submit side nudge; without it the reactor would
  /// block until some already armed descriptor fired and never notice new work.
  int wake_fd;
  /// The pool of armed, but not yet completed operations. epoll only reports
  /// readiness, not the request, so this is the backend's memory of "what to do
  /// when each fd fires". Its size is fixed at init from
  /// `reactor.ring_magnitude` (the per-channel ring capacity) times the channel
  /// count, which is the most operations that can be in flight, so the pool
  /// never needs to grow.
  epoll_op_t *ops;
  /// Number of slots in `ops`.
  size_t total_slots;
  /// Head of the intrusive free list, the first free slot, or SIZE_MAX if the
  /// pool is full.
  size_t free_head;
  /// fd -> `epoll_op_list_t`. Grouped ops live here so a readiness event finds
  /// every op on its fd. Persistent and arena-backed; reused via tombstones, so
  /// it grows only with distinct fd integers.
  hmap_t fdmap;
} epoll_state_t;
 
// -----------------------------------------------------------------------------
// fd -> op-list map
// -----------------------------------------------------------------------------
// The key is the fd integer stored in `hmap_key_t.data` by value, not by
// pointer. The default control would dereference `data` to hash and compare it;
// this control treats `data` as the value, so no separate key storage is
// needed.
 
/**
 * Compare two fd keys. The fd lives in `data` itself, so identity of the value
 * is identity of the key.
 */
static bool fdmap_cmp(const hmap_t *hmap, const hmap_key_t *a, const hmap_key_t *b) {
  UNUSED(hmap);
  assert(a != nullptr);
  assert(b != nullptr);
  // fd 0 is a valid descriptor and stores as a null `data`, so there is no
  // non-null invariant to assert on the key payload.
  return a->data == b->data;
}
 
/**
 * Hash an fd key, the fd is its own hash. Distinct fds give distinct hashes, so
 * there are no collision nodes, and small sequential fds fan out evenly across
 * the trie's low-order 5-bit chunks. Running them through a scrambling hash
 * would only add cost and could manufacture collisions the raw value avoids.
 */
static hmap_hash_t fdmap_hash(const hmap_t *hmap, const hmap_key_t *k) {
  UNUSED(hmap);
  assert(k != nullptr);
  // fd 0 is a valid descriptor and stores as a null `data`; its hash is 0.
  return (hmap_hash_t)(intptr_t)k->data;
}
 
/**
 * Copy an fd key into the map's context. The fd lives in `data` by value, so
 * cloning the struct is enough; the default byte copier would dereference it
 * as a pointer.
 */
static hmap_key_t *fdmap_copy_key(const hmap_t *hmap, const hmap_key_t *k) {
  hmap_key_t *copy = ALLOC(hmap->map_ctx, hmap_key_t);
  *copy            = *k;
  return copy;
}
 
static hmap_control_t fdmap_control = {
  .cmp      = fdmap_cmp,
  .hash     = fdmap_hash,
  .insert   = hmap_insert_ctl,
  .lookup   = hmap_lookup_ctl,
  .copy_key = fdmap_copy_key,
};
 
/**
 * Find the op list for `fd`, creating a dormant one on first sight. A freshly
 * created list is `deleted` (not yet registered in epoll); the first op to arm
 * it revives it. The fd is stored inside the list's key, which is what the map
 * points at.
 */
static epoll_op_list_t *fd_list(srn_context_t *ctx, epoll_state_t *st, int fd) {
  assert(ctx != nullptr);
  assert(st != nullptr);
 
  hmap_key_t probe = {.data = (void *)(intptr_t)fd, .len = sizeof(int)};
 
  epoll_op_list_t *l = hmap_lookup_ctl(&fdmap_control, &st->fdmap, &probe, nullptr);
 
  if (l != nullptr) {
    return l;
  }
 
  // Create the entry as it is not there
  l          = ALLOC(ctx, epoll_op_list_t);
  l->key     = (hmap_key_t){.data = (void *)(intptr_t)fd, .len = sizeof(int)};
  l->deleted = true; // dormant until the first op arms it
  l->ops     = nullptr;
  st->fdmap  = hmap_insert_ctl(&fdmap_control, &st->fdmap, &l->key, l);
  return l;
}
 
/**
 * The epoll interest an op waits on. Read-side ops (READ, RECVFROM, RECVMSG,
 * ACCEPT) wait for EPOLLIN; write-side ops (WRITE, SENDTO, SENDMSG, CONNECT)
 * wait for EPOLLOUT. POLL carries its own requested events in `flags`.
 */
static uint32_t op_events(const epoll_op_t *op) {
  switch (op->op) {
  case SRN_REACTOR_IO_READ:
  case SRN_REACTOR_IO_RECVFROM:
  case SRN_REACTOR_IO_RECVMSG:
  case SRN_REACTOR_IO_ACCEPT:
    return EPOLLIN;
  case SRN_REACTOR_IO_WRITE:
  case SRN_REACTOR_IO_SENDTO:
  case SRN_REACTOR_IO_SENDMSG:
  case SRN_REACTOR_IO_CONNECT:
    return EPOLLOUT;
  case SRN_REACTOR_IO_POLL:
    return (uint32_t)op->flags;
  default:
    // NOP/SLEEP/CANCEL never reach a backend; anything else is a routing bug.
    PANIC("epoll backend: operation has no readiness direction");
  }
}
 
/**
 * The union of every armed op's interest on this fd, what the fd is registered
 * for in epoll.
 */
static uint32_t list_mask(const epoll_op_list_t *l) {
  uint32_t m = 0;
  for (epoll_op_t *o = l->ops; o != nullptr; o = o->fd_next) {
    m |= op_events(o);
  }
  return m;
}
 
// -----------------------------------------------------------------------------
// Operation pool (intrusive free list)
// -----------------------------------------------------------------------------
 
/**
 * Take a free slot from the pool, or null when the pool is full. The caller
 * fails the operation back to its fiber (a posted completion), so there is no
 * need to allocate or carry an error here.
 */
static epoll_op_t *alloc_op(epoll_state_t *st) {
  if (st->free_head == SIZE_MAX) {
    return nullptr; // pool exhausted
  }
 
  // Pop the head, take the first free slot, advance the head to its successor.
  size_t idx    = st->free_head;
  st->free_head = st->ops[idx].next_free;
  return &st->ops[idx];
}
 
/**
 * Return a slot to the pool, pushing it onto the front of the free list. Takes
 * the op pointer (the wait loop holds the pointer, not the index) and recovers
 * the index by its offset in the fixed `ops` array.
 */
static void free_op(epoll_state_t *st, epoll_op_t *op) {
  size_t idx    = (size_t)(op - st->ops); // pointer back to its slot index
  op->active    = false;
  op->next_free = st->free_head; // point at the old head
  st->free_head = idx;           // become the new head
}
 
// -----------------------------------------------------------------------------
// IO helpers
// -----------------------------------------------------------------------------
 
/**
 * Empty the self-wake eventfd's counter. It is level-triggered, so epoll keeps
 * reporting it readable until it is drained back to zero.
 */
static void drain_wake(epoll_state_t *st) {
  assert(st != nullptr);
 
  // A read of an eventfd returns its accumulated counter and resets it to zero.
  // The fd is EFD_NONBLOCK, so once the counter is zero the read fails with
  // EAGAIN instead of blocking, which ends the loop. So this absorbs however
  // many wakes were coalesced into the counter and leaves it drained. And since
  // we are draining the `wake_fd`, this loop is really trivial.
  uint64_t v;
  while (read(st->wake_fd, &v, sizeof(v)) == (ssize_t)sizeof(v)) {
  }
}
 
/**
 * Perform the transfer for a ready operation and fill `*out`. Returns false on
 * EAGAIN/EWOULDBLOCK (the descriptor was not actually ready, so the op stays
 * armed for the next readiness event); otherwise true with the result set.
 *
 * This function deliberately only does the transfer and records the result. It
 * does NOT free the slot, deregister the fd, or post the completion. The wait
 * loop owns that ordering (it must deregister before the completion is visible,
 * so a worker that reaps it can safely close the fd).
 *
 * `ready` is the readiness mask epoll reported for the fd; only POLL uses it
 * (to report which events fired), the transfers ignore it.
 */
static bool perform_op(epoll_op_t *op, uint32_t ready, srn_reactor_io_completion_t *out) {
  assert(op != nullptr);
 
  // Success by default; the error paths set `maybe_error`. `result` and
  // `addrlen` are meaningful only on success, but default them so a failed op
  // never leaves them indeterminate. Only ACCEPT and RECVFROM fill `addrlen`.
  out->maybe_error = nullptr;
  out->result      = 0;
  out->addrlen     = 0;
 
  ssize_t n;
 
  switch (op->op) {
  case SRN_REACTOR_IO_READ:
    n = (op->offset < 0) ? read(op->fd, op->buf, op->len)
                         : pread(op->fd, op->buf, op->len, (off_t)op->offset);
    break;
  case SRN_REACTOR_IO_WRITE:
    n = (op->offset < 0) ? write(op->fd, op->buf, op->len)
                         : pwrite(op->fd, op->buf, op->len, (off_t)op->offset);
    break;
  case SRN_REACTOR_IO_ACCEPT: {
    // The result is the accepted descriptor, carried back in `result` like a
    // byte count. `addrlen` is value-result for accept4: pass the buffer
    // capacity in, report the actual peer length back (valid when n >= 0).
    socklen_t al = (socklen_t)op->addrlen;
    n            = accept4(
      op->fd, (struct sockaddr *)op->addr, (op->addr != nullptr) ? &al : nullptr, op->flags
    );
    out->addrlen = (uint32_t)al;
    break;
  }
  case SRN_REACTOR_IO_CONNECT: {
    // Readiness means the non-blocking connect finished; SO_ERROR carries the
    // verdict (0, or an errno). There is no transfer and no EAGAIN here.
    int soerr        = 0;
    socklen_t errlen = sizeof(soerr);
    if (getsockopt(op->fd, SOL_SOCKET, SO_ERROR, &soerr, &errlen) < 0) {
      soerr = errno;
    }
    out->fiber_data = op->fiber_data;
    if (soerr != 0) {
      out->maybe_error = srn_reactor_errors_static(srn_errno_to_tag(soerr, UNKNOWN_IO_ERROR));
    } else {
      out->result = 0;
    }
    return true;
  }
  case SRN_REACTOR_IO_RECVFROM: {
    // `addrlen` is value-result here too, report the sender address length
    // back.
    socklen_t al = (socklen_t)op->addrlen;
    n            = recvfrom(
      op->fd, op->buf, op->len, op->flags, (struct sockaddr *)op->addr,
      (op->addr != nullptr) ? &al : nullptr
    );
    out->addrlen = (uint32_t)al;
    break;
  }
  case SRN_REACTOR_IO_SENDTO:
    n = sendto(
      op->fd, op->buf, op->len, op->flags, (struct sockaddr *)op->addr, (socklen_t)op->addrlen
    );
    break;
  case SRN_REACTOR_IO_RECVMSG:
    n = recvmsg(op->fd, (struct msghdr *)op->buf, op->flags);
    break;
  case SRN_REACTOR_IO_SENDMSG:
    n = sendmsg(op->fd, (struct msghdr *)op->buf, op->flags);
    break;
  case SRN_REACTOR_IO_POLL:
    // No transfer, report which events are ready.
    out->fiber_data = op->fiber_data;
    out->result     = (size_t)ready;
    return true;
  default:
    // NOP/SLEEP/CANCEL are handled by the reactor core and never reach a
    // backend. Anything else is a routing bug, not a runtime condition.
    PANIC("epoll backend: unexpected operation kind");
  }
 
  if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
    return false; // spurious readiness; leave it armed
  }
 
  out->fiber_data = op->fiber_data;
  if (n < 0) {
    out->maybe_error = srn_reactor_errors_static(srn_errno_to_tag(errno, UNKNOWN_IO_ERROR));
  } else {
    out->result = (size_t)n; // bytes or accepted fd
  }
  return true;
}
 
// -----------------------------------------------------------------------------
// Lifecycle
// -----------------------------------------------------------------------------
 
/**
 * Bring the epoll backend up, build the operation pool and the fd map, open the
 * epoll instance, and register the self-wake eventfd.
 */
static srn_error_t *epoll_backend_init(srn_reactor_t *reactor) {
  PANIC_IF_NULL(reactor);
 
  srn_context_t *ctx = reactor->ctx;
 
  epoll_state_t *st = ALLOC(ctx, epoll_state_t);
  size_t cap        = (size_t)1 << ctx->engine->config.reactor.ring_magnitude;
  st->total_slots   = reactor->nchannels * cap;
  st->ops           = ALLOCN(ctx, epoll_op_t, st->total_slots);
  st->fdmap         = hmap_empty(ctx);
 
  // Chain every slot into the free list: 0 -> 1 -> ... -> last -> end.
  for (size_t i = 0; i < st->total_slots; i++) {
    st->ops[i].active    = false;
    st->ops[i].next_free = i + 1;
  }
  st->ops[st->total_slots - 1].next_free = SIZE_MAX;
  st->free_head                          = 0;
 
  st->epfd    = epoll_create1(EPOLL_CLOEXEC);
  st->wake_fd = -1;
 
  if (st->epfd == -1) {
    return ERR(ctx, EPOLL_INIT_FAILED, "Couldn't create the epoll file descriptor.");
  }
 
  st->wake_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
  if (st->wake_fd == -1) {
    close(st->epfd);
    return ERR(
      ctx, EPOLL_EVENT_FD_FAILED, "Couldn't create the event file descriptor of the reactor."
    );
  }
 
  // Register the wake fd with a NULL data pointer. Armed fds carry a non-NULL
  // pointer to their op list (below), so NULL uniquely marks "this is the wake
  // fd".
  struct epoll_event ee = {.events = EPOLLIN, .data.ptr = nullptr};
  if (epoll_ctl(st->epfd, EPOLL_CTL_ADD, st->wake_fd, &ee) == -1) {
    close(st->wake_fd);
    close(st->epfd);
    return ERR(ctx, EPOLL_FAILED_TO_ADD_FD, "Failed to add the wake_fd to the epfd");
  }
 
  reactor->backend_state = st;
  return nullptr;
}
 
/**
 * Tear the backend down, close the descriptors and detach the state. The pool,
 * the lists and the map are arena-allocated (from the reactor context), so they
 * are reclaimed in bulk with the context; there is nothing to free here.
 */
static void epoll_backend_shutdown(srn_reactor_t *reactor) {
  PANIC_IF_NULL(reactor);
 
  epoll_state_t *st = (epoll_state_t *)reactor->backend_state;
  if (st == nullptr) {
    return;
  }
 
  close(st->wake_fd);
  close(st->epfd);
  reactor->backend_state = nullptr;
}
 
// -----------------------------------------------------------------------------
// Submit / wait / wake
// -----------------------------------------------------------------------------
 
/**
 * Arm a descriptor operation. The op joins its fd's list, and the fd is
 * registered (or re-registered) for the union of its ops' interests. On a full
 * pool, or if epoll refuses the fd, the operation is completed with its
 * `maybe_error` set rather than aborting, so the waiting fiber wakes and can
 * react. Called by the reactor core on the reactor thread while draining
 * submissions.
 */
static void
epoll_backend_submit(srn_reactor_t *reactor, size_t channel, const srn_reactor_io_request_t *req) {
  PANIC_IF_NULL(reactor);
  PANIC_IF_NULL(reactor->backend_state);
 
  // A POLL that asks for no pollable event would register an empty interest
  // mask, nothing could ever match it, and the fiber would park forever.
  // Fail it back as malformed rather than arming it.
  if (req->op == SRN_REACTOR_IO_POLL && (req->flags & (EPOLLIN | EPOLLOUT)) == 0) {
    srn_reactor_post_completion(
      reactor, channel,
      (srn_reactor_io_completion_t){
        .fiber_data = req->fiber_data, .maybe_error = srn_reactor_errors_static(INVALID_ARGUMENT)
      }
    );
    return;
  }
 
  srn_context_t *ctx = reactor->ctx;
  epoll_state_t *st  = reactor->backend_state;
  epoll_op_t *op     = alloc_op(st);
 
  if (op == nullptr) {
    // Pool full, fail this op back to its fiber. -ENOBUFS ("no buffer space
    // available") is the closest errno for an exhausted op pool; it is
    // transient, so the fiber can retry once in-flight operations drain.
    srn_error_t *err = srn_reactor_errors_static(srn_errno_to_tag(ENOBUFS, UNKNOWN_IO_ERROR));
    srn_reactor_post_completion(
      reactor, channel,
      (srn_reactor_io_completion_t){.fiber_data = req->fiber_data, .maybe_error = err}
    );
    return;
  }
 
  op->active     = true;
  op->fd         = req->fd;
  op->op         = req->op;
  op->channel    = channel;
  op->fiber_data = req->fiber_data;
  op->buf        = req->buf;
  op->len        = req->len;
  op->offset     = req->offset;
  op->addr       = req->addr;
  op->addrlen    = req->addrlen;
  op->flags      = req->flags;
 
  // CONNECT issues the (non-blocking) connect now; the readiness later reports
  // its verdict via SO_ERROR. An immediate success, or a hard failure that is
  // not EINPROGRESS, completes here without arming anything.
  if (op->op == SRN_REACTOR_IO_CONNECT) {
    int rc = connect(op->fd, (struct sockaddr *)op->addr, (socklen_t)op->addrlen);
 
    if (rc == 0 || errno != EINPROGRESS) {
      srn_error_t *err =
        (rc == 0) ? nullptr : srn_reactor_errors_static(srn_errno_to_tag(errno, UNKNOWN_IO_ERROR));
      free_op(st, op);
      srn_reactor_post_completion(
        reactor, channel,
        (srn_reactor_io_completion_t){.fiber_data = req->fiber_data, .maybe_error = err}
      );
 
      return;
    }
    // EINPROGRESS: fall through and arm EPOLLOUT like any other op.
  }
 
  // Join the fd's list. A dormant (deleted) list is not yet in epoll, so this
  // is the first op on the fd and needs an ADD; otherwise the fd is already
  // registered and its interest mask only needs widening with a MOD.
  epoll_op_list_t *l = fd_list(ctx, st, req->fd);
  bool need_add      = l->deleted;
  l->deleted         = false;
  op->fd_next        = l->ops;
  l->ops             = op;
 
  // The list pointer rides in epoll's data, so a readiness event lands on every
  // op for the fd with no search of the pool.
  struct epoll_event ee = {.events = list_mask(l), .data.ptr = l};
  int rc = epoll_ctl(st->epfd, (int)need_add ? EPOLL_CTL_ADD : EPOLL_CTL_MOD, req->fd, &ee);
 
  if (rc == -1) {
    // Could not arm it (bad fd, ...). Pop the op back off the list, return the
    // slot and fail the op back; otherwise it would sit armed-in-software but
    // never fire and the fiber would hang. If that emptied the list, mark it
    // dormant again so the next op on this fd re-adds rather than mods.
    int err = errno;
    l->ops  = op->fd_next;
    free_op(st, op);
    if (l->ops == nullptr) {
      l->deleted = true;
    }
 
    srn_error_t *err1 = srn_reactor_errors_static(srn_errno_to_tag(err, UNKNOWN_IO_ERROR));
    srn_reactor_post_completion(
      reactor, channel,
      (srn_reactor_io_completion_t){.fiber_data = req->fiber_data, .maybe_error = err1}
    );
  }
}
 
/**
 * Wait for readiness up to `timeout_ms` (epoll's own unit for timout: -1 blocks
 * until an event, 0 polls), then perform and complete every ready operation.
 * Each event carries its fd's op list, so it resolves with no scan of the pool;
 * the loop walks the list and completes the ops whose direction is ready.
 * Called by the reactor thread's loop each pass.
 */
static void epoll_backend_wait(srn_reactor_t *reactor, int timeout_ms) {
  PANIC_IF_NULL(reactor);
  PANIC_IF_NULL(reactor->backend_state);
 
  epoll_state_t *st = reactor->backend_state;
  size_t batch      = reactor->ctx->engine->config.reactor.reap_batch;
 
  struct epoll_event events[batch];
  int n = epoll_wait(st->epfd, events, (int)batch, timeout_ms);
 
  if (n < 0) {
    // The only error expected here is EINTR: a signal interrupted the wait, and
    // the reactor loop simply calls wait again. Any other errno means epfd or
    // the arguments are invalid -- a setup bug, not a runtime condition -- and
    // silently returning would turn the reactor loop into a hot spin, so it
    // panics in every build.
    PANIC_IF(errno != EINTR, "reactor: epoll_wait failed");
    return;
  }
 
  if (n == 0) {
    return; // timed out; nothing was ready
  }
 
  for (int i = 0; i < n; i++) {
    // The wake fd is the one event registered with a NULL data pointer. It
    // carries no operation, someone (a worker that just submitted, or shutdown)
    // wrote the eventfd purely to force this epoll_wait to return so the
    // reactor thread re-checks its queues. Drain it and move on.
    if (events[i].data.ptr == nullptr) {
      drain_wake(st);
      continue;
    }
 
    epoll_op_list_t *l = (epoll_op_list_t *)events[i].data.ptr;
 
    // An error or hangup on the fd wakes both directions, a reader and a writer
    // on the same fd should both observe it.
    uint32_t ready = events[i].events;
    if (ready & (EPOLLERR | EPOLLHUP)) {
      ready |= (EPOLLIN | EPOLLOUT);
    }
 
    uint32_t before = list_mask(l);
 
    // Walk the fd's ops, completing those the readiness covers. `link` is the
    // address of the pointer that reaches the current op, so unlinking is a
    // single store and the walk continues from the same spot.
    epoll_op_t **link = &l->ops;
    while (*link != nullptr) {
      epoll_op_t *op = *link;
 
      if ((op_events(op) & ready) == 0) {
        link = &op->fd_next; // not this op's direction
        continue;
      }
 
      // === PERFORM THE ACTUAL OPERATION ===
      srn_reactor_io_completion_t c;
      if (!perform_op(op, ready, &c)) {
        link = &op->fd_next; // EAGAIN: still armed, will fire again
        continue;
      }
 
      // Take this op off the list. If it was the last op on the fd, drop the fd
      // from epoll and mark the list dormant -- and do both before posting the
      // completion. Once the completion is posted, a worker may reap it and
      // close the fd, so the fd has to be out of epoll by that point.
      //
      // The unlink already set `*link` to the next op, so the loop carries on
      // without moving the cursor forward.
      size_t ch = op->channel;
      *link     = op->fd_next;
      if (l->ops == nullptr) {
        epoll_ctl(st->epfd, EPOLL_CTL_DEL, op->fd, nullptr);
        l->deleted = true;
      }
      free_op(st, op);
      srn_reactor_post_completion(reactor, ch, c);
    }
 
    // If the fd is still armed but some ops finished, narrow its interest to
    // what remains. Level-triggered epoll would otherwise report a stale
    // EPOLLOUT every wait and spin.
    if (!l->deleted) {
      uint32_t now = list_mask(l);
      if (now != before) {
        int fd                = (int)(intptr_t)l->key.data;
        struct epoll_event ee = {.events = now, .data.ptr = l};
        epoll_ctl(st->epfd, EPOLL_CTL_MOD, fd, &ee);
      }
    }
  }
}
 
/**
 * Break a concurrent epoll_wait out of its sleep from any thread, by making the
 * self-wake eventfd readable. The reactor uses this as the submit-side nudge
 * and during shutdown to re-check `running`. Called from srn_reactor_submit
 * (after queuing a request) and srn_reactor_shutdown (to break the wait).
 */
static void epoll_backend_wake(srn_reactor_t *reactor) {
  PANIC_IF_NULL(reactor);
 
  epoll_state_t *st = (epoll_state_t *)reactor->backend_state;
 
  uint64_t one = 1;
  ssize_t rc   = write(st->wake_fd, &one, sizeof(one));
  // A full counter (EAGAIN) is fine, it is already readable, so the reactor
  // will still wake. Any other error is ignored for the same reason.
  UNUSED(rc);
}
 
// -----------------------------------------------------------------------------
// Vtable
// -----------------------------------------------------------------------------
const srn_reactor_backend_t srn_reactor_backend_epoll = {
  .name     = "epoll",
  .init     = epoll_backend_init,
  .shutdown = epoll_backend_shutdown,
  .submit   = epoll_backend_submit,
  .wait     = epoll_backend_wait,
  .wake     = epoll_backend_wake,
  .features = SRN_REACTOR_BACKEND_FEAT_FILE_OFFLOAD,
};
 
#endif
// NOLINTEND(performance-no-int-to-ptr)