io.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 fiber side of the reactor: the only layer that knows both fibers and the
 reactor.
 
 A fiber issues IO through a blocking looking wrapper here. The wrapper builds the
 request on the fiber's own stack, suspends, and the suspend's commit submits it
 to the reactor -- after the fiber has fully parked, so the completion can never
 race the park. Each completion carries a pointer to the fiber's on-stack
 waiter. The owning worker drains its channel (srn_reactor_consume) and, per
 completion, copies the outcome into the waiter, readies the fiber, and drops the
 inflight count. The reactor thread only posts completions and wakes the worker;
 it never touches fibers.
  */
 
#include "serene/rt/fiber/io.h"
 
#include <assert.h>
 
#include "rt/reactor/backend.h"
#include "rt/reactor/internal.h"
#include "serene/rt/context.h"
#include "serene/rt/engine.h"
#include "serene/rt/errors.h"
#include "serene/rt/fiber.h"
#include "serene/rt/reactor.h"
#include "serene/utils.h"
 
#ifdef __linux__
#  include <errno.h>
#  include <fcntl.h>
#  include <sys/socket.h>
#  include <sys/stat.h>
#  include <unistd.h>
#else
#  error "io.c needs to be patched to support this platform"
#endif
 
#define MICROSEC 1000ULL
#define MILLISEC 1000ULL * MICROSEC
#define SEC      1000ULL * MILLISEC
 
/**
 * Lives on the suspended fiber's stack; the worker fills it from the completion
 * and the fiber reads it on resume. It mirrors the completion's payload minus
 * the routing token, and is itself a fallible value (`maybe_error` null on
 * success).
 */
typedef struct io_waiter_t {
  ERROR_HEADER;
  srn_fiber_t *fiber;
  size_t result;
  uint32_t addrlen;
} io_waiter_t;
 
/**
 * Carries the submission across the suspend, on the fiber's stack.
 */
typedef struct submit_ctx_t {
  srn_reactor_t *reactor;
  srn_scheduler_t *scheduler;
  io_waiter_t *waiter;
  size_t channel;
  srn_reactor_io_request_t request;
} submit_ctx_t;
 
/**
 * Runs on the worker's loop once the fiber has switched out -- the race free
 * point to hand the request to the reactor. Returns true to stay suspended; the
 * worker readies the fiber when it consumes the completion.
 *
 * If the scheduler is winding down (`srn_sched_drain`/`srn_sched_stop`), the
 * submission is fenced, the request is never handed to the reactor, the waiter
 * is set to the CANCELED error, and the commit returns false so the worker
 * readies the fiber at once. The fiber then sees a cancelled result from its IO
 * call and unwinds, instead of parking on an op the wind-down would wait out.
 */
static bool reactor_submit_commit(srn_fiber_t *fiber, void *arg) {
  UNUSED(fiber);
  submit_ctx_t *s = (submit_ctx_t *)arg;
 
  if (!srn_sched_accepting_submissions(s->scheduler)) {
    FIBER_TRACEPOINT(fiber_io_canceled, s->channel, (int)s->request.op);
    s->waiter->maybe_error = srn_reactor_errors_static(CANCELED);
    return false; // decline the park; the worker readies us with the error
  }
 
  if (!srn_reactor_submit(s->reactor, s->channel, &s->request)) {
    // The channel is at its in-flight cap. Fail the op back to the fiber; the
    // condition is transient, so the fiber may retry once the channel drains.
    FIBER_TRACEPOINT(fiber_io_busy, s->channel, (int)s->request.op);
    s->waiter->maybe_error = srn_reactor_errors_static(CHANNEL_BUSY);
    return false;
  }
  return true; // stay parked; the worker wakes us on the completion
}
 
/**
 * Runs on the worker loop who owns the channel. Drains this worker's channel
 * and, for each completion, copies the outcome into the waiting fiber's waiter,
 * readies it, and only then drops the inflight count, so quiescence never fires
 * in the gap between a result being produced and its fiber being made runnable.
 * `srn_fiber_ready` here pushes onto the running worker's own deque, keeping the
 * woken fiber local.
 */
