Serene Runtime 1.0.0-dev
C runtime for the Serene programming language
Loading...
Searching...
No Matches
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_tsrn_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_tsrn_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_tsrn_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_tsrn_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_tsrn_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_tio_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_tsrn_fiber_bind (int fd, const struct sockaddr *addr, socklen_t addrlen)
 Bind fd to addr/addrlen, like bind(2).
srn_error_tsrn_fiber_listen (int fd, int backlog)
 Mark fd as a listening socket with the given backlog, like listen(2).
srn_error_tsrn_fiber_close (int fd)
 Close fd, like close(2).
srn_error_tsrn_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_tsrn_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_tsrn_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 174 of file io.c.

174 {
175#ifdef __linux__
176 assert(req != nullptr);
177 ssize_t n;
178
179 if (req->op == SRN_REACTOR_IO_READ) {
180 n = req->offset < 0 ? read(req->fd, req->buf, req->len)
181 : pread(req->fd, req->buf, req->len, (off_t)req->offset);
182 } else {
183 n = req->offset < 0 ? write(req->fd, req->buf, req->len)
184 : pwrite(req->fd, req->buf, req->len, (off_t)req->offset);
185 }
186
187 io_waiter_t w = {.fiber = nullptr, .maybe_error = nullptr, .result = 0, .addrlen = 0};
188 if (n < 0) {
190 w.maybe_error = srn_reactor_errors_static(err_tag);
191 } else {
192 w.result = (size_t)n;
193 }
194 return w;
195#else
196# error "You need to patch blocking_io for this platform"
197#endif
198}
int n
Definition acutest.h:525
srn_error_tag_t
The kind of a failure.
Definition errors.h:75
@ UNKNOWN_IO_ERROR
Definition errors.h:118
srn_error_tag_t srn_errno_to_tag(int errno_value, srn_error_tag_t default_tag)
Map a POSIX errno to error tag.
srn_error_t * srn_reactor_errors_static(srn_error_tag_t tag)
Return the shared, immutable error for an IO failure tag.
Definition reactor.c:125
@ SRN_REACTOR_IO_READ
Read from a descriptor into buf. Uses fd, buf, len, offset.
Definition reactor.h:68
Lives on the suspended fiber's stack; the worker fills it from the completion and the fiber reads it ...
Definition io.c:66
size_t result
Definition io.c:69
int fd
The descriptor the operation targets.
Definition reactor.h:117
srn_reactor_io_op_t op
Definition reactor.h:114
void * buf
The transfer buffer for READ, WRITE, RECVFROM, and SENDTO, and the struct msghdr * for RECVMSG and SE...
Definition reactor.h:120
size_t len
The transfer length, in bytes, for READ, WRITE, RECVFROM, and SENDTO.
Definition reactor.h:122
int64_t offset
The file offset for SRN_REACTOR_IO_READ and SRN_REACTOR_IO_WRITE; -1 reads at the current position.
Definition reactor.h:125
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 150 of file io.c.

150 {
151#ifdef __linux__
152 assert(reactor != nullptr);
154 return false;
155 }
156
157 struct stat st;
158 if (fstat(fd, &st) != 0) {
159 return false;
160 }
161 return S_ISREG(st.st_mode) || S_ISBLK(st.st_mode) || S_ISDIR(st.st_mode);
162#else
163# error "You need to patch fd_needs_blocking for this platform"
164#endif
165}
#define SRN_REACTOR_BACKEND_FEAT_FILE_OFFLOAD
The backend cannot wait on regular file readiness, so the core serves regular file READ and WRITE off...
Definition backend.h:49
uint32_t features
Bitmap of SRN_REACTOR_BACKEND_FEAT_* flags this backend requires from the reactor.
Definition backend.h:61
const srn_reactor_backend_t * backend
The chosen kernel mechanism and its private state, set at activation.
Definition internal.h:47
Here is the caller graph for this function:

◆ io_fd()

srn_io_fd_result_t io_fd ( int fd)
static

Definition at line 414 of file io.c.

414 {
415#ifdef __linux__
416 if (fd < 0) {
417 return (srn_io_fd_result_t){
419 };
420 }
421 return (srn_io_fd_result_t){.fd = fd};
422#else
423# error "You need to patch io_fd for this platform"
424#endif
425}
The result of an op that yields a descriptor (srn_fiber_socket, srn_fiber_open).
Definition io.h:66
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 405 of file io.c.

405 {
406#ifdef __linux__
407 return rc < 0 ? srn_reactor_errors_static(srn_errno_to_tag(errno, UNKNOWN_IO_ERROR)) : nullptr;
408#else
409# error "You need to patch io_status for this platform"
410#endif
411}
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 206 of file io.c.

206 {
208 PANIC_IF_NULL(self);
209
210 srn_reactor_t *reactor = self->ctx->engine->reactor;
211
212 // A READ or WRITE on an fd the backend cannot poll (a regular file on a
213 // readiness backend) runs synchronously on this worker. It never reaches the
214 // reactor, so it neither suspends nor touches the in-flight count -- the
215 // worker is simply busy in the syscall, like a compute-bound fiber.
216 if (
217 (request.op == SRN_REACTOR_IO_READ || request.op == SRN_REACTOR_IO_WRITE) &&
218 fd_needs_blocking(reactor, request.fd)
219 ) {
220 return blocking_io(&request);
221 }
222
224 PANIC_IF(wid == (srn_worker_id_t)-1, "reactor IO requested outside a worker");
225
226 // The waiter and the submit context live on this fiber's stack; the fiber
227 // stays parked (its stack preserved) until the completion comes back, so both
228 // outlive the suspend.
229 io_waiter_t waiter = {.fiber = self, .maybe_error = nullptr, .result = 0, .addrlen = 0};
230 request.fiber_data = &waiter;
231
232 submit_ctx_t s = {
233 .reactor = reactor,
234 .scheduler = self->ctx->engine->scheduler,
235 .waiter = &waiter,
236 .channel = (size_t)wid,
237 .request = request,
238 };
240 return waiter;
241}
static srn_fiber_result_t waiter(srn_context_t *ctx, void *arg)
Definition 03_wait_for.c:51
size_t srn_worker_id_t
Definition fiber.h:149
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.
Definition io.c:174
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 t...
Definition io.c:95
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.
Definition io.c:150
@ SRN_REACTOR_IO_WRITE
Write buf to a descriptor. Uses fd, buf, len, offset.
Definition reactor.h:70
srn_worker_id_t srn_sched_current_worker_id()
Return the id of the worker that the calling os thread is running, or SIZE_MAX when the calling threa...
Definition scheduler.c:1102
srn_fiber_t * srn_fiber_current(void)
The fiber currently running on this os thread, or null when the calling thread is not a worker or the...
Definition scheduler.c:1058
void srn_fiber_suspend(srn_fiber_park_fn commit, void *arg)
A suspended fiber is on no scheduler queue, and the scheduler does not track what it waits on – whoev...
Definition scheduler.c:1024
srn_engine_t * engine
Long term state of the compiler.
Definition context.h:49
srn_scheduler_t * scheduler
The fiber scheduler, that is the entry point of the fiber subsystem.
Definition engine.h:75
srn_reactor_t * reactor
The I/O reactor, that is in charge of handling everything I/O.
Definition engine.h:78
srn_fiber_result_t result
Set when state reaches SRN_FIBER_DONE.
Definition fiber.h:279
srn_context_t * ctx
Definition fiber.h:274
void * fiber_data
Opaque token, round tripped to the matching completion.
Definition reactor.h:133
Carries the submission across the suspend, on the fiber's stack.
Definition io.c:76
#define PANIC_IF_NULL(ptr)
Definition utils.h:66
#define PANIC_IF(cond, msg)
Definition utils.h:59
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.

95 {
96 UNUSED(fiber);
97 submit_ctx_t *s = (submit_ctx_t *)arg;
98
100 s->waiter->maybe_error = srn_reactor_errors_static(CANCELED);
101 return false; // decline the park; the worker readies us with the error
102 }
103
104 if (!srn_reactor_submit(s->reactor, s->channel, &s->request)) {
105 // The channel is at its in-flight cap. Fail the op back to the fiber; the
106 // condition is transient, so the fiber may retry once the channel drains.
108 return false;
109 }
110 return true; // stay parked; the worker wakes us on the completion
111}
@ CHANNEL_BUSY
The channel already has a full ring's worth of operations in flight.
Definition errors.h:117
@ CANCELED
Definition errors.h:97
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.
Definition reactor.c:174
bool srn_sched_accepting_submissions(srn_scheduler_t *sched)
Whether the scheduler still accepts new IO submissions.
Definition scheduler.c:1108
srn_reactor_t * reactor
Definition io.c:77
size_t channel
Definition io.c:80
io_waiter_t * waiter
Definition io.c:79
srn_reactor_io_request_t request
Definition io.c:81
srn_scheduler_t * scheduler
Definition io.c:78
#define UNUSED(x)
Definition utils.h:45
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 308 of file io.c.

308 {
309#ifdef __linux__
310 // Force the accepted socket non-blocking and close-on-exec, so further fiber
311 // IO on it never blocks the reactor thread and it does not leak across exec.
314 .fd = fd,
315 .addr = addr,
316 .addrlen = (uint32_t)addr_cap,
317 .flags = SOCK_NONBLOCK | SOCK_CLOEXEC,
318 };
319 io_waiter_t w = io_wait(req);
320 return (srn_io_accept_result_t){
321 .maybe_error = w.maybe_error,
322 .fd = (int)w.result,
323 .addrlen = (socklen_t)w.addrlen,
324 };
325#else
326# error "You need to patch srn_fiber_accept for this platform"
327#endif
328}
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...
Definition io.c:206
@ SRN_REACTOR_IO_ACCEPT
Accept a pending connection on the listening socket fd, with flags (the accept4 flags).
Definition reactor.h:79
uint32_t addrlen
Definition io.c:70
The result of srn_fiber_accept.
Definition io.h:86
A submission, one request a fiber places on its channel.
Definition reactor.h:113
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.

435 {
436#ifdef __linux__
437 return io_status(bind(fd, addr, addrlen));
438#else
439# error "You need to patch srn_fiber_bind for this platform"
440#endif
441}
static srn_error_t * io_status(int rc)
Definition io.c:405
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.

451 {
452#ifdef __linux__
453 return io_status(close(fd));
454#else
455# error "You need to patch srn_fiber_close for this platform"
456#endif
457}
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.

330 {
333 .fd = fd,
334 .addr = (void *)addr,
335 .addrlen = (uint32_t)addrlen,
336 };
337 return io_wait(req).maybe_error;
338}
@ SRN_REACTOR_IO_CONNECT
Connect fd to the address in addr/addrlen.
Definition reactor.h:83
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.

467 {
468#ifdef __linux__
469 return io_status(getsockopt(fd, level, optname, optval, optlen));
470#else
471# error "You need to patch srn_fiber_getsockopt for this platform"
472#endif
473}
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.

243 {
244 if (e == nullptr || e->tag != CHANNEL_BUSY) {
245 return false;
246 }
248 return true;
249}
void srn_fiber_yield(void)
Yield cooperatively, re-enqueue the running fiber and run the next ready one.
Definition scheduler.c:1000
srn_error_tag_t tag
Definition errors.h:126
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.

443 {
444#ifdef __linux__
445 return io_status(listen(fd, backlog));
446#else
447# error "You need to patch srn_fiber_listen for this platform"
448#endif
449}
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.

484 {
485#ifdef __linux__
486 // O_NONBLOCK keeps the reactor thread safe, a pollable descriptor (FIFO,
487 // character device) opened here must never make the reactor block in a
488 // transfer. Regular files ignore the flag and take the synchronous worker
489 // path anyway.
490 return io_fd(open(path, flags | O_NONBLOCK | O_CLOEXEC, mode));
491#else
492# error "You need to patch srn_fiber_open for this platform"
493#endif
494}
static srn_io_fd_result_t io_fd(int fd)
Definition io.c:414
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.

284 {
287 .fd = fd,
288 .buf = buf,
289 .len = len,
290 .offset = offset,
291 };
292 io_waiter_t w = io_wait(req);
293 return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
294}
The result of a byte transfer op (read/write/sendto/recvmsg/sendmsg).
Definition io.h:76
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.

342 {
345 .fd = fd,
346 .buf = buf,
347 .len = len,
348 .flags = flags,
349 .addr = addr,
350 .addrlen = (uint32_t)addr_cap,
351 };
352 io_waiter_t w = io_wait(req);
354 .maybe_error = w.maybe_error,
355 .count = w.result,
356 .addrlen = (socklen_t)w.addrlen,
357 };
358}
@ SRN_REACTOR_IO_RECVFROM
Receive a datagram from fd into buf/len with flags.
Definition reactor.h:88
The result of srn_fiber_recvfrom.
Definition io.h:97
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.

376 {
379 .fd = fd,
380 .buf = msg,
381 .flags = flags,
382 };
383 io_waiter_t w = io_wait(req);
384 return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
385}
@ SRN_REACTOR_IO_RECVMSG
Receive into the struct msghdr that buf points at, on fd, with flags (scatter/gather and ancillary da...
Definition reactor.h:96
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.

387 {
390 .fd = fd,
391 .buf = (void *)msg,
392 .flags = flags,
393 };
394 io_waiter_t w = io_wait(req);
395 return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
396}
@ SRN_REACTOR_IO_SENDMSG
Send the struct msghdr that buf points at, on fd, with flags.
Definition reactor.h:100
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.

362 {
365 .fd = fd,
366 .buf = (void *)buf,
367 .len = len,
368 .flags = flags,
369 .addr = (void *)addr,
370 .addrlen = (uint32_t)addrlen,
371 };
372 io_waiter_t w = io_wait(req);
373 return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
374}
@ SRN_REACTOR_IO_SENDTO
Send buf/len on fd to the address in addr/addrlen with flags.
Definition reactor.h:92
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.

476 {
477#ifdef __linux__
478 return io_status(setsockopt(fd, level, optname, optval, optlen));
479#else
480# error "You need to patch srn_fiber_setsockopt for this platform"
481#endif
482}
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.

459 {
460#ifdef __linux__
461 return io_status(shutdown(fd, how));
462#else
463# error "You need to patch srn_fiber_shutdown for this platform"
464#endif
465}
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.

282{ return srn_fiber_sleep_ns(to_ns(duration, SEC)); }
#define SEC
Definition io.c:58
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 m...
Definition io.c:251
static uint64_t to_ns(uint64_t duration, uint64_t ns_per_unit)
Convert a coarser duration to nanoseconds, saturating instead of wrapping.
Definition io.c:270
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.

278 {
279 return srn_fiber_sleep_ns(to_ns(duration_ms, MILLISEC));
280}
#define MILLISEC
Definition io.c:57
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.

251 {
252 uint64_t now = srn_now_ns();
253
254 // Saturate instead of wrapping, an absurd duration means "sleep past any
255 // horizon", not "wake at once".
256 uint64_t deadline = duration_ns > UINT64_MAX - now ? UINT64_MAX : now + duration_ns;
257
260 .deadline_ns = deadline,
261 };
262 return io_wait(req).maybe_error;
263}
@ SRN_REACTOR_IO_SLEEP
Wait out a deadline.
Definition reactor.h:66
uint64_t srn_now_ns(void)
The current time on the monotonic clock, in nanoseconds.
Definition utils.c:46
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.

274 {
275 return srn_fiber_sleep_ns(to_ns(duration_us, MICROSEC));
276}
#define MICROSEC
Definition io.c:56
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.

427 {
428#ifdef __linux__
429 return io_fd(socket(domain, type | SOCK_NONBLOCK | SOCK_CLOEXEC, protocol));
430#else
431# error "You need to patch srn_fiber_socket for this platform"
432#endif
433}
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.

296 {
299 .fd = fd,
300 .buf = (void *)buf,
301 .len = len,
302 .offset = offset,
303 };
304 io_waiter_t w = io_wait(req);
305 return (srn_io_size_result_t){.maybe_error = w.maybe_error, .count = w.result};
306}
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 121 of file io.c.

121 {
122 // The drain batch is the configured reap batch -- "most completions a worker
123 // drains from its channel in a single pass", which is exactly this loop.
124 size_t batch = reactor->ctx->engine->config.reactor.reap_batch;
125 srn_reactor_io_completion_t comps[batch];
126 size_t n;
127
128 while ((n = srn_reactor_reap(reactor, channel, comps, batch)) > 0) {
129 for (size_t i = 0; i < n; i++) {
130 io_waiter_t *w = (io_waiter_t *)comps[i].fiber_data;
131 w->maybe_error = comps[i].maybe_error;
132 w->result = comps[i].result;
133 w->addrlen = comps[i].addrlen;
135 srn_reactor_op_done(reactor, channel);
136 }
137 }
138}
void srn_reactor_op_done(srn_reactor_t *reactor, size_t channel)
Drop channel's and the reactor's in-flight counts by one.
Definition reactor.c:162
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 reactor.c:220
void srn_fiber_ready(srn_fiber_t *fiber)
Mark a suspended fiber runnable again, waking it when the event it awaited occurs.
Definition scheduler.c:1044
srn_fiber_t * fiber
Definition io.c:68
srn_reactor_config_t reactor
srn_configuration_t config
The runtime's tunable knobs, the single source for every configurable value (see configuration....
Definition engine.h:62
size_t reap_batch
Most completions a worker drains from its channel in a single pass.
A completion, the result of one submission, echoed back on the channel.
Definition reactor.h:156
uint32_t addrlen
For SRN_REACTOR_IO_ACCEPT and SRN_REACTOR_IO_RECVFROM, the actual length of the address written into ...
Definition reactor.h:169
size_t result
The result of the operation on success, bytes transferred for the transfers, the accepted descriptor ...
Definition reactor.h:163
srn_context_t * ctx
Definition internal.h:44
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 270 of file io.c.

270 {
271 return duration > UINT64_MAX / ns_per_unit ? UINT64_MAX : duration * ns_per_unit;
272}
Here is the caller graph for this function: