Handle proxy earlier

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;
 }

src/worker/request_handler.c

@@ -327,17 +327,17 @@ int respond(client_ctx_t *ctx) {
             close_proxy = 1;
         }
         info("WebSocket connection closed");
+    } else if (ctx->use_proxy) {
+        return 3;
     } else if (strcmp(req->method, "HEAD") != 0) {
         // default response
-        unsigned long snd_len = 0;
-        unsigned long len;
         if (ctx->msg_buf != NULL) {
             ret = sock_send(client, ctx->msg_buf, ctx->content_length, 0);
             if (ret <= 0) {
                 error("Unable to send: %s", sock_strerror(client));
             }
-            snd_len += ret;
         } else if (ctx->file != NULL) {
+            unsigned long len, snd_len = 0;
             while (snd_len < ctx->content_length) {
                 len = fread(buffer, 1, CHUNK_SIZE, ctx->file);
                 if (snd_len + len > ctx->content_length) {
@@ -352,8 +352,6 @@ int respond(client_ctx_t *ctx) {
             }
         } else if (ctx->use_fastcgi) {
             return 2;
-        } else if (ctx->use_proxy) {
-            return 3;
         }
 
         if (ret < 0) {