void srn_reactor_consume(srn_reactor_t *reactor, size_t channel) {
  // The drain batch is the configured reap batch -- "most completions a worker
  // drains from its channel in a single pass", which is exactly this loop.
  size_t batch = reactor->ctx->engine->config.reactor.reap_batch;
  srn_reactor_io_completion_t comps[batch];
  size_t n;
 
  while ((n = srn_reactor_reap(reactor, channel, comps, batch)) > 0) {
    for (size_t i = 0; i < n; i++) {
      io_waiter_t *w = (io_waiter_t *)comps[i].fiber_data;
      w->maybe_error = comps[i].maybe_error;
      w->result      = comps[i].result;
      w->addrlen     = comps[i].addrlen;
      // Emit before readying: once srn_fiber_ready runs, the woken fiber may
      // resume on another worker and unwind the stack that `w` points into, so
      // `w->fiber` is only safe to read here. `comps[i]` is a local copy and
      // stays valid regardless. Pairs with fiber_io_wait on the fiber pointer.
      FIBER_TRACEPOINT(
        fiber_io_complete, channel, comps[i].result, comps[i].maybe_error != nullptr,
        (void *)w->fiber
      );
      srn_fiber_ready(w->fiber);
      srn_reactor_op_done(reactor, channel);
    }
    FIBER_TRACEPOINT(fiber_io_drain, channel, n);
  }
}
 
/**
 * Whether `fd` must be served by a synchronous syscall on the calling worker
 * rather than the reactor. A readiness backend (e.g. epoll, kqueue) cannot wait
 * on a regular file, block device, or directory -- they are never "not ready"
 * -- so a backend that declares SRN_REACTOR_BACKEND_FEAT_FILE_OFFLOAD has those
 * fds served inline here instead. A backend without the flag (io_uring) handles
 * every fd type itself, so this is always false for it. A failed fstat (such as
 * a bad descriptor) returns false, leaving the reactor path to report the
 * error.
 */
static bool fd_needs_blocking(srn_reactor_t *reactor, int fd) {
#ifdef __linux__
  assert(reactor != nullptr);
  if ((reactor->backend->features & SRN_REACTOR_BACKEND_FEAT_FILE_OFFLOAD) == 0) {
    return false;
  }
 
  struct stat st;
  if (fstat(fd, &st) != 0) {
    return false;
  }
  return S_ISREG(st.st_mode) || S_ISBLK(st.st_mode) || S_ISDIR(st.st_mode);
#else
#  error "You need to patch fd_needs_blocking for this platform"
#endif
}
 
/**
 * Perform a READ or WRITE synchronously on the calling worker and return its outcome. This blocks
 * the worker os thread for the duration of the syscall; with a single worker it blocks the whole
 * pool. An `offset` of -1 reads or writes at the descriptor's current position. This is the one
 * place this layer maps a raw errno to an error tag, since the syscall runs here rather than in a
 * backend.
 */
static io_waiter_t blocking_io(const srn_reactor_io_request_t *req) {
#ifdef __linux__
  assert(req != nullptr);
  ssize_t n;
 
  if (req->op == SRN_REACTOR_IO_READ) {
    n = req->offset < 0 ? read(req->fd, req->buf, req->len)
                        : pread(req->fd, req->buf, req->len, (off_t)req->offset);
  } else {
    n = req->offset < 0 ? write(req->fd, req->buf, req->len)
                        : pwrite(req->fd, req->buf, req->len, (off_t)req->offset);
  }
 
  io_waiter_t w = {.fiber = nullptr, .maybe_error = nullptr, .result = 0, .addrlen = 0};
  if (n < 0) {
    srn_error_tag_t err_tag = srn_errno_to_tag(errno, UNKNOWN_IO_ERROR);
    w.maybe_error           = srn_reactor_errors_static(err_tag);
  } else {
    w.result = (size_t)n;
  }
  return w;
#else
#  error "You need to patch blocking_io for this platform"
#endif
}
 
