io.h File Reference

The fiber facing IO API: blocking looking calls a fiber uses to do IO. More...

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

Include dependency graph for io.h:

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

Go to the source code of this file.

Data Structures
struct	srn_io_fd_result_t
	The result of an op that yields a descriptor (srn_fiber_socket, srn_fiber_open). More...
struct	srn_io_size_result_t
	The result of a byte transfer op (read/write/sendto/recvmsg/sendmsg). More...
struct	srn_io_accept_result_t
	The result of srn_fiber_accept. More...
struct	srn_io_recvfrom_result_t
	The result of srn_fiber_recvfrom. More...

Macros
#define	SRN_WITH_RETRY(err, ...)
	Run block and retry it (yielding between attempts through srn_fiber_io_retry) while err holds the transient CHANNEL_BUSY error.

Typedefs
typedef struct srn_io_fd_result_t	srn_io_fd_result_t
	The result of an op that yields a descriptor (srn_fiber_socket, srn_fiber_open).
typedef struct srn_io_size_result_t	srn_io_size_result_t
	The result of a byte transfer op (read/write/sendto/recvmsg/sendmsg).
typedef struct srn_io_accept_result_t	srn_io_accept_result_t
	The result of srn_fiber_accept.
typedef struct srn_io_recvfrom_result_t	srn_io_recvfrom_result_t
	The result of srn_fiber_recvfrom.

Functions
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_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_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.
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_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.
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 facing IO API: blocking looking calls a fiber uses to do IO.

Each call builds a request, suspends the calling fiber, and hands the request to the reactor; the fiber resumes with the result once the operation completes. The calling worker is never blocked – it runs other fibers while this one is parked. These are the fiber world replacements for the raw read/write and other similar IO related syscalls.

Each call returns a small per-op result which is a fallible value (see errors.h): maybe_error is null on success and carries the failure otherwise, so the success fields are meaningful only when there is no error. Calls with no value to report (connect, sleep) return srn_error_t * directly – null on success.

Constraints: a suspending call must run on a fiber that is on a worker (it suspends the fiber, so there must be one). A pollable, non-blocking descriptor (socket, pipe, ...) goes through the reactor; a regular file is served by a synchronous syscall on the worker, since a readiness backend cannot poll it.

Backpressure: a channel admits at most one ring's worth of operations in flight, and a suspending call past that cap fails with the transient CHANNEL_BUSY error. SRN_WITH_RETRY wraps a block in the retry loop (yielding between attempts through srn_fiber_io_retry):

 srn_io_size_result_t r;
 SRN_WITH_RETRY(r.maybe_error, {
   r = srn_fiber_read(fd, buf, len, -1);
 });

At the end of this file there are a set of convenient non-suspending helpers (socket, bind, listen, close, shutdown, getsockopt, setsockopt, open), thin wrappers over the plain syscalls that share the same fallible result convention. They do not touch the reactor and may be called from any context. socket and open force O_NONBLOCK/O_CLOEXEC so a descriptor handed to the suspending calls above is always safe for the reactor.

Definition in file io.h.

Macro Definition Documentation

SRN_WITH_RETRY

#define SRN_WITH_RETRY	(		err,
			... )

Value:

do                             \
  __VA_ARGS__                  \
while (srn_fiber_io_retry(err))

Run block and retry it (yielding between attempts through srn_fiber_io_retry) while err holds the transient CHANNEL_BUSY error.

err is the error lvalue the block assigns, r.maybe_error for a fallible result struct, the variable itself for a plain srn_error_t *. Variadic so commas inside the block do not split the arguments.

srn_io_size_result_t r;
SRN_WITH_RETRY(r.maybe_error, {
  r = srn_fiber_read(fd, buf, len, -1);
});

srn_error_t *e;
SRN_WITH_RETRY(e, { e = srn_fiber_sleep(duration); });

Definition at line 132 of file io.h.

#define SRN_WITH_RETRY(err, ...) \
  do                             \
    __VA_ARGS__                  \
  while (srn_fiber_io_retry(err))

Typedef Documentation

srn_io_accept_result_t

typedef struct srn_io_accept_result_t srn_io_accept_result_t

The result of srn_fiber_accept.

On success fd is the accepted descriptor and, when an address buffer was given, addrlen is the length of the peer address written into it.

srn_io_fd_result_t

typedef struct srn_io_fd_result_t srn_io_fd_result_t

The result of an op that yields a descriptor (srn_fiber_socket, srn_fiber_open).

On success fd is the descriptor; on failure maybe_error is set.

srn_io_recvfrom_result_t

typedef struct srn_io_recvfrom_result_t srn_io_recvfrom_result_t

The result of srn_fiber_recvfrom.

On success count is the number of bytes received and, when an address buffer was given, addrlen is the length of the sender address written into it.

srn_io_size_result_t

typedef struct srn_io_size_result_t srn_io_size_result_t

The result of a byte transfer op (read/write/sendto/recvmsg/sendmsg).

On success count is the number of bytes moved, on failure maybe_error is set and count is unspecified.

Function Documentation

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 308 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 435 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 451 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 330 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 467 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 243 of file io.c.

                                        {
  if (e == nullptr || e->tag != CHANNEL_BUSY) {
    return false;
  }
  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 443 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 484 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 284 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 340 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 376 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 387 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 360 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 476 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 459 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 282 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 278 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 251 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 274 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 427 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 296 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: