reactor.h

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
 Reactor overview
 
 The reactor is in charge of all the IO in the runtime.
 
 It is completion based and modeled after
 [io_uring](https://man7.org/linux/man-pages/man7/io_uring.7.html); 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.
*/
 
#pragma once
 
#include <stddef.h>
#include <stdint.h>
 
#include "serene/rt/errors.h"
 
typedef struct srn_engine_t srn_engine_t;
typedef struct srn_reactor_backend_t srn_reactor_backend_t;
 
/// The implementation is internal to the runtime
typedef struct srn_reactor_t srn_reactor_t;
typedef struct srn_scheduler_t srn_scheduler_t;
 
/**
 * The operation a submission asks the reactor to perform.
 */
typedef enum srn_reactor_io_op_t {
  /// 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;
 
/**
 * 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.
 */
typedef struct srn_reactor_io_request_t {
  srn_reactor_io_op_t op;
  /// The descriptor the operation targets. Unused by SRN_REACTOR_IO_NOP and
  /// SRN_REACTOR_IO_SLEEP.
  int fd;
  /// The transfer buffer for READ, WRITE, RECVFROM, and SENDTO, and the
  /// `struct msghdr *` for RECVMSG and SENDMSG.
  void *buf;
  /// The transfer length, in bytes, for READ, WRITE, RECVFROM, and SENDTO.
  size_t len;
  /// The file offset for SRN_REACTOR_IO_READ and SRN_REACTOR_IO_WRITE; -1 reads
  /// at the current position.
  int64_t offset;
  /// The wake deadline for SRN_REACTOR_IO_SLEEP, on the monotonic clock, in
  /// nanoseconds.
  uint64_t deadline_ns;
  /// Opaque token, round tripped to the matching completion. The reactor never
  /// reads it. For SRN_REACTOR_IO_CANCEL it names the target operation. The
  /// workers or the scheduler set their data here and expect it back on the
  /// completion counterpart
  void *fiber_data;
  /// Socket address. Input for SRN_REACTOR_IO_CONNECT and SRN_REACTOR_IO_SENDTO
  /// (the peer to use); output for SRN_REACTOR_IO_ACCEPT and
  /// SRN_REACTOR_IO_RECVFROM (the peer that connected or sent), or null to
  /// ignore. A `struct sockaddr *`, kept as `void *` so this header stays free
  /// of socket headers.
  void *addr;
  /// Length of `addr`, the exact address length for SRN_REACTOR_IO_CONNECT and
  /// SRN_REACTOR_IO_SENDTO, the buffer capacity for SRN_REACTOR_IO_ACCEPT and
  /// SRN_REACTOR_IO_RECVFROM.
  uint32_t addrlen;
  /// Operation flags, the `MSG_*` flags for the datagram and message ops, the
  /// `accept4` flags for SRN_REACTOR_IO_ACCEPT, and the requested readiness
  /// (`EPOLLIN`/`EPOLLOUT`) for SRN_REACTOR_IO_POLL.
  int flags;
} srn_reactor_io_request_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.
 */
typedef struct srn_reactor_io_completion_t {
  ERROR_HEADER;
  /// The `fiber_data` of the submission this completes.
  void *fiber_data;
  /// The result of the operation on success, bytes transferred for the
  /// transfers, the accepted descriptor for SRN_REACTOR_IO_ACCEPT, 0 for SRN_REACTOR_IO_CONNECT and
  /// SRN_REACTOR_IO_NOP. Unspecified when `maybe_error` is set.
  size_t result;
  /// For SRN_REACTOR_IO_ACCEPT and SRN_REACTOR_IO_RECVFROM, the actual length of
  /// the address written into the request's `addr` (the kernel's value-result
  /// `addrlen`), which lets a caller interpret a variable or mixed-family
  /// address (IPv6, Unix-domain, a dual-stack listener). The consumer knows its
  /// op, so it reads this only for those two; unused otherwise.
  uint32_t addrlen;
} srn_reactor_io_completion_t;
 
/**
 * 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.
 */
typedef void (*srn_reactor_notify_fn)(srn_scheduler_t *sched, size_t channel);
 
// -----------------------------------------------------------------------------
// Lifecycle
// -----------------------------------------------------------------------------
/// Allocate the IO reactor from the engine.
[[nodiscard]] [[gnu::nonnull(1)]]
srn_reactor_t *srn_reactor_init(srn_engine_t *engine);
 
/**
 * 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.
 */
[[gnu::nonnull(1)]]
void srn_reactor_activate(srn_reactor_t *reactor, size_t nchannels, srn_reactor_notify_fn notify);
 
/**
 * Tear the reactor down, stop and join the reactor thread and release its
 * channels.
 */
[[gnu::nonnull(1)]]
void srn_reactor_shutdown(srn_reactor_t *reactor);
 
// -----------------------------------------------------------------------------
// Submit and consume
// -----------------------------------------------------------------------------
/**
 * 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.
 */
[[nodiscard]] [[gnu::nonnull(1, 3)]]
bool srn_reactor_submit(
  srn_reactor_t *reactor, size_t channel, const srn_reactor_io_request_t *request
);
 
/**
 * Drain up to `max` completions from `channel`'s completion queue into `out`,
 * returning how many were taken.
 */
[[gnu::nonnull(1, 3)]]
size_t srn_reactor_reap(
  srn_reactor_t *reactor, size_t channel, srn_reactor_io_completion_t *out, size_t max
);
 
/**
 * 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).
 */
[[gnu::nonnull(1)]]
void srn_reactor_op_done(srn_reactor_t *reactor, size_t channel);
 
/**
 * 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.
 */
[[nodiscard]] [[gnu::nonnull(1)]]
bool srn_reactor_idle(srn_reactor_t *reactor);
 
/**
 * 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.
 */
[[gnu::nonnull(1)]]
void srn_reactor_consume(srn_reactor_t *reactor, size_t channel);
 
/**
 * 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.
 */
[[nodiscard]] [[gnu::nonnull(1)]]
bool srn_reactor_channel_has_completions(srn_reactor_t *reactor, size_t channel);