/**
 * The general submit path, build the request on the calling fiber's stack,
 * submit it as the fiber parks, and return the waiter once the worker consumes
 * the completion and wakes it. Each public wrapper projects the returned waiter
 * into its own result type.
 */
static io_waiter_t io_wait(srn_reactor_io_request_t request) {
  srn_fiber_t *self = srn_fiber_current();
  PANIC_IF_NULL(self);
 
  srn_reactor_t *reactor = self->ctx->engine->reactor;
 
  // A READ or WRITE on an fd the backend cannot poll (a regular file on a
  // readiness backend) runs synchronously on this worker. It never reaches the
  // reactor, so it neither suspends nor touches the in-flight count -- the
  // worker is simply busy in the syscall, like a compute-bound fiber.
  if (
    (request.op == SRN_REACTOR_IO_READ || request.op == SRN_REACTOR_IO_WRITE) &&
    fd_needs_blocking(reactor, request.fd)
  ) {
    FIBER_TRACEPOINT(fiber_io_blocking, request.fd, (int)request.op);
    return blocking_io(&request);
  }
 
  srn_worker_id_t wid = srn_sched_current_worker_id();
  PANIC_IF(wid == (srn_worker_id_t)-1, "reactor IO requested outside a worker");
 
  // The fiber-side start of a reactor op, keyed on the fiber pointer. Pairs
  // with fiber_io_complete in srn_reactor_consume to measure the fiber's
  // perceived IO latency, which spans the suspend, the reactor round trip, and
  // the wake, not just the reactor thread's portion.
  FIBER_TRACEPOINT(fiber_io_wait, (int)request.op, request.fd, (void *)self);
 
  // The waiter and the submit context live on this fiber's stack; the fiber
  // stays parked (its stack preserved) until the completion comes back, so both
  // outlive the suspend.
  io_waiter_t waiter = {.fiber = self, .maybe_error = nullptr, .result = 0, .addrlen = 0};
  request.fiber_data = &waiter;
 
  submit_ctx_t s = {
    .reactor   = reactor,
    .scheduler = self->ctx->engine->scheduler,
    .waiter    = &waiter,
    .channel   = (size_t)wid,
    .request   = request,
  };
  srn_fiber_suspend(reactor_submit_commit, &s);
  return waiter;
}
 
bool srn_fiber_io_retry(srn_error_t *e) {
  if (e == nullptr || e->tag != CHANNEL_BUSY) {
    return false;
  }
  FIBER_TRACEPOINT(fiber_io_retry, (void *)srn_fiber_current());
  srn_fiber_yield();
  return true;
}
 
srn_error_t *srn_fiber_sleep_ns(uint64_t duration_ns) {
  uint64_t now = srn_now_ns();
 
  // Saturate instead of wrapping, an absurd duration means "sleep past any
  // horizon", not "wake at once".
  uint64_t deadline = duration_ns > UINT64_MAX - now ? UINT64_MAX : now + duration_ns;
 
  srn_reactor_io_request_t req = {
    .op          = SRN_REACTOR_IO_SLEEP,
    .deadline_ns = deadline,
  };
  return io_wait(req).maybe_error;
}
 
/**
 * Convert a coarser duration to nanoseconds, saturating instead of wrapping.
 * A wrapped product would turn an absurdly long sleep into a short one,
 * srn_fiber_sleep_ns saturates its deadline for the same reason.
 */
static uint64_t to_ns(uint64_t duration, uint64_t ns_per_unit) {
  return duration > UINT64_MAX / ns_per_unit ? UINT64_MAX : duration * ns_per_unit;
}
 
srn_error_t *srn_fiber_sleep_us(uint64_t duration_us) {
  return srn_fiber_sleep_ns(to_ns(duration_us, MICROSEC));
}
 
srn_error_t *srn_fiber_sleep_ms(uint64_t duration_ms) {
  return srn_fiber_sleep_ns(to_ns(duration_ms, MILLISEC));
}
 
srn_error_t *srn_fiber_sleep(uint64_t duration) { return srn_fiber_sleep_ns(to_ns(duration, SEC)); }
 
srn_io_size_result_t srn_fiber_read(int fd, void *buf, size_t len, int64_t offset) {
  srn_reactor_io_request_t req = {
    .op     = SRN_REACTOR_IO_READ,
    .fd     = fd,
    .buf    = buf,
    .len    = len,
    .offset = offset,
  };
  io_waiter_t w = io_wait(req);
  return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
}
 
srn_io_size_result_t srn_fiber_write(int fd, const void *buf, size_t len, int64_t offset) {
  srn_reactor_io_request_t req = {
    .op     = SRN_REACTOR_IO_WRITE,
    .fd     = fd,
    .buf    = (void *)buf,
    .len    = len,
    .offset = offset,
  };
  io_waiter_t w = io_wait(req);
  return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
}
 
srn_io_accept_result_t srn_fiber_accept(int fd, struct sockaddr *addr, socklen_t addr_cap) {
#ifdef __linux__
  // Force the accepted socket non-blocking and close-on-exec, so further fiber
  // IO on it never blocks the reactor thread and it does not leak across exec.
  srn_reactor_io_request_t req = {
    .op      = SRN_REACTOR_IO_ACCEPT,
    .fd      = fd,
    .addr    = addr,
    .addrlen = (uint32_t)addr_cap,
    .flags   = SOCK_NONBLOCK | SOCK_CLOEXEC,
  };
  io_waiter_t w = io_wait(req);
  return (srn_io_accept_result_t){
    .maybe_error = w.maybe_error,
    .fd          = (int)w.result,
    .addrlen     = (socklen_t)w.addrlen,
  };
#else
#  error "You need to patch srn_fiber_accept for this platform"
#endif
}
 
srn_error_t *srn_fiber_connect(int fd, const struct sockaddr *addr, socklen_t addrlen) {
  srn_reactor_io_request_t req = {
    .op      = SRN_REACTOR_IO_CONNECT,
    .fd      = fd,
    .addr    = (void *)addr,
    .addrlen = (uint32_t)addrlen,
  };
  return io_wait(req).maybe_error;
}
 
srn_io_recvfrom_result_t srn_fiber_recvfrom(
  int fd, void *buf, size_t len, int flags, struct sockaddr *addr, socklen_t addr_cap
) {
  srn_reactor_io_request_t req = {
    .op      = SRN_REACTOR_IO_RECVFROM,
    .fd      = fd,
    .buf     = buf,
    .len     = len,
    .flags   = flags,
    .addr    = addr,
    .addrlen = (uint32_t)addr_cap,
  };
  io_waiter_t w = io_wait(req);
  return (srn_io_recvfrom_result_t){
    .maybe_error = w.maybe_error,
    .count       = w.result,
    .addrlen     = (socklen_t)w.addrlen,
  };
}
 
srn_io_size_result_t srn_fiber_sendto(
  int fd, const void *buf, size_t len, int flags, const struct sockaddr *addr, socklen_t addrlen
) {
  srn_reactor_io_request_t req = {
    .op      = SRN_REACTOR_IO_SENDTO,
    .fd      = fd,
    .buf     = (void *)buf,
    .len     = len,
    .flags   = flags,
    .addr    = (void *)addr,
    .addrlen = (uint32_t)addrlen,
  };
  io_waiter_t w = io_wait(req);
  return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
}
 
srn_io_size_result_t srn_fiber_recvmsg(int fd, struct msghdr *msg, int flags) {
  srn_reactor_io_request_t req = {
    .op    = SRN_REACTOR_IO_RECVMSG,
    .fd    = fd,
    .buf   = msg,
    .flags = flags,
  };
  io_waiter_t w = io_wait(req);
  return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
}
 
