mpmc: Use volatile modifier on buffer differently

src/async.c

@@ -21,7 +21,8 @@
 #include <unistd.h>
 #include <openssl/ssl.h>
 
-#define ASYNC_MAX_EVENTS 16
+#define ASYNC_QUEUE_MAX_EVENTS 256
+#define ASYNC_EPOLL_MAX_EVENTS 64
 
 typedef struct {
     int fd;
@@ -36,7 +37,7 @@ typedef struct {
 
 typedef struct {
     int n;
-    evt_listen_t *q[ASYNC_MAX_EVENTS];
+    evt_listen_t *q[ASYNC_QUEUE_MAX_EVENTS];
 } listen_queue_t;
 
 static volatile listen_queue_t listen1, listen2, *listen_q = &listen1;
@@ -123,7 +124,13 @@ static int async_add_to_queue(evt_listen_t *evt) {
         }
     }
 
-    evt_listen_t *ptr = malloc(sizeof(evt_listen_t));
+    if (listen_q->n >= ASYNC_QUEUE_MAX_EVENTS) {
+        sem_post(&lock);
+        errno = ENOBUFS;
+        return -1;
+    }
+
+    evt_listen_t *ptr = malloc(sizeof(*evt));
     if (ptr == NULL) {
         sem_post(&lock);
         return -1;
@@ -198,9 +205,12 @@ 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)
+    int ret;
+    if ((ret = async_add_to_queue(evt)) != 0) {
+        alert("Unable to add event to async queue");
+    } else if (thread != -1) {
         pthread_kill(thread, SIGUSR1);
+    }
 
     return ret;
 }
