lorenz.stechauner/sesimos
1
1
Fork 0
Code Issues Pull Requests Projects Releases Activity

Update async to use epoll instead of poll

Browse Source
This commit is contained in:
Lorenz Stechauner
2023-01-05 20:33:59 +01:00
parent 6112e719e5
commit 1619e01174
5 changed files with 132 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Makefile
View File

@ -35,7 +35,8 @@ bin/worker:
bin/res:
mkdir -p bin/res
bin/test: test/mock_*.c test/test_*.c src/lib/utils.c src/lib/sock.c src/lib/list.c
bin/test: test/mock_*.c test/test_*.c \
src/lib/utils.c src/lib/sock.c src/lib/list.c src/lib/http.c src/lib/http_static.c src/logger.c
$(CC) -o $@ $(CFLAGS) $^ -lcriterion

150
src/async.c
View File

@ -11,32 +11,72 @@
#include "lib/list.h"
#include <poll.h>
#include <sys/epoll.h>
#include <signal.h>
#include <errno.h>
#include <memory.h>
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h>
#define ASYNC_MAX_EVENTS 16
typedef struct {
int fd;
sock *socket;
short events;
async_evt_t events;
int flags;
void (*cb)(void *);
void *arg;
void (*cb)(void *);
void (*err_cb)(void *);
void *err_arg;
} evt_listen_t;
typedef struct {
int n;
evt_listen_t q[64];
evt_listen_t *q[ASYNC_MAX_EVENTS];
} listen_queue_t;
static listen_queue_t listen1, listen2, *listen_q = &listen1;
static volatile sig_atomic_t alive = 1;
static pthread_t thread = -1;
static sem_t lock;
static int epoll_fd;
static short async_a2p(async_evt_t events) {
short ret = 0;
if (events & ASYNC_IN) ret |= POLLIN;
if (events & ASYNC_PRI) ret |= POLLPRI;
if (events & ASYNC_OUT) ret |= POLLOUT;
return ret;
}
static unsigned int async_a2e(async_evt_t events) {
unsigned int ret = 0;
if (events & ASYNC_IN) ret |= EPOLLIN;
if (events & ASYNC_PRI) ret |= EPOLLPRI;
if (events & ASYNC_OUT) ret |= EPOLLOUT;
return ret;
}
static async_evt_t async_p2a(short events) {
async_evt_t ret = 0;
if (events & POLLIN) ret |= ASYNC_IN;
if (events & POLLPRI) ret |= ASYNC_PRI;
if (events & POLLOUT) ret |= ASYNC_OUT;
if (events & POLLERR) ret |= ASYNC_ERR;
if (events & POLLHUP) ret |= ASYNC_HUP;
return ret;
}
static async_evt_t async_e2a(unsigned int events) {
async_evt_t ret = 0;
if (events & EPOLLIN) ret |= ASYNC_IN;
if (events & EPOLLPRI) ret |= ASYNC_PRI;
if (events & EPOLLOUT) ret |= ASYNC_OUT;
if (events & EPOLLERR) ret |= ASYNC_ERR;
if (events & EPOLLHUP) ret |= ASYNC_HUP;
return ret;
}
static int async_add_to_queue(evt_listen_t *evt) {
try_again:
@ -48,19 +88,26 @@ static int async_add_to_queue(evt_listen_t *evt) {
}
}
memcpy(&listen_q->q[listen_q->n++], evt, sizeof(*evt));
evt_listen_t *ptr = malloc(sizeof(evt_listen_t));
if (ptr == NULL) {
sem_post(&lock);
return -1;
}
memcpy(ptr, evt, sizeof(*evt));
listen_q->q[listen_q->n++] = ptr;
sem_post(&lock);
return 0;
}
static int async_exec(evt_listen_t *evt, short r_events) {
static int async_exec(evt_listen_t *evt, async_evt_t r_events) {
int ret, e = errno;
if (r_events & evt->events) {
// specified event(s) occurred
if (evt->socket && !sock_has_pending(evt->socket)) {
evt->err_cb(evt->err_arg);
evt->err_cb(evt->arg);
ret = 0;
} else {
evt->cb(evt->arg);
@ -68,7 +115,7 @@ static int async_exec(evt_listen_t *evt, short r_events) {
}
} else if (r_events & (POLLERR | POLLHUP | POLLNVAL)) {
// error occurred
evt->err_cb(evt->err_arg);
evt->err_cb(evt->arg);
ret = 0;
} else {
// no event occurred
@ -81,12 +128,15 @@ static int async_exec(evt_listen_t *evt, short r_events) {
}
static int async_check(evt_listen_t *evt) {
struct pollfd fds[1] = {{.fd = evt->fd, .events = evt->events}};
struct pollfd fds[1] = {{
.fd = evt->fd,
.events = async_a2p(evt->events)
}};
// check, if fd is already ready
if (poll(fds, 1, 0) == 1) {
// fd already read
if (async_exec(evt, fds[0].revents) == 0)
// fd already ready
if (async_exec(evt, async_p2a(fds[0].revents)) == 0)
return 1;
}
@ -104,30 +154,28 @@ static int async_add(evt_listen_t *evt) {
return ret;
}
int async_fd(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg) {
int async_fd(int fd, async_evt_t events, int flags, void *arg, void cb(void *), void err_cb(void *)) {
evt_listen_t evt = {
.fd = fd,
.socket = NULL,
.events = events,
.flags = flags,
.cb = cb,
.arg = arg,
.cb = cb,
.err_cb = err_cb,
.err_arg = err_arg,
};
return async_add(&evt);
}
int async(sock *s, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg) {
int async(sock *s, async_evt_t events, int flags, void *arg, void cb(void *), void err_cb(void *)) {
evt_listen_t evt = {
.fd = s->socket,
.socket = s,
.events = events,
.flags = flags,
.cb = cb,
.arg = arg,
.cb = cb,
.err_cb = err_cb,
.err_arg = err_arg,
};
return async_add(&evt);
}
@ -140,19 +188,29 @@ int async_init(void) {
listen1.n = 0;
listen2.n = 0;
if ((epoll_fd = epoll_create1(0)) == -1) {
async_free();
return -1;
}
return 0;
}
void async_free(void) {
int e = errno;
sem_destroy(&lock);
close(epoll_fd);
errno = e;
}
void async_thread(void) {
evt_listen_t *local = list_create(sizeof(evt_listen_t), 16);
// TODO use epoll instead of poll
struct epoll_event ev, events[ASYNC_MAX_EVENTS];
int num_fds;
evt_listen_t **local = list_create(sizeof(evt_listen_t *), 16);
if (local == NULL) {
critical("Unable to create async local list");
return;
}
thread = pthread_self();
@ -162,23 +220,27 @@ void async_thread(void) {
listen_queue_t *l = listen_q;
listen_q = (listen_q == &listen1) ? &listen2 : &listen1;
// fill local list with previously added queue entries
// fill local list and epoll instance with previously added queue entries
for (int i = 0; i < l->n; i++) {
local = list_append(local, &l->q[i]);
}
evt_listen_t *evt = l->q[i];
local = list_append(local, &evt);
if (local == NULL) {
critical("Unable to resize async local list");
return;
}
ev.events = async_a2e(evt->events);
ev.data.ptr = evt;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, evt->fd, &ev) == -1) {
critical("Unable to add file descriptor to epoll instance");
return;
}
}
// reset size of queue
l->n = 0;
// fill fds with newly added queue entries
int num_fds = 0;
struct pollfd fds[list_size(local)];
for (int i = 0; i < list_size(local); i++, num_fds++) {
fds[num_fds].fd = local[i].fd;
fds[num_fds].events = local[i].events;
}
if (poll(fds, num_fds, -1) < 0) {
if ((num_fds = epoll_wait(epoll_fd, events, ASYNC_MAX_EVENTS, -1)) == -1) {
if (errno == EINTR) {
// interrupt
errno = 0;
@ -190,13 +252,29 @@ void async_thread(void) {
}
}
for (int i = 0, j = 0; i < num_fds; i++, j++) {
evt_listen_t *evt = &local[j];
if (async_exec(evt, fds[i].revents) == 0)
local = list_remove(local, j--);
for (int i = 0; i < num_fds; i++) {
evt_listen_t *evt = events[i].data.ptr;
if (async_exec(evt, async_e2a(events[i].events)) == 0) {
if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, evt->fd, NULL) == -1) {
critical("Unable to remove file descriptor from epoll instance");
return;
}
local = list_delete(local, &evt);
if (local == NULL) {
critical("Unable to resize async local list");
return;
}
free(evt);
}
}
}
// cleanup
for (int i = 0; i < list_size(local); i++) {
free(local[i]);
}
list_free(local);
}

17
src/async.h
View File

@ -11,13 +11,22 @@
#include "lib/sock.h"
#include <poll.h>
#define ASYNC_KEEP 1
int async(sock *s, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg);
#define ASYNC_IN 0x01
#define ASYNC_PRI 0x02
#define ASYNC_OUT 0x04
#define ASYNC_ERR 0x08
#define ASYNC_HUP 0x10
int async_fd(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg);
#define ASYNC_WAIT_READ ASYNC_IN
#define ASYNC_WAIT_WRITE ASYNC_OUT
typedef unsigned int async_evt_t;
int async(sock *s, async_evt_t events, int flags, void *arg, void cb(void *), void err_cb(void *));
int async_fd(int fd, async_evt_t events, int flags, void *arg, void cb(void *), void err_cb(void *));
int async_init(void);

2
src/server.c
View File

@ -296,7 +296,7 @@ int main(int argc, char *const argv[]) {
workers_init();
for (int i = 0; i < NUM_SOCKETS; i++) {
async_fd(sockets[i], POLLIN, ASYNC_KEEP, accept_cb, &sockets[i], accept_err_cb, &sockets[i]);
async_fd(sockets[i], ASYNC_WAIT_READ, ASYNC_KEEP, &sockets[i], accept_cb, accept_err_cb);
}
notice("Ready to accept connections");

4
src/workers.c
View File

@ -53,7 +53,7 @@ static int handle_request_cb(client_ctx_t *ctx) {
int handle_request(client_ctx_t *ctx) {
if (ctx->c_keep_alive && ctx->s_keep_alive) {
return async(&ctx->socket, POLLIN, 0, (void (*)(void *)) handle_request_cb, ctx, (void (*)(void *)) tcp_close, ctx);
return async(&ctx->socket, ASYNC_WAIT_READ, 0, ctx, (void (*)(void *)) handle_request_cb, (void (*)(void *)) tcp_close);
} else {
tcp_close(ctx);
return 0;
@ -77,5 +77,5 @@ static int ws_handle_frame_cb(ws_ctx_t *ctx) {
}
int ws_handle_frame(ws_ctx_t *ctx) {
return async(ctx->socket, POLLIN, 0, (void (*)(void *)) ws_handle_frame_cb, ctx, (void (*)(void *)) ws_close, ctx);
return async(ctx->socket, ASYNC_WAIT_READ, 0, ctx, (void (*)(void *)) ws_handle_frame_cb, (void (*)(void *)) ws_close);
}