reactor.c

Go to the documentation of this file.

/* -*- C -*-
 * Serene programming language
 * Copyright (C) 2019-2026 Sameer Rahmani <lxsameer@lxsameer.com>
 *
 * 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/>.
 */
#include "serene/rt/reactor.h"
 
#include <assert.h>
#include <limits.h>
 
#include "backend.h"
#include "internal.h"
#include "serene/rt/engine.h"
#include "serene/rt/impl/spsc_ring.h"
#include "serene/rt/mm/interface.h"
#include "serene/utils.h"
 
// A magnitude of 8 corresponds to 256 slot on the ring
// -----------------------------------------------------------------------------
// Lifecycle
// -----------------------------------------------------------------------------
srn_reactor_t *srn_reactor_init(srn_engine_t *engine) {
  PANIC_IF_NULL(engine);
 
  srn_reactor_t *reactor =
    srn_mm_immortal_allocate_aligned(engine->mm, sizeof(srn_reactor_t), alignof(srn_reactor_t));
  PANIC_IF_NULL(reactor);
 
  // This will allocate a root context which will have its own block of memory
  reactor->ctx = srn_context_make(engine);
  PANIC_IF_NULL(reactor->ctx);
 
  switch (reactor->ctx->engine->config.reactor.backend) {
#ifdef __linux__
  case SRN_REACTOR_EPOLL:
    reactor->backend = &srn_reactor_backend_epoll;
    break;
#endif
  default:
    PANIC("IO backend is not implemented yet");
  }
 
  reactor->backend_state = nullptr;
  reactor->channels      = nullptr;
  reactor->nchannels     = 0;
  reactor->notify        = nullptr;
 
  atomic_init(&reactor->running, false);
  atomic_init(&reactor->inflight, 0);
 
  return reactor;
}
 
void srn_reactor_shutdown(srn_reactor_t *reactor) {
  // Never activated, or already shut down. Nothing to release.
  RETURN_IF_NULL(reactor)
  if (!atomic_load_explicit(&reactor->running, memory_order_acquire)) {
    return;
  }
 
  atomic_store_explicit(&reactor->running, false, memory_order_release);
  // Break the reactor out of its wait so it observes
  // `running == false` and exits, rather than blocking until some descriptor
  // happens to fire.
  reactor->backend->wake(reactor);
  srn_thread_join(&reactor->thread);
  reactor->backend->shutdown(reactor);
 
  // The reactor thread is joined, so the timer heap is ours now. Free any
  // timers that never fired (their fibers are reaped by the scheduler), then
  // the heap.
  void *out = nullptr;
  while (pqmh_pop(reactor->ctx, &reactor->timers, &out)) {
    srn_mm_free(reactor->ctx->engine->mm, out);
  }
  pqmh_free(reactor->ctx, &reactor->timers);
}
 
// -----------------------------------------------------------------------------
// Static IO errors
// -----------------------------------------------------------------------------
// An IO failure is fully described by its tag (a fixed tag and message), so it
// need not be allocated per occurrence. These shared, immutable instances let a
// completion or result point one at `maybe_error` with no allocation and no
// lifetime concern; they are read only and safe to share across threads. The
// last entry is the fallback for any tag without a dedicated one.
static srn_error_t io_errors[] = {
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = CONNECTION_REFUSED,
   .msg   = (char *)"connection refused",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = CONNECTION_RESET,
   .msg   = (char *)"connection reset by peer",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = CONNECTION_ABORTED,
   .msg   = (char *)"connection aborted",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = BROKEN_PIPE,
   .msg   = (char *)"broken pipe",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = BAD_DESCRIPTOR,
   .msg   = (char *)"bad file descriptor",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = TIMED_OUT,
   .msg   = (char *)"operation timed out",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = CANCELED,
   .msg   = (char *)"operation canceled",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = NOT_CONNECTED,
   .msg   = (char *)"socket is not connected",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = ADDRESS_IN_USE,
   .msg   = (char *)"address already in use",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = HOST_UNREACHABLE,
   .msg   = (char *)"host is unreachable",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = NETWORK_UNREACHABLE,
   .msg   = (char *)"network is unreachable",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = MESSAGE_TOO_LONG,
   .msg   = (char *)"message too long",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = TOO_MANY_OPEN_FILES,
   .msg   = (char *)"too many open files",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = PERMISSION_DENIED,
   .msg   = (char *)"permission denied",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = OUT_OF_MEMORY,
   .msg   = (char *)"out of memory",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = NOT_FOUND,
   .msg   = (char *)"no such file or directory",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = ALREADY_EXISTS,
   .msg   = (char *)"file already exists",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = IS_A_DIRECTORY,
   .msg   = (char *)"is a directory",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = NOT_A_DIRECTORY,
   .msg   = (char *)"not a directory",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = NO_SPACE,
   .msg   = (char *)"no space left on device",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = INVALID_ARGUMENT,
   .msg   = (char *)"invalid argument",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = CHANNEL_BUSY,
   .msg   = (char *)"reactor channel is at capacity",
   .as    = {.frames = nullptr}},
  {.layer = SRN_ERR_RUNTIME_LAYER,
   .tag   = UNKNOWN_IO_ERROR,
   .msg   = (char *)"unknown IO error",
   .as    = {.frames = nullptr}},
};
 
srn_error_t *srn_reactor_errors_static(srn_error_tag_t tag) {
 
  size_t count = sizeof(io_errors) / sizeof(io_errors[0]);
 
  for (size_t i = 0; i < count; i++) {
    if (io_errors[i].tag == tag) {
      return &io_errors[i];
    }
  }
  // the UNKNOWN_IO_ERROR fallback
  return &io_errors[count - 1];
}
 
// -----------------------------------------------------------------------------
// Completions
// -----------------------------------------------------------------------------
void srn_reactor_post_completion(
  srn_reactor_t *reactor, size_t channel, srn_reactor_io_completion_t completion
) {
  srn_reactor_io_channel_t *ch = &reactor->channels[channel];
  // Admission control bounds a channel's in-flight ops at the ring capacity,
  // and the op being posted is one of them and not yet in the CQ, so the CQ
  // always has a slot. A full CQ here means the accounting is broken; a spin
  // in its place would stall the whole reactor behind one busy worker.
  PANIC_IF(
    !srn_spsc_ring_push(ch->cq, &completion),
    "reactor: completion queue full despite admission control; this is a bug"
  );
 
  REACTOR_TRACEPOINT(
    reactor_complete, channel, completion.result, completion.maybe_error != nullptr,
    completion.fiber_data
  );
 
  // The in-flight count is dropped by the consumer through srn_reactor_op_done,
  // not here. A fiber's waiter must be readied before the op counts as gone, or
  // quiescence could fire in the gap between posting and readying.
  if (reactor->notify != nullptr) {
    reactor->notify(reactor->ctx->engine->scheduler, channel);
  }
}
 
void srn_reactor_op_done(srn_reactor_t *reactor, size_t channel) {
  PANIC_IF_NULL(reactor);
  PANIC_IF(channel >= reactor->nchannels, "reactor: channel id out of range");
  reactor->channels[channel].inflight--;
  atomic_fetch_sub(&reactor->inflight, 1);
  REACTOR_TRACEPOINT(reactor_op_done, channel);
}
 
bool srn_reactor_idle(srn_reactor_t *reactor) {
  PANIC_IF_NULL(reactor);
  return atomic_load(&reactor->inflight) == 0;
}
 
bool srn_reactor_submit(
  srn_reactor_t *reactor, size_t channel, const srn_reactor_io_request_t *request
) {
  PANIC_IF_NULL(reactor);
  PANIC_IF_NULL(request);
  PANIC_IF(channel >= reactor->nchannels, "reactor: channel id out of range");
 
  // A clean panic beats what follows otherwise, the backend state is gone
  // after shutdown, so the wake below would dereference null.
  PANIC_IF(
    !atomic_load_explicit(&reactor->running, memory_order_acquire),
    "reactor: submit on a reactor that is not running"
  );
 
  srn_reactor_io_channel_t *ch = &reactor->channels[channel];
 
  // Admission control. A channel holds at most one ring's worth of ops in
  // flight, counted from submit until the completion is consumed. The cap is
  // what makes ring overflow impossible, the SQ holds at most the admitted
  // ops, and the CQ holds at most the admitted ops minus the one being
  // posted, so neither the push below nor the post on the reactor thread can
  // find its ring full. A rejected submit is failed back to the caller as a
  // transient condition; it clears as the worker consumes completions.
  if (ch->inflight >= reactor->ring_capacity) {
    REACTOR_TRACEPOINT(reactor_submit_rejected, channel, ch->inflight);
    return false;
  }
  ch->inflight++;
 
  // Count the op as in flight BEFORE the request becomes visible to the reactor
  // thread. If the bump came after the push, the reactor could complete the op
  // and the consumer could call srn_reactor_op_done before the bump ran --
  // underflowing the counter and breaking quiescence.
  atomic_fetch_add(&reactor->inflight, 1);
 
  // The SQ is single-producer, the worker that owns this channel.
  PANIC_IF(
    !srn_spsc_ring_push(ch->sq, request),
    "reactor: submission queue full despite admission control; this is a bug"
  );
 
  REACTOR_TRACEPOINT(reactor_submit, channel, (int)request->op, request->fd, request->fiber_data);
 
  // Rouse the reactor so it drains the new request at once instead of waiting
  // out its idle timeout.
  reactor->backend->wake(reactor);
  return true;
}
 
