Update Makefile

Makefile

@@ -35,7 +35,8 @@ bin/worker:
 bin/res:
 	mkdir -p bin/res
 
-bin/test: test/mock_*.c test/test_*.c src/lib/utils.c src/lib/sock.c src/lib/list.c
+bin/test: test/mock_*.c test/test_*.c \
+          src/lib/utils.c src/lib/sock.c src/lib/list.c src/lib/http.c src/lib/http_static.c src/logger.c
 	$(CC) -o $@ $(CFLAGS) $^ -lcriterion
 
 
@@ -56,15 +57,15 @@ bin/res/%.txt: res/%.*
 	echo -ne "\x00" >> $@
 
 bin/sesimos: bin/server.o bin/logger.o bin/cache_handler.o bin/async.o bin/workers.o \
-			 bin/worker/request_handler.o bin/worker/tcp_acceptor.o \
-			 bin/worker/fastcgi_handler.o bin/worker/local_handler.o bin/worker/proxy_handler.o \
-			 bin/worker/ws_frame_handler.o \
-			 bin/lib/http_static.o bin/res/default.o bin/res/proxy.o bin/res/style.o \
-			 bin/res/icon_error.o bin/res/icon_info.o bin/res/icon_success.o bin/res/icon_warning.o \
-			 bin/res/globe.o \
-			 bin/lib/compress.o bin/lib/config.o bin/lib/fastcgi.o bin/lib/geoip.o \
-			 bin/lib/http.o  bin/lib/proxy.o bin/lib/sock.o bin/lib/uri.o \
-		     bin/lib/utils.o bin/lib/websocket.o bin/lib/mpmc.o bin/lib/list.o
+             bin/worker/request_handler.o bin/worker/tcp_acceptor.o \
+             bin/worker/fastcgi_handler.o bin/worker/local_handler.o bin/worker/proxy_handler.o \
+             bin/worker/ws_frame_handler.o \
+             bin/lib/http_static.o bin/res/default.o bin/res/proxy.o bin/res/style.o \
+             bin/res/icon_error.o bin/res/icon_info.o bin/res/icon_success.o bin/res/icon_warning.o \
+             bin/res/globe.o \
+             bin/lib/compress.o bin/lib/config.o bin/lib/fastcgi.o bin/lib/geoip.o \
+             bin/lib/http.o  bin/lib/proxy.o bin/lib/sock.o bin/lib/uri.o \
+             bin/lib/utils.o bin/lib/websocket.o bin/lib/mpmc.o bin/lib/list.o
 	$(CC) -o $@ $^ $(CFLAGS) $(LDFLAGS)
 
 

src/async.c

@@ -11,32 +11,72 @@
 #include "lib/list.h"
 
 #include <poll.h>
+#include <sys/epoll.h>
 #include <signal.h>
 #include <errno.h>
 #include <memory.h>
 #include <pthread.h>
 #include <semaphore.h>
+#include <unistd.h>
+
+#define ASYNC_MAX_EVENTS 16
 
 typedef struct {
     int fd;
     sock *socket;
-    short events;
+    async_evt_t events;
     int flags;
-    void (*cb)(void *);
     void *arg;
+    void (*cb)(void *);
     void (*err_cb)(void *);
-    void *err_arg;
 } evt_listen_t;
 
 typedef struct {
     int n;
-    evt_listen_t q[64];
+    evt_listen_t *q[ASYNC_MAX_EVENTS];
 } listen_queue_t;
 
 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 epoll_fd;
+
+static short async_a2p(async_evt_t events) {
+    short ret = 0;
+    if (events & ASYNC_IN)  ret |= POLLIN;
+    if (events & ASYNC_PRI) ret |= POLLPRI;
+    if (events & ASYNC_OUT) ret |= POLLOUT;
+    return ret;
+}
+
+static unsigned int async_a2e(async_evt_t events) {
+    unsigned int ret = 0;
+    if (events & ASYNC_IN)  ret |= EPOLLIN;
+    if (events & ASYNC_PRI) ret |= EPOLLPRI;
+    if (events & ASYNC_OUT) ret |= EPOLLOUT;
+    return ret;
+}
+
+static async_evt_t async_p2a(short events) {
+    async_evt_t ret = 0;
+    if (events & POLLIN)   ret |= ASYNC_IN;
+    if (events & POLLPRI)  ret |= ASYNC_PRI;
+    if (events & POLLOUT)  ret |= ASYNC_OUT;
+    if (events & POLLERR)  ret |= ASYNC_ERR;
+    if (events & POLLHUP)  ret |= ASYNC_HUP;
+    return ret;
+}
+
+static async_evt_t async_e2a(unsigned int events) {
+    async_evt_t ret = 0;
+    if (events & EPOLLIN)   ret |= ASYNC_IN;
+    if (events & EPOLLPRI)  ret |= ASYNC_PRI;
+    if (events & EPOLLOUT)  ret |= ASYNC_OUT;
+    if (events & EPOLLERR)  ret |= ASYNC_ERR;
+    if (events & EPOLLHUP)  ret |= ASYNC_HUP;
+    return ret;
+}
 
 static int async_add_to_queue(evt_listen_t *evt) {
     try_again:
@@ -48,19 +88,26 @@ static int async_add_to_queue(evt_listen_t *evt) {
         }
     }
 
-    memcpy(&listen_q->q[listen_q->n++], evt, sizeof(*evt));
+    evt_listen_t *ptr = malloc(sizeof(evt_listen_t));
+    if (ptr == NULL) {
+        sem_post(&lock);
+        return -1;
+    }
+
+    memcpy(ptr, evt, sizeof(*evt));
+    listen_q->q[listen_q->n++] = ptr;
 
     sem_post(&lock);
 
     return 0;
 }
 
-static int async_exec(evt_listen_t *evt, short r_events) {
+static int async_exec(evt_listen_t *evt, async_evt_t r_events) {
     int ret, e = errno;
     if (r_events & evt->events) {
         // specified event(s) occurred
         if (evt->socket && !sock_has_pending(evt->socket)) {
-            evt->err_cb(evt->err_arg);
+            evt->err_cb(evt->arg);
             ret = 0;
         } else {
             evt->cb(evt->arg);
@@ -68,7 +115,7 @@ static int async_exec(evt_listen_t *evt, short r_events) {
         }
     } else if (r_events & (POLLERR | POLLHUP | POLLNVAL)) {
         // error occurred
-        evt->err_cb(evt->err_arg);
+        evt->err_cb(evt->arg);
         ret = 0;
     } else {
         // no event occurred
@@ -81,12 +128,15 @@ static int async_exec(evt_listen_t *evt, short r_events) {
 }
 
 static int async_check(evt_listen_t *evt) {
-    struct pollfd fds[1] = {{.fd = evt->fd, .events = evt->events}};
+    struct pollfd fds[1] = {{
+        .fd = evt->fd,
+        .events = async_a2p(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)
+        // fd already ready
+        if (async_exec(evt, async_p2a(fds[0].revents)) == 0)
             return 1;
     }
 
@@ -104,30 +154,28 @@ static int async_add(evt_listen_t *evt) {
     return ret;
 }
 
-int async_fd(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg) {
+int async_fd(int fd, async_evt_t events, int flags, void *arg, void cb(void *), void err_cb(void *)) {
     evt_listen_t evt = {
             .fd = fd,
             .socket = NULL,
             .events = events,
             .flags = flags,
-            .cb = cb,
             .arg = arg,
+            .cb = cb,
             .err_cb = err_cb,
-            .err_arg = err_arg,
     };
     return async_add(&evt);
 }
 
-int async(sock *s, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg) {
+int async(sock *s, async_evt_t events, int flags, void *arg, void cb(void *), void err_cb(void *)) {
     evt_listen_t evt = {
             .fd = s->socket,
             .socket = s,
             .events = events,
             .flags = flags,
-            .cb = cb,
             .arg = arg,
+            .cb = cb,
             .err_cb = err_cb,
-            .err_arg = err_arg,
     };
     return async_add(&evt);
 }
@@ -140,19 +188,29 @@ int async_init(void) {
     listen1.n = 0;
     listen2.n = 0;
 
+    if ((epoll_fd = epoll_create1(0)) == -1) {
+        async_free();
+        return -1;
+    }
+
     return 0;
 }
 
 void async_free(void) {
     int e = errno;
     sem_destroy(&lock);
+    close(epoll_fd);
     errno = e;
 }
 
 void async_thread(void) {
-    evt_listen_t *local = list_create(sizeof(evt_listen_t), 16);
-
-    // TODO use epoll instead of poll
+    struct epoll_event ev, events[ASYNC_MAX_EVENTS];
+    int num_fds;
+    evt_listen_t **local = list_create(sizeof(evt_listen_t *), 16);
+    if (local == NULL) {
+        critical("Unable to create async local list");
+        return;
+    }
 
     thread = pthread_self();
 
@@ -162,23 +220,27 @@ void async_thread(void) {
         listen_queue_t *l = listen_q;
         listen_q = (listen_q == &listen1) ? &listen2 : &listen1;
 
-        // fill local list with previously added queue entries
+        // fill local list and epoll instance with previously added queue entries
         for (int i = 0; i < l->n; i++) {
-            local = list_append(local, &l->q[i]);
-        }
+            evt_listen_t *evt = l->q[i];
+            local = list_append(local, &evt);
+            if (local == NULL) {
+                critical("Unable to resize async local list");
+                return;
+            }
 
+            ev.events = async_a2e(evt->events);
+            ev.data.ptr = evt;
+
+            if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, evt->fd, &ev) == -1) {
+                critical("Unable to add file descriptor to epoll instance");
+                return;
+            }
+        }
         // reset size of queue
         l->n = 0;
 
-        // fill fds with newly added queue entries
-        int num_fds = 0;
-        struct pollfd fds[list_size(local)];
-        for (int i = 0; i < list_size(local); i++, num_fds++) {
-            fds[num_fds].fd = local[i].fd;
-            fds[num_fds].events = local[i].events;
-        }
-
-        if (poll(fds, num_fds, -1) < 0) {
+        if ((num_fds = epoll_wait(epoll_fd, events, ASYNC_MAX_EVENTS, -1)) == -1) {
             if (errno == EINTR) {
                 // interrupt
                 errno = 0;
@@ -190,13 +252,29 @@ void async_thread(void) {
             }
         }
 
-        for (int i = 0, j = 0; i < num_fds; i++, j++) {
-            evt_listen_t *evt = &local[j];
-            if (async_exec(evt, fds[i].revents) == 0)
-                local = list_remove(local, j--);
+        for (int i = 0; i < num_fds; i++) {
+            evt_listen_t *evt = events[i].data.ptr;
+            if (async_exec(evt, async_e2a(events[i].events)) == 0) {
+                if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, evt->fd, NULL) == -1) {
+                    critical("Unable to remove file descriptor from epoll instance");
+                    return;
+                }
+
+                local = list_delete(local, &evt);
+                if (local == NULL) {
+                    critical("Unable to resize async local list");
+                    return;
+                }
+
+                free(evt);
+            }
         }
     }
 
+    // cleanup
+    for (int i = 0; i < list_size(local); i++) {
+        free(local[i]);
+    }
     list_free(local);
 }
 

src/async.h

@@ -11,13 +11,22 @@
 
 #include "lib/sock.h"
 
-#include <poll.h>
-
 #define ASYNC_KEEP 1
 
-int async(sock *s, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg);
+#define ASYNC_IN   0x01
+#define ASYNC_PRI  0x02
+#define ASYNC_OUT  0x04
+#define ASYNC_ERR  0x08
+#define ASYNC_HUP  0x10
 
-int async_fd(int fd, short events, int flags, void cb(void *), void *arg, void err_cb(void *), void *err_arg);
+#define ASYNC_WAIT_READ  ASYNC_IN
+#define ASYNC_WAIT_WRITE ASYNC_OUT
+
+typedef unsigned int async_evt_t;
+
+int async(sock *s, async_evt_t events, int flags, void *arg, void cb(void *), void err_cb(void *));
+
+int async_fd(int fd, async_evt_t events, int flags, void *arg, void cb(void *), void err_cb(void *));
 
 int async_init(void);
 

src/lib/http.c

@@ -143,10 +143,81 @@ int http_parse_header_field(http_hdr *hdr, const char *buf, const char *end_ptr,
     return 0;
 }
 
-int http_receive_request(sock *client, http_req *req) {
-    long rcv_len, len;
-    char buf[CLIENT_MAX_HEADER_SIZE];
+int http_parse_request(char *buf, http_req *req) {
     char *ptr, *pos0 = buf, *pos1, *pos2;
+    long len;
+
+    unsigned long header_len = strstr(buf, "\r\n\r\n") - buf + 4;
+    if (header_len <= 0) {
+        error("Unable to parse http header: End of header not found");
+        return -5;
+    }
+
+    for (int i = 0; i < header_len; i++) {
+        if ((buf[i] >= 0x00 && buf[i] <= 0x1F && buf[i] != '\r' && buf[i] != '\n') || buf[i] == 0x7F) {
+            error("Unable to parse http header: Header contains illegal characters");
+            return -4;
+        }
+    }
+
+    ptr = buf;
+    while (header_len > (ptr - buf + 2)) {
+        pos0 = strstr(ptr, "\r\n");
+        if (pos0 == NULL) {
+            error("Unable to parse http header: Invalid header format");
+            return -1;
+        }
+
+        if (req->version[0] == 0) {
+            pos1 = (char *) strchr(ptr, ' ') + 1;
+            if (pos1 == NULL) goto err_hdr_fmt;
+
+            if (pos1 - ptr - 1 >= sizeof(req->method)) {
+                error("Unable to parse http header: Method name too long");
+                return -2;
+            }
+
+            for (int i = 0; i < (pos1 - ptr - 1); i++) {
+                if (ptr[i] < 'A' || ptr[i] > 'Z') {
+                    error("Unable to parse http header: Invalid method");
+                    return -2;
+                }
+            }
+            snprintf(req->method, sizeof(req->method), "%.*s", (int) (pos1 - ptr - 1), ptr);
+
+            pos2 = (char *) strchr(pos1, ' ') + 1;
+            if (pos2 == NULL) {
+                err_hdr_fmt:
+                error("Unable to parse http header: Invalid header format");
+                return -1;
+            }
+
+            if (memcmp(pos2, "HTTP/", 5) != 0 || memcmp(pos2 + 8, "\r\n", 2) != 0) {
+                error("Unable to parse http header: Invalid version");
+                return -3;
+            }
+
+            len = pos2 - pos1 - 1;
+            req->uri = malloc(len + 1);
+            sprintf(req->uri, "%.*s", (int) len, pos1);
+            sprintf(req->version, "%.3s", pos2 + 5);
+        } else {
+            int ret = http_parse_header_field(&req->hdr, ptr, pos0, HTTP_MERGE_FIELDS);
+            if (ret != 0) return -ret;
+        }
+        ptr = pos0 + 2;
+    }
+
+    if (pos0[2] == '\r' && pos0[3] == '\n') {
+        return (int) header_len;
+    }
+
+    return -1;
+}
+
+int http_receive_request(sock *client, http_req *req) {
+    long rcv_len;
+    char buf[CLIENT_MAX_HEADER_SIZE];
     memset(buf, 0, sizeof(buf));
     memset(req->method, 0, sizeof(req->method));
     memset(req->version, 0, sizeof(req->version));
@@ -154,79 +225,20 @@ int http_receive_request(sock *client, http_req *req) {
     req->hdr.last_field_num = -1;
     http_init_hdr(&req->hdr);
 
-    while (1) {
-        rcv_len = sock_recv(client, buf, CLIENT_MAX_HEADER_SIZE, MSG_PEEK);
-        if (rcv_len <= 0) {
-            error("Unable to receive http header: %s", sock_strerror(client));
-            return -1;
-        }
-
-        unsigned long header_len = strstr(buf, "\r\n\r\n") - buf + 4;
-        if (header_len <= 0) {
-            error("Unable to parse http header: End of header not found");
-            return 5;
-        } else {
-            rcv_len = sock_recv(client, buf, header_len, 0);
-        }
-
-        for (int i = 0; i < header_len; i++) {
-            if ((buf[i] >= 0x00 && buf[i] <= 0x1F && buf[i] != '\r' && buf[i] != '\n') || buf[i] == 0x7F) {
-                error("Unable to parse http header: Header contains illegal characters");
-                return 4;
-            }
-        }
-
-        ptr = buf;
-        while (header_len > (ptr - buf + 2)) {
-            pos0 = strstr(ptr, "\r\n");
-            if (pos0 == NULL) {
-                error("Unable to parse http header: Invalid header format");
-                return 1;
-            }
-
-            if (req->version[0] == 0) {
-                pos1 = (char *) memchr(ptr, ' ', rcv_len - (ptr - buf)) + 1;
-                if (pos1 == NULL) goto err_hdr_fmt;
-
-                if (pos1 - ptr - 1 >= sizeof(req->method)) {
-                    error("Unable to parse http header: Method name too long");
-                    return 2;
-                }
-
-                for (int i = 0; i < (pos1 - ptr - 1); i++) {
-                    if (ptr[i] < 'A' || ptr[i] > 'Z') {
-                        error("Unable to parse http header: Invalid method");
-                        return 2;
-                    }
-                }
-                snprintf(req->method, sizeof(req->method), "%.*s", (int) (pos1 - ptr - 1), ptr);
-
-                pos2 = (char *) memchr(pos1, ' ', rcv_len - (pos1 - buf)) + 1;
-                if (pos2 == NULL) {
-                    err_hdr_fmt:
-                    error("Unable to parse http header: Invalid header format");
-                    return 1;
-                }
-
-                if (memcmp(pos2, "HTTP/", 5) != 0 || memcmp(pos2 + 8, "\r\n", 2) != 0) {
-                    error("Unable to parse http header: Invalid version");
-                    return 3;
-                }
-
-                len = pos2 - pos1 - 1;
-                req->uri = malloc(len + 1);
-                sprintf(req->uri, "%.*s", (int) len, pos1);
-                sprintf(req->version, "%.3s", pos2 + 5);
-            } else {
-                int ret = http_parse_header_field(&req->hdr, ptr, pos0, HTTP_MERGE_FIELDS);
-                if (ret != 0) return ret;
-            }
-            ptr = pos0 + 2;
-        }
-        if (pos0[2] == '\r' && pos0[3] == '\n') {
-            break;
-        }
+    rcv_len = sock_recv(client, buf, CLIENT_MAX_HEADER_SIZE - 1, MSG_PEEK);
+    if (rcv_len <= 0) {
+        error("Unable to receive http header: %s", sock_strerror(client));
+        return -1;
     }
+    buf[rcv_len] = 0;
+
+    long header_len = http_parse_request(buf, req);
+    if (header_len < 0)
+        return (int) -header_len;
+
+    rcv_len = sock_recv(client, buf, header_len, 0);
+    if (rcv_len != header_len)
+        return -1;
 
     return 0;
 }

src/lib/http.h

@@ -134,6 +134,8 @@ void http_free_req(http_req *req);
 
 void http_free_res(http_res *res);
 
+int http_parse_request(char *buf, http_req *req);
+
 int http_receive_request(sock *client, http_req *req);
 
 int http_parse_header_field(http_hdr *hdr, const char *buf, const char *end_ptr, int flags);

src/lib/list.c

@@ -43,6 +43,19 @@ int list_size(const void *list_ptr) {
     return list->size;
 }
 
+int list_find(void *list_ptr, void *elem) {
+    list_meta_t *list = (void *) ((unsigned char *) list_ptr - sizeof(list_meta_t));
+    unsigned char *array = list_ptr;
+
+    for (int i = 0; i < list->size; i++) {
+        if (memcmp(array + i * list->elem_size, elem, list->elem_size) == 0) {
+            return i;
+        }
+    }
+
+    return -1;
+}
+
 void *list_insert(void *list_ptr, void *elem, int n) {
     void *ptr = NULL;
     list_ptr = list_insert_ptr(list_ptr, &ptr, n);
@@ -94,7 +107,7 @@ void *list_remove(void *list_ptr, int n) {
         memmove(array + n * list->elem_size, array + (n + 1) * list->elem_size, (list->size - n - 1) * list->elem_size);
 
     list->size--;
-    if (list->size < list->max_size / FACTOR && list->max_size / FACTOR >= list->init_size) {
+    if (list->size < list->max_size / FACTOR / 2 && list->max_size / FACTOR >= list->init_size) {
         if ((list = list_resize(list, list->max_size / FACTOR)) == NULL) {
             return NULL;
         }
@@ -103,6 +116,15 @@ void *list_remove(void *list_ptr, int n) {
     return (unsigned char *) list + sizeof(list_meta_t);
 }
 
+void *list_delete(void *list_ptr, void *elem) {
+    int idx = list_find(list_ptr, elem);
+    if (idx == -1) {
+        return list_ptr;
+    } else {
+        return list_remove(list_ptr, idx);
+    }
+}
+
 void *list_clear(void *list_ptr) {
     list_meta_t *list = (void *) ((unsigned char *) list_ptr - sizeof(list_meta_t));
     list->size = 0;

src/lib/list.h

@@ -6,6 +6,8 @@ void *list_create(int elem_size, int init_elem_n);
 
 int list_size(const void *list_ptr);
 
+int list_find(void *list_ptr, void *elem);
+
 void *list_insert(void *list_ptr, void *elem, int n);
 
 void *list_insert_ptr(void *list_ptr, void **elem, int n);
@@ -16,6 +18,8 @@ void *list_append_ptr(void *list_ptr, void **elem);
 
 void *list_remove(void *list_ptr, int n);
 
+void *list_delete(void *list_ptr, void *elem);
+
 void *list_clear(void *list_ptr);
 
 void list_free(void *list_ptr);

src/lib/utils.c

@@ -210,3 +210,10 @@ long clock_micros(void) {
     clock_gettime(CLOCK_MONOTONIC, &time);
     return time.tv_sec * 1000000 + time.tv_nsec / 1000;
 }
+
+
+long clock_cpu(void) {
+    struct timespec time;
+    clock_gettime(CLOCK_THREAD_CPUTIME_ID, &time);
+    return time.tv_sec * 1000000000 + time.tv_nsec;
+}

src/lib/utils.h

@@ -39,4 +39,6 @@ int base64_encode(void *data, unsigned long data_len, char *output, unsigned lon
 
 long clock_micros(void);
 
+long clock_cpu(void);
+
 #endif //SESIMOS_UTILS_H

src/server.c

@@ -296,7 +296,7 @@ int main(int argc, char *const argv[]) {
     workers_init();
 
     for (int i = 0; i < NUM_SOCKETS; i++) {
-        async_fd(sockets[i], POLLIN, ASYNC_KEEP, accept_cb, &sockets[i], accept_err_cb, &sockets[i]);
+        async_fd(sockets[i], ASYNC_WAIT_READ, ASYNC_KEEP, &sockets[i], accept_cb, accept_err_cb);
     }
 
     notice("Ready to accept connections");

src/workers.c

@@ -53,7 +53,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, POLLIN, 0, (void (*)(void *)) handle_request_cb, ctx, (void (*)(void *)) tcp_close, ctx);
+        return async(&ctx->socket, ASYNC_WAIT_READ, 0, ctx, (void (*)(void *)) handle_request_cb, (void (*)(void *)) tcp_close);
     } else {
         tcp_close(ctx);
         return 0;
@@ -77,5 +77,5 @@ static int ws_handle_frame_cb(ws_ctx_t *ctx) {
 }
 
 int ws_handle_frame(ws_ctx_t *ctx) {
-    return async(ctx->socket, POLLIN, 0, (void (*)(void *)) ws_handle_frame_cb, ctx, (void (*)(void *)) ws_close, ctx);
+    return async(ctx->socket, ASYNC_WAIT_READ, 0, ctx, (void (*)(void *)) ws_handle_frame_cb, (void (*)(void *)) ws_close);
 }