reactor.h File Reference

Reactor overview. More...

#include <stddef.h>
#include <stdint.h>
#include "serene/rt/errors.h"

Include dependency graph for reactor.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	srn_reactor_io_request_t
	A submission, one request a fiber places on its channel. More...
struct	srn_reactor_io_completion_t
	A completion, the result of one submission, echoed back on the channel. More...

Typedefs
typedef enum srn_reactor_io_op_t	srn_reactor_io_op_t
	The operation a submission asks the reactor to perform.
typedef struct srn_reactor_io_request_t	srn_reactor_io_request_t
	A submission, one request a fiber places on its channel.
typedef struct srn_reactor_io_completion_t	srn_reactor_io_completion_t
	A completion, the result of one submission, echoed back on the channel.
typedef void(*	srn_reactor_notify_fn) (srn_scheduler_t *sched, size_t channel)
	The interface that the scheduler supplies so the reactor can announce that a channel has completions, without the reactor knowing about workers or parking.

Enumerations
enum	srn_reactor_io_op_t {   SRN_REACTOR_IO_NOP = 0 , SRN_REACTOR_IO_SLEEP , SRN_REACTOR_IO_READ , SRN_REACTOR_IO_WRITE ,   SRN_REACTOR_IO_CANCEL , SRN_REACTOR_IO_ACCEPT , SRN_REACTOR_IO_CONNECT , SRN_REACTOR_IO_RECVFROM ,   SRN_REACTOR_IO_SENDTO , SRN_REACTOR_IO_RECVMSG , SRN_REACTOR_IO_SENDMSG , SRN_REACTOR_IO_POLL }
	The operation a submission asks the reactor to perform. More...

Functions
srn_reactor_t *	srn_reactor_init (srn_engine_t *engine)
	Allocate the IO reactor from the engine.
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.
void	srn_reactor_shutdown (srn_reactor_t *reactor)
	Tear the reactor down, stop and join the reactor thread and release its channels.
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.
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.
void	srn_reactor_consume (srn_reactor_t *reactor, size_t channel)
	Drain and act on channel's completions.
bool	srn_reactor_channel_has_completions (srn_reactor_t *reactor, size_t channel)
	Whether channel's completion queue has unconsumed completions.

Detailed Description

Reactor overview.

The reactor is in charge of all the IO in the runtime.

It is completion based and modeled after io_uring; a submission describes a whole request, and a completion reports its result. The reactor utilizes different backends for different platforms. A readiness backend such as epoll performs the transfer itself when the descriptor signals readiness and posts the finished result, so the contract is the common subset every platform can honour.

A worker and the reactor talk through a channel: a channel is a pair of submission queue (SQ) and a completion queue (CQ), a SQ carrying requests toward the reactor and a CQ carrying results back. There is one channel per worker. The reactor owns every channel; a worker holds only the channel id and never touches a raw queue.

Each request carries fiber_data, an opaque token the reactor round-trips untouched: it rides in on the submission and comes back out on the completion, and is how a finished result finds the fiber that was waiting.

Definition in file reactor.h.

Typedef Documentation

srn_reactor_io_completion_t

typedef struct srn_reactor_io_completion_t srn_reactor_io_completion_t

A completion, the result of one submission, echoed back on the channel.

It is a fallible value (see errors.h). maybe_error is null on success and points at the failure otherwise, so result/addrlen are meaningful only when there is no error. The backend sets maybe_error from a static IO error to avoid unnecessary memory allocation.

srn_reactor_io_op_t

typedef enum srn_reactor_io_op_t srn_reactor_io_op_t

The operation a submission asks the reactor to perform.

srn_reactor_io_request_t

typedef struct srn_reactor_io_request_t srn_reactor_io_request_t

A submission, one request a fiber places on its channel.

The reactor copies it into the submission queue, so the struct itself need not outlive the call; a buffer it points at (buf) must stay valid until the operation completes.

srn_reactor_notify_fn

typedef void(* srn_reactor_notify_fn) (srn_scheduler_t *sched, size_t channel)

The interface that the scheduler supplies so the reactor can announce that a channel has completions, without the reactor knowing about workers or parking.

The reactor calls it after placing completions on channel; the scheduler wakes that channel's worker if it is parked.

Definition at line 177 of file reactor.h.

Enumeration Type Documentation

srn_reactor_io_op_t

enum srn_reactor_io_op_t

The operation a submission asks the reactor to perform.

