Serene Runtime 1.0.0-dev
C runtime for the Serene programming language
Loading...
Searching...
No Matches
reactor.c File Reference
#include "serene/rt/reactor.h"
#include <assert.h>
#include <limits.h>
#include "backend.h"
#include "internal.h"
#include "serene/rt/engine.h"
#include "serene/rt/impl/spsc_ring.h"
#include "serene/rt/mm/interface.h"
#include "serene/utils.h"
Include dependency graph for reactor.c:

Go to the source code of this file.

Data Structures

struct  srn_timer_t

Typedefs

typedef struct srn_timer_t srn_timer_t

Functions

srn_reactor_t * srn_reactor_init (srn_engine_t *engine)
 Allocate the IO reactor from the engine.
void srn_reactor_shutdown (srn_reactor_t *reactor)
 Tear the reactor down, stop and join the reactor thread and release its channels.
srn_error_t * srn_reactor_errors_static (srn_error_tag_t tag)
 Return the shared, immutable error for an IO failure tag.
void srn_reactor_post_completion (srn_reactor_t *reactor, size_t channel, srn_reactor_io_completion_t completion)
 Post a finished operation to channel's completion queue and rouse the channel's worker through the notify mechanism.
void srn_reactor_op_done (srn_reactor_t *reactor, size_t channel)
 Drop channel's and the reactor's in-flight counts by one.
bool srn_reactor_idle (srn_reactor_t *reactor)
 Whether the reactor has no operations in flight.
bool srn_reactor_submit (srn_reactor_t *reactor, size_t channel, const srn_reactor_io_request_t *request)
 Place a request on channel's submission queue and rouse the reactor.
size_t srn_reactor_reap (srn_reactor_t *reactor, size_t channel, srn_reactor_io_completion_t *out, size_t max)
 Drain up to max completions from channel's completion queue into out, returning how many were taken.
bool srn_reactor_channel_has_completions (srn_reactor_t *reactor, size_t channel)
 Whether channel's completion queue has unconsumed completions.
static bool timer_less (srn_context_t *ctx, const void *a, const void *b)
 Order timers by deadline so the soonest sits at the heap root.
static void timer_add (srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req)
 Register a SLEEP: allocate a timer entry and push it onto the heap.
static int timer_next_timeout (srn_reactor_t *r)
 Milliseconds until the soonest deadline (rounded up), clamped to a valid epoll_wait timeout, 0 if a timer is already due, INT_MAX for a far-off deadline, and -1 (block indefinitely) when no timers are pending.
static void timer_fire_expired (srn_reactor_t *r)
 Fire every timer whose deadline has passed, pop it (in deadline order), post a zero result completion to its channel, and free the entry.
static void process_request (srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req)
static void drain_submissions (srn_reactor_t *r)
static void reactor_thread_main (void *arg)
void srn_reactor_activate (srn_reactor_t *reactor, size_t nchannels, srn_reactor_notify_fn notify)
 Bring the reactor up, allocate nchannels channels (one per worker) and start the reactor thread.

Variables

static srn_error_t io_errors []
static const pqmh_control_t timer_ctl = {.less = timer_less}

Typedef Documentation

◆ srn_timer_t

typedef struct srn_timer_t srn_timer_t

Function Documentation

◆ drain_submissions()

void drain_submissions ( srn_reactor_t * r)
static

Definition at line 447 of file reactor.c.

447 {
449 // TODO(lxsameer): Should we make this configurable via a policy or something?
450 // For example, pop from each channel once vs pop all the SQEs per channel
451 for (size_t c = 0; c < r->nchannels; c++) {
452 while (srn_spsc_ring_pop(r->channels[c].sq, &req)) {
453 process_request(r, c, &req);
454 }
455 }
456}
static void process_request(srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req)
Definition reactor.c:416
bool srn_spsc_ring_pop(srn_spsc_ring_t *r, void *out)
Consumer side.
Definition spsc_ring.c:83
srn_spsc_ring_t * sq
Definition internal.h:32
A submission, one request a fiber places on its channel.
Definition reactor.h:116
srn_reactor_io_channel_t * channels
One channel per worker, there is a 1to1 mapping between worker ids and channel ids.
Definition internal.h:56
size_t nchannels
Definition internal.h:57
Here is the call graph for this function:
Here is the caller graph for this function:

◆ process_request()

void process_request ( srn_reactor_t * r,
size_t channel,
const srn_reactor_io_request_t * req )
static

Definition at line 416 of file reactor.c.

416 {
417 switch (req->op) {
420 r, channel, (srn_reactor_io_completion_t){.fiber_data = req->fiber_data, .result = 0}
421 );
422 break;
432 // Descriptor ops go to the backend, it arms the fd and posts the completion
433 // when it is ready. The backend reports its own failures through a
434 // completion, so there is nothing to check here.
435 r->backend->submit(r, channel, req);
436 break;
437
439 timer_add(r, channel, req);
440 break;
441
443 PANIC("reactor: CANCEL is not implemented yet");
444 }
445}
static void timer_add(srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req)
Register a SLEEP: allocate a timer entry and push it onto the heap.
Definition reactor.c:352
void srn_reactor_post_completion(srn_reactor_t *reactor, size_t channel, srn_reactor_io_completion_t completion)
Post a finished operation to channel's completion queue and rouse the channel's worker through the no...
Definition reactor.c:210
@ SRN_REACTOR_IO_SLEEP
Wait out a deadline.
Definition reactor.h:69
@ SRN_REACTOR_IO_ACCEPT
Accept a pending connection on the listening socket fd, with flags (the accept4 flags).
Definition reactor.h:82
@ SRN_REACTOR_IO_READ
Read from a descriptor into buf. Uses fd, buf, len, offset.
Definition reactor.h:71
@ SRN_REACTOR_IO_SENDTO
Send buf/len on fd to the address in addr/addrlen with flags.
Definition reactor.h:95
@ SRN_REACTOR_IO_NOP
Completes immediately with a zero result.
Definition reactor.h:67
@ SRN_REACTOR_IO_CONNECT
Connect fd to the address in addr/addrlen.
Definition reactor.h:86
@ SRN_REACTOR_IO_CANCEL
Cancel one inflight operation, named by its fiber_data.
Definition reactor.h:77
@ SRN_REACTOR_IO_RECVFROM
Receive a datagram from fd into buf/len with flags.
Definition reactor.h:91
@ 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:99
@ SRN_REACTOR_IO_WRITE
Write buf to a descriptor. Uses fd, buf, len, offset.
Definition reactor.h:73
@ SRN_REACTOR_IO_POLL
Wait until fd is ready for the events requested in flags (EPOLLIN/EPOLLOUT) without performing any tr...
Definition reactor.h:108
@ SRN_REACTOR_IO_SENDMSG
Send the struct msghdr that buf points at, on fd, with flags.
Definition reactor.h:103
void(* submit)(srn_reactor_t *reactor, size_t channel, const srn_reactor_io_request_t *req)
Submit a new request to the backend.
Definition backend.h:71
A completion, the result of one submission, echoed back on the channel.
Definition reactor.h:159
srn_reactor_io_op_t op
Definition reactor.h:117
void * fiber_data
Opaque token, round tripped to the matching completion.
Definition reactor.h:136
const srn_reactor_backend_t * backend
The chosen kernel mechanism and its private state, set at activation.
Definition internal.h:47
#define PANIC(msg)
Definition utils.h:53
Here is the call graph for this function:
Here is the caller graph for this function:

◆ reactor_thread_main()

void reactor_thread_main ( void * arg)
static

Definition at line 458 of file reactor.c.

458 {
460 while (atomic_load_explicit(&r->running, memory_order_acquire)) {
462 // Wait until the soonest timer deadline (or indefinitely if none), unless a
463 // kernel event, a submission nudge, or shutdown breaks us out sooner. Then
464 // fire whatever timers came due.
467 }
468}
static void drain_submissions(srn_reactor_t *r)
Definition reactor.c:447
static int timer_next_timeout(srn_reactor_t *r)
Milliseconds until the soonest deadline (rounded up), clamped to a valid epoll_wait timeout,...
Definition reactor.c:370
static void timer_fire_expired(srn_reactor_t *r)
Fire every timer whose deadline has passed, pop it (in deadline order), post a zero result completion...
Definition reactor.c:395
void * arg
void(* wait)(srn_reactor_t *reactor, int timeout_ms)
Wait for readiness and complete any ready operations.
Definition backend.h:75
_Atomic bool running
Definition internal.h:66
Here is the call graph for this function:
Here is the caller graph for this function:

◆ srn_reactor_activate()

void srn_reactor_activate ( srn_reactor_t * reactor,
size_t nchannels,
srn_reactor_notify_fn notify )

Bring the reactor up, allocate nchannels channels (one per worker) and start the reactor thread.

notify is called with the scheduler and the channel id whenever a channel gains completions.

Definition at line 470 of file reactor.c.

470 {
471
472 PANIC_IF_NULL(reactor);
473 PANIC_IF(nchannels == 0, "reactor: at least one channel is required");
474
475 reactor->notify = notify;
476 reactor->nchannels = nchannels;
477
478 // The ring size is a runtime knob; both rings of every channel use it.
479 // Validate the knobs once here, so everything sized from them downstream
480 // (rings, the backend op pool, the reap stack buffers) is sound, a zero
481 // batch is a zero length VLA, and an oversized magnitude wraps the pool
482 // arithmetic.
483 const size_t ring_mag = reactor->ctx->engine->config.reactor.ring_magnitude;
484 PANIC_IF(ring_mag < 1 || ring_mag > 16, "reactor: ring_magnitude must be within 1..16");
485 reactor->ring_capacity = (size_t)1 << ring_mag;
486
487 const size_t batch = reactor->ctx->engine->config.reactor.reap_batch;
488 PANIC_IF(
489 batch < 1 || batch > reactor->ring_capacity,
490 "reactor: reap_batch must be within 1..ring capacity"
491 );
492
493 reactor->channels = ALLOCN(reactor->ctx, srn_reactor_io_channel_t, nchannels);
494 PANIC_IF_NULL(reactor->channels);
495
496 for (size_t c = 0; c < nchannels; c++) {
497 reactor->channels[c].inflight = 0;
498 // Setup the SQ of the channel
499 reactor->channels[c].sq =
500 srn_spsc_ring_make(reactor->ctx, ring_mag, sizeof(srn_reactor_io_request_t));
501 PANIC_IF_NULL(reactor->channels[c].sq);
502 PANIC_IF(HAS_ERROR(*reactor->channels[c].sq), reactor->channels[c].sq->maybe_error->msg);
503
504 // Setup the CQ of the channel
505 reactor->channels[c].cq =
506 srn_spsc_ring_make(reactor->ctx, ring_mag, sizeof(srn_reactor_io_completion_t));
507 PANIC_IF_NULL(reactor->channels[c].cq);
508 PANIC_IF(HAS_ERROR(*reactor->channels[c].cq), reactor->channels[c].cq->maybe_error->msg);
509 }
510
511 // The timer heap is reactor thread private. Create it before the thread
512 // spawns.
513 reactor->timers = pqmh_make(reactor->ctx, &timer_ctl, 0);
514
515 // Bring the backend up now that nchannels is known, its op pool is sized from
516 // the channel count, so it cannot be initialized before activation.
517 srn_error_t *err = reactor->backend->init(reactor);
518 PANIC_IF(err != nullptr, err->msg);
519
520 atomic_store_explicit(&reactor->running, true, memory_order_release);
521 PANIC_IF(
523 "reactor: failed to spawn the reactor thread"
524 );
525}
#define ALLOCN(ctx, T, N)
Definition context.h:96
#define HAS_ERROR(x)
True when x – a value carrying an ERROR_HEADER – has an error attached.
Definition errors.h:70
pqmh_t pqmh_make(srn_context_t *ctx, const pqmh_control_t *ctl, size_t initial_cap)
Create an empty queue ordered by ctl, with room for initial_cap elements (a small default is used whe...
Definition pqmh.c:81
static const pqmh_control_t timer_ctl
Definition reactor.c:347
static void reactor_thread_main(void *arg)
Definition reactor.c:458
srn_spsc_ring_t * srn_spsc_ring_make(srn_context_t *ctx, size_t magnitude, size_t elem_size)
Set up a ring by allocating the buf to be the size of cap * elem_size.
Definition spsc_ring.c:27
srn_reactor_config_t reactor
srn_engine_t * engine
Long term state of the compiler.
Definition context.h:49
srn_configuration_t config
The runtime's tunable knobs, the single source for every configurable value (see configuration....
Definition engine.h:65
A runtime error, a tag classifying the failure and a human-readable message.
Definition errors.h:173
char * msg
Definition errors.h:176
srn_error_t *(* init)(srn_reactor_t *reactor)
Initialize the backend in the given reactor.
Definition backend.h:64
size_t reap_batch
Most completions a worker drains from its channel in a single pass.
size_t ring_magnitude
Magnitude of each channel's SQ/CQ rings, capacity is 1 << magnitude.
A channel, the per-worker submission and completion ring pair.
Definition internal.h:30
srn_spsc_ring_t * cq
Definition internal.h:34
size_t inflight
Operations submitted on this channel and not yet consumed.
Definition internal.h:40
srn_thread_t thread
The reactor thread and its run flag.
Definition internal.h:65
pqmh_t timers
Pending timers (like SLEEP, and IO deadlines) as a min-heap keyed by absolute monotonic deadline.
Definition internal.h:77
srn_reactor_notify_fn notify
Wakes the worker that owns a channel a completion landed on.
Definition internal.h:52
srn_context_t * ctx
Definition internal.h:44
size_t ring_capacity
Capacity of every ring (1 << ring_magnitude), fixed at activation.
Definition internal.h:61
@ SRN_THREAD_OK
Definition thread.h:62
srn_thread_status_t srn_thread_spawn(srn_thread_t *t, void(*fn)(void *), void *arg)
Run fn(arg) on a new OS thread.
#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:

◆ srn_reactor_channel_has_completions()

bool srn_reactor_channel_has_completions ( srn_reactor_t * reactor,
size_t channel )
nodiscard

Whether channel's completion queue has unconsumed completions.

The worker checks this under the scheduler lock before parking, so a completion landing just before it sleeps is not missed.

Definition at line 315 of file reactor.c.

315 {
316 PANIC_IF_NULL(reactor);
317 PANIC_IF(channel >= reactor->nchannels, "reactor: channel id out of range");
318 return (!srn_spsc_ring_empty(reactor->channels[channel].cq)) != 0;
319}
bool srn_spsc_ring_empty(const srn_spsc_ring_t *r)
Whether the ring currently holds no elements.
Definition spsc_ring.c:103
Here is the call graph for this function:
Here is the caller graph for this function:

◆ srn_reactor_errors_static()

srn_error_t * srn_reactor_errors_static ( srn_error_tag_t tag)

Return the shared, immutable error for an IO failure tag.

These live in static storage, not a context, so a completion or result can point one at maybe_error with no allocation and no lifetime concern, and it is safe to share across threads. A backend maps its native code (via srn_errno_to_tag on POSIX) to a tag, then to one of these. Any tag without a dedicated entry yields the UNKNOWN_IO_ERROR error.

Because they are shared they are never written to. In particular they are not trace targets, srn_error_trace_push, PUSH_TRACE and PUSH_AND_MOVE would allocate frames in one caller's context and record unrelated call sites into every later failure with the same tag. A caller that wants a trace on an IO failure builds its own error from the static one with ERR_FROM and traces that.

Definition at line 194 of file reactor.c.

194 {
195
196 size_t count = sizeof(io_errors) / sizeof(io_errors[0]);
197
198 for (size_t i = 0; i < count; i++) {
199 if (io_errors[i].tag == tag) {
200 return &io_errors[i];
201 }
202 }
203 // the UNKNOWN_IO_ERROR fallback
204 return &io_errors[count - 1];
205}
static srn_error_t io_errors[]
Definition reactor.c:99
Here is the caller graph for this function:

◆ srn_reactor_idle()

bool srn_reactor_idle ( srn_reactor_t * reactor)
nodiscard

Whether the reactor has no operations in flight.

The scheduler folds this into quiescence, a pool with parked workers and an empty run queue is only truly done when the reactor is idle too.

Definition at line 244 of file reactor.c.

244 {
245 PANIC_IF_NULL(reactor);
246 return atomic_load(&reactor->inflight) == 0;
247}
_Atomic size_t inflight
Operations submitted but not yet consumed, across all channels.
Definition internal.h:71
Here is the caller graph for this function:

◆ srn_reactor_init()

srn_reactor_t * srn_reactor_init ( srn_engine_t * engine)
nodiscard

Allocate the IO reactor from the engine.

Definition at line 34 of file reactor.c.

34 {
35 PANIC_IF_NULL(engine);
36
37 srn_reactor_t *reactor =
39 PANIC_IF_NULL(reactor);
40
41 // This will allocate a root context which will have its own block of memory
42 reactor->ctx = srn_context_make(engine);
43 PANIC_IF_NULL(reactor->ctx);
44
45 switch (reactor->ctx->engine->config.reactor.backend) {
46#ifdef __linux__
49 break;
50#endif
51 default:
52 PANIC("IO backend is not implemented yet");
53 }
54
55 reactor->backend_state = nullptr;
56 reactor->channels = nullptr;
57 reactor->nchannels = 0;
58 reactor->notify = nullptr;
59
60 atomic_init(&reactor->running, false);
61 atomic_init(&reactor->inflight, 0);
62
63 return reactor;
64}
const srn_reactor_backend_t srn_reactor_backend_epoll
@ SRN_REACTOR_EPOLL
srn_context_t * srn_context_make(srn_engine_t *engine)
Make an empty context, by allocating a new memory block.
Definition context.c:39
void * srn_mm_immortal_allocate_aligned(srn_mm_t *mm, size_t size, size_t alignment)
Allocate memory on the importal block which will never gets freed.
Definition default.c:431
srn_mm_t * mm
Memory manager.
Definition engine.h:68
srn_reactor_backend_e backend
What IO backend to use.
void * backend_state
Definition internal.h:48
Here is the call graph for this function:
Here is the caller graph for this function:

◆ srn_reactor_op_done()

void srn_reactor_op_done ( srn_reactor_t * reactor,
size_t channel )

Drop channel's and the reactor's in-flight counts by one.

The consumer of a completion calls this after it has finished acting on it (for fibers, after the waiting fiber is readied), so that "in flight" reaches zero only once every result has been both produced and handled. Dropping the channel count is also what re-opens the channel's admission window (see srn_reactor_submit).

Definition at line 236 of file reactor.c.

236 {
237 PANIC_IF_NULL(reactor);
238 PANIC_IF(channel >= reactor->nchannels, "reactor: channel id out of range");
239 reactor->channels[channel].inflight--;
240 atomic_fetch_sub(&reactor->inflight, 1);
241 REACTOR_TRACEPOINT(reactor_op_done, channel);
242}
#define REACTOR_TRACEPOINT(...)
Definition reactor.h:58
Here is the caller graph for this function:

◆ srn_reactor_post_completion()

void srn_reactor_post_completion ( srn_reactor_t * reactor,
size_t channel,
srn_reactor_io_completion_t completion )

Post a finished operation to channel's completion queue and rouse the channel's worker through the notify mechanism.

Runs on the reactor thread, called by the core for ops it completes itself (such as NOP) and by a backend when a descriptor operation finishes. It does NOT drop the in-flight count: that is the consumer's job via srn_reactor_op_done, so an op counts as gone only once the waiting fiber has been readied, never in the gap between.

Definition at line 210 of file reactor.c.

212 {
213 srn_reactor_io_channel_t *ch = &reactor->channels[channel];
214 // Admission control bounds a channel's in-flight ops at the ring capacity,
215 // and the op being posted is one of them and not yet in the CQ, so the CQ
216 // always has a slot. A full CQ here means the accounting is broken; a spin
217 // in its place would stall the whole reactor behind one busy worker.
218 PANIC_IF(
219 !srn_spsc_ring_push(ch->cq, &completion),
220 "reactor: completion queue full despite admission control; this is a bug"
221 );
222
224 reactor_complete, channel, completion.result, completion.maybe_error != nullptr,
225 completion.fiber_data
226 );
227
228 // The in-flight count is dropped by the consumer through srn_reactor_op_done,
229 // not here. A fiber's waiter must be readied before the op counts as gone, or
230 // quiescence could fire in the gap between posting and readying.
231 if (reactor->notify != nullptr) {
232 reactor->notify(reactor->ctx->engine->scheduler, channel);
233 }
234}
bool srn_spsc_ring_push(srn_spsc_ring_t *r, const void *elem)
Producer side.
Definition spsc_ring.c:61
srn_scheduler_t * scheduler
The fiber scheduler, that is the entry point of the fiber subsystem.
Definition engine.h:78
void * fiber_data
The fiber_data of the submission this completes.
Definition reactor.h:162
size_t result
The result of the operation on success, bytes transferred for the transfers, the accepted descriptor ...
Definition reactor.h:166
Here is the call graph for this function:
Here is the caller graph for this function:

◆ srn_reactor_reap()

size_t srn_reactor_reap ( srn_reactor_t * reactor,
size_t channel,
srn_reactor_io_completion_t * out,
size_t max )

Drain up to max completions from channel's completion queue into out, returning how many were taken.

Definition at line 298 of file reactor.c.

300 {
301 PANIC_IF_NULL(reactor);
302 PANIC_IF_NULL(out);
303 PANIC_IF(channel >= reactor->nchannels, "reactor: channel id out of range");
304
305 srn_reactor_io_channel_t *ch = &reactor->channels[channel];
306 size_t n = 0;
307
308 while (n < max && (int)srn_spsc_ring_pop(ch->cq, &out[n])) {
309 n++;
310 }
311
312 return n;
313}
int n
Definition acutest.h:525
Here is the call graph for this function:
Here is the caller graph for this function:

◆ srn_reactor_shutdown()

void srn_reactor_shutdown ( srn_reactor_t * reactor)

Tear the reactor down, stop and join the reactor thread and release its channels.

Definition at line 66 of file reactor.c.

66 {
67 // Never activated, or already shut down. Nothing to release.
68 RETURN_IF_NULL(reactor)
69 if (!atomic_load_explicit(&reactor->running, memory_order_acquire)) {
70 return;
71 }
72
73 atomic_store_explicit(&reactor->running, false, memory_order_release);
74 // Break the reactor out of its wait so it observes
75 // `running == false` and exits, rather than blocking until some descriptor
76 // happens to fire.
77 reactor->backend->wake(reactor);
78 srn_thread_join(&reactor->thread);
79 reactor->backend->shutdown(reactor);
80
81 // The reactor thread is joined, so the timer heap is ours now. Free any
82 // timers that never fired (their fibers are reaped by the scheduler), then
83 // the heap.
84 void *out = nullptr;
85 while (pqmh_pop(reactor->ctx, &reactor->timers, &out)) {
86 srn_mm_free(reactor->ctx->engine->mm, out);
87 }
88 pqmh_free(reactor->ctx, &reactor->timers);
89}
void srn_mm_free(srn_mm_t *mm, void *ptr)
Release a pointer previously returned by srn_mm_malloc or srn_mm_reallocate.
Definition default.c:168
void pqmh_free(srn_context_t *ctx, pqmh_t *pq)
Release the backing array and reset the queue to empty.
Definition pqmh.c:142
bool pqmh_pop(srn_context_t *ctx, pqmh_t *pq, void **out)
Remove the highest priority element into *out and return true, or return false (leaving *out untouche...
Definition pqmh.c:111
void(* shutdown)(srn_reactor_t *reactor)
Clean up after the backend.
Definition backend.h:67
void(* wake)(srn_reactor_t *reactor)
Definition backend.h:77
srn_thread_status_t srn_thread_join(srn_thread_t *t)
Block until the thread started for t returns.
#define RETURN_IF_NULL(ptr)
Definition utils.h:67
Here is the call graph for this function:
Here is the caller graph for this function:

◆ srn_reactor_submit()

bool srn_reactor_submit ( srn_reactor_t * reactor,
size_t channel,
const srn_reactor_io_request_t * request )
nodiscard

Place a request on channel's submission queue and rouse the reactor.

Returns false without submitting when the channel already has a full ring's worth of operations in flight. The condition is transient, it clears as the channel's worker consumes completions, so the caller can fail the operation back with CHANNEL_BUSY and let the fiber retry.

Definition at line 249 of file reactor.c.

251 {
252 PANIC_IF_NULL(reactor);
253 PANIC_IF_NULL(request);
254 PANIC_IF(channel >= reactor->nchannels, "reactor: channel id out of range");
255
256 // A clean panic beats what follows otherwise, the backend state is gone
257 // after shutdown, so the wake below would dereference null.
258 PANIC_IF(
259 !atomic_load_explicit(&reactor->running, memory_order_acquire),
260 "reactor: submit on a reactor that is not running"
261 );
262
263 srn_reactor_io_channel_t *ch = &reactor->channels[channel];
264
265 // Admission control. A channel holds at most one ring's worth of ops in
266 // flight, counted from submit until the completion is consumed. The cap is
267 // what makes ring overflow impossible, the SQ holds at most the admitted
268 // ops, and the CQ holds at most the admitted ops minus the one being
269 // posted, so neither the push below nor the post on the reactor thread can
270 // find its ring full. A rejected submit is failed back to the caller as a
271 // transient condition; it clears as the worker consumes completions.
272 if (ch->inflight >= reactor->ring_capacity) {
273 REACTOR_TRACEPOINT(reactor_submit_rejected, channel, ch->inflight);
274 return false;
275 }
276 ch->inflight++;
277
278 // Count the op as in flight BEFORE the request becomes visible to the reactor
279 // thread. If the bump came after the push, the reactor could complete the op
280 // and the consumer could call srn_reactor_op_done before the bump ran --
281 // underflowing the counter and breaking quiescence.
282 atomic_fetch_add(&reactor->inflight, 1);
283
284 // The SQ is single-producer, the worker that owns this channel.
285 PANIC_IF(
286 !srn_spsc_ring_push(ch->sq, request),
287 "reactor: submission queue full despite admission control; this is a bug"
288 );
289
290 REACTOR_TRACEPOINT(reactor_submit, channel, (int)request->op, request->fd, request->fiber_data);
291
292 // Rouse the reactor so it drains the new request at once instead of waiting
293 // out its idle timeout.
294 reactor->backend->wake(reactor);
295 return true;
296}
int fd
The descriptor the operation targets.
Definition reactor.h:120
Here is the call graph for this function:
Here is the caller graph for this function:

◆ timer_add()

void timer_add ( srn_reactor_t * r,
size_t channel,
const srn_reactor_io_request_t * req )
static

Register a SLEEP: allocate a timer entry and push it onto the heap.

Definition at line 352 of file reactor.c.

352 {
353 assert(r != nullptr);
354
355 srn_timer_t *t = srn_mm_malloc(r->ctx->engine->mm, sizeof(srn_timer_t));
356 PANIC_IF_NULL(t);
357
358 t->deadline_ns = req->deadline_ns;
359 t->channel = channel;
360 t->fiber_data = req->fiber_data;
361 REACTOR_TRACEPOINT(reactor_timer_armed, channel, req->deadline_ns);
362 pqmh_push(r->ctx, &r->timers, t);
363}
void * srn_mm_malloc(srn_mm_t *mm, size_t size)
Generic allocations that do not participate in the block based pools.
Definition default.c:155
void pqmh_push(srn_context_t *ctx, pqmh_t *pq, void *elem)
Insert elem, growing the backing array (doubling) if it is full.
Definition pqmh.c:92
uint64_t deadline_ns
The wake deadline for SRN_REACTOR_IO_SLEEP, on the monotonic clock, in nanoseconds.
Definition reactor.h:131
void * fiber_data
Round tripped to the waiting fiber.
Definition reactor.c:336
size_t channel
Channel the completion is posted to.
Definition reactor.c:334
uint64_t deadline_ns
Absolute CLOCK_MONOTONIC deadline.
Definition reactor.c:332
Here is the call graph for this function:
Here is the caller graph for this function:

◆ timer_fire_expired()

void timer_fire_expired ( srn_reactor_t * r)
static

Fire every timer whose deadline has passed, pop it (in deadline order), post a zero result completion to its channel, and free the entry.

Definition at line 395 of file reactor.c.

395 {
396 assert(r != nullptr);
397
398 uint64_t now = srn_now_ns();
399 const srn_timer_t *top;
400
401 while ((top = pqmh_peek(&r->timers)) != nullptr && top->deadline_ns <= now) {
402 void *out = nullptr;
403 UNUSED(pqmh_pop(r->ctx, &r->timers, &out));
404
405 srn_timer_t *t = (srn_timer_t *)out;
406 REACTOR_TRACEPOINT(reactor_timer_fired, t->channel, t->deadline_ns);
407
409 r, t->channel, (srn_reactor_io_completion_t){.fiber_data = t->fiber_data, .result = 0}
410 );
411 srn_mm_free(r->ctx->engine->mm, t);
412 }
413}
void * pqmh_peek(const pqmh_t *pq)
The highest priority element without removing it, or NULL when the queue is empty.
Definition pqmh.c:132
uint64_t srn_now_ns(void)
The current time on the monotonic clock, in nanoseconds.
Definition utils.c:45
#define UNUSED(x)
Definition utils.h:45
Here is the call graph for this function:
Here is the caller graph for this function:

◆ timer_less()

bool timer_less ( srn_context_t * ctx,
const void * a,
const void * b )
static

Order timers by deadline so the soonest sits at the heap root.

Definition at line 342 of file reactor.c.

342 {
343 UNUSED(ctx);
344 return ((const srn_timer_t *)a)->deadline_ns < ((const srn_timer_t *)b)->deadline_ns;
345}

◆ timer_next_timeout()

int timer_next_timeout ( srn_reactor_t * r)
static

Milliseconds until the soonest deadline (rounded up), clamped to a valid epoll_wait timeout, 0 if a timer is already due, INT_MAX for a far-off deadline, and -1 (block indefinitely) when no timers are pending.

Definition at line 370 of file reactor.c.

370 {
371 assert(r != nullptr);
372
373 const srn_timer_t *t = pqmh_peek(&r->timers);
374
375 if (t == nullptr) {
376 return -1;
377 }
378
379 uint64_t now = srn_now_ns();
380 if (t->deadline_ns <= now) {
381 return 0;
382 }
383
384 // Convert the ns remaining to the ms epoll_wait timeout, rounding UP (add
385 // NS_PER_MS-1) so a sub-millisecond remainder waits 1ms rather than 0 -- we
386 // must never wake early and busy-spin firing nothing.
387 uint64_t ms = ((t->deadline_ns - now) + (SRN_NS_PER_MS - 1)) / SRN_NS_PER_MS;
388 return (ms > (uint64_t)INT_MAX) ? INT_MAX : (int)ms;
389}
#define SRN_NS_PER_MS
nanoseconds in one millisecond
Definition utils.h:308
Here is the call graph for this function:
Here is the caller graph for this function:

Variable Documentation

◆ io_errors

srn_error_t io_errors[]
static

Definition at line 99 of file reactor.c.

99 {
100 {.layer = SRN_ERR_RUNTIME_LAYER,
101 .tag = CONNECTION_REFUSED,
102 .msg = (char *)"connection refused",
103 .as = {.frames = nullptr}},
104 {.layer = SRN_ERR_RUNTIME_LAYER,
105 .tag = CONNECTION_RESET,
106 .msg = (char *)"connection reset by peer",
107 .as = {.frames = nullptr}},
108 {.layer = SRN_ERR_RUNTIME_LAYER,
109 .tag = CONNECTION_ABORTED,
110 .msg = (char *)"connection aborted",
111 .as = {.frames = nullptr}},
112 {.layer = SRN_ERR_RUNTIME_LAYER,
113 .tag = BROKEN_PIPE,
114 .msg = (char *)"broken pipe",
115 .as = {.frames = nullptr}},
116 {.layer = SRN_ERR_RUNTIME_LAYER,
117 .tag = BAD_DESCRIPTOR,
118 .msg = (char *)"bad file descriptor",
119 .as = {.frames = nullptr}},
120 {.layer = SRN_ERR_RUNTIME_LAYER,
121 .tag = TIMED_OUT,
122 .msg = (char *)"operation timed out",
123 .as = {.frames = nullptr}},
124 {.layer = SRN_ERR_RUNTIME_LAYER,
125 .tag = CANCELED,
126 .msg = (char *)"operation canceled",
127 .as = {.frames = nullptr}},
128 {.layer = SRN_ERR_RUNTIME_LAYER,
129 .tag = NOT_CONNECTED,
130 .msg = (char *)"socket is not connected",
131 .as = {.frames = nullptr}},
132 {.layer = SRN_ERR_RUNTIME_LAYER,
133 .tag = ADDRESS_IN_USE,
134 .msg = (char *)"address already in use",
135 .as = {.frames = nullptr}},
136 {.layer = SRN_ERR_RUNTIME_LAYER,
137 .tag = HOST_UNREACHABLE,
138 .msg = (char *)"host is unreachable",
139 .as = {.frames = nullptr}},
140 {.layer = SRN_ERR_RUNTIME_LAYER,
141 .tag = NETWORK_UNREACHABLE,
142 .msg = (char *)"network is unreachable",
143 .as = {.frames = nullptr}},
144 {.layer = SRN_ERR_RUNTIME_LAYER,
145 .tag = MESSAGE_TOO_LONG,
146 .msg = (char *)"message too long",
147 .as = {.frames = nullptr}},
148 {.layer = SRN_ERR_RUNTIME_LAYER,
149 .tag = TOO_MANY_OPEN_FILES,
150 .msg = (char *)"too many open files",
151 .as = {.frames = nullptr}},
152 {.layer = SRN_ERR_RUNTIME_LAYER,
153 .tag = PERMISSION_DENIED,
154 .msg = (char *)"permission denied",
155 .as = {.frames = nullptr}},
156 {.layer = SRN_ERR_RUNTIME_LAYER,
157 .tag = OUT_OF_MEMORY,
158 .msg = (char *)"out of memory",
159 .as = {.frames = nullptr}},
160 {.layer = SRN_ERR_RUNTIME_LAYER,
161 .tag = NOT_FOUND,
162 .msg = (char *)"no such file or directory",
163 .as = {.frames = nullptr}},
164 {.layer = SRN_ERR_RUNTIME_LAYER,
165 .tag = ALREADY_EXISTS,
166 .msg = (char *)"file already exists",
167 .as = {.frames = nullptr}},
168 {.layer = SRN_ERR_RUNTIME_LAYER,
169 .tag = IS_A_DIRECTORY,
170 .msg = (char *)"is a directory",
171 .as = {.frames = nullptr}},
172 {.layer = SRN_ERR_RUNTIME_LAYER,
173 .tag = NOT_A_DIRECTORY,
174 .msg = (char *)"not a directory",
175 .as = {.frames = nullptr}},
176 {.layer = SRN_ERR_RUNTIME_LAYER,
177 .tag = NO_SPACE,
178 .msg = (char *)"no space left on device",
179 .as = {.frames = nullptr}},
180 {.layer = SRN_ERR_RUNTIME_LAYER,
181 .tag = INVALID_ARGUMENT,
182 .msg = (char *)"invalid argument",
183 .as = {.frames = nullptr}},
184 {.layer = SRN_ERR_RUNTIME_LAYER,
185 .tag = CHANNEL_BUSY,
186 .msg = (char *)"reactor channel is at capacity",
187 .as = {.frames = nullptr}},
188 {.layer = SRN_ERR_RUNTIME_LAYER,
189 .tag = UNKNOWN_IO_ERROR,
190 .msg = (char *)"unknown IO error",
191 .as = {.frames = nullptr}},
192};
@ BROKEN_PIPE
Definition errors.h:121
@ HOST_UNREACHABLE
Definition errors.h:127
@ NETWORK_UNREACHABLE
Definition errors.h:128
@ IS_A_DIRECTORY
Definition errors.h:135
@ INVALID_ARGUMENT
The request itself is malformed (an empty POLL interest mask, for example), so retrying it unchanged ...
Definition errors.h:140
@ CHANNEL_BUSY
The channel already has a full ring's worth of operations in flight.
Definition errors.h:144
@ PERMISSION_DENIED
Definition errors.h:131
@ CONNECTION_RESET
Definition errors.h:119
@ UNKNOWN_IO_ERROR
Definition errors.h:147
@ TOO_MANY_OPEN_FILES
Definition errors.h:130
@ NOT_CONNECTED
Definition errors.h:125
@ CONNECTION_REFUSED
Definition errors.h:118
@ CONNECTION_ABORTED
Definition errors.h:120
@ ADDRESS_IN_USE
Definition errors.h:126
@ BAD_DESCRIPTOR
Definition errors.h:122
@ OUT_OF_MEMORY
Definition errors.h:132
@ NOT_FOUND
Definition errors.h:133
@ CANCELED
Definition errors.h:124
@ NOT_A_DIRECTORY
Definition errors.h:136
@ TIMED_OUT
Definition errors.h:123
@ MESSAGE_TOO_LONG
Definition errors.h:129
@ ALREADY_EXISTS
Definition errors.h:134
@ NO_SPACE
Definition errors.h:137
@ SRN_ERR_RUNTIME_LAYER
Definition errors.h:151

◆ timer_ctl

const pqmh_control_t timer_ctl = {.less = timer_less}
static

Definition at line 347 of file reactor.c.

347{.less = timer_less};
static bool timer_less(srn_context_t *ctx, const void *a, const void *b)
Order timers by deadline so the soonest sits at the heap root.
Definition reactor.c:342