Tons of changes

Mainly for things that came up in other projects.

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

33 files changed, 376 insertions, 246 deletions

diff --git a/src/common.c b/src/common.c
index ae9edf7..3abba13 100644
--- a/src/common.c
+++ b/src/common.c
@@ -29,7 +29,9 @@ bool aki_common_init(void)
     al_set_page_size(info.dwPageSize);
 #endif
 
+#ifndef _WIN32
     signal(SIGPIPE, SIG_IGN);
+#endif
 
     return true;
 }
diff --git a/src/curl/curl.c b/src/curl/curl.c
index 3f4cb9e..5560487 100644
--- a/src/curl/curl.c
+++ b/src/curl/curl.c
@@ -13,7 +13,7 @@ static void curl_socket_action_callback(struct ev_loop *loop, ev_io *w, s32 reve
         al_log_error("curl", "curl_multi_socket_action() failed (%s).", curl_multi_strerror(mc));
         return;
     }
-    curl->handle_events(curl);
+    curl->handle_events(curl->userdata, curl);
 }
 
 static void set_sock(struct aki_curl *curl, s32 what)
@@ -53,7 +53,7 @@ static void timeout_callback(struct ev_loop *loop, ev_timer *w, s32 revents)
     curl->timer_started = false;
     s32 running;
     curl_multi_socket_action(curl->multi_handle, CURL_SOCKET_TIMEOUT, 0, &running);
-    curl->handle_events(curl);
+    curl->handle_events(curl->userdata, curl);
 }
 
 static s32 timer_callback(CURLM *multi, s64 timeout_ms, void *userp)
diff --git a/src/curl/curl.h b/src/curl/curl.h
index 1850dcf..5bb38c0 100644
--- a/src/curl/curl.h
+++ b/src/curl/curl.h
@@ -13,7 +13,8 @@ struct aki_curl {
     CURL *handle;
     CURLM *multi_handle;
     curl_socket_t sock;
-    void (*handle_events)(struct aki_curl *);
+    void (*handle_events)(void *, struct aki_curl *);
+    void *userdata;
 };
 
 bool aki_curl_init(struct aki_curl *curl);
diff --git a/src/curl/http.c b/src/curl/http.c
index c58d5a6..cdd740b 100644
--- a/src/curl/http.c
+++ b/src/curl/http.c
@@ -19,8 +19,9 @@ static void check_status_codes(struct aki_curl *curl)
     }
 }
 
-static void http_handle_events(struct aki_curl *curl)
+static void http_handle_events(void *userdata, struct aki_curl *curl)
 {
+    (void)userdata;
     s32 pending;
     CURLMsg *msg;
     struct aki_http *http = (struct aki_http *)curl;
diff --git a/src/curl/websocket.c b/src/curl/websocket.c
index 4d480c3..aede3de 100644
--- a/src/curl/websocket.c
+++ b/src/curl/websocket.c
@@ -3,8 +3,9 @@
 
 #include "websocket.h"
 
-static void websocket_handle_events(struct aki_curl *curl)
+static void websocket_handle_events(void *userdata, struct aki_curl *curl)
 {
+    struct aki_websocket *ws = (struct aki_websocket *)userdata;
     s32 pending;
     CURLMsg *msg;
     while ((msg = curl_multi_info_read(curl->multi_handle, &pending))) {
@@ -12,6 +13,8 @@ static void websocket_handle_events(struct aki_curl *curl)
         case CURLMSG_DONE:
             aki_curl_remove_handle(curl);
             aki_curl_close(curl);
+            aki_buffer_free(&ws->frame);
+            ws->callback(ws->userdata, NULL, NULL, 0);
             return;
         case CURLMSG_LAST:
             break;
@@ -51,13 +54,17 @@ void aki_websocket_set_url(struct aki_websocket *ws, str *url)
 bool aki_websocket_connect(struct aki_websocket *ws, struct aki_event_loop *loop,
     void (*callback)(void *, const struct curl_ws_frame *, u8 *, size_t), void *userdata)
 {
-    al_assert(ws->curl.loop == NULL);
+    // Connect blocking.
+    //curl_easy_setopt(ws->curl.handle, CURLOPT_CONNECT_ONLY, 2L);
+    //curl_easy_perform(ws->curl.handle);
+    curl_easy_setopt(ws->curl.handle, CURLOPT_CONNECT_ONLY, 0L);
     ws->curl.loop = loop;
     ws->callback = callback;
     ws->userdata = userdata;
     curl_easy_setopt(ws->curl.handle, CURLOPT_WRITEFUNCTION, stream_write_callback);
     curl_easy_setopt(ws->curl.handle, CURLOPT_WRITEDATA, ws);
     ws->curl.handle_events = websocket_handle_events;
+    ws->curl.userdata = ws;
     return aki_curl_add_handle(&ws->curl);
 }
 
@@ -69,13 +76,14 @@ void aki_websocket_disconnect(struct aki_websocket *ws, u16 status)
     al_assert(sent == sizeof(u16));
 }
 
-//aki_websocket_close();
-//aki_buffer_free(&ws->frame);
-
 bool aki_websocket_send_frame(struct aki_websocket *ws, u8 *data, size_t size)
 {
     size_t sent;
-    curl_ws_send(ws->curl.handle, data, size, &sent, 0, CURLWS_TEXT);
+    CURLcode code = curl_ws_send(ws->curl.handle, data, size, &sent, 0, CURLWS_TEXT);
+    if (code != CURLE_OK) {
+        al_log_error("curl", "curl_ws_send() failed (%s).", curl_easy_strerror(code));
+        return false;
+    }
     al_assert(sent == size);
     return true;
 }
diff --git a/src/curl/websocket.h b/src/curl/websocket.h
index 3dc740a..276f6bc 100644
--- a/src/curl/websocket.h
+++ b/src/curl/websocket.h
@@ -1,6 +1,8 @@
 #pragma once
 
-#include "http.h"
+#include "../util/buffer.h"
+
+#include "curl.h"
 
 struct aki_websocket {
     struct aki_curl curl;
diff --git a/src/ev_embed_compat.c b/src/ev_embed_compat.c
index ab98423..4c7a1f2 100644
--- a/src/ev_embed_compat.c
+++ b/src/ev_embed_compat.c
@@ -21,11 +21,14 @@ _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
+_Pragma("GCC diagnostic ignored \"-Wdangling-else\"")
+_Pragma("GCC diagnostic ignored \"-Wtype-limits\"")
+#elif _MSC_VER
+#pragma warning(push, 0)
 #endif
 #include <ev.c>
 #if defined(__clang__) || defined(__GNUC__)
 _Pragma("GCC diagnostic pop")
+#elif _MSC_VER
+#pragma warning(pop)
 #endif
diff --git a/src/fs_event/fs_event.h b/src/fs_event/fs_event.h
index 0240e6f..d4983d4 100644
--- a/src/fs_event/fs_event.h
+++ b/src/fs_event/fs_event.h
@@ -4,6 +4,10 @@
 
 #ifndef _WIN32
 #include <sys/inotify.h>
+#define AKI_CLOSE_WRITE IN_CLOSE_WRITE
+#else
+struct inotify_event {};
+#define AKI_CLOSE_WRITE 0
 #endif
 
 #include "../loop.h"
diff --git a/src/loop.c b/src/loop.c
index 6598768..54ffbcd 100644
--- a/src/loop.c
+++ b/src/loop.c
@@ -1,3 +1,5 @@
+#include "util/timer/timer.h"
+
 #include "loop.h"
 
 bool aki_event_loop_init(struct aki_event_loop *loop)
@@ -15,7 +17,20 @@ s32 aki_event_loop_run(struct aki_event_loop *loop)
 
 s32 aki_event_loop_run_once(struct aki_event_loop *loop)
 {
-    return ev_run(loop->ev, EVRUN_ONCE);
+    return ev_run(loop->ev, EVRUN_ONCE | EVRUN_NOWAIT);
+}
+
+void aki_event_loop_sleep(struct aki_event_loop *loop, aki_os_tstamp delay)
+{
+    al_assert(delay >= AKI_TS_FROM_USEC(1000));
+    u64 prev = aki_get_timestamp(), now;
+    while (delay >= 0.0) {
+        aki_event_loop_run_once(loop);
+        aki_thread_sleep(AKI_TS_FROM_USEC(1000));
+        now = aki_get_timestamp();
+        delay -= AKI_TS_FROM_USEC(now - prev);
+        prev = now;
+    }
 }
 
 void aki_event_loop_break_one(struct aki_event_loop *loop)
diff --git a/src/loop.h b/src/loop.h
index aba49d9..2b48e38 100644
--- a/src/loop.h
+++ b/src/loop.h
@@ -2,6 +2,8 @@
 
 #include <al/types.h>
 
+#include "util/thread/thread.h"
+
 #include "evwrap.h"
 
 struct aki_event_loop {
@@ -11,6 +13,7 @@ struct aki_event_loop {
 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_sleep(struct aki_event_loop *loop, aki_os_tstamp delay);
 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);
diff --git a/src/multiplex.c b/src/multiplex.c
index d3bb14e..ea8ca9f 100644
--- a/src/multiplex.c
+++ b/src/multiplex.c
@@ -31,7 +31,8 @@ static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 reven
 
 bool aki_multiplex_socket_listen(struct aki_multiplex_socket *multi, struct aki_event_loop *loop, str *addr, s32 port)
 {
-    if (!aki_socket_listen(&multi->sock, addr, port)) return false;
+    if (!aki_socket_bind(&multi->sock, addr, port)) return false;
+    if (!aki_socket_listen(&multi->sock)) return false;
     multi->loop = loop;
     multi->event.data = multi;
     ev_io_init(&multi->event, socket_connection_callback, aki_socket_get_fd(&multi->sock), EV_READ);
diff --git a/src/packet_cache.c b/src/packet_cache.c
index e08bee7..209b3a4 100644
--- a/src/packet_cache.c
+++ b/src/packet_cache.c
@@ -4,6 +4,7 @@ void aki_packet_cache_init(struct aki_packet_cache *cache, u32 size)
 {
     al_array_init(cache->cache);
     al_array_reserve(cache->cache, size);
+    cache->flush = size / 2;
     aki_cond_init(&cache->cond);
     aki_mutex_init(&cache->mutex);
     cache->disabled = false;
@@ -17,42 +18,44 @@ bool aki_packet_cache_send_packet(struct aki_packet_cache *cache, struct aki_pac
         return false;
     }
     al_array_push(cache->cache, packet);
-    if (aki_cond_is_waiting(&cache->cond)) {
+    bool flush = !packet || cache->cache.size >= cache->flush;
+    if (flush && 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)
+bool aki_packet_cache_wait(struct aki_packet_cache *cache, u32 *count)
 {
     aki_mutex_lock(&cache->mutex);
-    if (cache->disabled) {
-        aki_mutex_unlock(&cache->mutex);
-        return false;
-    }
-    if (cache->cache.size == 0) {
+    if (!cache->disabled && cache->cache.size == 0) {
         aki_cond_wait(&cache->cond, &cache->mutex);
     }
+    // If there was an active wait before calling disable(), expected
+    // behavior would be that wait() returns false.
     if (cache->disabled) {
         aki_mutex_unlock(&cache->mutex);
         return false;
     }
-    aki_mutex_unlock(&cache->mutex);
+    *count = cache->cache.size;
     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_unlock(struct aki_packet_cache *cache)
+{
+    aki_mutex_unlock(&cache->mutex);
+}
+
 void aki_packet_cache_disable(struct aki_packet_cache *cache)
 {
     aki_mutex_lock(&cache->mutex);
diff --git a/src/packet_cache.h b/src/packet_cache.h
index ff036b5..9045e54 100644
--- a/src/packet_cache.h
+++ b/src/packet_cache.h
@@ -7,6 +7,7 @@
 
 struct aki_packet_cache {
     array(struct aki_packet *) cache;
+    u32 flush;
     struct aki_cond cond;
     struct aki_mutex mutex;
     bool disabled;
@@ -14,8 +15,9 @@ struct aki_packet_cache {
 
 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);
+bool aki_packet_cache_wait(struct aki_packet_cache *cache, u32 *count);
 struct aki_packet *aki_packet_cache_pop(struct aki_packet_cache *cache);
+void aki_packet_cache_unlock(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
index 683efcb..a3ca57b 100644
--- a/src/packet_pool.c
+++ b/src/packet_pool.c
@@ -15,6 +15,10 @@ static void signal_callback(struct ev_loop *loop, ev_async *w, s32 revents)
     struct aki_packet_pool *pool = (struct aki_packet_pool *)w->data;
     (void)revents;
     aki_mutex_lock(&pool->mutex);
+    if (pool->disabled) {
+        aki_mutex_unlock(&pool->mutex);
+        return;
+    }
     while (pool->ready.size > 0) {
         struct aki_packet *packet;
         al_array_pop_at(pool->ready, 0, packet);
diff --git a/src/packet_stream.c b/src/packet_stream.c
index f653c54..5ae91f1 100644
--- a/src/packet_stream.c
+++ b/src/packet_stream.c
@@ -1,5 +1,12 @@
 #include "packet_stream.h"
 
+enum {
+    PACKET_STREAM_CONNECTING = 0,
+    PACKET_STREAM_CONNECTED,
+    PACKET_STREAM_DISCONNECTING,
+    PACKET_STREAM_DISCONNECTED
+};
+
 static void init_internal(struct aki_packet_stream *stream)
 {
     al_array_init(stream->out.queue);
@@ -20,7 +27,7 @@ bool aki_packet_stream_init(struct aki_packet_stream *stream, u8 type,
     }
     aki_socket_set_blocking(&stream->sock, false);
     stream->corked = false;
-    stream->connected = false;
+    stream->connect = PACKET_STREAM_DISCONNECTED;
     stream->connection_callback = connection_callback;
     stream->connection_closed_callback = connection_closed_callback;
     stream->packet_callback = NULL;
@@ -48,18 +55,23 @@ static void stop_write_internal(struct aki_packet_stream *stream)
 
 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->corked) {
-        ev_io_stop(stream->loop->ev, &stream->revent);
-    }
+    stream->connect = PACKET_STREAM_DISCONNECTED;
     if (stream->out.have_data) {
         stop_write_internal(stream);
     }
     aki_socket_close(&stream->sock);
+    if (stream->out.packet) {
+        stream->packet_sent_callback(stream->userdata, stream->out.packet);
+        stream->out.packet = NULL;
+    }
+    struct aki_packet *packet;
+    al_array_foreach(stream->out.queue, i, packet) {
+        stream->packet_sent_callback(stream->userdata, packet);
+    }
+    stream->out.queue.size = 0;
+    aki_packet_reset(stream->in.packet);
+    stream->in.have_size = false;
+    stream->in.index = 0;
     stream->connection_closed_callback(stream->userdata, stream);
 }
 
@@ -67,12 +79,13 @@ 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;
+    al_assert(revents & EV_READ); // Assert on EV_ERROR.
     // EV_READ can mean any of POLLIN, POLLERR, or POLLHUP.
     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);
+    u32 size = stream->in.have_size ? aki_packet_size(stream->in.packet) : AKI_PACKET_HEADER_LENGTH;
     ssize_t ret = aki_socket_read(&stream->sock, ptr, size - stream->in.index);
-    if (ret <= 0) {
+    if (ret <= 0 || stream->connect == PACKET_STREAM_DISCONNECTING) {
+        ev_io_stop(stream->loop->ev, &stream->revent);
         stop_internal(stream);
         return;
     }
@@ -88,7 +101,7 @@ static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
         stream->in.index = 0;
         stream->in.have_size = false;
         stream->packet_callback(stream->userdata, stream, packet);
-        // Stream could be invalid after this point.
+        // Stream could be invalid at this point.
     }
 }
 
@@ -96,19 +109,22 @@ 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;
+    al_assert(revents & EV_WRITE); // Assert on EV_ERROR.
     // EV_WRITE can mean any of POLLOUT, POLLERR, or POLLHUP.
-    if (UNLIKELY(!stream->connected)) {
-        stream->connected = true;
+    if (UNLIKELY(stream->connect == PACKET_STREAM_CONNECTING)) {
         // Shove in the multiplex ID byte here assuming POLLOUT means
         // we can write a minimum of 1 byte.
-        // If there is a case where POLLOUT can end up writing 0 bytes, we
-        // should wait to set stream->connected until write() returns >0 and
-        // return before connection_callback().
         ssize_t ret = aki_socket_write(&stream->sock, &stream->id, sizeof(u8));
-        // Assume any error (ret < 0) will be resolved in the read callback.
-        al_assert(ret <= 0 || ret == sizeof(u8));
-        // Packet callbacks are set in connection_callback.
+        if (ret < 0) {
+            // It seems POLLOUT can be signaled even if any write() will error.
+            stop_internal(stream);
+            return;
+        } else if (ret == 0) {
+            // Try again on next POLLOUT. Not sure if this can happen.
+            return;
+        }
+        stream->connect = PACKET_STREAM_CONNECTED;
+        // Callbacks must be set in connection_callback.
         stream->connection_callback(stream->userdata, stream);
         al_assert(stream->packet_callback);
         al_assert(stream->packet_sent_callback);
@@ -143,7 +159,7 @@ void aki_packet_stream_from_socket(struct aki_packet_stream *stream, struct aki_
 {
     stream->loop = loop;
     stream->corked = false;
-    stream->connected = true;
+    stream->connect = PACKET_STREAM_CONNECTED;
     stream->sock = *sock;
     aki_socket_set_blocking(&stream->sock, false);
     init_internal(stream);
@@ -157,16 +173,6 @@ void aki_packet_stream_from_socket(struct aki_packet_stream *stream, struct aki_
     ev_io_start(stream->loop->ev, &stream->revent);
 }
 
-/*
-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;
@@ -179,6 +185,7 @@ static void connect_internal(struct aki_packet_stream *stream, str *addr, s32 po
         stream->connection_closed_callback(stream->userdata, stream);
         return;
     }
+    stream->connect = PACKET_STREAM_CONNECTING;
     stream->revent.data = stream;
     ev_io_init(&stream->revent, stream_read_callback,
         aki_socket_get_fd(&stream->sock), EV_READ);
@@ -186,10 +193,6 @@ static void connect_internal(struct aki_packet_stream *stream, str *addr, s32 po
     ev_io_init(&stream->wevent, stream_write_callback,
         aki_socket_get_fd(&stream->sock), EV_WRITE);
     start_write_internal(stream);
-    //stream->timer.data = stream;
-    //ev_timer_init(&stream->timer, timeout_callback, 5.0, 0.0);
-    //stream->timer_started = true;
-    //ev_timer_start(loop->ev, &stream->timer);
 }
 
 void aki_packet_stream_connect(struct aki_packet_stream *stream, struct aki_event_loop *loop, str *addr, s32 port)
@@ -200,32 +203,35 @@ void aki_packet_stream_connect(struct aki_packet_stream *stream, struct aki_even
 
 void aki_packet_stream_reconnect(struct aki_packet_stream *stream, str *addr, s32 port)
 {
+    al_assert(stream->connect == PACKET_STREAM_DISCONNECTED);
     if (!aki_socket_init(&stream->sock)) {
         stream->connection_closed_callback(stream->userdata, stream);
         return;
     }
-    aki_packet_reset(stream->in.packet);
-    stream->in.have_size = false;
-    stream->in.index = 0;
     aki_socket_set_blocking(&stream->sock, false);
     connect_internal(stream, addr, port);
 }
 
 void aki_packet_stream_cork(struct aki_packet_stream *stream, bool cork)
 {
-    if (stream->connected) {
+    if (stream->connect == PACKET_STREAM_CONNECTED) {
         if (cork && !stream->corked) {
             ev_io_stop(stream->loop->ev, &stream->revent);
         } else if (!cork && stream->corked) {
             ev_io_start(stream->loop->ev, &stream->revent);
         }
     }
+    // Allow stream->corked to be changed even if it has no immediate effect.
     stream->corked = cork;
 }
 
 void aki_packet_stream_send_packet(struct aki_packet_stream *stream, struct aki_packet *packet)
 {
-    al_assert(stream->connected);
+    if (stream->connect == PACKET_STREAM_DISCONNECTING) {
+        stream->packet_sent_callback(stream->userdata, packet);
+        return;
+    }
+    al_assert(stream->connect == PACKET_STREAM_CONNECTED);
     aki_packet_write_size(packet);
     al_array_push(stream->out.queue, packet);
     if (!stream->out.have_data) {
@@ -235,23 +241,26 @@ void aki_packet_stream_send_packet(struct aki_packet_stream *stream, struct aki_
 
 void aki_packet_stream_disconnect(struct aki_packet_stream *stream)
 {
-    if (!stream->connected) return;
+    al_assert(stream->connect != PACKET_STREAM_DISCONNECTING);
+    // To reuse this stream is may need to be uncorked. That should
+    // be tracked externally.
+    if (stream->connect == PACKET_STREAM_DISCONNECTED) {
+        return;
+    }
+    stream->connect = PACKET_STREAM_DISCONNECTING;
+    // If we are not corked, this relies on stream_read_callback() to resolve and call stop_internal().
+    // NOTE: read() can still return data after shutdown().
     aki_socket_shutdown(&stream->sock);
-    stop_internal(stream);
+    if (stream->corked || stream->connect == PACKET_STREAM_CONNECTING) {
+        stop_internal(stream);
+    }
 }
 
 void aki_packet_stream_free(struct aki_packet_stream *stream)
 {
-    // Loosely assert that stop_internal has ran.
+    // Loosely assert that stop_internal has run.
     al_assert(!stream->out.have_data);
     al_assert(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
index b678ba7..f0f9cc2 100644
--- a/src/packet_stream.h
+++ b/src/packet_stream.h
@@ -14,9 +14,7 @@ struct aki_packet_stream {
     ev_io wevent;
     bool corked;
     u8 id; // multiplex
-    //ev_timer timer;
-    //bool timer_started;
-    bool connected;
+    u8 connect;
     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 *);
diff --git a/src/rpc.c b/src/rpc.c
index 0c0046a..d208f0b 100644
--- a/src/rpc.c
+++ b/src/rpc.c
@@ -23,6 +23,7 @@ void aki_rpc_add_command(struct aki_rpc *rpc, struct aki_rpc_command *command)
 static void packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet)
 {
     struct aki_rpc_connection *conn = (struct aki_rpc_connection *)userdata;
+    al_assert(!conn->disconnected);
     s8 op = aki_packet_read_s8(packet);
     u32 id = aki_packet_read_u32(packet);
     if (op == -1) { // Response
@@ -63,7 +64,9 @@ static void stream_connection_callback(void *userdata, struct aki_packet_stream
 {
     struct aki_rpc_connection *conn = (struct aki_rpc_connection *)userdata;
     struct aki_rpc *rpc = conn->rpc;
-    aki_packet_stream_set_nodelay(stream, 1);
+    if (stream->sock.type == AKI_SOCKET_TCP) {
+        aki_packet_stream_set_nodelay(stream, 1);
+    }
     stream->packet_callback = packet_callback;
     stream->packet_sent_callback = packet_sent_callback;
     stream->userdata = conn;
@@ -83,6 +86,7 @@ void aki_rpc_add_socket(struct aki_rpc *rpc, struct aki_socket *sock)
 {
     struct aki_rpc_connection *conn = al_alloc_object(struct aki_rpc_connection);
     conn->rpc = rpc;
+    conn->disconnected = false;
     al_array_init(conn->callbacks);
     conn->stream.connection_callback = stream_connection_callback;
     conn->stream.connection_closed_callback = stream_connection_closed_callback;
@@ -95,6 +99,7 @@ bool aki_rpc_prepare_client(struct aki_rpc *rpc, u8 type, u8 id)
     al_assert(!rpc->conn);
     struct aki_rpc_connection *conn = al_alloc_object(struct aki_rpc_connection);
     rpc->conn = conn;
+    conn->disconnected = false;
     al_array_push(rpc->connections, conn);
     if (!aki_packet_stream_init(&conn->stream, type, stream_connection_callback,
         stream_connection_closed_callback, conn)) {
@@ -102,7 +107,9 @@ bool aki_rpc_prepare_client(struct aki_rpc *rpc, u8 type, u8 id)
     }
     conn->rpc = rpc;
     al_array_init(conn->callbacks);
-    aki_packet_stream_set_nodelay(&conn->stream, 1);
+    if (conn->stream.sock.type == AKI_SOCKET_TCP) {
+        aki_packet_stream_set_nodelay(&conn->stream, 1);
+    }
     aki_packet_stream_set_multiplex(&conn->stream, id);
     return true;
 }
@@ -149,5 +156,6 @@ void aki_rpc_connection_command(struct aki_rpc_connection *conn, struct aki_pack
 
 void aki_rpc_conn_disconnect(struct aki_rpc_connection *conn)
 {
+    conn->disconnected = true;
     aki_packet_stream_disconnect(&conn->stream);
 }
diff --git a/src/rpc2.h b/src/rpc2.h
index fd98c2d..5ea8196 100644
--- a/src/rpc2.h
+++ b/src/rpc2.h
@@ -14,6 +14,7 @@ struct aki_rpc_callback {
 
 struct aki_rpc_connection {
     struct aki_packet_stream stream;
+    bool disconnected;
     array(struct aki_rpc_callback) callbacks;
     struct aki_rpc *rpc;
 };
@@ -45,6 +46,7 @@ void aki_rpc_connect(struct aki_rpc *rpc, str *addr, s32 port);
 struct aki_packet *aki_rpc_get_packet(struct aki_rpc *rpc, s8 op);
 void aki_rpc_free(struct aki_rpc *rpc);
 
+// The order of commands per connection will be respected.
 void aki_rpc_connection_command(struct aki_rpc_connection *conn, struct aki_packet *packet,
     void (*callback)(void *, struct aki_packet *), void *userdata);
 void aki_rpc_conn_disconnect(struct aki_rpc_connection *conn);
diff --git a/src/socket/socket.h b/src/socket/socket.h
index bb42a3d..18e3040 100644
--- a/src/socket/socket.h
+++ b/src/socket/socket.h
@@ -1,21 +1,20 @@
 #pragma once
 
+#include <al/types.h>
+#include <al/str.h>
+
 #ifndef _WIN32
 #include <unistd.h>
 #include <sys/socket.h>
+#include <sys/un.h>
+#include <sys/eventfd.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
-#include <netinet/in.h>
 #include <poll.h>
-#include <sys/un.h>
-#include <sys/eventfd.h>
 #else
 #include "../winwrap.h"
 #endif
 
-#include <al/types.h>
-#include <al/str.h>
-
 enum {
     AKI_SOCKET_TCP = 0,
     AKI_SOCKET_UDP,
@@ -26,10 +25,11 @@ struct aki_socket {
     u8 type;
 #ifndef _WIN32
     s32 fd;
-    struct sockaddr_in addr_in;
+    struct addrinfo *addrinfo;
     struct sockaddr_un addr_un;
 #else
     SOCKET fd;
+    s32 internal_fd;
     SOCKADDR_IN addr_in;
 #endif
 };
@@ -62,9 +62,11 @@ void aki_socket_set_nodelay(struct aki_socket *sock, s32 nodelay);
 void aki_socket_set_send_buf(struct aki_socket *sock, u32 sndbuf);
 u32 aki_socket_get_send_buf(struct aki_socket *sock);
 
-bool aki_socket_listen(struct aki_socket *sock, str *addr, s32 port);
+bool aki_socket_set(struct aki_socket *sock, str *addr, u16 port);
+bool aki_socket_bind(struct aki_socket *sock, str *addr, u16 port);
+bool aki_socket_listen(struct aki_socket *sock);
 bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *c);
-bool aki_socket_connect(struct aki_socket *sock, str *addr, s32 port);
+bool aki_socket_connect(struct aki_socket *sock, str *addr, u16 port);
 
 s32 aki_socket_get_fd(struct aki_socket *sock);
 
diff --git a/src/socket/socket_internal.h b/src/socket/socket_internal.h
new file mode 100644
index 0000000..33f6a7b
--- /dev/null
+++ b/src/socket/socket_internal.h
@@ -0,0 +1,11 @@
+#pragma once
+
+#include <al/str.h>
+
+static str *ipany = al_str_c("0.0.0.0");
+static inline str *aki_socket_display_addr(str *addr)
+{
+    return addr ? addr : ipany;
+}
+
+#define aki_socket_printf_addr(addr) AL_STR_PRINTF(aki_socket_display_addr(addr))
diff --git a/src/socket/socket_linux.c b/src/socket/socket_linux.c
index adab91b..753f6b9 100644
--- a/src/socket/socket_linux.c
+++ b/src/socket/socket_linux.c
@@ -1,5 +1,4 @@
 #include <al/log.h>
-#include <unistd.h>
 #include <sys/fcntl.h>
 #include <netinet/tcp.h>
 #include <netdb.h>
@@ -7,6 +6,7 @@
 #include "../util/error.h"
 
 #include "socket.h"
+#include "socket_internal.h"
 
 void aki_fd_set_blocking(s32 fd, bool blocking)
 {
@@ -31,7 +31,7 @@ bool aki_socket_init(struct aki_socket *sock)
 {
     switch (sock->type) {
     case AKI_SOCKET_TCP:
-        sock->fd = socket(AF_INET, SOCK_STREAM, 0);
+        sock->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
         break;
     case AKI_SOCKET_UDP:
         sock->fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
@@ -39,6 +39,8 @@ bool aki_socket_init(struct aki_socket *sock)
     case AKI_SOCKET_UNIX:
         sock->fd = socket(AF_UNIX, SOCK_STREAM, 0);
         break;
+    default:
+        al_assert_and_return(false);
     }
 
     if (sock->fd < 0) {
@@ -46,6 +48,8 @@ bool aki_socket_init(struct aki_socket *sock)
         return false;
     }
 
+    sock->addrinfo = NULL;
+
     return true;
 }
 
@@ -56,8 +60,7 @@ void aki_socket_set_blocking(struct aki_socket *sock, bool blocking)
 
 void aki_socket_set_nodelay(struct aki_socket *sock, s32 nodelay)
 {
-    //al_assert(sock->type == AKI_SOCKET_TCP);
-    if (sock->type != AKI_SOCKET_TCP) return; // TODO: 
+    al_assert(sock->type == AKI_SOCKET_TCP);
     setsockopt(sock->fd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay));
 }
 
@@ -91,141 +94,155 @@ void aki_socket_set_recv_buf(struct aki_socket *sock, u32 rcvbuf)
     al_assert(aki_socket_get_recv_buf(sock) == receive_queue_size);
 }
 
-#define al_log_socket_err(func) \
-    al_log_error("socket", #func"(%.*s:%d) failed: %s (%d).", \
-        AL_STR_PRINTF(addr), port, aki_strerror(errno), errno);
+static bool parse_address(struct aki_socket *sock, str *addr, u16 port)
+{
+    struct addrinfo hints = { 0 };
+    hints.ai_family = AF_INET;
+    switch (sock->type) {
+    case AKI_SOCKET_TCP:
+        hints.ai_socktype = SOCK_STREAM;
+        hints.ai_protocol = IPPROTO_TCP;
+        break;
+    case AKI_SOCKET_UDP:
+        hints.ai_socktype = SOCK_DGRAM;
+        hints.ai_protocol = IPPROTO_UDP;
+        break;
+    }
+
+    char *c_str = addr ? al_str_to_c_str(addr) : NULL;
+    if (!c_str) hints.ai_flags = AI_PASSIVE;
+    char port_str[6];
+    al_sprintf(port_str, "%hu", port);
+    s32 status = getaddrinfo(c_str, port_str, &hints, &sock->addrinfo);
+    al_free(c_str);
+    if (status != 0) {
+        al_log_error("socket", "Failed to parse address (%s).", gai_strerror(status));
+        return false;
+    }
+
+    return true;
+}
+
+bool aki_socket_set(struct aki_socket *sock, str *addr, u16 port)
+{
+    al_assert(sock->type == AKI_SOCKET_UDP);
+    return parse_address(sock, addr, port);
+}
 
-bool aki_socket_listen(struct aki_socket *sock, str *addr, s32 port)
+bool aki_socket_bind(struct aki_socket *sock, str *addr, u16 port)
 {
     struct sockaddr *saddr = NULL;
     socklen_t addrlen = 0;
 
     switch (sock->type) {
-    case AKI_SOCKET_TCP: {
-        sock->addr_in.sin_family = AF_INET;
-        sock->addr_in.sin_addr.s_addr = htonl(INADDR_ANY);
-        sock->addr_in.sin_port = htons(port);
-        saddr = (struct sockaddr *)&sock->addr_in;
-        addrlen = sizeof(sock->addr_in);
+    case AKI_SOCKET_TCP:
+    case AKI_SOCKET_UDP: {
+        if (!parse_address(sock, addr, port)) return false;
+        saddr = sock->addrinfo->ai_addr;
+        addrlen = sock->addrinfo->ai_addrlen;
         break;
     }
     case AKI_SOCKET_UNIX: {
+        al_assert(addr);
         sock->addr_un.sun_family = AF_UNIX;
         char *c_str = al_str_to_c_str(addr);
         al_memcpy(sock->addr_un.sun_path, c_str, al_strlen(c_str) + 1);
+        unlink(c_str);
         al_free(c_str);
         saddr = (struct sockaddr *)&sock->addr_un;
         addrlen = sizeof(sock->addr_un);
         break;
     }
-    case AKI_SOCKET_UDP:
-        al_assert(false);
+    default:
+        al_assert_and_return(false);
     }
 
     if (bind(sock->fd, saddr, addrlen) < 0) {
-        al_log_socket_err(bind)
+        al_log_error("socket", "bind(%.*s:%hu) failed: %s (%d).",
+            aki_socket_printf_addr(addr), port, aki_strerror(errno), errno);
         return false;
     }
 
+    al_log_info("socket", "Socket bound to %.*s:%hu.", aki_socket_printf_addr(addr), port);
+
+    return true;
+}
+
+bool aki_socket_listen(struct aki_socket *sock)
+{
+    al_assert(sock->type == AKI_SOCKET_TCP || sock->type == AKI_SOCKET_UNIX);
+
     if (listen(sock->fd, SOMAXCONN) < 0) {
-        al_log_socket_err(listen)
+        al_log_error("socket", "listen() failed: %s (%d).", aki_strerror(errno), errno);
         return false;
     }
 
-    al_log_info("socket", "Listening on %.*s:%d.", AL_STR_PRINTF(addr), port);
+    al_log_info("socket", "Listening.");
 
     return true;
 }
 
 bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *cl)
 {
-    cl->type = sock->type;
+    al_assert(sock->type == AKI_SOCKET_TCP || sock->type == AKI_SOCKET_UNIX);
 
+    cl->type = sock->type;
     struct sockaddr_storage addr;
-    socklen_t len = sizeof(addr);
-    if ((cl->fd = accept(sock->fd, (struct sockaddr *)&addr, &len)) == -1) {
+    socklen_t addrlen = sizeof(addr);
+    if ((cl->fd = accept(sock->fd, (struct sockaddr *)&addr, &addrlen)) == -1) {
         al_log_error("socket", "accept() failed: %s (%d).", aki_strerror(errno), errno);
         return false;
     }
 
+    cl->addrinfo = NULL;
+
     switch (cl->type) {
     case AKI_SOCKET_TCP: {
-        s32 port = 0;
-        char ipstr[INET6_ADDRSTRLEN] = { 0 };
+        u16 port = 0;
+        char addr_str[INET6_ADDRSTRLEN] = { 0 };
         if (addr.ss_family == AF_INET) {
             struct sockaddr_in *saddr = (struct sockaddr_in *)&addr;
             port = ntohs(saddr->sin_port);
-            inet_ntop(AF_INET, &saddr->sin_addr, ipstr, sizeof(ipstr));
+            inet_ntop(AF_INET, &saddr->sin_addr, addr_str, sizeof(addr_str));
         } else if (addr.ss_family == AF_INET6) {
             struct sockaddr_in6 *saddr = (struct sockaddr_in6 *)&addr;
             port = ntohs(saddr->sin6_port);
-            inet_ntop(AF_INET6, &saddr->sin6_addr, ipstr, sizeof(ipstr));
+            inet_ntop(AF_INET6, &saddr->sin6_addr, addr_str, sizeof(addr_str));
         }
-        al_log_info("socket", "Connection from %s:%d.", ipstr, port);
+        al_log_info("socket", "Connection from %s:%hu.", addr_str, port);
         break;
     }
     case AKI_SOCKET_UNIX:
         al_log_info("socket", "Connection on UNIX socket.");
         break;
-    case AKI_SOCKET_UDP:
-        al_assert(false);
     }
 
     return true;
 }
 
-bool aki_socket_connect(struct aki_socket *sock, str *addr, s32 port)
+bool aki_socket_connect(struct aki_socket *sock, str *addr, u16 port)
 {
+    s32 ret;
     switch (sock->type) {
     case AKI_SOCKET_TCP:
-    case AKI_SOCKET_UDP: {
-        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;
-        }
-
-        sock->addr_in.sin_family = AF_INET;
-        sock->addr_in.sin_addr.s_addr = ((struct in_addr *)hptr->h_addr_list[0])->s_addr;
-        sock->addr_in.sin_port = htons(port);
-
-        break;
-    }
-    default:
-        break;
-    }
-
-    switch (sock->type) {
-    case AKI_SOCKET_TCP:
-    case AKI_SOCKET_UDP: {
-        s32 ret = connect(sock->fd, (struct sockaddr *)&sock->addr_in, sizeof(sock->addr_in));
-        if (ret != 0 && errno != EINPROGRESS) { 
-            al_log_socket_err(connect)
-            return false;
-        }
+    case AKI_SOCKET_UDP:
+        if (!parse_address(sock, addr, port)) return false;
+        ret = connect(sock->fd, sock->addrinfo->ai_addr, sock->addrinfo->ai_addrlen);
         break;
-    }
-    case AKI_SOCKET_UNIX: {
+    case AKI_SOCKET_UNIX:
         sock->addr_un.sun_family = AF_UNIX;
         al_memcpy(sock->addr_un.sun_path, addr->data, addr->len);
         sock->addr_un.sun_path[addr->len] = '\0';
-        if (connect(sock->fd, (struct sockaddr *)&sock->addr_un, sizeof(sock->addr_un)) < 0) {
-            al_log_socket_err(connect)
-            return false;
-        }
+        ret = connect(sock->fd, (struct sockaddr *)&sock->addr_un, sizeof(sock->addr_un));
         break;
-    }
     default:
-        al_assert(false);
-        break;
+        al_assert_and_return(false);
+    }
+
+    if (ret != 0 && errno != EINPROGRESS) {
+        al_log_error("socket", "connect(%.*s:%hu) failed: %s (%d).",
+            aki_socket_printf_addr(addr), port, aki_strerror(errno), errno);
+        return false;
     }
 
     return true;
@@ -243,13 +260,12 @@ ssize_t aki_socket_write(struct aki_socket *sock, void *buf, size_t size)
 
 ssize_t aki_socket_sendto(struct aki_socket *sock, void *buf, size_t size)
 {
-    return sendto(sock->fd, buf, size, 0, (const struct sockaddr *)&sock->addr_in, sizeof(sock->addr_in));
+    return sendto(sock->fd, buf, size, 0, sock->addrinfo->ai_addr, sock->addrinfo->ai_addrlen);
 }
 
 ssize_t aki_socket_recvfrom(struct aki_socket *sock, void *buf, size_t size)
 {
-    socklen_t addrlen = sizeof(sock->addr_in);
-    return recvfrom(sock->fd, buf, size, 0, (struct sockaddr *)&sock->addr_in, &addrlen);
+    return recvfrom(sock->fd, buf, size, 0, sock->addrinfo->ai_addr, &sock->addrinfo->ai_addrlen);
 }
 
 s32 aki_socket_get_fd(struct aki_socket *sock)
@@ -265,4 +281,8 @@ void aki_socket_shutdown(struct aki_socket *sock)
 void aki_socket_close(struct aki_socket *sock)
 {
     close(sock->fd);
+    if (sock->addrinfo) {
+        freeaddrinfo(sock->addrinfo);
+        sock->addrinfo = NULL;
+    }
 }
diff --git a/src/socket/socket_windows.c b/src/socket/socket_windows.c
index a7192d8..013a5bd 100644
--- a/src/socket/socket_windows.c
+++ b/src/socket/socket_windows.c
@@ -1,116 +1,142 @@
 #include <al/log.h>
 
 #include "socket.h"
+#include "socket_internal.h"
 
-bool aki_socket_init(struct aki_socket *s)
+bool aki_socket_init(struct aki_socket *sock)
 {
-    switch (s->type) {
+    al_assert(sock->type == AKI_SOCKET_TCP);
+
+    switch (sock->type) {
     case AKI_SOCKET_TCP:
-        s->sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+        sock->fd = 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) {
+    if (sock->fd == INVALID_SOCKET) {
         al_log_error("socket", "socket() failed: %d.", WSAGetLastError());
         return false;
     }
 
+    sock->internal_fd = _open_osfhandle(sock->fd, 0);
+
     return true;
 }
 
-void aki_socket_set_blocking(struct aki_socket *s, bool blocking)
+void aki_socket_set_blocking(struct aki_socket *sock, bool blocking)
 {
     u_long mode = blocking ? 0 : 1;
-    ioctlsocket(s->sock, FIONBIO, &mode);
+    ioctlsocket(sock->fd, FIONBIO, &mode);
 }
 
-void aki_socket_set_nodelay(struct aki_socket *s, s32 nodelay)
+void aki_socket_set_nodelay(struct aki_socket *sock, s32 nodelay)
 {
-    (void)s;
+    (void)sock;
     (void)nodelay;
 }
 
-#define al_log_socket_err(func) \
-    al_log_error("socket", #func"(%.*s:%d) failed: %d.", \
-        AL_STR_PRINTF(addr), port, WSAGetLastError());
+bool aki_socket_set(struct aki_socket *sock, str *addr, u16 port)
+{
+    (void)sock;
+    (void)addr;
+    (void)port;
+    return false;
+}
 
-bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port)
+bool aki_socket_bind(struct aki_socket *sock, str *addr, u16 port)
 {
-    s->addr_in.sin_family = AF_INET;
-    s->addr_in.sin_addr.s_addr = INADDR_ANY;
-    s->addr_in.sin_port = aki_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_socket_err(bind);
+    sock->addr_in.sin_family = AF_INET;
+    sock->addr_in.sin_addr.s_addr = INADDR_ANY;
+    sock->addr_in.sin_port = aki_htons(port);
+    al_memset(sock->addr_in.sin_zero, '\0', sizeof(sock->addr_in.sin_zero));
+    if (bind(sock->fd, (SOCKADDR *)&sock->addr_in, sizeof(sock->addr_in)) == SOCKET_ERROR) {
+        al_log_error("socket", "bind(%.*s:%hu) failed: %d.",
+            aki_socket_printf_addr(addr), port, WSAGetLastError());
         return false;
     }
-    if (listen(s->sock, SOMAXCONN) == SOCKET_ERROR) {
-        al_log_socket_err(listen);
+
+    al_log_info("socket", "Socket bound to %.*s:%hu.", aki_socket_printf_addr(addr), port);
+
+    return true;
+}
+
+bool aki_socket_listen(struct aki_socket *sock)
+{
+    if (listen(sock->fd, SOMAXCONN) == SOCKET_ERROR) {
+        al_log_error("socket", "listen() failed: %d.", WSAGetLastError());
         return false;
     }
+
+    al_log_info("socket", "Listening.");
+
     return true;
 }
 
-bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c)
+bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *cl)
 {
-    c->type = s->type;
-    if ((c->sock = accept(s->sock, NULL, NULL)) == INVALID_SOCKET) {
+    cl->type = sock->type;
+    if ((cl->fd = accept(sock->fd, NULL, NULL)) == INVALID_SOCKET) {
+        al_log_error("socket", "accept() failed: %d.", WSAGetLastError());
         return false;
     }
+    cl->internal_fd = _open_osfhandle(cl->fd, 0);
     return true;
 }
 
-bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
+bool aki_socket_connect(struct aki_socket *sock, str *addr, u16 port)
 {
-    s->addr_in.sin_family = AF_INET;
+    sock->addr_in.sin_family = AF_INET;
     char *c_str = al_str_to_c_str(addr);
-    s->addr_in.sin_addr.s_addr = inet_addr(c_str);
+    sock->addr_in.sin_addr.s_addr = inet_addr(c_str);
     al_free(c_str);
-    s->addr_in.sin_port = aki_htons(port);
-    al_memset(s->addr_in.sin_zero, '\0', 8);
-    s32 ret = connect(s->sock, (SOCKADDR *)&s->addr_in, sizeof(s->addr_in));
+    sock->addr_in.sin_port = aki_htons(port);
+    al_memset(sock->addr_in.sin_zero, '\0', sizeof(sock->addr_in.sin_zero));
+    s32 ret = connect(sock->fd, (SOCKADDR *)&sock->addr_in, sizeof(sock->addr_in));
     if (ret == SOCKET_ERROR && WSAGetLastError() != WSAEWOULDBLOCK) {
-        al_log_socket_err(connect);
+        al_log_error("socket", "connect() failed: %d.", WSAGetLastError());
         return false;
     }
     return true;
 }
 
-s32 aki_socket_get_fd(struct aki_socket *s)
+s32 aki_socket_get_fd(struct aki_socket *sock)
+{
+    return sock->internal_fd;
+}
+
+ssize_t aki_socket_read(struct aki_socket *sock, void *buf, size_t size)
 {
-    return _open_osfhandle(s->sock, 0);
+    return recv(sock->fd, buf, (s32)size, 0);
 }
 
-ssize_t aki_socket_read(struct aki_socket *s, void *buf, size_t size)
+ssize_t aki_socket_write(struct aki_socket *sock, void *buf, size_t size)
 {
-    return recv(s->sock, buf, (s32)size, 0);
+    return send(sock->fd, buf, (s32)size, 0);
 }
 
-ssize_t aki_socket_write(struct aki_socket *s, void *buf, size_t size)
+ssize_t aki_socket_sendto(struct aki_socket *sock, void *buf, size_t size)
 {
-    return send(s->sock, buf, (s32)size, 0);
+    (void)sock;
+    (void)buf;
+    (void)size;
+    return 0;
 }
 
-ssize_t aki_socket_sendto(struct aki_socket *s, void *buf, size_t size)
+ssize_t aki_socket_recvfrom(struct aki_socket *sock, void *buf, size_t size)
 {
-    (void)s;
+    (void)sock;
     (void)buf;
     (void)size;
     return 0;
 }
 
-void aki_socket_shutdown(struct aki_socket *s)
+void aki_socket_shutdown(struct aki_socket *sock)
 {
-    if (shutdown(s->sock, SD_SEND) == SOCKET_ERROR) {}
+    if (shutdown(sock->fd, SD_BOTH) == SOCKET_ERROR) {}
 }
 
-void aki_socket_close(struct aki_socket *s)
+void aki_socket_close(struct aki_socket *sock)
 {
-    closesocket(s->sock);
+    _close(sock->internal_fd);
+    //closesocket(sock->fd);
 }
diff --git a/src/timer.c b/src/timer.c
index 52cdcca..0057fd6 100644
--- a/src/timer.c
+++ b/src/timer.c
@@ -5,7 +5,7 @@ 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);
+    if (!timer->disabled) timer->callback(timer->userdata, timer);
 }
 
 void aki_timer_init(struct aki_timer *timer, struct aki_event_loop *loop,
diff --git a/src/timer.h b/src/timer.h
index 73080a4..364c82f 100644
--- a/src/timer.h
+++ b/src/timer.h
@@ -2,7 +2,6 @@
 
 #include "util/thread/thread.h"
 
-#include "evwrap.h"
 #include "loop.h"
 
 struct aki_timer {
diff --git a/src/util/buffer.c b/src/util/buffer.c
index 4f184b9..32a8230 100644
--- a/src/util/buffer.c
+++ b/src/util/buffer.c
@@ -1,13 +1,14 @@
 #include "buffer.h"
 
-#define INITIAL_SIZE 128
+#define BUFFER_INITIAL_SIZE 128
 
 void aki_buffer_ensure_space(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);
+    if (size > buf->alloc) {
+        if (size < BUFFER_INITIAL_SIZE) buf->alloc = BUFFER_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)
diff --git a/src/util/packet.c b/src/util/packet.c
index 373a3a3..ae36fcd 100644
--- a/src/util/packet.c
+++ b/src/util/packet.c
@@ -60,7 +60,8 @@ void aki_packet_write_str(struct aki_packet *packet, str *s)
 
 void aki_packet_write_buffer(struct aki_packet *packet, struct aki_buffer *buf)
 {
-    AKI_PACKET_WRITE_TYPE(packet, size_t, buf->size);
+    u64 size = (u64)buf->size;
+    AKI_PACKET_WRITE_TYPE(packet, u64, size);
     AKI_PACKET_WRITE_DATA(packet, aki_buffer_get_ptr(buf, 0), buf->size);
 }
 
@@ -92,12 +93,14 @@ void aki_packet_read_str(struct aki_packet *packet, str *s)
 
 void aki_packet_read_buffer(struct aki_packet *packet, struct aki_buffer *buf)
 {
-    AKI_PACKET_READ_TYPE(packet, size_t, buf->size);
+    u64 size;
+    AKI_PACKET_READ_TYPE(packet, u64, size);
+    al_assert((sizeof(size_t) > 4 || size <= INT32_MAX));
+    buf->size = (size_t)size;
     AKI_PACKET_READ_DATA(packet, buf->size, buf->data);
     buf->alloc = 0;
 }
 
-#ifdef AL_HAVE_WIDE_STRING
 void aki_packet_write_wstr(struct aki_packet *packet, wstr *w)
 {
     AKI_PACKET_WRITE_TYPE(packet, u32, w->len);
@@ -110,7 +113,6 @@ void aki_packet_read_wstr(struct aki_packet *packet, wstr *w)
     AKI_PACKET_READ_DATA(packet, w->len * sizeof(wchar_t), w->data);
     w->alloc = 0;
 }
-#endif
 
 void aki_packet_free(struct aki_packet *packet)
 {
diff --git a/src/util/packet.h b/src/util/packet.h
index cacd82c..4f5b28c 100644
--- a/src/util/packet.h
+++ b/src/util/packet.h
@@ -13,7 +13,6 @@ struct aki_packet {
     struct aki_buffer buffer;
     u32 rindex;
     u32 windex;
-    void *userdata;
 };
 
 #define AKI_PACKET_GET_ID(p) \
@@ -68,9 +67,7 @@ f64 aki_packet_read_f64(struct aki_packet *packet);
 void aki_packet_read_str(struct aki_packet *packet, str *s);
 void aki_packet_read_buffer(struct aki_packet *packet, struct aki_buffer *buf);
 
-#ifdef AL_HAVE_WIDE_STRING
 void aki_packet_write_wstr(struct aki_packet *packet, wstr *w);
 void aki_packet_read_wstr(struct aki_packet *packet, wstr *w);
-#endif
 
 void aki_packet_free(struct aki_packet *packet);
diff --git a/src/util/thread/thread.h b/src/util/thread/thread.h
index 8860d2d..6a04647 100644
--- a/src/util/thread/thread.h
+++ b/src/util/thread/thread.h
@@ -47,7 +47,7 @@ typedef unsigned long aki_thread_result;
 
 typedef aki_thread_result (AKI_THREADCALL * aki_thread_func)(void *);
 
-void aki_thread_sleep(aki_os_tstamp ts);
+void aki_thread_sleep(aki_os_tstamp delay);
 
 void aki_thread_create(struct aki_thread *thread, aki_thread_func func, void *userdata);
 void aki_thread_join(struct aki_thread *thread);
diff --git a/src/util/thread/thread_linux.c b/src/util/thread/thread_linux.c
index e1d694c..6df84e9 100644
--- a/src/util/thread/thread_linux.c
+++ b/src/util/thread/thread_linux.c
@@ -4,11 +4,11 @@
 
 #include "thread.h"
 
-void aki_thread_sleep(aki_os_tstamp ts)
+void aki_thread_sleep(aki_os_tstamp delay)
 {
     struct timespec tv;
-    tv.tv_sec = (s64)ts;
-    tv.tv_nsec = (s64)((ts - tv.tv_sec) * 1e9);
+    tv.tv_sec = (s64)delay;
+    tv.tv_nsec = (s64)((delay - tv.tv_sec) * 1e9);
     while (nanosleep(&tv, &tv) == -1 && errno == EINTR) {}
 }
 
diff --git a/src/util/thread/thread_windows.c b/src/util/thread/thread_windows.c
index 9ea9ead..a58e1b1 100644
--- a/src/util/thread/thread_windows.c
+++ b/src/util/thread/thread_windows.c
@@ -2,11 +2,11 @@
 
 NTSTATUS(__stdcall *NtDelayExecution)(BOOL Alertable, PLARGE_INTEGER DelayInterval);
 
-void aki_thread_sleep(aki_os_tstamp ts)
+void aki_thread_sleep(aki_os_tstamp delay)
 {
-    LARGE_INTEGER intr;
-    intr.QuadPart = -1 * (s64)(ts * 1e7);
-    NtDelayExecution(false, &intr);
+    LARGE_INTEGER li;
+    li.QuadPart = -(s64)(delay * 1e7L);
+    NtDelayExecution(false, &li);
 }
 
 void aki_thread_create(struct aki_thread *thread, aki_thread_func func, void *userdata)
diff --git a/src/util/timer/timer_windows.c b/src/util/timer/timer_windows.c
index 37a8fd8..007efed 100644
--- a/src/util/timer/timer_windows.c
+++ b/src/util/timer/timer_windows.c
@@ -8,7 +8,7 @@ f64 aki_get_tick(void)
 {
     LARGE_INTEGER result;
     QueryPerformanceCounter(&result);
-    return (result.QuadPart/(f64)performance_frequency.QuadPart);
+    return result.QuadPart / (f64)performance_frequency.QuadPart;
 }
 
 static SYSTEMTIME unix_epoch = {
@@ -35,5 +35,5 @@ u64 aki_get_timestamp(void)
         .LowPart = result1.dwLowDateTime,
         .HighPart = result1.dwHighDateTime
     };
-    return (li1.QuadPart - li0.QuadPart) / 10;
+    return (li1.QuadPart - li0.QuadPart) / 10L;
 }
diff --git a/src/winwrap.c b/src/winwrap.c
index dcfdef9..17fb160 100644
--- a/src/winwrap.c
+++ b/src/winwrap.c
@@ -3,17 +3,25 @@
 void aki_windows_init(void)
 {
     // https://learn.microsoft.com/en-us/windows/win32/api/objbase/ne-objbase-coinit
-    CoInitialize(NULL);
+    CoInitializeEx(NULL, COINIT_MULTITHREADED | COINIT_SPEED_OVER_MEMORY);
 
-    WSADATA data;
-    WSAStartup(MAKEWORD(2,2), &data);
+    WSADATA wsa_data;
+    WSAStartup(MAKEWORD(2, 2), &wsa_data);
 
-    ULONG actual_resolution;
+#if defined(__clang__) || defined(__GNUC__)
+_Pragma("GCC diagnostic push") \
+_Pragma("GCC diagnostic ignored \"-Wcast-function-type\"")
+#endif
     NTSTATUS(__stdcall *ZwSetTimerResolution)(IN ULONG RequestedResolution, IN BOOLEAN Set, OUT PULONG ActualResolution);
     ZwSetTimerResolution = (NTSTATUS(__stdcall *)(ULONG, BOOLEAN, PULONG))GetProcAddress(GetModuleHandleW(L"ntdll"), "ZwSetTimerResolution");
+
+    ULONG actual_resolution;
     ZwSetTimerResolution(1, TRUE, &actual_resolution);
 
     NtDelayExecution = (NTSTATUS(__stdcall *)(BOOL, PLARGE_INTEGER))GetProcAddress(GetModuleHandleW(L"ntdll"), "NtDelayExecution");
+#if defined(__clang__) || defined(__GNUC__)
+_Pragma("GCC diagnostic pop")
+#endif
 
     QueryPerformanceFrequency(&performance_frequency);
 }
diff --git a/src/winwrap.h b/src/winwrap.h
index 2c71dc0..bd7ad90 100644
--- a/src/winwrap.h
+++ b/src/winwrap.h
@@ -1,7 +1,5 @@
 #pragma once
 
-#include <al/lib.h>
-
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #include <io.h>