Enumerator
SRN_REACTOR_IO_NOP	Completes immediately with a zero result. Exercises the submit-and-complete path on its own, with no descriptor and no kernel mechanism.
SRN_REACTOR_IO_SLEEP	Wait out a deadline.
SRN_REACTOR_IO_READ	Read from a descriptor into buf. Uses fd, buf, len, offset.
SRN_REACTOR_IO_WRITE	Write buf to a descriptor. Uses fd, buf, len, offset.
SRN_REACTOR_IO_CANCEL	Cancel one inflight operation, named by its fiber_data. Not implemented, submitting it panics. It lands together with the fd deregistration work.
SRN_REACTOR_IO_ACCEPT	Accept a pending connection on the listening socket fd, with flags (the accept4 flags). The peer address is written to addr/addrlen when addr is non-null. The completion's result is the accepted descriptor; failures arrive through the completion's maybe_error.
SRN_REACTOR_IO_CONNECT	Connect fd to the address in addr/addrlen. The connect is started at submit time and completes once the socket is writable. result is 0 on success; failures arrive through the completion's maybe_error.
SRN_REACTOR_IO_RECVFROM	Receive a datagram from fd into buf/len with flags. The sender address is written to addr/addrlen when addr is non-null. result is bytes received; failures arrive through the completion's maybe_error.
SRN_REACTOR_IO_SENDTO	Send buf/len on fd to the address in addr/addrlen with flags. result is bytes sent; failures arrive through the completion's maybe_error.
SRN_REACTOR_IO_RECVMSG	Receive into the struct msghdr that buf points at, on fd, with flags (scatter/gather and ancillary data). result is bytes received; failures arrive through the completion's maybe_error.
SRN_REACTOR_IO_SENDMSG	Send the struct msghdr that buf points at, on fd, with flags. result is bytes sent; failures arrive through the completion's maybe_error.
SRN_REACTOR_IO_POLL	Wait until fd is ready for the events requested in flags (EPOLLIN/EPOLLOUT) without performing any transfer. result is the ready-events bitmask. The gateway for timerfd, signalfd, eventfd, inotify and pidfd.

Definition at line 60 of file reactor.h.

                                 {
  /// Completes immediately with a zero result. Exercises the
  /// submit-and-complete
  /// path on its own, with no descriptor and no kernel mechanism.
  SRN_REACTOR_IO_NOP = 0,
  /// Wait out a deadline.
  SRN_REACTOR_IO_SLEEP,
  /// Read from a descriptor into `buf`. Uses `fd`, `buf`, `len`, `offset`.
  SRN_REACTOR_IO_READ,
  /// Write `buf` to a descriptor. Uses `fd`, `buf`, `len`, `offset`.
  SRN_REACTOR_IO_WRITE,
  /// Cancel one inflight operation, named by its `fiber_data`. Not
  /// implemented, submitting it panics. It lands together with the fd
  /// deregistration work.
  SRN_REACTOR_IO_CANCEL,
  /// Accept a pending connection on the listening socket `fd`, with `flags`
  /// (the `accept4` flags). The peer address is written to `addr`/`addrlen`
  /// when `addr` is non-null. The completion's `result` is the accepted
  /// descriptor; failures arrive through the completion's `maybe_error`.
  SRN_REACTOR_IO_ACCEPT,
  /// Connect `fd` to the address in `addr`/`addrlen`. The connect is started at
  /// submit time and completes once the socket is writable. `result` is 0 on
  /// success; failures arrive through the completion's `maybe_error`.
  SRN_REACTOR_IO_CONNECT,
  /// Receive a datagram from `fd` into `buf`/`len` with `flags`. The sender
  /// address is written to `addr`/`addrlen` when `addr` is non-null. `result`
  /// is bytes received; failures arrive through the completion's
  /// `maybe_error`.
  SRN_REACTOR_IO_RECVFROM,
  /// Send `buf`/`len` on `fd` to the address in `addr`/`addrlen` with `flags`.
  /// `result` is bytes sent; failures arrive through the completion's
  /// `maybe_error`.
  SRN_REACTOR_IO_SENDTO,
  /// Receive into the `struct msghdr` that `buf` points at, on `fd`, with
  /// `flags` (scatter/gather and ancillary data). `result` is bytes received;
  /// failures arrive through the completion's `maybe_error`.
  SRN_REACTOR_IO_RECVMSG,
  /// Send the `struct msghdr` that `buf` points at, on `fd`, with `flags`.
  /// `result` is bytes sent; failures arrive through the completion's
  /// `maybe_error`.
  SRN_REACTOR_IO_SENDMSG,
  /// Wait until `fd` is ready for the events requested in `flags`
  /// (`EPOLLIN`/`EPOLLOUT`) without performing any transfer. `result` is the
  /// ready-events bitmask. The gateway for timerfd, signalfd, eventfd, inotify
  /// and pidfd.
  SRN_REACTOR_IO_POLL,
} srn_reactor_io_op_t;

Function Documentation

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_consume()

void srn_reactor_consume	(	srn_reactor_t *	reactor,
		size_t	channel )

Drain and act on channel's completions.

Called by the worker that owns the channel, from its loop. Per completion it sets the waiter's result, readies the fiber, and drops the in-flight count.

Drain and act on channel's completions.

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.

Definition at line 121 of file io.c.

                                                                 {
  // 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;
      srn_fiber_ready(w->fiber);
      srn_reactor_op_done(reactor, channel);
    }
  }
}

Here is the call graph for this function:

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_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: