Android build, packet stream fixes

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

22 files changed, 138 insertions, 77 deletions

diff --git a/meson.build b/meson.build
index e0af844..c779c33 100644
--- a/meson.build
+++ b/meson.build
@@ -114,6 +114,10 @@ if not is_windows
   if pthreads.found()
     akiyo_deps += [pthreads]
   endif
+  have_pthread_cancel = compiler.has_function('pthread_cancel', prefix: '#include <pthread.h>')
+  if have_pthread_cancel
+    akiyo_args += ['-DAKIYO_HAS_THREAD_CANCEL']
+  endif
   akiyo_args += ['-D_GNU_SOURCE']
   akiyo_has_mmap = true
 else
diff --git a/src/common.c b/src/common.c
index 3abba13..ec2c2a9 100644
--- a/src/common.c
+++ b/src/common.c
@@ -15,7 +15,9 @@ bool aki_common_init(void)
 {
     if (!setlocale(LC_ALL, "")) return false;
 
+#ifdef AL_MEMORY_TRACKING
     al_malloc_init();
+#endif
     al_rand_init();
 #ifdef _WIN32
     aki_windows_init();
@@ -41,5 +43,7 @@ void aki_common_close(void)
 #ifdef _WIN32
     aki_windows_close();
 #endif
+#ifdef AL_MEMORY_TRACKING
     al_malloc_close();
+#endif
 }
diff --git a/src/curl/curl.c b/src/curl/curl.c
index 5560487..b96d46f 100644
--- a/src/curl/curl.c
+++ b/src/curl/curl.c
@@ -2,6 +2,8 @@
 
 #include "curl.h"
 
+// TODO: https://curl.se/libcurl/c/externalsocket.html
+
 static void curl_socket_action_callback(struct ev_loop *loop, ev_io *w, s32 revents)
 {
     (void)loop;
@@ -36,10 +38,8 @@ static s32 sock_callback(CURL *e, curl_socket_t s, s32 what, void *cbp, void *so
         if (!sockp) {
             curl->sock = s;
             curl_multi_assign(curl->multi_handle, curl->sock, curl);
-            set_sock(curl, what);
-        } else {
-            set_sock(curl, what);
         }
+        set_sock(curl, what);
     }
     return 0;
 }
@@ -88,6 +88,7 @@ bool aki_curl_init(struct aki_curl *curl)
     curl_multi_setopt(curl->multi_handle, CURLMOPT_SOCKETDATA, curl);
     curl_multi_setopt(curl->multi_handle, CURLMOPT_TIMERFUNCTION, timer_callback);
     curl_multi_setopt(curl->multi_handle, CURLMOPT_TIMERDATA, curl);
+    curl->added = false;
     curl->timer_started = false;
     curl->loop = NULL;
     curl->event.data = curl;
@@ -107,6 +108,7 @@ void aki_curl_set_url(struct aki_curl *curl, str *url)
 
 bool aki_curl_add_handle(struct aki_curl *curl)
 {
+    al_assert(!curl->added);
     // 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(curl->multi_handle, curl->handle);
@@ -114,21 +116,27 @@ bool aki_curl_add_handle(struct aki_curl *curl)
         al_log_error("curl", "curl_multi_add_handle() failed (%s).", curl_multi_strerror(mc));
         return false;
     }
+    curl->added = true;
     return true;
 }
 
 bool aki_curl_remove_handle(struct aki_curl *curl)
 {
+    al_assert(curl->added);
     CURLMcode mc = curl_multi_remove_handle(curl->multi_handle, curl->handle);
     if (mc != CURLM_OK) {
         al_log_error("curl", "curl_multi_remove_handle() failed (%s).", curl_multi_strerror(mc));
         return false;
     }
+    curl->added = false;
     return true;
 }
 
 void aki_curl_close(struct aki_curl *curl)
 {
+    if (curl->handle && curl->multi_handle && curl->added) {
+        aki_curl_remove_handle(curl);
+    }
     if (curl->handle) curl_easy_cleanup(curl->handle);
     if (curl->multi_handle) curl_multi_cleanup(curl->multi_handle);
 }
diff --git a/src/curl/curl.h b/src/curl/curl.h
index 5bb38c0..ce14f3c 100644
--- a/src/curl/curl.h
+++ b/src/curl/curl.h
@@ -12,6 +12,7 @@ struct aki_curl {
     struct aki_event_loop *loop;
     CURL *handle;
     CURLM *multi_handle;
+    bool added;
     curl_socket_t sock;
     void (*handle_events)(void *, struct aki_curl *);
     void *userdata;
diff --git a/src/ev_embed_compat.c b/src/ev_embed_compat.c
index 552e25b..2bf262a 100644
--- a/src/ev_embed_compat.c
+++ b/src/ev_embed_compat.c
@@ -24,6 +24,7 @@ _Pragma("GCC diagnostic ignored \"-Wreturn-type\"")
 _Pragma("GCC diagnostic ignored \"-Wdeprecated-declarations\"")
 _Pragma("GCC diagnostic ignored \"-Wdangling-else\"")
 _Pragma("GCC diagnostic ignored \"-Wtype-limits\"")
+_Pragma("GCC diagnostic ignored \"-Wunknown-pragmas\"")
 #elif _MSC_VER
 #pragma warning(push, 0)
 #endif
diff --git a/src/loop.c b/src/loop.c
index 54ffbcd..0e650ba 100644
--- a/src/loop.c
+++ b/src/loop.c
@@ -4,7 +4,8 @@
 
 bool aki_event_loop_init(struct aki_event_loop *loop)
 {
-    loop->ev = ev_loop_new(EVFLAG_AUTO);
+    loop->ev = ev_default_loop(EVFLAG_AUTO);
+    //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;
diff --git a/src/multiplex.c b/src/multiplex.c
index ea8ca9f..4f5330f 100644
--- a/src/multiplex.c
+++ b/src/multiplex.c
@@ -29,7 +29,7 @@ static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 reven
     aki_socket_close(&sock);
 }
 
-bool aki_multiplex_socket_listen(struct aki_multiplex_socket *multi, struct aki_event_loop *loop, str *addr, s32 port)
+bool aki_multiplex_socket_listen(struct aki_multiplex_socket *multi, struct aki_event_loop *loop, str *addr, u16 port)
 {
     if (!aki_socket_bind(&multi->sock, addr, port)) return false;
     if (!aki_socket_listen(&multi->sock)) return false;
diff --git a/src/multiplex.h b/src/multiplex.h
index cd20c29..68d24a4 100644
--- a/src/multiplex.h
+++ b/src/multiplex.h
@@ -16,5 +16,5 @@ struct aki_multiplex_socket {
 
 bool aki_multiplex_socket_init(struct aki_multiplex_socket *multi, u8 type,
     bool (*connection_callback)(void *, u8, struct aki_socket *), void *userdata);
-bool aki_multiplex_socket_listen(struct aki_multiplex_socket *multi, struct aki_event_loop *loop, str *addr, s32 port);
+bool aki_multiplex_socket_listen(struct aki_multiplex_socket *multi, struct aki_event_loop *loop, str *addr, u16 port);
 void aki_multiplex_socket_close(struct aki_multiplex_socket *multi);
diff --git a/src/packet_stream.c b/src/packet_stream.c
index 8b5c611..110984b 100644
--- a/src/packet_stream.c
+++ b/src/packet_stream.c
@@ -3,6 +3,7 @@
 enum {
     PACKET_STREAM_CONNECTING = 0,
     PACKET_STREAM_CONNECTED,
+    PACKET_STREAM_FLUSHING,
     PACKET_STREAM_DISCONNECTING,
     PACKET_STREAM_DISCONNECTED
 };
@@ -11,7 +12,7 @@ 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->out.active = false;
     stream->in.packet = aki_packet_create();
     stream->in.have_size = false;
     stream->in.index = 0;
@@ -47,18 +48,29 @@ void aki_packet_stream_set_multiplex(struct aki_packet_stream *stream, u8 id)
     stream->id = id;
 }
 
+static void start_write_internal(struct aki_packet_stream *stream)
+{
+    stream->out.active = true;
+    ev_io_start(stream->loop->ev, &stream->wevent);
+}
+
 static void stop_write_internal(struct aki_packet_stream *stream)
 {
-    stream->out.have_data = false;
+    stream->out.active = false;
     ev_io_stop(stream->loop->ev, &stream->wevent);
 }
 
+static void shutdown_internal(struct aki_packet_stream *stream)
+{
+    stream->connect = PACKET_STREAM_DISCONNECTING;
+    // This assumes select()/poll() will return after shutdown().
+    aki_socket_shutdown(&stream->sock);
+}
+
 static void stop_internal(struct aki_packet_stream *stream)
 {
+    stream->corked = false;
     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);
@@ -82,17 +94,20 @@ static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 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_get_size(stream->in.packet) : AKI_PACKET_HEADER_LENGTH;
     ssize_t ret = aki_socket_read(&stream->sock, ptr, size - stream->in.index);
     if (ret <= 0 || stream->connect == PACKET_STREAM_DISCONNECTING) {
         ev_io_stop(stream->loop->ev, &stream->revent);
+        if (stream->out.active) {
+            stop_write_internal(stream);
+        }
         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);
+        size = aki_packet_get_size(stream->in.packet);
         aki_buffer_ensure_space(&stream->in.packet->buffer, size);
     }
     if (stream->in.have_size && stream->in.index >= size) {
@@ -115,6 +130,7 @@ static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents)
         ssize_t ret = aki_socket_write(&stream->sock, &stream->id, sizeof(u8));
         if (ret < 0) {
             // It seems POLLOUT can be signaled even if any write() will error.
+            stop_write_internal(stream);
             stop_internal(stream);
             return;
         } else if (ret == 0) {
@@ -137,11 +153,18 @@ static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents)
             stream->out.index = 0;
         } else {
             stop_write_internal(stream);
+            if (stream->connect == PACKET_STREAM_FLUSHING) {
+                if (stream->corked) {
+                    stop_internal(stream);
+                } else {
+                    shutdown_internal(stream);
+                }
+            }
             return;
         }
     }
     u8 *ptr = aki_buffer_get_ptr(&stream->out.packet->buffer, stream->out.index);
-    u32 size = aki_packet_size(stream->out.packet);
+    u32 size = aki_packet_get_size(stream->out.packet);
     ssize_t ret = aki_socket_write(&stream->sock, ptr, size - stream->out.index);
     if (ret <= 0) {
         stop_write_internal(stream);
@@ -172,15 +195,10 @@ void aki_packet_stream_from_socket(struct aki_packet_stream *stream, struct aki_
     ev_io_start(stream->loop->ev, &stream->revent);
 }
 
-static void start_write_internal(struct aki_packet_stream *stream)
-{
-    stream->out.have_data = true;
-    ev_io_start(stream->loop->ev, &stream->wevent);
-}
-
-static void connect_internal(struct aki_packet_stream *stream, str *addr, s32 port)
+static void connect_internal(struct aki_packet_stream *stream, str *addr, u16 port)
 {
     if (!aki_socket_connect(&stream->sock, addr, port)) {
+        aki_socket_close(&stream->sock);
         stream->connection_closed_callback(stream->userdata, stream);
         return;
     }
@@ -194,13 +212,13 @@ static void connect_internal(struct aki_packet_stream *stream, str *addr, s32 po
     start_write_internal(stream);
 }
 
-void aki_packet_stream_connect(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, u16 port)
 {
     stream->loop = loop;
     connect_internal(stream, addr, port);
 }
 
-void aki_packet_stream_reconnect(struct aki_packet_stream *stream, str *addr, s32 port)
+void aki_packet_stream_reconnect(struct aki_packet_stream *stream, str *addr, u16 port)
 {
     al_assert(stream->connect == PACKET_STREAM_DISCONNECTED);
     if (!aki_socket_init(&stream->sock)) {
@@ -214,30 +232,46 @@ void aki_packet_stream_reconnect(struct aki_packet_stream *stream, str *addr, s3
 void aki_packet_stream_cork(struct aki_packet_stream *stream, bool cork)
 {
     if (stream->connect == PACKET_STREAM_CONNECTED) {
-        if (cork && stream->revent.active) {
+        if (cork && !stream->corked) {
             ev_io_stop(stream->loop->ev, &stream->revent);
-        } else if (!cork && !stream->revent.active) {
+        } else if (!cork && stream->corked) {
             ev_io_start(stream->loop->ev, &stream->revent);
         }
+        stream->corked = cork;
     }
-    // Allow stream->corked to be changed even if it has no immediate effect.
-    stream->corked = cork;
 }
 
 bool aki_packet_stream_send_packet(struct aki_packet_stream *stream, struct aki_packet *packet)
 {
-    if (stream->connect == PACKET_STREAM_DISCONNECTING) {
+    if (stream->connect == PACKET_STREAM_FLUSHING ||
+        stream->connect == PACKET_STREAM_DISCONNECTING) {
         return false;
     }
     al_assert(stream->connect == PACKET_STREAM_CONNECTED);
     aki_packet_write_size(packet);
     al_array_push(stream->out.queue, packet);
-    if (!stream->out.have_data) {
+    if (!stream->out.active) {
         start_write_internal(stream);
     }
     return true;
 }
 
+void aki_packet_stream_flush(struct aki_packet_stream *stream)
+{
+    al_assert(stream->connect == PACKET_STREAM_CONNECTED ||
+        stream->connect == PACKET_STREAM_CONNECTING);
+    // If connect = CONNECTING, we will wait utill connected then flush.
+    if (!stream->out.active) {
+        if (stream->corked) {
+            stop_internal(stream);
+        } else {
+            shutdown_internal(stream);
+        }
+    } else {
+        stream->connect = PACKET_STREAM_FLUSHING;
+    }
+}
+
 void aki_packet_stream_disconnect(struct aki_packet_stream *stream)
 {
     al_assert(stream->connect != PACKET_STREAM_DISCONNECTING);
@@ -246,20 +280,24 @@ void aki_packet_stream_disconnect(struct aki_packet_stream *stream)
     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(). This is why we have to check if
     // connect = DISCONNECTING in send_packet().
-    aki_socket_shutdown(&stream->sock);
-    if (stream->corked || stream->connect == PACKET_STREAM_CONNECTING) {
+    if (stream->connect == PACKET_STREAM_CONNECTING) {
+        // wevent is always active when connect = CONNECTING.
+        stop_write_internal(stream);
+        stop_internal(stream);
+    } else if (stream->corked) {
         stop_internal(stream);
+    } else {
+        shutdown_internal(stream);
     }
 }
 
 void aki_packet_stream_free(struct aki_packet_stream *stream)
 {
     // Loosely assert that stop_internal has run.
-    al_assert(!stream->out.have_data);
+    al_assert(!stream->out.active);
     al_assert(stream->in.packet);
     aki_packet_free(stream->in.packet);
     al_array_free(stream->out.queue);
diff --git a/src/packet_stream.h b/src/packet_stream.h
index 80d8a2f..180c507 100644
--- a/src/packet_stream.h
+++ b/src/packet_stream.h
@@ -23,7 +23,7 @@ struct aki_packet_stream {
     struct {
         array(struct aki_packet *) queue;
         struct aki_packet *packet;
-        bool have_data;
+        bool active;
         size_t index;
     } out;
     struct {
@@ -39,9 +39,10 @@ bool aki_packet_stream_init(struct aki_packet_stream *stream, u8 type,
 void aki_packet_stream_set_nodelay(struct aki_packet_stream *stream, s32 no_delay);
 void aki_packet_stream_set_multiplex(struct aki_packet_stream *stream, u8 id);
 void aki_packet_stream_from_socket(struct aki_packet_stream *stream, struct aki_event_loop *loop, struct aki_socket *sock);
-void aki_packet_stream_connect(struct aki_packet_stream *stream, struct aki_event_loop *loop, str *addr, s32 port);
-void aki_packet_stream_reconnect(struct aki_packet_stream *stream, str *addr, s32 port);
+void aki_packet_stream_connect(struct aki_packet_stream *stream, struct aki_event_loop *loop, str *addr, u16 port);
+void aki_packet_stream_reconnect(struct aki_packet_stream *stream, str *addr, u16 port);
 void aki_packet_stream_cork(struct aki_packet_stream *stream, bool cork);
 bool 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 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
index d208f0b..e086d89 100644
--- a/src/rpc.c
+++ b/src/rpc.c
@@ -23,7 +23,6 @@ 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
@@ -86,7 +85,6 @@ 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;
@@ -99,7 +97,6 @@ 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)) {
@@ -114,12 +111,18 @@ bool aki_rpc_prepare_client(struct aki_rpc *rpc, u8 type, u8 id)
     return true;
 }
 
-void aki_rpc_connect(struct aki_rpc *rpc, str *addr, s32 port)
+void aki_rpc_connect(struct aki_rpc *rpc, str *addr, u16 port)
 {
     al_assert(rpc->conn);
     aki_packet_stream_connect(&rpc->conn->stream, rpc->loop, addr, port);
 }
 
+void aki_rpc_reconnect(struct aki_rpc *rpc, str *addr, u16 port)
+{
+    al_assert(rpc->conn);
+    aki_packet_stream_reconnect(&rpc->conn->stream, addr, port);
+}
+
 struct aki_packet *aki_rpc_get_packet(struct aki_rpc *rpc, s8 op)
 {
     struct aki_packet *packet = aki_packet_create();
@@ -154,8 +157,12 @@ void aki_rpc_connection_command(struct aki_rpc_connection *conn, struct aki_pack
     aki_packet_stream_send_packet(&conn->stream, packet);
 }
 
+void aki_rpc_conn_flush(struct aki_rpc_connection *conn)
+{
+    aki_packet_stream_flush(&conn->stream);
+}
+
 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 5ea8196..d03a8a7 100644
--- a/src/rpc2.h
+++ b/src/rpc2.h
@@ -14,7 +14,6 @@ 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;
 };
@@ -42,11 +41,13 @@ bool aki_rpc_init(struct aki_rpc *rpc, struct aki_event_loop *loop,
 void aki_rpc_add_command(struct aki_rpc *rpc, struct aki_rpc_command *command);
 void aki_rpc_add_socket(struct aki_rpc *rpc, struct aki_socket *sock);
 bool aki_rpc_prepare_client(struct aki_rpc *rpc, u8 type, u8 id);
-void aki_rpc_connect(struct aki_rpc *rpc, str *addr, s32 port);
+void aki_rpc_connect(struct aki_rpc *rpc, str *addr, u16 port);
+void aki_rpc_reconnect(struct aki_rpc *rpc, str *addr, u16 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_flush(struct aki_rpc_connection *conn);
 void aki_rpc_conn_disconnect(struct aki_rpc_connection *conn);
diff --git a/src/signal.c b/src/signal.c
index 0bc15b8..ad9c59d 100644
--- a/src/signal.c
+++ b/src/signal.c
@@ -19,7 +19,7 @@ void aki_signal_init(struct aki_signal *signal, void (*callback)(void *), void *
 void aki_signal_start(struct aki_signal *signal, struct aki_event_loop *loop)
 {
     signal->loop = loop;
-    ev_async_start(loop->ev, &signal->signal);
+    ev_async_start(signal->loop->ev, &signal->signal);
 }
 
 void aki_signal_send(struct aki_signal *signal)
diff --git a/src/util/packet.c b/src/util/packet.c
index 3b49ced..2658d38 100644
--- a/src/util/packet.c
+++ b/src/util/packet.c
@@ -30,7 +30,7 @@ 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)
+u32 aki_packet_get_size(struct aki_packet *packet)
 {
     return *((u32 *)aki_buffer_get_ptr(&packet->buffer, 0));
 }
diff --git a/src/util/packet.h b/src/util/packet.h
index 83b75a3..72543a6 100644
--- a/src/util/packet.h
+++ b/src/util/packet.h
@@ -39,7 +39,7 @@ 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);
+u32 aki_packet_get_size(struct aki_packet *packet);
 
 void aki_packet_write_bool(struct aki_packet *packet, bool v);
 void aki_packet_write_s8(struct aki_packet *packet, s8 v);
diff --git a/src/util/thread/thread.h b/src/util/thread/thread.h
index 0c5042e..5975c84 100644
--- a/src/util/thread/thread.h
+++ b/src/util/thread/thread.h
@@ -8,16 +8,11 @@
 #include "../../winwrap.h"
 #endif
 
-#ifndef _WIN32
+#ifdef AKIYO_HAS_THREAD_CANCEL
 enum {
     AKI_THREAD_CANCEL_DEFERRED = PTHREAD_CANCEL_DEFERRED,
     AKI_THREAD_CANCEL_ASYNCHRONOUS = PTHREAD_CANCEL_ASYNCHRONOUS
 };
-#else
-enum {
-    AKI_THREAD_CANCEL_DEFERRED = 0,
-    AKI_THREAD_CANCEL_ASYNCHRONOUS
-};
 #endif
 
 struct aki_thread {
@@ -63,8 +58,10 @@ 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);
+#ifdef AKIYO_HAS_THREAD_CANCEL
 void aki_thread_cancel(struct aki_thread *thread);
 void aki_thread_setcanceltype(s32 type);
+#endif
 //void aki_thread_detach(struct aki_thread *thread);
 
 void aki_mutex_init(struct aki_mutex *mutex);
diff --git a/src/util/thread/thread_linux.c b/src/util/thread/thread_linux.c
index 04b0868..26c3e18 100644
--- a/src/util/thread/thread_linux.c
+++ b/src/util/thread/thread_linux.c
@@ -22,6 +22,7 @@ void aki_thread_join(struct aki_thread *thread)
     pthread_join(thread->thread, NULL);
 }
 
+#ifdef AKIYO_HAS_THREAD_CANCEL
 void aki_thread_cancel(struct aki_thread *thread)
 {
     pthread_cancel(thread->thread);
@@ -31,6 +32,7 @@ void aki_thread_setcanceltype(s32 type)
 {
     pthread_setcanceltype(type, NULL);
 }
+#endif
 
 /*
 void aki_thread_detach(struct aki_thread *thread)
diff --git a/src/util/thread/thread_windows.c b/src/util/thread/thread_windows.c
index b7bc0ec..c7fdd62 100644
--- a/src/util/thread/thread_windows.c
+++ b/src/util/thread/thread_windows.c
@@ -18,6 +18,7 @@ void aki_thread_join(struct aki_thread *thread)
     CloseHandle(thread->thread);
 }
 
+#ifdef AKIYO_HAS_THREAD_CANCEL
 void aki_thread_cancel(struct aki_thread *thread)
 {
     (void)thread;
@@ -27,6 +28,7 @@ void aki_thread_setcanceltype(s32 type)
 {
     (void)type;
 }
+#endif
 
 /*
 void aki_thread_detach(struct aki_thread *thread)
diff --git a/src/winwrap.c b/src/winwrap.c
index 17fb160..601b0f7 100644
--- a/src/winwrap.c
+++ b/src/winwrap.c
@@ -28,7 +28,8 @@ _Pragma("GCC diagnostic pop")
 
 void aki_windows_close(void)
 {
-    WSACleanup();
+    // Still crashes on windows.
+    //WSACleanup();
     CoUninitialize();
 }
 
diff --git a/subprojects/libev.wrap b/subprojects/libev.wrap
index 02b8df5..56b5299 100644
--- a/subprojects/libev.wrap
+++ b/subprojects/libev.wrap
@@ -6,7 +6,7 @@ source_filename = libev-4.33.tar.gz
 source_hash = 507eb7b8d1015fbec5b935f34ebed15bf346bed04a11ab82b8eee848c4205aea
 
 patch_directory = libev
-diff_files = libev/libev_aki_sleep_no_monotonic.diff, libev/no_extern_on_init.diff
+diff_files = libev/libev_aki_thread_sleep.diff, libev/no_extern_on_init.diff, libev/no_monotonic.diff
 
 [provide]
 dependency_names = libev
diff --git a/subprojects/packagefiles/libev/libev_aki_sleep_no_monotonic.diff b/subprojects/packagefiles/libev/libev_aki_thread_sleep.diff
index 2080116..59d197c 100644
--- a/subprojects/packagefiles/libev/libev_aki_sleep_no_monotonic.diff
+++ b/subprojects/packagefiles/libev/libev_aki_thread_sleep.diff
@@ -9,15 +9,6 @@
  #ifdef EV_H
  # include EV_H
  #else
-@@ -437,7 +439,7 @@
- /* on linux, we can use a (slow) syscall to avoid a dependency on pthread, */
- /* which makes programs even slower. might work on other unices, too. */
- #if EV_USE_CLOCK_SYSCALL
--# include <sys/syscall.h>
-+# include <unistd.h>
- # ifdef SYS_clock_gettime
- #  define clock_gettime(id, ts) syscall (SYS_clock_gettime, (id), (ts))
- #  undef EV_USE_MONOTONIC
 @@ -2216,24 +2218,7 @@
  {
    if (delay > EV_TS_CONST (0.))
@@ -44,17 +35,3 @@
      }
  }
  
-@@ -4100,11 +4085,13 @@
-               waittime = EV_TS_CONST (MAX_BLOCKTIME2);
- #endif
- #if !EV_PERIODIC_ENABLE
-+#if EV_USE_MONOTONIC
-             /* without periodics but with monotonic clock there is no need */
-             /* for any time jump detection, so sleep longer */
-             if (ecb_expect_true (have_monotonic))
-               waittime = EV_TS_CONST (MAX_BLOCKTIME2);
- #endif
-+#endif
- 
-             if (timercnt)
-               {
diff --git a/subprojects/packagefiles/libev/no_monotonic.diff b/subprojects/packagefiles/libev/no_monotonic.diff
new file mode 100644
index 0000000..382c241
--- /dev/null
+++ b/subprojects/packagefiles/libev/no_monotonic.diff
@@ -0,0 +1,16 @@
+--- a/ev.c 2020-03-18 08:27:32.000000000 -0400
++++ b/ev.c 2024-01-16 15:42:39.360703755 -0400
+@@ -4100,11 +4085,13 @@
+               waittime = EV_TS_CONST (MAX_BLOCKTIME2);
+ #endif
+ #if !EV_PERIODIC_ENABLE
++#if EV_USE_MONOTONIC
+             /* without periodics but with monotonic clock there is no need */
+             /* for any time jump detection, so sleep longer */
+             if (ecb_expect_true (have_monotonic))
+               waittime = EV_TS_CONST (MAX_BLOCKTIME2);
+ #endif
++#endif
+ 
+             if (timercnt)
+               {