56# include <sys/epoll.h>
57# include <sys/eventfd.h>
58# include <sys/socket.h>
69typedef struct epoll_op_t {
80 struct epoll_op_t *fd_next;
105typedef struct epoll_op_list_t {
123typedef struct epoll_state_t {
169 assert(a !=
nullptr);
170 assert(b !=
nullptr);
184 assert(k !=
nullptr);
205 .copy_key = fdmap_copy_key,
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);
218 hmap_key_t probe = {.data = (
void *)(intptr_t)fd, .len =
sizeof(int)};
220 epoll_op_list_t *l =
hmap_lookup_ctl(&fdmap_control, &st->fdmap, &probe,
nullptr);
227 l =
ALLOC(ctx, epoll_op_list_t);
228 l->key = (
hmap_key_t){.data = (
void *)(intptr_t)fd, .len =
sizeof(int)};
240static uint32_t op_events(
const epoll_op_t *op) {
253 return (uint32_t)op->flags;
256 PANIC(
"epoll backend: operation has no readiness direction");
264static uint32_t list_mask(
const epoll_op_list_t *l) {
266 for (epoll_op_t *o = l->ops; o !=
nullptr; o = o->fd_next) {
281static epoll_op_t *alloc_op(epoll_state_t *st) {
282 if (st->free_head == SIZE_MAX) {
287 size_t idx = st->free_head;
288 st->free_head = st->ops[idx].next_free;
289 return &st->ops[idx];
297static void free_op(epoll_state_t *st, epoll_op_t *op) {
298 size_t idx = (size_t)(op - st->ops);
300 op->next_free = st->free_head;
312static void drain_wake(epoll_state_t *st) {
313 assert(st !=
nullptr);
321 while (read(st->wake_fd, &v,
sizeof(v)) == (ssize_t)
sizeof(v)) {
339 assert(op !=
nullptr);
344 out->maybe_error =
nullptr;
352 n = (op->offset < 0) ? read(op->fd, op->buf, op->len)
353 : pread(op->fd, op->buf, op->len, (off_t)op->offset);
356 n = (op->offset < 0) ? write(op->fd, op->buf, op->len)
357 : pwrite(op->fd, op->buf, op->len, (off_t)op->offset);
363 socklen_t al = (socklen_t)op->addrlen;
365 op->fd, (
struct sockaddr *)op->addr, (op->addr !=
nullptr) ? &al :
nullptr, op->flags
374 socklen_t errlen =
sizeof(soerr);
375 if (getsockopt(op->fd, SOL_SOCKET, SO_ERROR, &soerr, &errlen) < 0) {
389 socklen_t al = (socklen_t)op->addrlen;
391 op->fd, op->buf, op->len, op->flags, (
struct sockaddr *)op->addr,
392 (op->addr !=
nullptr) ? &al :
nullptr
399 op->fd, op->buf, op->len, op->flags, (
struct sockaddr *)op->addr, (socklen_t)op->addrlen
403 n = recvmsg(op->fd, (
struct msghdr *)op->buf, op->flags);
406 n = sendmsg(op->fd, (
struct msghdr *)op->buf, op->flags);
411 out->
result = (size_t)ready;
416 PANIC(
"epoll backend: unexpected operation kind");
419 if (
n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
445 epoll_state_t *st =
ALLOC(ctx, epoll_state_t);
447 st->total_slots = reactor->nchannels * cap;
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;
461 st->ops[st->total_slots - 1].next_free = SIZE_MAX;
464 st->epfd = epoll_create1(EPOLL_CLOEXEC);
467 if (st->epfd == -1) {
471 st->wake_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
472 if (st->wake_fd == -1) {
482 struct epoll_event ee = {.events = EPOLLIN, .data.ptr =
nullptr};
483 if (epoll_ctl(st->epfd, EPOLL_CTL_ADD, st->wake_fd, &ee) == -1) {
489 reactor->backend_state = st;
510 reactor->backend_state =
nullptr;
545 epoll_op_t *op = alloc_op(st);
562 op->channel = channel;
567 op->addr = req->
addr;
569 op->flags = req->
flags;
575 int rc = connect(op->fd, (
struct sockaddr *)op->addr, (socklen_t)op->addrlen);
577 if (rc == 0 || errno != EINPROGRESS) {
594 epoll_op_list_t *l = fd_list(ctx, st, req->
fd);
595 bool need_add = l->deleted;
597 op->fd_next = l->ops;
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);
611 l->ops = op->fd_next;
613 if (l->ops ==
nullptr) {
632static void epoll_backend_wait(
srn_reactor_t *reactor,
int timeout_ms) {
639 struct epoll_event events[batch];
640 int n = epoll_wait(st->epfd, events, (
int)batch, timeout_ms);
648 PANIC_IF(errno != EINTR,
"reactor: epoll_wait failed");
656 for (
int i = 0; i <
n; i++) {
661 if (events[i].data.ptr ==
nullptr) {
666 epoll_op_list_t *l = (epoll_op_list_t *)events[i].data.ptr;
670 uint32_t ready = events[i].events;
671 if (ready & (EPOLLERR | EPOLLHUP)) {
672 ready |= (EPOLLIN | EPOLLOUT);
675 uint32_t before = list_mask(l);
680 epoll_op_t **link = &l->ops;
681 while (*link !=
nullptr) {
682 epoll_op_t *op = *link;
684 if ((op_events(op) & ready) == 0) {
691 if (!perform_op(op, ready, &c)) {
703 size_t ch = op->channel;
705 if (l->ops ==
nullptr) {
706 epoll_ctl(st->epfd, EPOLL_CTL_DEL, op->fd,
nullptr);
717 uint32_t now = list_mask(l);
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);
739 ssize_t rc = write(st->wake_fd, &one,
sizeof(one));
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,
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...
void srn_mm_free(srn_mm_t *mm, void *ptr)
Release a pointer previously returned by srn_mm_malloc or srn_mm_reallocate.
void * srn_mm_malloc(srn_mm_t *mm, size_t size)
Generic allocations that do not participate in the block based pools.
Error handling for the runtime.
@ INVALID_ARGUMENT
The request itself is malformed (an empty POLL interest mask, for example), so retrying it unchanged ...
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)
hmap_t hmap_empty(const srn_context_t *ctx)
Create, initialize and return a new hashmap pinned to ctx.
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 ...
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 ...
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
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 no...
srn_reactor_io_op_t
The operation a submission asks the reactor to perform.
@ SRN_REACTOR_IO_ACCEPT
Accept a pending connection on the listening socket fd, with flags (the accept4 flags).
@ SRN_REACTOR_IO_READ
Read from a descriptor into buf. Uses fd, buf, len, offset.
@ SRN_REACTOR_IO_SENDTO
Send buf/len on fd to the address in addr/addrlen with flags.
@ SRN_REACTOR_IO_CONNECT
Connect fd to the address in addr/addrlen.
@ SRN_REACTOR_IO_RECVFROM
Receive a datagram from fd into buf/len with flags.
@ SRN_REACTOR_IO_RECVMSG
Receive into the struct msghdr that buf points at, on fd, with flags (scatter/gather and ancillary da...
@ SRN_REACTOR_IO_WRITE
Write buf to a descriptor. Uses fd, buf, len, offset.
@ SRN_REACTOR_IO_POLL
Wait until fd is ready for the events requested in flags (EPOLLIN/EPOLLOUT) without performing any tr...
@ SRN_REACTOR_IO_SENDMSG
Send the struct msghdr that buf points at, on fd, with flags.
If we ever want to modify some of these behaviours for a new instance of hashmap, we should use this ...
Note: For key equality we use the memcpy function.
void * data
len 0 -> data == nullptr
const srn_context_t * map_ctx
The context that owns every allocation the map retains.
srn_reactor_config_t reactor
srn_engine_t * engine
Long term state of the compiler.
srn_configuration_t config
The runtime's tunable knobs, the single source for every configurable value (see configuration....
srn_mm_t * mm
Memory manager.
A runtime error, a tag classifying the failure and a human-readable message.
One async system behind the reactor contract.
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.
void * fiber_data
The fiber_data of the submission this completes.
uint32_t addrlen
For SRN_REACTOR_IO_ACCEPT and SRN_REACTOR_IO_RECVFROM, the actual length of the address written into ...
size_t result
The result of the operation on success, bytes transferred for the transfers, the accepted descriptor ...
A submission, one request a fiber places on its channel.
int fd
The descriptor the operation targets.
int flags
Operation flags, the MSG_* flags for the datagram and message ops, the accept4 flags for SRN_REACTOR_...
void * addr
Socket address.
void * buf
The transfer buffer for READ, WRITE, RECVFROM, and SENDTO, and the struct msghdr * for RECVMSG and SE...
size_t len
The transfer length, in bytes, for READ, WRITE, RECVFROM, and SENDTO.
int64_t offset
The file offset for SRN_REACTOR_IO_READ and SRN_REACTOR_IO_WRITE; -1 reads at the current position.
void * fiber_data
Opaque token, round tripped to the matching completion.
uint32_t addrlen
Length of addr, the exact address length for SRN_REACTOR_IO_CONNECT and SRN_REACTOR_IO_SENDTO,...
#define PANIC_IF_NULL(ptr)
#define PANIC_IF(cond, msg)