From 51d85cc99fb1f18079d28e746fbba909228668ed Mon Sep 17 00:00:00 2001 From: Lorenz Stechauner Date: Mon, 2 Jan 2023 16:52:18 +0100 Subject: [PATCH] Async check if closed --- src/async.c | 113 +++++++++++++++++++++++++++++++------------------- src/async.h | 6 ++- src/server.c | 2 +- src/workers.c | 2 +- 4 files changed, 78 insertions(+), 45 deletions(-) diff --git a/src/async.c b/src/async.c index e759293..78d327d 100644 --- a/src/async.c +++ b/src/async.c @@ -17,6 +17,7 @@ typedef struct { int fd; + sock *socket; short events; int flags; void (*cb)(void *); @@ -30,20 +31,69 @@ typedef struct { evt_listen_t q[256]; } listen_queue_t; -static listen_queue_t listen1, listen2, *listen = &listen1; +static listen_queue_t listen1, listen2, *listen_q = &listen1; static volatile sig_atomic_t alive = 1; static pthread_t thread = -1; static int async_add_to_queue(evt_listen_t *evt) { // TODO locking - memcpy(&listen->q[listen->n++], evt, sizeof(*evt)); + memcpy(&listen_q->q[listen_q->n++], evt, sizeof(*evt)); return 0; } -int async(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg) { - struct pollfd fds[1] = {{.fd = fd, .events = events}}; +static int async_exec(evt_listen_t *evt, short r_events) { + if (r_events & evt->events) { + // specified event(s) occurred + if (evt->socket && !sock_check(evt->socket)) { + evt->err_cb(evt->err_arg); + errno = 0; + } else { + evt->cb(evt->arg); + errno = 0; + + if (evt->flags & ASYNC_KEEP) + return 1; + } + return 0; + } else if (r_events & (POLLERR | POLLHUP | POLLNVAL)) { + // error occurred + evt->err_cb(evt->err_arg); + errno = 0; + return 0; + } else { + // no event occurred + return -1; + } +} + +static int async_check(evt_listen_t *evt) { + struct pollfd fds[1] = {{.fd = evt->fd, .events = 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) + return 1; + } + + return 0; +} + +static int async_add(evt_listen_t *evt) { + if (async_check(evt) == 1) + return 0; + + int ret = async_add_to_queue(evt); + if (ret == 0 && thread != -1) + pthread_kill(thread, SIGUSR1); + + return ret; +} + +int async_fd(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg) { evt_listen_t evt = { .fd = fd, + .socket = NULL, .events = events, .flags = flags, .cb = cb, @@ -51,28 +101,21 @@ int async(int fd, short events, int flags, void cb(void *), void *arg, void err_ .err_cb = err_cb, .err_arg = err_arg, }; + return async_add(&evt); +} - // check, if fd is already ready - if (poll(fds, 1, 0) == 1) { - // fd already read - if (fds[0].revents & events) { - // specified event(s) occurred - cb(arg); - - if (!(flags & ASYNC_KEEP)) - return 0; - } else if (fds[0].revents & (POLLERR | POLLHUP | POLLNVAL)) { - // error occurred - err_cb(err_arg); - return 0; - } - } - - int ret = async_add_to_queue(&evt); - if (ret == 0 && thread != -1) - pthread_kill(thread, SIGUSR1); - - return ret; +int async(sock *s, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg) { + evt_listen_t evt = { + .fd = s->socket, + .socket = s, + .events = events, + .flags = flags, + .cb = cb, + .arg = arg, + .err_cb = err_cb, + .err_arg = err_arg, + }; + return async_add(&evt); } void async_thread(void) { @@ -84,8 +127,8 @@ void async_thread(void) { // main event loop while (alive) { // swap listen queue - listen_queue_t *l = listen; - listen = (listen == &listen1) ? &listen2 : &listen1; + listen_queue_t *l = listen_q; + listen_q = (listen_q == &listen1) ? &listen2 : &listen1; // fill fds with newly added queue entries for (num_fds = 0; num_fds < l->n; num_fds++) { @@ -106,22 +149,8 @@ void async_thread(void) { for (int i = 0; i < num_fds; i++) { evt_listen_t *e = &l->q[i]; - if (fds[i].revents & e->events) { - // specified event(s) occurred - e->cb(e->arg); - - if (e->flags & ASYNC_KEEP) - async_add_to_queue(e); - } else if (fds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) { - // error occurred - e->err_cb(e->err_arg); - } else { - // no event occurred + if (async_exec(e, fds[i].revents) != 0) async_add_to_queue(e); - } - - // reset errno to prevent strange behaviour - errno = 0; } // reset size of queue diff --git a/src/async.h b/src/async.h index 89f71d7..f706b19 100644 --- a/src/async.h +++ b/src/async.h @@ -9,11 +9,15 @@ #ifndef SESIMOS_ASYNC_H #define SESIMOS_ASYNC_H +#include "lib/sock.h" + #include #define ASYNC_KEEP 1 -int async(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg); +int async(sock *s, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg); + +int async_fd(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg); void async_thread(void); diff --git a/src/server.c b/src/server.c index be701a3..c4b5e99 100644 --- a/src/server.c +++ b/src/server.c @@ -257,7 +257,7 @@ int main(int argc, char *const argv[]) { workers_init(); for (int i = 0; i < NUM_SOCKETS; i++) { - async(sockets[i], POLLIN, ASYNC_KEEP, accept_cb, &sockets[i], accept_err_cb, &sockets[i]); + async_fd(sockets[i], POLLIN, ASYNC_KEEP, accept_cb, &sockets[i], accept_err_cb, &sockets[i]); } notice("Ready to accept connections"); diff --git a/src/workers.c b/src/workers.c index 9cda320..889cf17 100644 --- a/src/workers.c +++ b/src/workers.c @@ -50,7 +50,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.socket, POLLIN, 0, (void (*)(void *)) handle_request_cb, ctx, (void (*)(void *)) tcp_close, ctx); + return async(&ctx->socket, POLLIN, 0, (void (*)(void *)) handle_request_cb, ctx, (void (*)(void *)) tcp_close, ctx); } else { tcp_close(ctx); return 0;