napi: implement uv_async_t on POSIX by bridging to Bun's event loop

src/jsc/bindings/libuv/generate_uv_posix_stubs_constants.ts

@@ -14,8 +14,9 @@ export const test_skipped = [
 
 export const symbols = [
   "uv_accept",
-  "uv_async_init",
-  "uv_async_send",
+  // Defined in uv-posix-polyfills.c
+  // "uv_async_init",
+  // "uv_async_send",
   "uv_available_parallelism",
   "uv_backend_fd",
   "uv_backend_timeout",
@@ -29,7 +30,8 @@ export const symbols = [
   "uv_check_start",
   "uv_check_stop",
   "uv_clock_gettime",
-  "uv_close",
+  // Defined in uv-posix-polyfills.c
+  // "uv_close",
   "uv_cond_broadcast",
   "uv_cond_destroy",
   "uv_cond_init",
@@ -39,7 +41,8 @@ export const symbols = [
   "uv_cpu_info",
   "uv_cpumask_size",
   "uv_cwd",
-  "uv_default_loop",
+  // Defined in uv-posix-polyfills.c
+  // "uv_default_loop",
   "uv_disable_stdio_inheritance",
   "uv_dlclose",
   "uv_dlerror",
@@ -116,13 +119,15 @@ export const symbols = [
   "uv_getrusage_thread",
   "uv_gettimeofday",
   "uv_guess_handle",
-  "uv_handle_get_data",
-  "uv_handle_get_loop",
-  "uv_handle_get_type",
-  "uv_handle_set_data",
+  // Defined in uv-posix-polyfills.c
+  // "uv_handle_get_data",
+  // "uv_handle_get_loop",
+  // "uv_handle_get_type",
+  // "uv_handle_set_data",
   "uv_handle_size",
   "uv_handle_type_name",
-  "uv_has_ref",
+  // Defined in uv-posix-polyfills.c
+  // "uv_has_ref",
   // Defined in uv-posix-polyfills.cpp
   // "uv_hrtime",
   "uv_idle_init",
@@ -138,8 +143,9 @@ export const symbols = [
   "uv_ip6_addr",
   "uv_ip6_name",
   "uv_ip_name",
-  "uv_is_active",
-  "uv_is_closing",
+  // Defined in uv-posix-polyfills.c
+  // "uv_is_active",
+  // "uv_is_closing",
   "uv_is_readable",
   "uv_is_writable",
   "uv_key_create",
@@ -222,7 +228,8 @@ export const symbols = [
   "uv_read_start",
   "uv_read_stop",
   "uv_recv_buffer_size",
-  "uv_ref",
+  // Defined in uv-posix-polyfills.c
+  // "uv_ref",
   "uv_replace_allocator",
   "uv_req_get_data",
   "uv_req_get_type",
@@ -323,7 +330,8 @@ export const symbols = [
   "uv_udp_try_send",
   "uv_udp_try_send2",
   "uv_udp_using_recvmmsg",
-  "uv_unref",
+  // Defined in uv-posix-polyfills.c
+  // "uv_unref",
   "uv_update_time",
   "uv_uptime",
   "uv_utf16_length_as_wtf8",

src/jsc/bindings/uv-posix-polyfills.c

@@ -2,7 +2,9 @@
 
 #if OS(LINUX) || OS(DARWIN) || OS(FREEBSD)
 
+#include <assert.h>
 #include <pthread.h>
+#include <sched.h>
 #include <unistd.h>
 #include <stdlib.h>
 
@@ -138,4 +140,208 @@ UV_EXTERN void uv_mutex_unlock(uv_mutex_t* mutex)
         abort();
 }
 
+// ---------------------------------------------------------------------------
+// uv_handle_t / uv_async_t
+// ---------------------------------------------------------------------------
+//
+// On POSIX Bun does not run a libuv event loop. The `uv_loop_t*` that
+// `napi_get_uv_event_loop` / `uv_default_loop` hand out is really Bun's
+// `*mut EventLoop` (see `napi_get_uv_event_loop` in
+// src/runtime/napi/napi_body.rs). The shims below schedule work onto that
+// loop and adjust its keep-alive refcount.
+extern void Bun__uv_handle_schedule(uv_loop_t* loop, uv_handle_t* handle);
+extern void Bun__uv_handle_ref(uv_loop_t* loop, int delta);
+extern uv_loop_t* Bun__uv_default_loop(void);
+
+// Match the bits libuv uses so addons that peek at handle->flags see what
+// they expect (libuv's src/uv-common.h).
+#define BUN_UV_HANDLE_CLOSING 0x00000001
+#define BUN_UV_HANDLE_CLOSED 0x00000002
+#define BUN_UV_HANDLE_ACTIVE 0x00000004
+#define BUN_UV_HANDLE_REF 0x00000008
+
+// uv_async_t->pending is our "task is queued" bit: 0 idle, 1 queued. uv_close
+// sets it to 2 so subsequent uv_async_send calls observe non-zero and skip
+// scheduling.
+#define BUN_UV_ASYNC_CLOSING 2
+
+static int bun__is_supported_handle(const uv_handle_t* handle)
+{
+    return handle->type == UV_ASYNC;
+}
+
+UV_EXTERN uv_loop_t* uv_default_loop(void)
+{
+    return Bun__uv_default_loop();
+}
+
+UV_EXTERN void* uv_handle_get_data(const uv_handle_t* handle)
+{
+    return handle->data;
+}
+
+UV_EXTERN void uv_handle_set_data(uv_handle_t* handle, void* data)
+{
+    handle->data = data;
+}
+
+UV_EXTERN uv_loop_t* uv_handle_get_loop(const uv_handle_t* handle)
+{
+    return handle->loop;
+}
+
+UV_EXTERN uv_handle_type uv_handle_get_type(const uv_handle_t* handle)
+{
+    return handle->type;
+}
+
+UV_EXTERN int uv_has_ref(const uv_handle_t* handle)
+{
+    return (handle->flags & BUN_UV_HANDLE_REF) != 0;
+}
+
+UV_EXTERN int uv_is_active(const uv_handle_t* handle)
+{
+    return (handle->flags & BUN_UV_HANDLE_ACTIVE) != 0;
+}
+
+UV_EXTERN int uv_is_closing(const uv_handle_t* handle)
+{
+    return (handle->flags & (BUN_UV_HANDLE_CLOSING | BUN_UV_HANDLE_CLOSED)) != 0;
+}
+
+UV_EXTERN void uv_ref(uv_handle_t* handle)
+{
+    if (!bun__is_supported_handle(handle)) {
+        __bun_throw_not_implemented("uv_ref");
+    }
+    if (handle->flags & BUN_UV_HANDLE_REF)
+        return;
+    handle->flags |= BUN_UV_HANDLE_REF;
+    if ((handle->flags & BUN_UV_HANDLE_ACTIVE) && !(handle->flags & BUN_UV_HANDLE_CLOSING))
+        Bun__uv_handle_ref(handle->loop, 1);
+}
+
+UV_EXTERN void uv_unref(uv_handle_t* handle)
+{
+    if (!bun__is_supported_handle(handle)) {
+        __bun_throw_not_implemented("uv_unref");
+    }
+    if (!(handle->flags & BUN_UV_HANDLE_REF))
+        return;
+    handle->flags &= ~BUN_UV_HANDLE_REF;
+    if ((handle->flags & BUN_UV_HANDLE_ACTIVE) && !(handle->flags & BUN_UV_HANDLE_CLOSING))
+        Bun__uv_handle_ref(handle->loop, -1);
+}
+
+UV_EXTERN int uv_async_init(uv_loop_t* loop, uv_async_t* handle, uv_async_cb async_cb)
+{
+    if (loop == NULL)
+        return UV_EINVAL;
+    handle->loop = loop;
+    handle->type = UV_ASYNC;
+    handle->close_cb = NULL;
+    handle->next_closing = NULL;
+    // libuv: uv__handle_init sets REF, then uv__handle_start sets ACTIVE and
+    // bumps loop->active_handles. Mirror that so a freshly initialized async
+    // handle keeps the loop alive.
+    handle->flags = BUN_UV_HANDLE_REF | BUN_UV_HANDLE_ACTIVE;
+    handle->async_cb = async_cb;
+    handle->u.fd = 0;
+    __atomic_store_n(&handle->pending, 0, __ATOMIC_SEQ_CST);
+    Bun__uv_handle_ref(loop, 1);
+    return 0;
+}
+
+UV_EXTERN int uv_async_send(uv_async_t* handle)
+{
+    // libuv coalesces: only the 0->1 transition schedules. handle->u.fd is the
+    // "busy" counter libuv uses so uv_close can spin until no thread is between
+    // the exchange and the schedule below.
+    if (__atomic_load_n(&handle->pending, __ATOMIC_RELAXED) != 0)
+        return 0;
+    __atomic_fetch_add(&handle->u.fd, 1, __ATOMIC_SEQ_CST);
+    if (__atomic_exchange_n(&handle->pending, 1, __ATOMIC_SEQ_CST) == 0)
+        Bun__uv_handle_schedule(handle->loop, (uv_handle_t*)handle);
+    __atomic_fetch_sub(&handle->u.fd, 1, __ATOMIC_SEQ_CST);
+    return 0;
+}
+
+// Called from the event loop's task dispatcher on the loop thread.
+UV_EXTERN void Bun__uv_handle_dispatch(uv_handle_t* handle)
+{
+    if (handle->type != UV_ASYNC)
+        return;
+    uv_async_t* async = (uv_async_t*)handle;
+    if (handle->flags & BUN_UV_HANDLE_CLOSING) {
+        // uv_close ran; this is the deferred close. Leave pending non-zero so
+        // a racing uv_async_send cannot schedule a second task behind the
+        // close callback. Release the loop ref uv_close held (or took) so the
+        // process can exit once the close callback returns.
+        handle->flags |= BUN_UV_HANDLE_CLOSED;
+        Bun__uv_handle_ref(handle->loop, -1);
+        if (handle->close_cb != NULL)
+            handle->close_cb(handle);
+        return;
+    }
+    // Reset before the callback so a send inside the callback schedules again,
+    // matching libuv's uv__async_io.
+    __atomic_store_n(&async->pending, 0, __ATOMIC_SEQ_CST);
+    if (async->async_cb != NULL)
+        async->async_cb(async);
+}
+
+static void bun__uv_async_spin(uv_async_t* handle)
+{
+    int i;
+    for (;;) {
+        for (i = 0; i < 997; i++) {
+            if (__atomic_load_n(&handle->u.fd, __ATOMIC_SEQ_CST) == 0)
+                return;
+#if defined(__i386__) || defined(__x86_64__)
+            __asm__ __volatile__("pause" ::: "memory");
+#elif defined(__aarch64__) || defined(__arm__)
+            __asm__ __volatile__("yield" ::: "memory");
+#endif
+        }
+        sched_yield();
+    }
+}
+
+UV_EXTERN void uv_close(uv_handle_t* handle, uv_close_cb close_cb)
+{
+    if (!bun__is_supported_handle(handle)) {
+        __bun_throw_not_implemented("uv_close");
+    }
+    // libuv: assert(!uv__is_closing(handle)) before any field write.
+    if (handle->flags & (BUN_UV_HANDLE_CLOSING | BUN_UV_HANDLE_CLOSED)) {
+        assert(0);
+        return;
+    }
+    handle->close_cb = close_cb;
+    // A closing handle keeps the loop alive until close_cb runs (libuv's
+    // closing_handles list). If the handle was unref'd, take a ref back for
+    // the duration of the deferred close; Bun__uv_handle_dispatch drops it.
+    int had_active_ref = (handle->flags & BUN_UV_HANDLE_REF) && (handle->flags & BUN_UV_HANDLE_ACTIVE);
+    if (!had_active_ref)
+        Bun__uv_handle_ref(handle->loop, 1);
+    handle->flags |= BUN_UV_HANDLE_CLOSING;
+    handle->flags &= ~BUN_UV_HANDLE_ACTIVE;
+
+    uv_async_t* async = (uv_async_t*)handle;
+    // Force pending non-zero so no uv_async_send after this point schedules a
+    // new task, then wait for any send that is mid-flight between its exchange
+    // and its schedule call.
+    int prev = __atomic_exchange_n(&async->pending, BUN_UV_ASYNC_CLOSING, __ATOMIC_SEQ_CST);
+    bun__uv_async_spin(async);
+    if (prev == 0) {
+        // No send task is or will be queued; schedule the close ourselves.
+        Bun__uv_handle_schedule(handle->loop, handle);
+    } else {
+        // A send already queued a task (or is about to, within the spin window
+        // we just waited out). That task will observe CLOSING above and run the
+        // close path instead of async_cb.
+    }
+}
+
 #endif