@@ -257,7 +267,7 @@ void async_free(void) {
 }
 
 void async_thread(void) {
-    struct epoll_event ev, events[ASYNC_MAX_EVENTS];
+    struct epoll_event ev, events[ASYNC_EPOLL_MAX_EVENTS];
     int num_fds, idx;
     long ts, min_ts, cur_ts;
     volatile listen_queue_t *l;
@@ -343,7 +353,7 @@ void async_thread(void) {
         }
 
         // epoll is used in level-triggered mode, so buffers are taken into account
-        if ((num_fds = epoll_wait(epoll_fd, events, ASYNC_MAX_EVENTS, (int) (min_ts / 1000))) == -1) {
+        if ((num_fds = epoll_wait(epoll_fd, events, ASYNC_EPOLL_MAX_EVENTS, (int) (min_ts / 1000))) == -1) {
             if (errno == EINTR) {
                 // interrupt
                 errno = 0;

src/lib/http.c

@@ -132,7 +132,7 @@ int http_parse_header_field(http_hdr *hdr, const char *buf, const char *end_ptr,
     if (hdr->last_field_num > list_size(hdr->fields))
         return http_error(HTTP_ERROR_GENERAL);
 
-    char *pos1 = (char *) buf, *pos2 = (char *) end_ptr;
+    const char *pos1 = buf, *pos2 = end_ptr;
     if (buf[0] == ' ' || buf[0] == '\t') {
         if (hdr->last_field_num == -1)
             return http_error(HTTP_ERROR_GENERAL);

src/lib/mpmc.c

@@ -10,7 +10,7 @@
 
 static void *mpmc_worker(void *arg);
 
-int mpmc_init(mpmc_t *ctx, int n_workers, int buf_size, void (*consumer)(void *obj), const char *name) {
+int mpmc_init(mpmc_t *ctx, const int n_workers, const int buf_size, void (*consumer)(void *), const char *name) {
     ctx->alive = 1;
     ctx->n_workers = n_workers;
     ctx->size = buf_size, ctx->max_size = buf_size;
@@ -35,7 +35,7 @@ int mpmc_init(mpmc_t *ctx, int n_workers, int buf_size, void (*consumer)(void *o
         return -1;
     }
 
-    memset(ctx->buffer,   0, ctx->size      * sizeof(void *));
+    memset((void *)ctx->buffer, 0, ctx->size      * sizeof(void *));
     memset(ctx->workers,       -1, ctx->n_workers * sizeof(pthread_t));
 
     for (int i = 0; i < ctx->n_workers; i++) {
@@ -72,7 +72,7 @@ int mpmc_queue(mpmc_t *ctx, void *obj) {
         }
     }
 
-    int p = ctx->wr;
+    const int p = ctx->wr;
     ctx->wr = (ctx->wr + 1) % ctx->size;
 
     // unlock wr field
@@ -121,7 +121,7 @@ static void *mpmc_worker(void *arg) {
             }
         }
 
-        int p = ctx->rd;
+        const int p = ctx->rd;
         ctx->rd = (ctx->rd + 1) % ctx->size;
 
         // unlock rd field
@@ -143,9 +143,9 @@ void mpmc_stop(mpmc_t *ctx) {
 }
 
 void mpmc_destroy(mpmc_t *ctx) {
-    int e = errno;
+    const int e = errno;
 
-    // stop threads, if running
+    // stop threads if running
     mpmc_stop(ctx);
     for (int i = 0; i < ctx->n_workers; i++) {
         if (ctx->workers[i] == -1) break;
@@ -158,7 +158,7 @@ void mpmc_destroy(mpmc_t *ctx) {
     sem_destroy(&ctx->used);
     sem_destroy(&ctx->lck_rd);
     sem_destroy(&ctx->lck_wr);
-    free(ctx->buffer);
+    free((void *)ctx->buffer);
     free(ctx->workers);
 
     // reset errno

src/lib/mpmc.h

@@ -5,18 +5,18 @@
 #include <semaphore.h>
 
 typedef struct {
-    unsigned char alive;
+    volatile unsigned int alive:1;
     int n_workers;
-    int rd, wr;
+    volatile int rd, wr;
     sem_t free, used, lck_rd, lck_wr;
     int size, max_size;
-    void **buffer;
+    void *volatile *buffer;
     pthread_t *workers;
     void (*consumer)(void *obj);
     const char* name;
 } mpmc_t;
 
-int mpmc_init(mpmc_t *ctx, int n_workers, int buf_size, void (*consumer)(void *obj), const char *name);
+int mpmc_init(mpmc_t *ctx, int n_workers, int buf_size, void (*consumer)(void *), const char *name);
 
 int mpmc_queue(mpmc_t *ctx, void *obj);
 

src/lib/proxy.c

@@ -211,13 +211,13 @@ int proxy_request_header(http_req *req, sock *sock) {
         if (forwarded == NULL) {
             http_add_header_field(&req->hdr, "X-Forwarded-Host", http_get_header_field(&req->hdr, "Host"));
         } else {
-            char *ptr = strchr(forwarded, ',');
+            const char *ptr = strchr(forwarded, ',');
             unsigned long len;
             if (ptr != NULL) len = ptr - forwarded;
             else len = strlen(forwarded);
             ptr = strstr(forwarded, "host=");
             if ((ptr - forwarded) < len) {
-                char *end = strchr(ptr, ';');
+                const char *end = strchr(ptr, ';');
                 if (end == NULL) len -= (ptr - forwarded);
                 else len = (end - ptr);
                 len -= 5;
@@ -233,13 +233,13 @@ int proxy_request_header(http_req *req, sock *sock) {
         if (forwarded == NULL) {
             http_add_header_field(&req->hdr, "X-Forwarded-Proto", sock->enc ? "https" : "http");
         } else {
-            char *ptr = strchr(forwarded, ',');
+            const char *ptr = strchr(forwarded, ',');
             unsigned long len;
             if (ptr != NULL) len = ptr - forwarded;
             else len = strlen(forwarded);
             ptr = strstr(forwarded, "proto=");
             if ((ptr - forwarded) < len) {
-                char *end = strchr(ptr, ';');
+                const char *end = strchr(ptr, ';');
                 if (end == NULL) len -= (ptr - forwarded);
                 else len = (end - ptr);
                 len -= 6;

src/lib/sock.c

@@ -430,7 +430,7 @@ long sock_recv_chunk_header(sock *s) {
         buf_ptr[ret1] = 0;
 
         if ((ret2 = parse_chunk_header(buf, (buf_ptr - buf) + ret1, &len)) == -1) {
-            if (errno == EPROTO) {
+            if (errno != EAGAIN) {
                 return -1;
             } else {
                 if (sock_recv_x(s, buf_ptr, ret1, 0) == -1)

src/lib/uri.c

@@ -38,7 +38,7 @@ int path_exists(const char *path) {
     return ret == 0;
 }
 
-int uri_init(http_uri *uri, const char *webroot, const char *uri_str, int dir_mode) {
+int uri_init(http_uri *uri, const char *webroot, char *uri_str, int dir_mode) {
     char buf0[1024], buf1[1024], buf2[1024], buf3[1024], buf4[1024];
     int p_len;
 

src/lib/uri.h

@@ -42,7 +42,7 @@ typedef struct {
 } http_uri;
 
 
-int uri_init(http_uri *uri, const char *webroot, const char *uri_str, int dir_mode);
+int uri_init(http_uri *uri, const char *webroot, char *uri_str, int dir_mode);
 
 int uri_init_cache(http_uri *uri);
 

src/lib/utils.c

@@ -193,7 +193,7 @@ int strcpy_rem_webroot(char *dst, const char *src, const char *webroot) {
     return 0;
 }
 
-int str_trim(char **start, char **end) {
+int str_trim(const char **start, const char **end) {
     if (start == NULL || end == NULL || *start == NULL || *end == NULL)
         return -1;
 
@@ -204,7 +204,7 @@ int str_trim(char **start, char **end) {
     return 0;
 }
 
-int str_trim_lws(char **start, char **end) {
+int str_trim_lws(const char **start, const char **end) {
     if (start == NULL || end == NULL || *start == NULL || *end == NULL)
         return -1;
 
@@ -424,9 +424,9 @@ long ftelll(FILE *file) {
     return lines + 1;
 }
 
-long parse_chunk_header(const char *buf, size_t len, size_t *ret_len) {
+long parse_chunk_header(const char *buf, const size_t len, size_t *ret_len) {
     for (int i = 0; i < len; i++) {
-        char ch = buf[i];
+        const char ch = buf[i];
         if (ch == '\r') {
             continue;
         } else if (ch == '\n' && i > 1 && buf[i - 1] == '\r') {
@@ -437,5 +437,6 @@ long parse_chunk_header(const char *buf, size_t len, size_t *ret_len) {
             return -1;
         }
     }
+    errno = EAGAIN;
     return -1;
 }

src/lib/utils.h

@@ -33,9 +33,9 @@ int mime_is_text(const char *restrict type);
 
 int strcpy_rem_webroot(char *dst, const char *str, const char *webroot);
 
-int str_trim(char **start, char **end);
+int str_trim(const char **start, const char **end);
 
-int str_trim_lws(char **start, char **end);
+int str_trim_lws(const char **start, const char **end);
 
 int streq(const char *restrict str1, const char *restrict str2);
 

src/server.c

@@ -411,7 +411,19 @@ int main(int argc, char *const argv[]) {
 
     logger_set_name("main");
 
-    workers_init();
+    if (workers_init() != 0) {
+        critical("Unable to initialize workers");
+        ssl_free();
+        list_free(clients);
+        sem_destroy(&sem_clients_lock);
+        geoip_free();
+        proxy_unload();
+        cache_join();
+        async_free();
+        logger_stop();
+        logger_join();
+        return 1;
+    }
 
     for (int i = 0; i < NUM_SOCKETS; i++) {
         async_fd(sockets[i], ASYNC_WAIT_READ, ASYNC_KEEP, &sockets[i], accept_cb, accept_err_cb, accept_err_cb);

src/server.h

@@ -12,8 +12,8 @@
 #include "worker/func.h"
 
 #define NUM_SOCKETS 2
-#define LISTEN_BACKLOG 16
-#define REQ_PER_CONNECTION 200
+#define LISTEN_BACKLOG 256
+#define REQ_PER_CONNECTION 256
 
 #define SOCKET_TIMEOUT 1
 #define CLIENT_TIMEOUT 3600
@@ -21,9 +21,6 @@
 #define SERVER_SOCKET_TIMEOUT_RES 60
 #define SERVER_TIMEOUT 3600
 
-#define CNX_HANDLER_WORKERS 8
-#define REQ_HANDLER_WORKERS 16
-
 void server_free_client(client_ctx_t *ctx);
 
 #endif //SESIMOS_SERVER_H

src/worker/fastcgi_handler.c

@@ -16,30 +16,30 @@
 #include <errno.h>
 #include <unistd.h>
 
-static int fastcgi_handler_1(client_ctx_t *ctx, fastcgi_cnx_t **fcgi_cnx);
-static int fastcgi_handler_2(client_ctx_t *ctx, fastcgi_cnx_t *fcgi_cnx);
+static int fastcgi_handler_1(client_ctx_t *ctx);
+static int fastcgi_handler_2(client_ctx_t *ctx);
 
 void fastcgi_handler_func(client_ctx_t *ctx) {
     logger_set_prefix("[%s%*s%s]%s", BLD_STR, ADDRSTRLEN, ctx->req_host, CLR_STR, ctx->log_prefix);
 
-    if (!ctx->chunks_transferred) {
-        fastcgi_cnx_t *fcgi_cnx = NULL;
-        int ret = fastcgi_handler_1(ctx, &fcgi_cnx);
+    const int ret = fastcgi_handler_1(ctx);
     respond(ctx);
+
     if (ret == 0) {
-            fastcgi_handler_2(ctx, fcgi_cnx);
+        fastcgi_handler_2(ctx);
         return;
-        } else if (ctx->fcgi_ctx != NULL) {
+    }
+
+    if (ctx->fcgi_ctx != NULL) {
         fastcgi_close(ctx->fcgi_ctx);
         ctx->fcgi_ctx = NULL;
     }
-    }
 
     request_complete(ctx);
     handle_request(ctx);
 }
 
-static int fastcgi_handler_1(client_ctx_t *ctx, fastcgi_cnx_t **fcgi_cnx) {
+static int fastcgi_handler_1(client_ctx_t *ctx) {
     http_res *res = &ctx->res;
     http_req *req = &ctx->req;
     http_uri *uri = &ctx->uri;
@@ -69,8 +69,8 @@ static int fastcgi_handler_1(client_ctx_t *ctx, fastcgi_cnx_t **fcgi_cnx) {
         return 3;
     }
 
-    (*fcgi_cnx) = &fcgi_cnx_buf;
-    fastcgi_handle_connection(ctx, fcgi_cnx);
+    fastcgi_cnx_t *fcgi_cnx = &fcgi_cnx_buf;
+    fastcgi_handle_connection(ctx, &fcgi_cnx);
 
     int expect_100_continue = 0;
     const char *client_expect = http_get_header_field(&req->hdr, "Expect");
@@ -89,13 +89,13 @@ static int fastcgi_handler_1(client_ctx_t *ctx, fastcgi_cnx_t **fcgi_cnx) {
             http_send_100_continue(client);
         }
         unsigned long client_content_len = strtoul(client_content_length, NULL, 10);
-        ret = fastcgi_receive(*fcgi_cnx, client, client_content_len);
+        ret = fastcgi_receive(fcgi_cnx, client, client_content_len);
     } else if (strcontains(client_transfer_encoding, "chunked")) {
         if (expect_100_continue) {
             info(HTTP_1XX_STR "100 Continue" CLR_STR);
             http_send_100_continue(client);
         }
-        ret = fastcgi_receive_chunked(*fcgi_cnx, client);
+        ret = fastcgi_receive_chunked(fcgi_cnx, client);
     } else if (expect_100_continue) {
         fastcgi_close_cnx((&fcgi_cnx_buf));
         res->status = http_get_status(417);
@@ -112,9 +112,9 @@ static int fastcgi_handler_1(client_ctx_t *ctx, fastcgi_cnx_t **fcgi_cnx) {
         res->status = http_get_status(502);
         return 2;
     }
-    fastcgi_close_stdin(*fcgi_cnx);
+    fastcgi_close_stdin(fcgi_cnx);
 
-    if ((ret = fastcgi_header(*fcgi_cnx, res, err_msg)) != 0) {
+    if ((ret = fastcgi_header(fcgi_cnx, res, err_msg)) != 0) {
         if (ret == -1) res->status = http_get_status(502);
         return ret;
     }
@@ -147,7 +147,7 @@ static int fastcgi_handler_1(client_ctx_t *ctx, fastcgi_cnx_t **fcgi_cnx) {
         ctx->content_length != -1 &&
         ctx->content_length < sizeof(ctx->msg_content))
     {
-        fastcgi_dump(*fcgi_cnx, ctx->msg_content, ctx->content_length);
+        fastcgi_dump(fcgi_cnx, ctx->msg_content, ctx->content_length);
         ctx->msg_content[ctx->content_length] = 0;
         return 1;
     }
@@ -163,12 +163,15 @@ static int fastcgi_handler_1(client_ctx_t *ctx, fastcgi_cnx_t **fcgi_cnx) {
 }
 
 static void fastcgi_next_cb(chunk_ctx_t *ctx) {
+    logger_set_prefix("[%s%*s%s]%s", BLD_STR, ADDRSTRLEN, ctx->client->req_host, CLR_STR, ctx->client->log_prefix);
+
     if (ctx->client->fcgi_ctx) {
         fastcgi_close(ctx->client->fcgi_ctx);
         ctx->client->fcgi_ctx = NULL;
     }
 
-    fastcgi_handle(ctx->client);
+    request_complete(ctx->client);
+    handle_request(ctx->client);
 }
 
 static void fastcgi_error_cb(chunk_ctx_t *ctx) {
@@ -178,18 +181,19 @@ static void fastcgi_error_cb(chunk_ctx_t *ctx) {
     logger_set_prefix("[%s%*s%s]%s", BLD_STR, ADDRSTRLEN, ctx->client->req_host, CLR_STR, ctx->client->log_prefix);
 
     warning("Closing connection due to FastCGI error");
+    errno = 0;
+
     if(ctx->client->fcgi_ctx) {
         fastcgi_close_error(ctx->client->fcgi_ctx);
         ctx->client->fcgi_ctx = NULL;
     }
 
+    request_complete(ctx->client);
     tcp_close(ctx->client);
-
-    errno = 0;
 }
 
-static int fastcgi_handler_2(client_ctx_t *ctx, fastcgi_cnx_t *fcgi_cnx) {
-    int chunked = strcontains(http_get_header_field(&ctx->res.hdr, "Transfer-Encoding"), "chunked");
-    handle_chunks(ctx, &fcgi_cnx->out, chunked ? SOCK_CHUNKED : 0, fastcgi_next_cb, fastcgi_error_cb);
+static int fastcgi_handler_2(client_ctx_t *ctx) {
+    const int chunked = strcontains(http_get_header_field(&ctx->res.hdr, "Transfer-Encoding"), "chunked");
+    handle_chunks(ctx, &ctx->fcgi_ctx->cnx.out, chunked ? SOCK_CHUNKED : 0, fastcgi_next_cb, fastcgi_error_cb);
     return 1;
 }

src/worker/func.h

@@ -17,6 +17,12 @@
 #include "../lib/proxy.h"
 #include "../lib/fastcgi.h"
 
+typedef struct {
+    unsigned char closed:4;
+    char log_prefix[128];
+    fastcgi_cnx_t cnx;
+} fastcgi_ctx_t;
+
 typedef struct {
     sock socket;
     int req_num;
@@ -36,7 +42,7 @@ typedef struct {
     long content_length, transferred_length;
     char *msg_buf, *msg_buf_ptr, msg_content[1024];
     proxy_ctx_t *proxy;
-    void *fcgi_ctx;
+    fastcgi_ctx_t *fcgi_ctx;
 } client_ctx_t;
 
 typedef struct {
@@ -45,12 +51,6 @@ typedef struct {
     void *other;
 } ws_ctx_t;
 
-typedef struct {
-    unsigned char closed:4;
-    char log_prefix[128];
-    fastcgi_cnx_t cnx;
-} fastcgi_ctx_t;
-
 typedef struct {
     client_ctx_t *client;
     sock *socket;

src/worker/local_handler.c

@@ -39,7 +39,7 @@ void local_handler_func(client_ctx_t *ctx) {
 static int range_handler(client_ctx_t *ctx) {
     char buf[64];
     long num0, num1, num2;
-    char *ptr;
+    const char *ptr;
     int mode;
     const char *range = http_get_header_field(&ctx->req.hdr, "Range");
 

src/worker/proxy_handler.c

@@ -118,6 +118,7 @@ static int proxy_handler_1(client_ctx_t *ctx) {
 }
 
 static void proxy_chunk_next_cb(chunk_ctx_t *ctx) {
+    logger_set_prefix("[%s%*s%s]%s", BLD_STR, ADDRSTRLEN, ctx->client->req_host, CLR_STR, ctx->client->log_prefix);
     if (proxy_unlock_ctx(ctx->client->proxy) == 1)
         proxy_peer_handle(ctx->client->proxy);
 
@@ -127,6 +128,8 @@ static void proxy_chunk_next_cb(chunk_ctx_t *ctx) {
 }
 
 static void proxy_chunk_err_cb(chunk_ctx_t *ctx) {
+    logger_set_prefix("[%s%*s%s]%s", BLD_STR, ADDRSTRLEN, ctx->client->req_host, CLR_STR, ctx->client->log_prefix);
+
     proxy_close(ctx->client->proxy);
     proxy_unlock_ctx(ctx->client->proxy);
 

src/workers.c

@@ -16,16 +16,26 @@ static mpmc_t tcp_acceptor_ctx, request_handler_ctx, local_handler_ctx, fastcgi_
               proxy_peer_handler_ctx, ws_frame_handler_ctx, chunk_handler_ctx, fastcgi_frame_handler_ctx;
 
 int workers_init(void) {
-    mpmc_init(&tcp_acceptor_ctx,          8, 64, (void (*)(void *)) tcp_acceptor_func,          "tcp");
-    mpmc_init(&request_handler_ctx,       8, 64, (void (*)(void *)) request_handler_func,       "req");
-    mpmc_init(&local_handler_ctx,         8, 64, (void (*)(void *)) local_handler_func,         "local");
-    mpmc_init(&fastcgi_handler_ctx,       8, 64, (void (*)(void *)) fastcgi_handler_func,       "fcgi");
-    mpmc_init(&proxy_handler_ctx,         8, 64, (void (*)(void *)) proxy_handler_func,         "proxy");
-    mpmc_init(&proxy_peer_handler_ctx,    1,  8, (void (*)(void *)) proxy_peer_handler_func,    "prxy_p");
-    mpmc_init(&ws_frame_handler_ctx,      8, 64, (void (*)(void *)) ws_frame_handler_func,      "ws");
-    mpmc_init(&chunk_handler_ctx,         8, 64, (void (*)(void *)) chunk_handler_func,         "chunk");
-    mpmc_init(&fastcgi_frame_handler_ctx, 8, 64, (void (*)(void *)) fastcgi_frame_handler_func, "fcgi_f");
-    return -1;
+    int ret;
+    if ((ret = mpmc_init(&tcp_acceptor_ctx,          8, 64, (void (*)(void *)) tcp_acceptor_func,          "tcp")) != 0)
+        return ret;
+    if ((ret = mpmc_init(&request_handler_ctx,       8, 64, (void (*)(void *)) request_handler_func,       "req")) != 0)
+        return ret;
+    if ((ret = mpmc_init(&local_handler_ctx,         8, 64, (void (*)(void *)) local_handler_func,         "local")) != 0)
+        return ret;
+    if ((ret = mpmc_init(&fastcgi_handler_ctx,       8, 64, (void (*)(void *)) fastcgi_handler_func,       "fcgi")) != 0)
+        return ret;
+    if ((ret = mpmc_init(&proxy_handler_ctx,         8, 64, (void (*)(void *)) proxy_handler_func,         "proxy")) != 0)
+        return ret;
+    if ((ret = mpmc_init(&proxy_peer_handler_ctx,    1,  8, (void (*)(void *)) proxy_peer_handler_func,    "prxy_p")) != 0)
+        return ret;
+    if ((ret = mpmc_init(&ws_frame_handler_ctx,      8, 64, (void (*)(void *)) ws_frame_handler_func,      "ws")) != 0)
+        return ret;
+    if ((ret = mpmc_init(&chunk_handler_ctx,         8, 64, (void (*)(void *)) chunk_handler_func,         "chunk")) != 0)
+        return ret;
+    if ((ret = mpmc_init(&fastcgi_frame_handler_ctx, 8, 64, (void (*)(void *)) fastcgi_frame_handler_func, "fcgi_f")) != 0)
+        return ret;
+    return 0;
 }
 
 void workers_stop(void) {