From 1619e01174dc5524847ef5145d26f1c47dce3942 Mon Sep 17 00:00:00 2001 From: Lorenz Stechauner Date: Thu, 5 Jan 2023 20:33:59 +0100 Subject: [PATCH] Update async to use epoll instead of poll --- Makefile | 3 +- src/async.c | 150 ++++++++++++++++++++++++++++++++++++++------------ src/async.h | 17 ++++-- src/server.c | 2 +- src/workers.c | 4 +- 5 files changed, 132 insertions(+), 44 deletions(-) diff --git a/Makefile b/Makefile index 7c036af..cfcb2bb 100644 --- a/Makefile +++ b/Makefile @@ -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 diff --git a/src/async.c b/src/async.c index 07d6eda..f578459 100644 --- a/src/async.c +++ b/src/async.c @@ -11,32 +11,72 @@ #include "lib/list.h" #include +#include #include #include #include #include #include +#include + +#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); } diff --git a/src/async.h b/src/async.h index b845e94..0ac5c4a 100644 --- a/src/async.h +++ b/src/async.h @@ -11,13 +11,22 @@ #include "lib/sock.h" -#include - #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); diff --git a/src/server.c b/src/server.c index 49a531c..1c33faf 100644 --- a/src/server.c +++ b/src/server.c @@ -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"); diff --git a/src/workers.c b/src/workers.c index 34c62ce..4cd6b2b 100644 --- a/src/workers.c +++ b/src/workers.c @@ -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); }