Add libakiyo

Signed-off-by: Andrew Opalach <andrew@akon.city>

45 files changed, 3004 insertions, 0 deletions

diff --git a/src/common.c b/src/common.c
new file mode 100644
index 0000000..b9ee5c2
--- /dev/null
+++ b/src/common.c
@@ -0,0 +1,34 @@
+#include <al/lib.h>
+#include <al/random.h>
+
+#ifndef _WIN32
+#include <unistd.h>
+#else
+#include "winwrap.h"
+#endif
+
+#include "common.h"
+#include "winwrap.h"
+
+void aki_common_init(void)
+{
+    al_malloc_init();
+    al_rand_init();
+
+    aki_windows_init();
+
+#ifndef _WIN32
+    al_set_page_size(sysconf(_SC_PAGESIZE));
+#else
+    SYSTEM_INFO info;
+    GetSystemInfo(&info);
+    al_set_page_size(info.dwPageSize);
+#endif
+}
+
+void aki_common_close(void)
+{
+    aki_windows_close();
+
+    al_malloc_close();
+}
diff --git a/src/common.h b/src/common.h
new file mode 100644
index 0000000..da4b3e3
--- /dev/null
+++ b/src/common.h
@@ -0,0 +1,4 @@
+#pragma once
+
+void aki_common_init(void);
+void aki_common_close(void);
diff --git a/src/ev_embed_compat.c b/src/ev_embed_compat.c
new file mode 100644
index 0000000..7fb39fb
--- /dev/null
+++ b/src/ev_embed_compat.c
@@ -0,0 +1,29 @@
+#define EV_STANDALONE 1
+#define EV_USE_NANOSLEEP 1
+#define EV_USE_REALTIME 1
+#ifdef _WIN32
+#define FD_SETSIZE 2048
+#define EV_SELECT_IS_WINSOCKET 1
+#else
+#define EV_AVOID_STDIO 1
+#endif
+
+#include "evwrap.h"
+#if defined(__clang__) || defined(__GNUC__)
+_Pragma("GCC diagnostic push") \
+_Pragma("GCC diagnostic ignored \"-Wstrict-aliasing\"")
+_Pragma("GCC diagnostic ignored \"-Wcomment\"")
+_Pragma("GCC diagnostic ignored \"-Wunused-value\"")
+_Pragma("GCC diagnostic ignored \"-Wparentheses\"")
+_Pragma("GCC diagnostic ignored \"-Wunused-parameter\"")
+_Pragma("GCC diagnostic ignored \"-Wsign-compare\"")
+_Pragma("GCC diagnostic ignored \"-Wreturn-type\"")
+_Pragma("GCC diagnostic ignored \"-Wdeprecated-declarations\"")
+#ifdef _WIN32
+_Pragma("GCC diagnostic ignored \"-Wextern-initializer\"")
+#endif
+#endif
+#include <ev.c>
+#if defined(__clang__) || defined(__GNUC__)
+_Pragma("GCC diagnostic pop")
+#endif
diff --git a/src/evwrap.h b/src/evwrap.h
new file mode 100644
index 0000000..331da06
--- /dev/null
+++ b/src/evwrap.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include <al/lib.h>
+
+#define EV_MULTIPLICITY 1
+
+#define EV_STAT_ENABLE 0
+//#define EV_PERIODIC_ENABLE 0
+#define EV_SIGNAL_ENABLE 0
+#define EV_CHILD_ENABLE 0
+#define EV_IDLE_ENABLE 0
+#define EV_PREPARE_ENABLE 0
+#define EV_FORK_ENABLE 0
+#define EV_CLEANUP_ENABLE 0
+#define EV_EMBED_ENABLE 0
+
+#include <ev.h>
diff --git a/src/http.c b/src/http.c
new file mode 100644
index 0000000..0036af2
--- /dev/null
+++ b/src/http.c
@@ -0,0 +1,256 @@
+#include <al/log.h>
+
+#include "http.h"
+
+#define USER_AGENT "Mozilla/5.0 (X11; Linux x86_64; rv:96.0) Gecko/20100101 Firefox/96.0"
+
+static void curl_socket_action_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+    (void)loop;
+    struct aki_http *http = (struct aki_http *)w->data;
+
+    s32 action = ((revents & EV_READ) ? CURL_CSELECT_IN : 0) | ((revents & EV_WRITE) ? CURL_CSELECT_OUT : 0);
+
+    s32 running;
+    CURLMcode mc = curl_multi_socket_action(http->multi_handle, http->sock, action, &running);
+    if (mc != CURLM_OK) {
+        al_log_error("http", "curl_multi_socket_action() failed (%s).", curl_multi_strerror(mc));
+        return;
+    }
+
+    http->handle_events(http);
+}
+
+static void set_sock(struct aki_http *http, s32 what)
+{
+    http->event.data = http;
+
+    s32 action = ((what & CURL_POLL_IN) ? EV_READ : 0) | ((what & CURL_POLL_OUT) ? EV_WRITE : 0);
+
+    if (http->event.active) ev_io_stop(http->loop->ev, &http->event);
+
+    ev_io_init(&http->event, curl_socket_action_callback, http->sock, action);
+    ev_io_start(http->loop->ev, &http->event);
+}
+
+static s32 sock_callback(CURL *e, curl_socket_t s, s32 what, void *cbp, void *sockp)
+{
+    (void)e;
+    struct aki_http *http = (struct aki_http *)cbp;
+    if (what == CURL_POLL_REMOVE) {
+        http->sock = 0;
+        ev_io_stop(http->loop->ev, &http->event);
+        curl_multi_assign(http->multi_handle, http->sock, NULL);
+    } else {
+        if (!sockp) {
+            http->sock = s;
+            set_sock(http, what);
+            curl_multi_assign(http->multi_handle, http->sock, http);
+        } else {
+            set_sock(http, what);
+        }
+    }
+    return 0;
+}
+
+static void timeout_callback(struct ev_loop *loop, ev_timer *w, s32 revents)
+{
+    (void)loop;
+    struct aki_http *http = (struct aki_http *)w->data;
+    (void)revents;
+    ev_timer_stop(http->loop->ev, w);
+    http->timer_started = false;
+    s32 running;
+    curl_multi_socket_action(http->multi_handle, CURL_SOCKET_TIMEOUT, 0, &running);
+    http->handle_events(http);
+}
+
+static s32 timer_callback(CURLM *multi, s64 timeout_ms, void *userp)
+{
+    (void)multi;
+    struct aki_http *http = (struct aki_http *)userp;
+    if (timeout_ms == -1) {
+        ev_timer_stop(http->loop->ev, &http->timer);
+        return 0;
+    }
+    if (http->timer_started) {
+        ev_timer_stop(http->loop->ev, &http->timer);
+    }
+    http->timer.data = http;
+    ev_timer_init(&http->timer, timeout_callback, timeout_ms / 1000.0, 0.0);
+    http->timer_started = true;
+    ev_timer_start(http->loop->ev, &http->timer);
+    return 0;
+}
+
+bool aki_http_init(struct aki_http *http)
+{
+    http->handle = curl_easy_init();
+    if (!http->handle) goto err;
+    //curl_easy_setopt(http->handle, CURLOPT_VERBOSE, 1L);
+    curl_easy_setopt(http->handle, CURLOPT_NOPROGRESS, 1L);
+    curl_easy_setopt(http->handle, CURLOPT_USERAGENT, USER_AGENT);
+    http->multi_handle = curl_multi_init();
+    if (!http->multi_handle) goto err;
+    curl_multi_setopt(http->multi_handle, CURLMOPT_SOCKETFUNCTION, sock_callback);
+    curl_multi_setopt(http->multi_handle, CURLMOPT_SOCKETDATA, http);
+    curl_multi_setopt(http->multi_handle, CURLMOPT_TIMERFUNCTION, timer_callback);
+    curl_multi_setopt(http->multi_handle, CURLMOPT_TIMERDATA, http);
+    http->url = NULL;
+    http->headers = NULL;
+    http->timer_started = false;
+    http->response_code = -1;
+    return true;
+err:
+    if (http->handle) curl_easy_cleanup(http->handle);
+    if (http->multi_handle) curl_multi_cleanup(http->multi_handle);
+    return false;
+}
+
+void aki_http_add_header(struct aki_http *http, str *header)
+{
+    char *c_str = al_str_to_c_str(header);
+    http->headers = curl_slist_append(http->headers, c_str);
+    al_free(c_str);
+}
+
+void aki_http_set_url(struct aki_http *http, str *url)
+{
+    if (http->url) al_free(http->url);
+    http->url = al_str_to_c_str(url);
+    curl_easy_setopt(http->handle, CURLOPT_URL, http->url);
+}
+
+void aki_http_set_range(struct aki_http *http, ssize_t start, ssize_t end)
+{
+    char range[64];
+    if (start < 0) sprintf(range, "-%zd", end);
+    else if (end < 0) sprintf(range, "%zd-", start);
+    else sprintf(range, "%zd-%zd", start, end);
+    curl_easy_setopt(http->handle, CURLOPT_RANGE, range);
+}
+
+void aki_http_close(struct aki_http *http)
+{
+    curl_easy_cleanup(http->handle);
+    curl_multi_cleanup(http->multi_handle);
+    al_free(http->url);
+}
+
+static void stream_handle_events(struct aki_http *http)
+{
+    CURLMsg *msg;
+    s32 pending;
+    struct aki_http_stream *stream = (struct aki_http_stream *)http;
+    while ((msg = curl_multi_info_read(http->multi_handle, &pending))) {
+        switch (msg->msg) {
+        case CURLMSG_DONE:
+            al_assert(msg->easy_handle == http->handle);
+            if (msg->data.result != CURLE_OK) {
+                al_log_error("http", "Error handling curl_multi event (%s).", curl_easy_strerror(msg->data.result));
+                return;
+            }
+            CURLMcode mc = curl_multi_remove_handle(http->multi_handle, msg->easy_handle);
+            if (mc != CURLM_OK) {
+                al_log_error("http", "curl_multi_remove_handle() failed (%s).", curl_multi_strerror(mc));
+                return;
+            }
+            if (http->response_code < 0) {
+                curl_easy_getinfo(http->handle, CURLINFO_RESPONSE_CODE, &http->response_code);
+            }
+            if (http->response_code == 200) {
+                stream->callback(stream->userdata, AKI_HTTP_FINISHED, NULL, http->response_code);
+            } else if (http->response_code / 100 == 3) {
+                char *redirect_url = NULL;
+                curl_easy_getinfo(http->handle, CURLINFO_REDIRECT_URL, &redirect_url);
+                if (!redirect_url) {
+                    stream->callback(stream->userdata, AKI_HTTP_ERROR, NULL, http->response_code);
+                    return;
+                }
+                aki_http_set_url(http, al_str_c(redirect_url));
+                mc = curl_multi_add_handle(http->multi_handle, http->handle);
+                if (mc != CURLM_OK) {
+                    al_log_error("http", "curl_multi_add_handle() failed (%s).", curl_multi_strerror(mc));
+                    return;
+                }
+                stream->callback(stream->userdata, AKI_HTTP_REDIRECT, NULL, http->response_code);
+            } else {
+                stream->callback(stream->userdata, AKI_HTTP_ERROR, NULL, http->response_code);
+            }
+            return;
+        case CURLMSG_LAST:
+            al_log_warn("http", "Unhandled CURLMSG_LAST.");
+            break;
+        case CURLMSG_NONE:
+            al_log_warn("http", "Unhandled CURLMSG_NONE.");
+            break;
+        }
+    }
+}
+
+bool aki_http_stream_init(struct aki_http_stream *stream)
+{
+    al_memset(stream, 0, sizeof(struct aki_http_stream));
+    if (!aki_http_init(&stream->http)) return false;
+    stream->http.handle_events = stream_handle_events;
+    return true;
+}
+
+static size_t stream_write_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
+{
+    struct aki_http_stream *stream = (struct aki_http_stream *)userdata;
+    if (stream->http.response_code < 0) {
+        curl_easy_getinfo(stream->http.handle, CURLINFO_RESPONSE_CODE, &stream->http.response_code);
+        if (stream->http.response_code > 0) {
+            stream->callback(stream->userdata, AKI_HTTP_RESPONSE_CODE, NULL, stream->http.response_code);
+            curl_easy_getinfo(stream->http.handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &stream->http.content_length);
+            if (stream->http.content_length >= 0) {
+                stream->callback(stream->userdata, AKI_HTTP_CONTENT_LENGTH, NULL, (s64)stream->http.content_length);
+            }
+        }
+    }
+    return stream->callback(stream->userdata, AKI_HTTP_WRITE, (u8 *)buffer, size * nmemb);
+}
+
+static size_t stream_read_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
+{
+    struct aki_http_stream *stream = (struct aki_http_stream *)userdata;
+    return stream->callback(stream->userdata, AKI_HTTP_READ, (u8 *)buffer, size * nmemb);
+}
+
+bool aki_http_stream_request(struct aki_http_stream *stream, u8 method, struct aki_event_loop *loop,
+    size_t (*callback)(void *, u8, u8 *, s64), void *userdata)
+{
+    al_assert(stream->http.url != NULL);
+    al_assert(stream->http.loop == NULL);
+    stream->http.loop = loop;
+    stream->callback = callback;
+    stream->userdata = userdata;
+    switch (method) {
+    case AKI_HTTP_HEAD:
+        curl_easy_setopt(stream->http.handle, CURLOPT_NOBODY, 1L);
+        break;
+    case AKI_HTTP_GET:
+        curl_easy_setopt(stream->http.handle, CURLOPT_HTTPGET, 1L);
+        break;
+    case AKI_HTTP_POST:
+        al_assert(false); // Not implemented.
+    }
+    curl_easy_setopt(stream->http.handle, CURLOPT_WRITEFUNCTION, stream_write_callback);
+    curl_easy_setopt(stream->http.handle, CURLOPT_WRITEDATA, stream);
+    curl_easy_setopt(stream->http.handle, CURLOPT_READFUNCTION, stream_read_callback);
+    curl_easy_setopt(stream->http.handle, CURLOPT_READDATA, stream);
+    // CURLM_RECURSIVE_API_CALL (8) also means we can't even call this
+    // while a callback is running on another thread.
+    CURLMcode mc = curl_multi_add_handle(stream->http.multi_handle, stream->http.handle);
+    if (mc != CURLM_OK) {
+        al_log_error("http", "curl_multi_add_handle() failed (%s).", curl_multi_strerror(mc));
+        return false;
+    }
+    return true;
+}
+
+void aki_http_stream_close(struct aki_http_stream *stream)
+{
+    aki_http_close(&stream->http);
+}
diff --git a/src/http.h b/src/http.h
new file mode 100644
index 0000000..802a7f2
--- /dev/null
+++ b/src/http.h
@@ -0,0 +1,56 @@
+#pragma once
+
+#include <curl/curl.h>
+#include <al/str.h>
+
+#include "util/buffer.h"
+
+#include "loop.h"
+
+enum {
+    AKI_HTTP_HEAD = 0,
+    AKI_HTTP_GET,
+    AKI_HTTP_POST,
+};
+
+struct aki_http {
+    ev_io event;
+    ev_timer timer;
+    bool timer_started;
+    struct aki_event_loop *loop;
+    CURL *handle;
+    CURLM *multi_handle;
+    char *url;
+    struct curl_slist *headers;
+    s64 response_code;
+    curl_off_t content_length;
+    curl_socket_t sock;
+    void (*handle_events)(struct aki_http *);
+};
+
+enum {
+    AKI_HTTP_WRITE = 0,
+    AKI_HTTP_READ,
+    AKI_HTTP_RESPONSE_CODE,
+    AKI_HTTP_CONTENT_LENGTH,
+    AKI_HTTP_FINISHED,
+    AKI_HTTP_REDIRECT,
+    AKI_HTTP_ERROR
+};
+
+struct aki_http_stream {
+    struct aki_http http;
+    size_t (*callback)(void *, u8, u8 *, s64);
+    void *userdata;
+};
+
+bool aki_http_init(struct aki_http *http);
+void aki_http_add_header(struct aki_http *http, str *header);
+void aki_http_set_url(struct aki_http *http, str *url);
+void aki_http_set_range(struct aki_http *http, ssize_t start, ssize_t end);
+void aki_http_close(struct aki_http *http);
+
+bool aki_http_stream_init(struct aki_http_stream *stream);
+bool aki_http_stream_request(struct aki_http_stream *stream, u8 method, struct aki_event_loop *loop,
+    size_t (*callback)(void *, u8, u8 *, s64), void *userdata);
+void aki_http_stream_close(struct aki_http_stream *stream);
diff --git a/src/http_server.h b/src/http_server.h
new file mode 100644
index 0000000..3f59c93
--- /dev/null
+++ b/src/http_server.h
@@ -0,0 +1,2 @@
+#pragma once
+
diff --git a/src/line_processor.c b/src/line_processor.c
new file mode 100644
index 0000000..444142a
--- /dev/null
+++ b/src/line_processor.c
@@ -0,0 +1,123 @@
+#include <al/log.h>
+
+#include "line_processor.h"
+
+// Read from a socket/fd and callback for each line.
+// This isn't meant to be used seriously. In it's current state it's just to make testing easier.
+
+#define CHUNK_SIZE 512
+#define END al_str_c("X")
+
+struct line_processor_client {
+    ev_io event;
+    struct aki_socket s;
+    size_t index;
+    struct aki_buffer buf;
+    size_t mark;
+    struct aki_line_processor *pro;
+};
+
+static void read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+    (void)loop;
+    struct line_processor_client *client = (struct line_processor_client *)w->data;
+    (void)revents;
+
+    aki_buffer_ensure_size(&client->buf, client->index + CHUNK_SIZE);
+    u8 *ptr = aki_buffer_get_ptr(&client->buf, client->index);
+    ssize_t ret;
+    if (client->pro->type == AKI_LINE_PROCESSOR_SOCKET) {
+        ret = aki_socket_read(&client->s, ptr, CHUNK_SIZE);
+    } else {
+        ret = aki_read(client->pro->fd, ptr, CHUNK_SIZE);
+    }
+    if (ret <= 0) return;
+
+    client->index += ret;
+
+    str *delim = &client->pro->delim;
+    u32 offset = delim->len - 1;
+    ptr = aki_buffer_get_ptr(&client->buf, 0);
+    for (size_t i = client->mark; i < client->index - offset; i++) {
+        if (al_str_cmp(al_str_w((char *)ptr, i, (u32)(client->index - i)), delim, 0, delim->len) == 0) {
+            str *s = al_str_w((char *)ptr, 0, i);
+            if (al_str_eq(s, END)) {
+                ev_io_stop(client->pro->loop->ev, &client->event);
+                if (client->pro->type == AKI_LINE_PROCESSOR_SOCKET) {
+                    aki_socket_close(&client->s);
+                }
+                aki_buffer_free(&client->buf);
+                al_free(client);
+                return;
+            }
+            client->pro->callback(client->pro->userdata, s);
+            size_t skip = i + offset + 1;
+            client->index = client->index - skip;
+            al_memmove(ptr, &ptr[skip], client->index);
+            client->mark = i = 0;
+        }
+    }
+
+    client->mark = client->index;
+}
+
+static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+    (void)loop;
+    struct aki_line_processor *pro = (struct aki_line_processor *)w->data;
+    (void)revents;
+    struct line_processor_client *client = al_alloc_object(struct line_processor_client);
+    client->s.config.blocking = false;
+    client->s.config.no_delay = false;
+    if (!aki_socket_accept(pro->s, &client->s)) {
+        al_free(client);
+        return;
+    }
+    client->index = 0;
+    aki_buffer_init(&client->buf);
+    client->mark = 0;
+    client->pro = pro;
+    client->event.data = client;
+    ev_io_init(&client->event, read_callback, aki_socket_get_fd(&client->s), EV_READ);
+    ev_io_start(pro->loop->ev, &client->event);
+}
+
+void aki_line_processor_init(struct aki_line_processor *pro, str *delim)
+{
+    al_str_clone(&pro->delim, delim);
+}
+
+void aki_line_processor_open_socket(struct aki_line_processor *pro, struct aki_socket *socket)
+{
+    pro->type = AKI_LINE_PROCESSOR_SOCKET;
+    pro->s = socket;
+}
+
+void aki_line_processor_open_fd(struct aki_line_processor *pro, s32 fd)
+{
+    pro->type = AKI_LINE_PROCESSOR_FD;
+    pro->fd = fd;
+}
+
+void aki_line_processor_run(struct aki_line_processor *pro, struct aki_event_loop *loop)
+{
+    pro->loop = loop;
+    switch (pro->type) {
+    case AKI_LINE_PROCESSOR_SOCKET:
+        pro->event.data = pro;
+        ev_io_init(&pro->event, socket_connection_callback, aki_socket_get_fd(pro->s), EV_READ);
+        ev_io_start(pro->loop->ev, &pro->event);
+        break;
+    case AKI_LINE_PROCESSOR_FD: { // Not supported on windows.
+        struct line_processor_client *client = al_alloc_object(struct line_processor_client);
+        client->index = 0;
+        aki_buffer_init(&client->buf);
+        client->mark = 0;
+        client->pro = pro;
+        client->event.data = client;
+        ev_io_init(&client->event, read_callback, pro->fd, EV_READ);
+        ev_io_start(pro->loop->ev, &client->event);
+        break;
+    }
+    }
+}
diff --git a/src/line_processor.h b/src/line_processor.h
new file mode 100644
index 0000000..7ce3ca3
--- /dev/null
+++ b/src/line_processor.h
@@ -0,0 +1,34 @@
+#pragma once
+
+#include <al/types.h>
+#include <al/str.h>
+#include <al/array.h>
+
+#include "util/buffer.h"
+#include "util/thread/thread.h"
+#include "util/file/file.h"
+#include "socket/socket.h"
+
+#include "loop.h"
+#include "signal.h"
+
+enum {
+    AKI_LINE_PROCESSOR_SOCKET = 0,
+    AKI_LINE_PROCESSOR_FD
+};
+
+struct aki_line_processor {
+    struct aki_event_loop *loop;
+    u8 type;
+    ev_io event;
+    struct aki_socket *s;
+    s32 fd;
+    str delim;
+    void (*callback)(void *, str *line);
+    void *userdata;
+};
+
+void aki_line_processor_init(struct aki_line_processor *pro, str *delim);
+void aki_line_processor_open_socket(struct aki_line_processor *pro, struct aki_socket *socket);
+void aki_line_processor_open_fd(struct aki_line_processor *pro, s32 fd);
+void aki_line_processor_run(struct aki_line_processor *pro, struct aki_event_loop *loop);
diff --git a/src/loop.c b/src/loop.c
new file mode 100644
index 0000000..6598768
--- /dev/null
+++ b/src/loop.c
@@ -0,0 +1,35 @@
+#include "loop.h"
+
+bool aki_event_loop_init(struct aki_event_loop *loop)
+{
+    loop->ev = ev_loop_new(EVFLAG_AUTO);
+    //ev_set_timeout_collect_interval(loop->ev, 0.1);
+    //ev_set_io_collect_interval(loop->ev, 0.05);
+    return true;
+}
+
+s32 aki_event_loop_run(struct aki_event_loop *loop)
+{
+    return ev_run(loop->ev, 0);
+}
+
+s32 aki_event_loop_run_once(struct aki_event_loop *loop)
+{
+    return ev_run(loop->ev, EVRUN_ONCE);
+}
+
+void aki_event_loop_break_one(struct aki_event_loop *loop)
+{
+    ev_break(loop->ev, EVBREAK_ONE);
+}
+
+void aki_event_loop_break(struct aki_event_loop *loop)
+{
+    ev_break(loop->ev, EVBREAK_ALL);
+}
+
+void aki_event_loop_destroy(struct aki_event_loop *loop)
+{
+    ev_loop_destroy(loop->ev);
+    loop->ev = NULL;
+}
diff --git a/src/loop.h b/src/loop.h
new file mode 100644
index 0000000..aba49d9
--- /dev/null
+++ b/src/loop.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include <al/types.h>
+
+#include "evwrap.h"
+
+struct aki_event_loop {
+    struct ev_loop *ev;
+};
+
+bool aki_event_loop_init(struct aki_event_loop *loop);
+s32 aki_event_loop_run(struct aki_event_loop *loop);
+s32 aki_event_loop_run_once(struct aki_event_loop *loop);
+void aki_event_loop_break_one(struct aki_event_loop *loop);
+void aki_event_loop_break(struct aki_event_loop *loop);
+bool aki_event_loop_is_broken(struct aki_event_loop *loop);
+void aki_event_loop_destroy(struct aki_event_loop *loop);
diff --git a/src/packet_cache.c b/src/packet_cache.c
new file mode 100644
index 0000000..e08bee7
--- /dev/null
+++ b/src/packet_cache.c
@@ -0,0 +1,78 @@
+#include "packet_cache.h"
+
+void aki_packet_cache_init(struct aki_packet_cache *cache, u32 size)
+{
+    al_array_init(cache->cache);
+    al_array_reserve(cache->cache, size);
+    aki_cond_init(&cache->cond);
+    aki_mutex_init(&cache->mutex);
+    cache->disabled = false;
+}
+
+bool aki_packet_cache_send_packet(struct aki_packet_cache *cache, struct aki_packet *packet)
+{
+    aki_mutex_lock(&cache->mutex);
+    if (cache->disabled) {
+        aki_mutex_unlock(&cache->mutex);
+        return false;
+    }
+    al_array_push(cache->cache, packet);
+    if (aki_cond_is_waiting(&cache->cond)) {
+        aki_cond_signal(&cache->cond);
+    }
+    aki_mutex_unlock(&cache->mutex);
+    return true;
+}
+
+bool aki_packet_cache_wait(struct aki_packet_cache *cache)
+{
+    aki_mutex_lock(&cache->mutex);
+    if (cache->disabled) {
+        aki_mutex_unlock(&cache->mutex);
+        return false;
+    }
+    if (cache->cache.size == 0) {
+        aki_cond_wait(&cache->cond, &cache->mutex);
+    }
+    if (cache->disabled) {
+        aki_mutex_unlock(&cache->mutex);
+        return false;
+    }
+    aki_mutex_unlock(&cache->mutex);
+    return true;
+}
+
+struct aki_packet *aki_packet_cache_pop(struct aki_packet_cache *cache)
+{
+    aki_mutex_lock(&cache->mutex);
+    struct aki_packet *packet = NULL;
+    if (cache->cache.size > 0) {
+        al_array_pop_at(cache->cache, 0, packet);
+    }
+    aki_mutex_unlock(&cache->mutex);
+    return packet;
+}
+
+void aki_packet_cache_disable(struct aki_packet_cache *cache)
+{
+    aki_mutex_lock(&cache->mutex);
+    cache->disabled = true;
+    if (aki_cond_is_waiting(&cache->cond)) {
+        aki_cond_signal(&cache->cond);
+    }
+    aki_mutex_unlock(&cache->mutex);
+}
+
+void aki_packet_cache_enable(struct aki_packet_cache *cache)
+{
+    aki_mutex_lock(&cache->mutex);
+    cache->disabled = false;
+    aki_mutex_unlock(&cache->mutex);
+}
+
+void aki_packet_cache_free(struct aki_packet_cache *cache)
+{
+    al_array_free(cache->cache);
+    aki_mutex_destroy(&cache->mutex);
+    aki_cond_destroy(&cache->cond);
+}
diff --git a/src/packet_cache.h b/src/packet_cache.h
new file mode 100644
index 0000000..980c586
--- /dev/null
+++ b/src/packet_cache.h
@@ -0,0 +1,24 @@
+#pragma once
+
+#include <al/array.h>
+
+#include "util/packet.h"
+#include "util/thread/thread.h"
+
+#include "loop.h"
+#include "evwrap.h"
+
+struct aki_packet_cache {
+    array(struct aki_packet *) cache;
+    struct aki_cond cond;
+    struct aki_mutex mutex;
+    bool disabled;
+};
+
+void aki_packet_cache_init(struct aki_packet_cache *cache, u32 size);
+bool aki_packet_cache_send_packet(struct aki_packet_cache *cache, struct aki_packet *packet);
+bool aki_packet_cache_wait(struct aki_packet_cache *cache);
+struct aki_packet *aki_packet_cache_pop(struct aki_packet_cache *cache);
+void aki_packet_cache_disable(struct aki_packet_cache *cache);
+void aki_packet_cache_enable(struct aki_packet_cache *cache);
+void aki_packet_cache_free(struct aki_packet_cache *cache);
diff --git a/src/packet_pool.c b/src/packet_pool.c
new file mode 100644
index 0000000..4c942b1
--- /dev/null
+++ b/src/packet_pool.c
@@ -0,0 +1,169 @@
+#include "packet_pool.h"
+
+static void return_internal(struct aki_packet_pool *pool, struct aki_packet *packet)
+{
+    aki_packet_reset(packet);
+    al_array_push(pool->empty, packet);
+    if (aki_cond_is_waiting(&pool->cond)) {
+        aki_cond_signal(&pool->cond);
+    }
+}
+
+static void signal_callback(struct ev_loop *loop, ev_async *w, s32 revents)
+{
+    (void)loop;
+    struct aki_packet_pool *pool = (struct aki_packet_pool *)w->data;
+    (void)revents;
+    aki_mutex_lock(&pool->mutex);
+    while (pool->ready.size > 0) {
+        struct aki_packet *packet;
+        al_array_pop_at(pool->ready, 0, packet);
+        aki_mutex_unlock(&pool->mutex);
+        u8 ret = pool->callback(pool->userdata, packet);
+        al_assert(ret != AKI_PACKET_POOL_CLOSED);
+        aki_mutex_lock(&pool->mutex);
+        switch (ret) {
+        case AKI_PACKET_POOL_KEEP: // AKI_PACKET_POOL_NOP
+            break;
+        case AKI_PACKET_POOL_RETURN:
+            al_assert(pool->mode == AKI_PACKET_POOL_MODE_POOL);
+            return_internal(pool, packet);
+            break;
+        case AKI_PACKET_POOL_DISABLE:
+            al_assert(pool->mode == AKI_PACKET_POOL_MODE_POOL);
+            pool->disabled = true;
+            return_internal(pool, packet);
+            aki_mutex_unlock(&pool->mutex);
+            ret = pool->callback(pool->userdata, NULL);
+            al_assert(ret == AKI_PACKET_POOL_CLOSED);
+            // signal_callback must never be called again.
+            return;
+        }
+    }
+    aki_mutex_unlock(&pool->mutex);
+}
+
+static void init_internal(struct aki_packet_pool *pool, u32 size, struct aki_event_loop *loop,
+    u8 mode, u8 (*callback)(void *, struct aki_packet *), void *userdata)
+{
+    pool->mode = mode;
+    pool->loop = loop;
+    al_array_init(pool->ready);
+    al_array_reserve(pool->ready, size);
+    if (pool->mode == AKI_PACKET_POOL_MODE_POOL) {
+        al_array_init(pool->empty);
+        al_array_reserve(pool->empty, size);
+        for (u32 i = 0; i < size; i++) {
+            al_array_push(pool->empty, aki_packet_create());
+        }
+    }
+    pool->disabled = false;
+    aki_cond_init(&pool->cond);
+    aki_mutex_init(&pool->mutex);
+    pool->signal.data = pool;
+    ev_async_init(&pool->signal, signal_callback);
+    ev_async_start(pool->loop->ev, &pool->signal);
+    pool->callback = callback;
+    pool->userdata = userdata;
+}
+
+void aki_packet_pool_init(struct aki_packet_pool *pool, u32 size, struct aki_event_loop *loop,
+    u8 (*callback)(void *, struct aki_packet *), void *userdata)
+{
+    init_internal(pool, size, loop, AKI_PACKET_POOL_MODE_POOL, callback, userdata);
+}
+
+void aki_packet_pool_init_ex(struct aki_packet_pool *pool, u32 size, struct aki_event_loop *loop,
+    u8 mode, u8 (*callback)(void *, struct aki_packet *), void *userdata)
+{
+    init_internal(pool, size, loop, mode, callback, userdata);
+}
+
+struct aki_packet *aki_packet_pool_get(struct aki_packet_pool *pool)
+{
+    al_assert(pool->mode == AKI_PACKET_POOL_MODE_POOL);
+    aki_mutex_lock(&pool->mutex);
+    if (!pool->disabled && !pool->empty.size) {
+        aki_cond_wait(&pool->cond, &pool->mutex);
+    }
+    if (pool->disabled) {
+        aki_mutex_unlock(&pool->mutex);
+        return NULL;
+    }
+    struct aki_packet *packet = al_array_pop(pool->empty);
+    aki_mutex_unlock(&pool->mutex);
+    return packet;
+}
+
+void aki_packet_pool_submit(struct aki_packet_pool *pool, struct aki_packet *packet)
+{
+    aki_mutex_lock(&pool->mutex);
+    if (pool->disabled) {
+        aki_mutex_unlock(&pool->mutex);
+        return;
+    }
+    al_array_push(pool->ready, packet);
+    aki_mutex_unlock(&pool->mutex);
+    ev_async_send(pool->loop->ev, &pool->signal);
+}
+
+void aki_packet_pool_return(struct aki_packet_pool *pool, struct aki_packet *packet)
+{
+    al_assert(pool->mode == AKI_PACKET_POOL_MODE_POOL);
+    aki_mutex_lock(&pool->mutex);
+    return_internal(pool, packet);
+    aki_mutex_unlock(&pool->mutex);
+}
+
+void aki_packet_pool_disable(struct aki_packet_pool *pool)
+{
+    al_assert(pool->mode == AKI_PACKET_POOL_MODE_POOL);
+    aki_mutex_lock(&pool->mutex);
+    pool->disabled = true;
+    struct aki_packet *packet;
+    al_array_foreach(pool->ready, i, packet) {
+        aki_packet_reset(packet);
+        al_array_push(pool->empty, packet);
+    }
+    pool->ready.size = 0;
+    if (aki_cond_is_waiting(&pool->cond)) {
+        aki_cond_signal(&pool->cond);
+    }
+    aki_mutex_unlock(&pool->mutex);
+}
+
+void aki_packet_pool_enable(struct aki_packet_pool *pool)
+{
+    al_assert(pool->mode == AKI_PACKET_POOL_MODE_POOL);
+    aki_mutex_lock(&pool->mutex);
+    pool->disabled = false;
+    aki_mutex_unlock(&pool->mutex);
+}
+
+struct aki_packet *aki_packet_pool_pop(struct aki_packet_pool *pool)
+{
+    al_assert(pool->mode == AKI_PACKET_POOL_MODE_PASSTHROUGH);
+    struct aki_packet *packet = NULL;
+    if (pool->ready.size > 0) {
+        al_array_pop_at(pool->ready, 0, packet);
+    }
+    return packet;
+}
+
+void aki_packet_pool_free(struct aki_packet_pool *pool)
+{
+    ev_async_stop(pool->loop->ev, &pool->signal);
+    aki_mutex_destroy(&pool->mutex);
+    aki_cond_destroy(&pool->cond);
+    struct aki_packet *packet;
+    if (pool->mode == AKI_PACKET_POOL_MODE_POOL) {
+        al_array_foreach(pool->empty, i, packet) {
+            aki_packet_free(packet);
+        }
+        al_array_free(pool->empty);
+        al_array_foreach(pool->ready, i, packet) {
+            aki_packet_free(packet);
+        }
+    }
+    al_array_free(pool->ready);
+}
diff --git a/src/packet_pool.h b/src/packet_pool.h
new file mode 100644
index 0000000..2a2e608
--- /dev/null
+++ b/src/packet_pool.h
@@ -0,0 +1,63 @@
+#pragma once
+
+#include <al/array.h>
+
+#include "util/packet.h"
+#include "util/thread/thread.h"
+
+#include "loop.h"
+
+// Steps (MODE_POOL):
+//  packet_get (block)           \  Thread 2
+//  packet_submit (signal async) /
+//  callback called              \  Thread 1
+//  packet_return (unblock)      /
+//  pool_disable()
+//  join(thread1) && join(thread2)
+//  pool_free()
+
+// Steps (MODE_PASSTHROUGH):
+//  packet_submit (signal async) -> Thread 2
+//  callback called              -> Thread 1
+//  join(thread1) && join(thread2)
+//  while (pop()) {} (empty pool)
+//  pool_free()
+
+enum {
+    AKI_PACKET_POOL_MODE_POOL = 0,
+    AKI_PACKET_POOL_MODE_PASSTHROUGH
+};
+
+enum {
+    AKI_PACKET_POOL_KEEP = 0,
+    AKI_PACKET_POOL_RETURN,
+    AKI_PACKET_POOL_DISABLE,
+    AKI_PACKET_POOL_CLOSED
+};
+
+#define AKI_PACKET_POOL_NOP AKI_PACKET_POOL_KEEP
+
+struct aki_packet_pool {
+    u8 mode;
+    array(struct aki_packet *) empty;
+    array(struct aki_packet *) ready;
+    struct aki_event_loop *loop;
+    struct aki_cond cond;
+    struct aki_mutex mutex;
+    ev_async signal;
+    bool disabled;
+    u8 (*callback)(void *, struct aki_packet *);
+    void *userdata;
+};
+
+void aki_packet_pool_init(struct aki_packet_pool *pool, u32 size, struct aki_event_loop *loop,
+    u8 (*callback)(void *, struct aki_packet *), void *userdata);
+void aki_packet_pool_init_ex(struct aki_packet_pool *pool, u32 size, struct aki_event_loop *loop,
+    u8 mode, u8 (*callback)(void *, struct aki_packet *), void *userdata);
+struct aki_packet *aki_packet_pool_get(struct aki_packet_pool *pool);
+void aki_packet_pool_submit(struct aki_packet_pool *pool, struct aki_packet *packet);
+void aki_packet_pool_return(struct aki_packet_pool *pool, struct aki_packet *packet);
+void aki_packet_pool_disable(struct aki_packet_pool *pool);
+void aki_packet_pool_enable(struct aki_packet_pool *pool);
+struct aki_packet *aki_packet_pool_pop(struct aki_packet_pool *pool);
+void aki_packet_pool_free(struct aki_packet_pool *pool);
diff --git a/src/packet_stream.c b/src/packet_stream.c
new file mode 100644
index 0000000..8da342f
--- /dev/null
+++ b/src/packet_stream.c
@@ -0,0 +1,260 @@
+#include "packet_stream.h"
+
+static void init_internal(struct aki_packet_stream *stream)
+{
+    al_array_init(stream->out.queue);
+    stream->out.packet = NULL;
+    stream->out.have_data = false;
+    stream->in.packet = aki_packet_create();
+    stream->in.index = 0;
+    stream->in.have_size = false;
+}
+
+bool aki_packet_stream_init(struct aki_packet_stream *stream, u8 type,
+    void (*connection_callback)(void *, struct aki_packet_stream *),
+    void (*connection_closed_callback)(void *, struct aki_packet_stream *),
+    void (*packet_callback)(void *, struct aki_packet_stream *, struct aki_packet *),
+    void (*packet_sent_callback)(void *, struct aki_packet *), void *userdata)
+{
+    stream->connection_callback = connection_callback;
+    stream->connection_closed_callback = connection_closed_callback;
+    stream->packet_callback = packet_callback;
+    stream->packet_sent_callback = packet_sent_callback;
+    stream->flushed_callback = NULL;
+    stream->userdata = userdata;
+    stream->s.type = type;
+    stream->s.config.blocking = false;
+    stream->s.config.no_delay = false;
+    if (!aki_socket_init(&stream->s)) return false;
+    init_internal(stream);
+    stream->connected = false;
+    return true;
+}
+
+void aki_packet_stream_set_no_delay(struct aki_packet_stream *stream, bool no_delay)
+{
+    aki_socket_set_no_delay(&stream->s, no_delay);
+}
+
+static void stop_write_internal(struct aki_packet_stream *stream)
+{
+    stream->out.have_data = false;
+    ev_io_stop(stream->loop->ev, &stream->wevent);
+}
+
+static void stop_internal(struct aki_packet_stream *stream)
+{
+    stream->connected = false;
+    if (stream->timer_started) {
+        ev_timer_stop(stream->loop->ev, &stream->timer);
+        stream->timer_started = false;
+    }
+    if (stream->revent.active) {
+        ev_io_stop(stream->loop->ev, &stream->revent);
+    }
+    if (stream->out.have_data) {
+        stop_write_internal(stream);
+    }
+    aki_socket_close(&stream->s);
+    stream->connection_closed_callback(stream->userdata, stream);
+}
+
+static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+    (void)loop;
+    struct aki_packet_stream *stream = (struct aki_packet_stream *)w->data;
+    (void)revents;
+    u8 *ptr = aki_buffer_get_ptr(&stream->in.packet->buffer, stream->in.index);
+    u32 size = ((stream->in.have_size) ? aki_packet_size(stream->in.packet) : AKI_PACKET_HEADER_LENGTH);
+    ssize_t ret = aki_socket_read(&stream->s, ptr, size - stream->in.index);
+    if (ret <= 0) {
+        stop_internal(stream);
+        return;
+    }
+    stream->in.index += ret;
+    if (!stream->in.have_size && stream->in.index >= AKI_PACKET_HEADER_LENGTH) {
+        stream->in.have_size = true;
+        size = aki_packet_size(stream->in.packet);
+        aki_buffer_ensure_size(&stream->in.packet->buffer, size);
+    }
+    if (stream->in.have_size && stream->in.index >= size) {
+        struct aki_packet *packet = stream->in.packet;
+        stream->in.packet = aki_packet_create();
+        stream->in.index = 0;
+        stream->in.have_size = false;
+        stream->packet_callback(stream->userdata, stream, packet);
+        // Stream could be invalid after this point.
+    }
+}
+
+static void start_read_internal(struct aki_packet_stream *stream)
+{
+    ev_io_start(stream->loop->ev, &stream->revent);
+}
+
+static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+    (void)loop;
+    struct aki_packet_stream *stream = (struct aki_packet_stream *)w->data;
+    (void)revents;
+    if (UNLIKELY(!stream->connected)) {
+        stream->connected = true;
+        stop_write_internal(stream);
+        // This needs to be called before starting to read because setting
+        // connection_closed_callback and/or packet_callback here should be valid.
+        stream->connection_callback(stream->userdata, stream);
+        start_read_internal(stream);
+        return;
+    }
+    if (!stream->out.packet) {
+        if (stream->out.queue.size > 0) {
+            al_array_pop_at(stream->out.queue, 0, stream->out.packet);
+            stream->out.index = 0;
+        } else {
+            stop_write_internal(stream);
+            return;
+        }
+    }
+    u8 *ptr = aki_buffer_get_ptr(&stream->out.packet->buffer, stream->out.index);
+    u32 size = aki_packet_size(stream->out.packet);
+    ssize_t ret = aki_socket_write(&stream->s, ptr, size - stream->out.index);
+    if (ret <= 0) {
+        stop_write_internal(stream);
+        return;
+    }
+    stream->out.index += ret;
+    if (stream->out.index >= size) {
+        stream->packet_sent_callback(stream->userdata, stream->out.packet);
+        stream->out.packet = NULL;
+        if (stream->out.queue.size == 0 && stream->flushed_callback) {
+            stream->flushed_callback(stream->userdata);
+        }
+    }
+}
+
+static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+    (void)loop;
+    struct aki_packet_stream *server = (struct aki_packet_stream *)w->data;
+    (void)revents;
+    struct aki_packet_stream *client = al_alloc_object(struct aki_packet_stream);
+    client->s.config.blocking = false;
+    client->s.config.no_delay = false;
+    if (!aki_socket_accept(&server->s, &client->s)) {
+        al_free(client);
+        return;
+    }
+    client->connected = true;
+    client->loop = server->loop;
+    init_internal(client);
+    server->connection_callback(server->userdata, client);
+    client->wevent.data = client;
+    ev_io_init(&client->wevent, stream_write_callback,
+        aki_socket_get_fd(&client->s), EV_WRITE);
+    client->revent.data = client;
+    ev_io_init(&client->revent, stream_read_callback,
+        aki_socket_get_fd(&client->s), EV_READ);
+    ev_io_start(client->loop->ev, &client->revent);
+}
+
+void aki_packet_stream_listen(struct aki_packet_stream *stream,
+    struct aki_event_loop *loop, str *addr, s32 port)
+{
+    stream->loop = loop;
+    // wevent not used for listening.
+    stream->revent.data = stream;
+    ev_io_init(&stream->revent, socket_connection_callback,
+        aki_socket_get_fd(&stream->s), EV_READ);
+    ev_io_start(loop->ev, &stream->revent);
+    aki_socket_listen(&stream->s, addr, port);
+}
+
+static void timeout_callback(struct ev_loop *loop, ev_timer *w, s32 revents)
+{
+    (void)loop;
+    struct aki_packet_stream *stream = (struct aki_packet_stream *)w->data;
+    (void)revents;
+    if (!stream->connected) stop_internal(stream);
+}
+
+static void start_write_internal(struct aki_packet_stream *stream)
+{
+    stream->out.have_data = true;
+    ev_io_start(stream->loop->ev, &stream->wevent);
+}
+
+void aki_packet_stream_connect(struct aki_packet_stream *stream,
+    struct aki_event_loop *loop, str *addr, s32 port)
+{
+    stream->loop = loop;
+    stream->revent.data = stream;
+    ev_io_init(&stream->revent, stream_read_callback,
+        aki_socket_get_fd(&stream->s), EV_READ);
+    stream->wevent.data = stream;
+    ev_io_init(&stream->wevent, stream_write_callback,
+        aki_socket_get_fd(&stream->s), EV_WRITE);
+    start_write_internal(stream);
+    stream->timer.data = stream;
+    ev_timer_init(&stream->timer, timeout_callback, 5.0, 0.0);
+    if (!aki_socket_connect(&stream->s, addr, port)) {
+        stop_internal(stream);
+        return;
+    }
+    stream->timer_started = true;
+    ev_timer_start(loop->ev, &stream->timer);
+}
+
+void aki_packet_stream_pause(struct aki_packet_stream *stream)
+{
+    if (stream->connected && stream->revent.active) {
+        ev_io_stop(stream->loop->ev, &stream->revent);
+    }
+}
+
+void aki_packet_stream_resume(struct aki_packet_stream *stream)
+{
+    if (stream->connected && !stream->revent.active) {
+        ev_io_start(stream->loop->ev, &stream->revent);
+    }
+}
+
+void aki_packet_stream_send_packet(struct aki_packet_stream *stream, struct aki_packet *packet)
+{
+    al_assert(stream->connected);
+    aki_packet_write_size(packet);
+    al_array_push(stream->out.queue, packet);
+    if (!stream->out.have_data) {
+        start_write_internal(stream);
+    }
+}
+
+void aki_packet_stream_flush(struct aki_packet_stream *stream, void (*flushed_callback)(void *))
+{
+    stream->flushed_callback = flushed_callback;
+    if (!stream->out.have_data) {
+        stream->flushed_callback(stream->userdata);
+    }
+}
+
+void aki_packet_stream_disconnect(struct aki_packet_stream *stream)
+{
+    if (stream->connected) {
+        aki_socket_shutdown(&stream->s);
+        aki_packet_stream_resume(stream);
+    }
+}
+
+void aki_packet_stream_free(struct aki_packet_stream *stream)
+{
+    if (stream->in.packet) {
+        aki_packet_free(stream->in.packet);
+    }
+    if (stream->out.packet) {
+        stream->packet_sent_callback(stream->userdata, stream->out.packet);
+    }
+    struct aki_packet *packet;
+    al_array_foreach(stream->out.queue, i, packet) {
+        stream->packet_sent_callback(stream->userdata, packet);
+    }
+    al_array_free(stream->out.queue);
+}
diff --git a/src/packet_stream.h b/src/packet_stream.h
new file mode 100644
index 0000000..027fac0
--- /dev/null
+++ b/src/packet_stream.h
@@ -0,0 +1,52 @@
+#pragma once
+
+#include <al/array.h>
+
+#include "util/packet.h"
+#include "socket/socket.h"
+
+#include "loop.h"
+
+struct aki_packet_stream {
+    struct aki_socket s;
+    struct aki_event_loop *loop;
+    ev_io revent;
+    ev_io wevent;
+    ev_timer timer;
+    bool timer_started;
+    bool connected;
+    void (*connection_callback)(void *, struct aki_packet_stream *);
+    void (*connection_closed_callback)(void *, struct aki_packet_stream *);
+    void (*packet_callback)(void *, struct aki_packet_stream *, struct aki_packet *);
+    void (*packet_sent_callback)(void *, struct aki_packet *);
+    void (*flushed_callback)(void *);
+    void *userdata;
+    struct {
+        array(struct aki_packet *) queue;
+        struct aki_packet *packet;
+        bool have_data;
+        size_t index;
+    } out;
+    struct {
+        struct aki_packet *packet;
+        bool have_size;
+        size_t index;
+    } in;
+};
+
+bool aki_packet_stream_init(struct aki_packet_stream *stream, u8 type,
+    void (*connection_callback)(void *, struct aki_packet_stream *),
+    void (*connection_closed_callback)(void *, struct aki_packet_stream *),
+    void (*packet_callback)(void *, struct aki_packet_stream *, struct aki_packet *),
+    void (*packet_sent_callback)(void *, struct aki_packet *), void *userdata);
+void aki_packet_stream_set_no_delay(struct aki_packet_stream *stream, bool no_delay);
+void aki_packet_stream_listen(struct aki_packet_stream *stream,
+    struct aki_event_loop *loop, str *addr, s32 port);
+void aki_packet_stream_connect(struct aki_packet_stream *stream,
+    struct aki_event_loop *loop, str *addr, s32 port);
+void aki_packet_stream_pause(struct aki_packet_stream *stream);
+void aki_packet_stream_resume(struct aki_packet_stream *stream);
+void aki_packet_stream_send_packet(struct aki_packet_stream *stream, struct aki_packet *packet);
+void aki_packet_stream_flush(struct aki_packet_stream *stream, void (*flushed_callback)(void *));
+void aki_packet_stream_disconnect(struct aki_packet_stream *stream);
+void aki_packet_stream_free(struct aki_packet_stream *stream);
diff --git a/src/rpc.c b/src/rpc.c
new file mode 100644
index 0000000..edf73c0
--- /dev/null
+++ b/src/rpc.c
@@ -0,0 +1,146 @@
+#include "rpc2.h"
+
+static void connection_closed_callback(void *userdata, struct aki_packet_stream *s)
+{
+    struct aki_rpc_connection *conn = (struct aki_rpc_connection *)userdata;
+    al_assert(s == conn->stream);
+    conn->rpc->connection_closed_callback(conn->rpc->userdata, conn);
+}
+
+static void packet_callback(void *userdata, struct aki_packet_stream *s, struct aki_packet *packet)
+{
+    struct aki_rpc_connection *conn = (struct aki_rpc_connection *)userdata;
+    s8 op = aki_packet_read_s8(packet);
+    u32 id = aki_packet_read_u32(packet);
+    if (op == -1) { // Response
+        struct aki_rpc_callback *callback;
+        al_array_foreach_ptr(conn->callbacks, i, callback) {
+            if (callback->id == id) {
+                callback->callback(callback->userdata, packet);
+                al_array_remove_at_iter(conn->callbacks, i);
+                break;
+            }
+        }
+    } else { // Command
+        struct aki_rpc_command *command;
+        al_array_foreach_ptr(conn->rpc->commands, i, command) {
+            if (command->op == op) {
+                struct aki_packet *rpacket = aki_packet_create();
+                aki_packet_write_s8(rpacket, -1);
+                aki_packet_write_u32(rpacket, id);
+                if (command->callback(command->userdata, conn, packet, rpacket)) {
+                    aki_packet_stream_send_packet(s, rpacket);
+                } else {
+                    aki_packet_free(rpacket);
+                }
+            }
+        }
+    }
+}
+
+static void packet_sent_callback(void *userdata, struct aki_packet *packet)
+{
+    (void)userdata;
+    aki_packet_free(packet);
+}
+
+static void connection_callback(void *userdata, struct aki_packet_stream *s)
+{
+    struct aki_rpc *rpc = (struct aki_rpc *)userdata;
+    aki_packet_stream_set_no_delay(s, true);
+    struct aki_rpc_connection *conn = al_alloc_object(struct aki_rpc_connection);
+    al_array_push(rpc->connections, conn);
+    conn->rpc = rpc;
+    conn->stream = s;
+    // connection_callback will never be called twice for the same stream.
+    s->connection_closed_callback = connection_closed_callback;
+    s->packet_callback = packet_callback;
+    s->packet_sent_callback = packet_sent_callback;
+    s->flushed_callback = NULL;
+    // For aki_rpc_connect this overwrites rpc as userdata, whereas for aki_rpc_listen
+    // this is a new aki_packet_stream object.
+    s->userdata = conn;
+    al_array_init(conn->callbacks);
+    rpc->connection_callback(rpc->userdata, conn);
+}
+
+bool aki_rpc_init(struct aki_rpc *rpc, u8 type,
+    void (*connection_callback)(void *, struct aki_rpc_connection *),
+    void (*connection_closed_callback)(void *, struct aki_rpc_connection *), void *userdata)
+{
+    rpc->type = type;
+    rpc->id = 0;
+    rpc->connection_callback = connection_callback;
+    rpc->connection_closed_callback = connection_closed_callback;
+    rpc->userdata = userdata;
+    al_array_init(rpc->connections);
+    al_array_init(rpc->commands);
+    return true;
+}
+
+void aki_rpc_add_command(struct aki_rpc *rpc, struct aki_rpc_command *command)
+{
+    al_array_push(rpc->commands, *command);
+}
+
+bool aki_rpc_listen(struct aki_rpc *rpc, struct aki_event_loop *loop, str *addr, s32 port)
+{
+    if (!aki_packet_stream_init(&rpc->stream, rpc->type, connection_callback, NULL, NULL, NULL, rpc)) {
+        return false;
+    }
+    aki_packet_stream_set_no_delay(&rpc->stream, true);
+    aki_packet_stream_listen(&rpc->stream, loop, addr, port);
+    return true;
+}
+
+bool aki_rpc_connect(struct aki_rpc *rpc, struct aki_event_loop *loop, str *addr, s32 port)
+{
+    if (!aki_packet_stream_init(&rpc->stream, rpc->type, connection_callback, NULL, NULL, NULL, rpc)) {
+        return false;
+    }
+    aki_packet_stream_set_no_delay(&rpc->stream, true);
+    aki_packet_stream_connect(&rpc->stream, loop, addr, port);
+    return true;
+}
+
+struct aki_packet *aki_rpc_get_packet(struct aki_rpc *rpc, s8 op)
+{
+    struct aki_packet *packet = aki_packet_create();
+    aki_packet_write_s8(packet, op);
+    aki_packet_write_u32(packet, rpc->id);
+    rpc->id = al_u32_inc_wrap(rpc->id);
+    return packet;
+}
+
+void aki_rpc_disconnect(struct aki_rpc *rpc)
+{
+    aki_packet_stream_disconnect(&rpc->stream);
+}
+
+void aki_rpc_free(struct aki_rpc *rpc)
+{
+    aki_packet_stream_free(&rpc->stream);
+    struct aki_rpc_connection *conn;
+    al_array_foreach(rpc->connections, i, conn) {
+        al_array_free(conn->callbacks);
+        al_free(conn);
+    }
+    al_array_free(rpc->connections);
+    al_array_free(rpc->commands);
+}
+
+#define get_packet_id(p) \
+    *((u32 *)aki_buffer_get_ptr(&p->buffer, AKI_PACKET_HEADER_LENGTH + sizeof(s8)))
+
+void aki_rpc_connection_command(struct aki_rpc_connection *conn, struct aki_packet *packet,
+    void (*callback)(void *, struct aki_packet *), void *userdata)
+{
+    if (callback) {
+        al_array_push(conn->callbacks, ((struct aki_rpc_callback){
+            .id = get_packet_id(packet),
+            .callback = callback,
+            .userdata = userdata
+        }));
+    }
+    aki_packet_stream_send_packet(conn->stream, packet);
+}
diff --git a/src/rpc2.h b/src/rpc2.h
new file mode 100644
index 0000000..482ad56
--- /dev/null
+++ b/src/rpc2.h
@@ -0,0 +1,49 @@
+#pragma once
+
+#include <al/array.h>
+
+#include "util/packet.h"
+
+#include "packet_stream.h"
+
+struct aki_rpc_callback {
+    u32 id;
+    void (*callback)(void *, struct aki_packet *);
+    void *userdata;
+};
+
+struct aki_rpc_connection {
+    struct aki_rpc *rpc;
+    struct aki_packet_stream *stream;
+    array(struct aki_rpc_callback) callbacks;
+};
+
+struct aki_rpc_command {
+    s8 op;
+    bool (*callback)(void *, struct aki_rpc_connection *, struct aki_packet *, struct aki_packet *);
+    void *userdata;
+};
+
+struct aki_rpc {
+    u8 type;
+    struct aki_packet_stream stream;
+    array(struct aki_rpc_command) commands;
+    array(struct aki_rpc_connection *) connections;
+    u32 id;
+    void (*connection_callback)(void *, struct aki_rpc_connection *);
+    void (*connection_closed_callback)(void *, struct aki_rpc_connection *);
+    void *userdata;
+};
+
+bool aki_rpc_init(struct aki_rpc *rpc, u8 type,
+    void (*connection_callback)(void *, struct aki_rpc_connection *),
+    void (*connection_closed_callback)(void *, struct aki_rpc_connection *), void *userdata);
+void aki_rpc_add_command(struct aki_rpc *rpc, struct aki_rpc_command *command);
+bool aki_rpc_listen(struct aki_rpc *rpc, struct aki_event_loop *loop, str *addr, s32 port);
+bool aki_rpc_connect(struct aki_rpc *rpc, struct aki_event_loop *loop, str *addr, s32 port);
+struct aki_packet *aki_rpc_get_packet(struct aki_rpc *rpc, s8 op);
+void aki_rpc_disconnect(struct aki_rpc *rpc);
+void aki_rpc_free(struct aki_rpc *rpc);
+
+void aki_rpc_connection_command(struct aki_rpc_connection *conn, struct aki_packet *packet,
+    void (*callback)(void *, struct aki_packet *), void *userdata);
diff --git a/src/signal.c b/src/signal.c
new file mode 100644
index 0000000..0bc15b8
--- /dev/null
+++ b/src/signal.c
@@ -0,0 +1,33 @@
+#include "signal.h"
+
+static void signal_callback(struct ev_loop *loop, ev_async *w, s32 revents)
+{
+    (void)loop;
+    struct aki_signal *signal = (struct aki_signal *)w->data;
+    (void)revents;
+    signal->callback(signal->userdata);
+}
+
+void aki_signal_init(struct aki_signal *signal, void (*callback)(void *), void *userdata)
+{
+    signal->callback = callback;
+    signal->userdata = userdata;
+    signal->signal.data = signal;
+    ev_async_init(&signal->signal, signal_callback);
+}
+
+void aki_signal_start(struct aki_signal *signal, struct aki_event_loop *loop)
+{
+    signal->loop = loop;
+    ev_async_start(loop->ev, &signal->signal);
+}
+
+void aki_signal_send(struct aki_signal *signal)
+{
+    ev_async_send(signal->loop->ev, &signal->signal);
+}
+
+void aki_signal_stop(struct aki_signal *signal)
+{
+    ev_async_stop(signal->loop->ev, &signal->signal);
+}
diff --git a/src/signal.h b/src/signal.h
new file mode 100644
index 0000000..46bbbdb
--- /dev/null
+++ b/src/signal.h
@@ -0,0 +1,15 @@
+#pragma once
+
+#include "loop.h"
+
+struct aki_signal {
+    ev_async signal;
+    struct aki_event_loop *loop;
+    void (*callback)(void *);
+    void *userdata;
+};
+
+void aki_signal_init(struct aki_signal *signal, void (*callback)(void *), void *userdata);
+void aki_signal_start(struct aki_signal *signal, struct aki_event_loop *loop);
+void aki_signal_send(struct aki_signal *signal);
+void aki_signal_stop(struct aki_signal *signal);
diff --git a/src/socket/socket.h b/src/socket/socket.h
new file mode 100644
index 0000000..ea307e7
--- /dev/null
+++ b/src/socket/socket.h
@@ -0,0 +1,66 @@
+#pragma once
+
+#ifndef _WIN32
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/un.h>
+#else
+#include "../winwrap.h"
+#endif
+
+#include <al/types.h>
+#include <al/str.h>
+
+enum {
+    AKI_SOCKET_TCP = 0,
+    AKI_SOCKET_UDP,
+    AKI_SOCKET_UNIX
+};
+
+struct aki_socket {
+    u8 type;
+    struct {
+        bool blocking;
+        bool no_delay;
+    } config;
+#ifndef _WIN32
+    s32 sock;
+    struct sockaddr_in addr_in;
+    struct sockaddr_un addr_un;
+#else
+    SOCKET sock;
+    SOCKADDR_IN addr_in;
+#endif
+};
+
+#ifndef _WIN32
+#define aki_read read
+#else
+#define aki_read _read
+#endif
+
+#ifndef _WIN32
+#define aki_write write
+#else
+#define aki_write _write
+#endif
+
+bool aki_socket_init(struct aki_socket *s);
+
+void aki_socket_set_blocking(struct aki_socket *s, bool blocking);
+void aki_socket_set_no_delay(struct aki_socket *s, bool no_delay);
+
+bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port);
+bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c);
+bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port);
+
+s32 aki_socket_get_fd(struct aki_socket *s);
+
+ssize_t aki_socket_read(struct aki_socket *s, void *buf, size_t size);
+ssize_t aki_socket_write(struct aki_socket *s, void *buf, size_t size);
+ssize_t aki_socket_sendto(struct aki_socket *s, void *buf, size_t size);
+
+void aki_socket_shutdown(struct aki_socket *s);
+void aki_socket_close(struct aki_socket *s);
diff --git a/src/socket/socket_linux.c b/src/socket/socket_linux.c
new file mode 100644
index 0000000..b466dea
--- /dev/null
+++ b/src/socket/socket_linux.c
@@ -0,0 +1,259 @@
+#include <unistd.h>
+#include <sys/fcntl.h>
+#include <netinet/tcp.h>
+#include <netdb.h>
+#include <signal.h>
+#include <al/log.h>
+
+#include "../util/error.h"
+
+#include "socket.h"
+
+static void socket_set_blocking(struct aki_socket *s, bool blocking)
+{
+    s32 flags = fcntl(s->sock, F_GETFL, 0);
+    al_assert(flags != -1);
+    flags = blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK);
+    s32 ret = fcntl(s->sock, F_SETFL, flags);
+    al_assert(ret == 0);
+}
+
+static void socket_set_no_delay(struct aki_socket *s, s32 no_delay)
+{
+    if (s->type == AKI_SOCKET_TCP) {
+        setsockopt(s->sock, IPPROTO_TCP, TCP_NODELAY, &no_delay, sizeof(no_delay));
+    }
+}
+
+static void socket_set_send_buf(struct aki_socket *s, u32 sndbuf)
+{
+    u32 send_queue_size = sndbuf;
+    setsockopt(s->sock, SOL_SOCKET, SO_SNDBUF, &send_queue_size, sizeof(send_queue_size));
+
+    socklen_t optlen = sizeof(send_queue_size);
+    getsockopt(s->sock, SOL_SOCKET, SO_SNDBUF, &send_queue_size, &optlen);
+
+    if (send_queue_size != sndbuf * 2) {
+        al_log_warn("socket", "Failed to set SO_SNDBUF.");
+    }
+}
+
+static void socket_set_recv_buf(struct aki_socket *s, u32 rcvbuf)
+{
+    u32 receive_queue_size = rcvbuf;
+    setsockopt(s->sock, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, sizeof(receive_queue_size));
+
+    socklen_t optlen = sizeof(receive_queue_size);
+    getsockopt(s->sock, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, &optlen);
+
+    if (receive_queue_size != rcvbuf * 2) {
+        al_log_warn("socket", "Failed to set SO_RCVBUF.");
+    }
+}
+
+bool aki_socket_init(struct aki_socket *s)
+{
+    switch (s->type) {
+    case AKI_SOCKET_TCP:
+        s->sock = socket(AF_INET, SOCK_STREAM, 0);
+        break;
+    case AKI_SOCKET_UDP:
+        s->sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+        break;
+    case AKI_SOCKET_UNIX:
+        s->sock = socket(AF_UNIX, SOCK_STREAM, 0);
+        break;
+    }
+
+    if (s->sock < 0) {
+        al_log_error("socket", "Failed to create socket.");
+        return false;
+    }
+
+    aki_socket_set_blocking(s, s->config.blocking);
+    aki_socket_set_no_delay(s, s->config.no_delay);
+
+    return true;
+}
+
+void aki_socket_set_blocking(struct aki_socket *s, bool blocking)
+{
+    s->config.blocking = blocking;
+    socket_set_blocking(s, blocking);
+}
+
+void aki_socket_set_no_delay(struct aki_socket *s, bool no_delay)
+{
+    s->config.no_delay = no_delay;
+    socket_set_no_delay(s, no_delay);
+}
+
+bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c)
+{
+    c->type = s->type;
+
+    struct sockaddr_storage addr;
+    socklen_t len = sizeof(addr);
+    if((c->sock = accept(s->sock, (struct sockaddr *)&addr, &len)) == -1) {
+        return false;
+    }
+
+    aki_socket_set_blocking(c, c->config.blocking);
+    aki_socket_set_no_delay(c, c->config.no_delay);
+
+    switch (s->type) {
+    case AKI_SOCKET_TCP: {
+        s32 port = 0;
+        char ipstr[INET6_ADDRSTRLEN] = {0};
+        if (addr.ss_family == AF_INET) {
+            struct sockaddr_in *s = (struct sockaddr_in *)&addr;
+            port = ntohs(s->sin_port);
+            inet_ntop(AF_INET, &s->sin_addr, ipstr, sizeof(ipstr));
+        } else if (addr.ss_family == AF_INET6) {
+            struct sockaddr_in6 *s = (struct sockaddr_in6 *)&addr;
+            port = ntohs(s->sin6_port);
+            inet_ntop(AF_INET6, &s->sin6_addr, ipstr, sizeof(ipstr));
+        }
+        al_log_info("socket", "Connection from %s:%d.", ipstr, port);
+        break;
+    }
+    case AKI_SOCKET_UNIX: {
+        al_log_info("socket", "Connection from a UNIX socket.");
+        break;
+    }
+    case AKI_SOCKET_UDP: {
+        al_assert(false);
+    }
+    }
+
+    return true;
+}
+
+bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port)
+{
+    struct sockaddr *saddr = NULL;
+    socklen_t addrlen = 0;
+
+    switch (s->type) {
+    case AKI_SOCKET_TCP: {
+        s->addr_in.sin_family = AF_INET;
+        s->addr_in.sin_addr.s_addr = htonl(INADDR_ANY);
+        s->addr_in.sin_port = htons(port);
+        saddr = (struct sockaddr *)&s->addr_in;
+        addrlen = sizeof(s->addr_in);
+        break;
+    }
+    case AKI_SOCKET_UNIX: {
+        char *c_str = al_str_to_c_str(addr);
+        s->addr_un.sun_family = AF_UNIX;
+        size_t len = al_strlen(c_str);
+        al_memcpy(s->addr_un.sun_path, c_str, len);
+        s->addr_un.sun_path[len] = '\0';
+        al_free(c_str);
+        saddr = (struct sockaddr *)&s->addr_un;
+        addrlen = sizeof(s->addr_un);
+        break;
+    }
+    case AKI_SOCKET_UDP: {
+        al_assert(false);
+    }
+    }
+
+    if (bind(s->sock, saddr, addrlen) < 0) {
+        al_log_error("socket", "Failed to bind (%.*s:%d).", AL_STR_PRINTF(addr), port);
+        return false;
+    }
+
+    if (listen(s->sock, SOMAXCONN) < 0) {
+        al_log_error("socket", "Failed to listen (%.*s:%d).", AL_STR_PRINTF(addr), port);
+        return false;
+    }
+
+    al_log_info("socket", "Listening on %.*s:%d.", AL_STR_PRINTF(addr), port);
+
+    signal(SIGPIPE, SIG_IGN);
+
+    return true;
+}
+
+bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
+{
+    switch (s->type) {
+    case AKI_SOCKET_TCP: {
+        char *c_str = al_str_to_c_str(addr);
+        struct hostent *hptr = gethostbyname(c_str);
+        al_free(c_str);
+
+        if (!hptr) {
+            al_log_error("socket", "Failed to parse address.");
+            return false;
+        }
+
+        if (hptr->h_addrtype != AF_INET) {
+            al_log_error("socket", "Bad address family.");
+            return false;
+        }
+
+        s->addr_in.sin_family = AF_INET;
+        s->addr_in.sin_addr.s_addr = ((struct in_addr *)hptr->h_addr_list[0])->s_addr;
+        s->addr_in.sin_port = htons(port);
+
+        s32 ret = connect(s->sock, (struct sockaddr *)&s->addr_in, sizeof(s->addr_in));
+
+        if (ret != 0 && errno != 115) {
+            al_log_error("socket", "Failed to connect: %s (%d).", aki_strerror(errno), errno);
+            return false;
+        }
+
+        break;
+    }
+    case AKI_SOCKET_UNIX: {
+        s->addr_un.sun_family = AF_UNIX;
+        al_memcpy(s->addr_un.sun_path, addr->data, addr->len);
+        s->addr_un.sun_path[addr->len] = '\0';
+        if (connect(s->sock, (struct sockaddr *)&s->addr_un, sizeof(s->addr_un)) < 0) {
+            al_log_error("socket", "Failed to connect: %s (%d).", aki_strerror(errno), errno);
+            return false;
+        }
+        break;
+    }
+    case AKI_SOCKET_UDP: {
+        al_assert(false);
+    }
+    }
+
+    signal(SIGPIPE, SIG_IGN);
+
+    return true;
+}
+
+ssize_t aki_socket_read(struct aki_socket *s, void *buf, size_t size)
+{
+    return recv(s->sock, buf, size, 0);
+}
+
+ssize_t aki_socket_write(struct aki_socket *s, void *buf, size_t size)
+{
+    return send(s->sock, buf, size, MSG_NOSIGNAL);
+}
+
+ssize_t aki_socket_sendto(struct aki_socket *s, void *buf, size_t size)
+{
+    return sendto(s->sock, buf, size, 0, (const struct sockaddr *)&s->addr_un, sizeof(s->addr_un));
+}
+
+s32 aki_socket_get_fd(struct aki_socket *s)
+{
+    return s->sock;
+}
+
+void aki_socket_shutdown(struct aki_socket *s)
+{
+    shutdown(s->sock, SHUT_RDWR);
+}
+
+void aki_socket_close(struct aki_socket *s)
+{
+    close(s->sock);
+}
+
diff --git a/src/socket/socket_windows.c b/src/socket/socket_windows.c
new file mode 100644
index 0000000..6e72aea
--- /dev/null
+++ b/src/socket/socket_windows.c
@@ -0,0 +1,130 @@
+#include <al/log.h>
+
+#include "socket.h"
+
+static void socket_set_blocking(struct aki_socket *s, bool blocking)
+{
+    u_long mode = (blocking) ? 0 : 1;
+    ioctlsocket(s->sock, FIONBIO, &mode);
+}
+
+static void socket_set_no_delay(struct aki_socket *s, s32 no_delay)
+{
+    (void)s;
+    (void)no_delay;
+}
+
+bool aki_socket_init(struct aki_socket *s)
+{
+    switch (s->type) {
+    case AKI_SOCKET_TCP:
+        s->sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+        break;
+    case AKI_SOCKET_UDP:
+        al_assert(false);
+        return false;
+    case AKI_SOCKET_UNIX:
+        al_assert(false);
+        return false;
+    }
+
+    if (s->sock == INVALID_SOCKET) {
+        al_log_error("socket", "Failed to create socket.");
+        return false;
+    }
+
+    aki_socket_set_blocking(s, s->config.blocking);
+    aki_socket_set_no_delay(s, s->config.no_delay);
+
+    return true;
+}
+
+void aki_socket_set_blocking(struct aki_socket *s, bool blocking)
+{
+    s->config.blocking = blocking;
+    socket_set_blocking(s, blocking);
+}
+
+void aki_socket_set_no_delay(struct aki_socket *s, bool no_delay)
+{
+    s->config.no_delay = no_delay;
+    socket_set_no_delay(s, no_delay);
+}
+
+bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port)
+{
+    s->addr_in.sin_family = AF_INET;
+    s->addr_in.sin_addr.s_addr = INADDR_ANY;
+    s->addr_in.sin_port = htons(port);
+    al_memset(&(s->addr_in.sin_zero), '\0', 8);
+    if (bind(s->sock, (SOCKADDR *)&s->addr_in, sizeof(s->addr_in)) == SOCKET_ERROR) {
+        al_log_error("socket", "Failed to bind (%.*s:%d).", AL_STR_PRINTF(addr), port);
+        return false;
+    }
+    if (listen(s->sock, SOMAXCONN) == SOCKET_ERROR) {
+        al_log_error("socket", "Failed to listen (%.*s:%d).", AL_STR_PRINTF(addr), port);
+        return false;
+    }
+    return true;
+}
+
+bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c)
+{
+    c->type = s->type;
+
+    if ((c->sock = accept(s->sock, NULL, NULL)) == INVALID_SOCKET) {
+        return false;
+    }
+
+    aki_socket_set_blocking(c, c->config.blocking);
+    aki_socket_set_no_delay(c, c->config.no_delay);
+
+    return true;
+}
+
+bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
+{
+    s->addr_in.sin_family = AF_INET;
+    char *c_str = al_str_to_c_str(addr);
+    InetPton(AF_INET, c_str, &s->addr_in.sin_addr.s_addr);
+    al_free(c_str);
+    s->addr_in.sin_port = htons(port);
+    al_memset(&(s->addr_in.sin_zero), '\0', 8);
+    if (connect(s->sock, (SOCKADDR *)&s->addr_in, sizeof(s->addr_in)) == SOCKET_ERROR) {
+        return false;
+    }
+    return true;
+}
+
+s32 aki_socket_get_fd(struct aki_socket *s)
+{
+    return _open_osfhandle(s->sock, 0);
+}
+
+ssize_t aki_socket_read(struct aki_socket *s, void *buf, size_t size)
+{
+    return recv(s->sock, buf, (s32)size, 0);
+}
+
+ssize_t aki_socket_write(struct aki_socket *s, void *buf, size_t size)
+{
+    return send(s->sock, buf, (s32)size, 0);
+}
+
+ssize_t aki_socket_sendto(struct aki_socket *s, void *buf, size_t size)
+{
+    (void)s;
+    (void)buf;
+    (void)size;
+    return 0;
+}
+
+void aki_socket_shutdown(struct aki_socket *s)
+{
+    if (shutdown(s->sock, SD_SEND) == SOCKET_ERROR) {}
+}
+
+void aki_socket_close(struct aki_socket *s)
+{
+    closesocket(s->sock);
+}
diff --git a/src/timer.c b/src/timer.c
new file mode 100644
index 0000000..e730ad5
--- /dev/null
+++ b/src/timer.c
@@ -0,0 +1,25 @@
+#include "timer.h"
+
+static void timer_callback(struct ev_loop *loop, ev_timer *w, s32 revents)
+{
+    (void)loop;
+    struct aki_timer *timer = (struct aki_timer *)w->data;
+    (void)revents;
+    timer->callback(timer->userdata, timer);
+}
+
+void aki_timer_init(struct aki_timer *timer, struct aki_event_loop *loop,
+    aki_os_tstamp repeat, void (*callback)(void *, struct aki_timer *), void *userdata)
+{
+    timer->loop = loop;
+    timer->callback = callback;
+    timer->userdata = userdata;
+    timer->timer.data = timer;
+    timer->timer.repeat = repeat;
+    ev_init(&timer->timer, timer_callback);
+}
+
+void aki_timer_again(struct aki_timer *timer)
+{
+    ev_timer_again(timer->loop->ev, &timer->timer);
+}
diff --git a/src/timer.h b/src/timer.h
new file mode 100644
index 0000000..03b2f43
--- /dev/null
+++ b/src/timer.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include "util/thread/thread.h"
+
+#include "evwrap.h"
+#include "loop.h"
+
+struct aki_timer {
+    ev_timer timer;
+    struct aki_event_loop *loop;
+    void (*callback)(void *, struct aki_timer *);
+    void *userdata;
+};
+
+void aki_timer_init(struct aki_timer *timer, struct aki_event_loop *loop,
+    aki_os_tstamp repeat, void (*callback)(void *, struct aki_timer *), void *userdata);
+void aki_timer_again(struct aki_timer *timer);
diff --git a/src/util/buffer.c b/src/util/buffer.c
new file mode 100644
index 0000000..26e242c
--- /dev/null
+++ b/src/util/buffer.c
@@ -0,0 +1,42 @@
+#include "buffer.h"
+
+#define INITIAL_SIZE 128
+
+void aki_buffer_ensure_size(struct aki_buffer *buf, size_t size)
+{
+    if (buf->alloc >= size) return;
+    if (size < INITIAL_SIZE) buf->alloc = INITIAL_SIZE;
+    else buf->alloc = al_next_power_of_two(size);
+    buf->data = buf->data ? al_realloc(buf->data, buf->alloc) : al_malloc(buf->alloc);
+}
+
+void aki_buffer_init(struct aki_buffer *buf)
+{
+    buf->data = NULL;
+    buf->size = 0;
+    buf->alloc = 0;
+}
+
+void aki_buffer_write(struct aki_buffer *buf, void *data, size_t index, size_t size)
+{
+    size_t reach = index + size;
+    aki_buffer_ensure_size(buf, reach);
+    if (reach > buf->size) buf->size = reach;
+    al_memcpy(&buf->data[index], data, size);
+}
+
+u8 *aki_buffer_get_ptr(struct aki_buffer *buf, size_t index)
+{
+    return &buf->data[index];
+}
+
+void aki_buffer_read(struct aki_buffer *buf, void *ptr, size_t index, size_t size)
+{
+    al_assert(index + size <= buf->size);
+    al_memcpy(ptr, &buf->data[index], size);
+}
+
+void aki_buffer_free(struct aki_buffer *buf)
+{
+    if (buf->data && buf->alloc > 0) al_free(buf->data);
+}
diff --git a/src/util/buffer.h b/src/util/buffer.h
new file mode 100644
index 0000000..3c40b3a
--- /dev/null
+++ b/src/util/buffer.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include <al/types.h>
+#include <al/lib.h>
+
+struct aki_buffer {
+    u8 *data;
+    size_t size;
+    size_t alloc;
+};
+
+void aki_buffer_init(struct aki_buffer *buf);
+void aki_buffer_ensure_size(struct aki_buffer *buf, size_t size);
+void aki_buffer_write(struct aki_buffer *buf, void *data, size_t index, size_t size);
+u8 *aki_buffer_get_ptr(struct aki_buffer *buf, size_t index);
+void aki_buffer_read(struct aki_buffer *buf, void *ptr, size_t index, size_t size);
+void aki_buffer_free(struct aki_buffer *buf);
diff --git a/src/util/error.c b/src/util/error.c
new file mode 100644
index 0000000..f055416
--- /dev/null
+++ b/src/util/error.c
@@ -0,0 +1,7 @@
+#include "error.h"
+
+#ifndef _WIN32
+__thread char errorbuf[ERROR_STR_MAXLEN];
+#else
+__declspec(thread) char errorbuf[ERROR_STR_MAXLEN];
+#endif
diff --git a/src/util/error.h b/src/util/error.h
new file mode 100644
index 0000000..d014faa
--- /dev/null
+++ b/src/util/error.h
@@ -0,0 +1,21 @@
+#include <al/lib.h>
+
+#include <errno.h>
+
+#define ERROR_STR_MAXLEN 94
+
+#ifndef _WIN32
+extern __thread char errorbuf[ERROR_STR_MAXLEN];
+#else
+extern __declspec(thread) char errorbuf[ERROR_STR_MAXLEN];
+#endif
+
+static inline char *aki_strerror(s32 errnum)
+{
+#ifndef _WIN32
+    strerror_r(errnum, errorbuf, ERROR_STR_MAXLEN);
+#else
+    strerror_s(errorbuf, ERROR_STR_MAXLEN, errnum);
+#endif
+    return errorbuf;
+}
diff --git a/src/util/file/file.c b/src/util/file/file.c
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/src/util/file/file.c
diff --git a/src/util/file/file.h b/src/util/file/file.h
new file mode 100644
index 0000000..465d76a
--- /dev/null
+++ b/src/util/file/file.h
@@ -0,0 +1,82 @@
+#pragma once
+
+#ifdef AKIYO_FILE_STDIO_COMPAT
+#include <stdio.h>
+#else
+#ifndef _WIN32
+#include <dirent.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/stat.h>
+#include <sys/stat.h>
+#include <sys/mman.h>
+#endif
+#endif
+
+#include <al/str.h>
+
+enum {
+    AKI_ENTRY_FILE = 0,
+    AKI_ENTRY_DIR,
+    AKI_ENTRY_OTHER
+};
+
+struct aki_file {
+#ifdef AKIYO_FILE_STDIO_COMPAT
+    FILE *file;
+#else
+#ifndef _WIN32
+    s32 fd;
+#else
+    // WINDOWS
+#endif
+#endif
+    size_t filesize;
+};
+
+struct aki_dir {
+#ifdef AKIYO_FILE_STDIO_COMPAT
+#else
+#ifndef _WIN32
+    DIR *dir;
+#else
+    // WINDOWS
+#endif
+#endif
+    str path;
+};
+
+struct aki_dir_entry {
+    u8 type;
+#ifdef AKIYO_FILE_STDIO_COMPAT
+#else
+#ifndef _WIN32
+    struct dirent *entry;
+#else
+    // WINDOWS
+#endif
+#endif
+    str name;
+    str path;
+};
+
+bool aki_file_open(struct aki_file *file, str *path, bool create);
+bool aki_file_exists(str *path);
+bool aki_file_truncate(struct aki_file *file, off_t size);
+off_t aki_file_seek(struct aki_file *file, off_t offset, s32 whence);
+size_t aki_file_get_filesize(struct aki_file *file);
+bool aki_file_write(struct aki_file *file, void *buf, size_t size);
+bool aki_file_read(struct aki_file *file, void *buf, size_t size);
+s32 aki_file_read_as_str(struct aki_file *file, str *out);
+s32 aki_file_read_as_c_str(struct aki_file *file, char **out);
+void *aki_file_mmap(struct aki_file *file);
+void *aki_file_mremap(struct aki_file *file, size_t size, void *old_map);
+void aki_file_munmap(struct aki_file *file, void *map);
+void aki_file_close(struct aki_file *file);
+
+bool aki_dir_open(struct aki_dir *dir, str *path);
+void aki_dir_close(struct aki_dir *dir);
+bool aki_dir_exists(str *path);
+bool aki_dir_create(str *path);
+bool aki_dir_read(struct aki_dir *dir, struct aki_dir_entry *entry);
+void aki_dir_entry_free(struct aki_dir_entry *entry);
diff --git a/src/util/file/file_linux.c b/src/util/file/file_linux.c
new file mode 100644
index 0000000..97e9688
--- /dev/null
+++ b/src/util/file/file_linux.c
@@ -0,0 +1,203 @@
+#include <sys/mman.h>
+#include <al/log.h>
+
+#include "../error.h"
+
+#include "file.h"
+
+bool aki_file_open(struct aki_file *file, str *path, bool create)
+{
+    char *c_str = al_str_to_c_str(path);
+    s32 flags = O_RDWR;
+    if (create) flags |= O_CREAT;
+    file->fd = open(c_str, flags, 0666);
+    al_free(c_str);
+    if (file->fd == -1) {
+        al_log_error("file", "open(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
+        return false;
+    }
+    struct stat sb;
+    s32 res = fstat(file->fd, &sb);
+    if (res == -1) {
+        al_log_error("file", "fstat() failed (%s).", aki_strerror(errno));
+        close(file->fd);
+        return false;
+    }
+    file->filesize = sb.st_size;
+    return true;
+}
+
+bool aki_file_exists(str *path)
+{
+    char *c_str = al_str_to_c_str(path);
+    s32 ret = access(c_str, F_OK);
+    al_free(c_str);
+    if (ret != 0) {
+        al_log_error("file", "access(%.*s, F_OK) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
+    }
+    return ret == 0;
+}
+
+off_t aki_file_seek(struct aki_file *file, off_t offset, s32 whence)
+{
+    off_t res = lseek(file->fd, offset, whence);
+    if (res == (off_t)-1) {
+        al_log_error("file", "lseek(%jd, %d) failed (%s).", (intmax_t)offset, whence, aki_strerror(errno));
+    }
+    return res;
+}
+
+bool aki_file_truncate(struct aki_file *file, off_t size)
+{
+    s32 ret = ftruncate(file->fd, size);
+    if (ret != 0) {
+        al_log_error("file", "ftruncate(%jd) failed (%s).", (intmax_t)size, aki_strerror(errno));
+        return false;
+    }
+    file->filesize = size;
+    return true;
+}
+
+size_t aki_file_get_filesize(struct aki_file *file)
+{
+    return file->filesize;
+}
+
+#define file_io_loop(call, fail_case) \
+    size_t bc = 0; \
+    ssize_t res; \
+    do { \
+        res = call(file->fd, buf, size - bc); \
+        if (fail_case) { \
+            al_log_error("file", ""#call"() failed (%s).", aki_strerror(errno)); \
+            return false; \
+        } \
+        bc += res; \
+    } while (bc < size); \
+    return true
+
+bool aki_file_write(struct aki_file *file, void *buf, size_t size)
+{
+    file_io_loop(write, (res == -1));
+}
+
+bool aki_file_read(struct aki_file *file, void *buf, size_t size)
+{
+    file_io_loop(read, (res == -1));
+}
+
+void *aki_file_mmap(struct aki_file *file)
+{
+    void *map = mmap(NULL, file->filesize, PROT_READ | PROT_WRITE, MAP_SHARED, file->fd, 0L);
+    if (map == MAP_FAILED) {
+        al_log_error("file", "mmap() failed (%s).", aki_strerror(errno));
+        return NULL;
+    }
+    return map;
+}
+
+void *aki_file_mremap(struct aki_file *file, size_t size, void *old_map)
+{
+    printf("old_size: %zu, new_size: %zu\n", file->filesize, size);
+    void *map = mremap(old_map, file->filesize, size, MREMAP_MAYMOVE);
+    if (map == MAP_FAILED) {
+        al_log_error("file", "mremap(%p, %zu, %zu) failed (%s).", old_map,
+            file->filesize, size, aki_strerror(errno));
+        return NULL;
+    }
+    return map;
+}
+
+void aki_file_munmap(struct aki_file *file, void *map)
+{
+    s32 ret = munmap(map, file->filesize);
+    if (ret != 0) {
+        al_log_error("file", "munmap(%p, %zu) failed (%s).", map, file->filesize, aki_strerror(errno));
+    }
+}
+
+void aki_file_close(struct aki_file *file)
+{
+    close(file->fd);
+}
+
+bool aki_dir_open(struct aki_dir *dir, str *path)
+{
+    char *c_str = al_str_to_c_str(path);
+    dir->dir = opendir(c_str);
+    al_free(c_str);
+    if (!dir->dir) {
+        al_log_error("file", "opendir(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
+        return false;
+    }
+    al_str_clone(&dir->path, path);
+    return true;
+}
+
+void aki_dir_close(struct aki_dir *dir)
+{
+    al_str_free(&dir->path);
+    closedir(dir->dir);
+}
+
+bool aki_dir_exists(str *path)
+{
+    struct aki_dir dir;
+    char *c_str = al_str_to_c_str(path);
+    dir.dir = opendir(c_str);
+    al_free(c_str);
+    if (!dir.dir) {
+        if (errno != ENOENT) {
+            al_log_error("file", "opendir(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
+        }
+        return false;
+    }
+    return true;
+}
+
+bool aki_dir_create(str *path)
+{
+    char *c_str = al_str_to_c_str(path);
+    s32 ret = mkdir(c_str, 0700);
+    al_free(c_str);
+    if (ret == -1) {
+        al_log_error("file", "mkdir(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
+        return false;
+    }
+    return true;
+}
+
+bool aki_dir_read(struct aki_dir *dir, struct aki_dir_entry *entry)
+{
+    errno = 0;
+    entry->entry = readdir(dir->dir);
+    if (!entry->entry) {
+        if (errno) {
+            al_log_error("file", "readdir() failed (%s).", aki_strerror(errno));
+        }
+        return false;
+    }
+    switch (entry->entry->d_type) {
+    case DT_REG:
+    case DT_UNKNOWN:
+        entry->type = AKI_ENTRY_FILE;
+        break;
+    case DT_DIR:
+        entry->type = AKI_ENTRY_DIR;
+        break;
+    default:
+        entry->type = AKI_ENTRY_OTHER;
+        break;
+    }
+    al_str_clone(&entry->name, al_str_c(entry->entry->d_name));
+    al_str_clone(&entry->path, &dir->path);
+    al_str_cat(&entry->path, al_str_c("/"));
+    al_str_cat(&entry->path, &entry->name);
+    return true;
+}
+
+void aki_dir_entry_free(struct aki_dir_entry *entry)
+{
+    al_str_free(&entry->name);
+    al_str_free(&entry->path);
+}
diff --git a/src/util/file/file_stdio.c b/src/util/file/file_stdio.c
new file mode 100644
index 0000000..52d8761
--- /dev/null
+++ b/src/util/file/file_stdio.c
@@ -0,0 +1,100 @@
+#include <al/log.h>
+
+#include "../error.h"
+
+#include "file.h"
+
+bool aki_file_open(struct aki_file *file, str *path, bool create)
+{
+    char *c_str = al_str_to_c_str(path);
+    const char *mode = create ? "wb+" : "rb+";
+#ifndef _WIN32
+    file->file = fopen(c_str, mode);
+#else
+    errno_t ret = fopen_s(&file->file, c_str, mode);
+#endif
+    al_free(c_str);
+#ifndef _WIN32
+    if (!file->file) {
+#else
+    if (ret != 0) {
+#endif
+        al_log_error("file_stdio", "fopen(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
+        return false;
+    }
+    off_t pos = aki_file_seek(file, 0, SEEK_END);
+    if (pos == -1) {
+        aki_file_close(file);
+        return false;
+    }
+    file->filesize = pos;
+    rewind(file->file);
+    return true;
+}
+
+bool aki_file_exists(str *path)
+{
+    (void)path;
+    return false;
+}
+
+off_t aki_file_seek(struct aki_file *file, off_t offset, s32 whence)
+{
+    if (fseek(file->file, offset, whence) != 0) {
+        al_log_error("file_stdio", "fseek(%jd, %d) failed (%s).", (intmax_t)offset,
+            whence, aki_strerror(errno));
+        return -1;
+    }
+    return ftell(file->file);
+}
+
+bool aki_file_truncate(struct aki_file *file, off_t size)
+{
+    (void)file;
+    (void)size;
+    return true;
+}
+
+size_t aki_file_get_filesize(struct aki_file *file)
+{
+    return file->filesize;
+}
+
+bool aki_file_write(struct aki_file *file, void *buf, size_t size)
+{
+    size_t ret = fwrite(buf, size, 1, file->file);
+    if (ret == 0) {
+        al_log_error("file_stdio", "fwrite(%zu) failed (%s).", size, aki_strerror(errno));
+        return false;
+    }
+    return true;
+}
+
+bool aki_file_read(struct aki_file *file, void *buf, size_t size)
+{
+    size_t ret = fread(buf, size, 1, file->file);
+    if (ret == 0) {
+        al_log_error("file_stdio", "fread(%zu) failed (%s).", size, aki_strerror(errno));
+        return false;
+    }
+    return true;
+}
+
+void *aki_file_mmap(struct aki_file *file)
+{
+    (void)file;
+    al_assert(false);
+    return NULL;
+}
+
+void aki_file_close(struct aki_file *file)
+{
+    fclose(file->file);
+}
+
+bool aki_dir_open(struct aki_dir *dir, str *path) { (void)dir; (void)path; return false; }
+void aki_dir_close(struct aki_dir *dir) { (void)dir; }
+bool aki_dir_exists(str *path) { (void)path; return false; }
+bool aki_dir_create(str *path) { (void)path; return false; }
+bool aki_dir_read(struct aki_dir *dir, struct aki_dir_entry *entry) { (void)dir; (void)entry; return false; }
+void aki_dir_entry_free(struct aki_dir_entry *entry) { (void)entry; }
diff --git a/src/util/file/util.c b/src/util/file/util.c
new file mode 100644
index 0000000..e893de7
--- /dev/null
+++ b/src/util/file/util.c
@@ -0,0 +1,25 @@
+#include "file.h"
+
+s32 aki_file_read_as_str(struct aki_file *file, str *out)
+{
+    size_t size = file->filesize;
+    al_str_sized(out, size);
+    if (!aki_file_read(file, (void *)out->data, size)) {
+        al_str_free(out);
+        return -1;
+    }
+    out->len = size;
+    return size;
+}
+
+s32 aki_file_read_as_c_str(struct aki_file *file, char **out)
+{
+    size_t size = file->filesize;
+    *out = al_malloc(size + 1);
+    if (!aki_file_read(file, (void *)*out, size)) {
+        al_free(*out);
+        return -1;
+    }
+    (*out)[size] = '\n';
+    return size;
+}
diff --git a/src/util/packet.c b/src/util/packet.c
new file mode 100644
index 0000000..c58d656
--- /dev/null
+++ b/src/util/packet.c
@@ -0,0 +1,104 @@
+#include "packet.h"
+
+struct aki_packet *aki_packet_create(void)
+{
+    struct aki_packet *packet = (struct aki_packet *)al_malloc(sizeof(struct aki_packet));
+    aki_buffer_init(&packet->buffer);
+    aki_packet_reset(packet);
+    return packet;
+}
+
+struct aki_packet *aki_packet_clone(struct aki_packet *packet)
+{
+    struct aki_packet *c = aki_packet_create();
+    aki_buffer_ensure_size(&c->buffer, packet->buffer.size);
+    al_memcpy(aki_buffer_get_ptr(&c->buffer, 0), aki_buffer_get_ptr(&packet->buffer, 0), packet->buffer.size);
+    c->windex = packet->windex;
+    c->rindex = packet->rindex;
+    return c;
+}
+
+void aki_packet_reset(struct aki_packet *packet)
+{
+    aki_buffer_ensure_size(&packet->buffer, AKI_PACKET_HEADER_LENGTH);
+    packet->rindex = AKI_PACKET_HEADER_LENGTH;
+    packet->windex = AKI_PACKET_HEADER_LENGTH;
+}
+
+void aki_packet_write_size(struct aki_packet *packet)
+{
+    *((u32 *)aki_buffer_get_ptr(&packet->buffer, 0)) = packet->windex;
+}
+
+u32 aki_packet_size(struct aki_packet *packet)
+{
+    return *((u32 *)aki_buffer_get_ptr(&packet->buffer, 0));
+}
+
+#define DEFINE_PACKET_WRITE_FUNC(type) \
+    void aki_packet_write_##type(struct aki_packet *packet, type v) \
+    { \
+        AKI_PACKET_WRITE_TYPE(packet, type, v); \
+    }
+
+DEFINE_PACKET_WRITE_FUNC(u8)
+DEFINE_PACKET_WRITE_FUNC(s8)
+DEFINE_PACKET_WRITE_FUNC(u16)
+DEFINE_PACKET_WRITE_FUNC(s16)
+DEFINE_PACKET_WRITE_FUNC(u32)
+DEFINE_PACKET_WRITE_FUNC(s32)
+DEFINE_PACKET_WRITE_FUNC(u64)
+DEFINE_PACKET_WRITE_FUNC(s64)
+DEFINE_PACKET_WRITE_FUNC(f32)
+DEFINE_PACKET_WRITE_FUNC(f64)
+
+void aki_packet_write_string(struct aki_packet *packet, str *s)
+{
+    AKI_PACKET_WRITE_TYPE(packet, u32, s->len);
+    AKI_PACKET_WRITE_DATA(packet, s->data, s->len);
+}
+
+void aki_packet_write_buffer(struct aki_packet *packet, struct aki_buffer *buf)
+{
+    AKI_PACKET_WRITE_TYPE(packet, size_t, buf->size);
+    AKI_PACKET_WRITE_DATA(packet, aki_buffer_get_ptr(buf, 0), buf->size);
+}
+
+#define DEFINE_PACKET_READ_FUNC(type) \
+    type aki_packet_read_##type(struct aki_packet *packet) \
+    { \
+      type r; \
+      AKI_PACKET_READ_TYPE(packet, type, r); \
+      return r; \
+    }
+
+DEFINE_PACKET_READ_FUNC(u8)
+DEFINE_PACKET_READ_FUNC(s8)
+DEFINE_PACKET_READ_FUNC(u16)
+DEFINE_PACKET_READ_FUNC(s16)
+DEFINE_PACKET_READ_FUNC(u32)
+DEFINE_PACKET_READ_FUNC(s32)
+DEFINE_PACKET_READ_FUNC(u64)
+DEFINE_PACKET_READ_FUNC(s64)
+DEFINE_PACKET_READ_FUNC(f32)
+DEFINE_PACKET_READ_FUNC(f64)
+
+void aki_packet_read_string(struct aki_packet *packet, str *s)
+{
+    AKI_PACKET_READ_TYPE(packet, u32, s->len);
+    AKI_PACKET_READ_DATA(packet, s->len, s->data);
+    s->alloc = 0;
+}
+
+void aki_packet_read_buffer(struct aki_packet *packet, struct aki_buffer *buf)
+{
+    AKI_PACKET_READ_TYPE(packet, size_t, buf->size);
+    AKI_PACKET_READ_DATA(packet, buf->size, buf->data);
+    buf->alloc = 0;
+}
+
+void aki_packet_free(struct aki_packet *packet)
+{
+    aki_buffer_free(&packet->buffer);
+    al_free(packet);
+}
diff --git a/src/util/packet.h b/src/util/packet.h
new file mode 100644
index 0000000..14acf68
--- /dev/null
+++ b/src/util/packet.h
@@ -0,0 +1,65 @@
+#pragma once
+
+#include <al/types.h>
+#include <al/str.h>
+#include <al/array.h>
+
+#include "../util/buffer.h"
+
+#define AKI_PACKET_HEADER_LENGTH (u32)(sizeof(u32))
+
+struct aki_packet {
+    struct aki_buffer buffer;
+    u32 rindex;
+    u32 windex;
+    void *userdata;
+};
+
+#define AKI_PACKET_WRITE_TYPE(p, type, v) \
+    aki_buffer_write(&(p)->buffer, &(v), (p)->windex, sizeof(type)); \
+    (p)->windex += (u32)sizeof(type)
+
+#define AKI_PACKET_WRITE_DATA(p, data, len) \
+    aki_buffer_write(&(p)->buffer, (data), (p)->windex, (len)); \
+    (p)->windex += (u32)len
+
+#define AKI_PACKET_READ_TYPE(p, type, r) \
+    r = *((type *)aki_buffer_get_ptr(&(p)->buffer, (p)->rindex)); \
+    (p)->rindex += (u32)sizeof(type);
+
+#define AKI_PACKET_READ_DATA(p, len, r) \
+    r = (void *)aki_buffer_get_ptr(&(p)->buffer, (p)->rindex);  \
+    (p)->rindex += (u32)len;
+
+struct aki_packet *aki_packet_create(void);
+struct aki_packet *aki_packet_clone(struct aki_packet *packet);
+void aki_packet_reset(struct aki_packet *packet);
+
+void aki_packet_write_size(struct aki_packet *packet);
+u32 aki_packet_size(struct aki_packet *packet);
+
+void aki_packet_write_s8(struct aki_packet *packet, s8 v);
+void aki_packet_write_u8(struct aki_packet *packet, u8 v);
+void aki_packet_write_s16(struct aki_packet *packet, s16 v);
+void aki_packet_write_u16(struct aki_packet *packet, u16 v);
+void aki_packet_write_s32(struct aki_packet *packet, s32 v);
+void aki_packet_write_u32(struct aki_packet *packet, u32 v);
+void aki_packet_write_s64(struct aki_packet *packet, s64 v);
+void aki_packet_write_u64(struct aki_packet *packet, u64 v);
+void aki_packet_write_f32(struct aki_packet *packet, f32 v);
+void aki_packet_write_string(struct aki_packet *packet, str *s);
+void aki_packet_write_buffer(struct aki_packet *packet, struct aki_buffer *buf);
+
+s8 aki_packet_read_s8(struct aki_packet *packet);
+u8 aki_packet_read_u8(struct aki_packet *packet);
+s16 aki_packet_read_s16(struct aki_packet *packet);
+u16 aki_packet_read_u16(struct aki_packet *packet);
+s32 aki_packet_read_s32(struct aki_packet *packet);
+u32 aki_packet_read_u32(struct aki_packet *packet);
+s64 aki_packet_read_s64(struct aki_packet *packet);
+u64 aki_packet_read_u64(struct aki_packet *packet);
+f32 aki_packet_read_f32(struct aki_packet *packet);
+void aki_packet_read_string(struct aki_packet *packet, str *s);
+void aki_packet_read_buffer(struct aki_packet *packet, struct aki_buffer *buf);
+
+void aki_packet_free(struct aki_packet *packet);
diff --git a/src/util/thread/thread.h b/src/util/thread/thread.h
new file mode 100644
index 0000000..eb63b8d
--- /dev/null
+++ b/src/util/thread/thread.h
@@ -0,0 +1,66 @@
+#pragma once
+
+#include <al/types.h>
+
+#ifndef _WIN32
+#include <pthread.h>
+#else
+#include "../../winwrap.h"
+#endif
+
+struct aki_thread {
+#ifndef _WIN32
+    pthread_t thread;
+#else
+    HANDLE thread;
+#endif
+};
+
+struct aki_mutex {
+#ifndef _WIN32
+    pthread_mutex_t mutex;
+#else
+    CRITICAL_SECTION mutex;
+#endif
+};
+
+struct aki_cond {
+#ifndef _WIN32
+    pthread_cond_t cond;
+#else
+    CONDITION_VARIABLE cond;
+#endif
+    bool condition;
+};
+
+typedef f64 aki_os_tstamp;
+
+#define AKI_TS_FROM_USEC(usec) (usec * 1e-6)
+
+#ifndef _WIN32
+#define AKI_THREADCALL
+typedef void * aki_thread_result;
+#else
+#define AKI_THREADCALL __stdcall
+typedef unsigned long aki_thread_result;
+#endif
+
+typedef aki_thread_result (AKI_THREADCALL * aki_thread_func)(void *);
+
+void aki_thread_sleep(aki_os_tstamp ts);
+
+void aki_thread_create(struct aki_thread *thread, aki_thread_func func, void *userdata);
+void aki_thread_join(struct aki_thread *thread);
+//void aki_thread_cancel(struct aki_thread *thread);
+//void aki_thread_detach(struct aki_thread *thread);
+
+void aki_mutex_init(struct aki_mutex *mutex);
+void aki_mutex_lock(struct aki_mutex *mutex);
+void aki_mutex_unlock(struct aki_mutex *mutex);
+void aki_mutex_destroy(struct aki_mutex *mutex);
+
+void aki_cond_init(struct aki_cond *cond);
+void aki_cond_wait(struct aki_cond *cond, struct aki_mutex *mutex);
+bool aki_cond_is_waiting(struct aki_cond *cond);
+void aki_cond_signal(struct aki_cond *cond);
+void aki_cond_destroy(struct aki_cond *cond);
diff --git a/src/util/thread/thread_linux.c b/src/util/thread/thread_linux.c
new file mode 100644
index 0000000..0214ed8
--- /dev/null
+++ b/src/util/thread/thread_linux.c
@@ -0,0 +1,86 @@
+#include <al/types.h>
+#include <time.h>
+
+#include "../error.h"
+
+#include "thread.h"
+
+void aki_thread_sleep(aki_os_tstamp ts)
+{
+    struct timespec tv;
+    tv.tv_sec = (s64)ts;
+    tv.tv_nsec = (s64)((ts - tv.tv_sec) * 1e9);
+    while (nanosleep(&tv, &tv) == -1 && errno == EINTR) {}
+}
+
+void aki_thread_create(struct aki_thread *thread, aki_thread_func func, void *userdata)
+{
+    pthread_create(&thread->thread, NULL, func, userdata);
+}
+
+void aki_thread_join(struct aki_thread *thread)
+{
+    pthread_join(thread->thread, NULL);
+}
+
+/*
+void aki_thread_cancel(struct aki_thread *thread)
+{
+    pthread_cancel(thread->thread);
+}
+
+void aki_thread_detach(struct aki_thread *thread)
+{
+    pthread_detach(thread->thread);
+}
+*/
+
+void aki_mutex_init(struct aki_mutex *mutex)
+{
+    pthread_mutex_init(&mutex->mutex, NULL);
+}
+
+void aki_mutex_lock(struct aki_mutex *mutex)
+{
+    pthread_mutex_lock(&mutex->mutex);
+}
+
+void aki_mutex_unlock(struct aki_mutex *mutex)
+{
+    pthread_mutex_unlock(&mutex->mutex);
+}
+
+void aki_mutex_destroy(struct aki_mutex *mutex)
+{
+    pthread_mutex_destroy(&mutex->mutex);
+}
+
+void aki_cond_init(struct aki_cond *cond)
+{
+    cond->condition = true;
+    pthread_cond_init(&cond->cond, NULL);
+}
+
+void aki_cond_wait(struct aki_cond *cond, struct aki_mutex *mutex)
+{
+    cond->condition = false;
+    while (!cond->condition) {
+        pthread_cond_wait(&cond->cond, &mutex->mutex);
+    }
+}
+
+bool aki_cond_is_waiting(struct aki_cond *cond)
+{
+    return !cond->condition;
+}
+
+void aki_cond_signal(struct aki_cond *cond)
+{
+    cond->condition = true;
+    pthread_cond_signal(&cond->cond);
+}
+
+void aki_cond_destroy(struct aki_cond *cond)
+{
+    pthread_cond_destroy(&cond->cond);
+}
diff --git a/src/util/thread/thread_windows.c b/src/util/thread/thread_windows.c
new file mode 100644
index 0000000..9ea9ead
--- /dev/null
+++ b/src/util/thread/thread_windows.c
@@ -0,0 +1,81 @@
+#include "thread.h"
+
+NTSTATUS(__stdcall *NtDelayExecution)(BOOL Alertable, PLARGE_INTEGER DelayInterval);
+
+void aki_thread_sleep(aki_os_tstamp ts)
+{
+    LARGE_INTEGER intr;
+    intr.QuadPart = -1 * (s64)(ts * 1e7);
+    NtDelayExecution(false, &intr);
+}
+
+void aki_thread_create(struct aki_thread *thread, aki_thread_func func, void *userdata)
+{
+    thread->thread = CreateThread(NULL, 0, func, userdata, 0, NULL);
+}
+
+void aki_thread_join(struct aki_thread *thread)
+{
+    WaitForSingleObject(thread->thread, INFINITE);
+    CloseHandle(thread->thread);
+}
+
+/*
+void aki_thread_cancel(struct aki_thread *thread)
+{
+}
+
+void aki_thread_detach(struct aki_thread *thread)
+{
+}
+*/
+
+void aki_mutex_init(struct aki_mutex *mutex)
+{
+    InitializeCriticalSection(&mutex->mutex);
+}
+
+void aki_mutex_lock(struct aki_mutex *mutex)
+{
+    EnterCriticalSection(&mutex->mutex);
+}
+
+void aki_mutex_unlock(struct aki_mutex *mutex)
+{
+    LeaveCriticalSection(&mutex->mutex);
+}
+
+void aki_mutex_destroy(struct aki_mutex *mutex)
+{
+    DeleteCriticalSection(&mutex->mutex);
+}
+
+void aki_cond_init(struct aki_cond *cond)
+{
+    cond->condition = true;
+    InitializeConditionVariable(&cond->cond);
+}
+
+void aki_cond_wait(struct aki_cond *cond, struct aki_mutex *mutex)
+{
+    cond->condition = false;
+    while (!cond->condition) {
+        SleepConditionVariableCS(&cond->cond, &mutex->mutex, INFINITE);
+    }
+}
+
+bool aki_cond_is_waiting(struct aki_cond *cond)
+{
+    return !cond->condition;
+}
+
+void aki_cond_signal(struct aki_cond *cond)
+{
+    cond->condition = true;
+    WakeConditionVariable(&cond->cond);
+}
+
+void aki_cond_destroy(struct aki_cond *cond)
+{
+    (void)cond;
+}
diff --git a/src/util/timer/timer.h b/src/util/timer/timer.h
new file mode 100644
index 0000000..2bf6b95
--- /dev/null
+++ b/src/util/timer/timer.h
@@ -0,0 +1,6 @@
+#pragma once
+
+#include <al/types.h>
+
+u64 aki_get_timestamp(void);
+f64 aki_get_tick(void);
diff --git a/src/util/timer/timer_linux.c b/src/util/timer/timer_linux.c
new file mode 100644
index 0000000..d93bfdc
--- /dev/null
+++ b/src/util/timer/timer_linux.c
@@ -0,0 +1,17 @@
+#include <time.h>
+
+#include "timer.h"
+
+u64 aki_get_timestamp(void)
+{
+    struct timespec spec;
+    clock_gettime(CLOCK_REALTIME, &spec);
+    return 1e6 * spec.tv_sec + spec.tv_nsec * 1e-3;
+}
+
+f64 aki_get_tick(void)
+{
+    struct timespec spec;
+    clock_gettime(CLOCK_MONOTONIC, &spec);
+    return spec.tv_sec + spec.tv_nsec * 1e-9;
+}
diff --git a/src/util/timer/timer_windows.c b/src/util/timer/timer_windows.c
new file mode 100644
index 0000000..37a8fd8
--- /dev/null
+++ b/src/util/timer/timer_windows.c
@@ -0,0 +1,39 @@
+#include <al/lib.h>
+
+#include "../../winwrap.h"
+
+#include "timer.h"
+
+f64 aki_get_tick(void)
+{
+    LARGE_INTEGER result;
+    QueryPerformanceCounter(&result);
+    return (result.QuadPart/(f64)performance_frequency.QuadPart);
+}
+
+static SYSTEMTIME unix_epoch = {
+    .wYear = 1970,
+    .wMonth = 1,
+    .wDayOfWeek = 4,
+    .wDay = 1,
+    .wHour = 0,
+    .wMinute = 0,
+    .wSecond = 0,
+    .wMilliseconds = 0
+};
+
+u64 aki_get_timestamp(void)
+{
+    FILETIME result0, result1;
+    SystemTimeToFileTime(&unix_epoch, &result0);
+    ULARGE_INTEGER li0 = {
+        .LowPart = result0.dwLowDateTime,
+        .HighPart = result0.dwHighDateTime
+    };
+    GetSystemTimePreciseAsFileTime(&result1);
+    ULARGE_INTEGER li1 = {
+        .LowPart = result1.dwLowDateTime,
+        .HighPart = result1.dwHighDateTime
+    };
+    return (li1.QuadPart - li0.QuadPart) / 10;
+}
diff --git a/src/winwrap.c b/src/winwrap.c
new file mode 100644
index 0000000..a0a1229
--- /dev/null
+++ b/src/winwrap.c
@@ -0,0 +1,35 @@
+#include "winwrap.h"
+
+void aki_windows_init(void)
+{
+#ifdef _WIN32
+    CoInitialize(NULL);
+
+    WSADATA data;
+    WSAStartup(MAKEWORD(2,2), &data);
+
+    ULONG actual_resolution;
+    NTSTATUS(__stdcall *ZwSetTimerResolution)(
+        IN ULONG RequestedResolution, IN BOOLEAN Set, OUT PULONG ActualResolution) = (
+        NTSTATUS(__stdcall*)(ULONG, BOOLEAN, PULONG))GetProcAddress(
+        GetModuleHandleW(L"ntdll"), "ZwSetTimerResolution");
+    ZwSetTimerResolution(1, TRUE, &actual_resolution);
+
+    NtDelayExecution = (NTSTATUS(__stdcall*)(BOOL, PLARGE_INTEGER))GetProcAddress(
+        GetModuleHandleW(L"ntdll"), "NtDelayExecution");
+
+    QueryPerformanceFrequency(&performance_frequency);
+#endif
+}
+
+void aki_windows_close(void)
+{
+#ifdef _WIN32
+    WSACleanup();
+    CoUninitialize();
+#endif
+}
+
+#ifdef _WIN32
+LARGE_INTEGER performance_frequency;
+#endif
diff --git a/src/winwrap.h b/src/winwrap.h
new file mode 100644
index 0000000..83cf7ea
--- /dev/null
+++ b/src/winwrap.h
@@ -0,0 +1,15 @@
+#pragma once
+
+#include <al/lib.h>
+
+#ifdef _WIN32
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#include <io.h>
+#include <windows.h>
+extern NTSTATUS(__stdcall *NtDelayExecution)(BOOL Alertable, PLARGE_INTEGER DelayInterval);
+extern LARGE_INTEGER performance_frequency;
+#endif
+
+void aki_windows_init(void);
+void aki_windows_close(void);