io.c File Reference

The fiber side of the reactor: the only layer that knows both fibers and the reactor. More...

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

Include dependency graph for io.c:

Go to the source code of this file.

Data Structures
struct	io_waiter_t
	Lives on the suspended fiber's stack; the worker fills it from the completion and the fiber reads it on resume. More...
struct	submit_ctx_t
	Carries the submission across the suspend, on the fiber's stack. More...

Macros
#define	MICROSEC   1000ULL
#define	MILLISEC   1000ULL * MICROSEC
#define	SEC   1000ULL * MILLISEC

Typedefs
typedef struct io_waiter_t	io_waiter_t
	Lives on the suspended fiber's stack; the worker fills it from the completion and the fiber reads it on resume.
typedef struct submit_ctx_t	submit_ctx_t
	Carries the submission across the suspend, on the fiber's stack.

Functions
static bool	reactor_submit_commit (srn_fiber_t *fiber, void *arg)
	Runs on the worker's loop once the fiber has switched out – the race free point to hand the request to the reactor.
void	srn_reactor_consume (srn_reactor_t *reactor, size_t channel)
	Runs on the worker loop who owns the channel.
static bool	fd_needs_blocking (srn_reactor_t *reactor, int fd)
	Whether fd must be served by a synchronous syscall on the calling worker rather than the reactor.
static io_waiter_t	blocking_io (const srn_reactor_io_request_t *req)
	Perform a READ or WRITE synchronously on the calling worker and return its outcome.
static io_waiter_t	io_wait (srn_reactor_io_request_t request)
	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.
bool	srn_fiber_io_retry (srn_error_t *e)
	Whether a suspending IO call should be retried, true exactly when e is the transient CHANNEL_BUSY error, in which case the calling fiber yields first so its worker can consume completions (which is what reopens the channel's admission window) and run other fibers before the retry.
srn_error_t *	srn_fiber_sleep_ns (uint64_t duration_ns)
	Suspend the calling fiber for at least duration_ns nanoseconds, letting its worker run other fibers meanwhile.
static uint64_t	to_ns (uint64_t duration, uint64_t ns_per_unit)
	Convert a coarser duration to nanoseconds, saturating instead of wrapping.
srn_error_t *	srn_fiber_sleep_us (uint64_t duration_us)
	Suspend the calling fiber for at least duration_us microseconds, letting its worker run other fibers meanwhile.
srn_error_t *	srn_fiber_sleep_ms (uint64_t duration_ms)
	Suspend the calling fiber for at least duration_ms miliseconds, letting its worker run other fibers meanwhile.
srn_error_t *	srn_fiber_sleep (uint64_t duration)
	Suspend the calling fiber for at least duration seconds, letting its worker run other fibers meanwhile.
srn_io_size_result_t	srn_fiber_read (int fd, void *buf, size_t len, int64_t offset)
	Read up to len bytes from fd into buf, suspending the calling fiber until the operation completes rather than blocking its worker.
srn_io_size_result_t	srn_fiber_write (int fd, const void *buf, size_t len, int64_t offset)
	Write len bytes from buf to fd, suspending the calling fiber until the operation completes rather than blocking its worker.
srn_io_accept_result_t	srn_fiber_accept (int fd, struct sockaddr *addr, socklen_t addr_cap)
	Accept a pending connection on the listening socket fd, suspending the calling fiber until one arrives.
srn_error_t *	srn_fiber_connect (int fd, const struct sockaddr *addr, socklen_t addrlen)
	Connect the socket fd to addr/addrlen, suspending the calling fiber until the connection completes rather than blocking its worker.
srn_io_recvfrom_result_t	srn_fiber_recvfrom (int fd, void *buf, size_t len, int flags, struct sockaddr *addr, socklen_t addr_cap)
	Receive a datagram from fd into buf/len with flags (the MSG_* flags), suspending the calling fiber until one arrives.
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)
	Send buf/len on fd to addr/addrlen with flags (the MSG_* flags), suspending the calling fiber until the send completes rather than blocking its worker.
srn_io_size_result_t	srn_fiber_recvmsg (int fd, struct msghdr *msg, int flags)
	Receive into the struct msghdr msg on fd with flags, suspending the calling fiber until data arrives.
srn_io_size_result_t	srn_fiber_sendmsg (int fd, const struct msghdr *msg, int flags)
	Send the struct msghdr msg on fd with flags, suspending the calling fiber until the send completes.
static srn_error_t *	io_status (int rc)
static srn_io_fd_result_t	io_fd (int fd)
srn_io_fd_result_t	srn_fiber_socket (int domain, int type, int protocol)
	Create a socket, like socket(2), additionally forcing SOCK_NONBLOCK and SOCK_CLOEXEC so the descriptor is safe to hand to the suspending calls (a blocking socket would stall the reactor thread).
srn_error_t *	srn_fiber_bind (int fd, const struct sockaddr *addr, socklen_t addrlen)
	Bind fd to addr/addrlen, like bind(2).
srn_error_t *	srn_fiber_listen (int fd, int backlog)
	Mark fd as a listening socket with the given backlog, like listen(2).
srn_error_t *	srn_fiber_close (int fd)
	Close fd, like close(2).
srn_error_t *	srn_fiber_shutdown (int fd, int how)
	Shut down part or all of the connection on fd (how is SHUT_RD/SHUT_WR/ SHUT_RDWR), like shutdown(2).
srn_error_t *	srn_fiber_getsockopt (int fd, int level, int optname, void *optval, socklen_t *optlen)
	Read a socket option into optval/optlen, like getsockopt(2).
srn_error_t *	srn_fiber_setsockopt (int fd, int level, int optname, const void *optval, socklen_t optlen)
	Set a socket option from optval/optlen, like setsockopt(2).
srn_io_fd_result_t	srn_fiber_open (const char *path, int flags, mode_t mode)
	Open path, like open(2), additionally forcing O_NONBLOCK and O_CLOEXEC so the descriptor is always safe to hand to the suspending calls above.

Detailed Description

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.

Definition in file io.c.

Macro Definition Documentation

MICROSEC

#define MICROSEC   1000ULL

Definition at line 56 of file io.c.

MILLISEC

#define MILLISEC   1000ULL * MICROSEC

Definition at line 57 of file io.c.

SEC

#define SEC   1000ULL * MILLISEC

Definition at line 58 of file io.c.

Typedef Documentation

io_waiter_t

typedef struct io_waiter_t io_waiter_t

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).

submit_ctx_t

typedef struct submit_ctx_t submit_ctx_t

Carries the submission across the suspend, on the fiber's stack.

Function Documentation

blocking_io()

io_waiter_t blocking_io ( const srn_reactor_io_request_t * req )

static

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.

Definition at line 185 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

fd_needs_blocking()

bool fd_needs_blocking ( srn_reactor_t * reactor, int fd )

static

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.

Definition at line 161 of file io.c.

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

Here is the caller graph for this function:

io_fd()

srn_io_fd_result_t io_fd ( int fd )

static

Definition at line 433 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

io_status()

srn_error_t * io_status ( int rc )

static

Definition at line 424 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

io_wait()

io_waiter_t io_wait ( srn_reactor_io_request_t request )

static

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.

Definition at line 217 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

reactor_submit_commit()

bool reactor_submit_commit ( srn_fiber_t * fiber, void * arg )

static

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.

Definition at line 95 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_accept()

srn_io_accept_result_t srn_fiber_accept	(	int	fd,
		struct sockaddr *	addr,
		socklen_t	addr_cap )

Accept a pending connection on the listening socket fd, suspending the calling fiber until one arrives.

When addr is non-null the peer address is written into it, up to addr_cap bytes, and the result's addrlen reports how many were written. The accepted socket is created non-blocking and close on exec, so it is ready for further fiber IO. On success the result's fd is the accepted descriptor. Must run on a fiber that is on a worker.

Definition at line 327 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_bind()

srn_error_t * srn_fiber_bind	(	int	fd,
		const struct sockaddr *	addr,
		socklen_t	addrlen )

Bind fd to addr/addrlen, like bind(2).

Returns null on success, or the error.

Definition at line 454 of file io.c.

                                                                                    {
#ifdef __linux__
  return io_status(bind(fd, addr, addrlen));
#else
#  error "You need to patch srn_fiber_bind for this platform"
#endif
}

Here is the call graph for this function:

srn_fiber_close()

srn_error_t * srn_fiber_close

(

int

fd

)

Close fd, like close(2).

Returns null on success, or the error.

Definition at line 470 of file io.c.

                                     {
#ifdef __linux__
  return io_status(close(fd));
#else
#  error "You need to patch srn_fiber_close for this platform"
#endif
}

Here is the call graph for this function:

srn_fiber_connect()

srn_error_t * srn_fiber_connect	(	int	fd,
		const struct sockaddr *	addr,
		socklen_t	addrlen )

Connect the socket fd to addr/addrlen, suspending the calling fiber until the connection completes rather than blocking its worker.

Returns null on success, or the error (such as CONNECTION_REFUSED). Must run on a fiber that is on a worker.

Definition at line 349 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_getsockopt()

srn_error_t * srn_fiber_getsockopt	(	int	fd,
		int	level,
		int	optname,
		void *	optval,
		socklen_t *	optlen )

Read a socket option into optval/optlen, like getsockopt(2).

Returns null on success, or the error.

Definition at line 486 of file io.c.

                                                                                                   {
#ifdef __linux__
  return io_status(getsockopt(fd, level, optname, optval, optlen));
#else
#  error "You need to patch srn_fiber_getsockopt for this platform"
#endif
}

Here is the call graph for this function:

srn_fiber_io_retry()

bool srn_fiber_io_retry

(

srn_error_t *

e

)

Whether a suspending IO call should be retried, true exactly when e is the transient CHANNEL_BUSY error, in which case the calling fiber yields first so its worker can consume completions (which is what reopens the channel's admission window) and run other fibers before the retry.

Any other error, and success, return false, so a do/while over this helper (see the file docs) retries precisely the backpressure case and hands every real outcome through. A retrying fiber with no runnable peers and no due completions degenerates to a paced busy loop; if that shows up in practice the upgrade is suspending on the channel until a slot frees, instead of polling.

Definition at line 261 of file io.c.

                                        {
  if (e == nullptr || e->tag != CHANNEL_BUSY) {
    return false;
  }
  FIBER_TRACEPOINT(fiber_io_retry, (void *)srn_fiber_current());
  srn_fiber_yield();
  return true;
}

Here is the call graph for this function:

srn_fiber_listen()

srn_error_t * srn_fiber_listen	(	int	fd,
		int	backlog )

Mark fd as a listening socket with the given backlog, like listen(2).

Returns null on success, or the error.

Definition at line 462 of file io.c.

                                                   {
#ifdef __linux__
  return io_status(listen(fd, backlog));
#else
#  error "You need to patch srn_fiber_listen for this platform"
#endif
}

Here is the call graph for this function:

srn_fiber_open()

srn_io_fd_result_t srn_fiber_open	(	const char *	path,
		int	flags,
		mode_t	mode )

Open path, like open(2), additionally forcing O_NONBLOCK and O_CLOEXEC so the descriptor is always safe to hand to the suspending calls above.

mode applies only when flags includes O_CREAT. The open runs synchronously on the calling thread. On success fd is the new descriptor. Regular files ignore O_NONBLOCK (they are served on the worker, not the reactor). FIFOs follow the open(2) O_NONBLOCK semantics: a read end opens at once with no writer present, and a write end with no reader fails instead of blocking until one appears.

Definition at line 503 of file io.c.

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

Here is the call graph for this function:

srn_fiber_read()

srn_io_size_result_t srn_fiber_read	(	int	fd,
		void *	buf,
		size_t	len,
		int64_t	offset )

Read up to len bytes from fd into buf, suspending the calling fiber until the operation completes rather than blocking its worker.

offset is the file offset, or -1 to read at the descriptor's current position. On success count is the number of bytes read (0 at end of file).

Definition at line 303 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_recvfrom()

srn_io_recvfrom_result_t srn_fiber_recvfrom	(	int	fd,
		void *	buf,
		size_t	len,
		int	flags,
		struct sockaddr *	addr,
		socklen_t	addr_cap )

Receive a datagram from fd into buf/len with flags (the MSG_* flags), suspending the calling fiber until one arrives.

When addr is non-null the sender address is written into it, up to addr_cap bytes, and the result's addrlen reports how many. On success count is the number of bytes received. Must run on a fiber that is on a worker.

Definition at line 359 of file io.c.

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

Here is the call graph for this function:

srn_fiber_recvmsg()

srn_io_size_result_t srn_fiber_recvmsg	(	int	fd,
		struct msghdr *	msg,
		int	flags )

Receive into the struct msghdr msg on fd with flags, suspending the calling fiber until data arrives.

Scatters across msg's iovecs and fills its msg_name/msg_control as requested. On success count is the number of bytes received. Must run on a fiber that is on a worker.

Definition at line 395 of file io.c.

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

Here is the call graph for this function:

srn_fiber_sendmsg()

srn_io_size_result_t srn_fiber_sendmsg	(	int	fd,
		const struct msghdr *	msg,
		int	flags )

Send the struct msghdr msg on fd with flags, suspending the calling fiber until the send completes.

Gathers from msg's iovecs. On success count is the number of bytes sent. Must run on a fiber that is on a worker.

Definition at line 406 of file io.c.

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

Here is the call graph for this function:

srn_fiber_sendto()

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 )

Send buf/len on fd to addr/addrlen with flags (the MSG_* flags), suspending the calling fiber until the send completes rather than blocking its worker.

On success count is the number of bytes sent. Must run on a fiber that is on a worker.

Definition at line 379 of file io.c.

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

Here is the call graph for this function:

srn_fiber_setsockopt()

srn_error_t * srn_fiber_setsockopt	(	int	fd,
		int	level,
		int	optname,
		const void *	optval,
		socklen_t	optlen )

Set a socket option from optval/optlen, like setsockopt(2).

Returns null on success, or the error.

Definition at line 495 of file io.c.

                                                                                           {
#ifdef __linux__
  return io_status(setsockopt(fd, level, optname, optval, optlen));
#else
#  error "You need to patch srn_fiber_setsockopt for this platform"
#endif
}

Here is the call graph for this function:

srn_fiber_shutdown()

srn_error_t * srn_fiber_shutdown	(	int	fd,
		int	how )

Shut down part or all of the connection on fd (how is SHUT_RD/SHUT_WR/ SHUT_RDWR), like shutdown(2).

Returns null on success, or the error.

Definition at line 478 of file io.c.

                                                 {
#ifdef __linux__
  return io_status(shutdown(fd, how));
#else
#  error "You need to patch srn_fiber_shutdown for this platform"
#endif
}

Here is the call graph for this function:

srn_fiber_sleep()

srn_error_t * srn_fiber_sleep

(

uint64_t

duration

)

Suspend the calling fiber for at least duration seconds, letting its worker run other fibers meanwhile.

Returns null on a normal wake, or the CANCELED error if the sleep was cut short by a scheduler wind down.

Definition at line 301 of file io.c.

{ return srn_fiber_sleep_ns(to_ns(duration, SEC)); }

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_sleep_ms()

srn_error_t * srn_fiber_sleep_ms

(

uint64_t

duration_ms

)

Suspend the calling fiber for at least duration_ms miliseconds, letting its worker run other fibers meanwhile.

Returns null on a normal wake, or the CANCELED error if the sleep was cut short by a scheduler wind down.

Definition at line 297 of file io.c.

                                                      {
  return srn_fiber_sleep_ns(to_ns(duration_ms, MILLISEC));
}

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_sleep_ns()

srn_error_t * srn_fiber_sleep_ns

(

uint64_t

duration_ns

)

Suspend the calling fiber for at least duration_ns nanoseconds, letting its worker run other fibers meanwhile.

Returns null on a normal wake, or the CANCELED error if the sleep was cut short by a scheduler wind down.

Definition at line 270 of file io.c.

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

Here is the call graph for this function:

Here is the caller graph for this function:

srn_fiber_sleep_us()

srn_error_t * srn_fiber_sleep_us

(

uint64_t

duration_us

)

Suspend the calling fiber for at least duration_us microseconds, letting its worker run other fibers meanwhile.

Returns null on a normal wake, or the CANCELED error if the sleep was cut short by a scheduler wind down.

Definition at line 293 of file io.c.

                                                      {
  return srn_fiber_sleep_ns(to_ns(duration_us, MICROSEC));
}

Here is the call graph for this function:

srn_fiber_socket()

srn_io_fd_result_t srn_fiber_socket	(	int	domain,
		int	type,
		int	protocol )

Create a socket, like socket(2), additionally forcing SOCK_NONBLOCK and SOCK_CLOEXEC so the descriptor is safe to hand to the suspending calls (a blocking socket would stall the reactor thread).

On success fd is the new descriptor.

Definition at line 446 of file io.c.

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

Here is the call graph for this function:

srn_fiber_write()

srn_io_size_result_t srn_fiber_write	(	int	fd,
		const void *	buf,
		size_t	len,
		int64_t	offset )

Write len bytes from buf to fd, suspending the calling fiber until the operation completes rather than blocking its worker.

offset is the file offset, or -1 to write at the descriptor's current position. On success count is the number of bytes written.

Definition at line 315 of file io.c.

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

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 )

Runs on the worker loop who owns the channel.

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

Here is the call graph for this function:

Here is the caller graph for this function:

to_ns()

uint64_t to_ns ( uint64_t duration, uint64_t ns_per_unit )

static

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.

Definition at line 289 of file io.c.

                                                               {
  return duration > UINT64_MAX / ns_per_unit ? UINT64_MAX : duration * ns_per_unit;
}

Here is the caller graph for this function: