Serene Runtime 1.0.0-dev
C runtime for the Serene programming language
Loading...
Searching...
No Matches
epoll.c
Go to the documentation of this file.
1/* -*- C -*-
2 * Serene programming language
3 * Copyright (C) 2019-2026 Sameer Rahmani <[email protected]>
4 *
5 * This library is free software: you can redistribute it and/or modify
6 * it under the terms of the GNU Lesser General Public License as published by
7 * the Free Software Foundation, either version 3 of the License, or
8 * (at your option) any later version.
9 *
10 * This library is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU Lesser General Public License for more details.
14 *
15 * You should have received a copy of the GNU Lesser General Public License
16 * along with this library. If not, see <https://www.gnu.org/licenses/>.
17 */
18
19/** @file
20## The epoll backend. Linux only.
21
22It arms a descriptor, and when epoll reports it ready it performs the transfer
23itself and posts the finished result. A self-wake `eventfd`, registered in the
24epoll set, lets a submission on any thread break the reactor thread out of
25`epoll_wait`.
26
27## Notes:
28- Operations (`epoll_op_t`) live in a fixed pool sized from the reactor ring
29magnitude, with an intrusive free list. No need for growth as we're limited by
30the magnitude of the reactor.
31
32- Several operations can share one descriptor (two fibers reading and writing
33the same socket, many fibers accepting on one listener). They are grouped in a
34per-fd list (`epoll_op_list_t`) and the fd is registered once with the union of
35their interests. `epoll_event.data.ptr` carries the list, so a readiness event
36resolves to the fd's ops with no search. The wait loop then walks the list and
37completes those whose direction is ready. The wake fd is distinguished by a NULL
38data pointer (a real list pointer is never NULL).
39
40- The fd -> list mapping is a hashmap (check the serene/rt/impl/hashmap.h). The
41fd integer is stuffed straight into the key's `data` field (it fits in a
42pointer), so the map needs no separate key storage; a custom control hashes and
43compares `data` by value instead of dereferencing it.
44*/
45
46#include <assert.h>
47
48#include "serene/rt/engine.h"
49#include "serene/rt/errors.h"
51
52#ifdef __linux__
53
54# include <errno.h>
55# include <stdint.h>
56# include <sys/epoll.h>
57# include <sys/eventfd.h>
58# include <sys/socket.h>
59# include <unistd.h>
60
61# include "../backend.h"
62# include "../internal.h"
64# include "serene/utils.h"
65
66// Since use fd value as (void *) for the map keys. We don't use them as
67// pointers so we're safe really.
68// NOLINTBEGIN(performance-no-int-to-ptr)
69typedef struct epoll_op_t {
70 /// Whether this slot holds a live operation. The discriminator between the
71 /// slot's two lives, read op fields only when true, `next_free` only when
72 /// false.
73 bool active;
74 /// While this slot is free, the index of the next free slot (SIZE_MAX ends
75 /// the chain). Meaningless while active.
76 size_t next_free;
77
78 int fd;
79 /// Next op on the same fd, or NULL. The per-fd list is threaded through here.
80 struct epoll_op_t *fd_next;
81
82 /// The kind of transfer. The epoll interest (EPOLLIN/EPOLLOUT) is derived
83 /// from it by `op_events`, so it is not stored twice.
85 /// Where the completion is posted back.
86 size_t channel;
87
88 void *fiber_data;
89 void *buf;
90 size_t len;
91 int64_t offset;
92
93 /// Socket address buffer, the peer to use for CONNECT/SENDTO, the peer to
94 /// fill in for ACCEPT/RECVFROM, or NULL. A `struct sockaddr *` held as `void
95 /// *`.
96 void *addr;
97 /// Length of `addr`, the exact length for CONNECT/SENDTO, the buffer capacity
98 /// for ACCEPT/RECVFROM.
99 uint32_t addrlen;
100 /// Operation flags, the `MSG_*` flags for the datagram and message ops, the
101 /// `accept4` flags for ACCEPT, and the requested readiness for POLL.
102 int flags;
103} epoll_op_t;
104
105typedef struct epoll_op_list_t {
106 /// The map key for this fd. `key.data` holds the fd value itself (see the
107 /// custom control), so it doubles as the fd's stable storage. The map keeps a
108 /// pointer to it, so it must live as long as the entry, it does, inside this
109 /// arena-allocated list.
110 hmap_key_t key;
111 /// Tombstone. The map is never deleted from (that keeps allocation bounded by
112 /// distinct fd integers). Instead, when the last op on an fd finishes, the fd
113 /// is removed from epoll and the list is marked dormant. A later op on the
114 /// same fd integer revives this same list. A dormant list (`deleted == true`)
115 /// is not registered in epoll and has no ops. And since POSIX specifies that
116 /// when allocating a FD, the smallest available FD should be chosen. So
117 /// We will end up recycling the deleted slots quite often.
118 bool deleted;
119 /// Head of the ops waiting on this fd
120 epoll_op_t *ops;
121} epoll_op_list_t;
122
123typedef struct epoll_state_t {
124 /// The main epoll descriptor.
125 int epfd;
126 /// Self-wake eventfd, registered in `epfd` with EPOLLIN. The reactor thread
127 /// sleeps in `epoll_wait`, which the kernel only breaks for IO readiness --
128 /// but two non-IO events must also break it, a worker submitting a new
129 /// request (so the reactor drains the SQ and arms the new fd) and shutdown
130 /// (so the thread observes `running == false` and exits).
131 /// `epoll_backend_wake` writes this fd from any thread, making it readable so
132 /// `epoll_wait` returns at once. The wait loop recognises it, drains the
133 /// counter (`drain_wake`) to silence the level-triggered fd, and treats it as
134 /// a no-op event. This is the submit side nudge; without it the reactor would
135 /// block until some already armed descriptor fired and never notice new work.
136 int wake_fd;
137 /// The pool of armed, but not yet completed operations. epoll only reports
138 /// readiness, not the request, so this is the backend's memory of "what to do
139 /// when each fd fires". Its size is fixed at init from
140 /// `reactor.ring_magnitude` (the per-channel ring capacity) times the channel
141 /// count, which is the most operations that can be in flight, so the pool
142 /// never needs to grow.
143 epoll_op_t *ops;
144 /// Number of slots in `ops`.
145 size_t total_slots;
146 /// Head of the intrusive free list, the first free slot, or SIZE_MAX if the
147 /// pool is full.
148 size_t free_head;
149 /// fd -> `epoll_op_list_t`. Grouped ops live here so a readiness event finds
150 /// every op on its fd. Persistent and arena-backed; reused via tombstones, so
151 /// it grows only with distinct fd integers.
152 hmap_t fdmap;
153} epoll_state_t;
154
155// -----------------------------------------------------------------------------
156// fd -> op-list map
157// -----------------------------------------------------------------------------
158// The key is the fd integer stored in `hmap_key_t.data` by value, not by
159// pointer. The default control would dereference `data` to hash and compare it;
160// this control treats `data` as the value, so no separate key storage is
161// needed.
162
163/**
164 * Compare two fd keys. The fd lives in `data` itself, so identity of the value
165 * is identity of the key.
166 */
167static bool fdmap_cmp(const hmap_t *hmap, const hmap_key_t *a, const hmap_key_t *b) {
168 UNUSED(hmap);
169 assert(a != nullptr);
170 assert(b != nullptr);
171 // fd 0 is a valid descriptor and stores as a null `data`, so there is no
172 // non-null invariant to assert on the key payload.
173 return a->data == b->data;
174}
175
176/**
177 * Hash an fd key, the fd is its own hash. Distinct fds give distinct hashes, so
178 * there are no collision nodes, and small sequential fds fan out evenly across
179 * the trie's low-order 5-bit chunks. Running them through a scrambling hash
180 * would only add cost and could manufacture collisions the raw value avoids.
181 */
182static hmap_hash_t fdmap_hash(const hmap_t *hmap, const hmap_key_t *k) {
183 UNUSED(hmap);
184 assert(k != nullptr);
185 // fd 0 is a valid descriptor and stores as a null `data`; its hash is 0.
186 return (hmap_hash_t)(intptr_t)k->data;
187}
188
189/**
190 * Copy an fd key into the map's context. The fd lives in `data` by value, so
191 * cloning the struct is enough; the default byte copier would dereference it
192 * as a pointer.
193 */
194static hmap_key_t *fdmap_copy_key(const hmap_t *hmap, const hmap_key_t *k) {
195 hmap_key_t *copy = ALLOC(hmap->map_ctx, hmap_key_t);
196 *copy = *k;
197 return copy;
198}
199
200static hmap_control_t fdmap_control = {
201 .cmp = fdmap_cmp,
202 .hash = fdmap_hash,
203 .insert = hmap_insert_ctl,
204 .lookup = hmap_lookup_ctl,
205 .copy_key = fdmap_copy_key,
206};
207
208/**
209 * Find the op list for `fd`, creating a dormant one on first sight. A freshly
210 * created list is `deleted` (not yet registered in epoll); the first op to arm
211 * it revives it. The fd is stored inside the list's key, which is what the map
212 * points at.
213 */
214static epoll_op_list_t *fd_list(srn_context_t *ctx, epoll_state_t *st, int fd) {
215 assert(ctx != nullptr);
216 assert(st != nullptr);
217
218 hmap_key_t probe = {.data = (void *)(intptr_t)fd, .len = sizeof(int)};
219
220 epoll_op_list_t *l = hmap_lookup_ctl(&fdmap_control, &st->fdmap, &probe, nullptr);
221
222 if (l != nullptr) {
223 return l;
224 }
225
226 // Create the entry as it is not there
227 l = ALLOC(ctx, epoll_op_list_t);
228 l->key = (hmap_key_t){.data = (void *)(intptr_t)fd, .len = sizeof(int)};
229 l->deleted = true; // dormant until the first op arms it
230 l->ops = nullptr;
231 st->fdmap = hmap_insert_ctl(&fdmap_control, &st->fdmap, &l->key, l);
232 return l;
233}
234
235/**
236 * The epoll interest an op waits on. Read-side ops (READ, RECVFROM, RECVMSG,
237 * ACCEPT) wait for EPOLLIN; write-side ops (WRITE, SENDTO, SENDMSG, CONNECT)
238 * wait for EPOLLOUT. POLL carries its own requested events in `flags`.
239 */
240static uint32_t op_events(const epoll_op_t *op) {
241 switch (op->op) {
246 return EPOLLIN;
251 return EPOLLOUT;
253 return (uint32_t)op->flags;
254 default:
255 // NOP/SLEEP/CANCEL never reach a backend; anything else is a routing bug.
256 PANIC("epoll backend: operation has no readiness direction");
257 }
258}
259
260/**
261 * The union of every armed op's interest on this fd, what the fd is registered
262 * for in epoll.
263 */
264static uint32_t list_mask(const epoll_op_list_t *l) {
265 uint32_t m = 0;
266 for (epoll_op_t *o = l->ops; o != nullptr; o = o->fd_next) {
267 m |= op_events(o);
268 }
269 return m;
270}
271
272// -----------------------------------------------------------------------------
273// Operation pool (intrusive free list)
274// -----------------------------------------------------------------------------
275
276/**
277 * Take a free slot from the pool, or null when the pool is full. The caller
278 * fails the operation back to its fiber (a posted completion), so there is no
279 * need to allocate or carry an error here.
280 */
281static epoll_op_t *alloc_op(epoll_state_t *st) {
282 if (st->free_head == SIZE_MAX) {
283 return nullptr; // pool exhausted
284 }
285
286 // Pop the head, take the first free slot, advance the head to its successor.
287 size_t idx = st->free_head;
288 st->free_head = st->ops[idx].next_free;
289 return &st->ops[idx];
290}
291
292/**
293 * Return a slot to the pool, pushing it onto the front of the free list. Takes
294 * the op pointer (the wait loop holds the pointer, not the index) and recovers
295 * the index by its offset in the fixed `ops` array.
296 */
297static void free_op(epoll_state_t *st, epoll_op_t *op) {
298 size_t idx = (size_t)(op - st->ops); // pointer back to its slot index
299 op->active = false;
300 op->next_free = st->free_head; // point at the old head
301 st->free_head = idx; // become the new head
302}
303
304// -----------------------------------------------------------------------------
305// IO helpers
306// -----------------------------------------------------------------------------
307
308/**
309 * Empty the self-wake eventfd's counter. It is level-triggered, so epoll keeps
310 * reporting it readable until it is drained back to zero.
311 */
312static void drain_wake(epoll_state_t *st) {
313 assert(st != nullptr);
314
315 // A read of an eventfd returns its accumulated counter and resets it to zero.
316 // The fd is EFD_NONBLOCK, so once the counter is zero the read fails with
317 // EAGAIN instead of blocking, which ends the loop. So this absorbs however
318 // many wakes were coalesced into the counter and leaves it drained. And since
319 // we are draining the `wake_fd`, this loop is really trivial.
320 uint64_t v;
321 while (read(st->wake_fd, &v, sizeof(v)) == (ssize_t)sizeof(v)) {
322 }
323}
324
325/**
326 * Perform the transfer for a ready operation and fill `*out`. Returns false on
327 * EAGAIN/EWOULDBLOCK (the descriptor was not actually ready, so the op stays
328 * armed for the next readiness event); otherwise true with the result set.
329 *
330 * This function deliberately only does the transfer and records the result. It
331 * does NOT free the slot, deregister the fd, or post the completion. The wait
332 * loop owns that ordering (it must deregister before the completion is visible,
333 * so a worker that reaps it can safely close the fd).
334 *
335 * `ready` is the readiness mask epoll reported for the fd; only POLL uses it
336 * (to report which events fired), the transfers ignore it.
337 */
338static bool perform_op(epoll_op_t *op, uint32_t ready, srn_reactor_io_completion_t *out) {
339 assert(op != nullptr);
340
341 // Success by default; the error paths set `maybe_error`. `result` and
342 // `addrlen` are meaningful only on success, but default them so a failed op
343 // never leaves them indeterminate. Only ACCEPT and RECVFROM fill `addrlen`.
344 out->maybe_error = nullptr;
345 out->result = 0;
346 out->addrlen = 0;
347
348 ssize_t n;
349
350 switch (op->op) {
352 n = (op->offset < 0) ? read(op->fd, op->buf, op->len)
353 : pread(op->fd, op->buf, op->len, (off_t)op->offset);
354 break;
356 n = (op->offset < 0) ? write(op->fd, op->buf, op->len)
357 : pwrite(op->fd, op->buf, op->len, (off_t)op->offset);
358 break;
360 // The result is the accepted descriptor, carried back in `result` like a
361 // byte count. `addrlen` is value-result for accept4: pass the buffer
362 // capacity in, report the actual peer length back (valid when n >= 0).
363 socklen_t al = (socklen_t)op->addrlen;
364 n = accept4(
365 op->fd, (struct sockaddr *)op->addr, (op->addr != nullptr) ? &al : nullptr, op->flags
366 );
367 out->addrlen = (uint32_t)al;
368 break;
369 }
371 // Readiness means the non-blocking connect finished; SO_ERROR carries the
372 // verdict (0, or an errno). There is no transfer and no EAGAIN here.
373 int soerr = 0;
374 socklen_t errlen = sizeof(soerr);
375 if (getsockopt(op->fd, SOL_SOCKET, SO_ERROR, &soerr, &errlen) < 0) {
376 soerr = errno;
377 }
378 out->fiber_data = op->fiber_data;
379 if (soerr != 0) {
381 } else {
382 out->result = 0;
383 }
384 return true;
385 }
387 // `addrlen` is value-result here too, report the sender address length
388 // back.
389 socklen_t al = (socklen_t)op->addrlen;
390 n = recvfrom(
391 op->fd, op->buf, op->len, op->flags, (struct sockaddr *)op->addr,
392 (op->addr != nullptr) ? &al : nullptr
393 );
394 out->addrlen = (uint32_t)al;
395 break;
396 }
398 n = sendto(
399 op->fd, op->buf, op->len, op->flags, (struct sockaddr *)op->addr, (socklen_t)op->addrlen
400 );
401 break;
403 n = recvmsg(op->fd, (struct msghdr *)op->buf, op->flags);
404 break;
406 n = sendmsg(op->fd, (struct msghdr *)op->buf, op->flags);
407 break;
409 // No transfer, report which events are ready.
410 out->fiber_data = op->fiber_data;
411 out->result = (size_t)ready;
412 return true;
413 default:
414 // NOP/SLEEP/CANCEL are handled by the reactor core and never reach a
415 // backend. Anything else is a routing bug, not a runtime condition.
416 PANIC("epoll backend: unexpected operation kind");
417 }
418
419 if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
420 return false; // spurious readiness; leave it armed
421 }
422
423 out->fiber_data = op->fiber_data;
424 if (n < 0) {
426 } else {
427 out->result = (size_t)n; // bytes or accepted fd
428 }
429 return true;
430}
431
432// -----------------------------------------------------------------------------
433// Lifecycle
434// -----------------------------------------------------------------------------
435
436/**
437 * Bring the epoll backend up, build the operation pool and the fd map, open the
438 * epoll instance, and register the self-wake eventfd.
439 */
440static srn_error_t *epoll_backend_init(srn_reactor_t *reactor) {
441 PANIC_IF_NULL(reactor);
442
443 srn_context_t *ctx = reactor->ctx;
444
445 epoll_state_t *st = ALLOC(ctx, epoll_state_t);
446 size_t cap = (size_t)1 << ctx->engine->config.reactor.ring_magnitude;
447 st->total_slots = reactor->nchannels * cap;
448 // The pool scales with the channel count and outgrows an arena block on
449 // many-core machines (a single arena allocation is bounded by the block
450 // size). Its size is known here and it dies at backend shutdown, so it is
451 // a plain allocation, like the reactor's timers.
452 st->ops = srn_mm_malloc(ctx->engine->mm, st->total_slots * sizeof(epoll_op_t));
453 PANIC_IF_NULL(st->ops);
454 st->fdmap = hmap_empty(ctx);
455
456 // Chain every slot into the free list: 0 -> 1 -> ... -> last -> end.
457 for (size_t i = 0; i < st->total_slots; i++) {
458 st->ops[i].active = false;
459 st->ops[i].next_free = i + 1;
460 }
461 st->ops[st->total_slots - 1].next_free = SIZE_MAX;
462 st->free_head = 0;
463
464 st->epfd = epoll_create1(EPOLL_CLOEXEC);
465 st->wake_fd = -1;
466
467 if (st->epfd == -1) {
468 return ERR(ctx, EPOLL_INIT_FAILED, "Couldn't create the epoll file descriptor.");
469 }
470
471 st->wake_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
472 if (st->wake_fd == -1) {
473 close(st->epfd);
474 return ERR(
475 ctx, EPOLL_EVENT_FD_FAILED, "Couldn't create the event file descriptor of the reactor."
476 );
477 }
478
479 // Register the wake fd with a NULL data pointer. Armed fds carry a non-NULL
480 // pointer to their op list (below), so NULL uniquely marks "this is the wake
481 // fd".
482 struct epoll_event ee = {.events = EPOLLIN, .data.ptr = nullptr};
483 if (epoll_ctl(st->epfd, EPOLL_CTL_ADD, st->wake_fd, &ee) == -1) {
484 close(st->wake_fd);
485 close(st->epfd);
486 return ERR(ctx, EPOLL_FAILED_TO_ADD_FD, "Failed to add the wake_fd to the epfd");
487 }
488
489 reactor->backend_state = st;
490 return nullptr;
491}
492
493/**
494 * Tear the backend down, close the descriptors and detach the state. The pool,
495 * the lists and the map are arena-allocated (from the reactor context), so they
496 * are reclaimed in bulk with the context; there is nothing to free here.
497 */
498static void epoll_backend_shutdown(srn_reactor_t *reactor) {
499 PANIC_IF_NULL(reactor);
500
501 epoll_state_t *st = (epoll_state_t *)reactor->backend_state;
502 if (st == nullptr) {
503 return;
504 }
505
506 close(st->wake_fd);
507 close(st->epfd);
508 srn_mm_free(reactor->ctx->engine->mm, st->ops);
509 st->ops = nullptr;
510 reactor->backend_state = nullptr;
511}
512
513// -----------------------------------------------------------------------------
514// Submit / wait / wake
515// -----------------------------------------------------------------------------
516
517/**
518 * Arm a descriptor operation. The op joins its fd's list, and the fd is
519 * registered (or re-registered) for the union of its ops' interests. On a full
520 * pool, or if epoll refuses the fd, the operation is completed with its
521 * `maybe_error` set rather than aborting, so the waiting fiber wakes and can
522 * react. Called by the reactor core on the reactor thread while draining
523 * submissions.
524 */
525static void
526epoll_backend_submit(srn_reactor_t *reactor, size_t channel, const srn_reactor_io_request_t *req) {
527 PANIC_IF_NULL(reactor);
529
530 // A POLL that asks for no pollable event would register an empty interest
531 // mask, nothing could ever match it, and the fiber would park forever.
532 // Fail it back as malformed rather than arming it.
533 if (req->op == SRN_REACTOR_IO_POLL && (req->flags & (EPOLLIN | EPOLLOUT)) == 0) {
535 reactor, channel,
537 .fiber_data = req->fiber_data, .maybe_error = srn_reactor_errors_static(INVALID_ARGUMENT)
538 }
539 );
540 return;
541 }
542
543 srn_context_t *ctx = reactor->ctx;
544 epoll_state_t *st = reactor->backend_state;
545 epoll_op_t *op = alloc_op(st);
546
547 if (op == nullptr) {
548 // Pool full, fail this op back to its fiber. -ENOBUFS ("no buffer space
549 // available") is the closest errno for an exhausted op pool; it is
550 // transient, so the fiber can retry once in-flight operations drain.
553 reactor, channel,
554 (srn_reactor_io_completion_t){.fiber_data = req->fiber_data, .maybe_error = err}
555 );
556 return;
557 }
558
559 op->active = true;
560 op->fd = req->fd;
561 op->op = req->op;
562 op->channel = channel;
563 op->fiber_data = req->fiber_data;
564 op->buf = req->buf;
565 op->len = req->len;
566 op->offset = req->offset;
567 op->addr = req->addr;
568 op->addrlen = req->addrlen;
569 op->flags = req->flags;
570
571 // CONNECT issues the (non-blocking) connect now; the readiness later reports
572 // its verdict via SO_ERROR. An immediate success, or a hard failure that is
573 // not EINPROGRESS, completes here without arming anything.
574 if (op->op == SRN_REACTOR_IO_CONNECT) {
575 int rc = connect(op->fd, (struct sockaddr *)op->addr, (socklen_t)op->addrlen);
576
577 if (rc == 0 || errno != EINPROGRESS) {
578 srn_error_t *err =
579 (rc == 0) ? nullptr : srn_reactor_errors_static(srn_errno_to_tag(errno, UNKNOWN_IO_ERROR));
580 free_op(st, op);
582 reactor, channel,
583 (srn_reactor_io_completion_t){.fiber_data = req->fiber_data, .maybe_error = err}
584 );
585
586 return;
587 }
588 // EINPROGRESS: fall through and arm EPOLLOUT like any other op.
589 }
590
591 // Join the fd's list. A dormant (deleted) list is not yet in epoll, so this
592 // is the first op on the fd and needs an ADD; otherwise the fd is already
593 // registered and its interest mask only needs widening with a MOD.
594 epoll_op_list_t *l = fd_list(ctx, st, req->fd);
595 bool need_add = l->deleted;
596 l->deleted = false;
597 op->fd_next = l->ops;
598 l->ops = op;
599
600 // The list pointer rides in epoll's data, so a readiness event lands on every
601 // op for the fd with no search of the pool.
602 struct epoll_event ee = {.events = list_mask(l), .data.ptr = l};
603 int rc = epoll_ctl(st->epfd, (int)need_add ? EPOLL_CTL_ADD : EPOLL_CTL_MOD, req->fd, &ee);
604
605 if (rc == -1) {
606 // Could not arm it (bad fd, ...). Pop the op back off the list, return the
607 // slot and fail the op back; otherwise it would sit armed-in-software but
608 // never fire and the fiber would hang. If that emptied the list, mark it
609 // dormant again so the next op on this fd re-adds rather than mods.
610 int err = errno;
611 l->ops = op->fd_next;
612 free_op(st, op);
613 if (l->ops == nullptr) {
614 l->deleted = true;
615 }
616
619 reactor, channel,
620 (srn_reactor_io_completion_t){.fiber_data = req->fiber_data, .maybe_error = err1}
621 );
622 }
623}
624
625/**
626 * Wait for readiness up to `timeout_ms` (epoll's own unit for timout: -1 blocks
627 * until an event, 0 polls), then perform and complete every ready operation.
628 * Each event carries its fd's op list, so it resolves with no scan of the pool;
629 * the loop walks the list and completes the ops whose direction is ready.
630 * Called by the reactor thread's loop each pass.
631 */
632static void epoll_backend_wait(srn_reactor_t *reactor, int timeout_ms) {
633 PANIC_IF_NULL(reactor);
635
636 epoll_state_t *st = reactor->backend_state;
637 size_t batch = reactor->ctx->engine->config.reactor.reap_batch;
638
639 struct epoll_event events[batch];
640 int n = epoll_wait(st->epfd, events, (int)batch, timeout_ms);
641
642 if (n < 0) {
643 // The only error expected here is EINTR: a signal interrupted the wait, and
644 // the reactor loop simply calls wait again. Any other errno means epfd or
645 // the arguments are invalid -- a setup bug, not a runtime condition -- and
646 // silently returning would turn the reactor loop into a hot spin, so it
647 // panics in every build.
648 PANIC_IF(errno != EINTR, "reactor: epoll_wait failed");
649 return;
650 }
651
652 if (n == 0) {
653 return; // timed out; nothing was ready
654 }
655
656 for (int i = 0; i < n; i++) {
657 // The wake fd is the one event registered with a NULL data pointer. It
658 // carries no operation, someone (a worker that just submitted, or shutdown)
659 // wrote the eventfd purely to force this epoll_wait to return so the
660 // reactor thread re-checks its queues. Drain it and move on.
661 if (events[i].data.ptr == nullptr) {
662 drain_wake(st);
663 continue;
664 }
665
666 epoll_op_list_t *l = (epoll_op_list_t *)events[i].data.ptr;
667
668 // An error or hangup on the fd wakes both directions, a reader and a writer
669 // on the same fd should both observe it.
670 uint32_t ready = events[i].events;
671 if (ready & (EPOLLERR | EPOLLHUP)) {
672 ready |= (EPOLLIN | EPOLLOUT);
673 }
674
675 uint32_t before = list_mask(l);
676
677 // Walk the fd's ops, completing those the readiness covers. `link` is the
678 // address of the pointer that reaches the current op, so unlinking is a
679 // single store and the walk continues from the same spot.
680 epoll_op_t **link = &l->ops;
681 while (*link != nullptr) {
682 epoll_op_t *op = *link;
683
684 if ((op_events(op) & ready) == 0) {
685 link = &op->fd_next; // not this op's direction
686 continue;
687 }
688
689 // === PERFORM THE ACTUAL OPERATION ===
691 if (!perform_op(op, ready, &c)) {
692 link = &op->fd_next; // EAGAIN: still armed, will fire again
693 continue;
694 }
695
696 // Take this op off the list. If it was the last op on the fd, drop the fd
697 // from epoll and mark the list dormant -- and do both before posting the
698 // completion. Once the completion is posted, a worker may reap it and
699 // close the fd, so the fd has to be out of epoll by that point.
700 //
701 // The unlink already set `*link` to the next op, so the loop carries on
702 // without moving the cursor forward.
703 size_t ch = op->channel;
704 *link = op->fd_next;
705 if (l->ops == nullptr) {
706 epoll_ctl(st->epfd, EPOLL_CTL_DEL, op->fd, nullptr);
707 l->deleted = true;
708 }
709 free_op(st, op);
710 srn_reactor_post_completion(reactor, ch, c);
711 }
712
713 // If the fd is still armed but some ops finished, narrow its interest to
714 // what remains. Level-triggered epoll would otherwise report a stale
715 // EPOLLOUT every wait and spin.
716 if (!l->deleted) {
717 uint32_t now = list_mask(l);
718 if (now != before) {
719 int fd = (int)(intptr_t)l->key.data;
720 struct epoll_event ee = {.events = now, .data.ptr = l};
721 epoll_ctl(st->epfd, EPOLL_CTL_MOD, fd, &ee);
722 }
723 }
724 }
725}
726
727/**
728 * Break a concurrent epoll_wait out of its sleep from any thread, by making the
729 * self-wake eventfd readable. The reactor uses this as the submit-side nudge
730 * and during shutdown to re-check `running`. Called from srn_reactor_submit
731 * (after queuing a request) and srn_reactor_shutdown (to break the wait).
732 */
733static void epoll_backend_wake(srn_reactor_t *reactor) {
734 PANIC_IF_NULL(reactor);
735
736 epoll_state_t *st = (epoll_state_t *)reactor->backend_state;
737
738 uint64_t one = 1;
739 ssize_t rc = write(st->wake_fd, &one, sizeof(one));
740 // A full counter (EAGAIN) is fine, it is already readable, so the reactor
741 // will still wake. Any other error is ignored for the same reason.
742 UNUSED(rc);
743}
744
745// -----------------------------------------------------------------------------
746// Vtable
747// -----------------------------------------------------------------------------
749 .name = "epoll",
750 .init = epoll_backend_init,
751 .shutdown = epoll_backend_shutdown,
752 .submit = epoll_backend_submit,
753 .wait = epoll_backend_wait,
754 .wake = epoll_backend_wake,
756};
757
758#endif
759// NOLINTEND(performance-no-int-to-ptr)
int n
Definition acutest.h:525
Internal reactor backend interface.
const srn_reactor_backend_t srn_reactor_backend_epoll
#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
#define ALLOC(ctx, T)
Definition context.h:84
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:169
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
struct hmap_t hmap_t
Definition engine.h:33
Error handling for the runtime.
@ EPOLL_EVENT_FD_FAILED
Definition errors.h:86
@ INVALID_ARGUMENT
The request itself is malformed (an empty POLL interest mask, for example), so retrying it unchanged ...
Definition errors.h:113
@ UNKNOWN_IO_ERROR
Definition errors.h:118
@ EPOLL_FAILED_TO_ADD_FD
Definition errors.h:87
@ EPOLL_INIT_FAILED
Definition errors.h:85
srn_error_tag_t srn_errno_to_tag(int errno_value, srn_error_tag_t default_tag)
Map a POSIX errno to error tag.
#define ERR(ctx, err, msg)
Definition errors.h:148
hmap_t hmap_empty(const srn_context_t *ctx)
Create, initialize and return a new hashmap pinned to ctx.
Definition hashmap.c:655
void * hmap_lookup_ctl(const hmap_control_t *ctl, const hmap_t *hmap, const hmap_key_t *k, void *default_value)
Just like the hmap_lookup function but, it receives a hmap_control_t to customize the comparison and ...
Definition hashmap.c:638
hmap_t hmap_insert_ctl(const hmap_control_t *ctl, const hmap_t *hmap, hmap_key_t *k, void *v)
Just like the hmap_insert function but, it receives a hmap_control_t to customize the comparison and ...
Definition hashmap.c:594
This is an implementation of Compressed Hash-Array Mapped Prefix-tree, which is a bit-partitioned,...
struct hmap_key_t hmap_key_t
Note: For key equality we use the memcpy function.
HMAP_HASH_TYPE hmap_hash_t
Definition hashmap.h:61
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
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:141
srn_reactor_io_op_t
The operation a submission asks the reactor to perform.
Definition reactor.h:63
@ 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_CONNECT
Connect fd to the address in addr/addrlen.
Definition reactor.h:86
@ 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
If we ever want to modify some of these behaviours for a new instance of hashmap, we should use this ...
Definition hashmap.h:109
Note: For key equality we use the memcpy function.
Definition hashmap.h:66
void * data
len 0 -> data == nullptr
Definition hashmap.h:68
const srn_context_t * map_ctx
The context that owns every allocation the map retains.
Definition hashmap.h:104
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:62
srn_mm_t * mm
Memory manager.
Definition engine.h:65
A runtime error, a tag classifying the failure and a human-readable message.
Definition errors.h:125
One async system behind the reactor contract.
Definition backend.h:55
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 completion, the result of one submission, echoed back on the channel.
Definition reactor.h:159
void * fiber_data
The fiber_data of the submission this completes.
Definition reactor.h:162
uint32_t addrlen
For SRN_REACTOR_IO_ACCEPT and SRN_REACTOR_IO_RECVFROM, the actual length of the address written into ...
Definition reactor.h:172
size_t result
The result of the operation on success, bytes transferred for the transfers, the accepted descriptor ...
Definition reactor.h:166
A submission, one request a fiber places on its channel.
Definition reactor.h:116
int fd
The descriptor the operation targets.
Definition reactor.h:120
int flags
Operation flags, the MSG_* flags for the datagram and message ops, the accept4 flags for SRN_REACTOR_...
Definition reactor.h:150
void * addr
Socket address.
Definition reactor.h:142
srn_reactor_io_op_t op
Definition reactor.h:117
void * buf
The transfer buffer for READ, WRITE, RECVFROM, and SENDTO, and the struct msghdr * for RECVMSG and SE...
Definition reactor.h:123
size_t len
The transfer length, in bytes, for READ, WRITE, RECVFROM, and SENDTO.
Definition reactor.h:125
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:128
void * fiber_data
Opaque token, round tripped to the matching completion.
Definition reactor.h:136
uint32_t addrlen
Length of addr, the exact address length for SRN_REACTOR_IO_CONNECT and SRN_REACTOR_IO_SENDTO,...
Definition reactor.h:146
srn_context_t * ctx
Definition internal.h:44
void * backend_state
Definition internal.h:48
#define PANIC_IF_NULL(ptr)
Definition utils.h:66
#define PANIC_IF(cond, msg)
Definition utils.h:59
#define UNUSED(x)
Definition utils.h:45
#define PANIC(msg)
Definition utils.h:53