reactor.c File Reference

#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"

Include dependency graph for reactor.c:

Go to the source code of this file.

Data Structures
struct	srn_timer_t

Typedefs
typedef struct srn_timer_t	srn_timer_t

Functions
srn_reactor_t *	srn_reactor_init (srn_engine_t *engine)
	Allocate the IO reactor from the engine.
void	srn_reactor_shutdown (srn_reactor_t *reactor)
	Tear the reactor down, stop and join the reactor thread and release its channels.
srn_error_t *	srn_reactor_errors_static (srn_error_tag_t tag)
	Return the shared, immutable error for an IO failure tag.
void	srn_reactor_post_completion (srn_reactor_t *reactor, size_t channel, srn_reactor_io_completion_t completion)
	Post a finished operation to channel's completion queue and rouse the channel's worker through the notify mechanism.
void	srn_reactor_op_done (srn_reactor_t *reactor, size_t channel)
	Drop channel's and the reactor's in-flight counts by one.
bool	srn_reactor_idle (srn_reactor_t *reactor)
	Whether the reactor has no operations in flight.
bool	srn_reactor_submit (srn_reactor_t *reactor, size_t channel, const srn_reactor_io_request_t *request)
	Place a request on channel's submission queue and rouse the reactor.
size_t	srn_reactor_reap (srn_reactor_t *reactor, size_t channel, srn_reactor_io_completion_t *out, size_t max)
	Drain up to max completions from channel's completion queue into out, returning how many were taken.
bool	srn_reactor_channel_has_completions (srn_reactor_t *reactor, size_t channel)
	Whether channel's completion queue has unconsumed completions.
static bool	timer_less (srn_context_t *ctx, const void *a, const void *b)
	Order timers by deadline so the soonest sits at the heap root.
static void	timer_add (srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req)
	Register a SLEEP: allocate a timer entry and push it onto the heap.
static int	timer_next_timeout (srn_reactor_t *r)
	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 void	timer_fire_expired (srn_reactor_t *r)
	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	process_request (srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req)
static void	drain_submissions (srn_reactor_t *r)
static void	reactor_thread_main (void *arg)
void	srn_reactor_activate (srn_reactor_t *reactor, size_t nchannels, srn_reactor_notify_fn notify)
	Bring the reactor up, allocate nchannels channels (one per worker) and start the reactor thread.

Variables
static srn_error_t	io_errors []
static const pqmh_control_t	timer_ctl = {.less = timer_less}

Typedef Documentation

srn_timer_t

typedef struct srn_timer_t srn_timer_t

Function Documentation

drain_submissions()

void drain_submissions ( srn_reactor_t * r )

static

Definition at line 367 of file reactor.c.

                                                {
  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);
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

process_request()

void process_request ( srn_reactor_t * r, size_t channel, const srn_reactor_io_request_t * req )

static

Definition at line 336 of file reactor.c.

                                                                                                   {
  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");
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

reactor_thread_main()

void reactor_thread_main ( void * arg )

static

Definition at line 378 of file reactor.c.

                                           {
  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);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_reactor_activate()

void srn_reactor_activate	(	srn_reactor_t *	reactor,
		size_t	nchannels,
		srn_reactor_notify_fn	notify )

Bring the reactor up, allocate nchannels channels (one per worker) and start the reactor thread.

notify is called with the scheduler and the channel id whenever a channel gains completions.

Definition at line 390 of file reactor.c.

                                                                                                  {
 
  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"
  );
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_reactor_channel_has_completions()

bool srn_reactor_channel_has_completions ( srn_reactor_t * reactor, size_t channel )

nodiscard

Whether channel's completion queue has unconsumed completions.

The worker checks this under the scheduler lock before parking, so a completion landing just before it sleeps is not missed.

Definition at line 237 of file reactor.c.

                                                                                 {
  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;
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_reactor_errors_static()

srn_error_t * srn_reactor_errors_static

(

srn_error_tag_t

tag

)

Return the shared, immutable error for an IO failure tag.

These live in static storage, not a context, so a completion or result can point one at maybe_error with no allocation and no lifetime concern, and it is safe to share across threads. A backend maps its native code (via srn_errno_to_tag on POSIX) to a tag, then to one of these. Any tag without a dedicated entry yields the UNKNOWN_IO_ERROR error.

Definition at line 125 of file reactor.c.

                                                            {
 
  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];
}

Here is the caller graph for this function:

srn_reactor_idle()

bool srn_reactor_idle ( srn_reactor_t * reactor )

nodiscard

Whether the reactor has no operations in flight.

The scheduler folds this into quiescence, a pool with parked workers and an empty run queue is only truly done when the reactor is idle too.

Definition at line 169 of file reactor.c.

                                              {
  PANIC_IF_NULL(reactor);
  return atomic_load(&reactor->inflight) == 0;
}

Here is the caller graph for this function:

srn_reactor_init()

srn_reactor_t * srn_reactor_init ( srn_engine_t * engine )

nodiscard

Allocate the IO reactor from the engine.

Definition at line 34 of file reactor.c.

                                                      {
  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;
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_reactor_op_done()

void srn_reactor_op_done	(	srn_reactor_t *	reactor,
		size_t	channel )

Drop channel's and the reactor's in-flight counts by one.

The consumer of a completion calls this after it has finished acting on it (for fibers, after the waiting fiber is readied), so that "in flight" reaches zero only once every result has been both produced and handled. Dropping the channel count is also what re-opens the channel's admission window (see srn_reactor_submit).

Definition at line 162 of file reactor.c.

                                                                 {
  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);
}

Here is the caller graph for this function:

srn_reactor_post_completion()

void srn_reactor_post_completion	(	srn_reactor_t *	reactor,
		size_t	channel,
		srn_reactor_io_completion_t	completion )

Post a finished operation to channel's completion queue and rouse the channel's worker through the notify mechanism.

Runs on the reactor thread, called by the core for ops it completes itself (such as NOP) and by a backend when a descriptor operation finishes. It does NOT drop the in-flight count: that is the consumer's job via srn_reactor_op_done, so an op counts as gone only once the waiting fiber has been readied, never in the gap between.

Definition at line 141 of file reactor.c.

  {
  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"
  );
 
  // 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);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_reactor_reap()

size_t srn_reactor_reap	(	srn_reactor_t *	reactor,
		size_t	channel,
		srn_reactor_io_completion_t *	out,
		size_t	max )

Drain up to max completions from channel's completion queue into out, returning how many were taken.

Definition at line 220 of file reactor.c.

  {
  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;
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_reactor_shutdown()

void srn_reactor_shutdown

(

srn_reactor_t *

reactor

)

Tear the reactor down, stop and join the reactor thread and release its channels.

Definition at line 66 of file reactor.c.

                                                  {
  // 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);
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_reactor_submit()

bool srn_reactor_submit ( srn_reactor_t * reactor, size_t channel, const srn_reactor_io_request_t * request )

nodiscard

Place a request on channel's submission queue and rouse the reactor.

Returns false without submitting when the channel already has a full ring's worth of operations in flight. The condition is transient, it clears as the channel's worker consumes completions, so the caller can fail the operation back with CHANNEL_BUSY and let the fiber retry.

Definition at line 174 of file reactor.c.

  {
  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) {
    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"
  );
 
  // Rouse the reactor so it drains the new request at once instead of waiting
  // out its idle timeout.
  reactor->backend->wake(reactor);
  return true;
}

Here is the call graph for this function:

Here is the caller graph for this function:

timer_add()

void timer_add ( srn_reactor_t * r, size_t channel, const srn_reactor_io_request_t * req )

static

Register a SLEEP: allocate a timer entry and push it onto the heap.

Definition at line 274 of file reactor.c.

                                                                                             {
  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;
  pqmh_push(r->ctx, &r->timers, t);
}

Here is the call graph for this function:

Here is the caller graph for this function:

timer_fire_expired()

void timer_fire_expired ( srn_reactor_t * r )

static

Fire every timer whose deadline has passed, pop it (in deadline order), post a zero result completion to its channel, and free the entry.

Definition at line 316 of file reactor.c.

                                                 {
  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;
 
    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);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

timer_less()

bool timer_less ( srn_context_t * ctx, const void * a, const void * b )

static

Order timers by deadline so the soonest sits at the heap root.

Definition at line 264 of file reactor.c.

                                                                         {
  UNUSED(ctx);
  return ((const srn_timer_t *)a)->deadline_ns < ((const srn_timer_t *)b)->deadline_ns;
}

timer_next_timeout()

int timer_next_timeout ( srn_reactor_t * r )

static

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.

Definition at line 291 of file reactor.c.

                                                {
  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;
}

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

io_errors

srn_error_t io_errors[]

static

Initial value:

                             = {
  {CONNECTION_REFUSED, (char *)"connection refused"},
  {CONNECTION_RESET, (char *)"connection reset by peer"},
  {CONNECTION_ABORTED, (char *)"connection aborted"},
  {BROKEN_PIPE, (char *)"broken pipe"},
  {BAD_DESCRIPTOR, (char *)"bad file descriptor"},
  {TIMED_OUT, (char *)"operation timed out"},
  {CANCELED, (char *)"operation canceled"},
  {NOT_CONNECTED, (char *)"socket is not connected"},
  {ADDRESS_IN_USE, (char *)"address already in use"},
  {HOST_UNREACHABLE, (char *)"host is unreachable"},
  {NETWORK_UNREACHABLE, (char *)"network is unreachable"},
  {MESSAGE_TOO_LONG, (char *)"message too long"},
  {TOO_MANY_OPEN_FILES, (char *)"too many open files"},
  {PERMISSION_DENIED, (char *)"permission denied"},
  {OUT_OF_MEMORY, (char *)"out of memory"},
  {NOT_FOUND, (char *)"no such file or directory"},
  {ALREADY_EXISTS, (char *)"file already exists"},
  {IS_A_DIRECTORY, (char *)"is a directory"},
  {NOT_A_DIRECTORY, (char *)"not a directory"},
  {NO_SPACE, (char *)"no space left on device"},
  {INVALID_ARGUMENT, (char *)"invalid argument"},
  {CHANNEL_BUSY, (char *)"reactor channel is at capacity"},
  {UNKNOWN_IO_ERROR, (char *)"unknown IO error"},
}

Definition at line 99 of file reactor.c.

                               {
  {CONNECTION_REFUSED, (char *)"connection refused"},
  {CONNECTION_RESET, (char *)"connection reset by peer"},
  {CONNECTION_ABORTED, (char *)"connection aborted"},
  {BROKEN_PIPE, (char *)"broken pipe"},
  {BAD_DESCRIPTOR, (char *)"bad file descriptor"},
  {TIMED_OUT, (char *)"operation timed out"},
  {CANCELED, (char *)"operation canceled"},
  {NOT_CONNECTED, (char *)"socket is not connected"},
  {ADDRESS_IN_USE, (char *)"address already in use"},
  {HOST_UNREACHABLE, (char *)"host is unreachable"},
  {NETWORK_UNREACHABLE, (char *)"network is unreachable"},
  {MESSAGE_TOO_LONG, (char *)"message too long"},
  {TOO_MANY_OPEN_FILES, (char *)"too many open files"},
  {PERMISSION_DENIED, (char *)"permission denied"},
  {OUT_OF_MEMORY, (char *)"out of memory"},
  {NOT_FOUND, (char *)"no such file or directory"},
  {ALREADY_EXISTS, (char *)"file already exists"},
  {IS_A_DIRECTORY, (char *)"is a directory"},
  {NOT_A_DIRECTORY, (char *)"not a directory"},
  {NO_SPACE, (char *)"no space left on device"},
  {INVALID_ARGUMENT, (char *)"invalid argument"},
  {CHANNEL_BUSY, (char *)"reactor channel is at capacity"},
  {UNKNOWN_IO_ERROR, (char *)"unknown IO error"},
};

timer_ctl

const pqmh_control_t timer_ctl = {.less = timer_less}

static

Definition at line 269 of file reactor.c.

{.less = timer_less};