size_t srn_reactor_reap(
  srn_reactor_t *reactor, size_t channel, srn_reactor_io_completion_t *out, size_t max
) {
  PANIC_IF_NULL(reactor);
  PANIC_IF_NULL(out);
  PANIC_IF(channel >= reactor->nchannels, "reactor: channel id out of range");
 
  srn_reactor_io_channel_t *ch = &reactor->channels[channel];
  size_t n                     = 0;
 
  while (n < max && (int)srn_spsc_ring_pop(ch->cq, &out[n])) {
    n++;
  }
 
  return n;
}
 
bool srn_reactor_channel_has_completions(srn_reactor_t *reactor, size_t channel) {
  PANIC_IF_NULL(reactor);
  PANIC_IF(channel >= reactor->nchannels, "reactor: channel id out of range");
  return (!srn_spsc_ring_empty(reactor->channels[channel].cq)) != 0;
}
 
// -----------------------------------------------------------------------------
// The reactor thread
// -----------------------------------------------------------------------------
// Timers
// -----------------------------------------------------------------------------
// A min-heap of pending timers keyed by absolute monotonic deadline. The heap
// stores pointers, so each entry is its own srn_mm_malloc allocation, owned by
// the reactor and freed when the timer fires (or at shutdown). Touched only on
// the reactor thread, so no locking.
typedef struct srn_timer_t {
  /// Absolute CLOCK_MONOTONIC deadline
  uint64_t deadline_ns;
  /// Channel the completion is posted to
  size_t channel;
  /// Round tripped to the waiting fiber
  void *fiber_data;
} srn_timer_t;
 
/**
 * Order timers by deadline so the soonest sits at the heap root.
 */
static bool timer_less(srn_context_t *ctx, const void *a, const void *b) {
  UNUSED(ctx);
  return ((const srn_timer_t *)a)->deadline_ns < ((const srn_timer_t *)b)->deadline_ns;
}
 
static const pqmh_control_t timer_ctl = {.less = timer_less};
 
/**
 * Register a SLEEP: allocate a timer entry and push it onto the heap.
 */
static void timer_add(srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req) {
  assert(r != nullptr);
 
  srn_timer_t *t = srn_mm_malloc(r->ctx->engine->mm, sizeof(srn_timer_t));
  PANIC_IF_NULL(t);
 
  t->deadline_ns = req->deadline_ns;
  t->channel     = channel;
  t->fiber_data  = req->fiber_data;
  REACTOR_TRACEPOINT(reactor_timer_armed, channel, req->deadline_ns);
  pqmh_push(r->ctx, &r->timers, t);
}
 
/**
 * Milliseconds until the soonest deadline (rounded up), clamped to a valid
 * `epoll_wait` timeout, 0 if a timer is already due, INT_MAX for a far-off
 * deadline, and -1 (block indefinitely) when no timers are pending.
 */
static int timer_next_timeout(srn_reactor_t *r) {
  assert(r != nullptr);
 
  const srn_timer_t *t = pqmh_peek(&r->timers);
 
  if (t == nullptr) {
    return -1;
  }
 
  uint64_t now = srn_now_ns();
  if (t->deadline_ns <= now) {
    return 0;
  }
 
  // Convert the ns remaining to the ms epoll_wait timeout, rounding UP (add
  // NS_PER_MS-1) so a sub-millisecond remainder waits 1ms rather than 0 -- we
  // must never wake early and busy-spin firing nothing.
  uint64_t ms = ((t->deadline_ns - now) + (SRN_NS_PER_MS - 1)) / SRN_NS_PER_MS;
  return (ms > (uint64_t)INT_MAX) ? INT_MAX : (int)ms;
}
 
/**
 * Fire every timer whose deadline has passed, pop it (in deadline order), post
 * a zero result completion to its channel, and free the entry.
 */
static void timer_fire_expired(srn_reactor_t *r) {
  assert(r != nullptr);
 
  uint64_t now = srn_now_ns();
  const srn_timer_t *top;
 
  while ((top = pqmh_peek(&r->timers)) != nullptr && top->deadline_ns <= now) {
    void *out = nullptr;
    UNUSED(pqmh_pop(r->ctx, &r->timers, &out));
 
    srn_timer_t *t = (srn_timer_t *)out;
    REACTOR_TRACEPOINT(reactor_timer_fired, t->channel, t->deadline_ns);
 
    srn_reactor_post_completion(
      r, t->channel, (srn_reactor_io_completion_t){.fiber_data = t->fiber_data, .result = 0}
    );
    srn_mm_free(r->ctx->engine->mm, t);
  }
}
 
// -----------------------------------------------------------------------------
static void process_request(srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req) {
  switch (req->op) {
  case SRN_REACTOR_IO_NOP:
    srn_reactor_post_completion(
      r, channel, (srn_reactor_io_completion_t){.fiber_data = req->fiber_data, .result = 0}
    );
    break;
  case SRN_REACTOR_IO_READ:
  case SRN_REACTOR_IO_WRITE:
  case SRN_REACTOR_IO_ACCEPT:
  case SRN_REACTOR_IO_CONNECT:
  case SRN_REACTOR_IO_RECVFROM:
  case SRN_REACTOR_IO_SENDTO:
  case SRN_REACTOR_IO_RECVMSG:
  case SRN_REACTOR_IO_SENDMSG:
  case SRN_REACTOR_IO_POLL:
    // Descriptor ops go to the backend, it arms the fd and posts the completion
    // when it is ready. The backend reports its own failures through a
    // completion, so there is nothing to check here.
    r->backend->submit(r, channel, req);
    break;
 
  case SRN_REACTOR_IO_SLEEP:
    timer_add(r, channel, req);
    break;
 
  case SRN_REACTOR_IO_CANCEL:
    PANIC("reactor: CANCEL is not implemented yet");
  }
}
 
static void drain_submissions(srn_reactor_t *r) {
  srn_reactor_io_request_t req;
  // TODO(lxsameer): Should we make this configurable via a policy or something?
  // For example, pop from each channel once vs pop all the SQEs per channel
  for (size_t c = 0; c < r->nchannels; c++) {
    while (srn_spsc_ring_pop(r->channels[c].sq, &req)) {
      process_request(r, c, &req);
    }
  }
}
 
static void reactor_thread_main(void *arg) {
  srn_reactor_t *r = (srn_reactor_t *)arg;
  while (atomic_load_explicit(&r->running, memory_order_acquire)) {
    drain_submissions(r);
    // Wait until the soonest timer deadline (or indefinitely if none), unless a
    // kernel event, a submission nudge, or shutdown breaks us out sooner. Then
    // fire whatever timers came due.
    r->backend->wait(r, timer_next_timeout(r));
    timer_fire_expired(r);
  }
}
 
void srn_reactor_activate(srn_reactor_t *reactor, size_t nchannels, srn_reactor_notify_fn notify) {
 
  PANIC_IF_NULL(reactor);
  PANIC_IF(nchannels == 0, "reactor: at least one channel is required");
 
  reactor->notify    = notify;
  reactor->nchannels = nchannels;
 
  // The ring size is a runtime knob; both rings of every channel use it.
  // Validate the knobs once here, so everything sized from them downstream
  // (rings, the backend op pool, the reap stack buffers) is sound, a zero
  // batch is a zero length VLA, and an oversized magnitude wraps the pool
  // arithmetic.
  const size_t ring_mag = reactor->ctx->engine->config.reactor.ring_magnitude;
  PANIC_IF(ring_mag < 1 || ring_mag > 16, "reactor: ring_magnitude must be within 1..16");
  reactor->ring_capacity = (size_t)1 << ring_mag;
 
  const size_t batch = reactor->ctx->engine->config.reactor.reap_batch;
  PANIC_IF(
    batch < 1 || batch > reactor->ring_capacity,
    "reactor: reap_batch must be within 1..ring capacity"
  );
 
  reactor->channels = ALLOCN(reactor->ctx, srn_reactor_io_channel_t, nchannels);
  PANIC_IF_NULL(reactor->channels);
 
  for (size_t c = 0; c < nchannels; c++) {
    reactor->channels[c].inflight = 0;
    // Setup the SQ of the channel
    reactor->channels[c].sq =
      srn_spsc_ring_make(reactor->ctx, ring_mag, sizeof(srn_reactor_io_request_t));
    PANIC_IF_NULL(reactor->channels[c].sq);
    PANIC_IF(HAS_ERROR(*reactor->channels[c].sq), reactor->channels[c].sq->maybe_error->msg);
 
    // Setup the CQ of the channel
    reactor->channels[c].cq =
      srn_spsc_ring_make(reactor->ctx, ring_mag, sizeof(srn_reactor_io_completion_t));
    PANIC_IF_NULL(reactor->channels[c].cq);
    PANIC_IF(HAS_ERROR(*reactor->channels[c].cq), reactor->channels[c].cq->maybe_error->msg);
  }
 
  // The timer heap is reactor thread private. Create it before the thread
  // spawns.
  reactor->timers = pqmh_make(reactor->ctx, &timer_ctl, 0);
 
  // Bring the backend up now that nchannels is known, its op pool is sized from
  // the channel count, so it cannot be initialized before activation.
  srn_error_t *err = reactor->backend->init(reactor);
  PANIC_IF(err != nullptr, err->msg);
 
  atomic_store_explicit(&reactor->running, true, memory_order_release);
  PANIC_IF(
    srn_thread_spawn(&reactor->thread, reactor_thread_main, reactor) != SRN_THREAD_OK,
    "reactor: failed to spawn the reactor thread"
  );
}