srn_io_size_result_t srn_fiber_sendmsg(int fd, const struct msghdr *msg, int flags) {
  srn_reactor_io_request_t req = {
    .op    = SRN_REACTOR_IO_SENDMSG,
    .fd    = fd,
    .buf   = (void *)msg,
    .flags = flags,
  };
  io_waiter_t w = io_wait(req);
  return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
}
 
// ---------------------------------------------------------------------------
// Non-suspending helpers. Thin wrappers over the plain syscalls, they map a -1
// return to the same static IO error the reactor uses, so callers see one
// uniform convention. No reactor, no suspend, callable from any context.
// ---------------------------------------------------------------------------
 
// Turn a 0/-1 syscall return into null or error.
static srn_error_t *io_status(int rc) {
#ifdef __linux__
  return rc < 0 ? srn_reactor_errors_static(srn_errno_to_tag(errno, UNKNOWN_IO_ERROR)) : nullptr;
#else
#  error "You need to patch io_status for this platform"
#endif
}
 
// Turn an fd-or-(-1) syscall return into an fd result.
static srn_io_fd_result_t io_fd(int fd) {
#ifdef __linux__
  if (fd < 0) {
    return (srn_io_fd_result_t){
      .maybe_error = srn_reactor_errors_static(srn_errno_to_tag(errno, UNKNOWN_IO_ERROR))
    };
  }
  return (srn_io_fd_result_t){.fd = fd};
#else
#  error "You need to patch io_fd for this platform"
#endif
}
 
srn_io_fd_result_t srn_fiber_socket(int domain, int type, int protocol) {
#ifdef __linux__
  return io_fd(socket(domain, type | SOCK_NONBLOCK | SOCK_CLOEXEC, protocol));
#else
#  error "You need to patch srn_fiber_socket for this platform"
#endif
}
 
srn_error_t *srn_fiber_bind(int fd, const struct sockaddr *addr, socklen_t addrlen) {
#ifdef __linux__
  return io_status(bind(fd, addr, addrlen));
#else
#  error "You need to patch srn_fiber_bind for this platform"
#endif
}
 
srn_error_t *srn_fiber_listen(int fd, int backlog) {
#ifdef __linux__
  return io_status(listen(fd, backlog));
#else
#  error "You need to patch srn_fiber_listen for this platform"
#endif
}
 
srn_error_t *srn_fiber_close(int fd) {
#ifdef __linux__
  return io_status(close(fd));
#else
#  error "You need to patch srn_fiber_close for this platform"
#endif
}
 
srn_error_t *srn_fiber_shutdown(int fd, int how) {
#ifdef __linux__
  return io_status(shutdown(fd, how));
#else
#  error "You need to patch srn_fiber_shutdown for this platform"
#endif
}
 
srn_error_t *srn_fiber_getsockopt(int fd, int level, int optname, void *optval, socklen_t *optlen) {
#ifdef __linux__
  return io_status(getsockopt(fd, level, optname, optval, optlen));
#else
#  error "You need to patch srn_fiber_getsockopt for this platform"
#endif
}
 
srn_error_t *
srn_fiber_setsockopt(int fd, int level, int optname, const void *optval, socklen_t optlen) {
#ifdef __linux__
  return io_status(setsockopt(fd, level, optname, optval, optlen));
#else
#  error "You need to patch srn_fiber_setsockopt for this platform"
#endif
}
 
srn_io_fd_result_t srn_fiber_open(const char *path, int flags, mode_t mode) {
#ifdef __linux__
  // O_NONBLOCK keeps the reactor thread safe, a pollable descriptor (FIFO,
  // character device) opened here must never make the reactor block in a
  // transfer. Regular files ignore the flag and take the synchronous worker
  // path anyway.
  return io_fd(open(path, flags | O_NONBLOCK | O_CLOEXEC, mode));
#else
#  error "You need to patch srn_fiber_open for this platform"
#endif
}