Async check if closed

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) {
+static int async_exec(evt_listen_t *evt, short r_events) {
-    struct pollfd fds[1] = {{.fd = fd, .events = 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
+int async(sock *s, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg) {
-    if (poll(fds, 1, 0) == 1) {
+    evt_listen_t evt = {
-        // fd already read
+            .fd = s->socket,
-        if (fds[0].revents & events) {
+            .socket = s,
-            // specified event(s) occurred
+            .events = events,
-            cb(arg);
+            .flags = flags,
-
+            .cb = cb,
-            if (!(flags & ASYNC_KEEP))
+            .arg = arg,
-                return 0;
+            .err_cb = err_cb,
-        } else if (fds[0].revents & (POLLERR | POLLHUP | POLLNVAL)) {
+            .err_arg = err_arg,
-            // error occurred
+    };
-            err_cb(err_arg);
+    return async_add(&evt);
-            return 0;
-        }
-    }
-
-    int ret = async_add_to_queue(&evt);
-    if (ret == 0 && thread != -1)
-        pthread_kill(thread, SIGUSR1);
-
-    return ret;
 }
 
 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_queue_t *l = listen_q;
-        listen = (listen == &listen1) ? &listen2 : &listen1;
+        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) {
+            if (async_exec(e, fds[i].revents) != 0)
-                // 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
                 async_add_to_queue(e);
-            }
-
-            // reset errno to prevent strange behaviour
-            errno = 0;
         }
 
         // reset size of queue

src/async.h

@@ -9,11 +9,15 @@
 #ifndef SESIMOS_ASYNC_H
 #define SESIMOS_ASYNC_H
 
+#include "lib/sock.h"
+
 #include <poll.h>
 
 #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);
 

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");

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;