Async locking

src/async.c

@@ -14,6 +14,7 @@
 #include <errno.h>
 #include <memory.h>
 #include <pthread.h>
+#include <semaphore.h>
 
 typedef struct {
     int fd;
@@ -34,10 +35,22 @@ typedef struct {
 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 async_add_to_queue(evt_listen_t *evt) {
-    // TODO locking
+    try_again:
+    if (sem_wait(&lock) != 0) {
+        if (errno == EINTR) {
+            goto try_again;
+        } else {
+            return -1;
+        }
+    }
+
     memcpy(&listen_q->q[listen_q->n++], evt, sizeof(*evt));
+
+    sem_post(&lock);
+
     return 0;
 }
 
@@ -118,6 +131,20 @@ int async(sock *s, short events, int flags, void cb(void *), void *arg, void err
     return async_add(&evt);
 }
 
+int async_init(void) {
+    if (sem_init(&lock, 0, 1) != 0) {
+        return -1;
+    }
+
+    return 0;
+}
+
+void async_free(void) {
+    int e = errno;
+    sem_destroy(&lock);
+    errno = e;
+}
+
 void async_thread(void) {
     int num_fds;
     struct pollfd fds[256];  // TODO dynamic

src/async.h

@@ -19,6 +19,10 @@ int async(sock *s, short events, int flags, void cb(void *), void *arg, void err
 
 int async_fd(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg);
 
+int async_init(void);
+
+void async_free(void);
+
 void async_thread(void);
 
 void async_stop(void);

src/server.c

@@ -244,12 +244,19 @@ int main(int argc, char *const argv[]) {
         }
     }
 
+    if (async_init() != 0) {
+        critical("Unable to initialize async thread");
+        geoip_free();
+        return 1;
+    }
+
     proxy_preload();
 
     for (int i = 0; i < NUM_SOCKETS; i++) {
         if (listen(sockets[i], LISTEN_BACKLOG) < 0) {
             critical("Unable to listen on socket %i", i);
             geoip_free();
+            proxy_unload();
             return 1;
         }
     }
@@ -264,7 +271,12 @@ int main(int argc, char *const argv[]) {
 
     async_thread();
 
+    warning("Async thread finished");
+    notice("Goodbye?");
+
     // cleanup
     geoip_free();
+    proxy_unload();
+    async_free();
     return 0;
 }