From 32d33e46d9762532bd0d0e45280c0e9e2b4dbed7 Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Fri, 14 Aug 2026 18:46:37 +0530 Subject: [PATCH 01/11] Swap worklet IPC from pipes to the in-process queue Bare Kit is in-process - worklets are threads - so the pipe transport, with its async close on exit, was the wrong tool and the root of the EOF and win32 problems (holepunchto/bare#160, #92). Bare.IPC becomes a bare-stream Duplex over the queue, and the worklet closes its end synchronously after bare_run, so the host reaches EOF on every exit path with no loop pump and nothing scheduled from the exit event. The host polls readiness rather than reading a pipe: apple keeps its dispatch source, linux and win32 now share one uv implementation instead of a hand-rolled epoll thread and a pipe loop, and android keeps its looper with the ipc owning the eventfd. The two poll tests closed their loop while their poll was still armed. Pipes hid that: their EOF arrived asynchronously at thread teardown, while the queue closes synchronously, so the wake lands in the window and the callback posts to a closed loop - a segfault on linux-x64. The poll is retired first now. bare_ipc_get_incoming and bare_ipc_get_outgoing go with the pipes; the win32 exports follow. Both EOF tests now run on win32, which is what this was for. --- shared/CMakeLists.txt | 14 +- shared/android/ipc.c | 105 +++++++--- shared/apple/ipc.h | 4 - shared/apple/ipc.m | 147 ++++++++------ shared/ipc.c | 67 +++++- shared/ipc.h | 50 +++-- shared/linux/ipc.c | 184 ----------------- shared/linux/ipc.h | 34 ---- shared/posix/ipc.c | 42 ---- shared/posix/ipc.h | 21 -- shared/uv/ipc.c | 244 ++++++++++++++++++++++ shared/uv/ipc.h | 24 +++ shared/win32/ipc.c | 280 -------------------------- shared/win32/ipc.h | 58 ------ shared/worklet.c | 213 ++++++++++++++++++-- shared/worklet.h | 10 +- shared/worklet.js | 101 +++++++++- test/shared/CMakeLists.txt | 11 +- test/shared/worklet-ipc-large-write.c | 6 +- test/shared/worklet-ipc.c | 6 +- test/shared/worklet-jsend-eof.c | 8 +- test/shared/worklet-suspend-resume.c | 3 +- win32/ipc.def | 2 - 23 files changed, 844 insertions(+), 790 deletions(-) delete mode 100644 shared/linux/ipc.c delete mode 100644 shared/linux/ipc.h delete mode 100644 shared/posix/ipc.c delete mode 100644 shared/posix/ipc.h create mode 100644 shared/uv/ipc.c create mode 100644 shared/uv/ipc.h delete mode 100644 shared/win32/ipc.c delete mode 100644 shared/win32/ipc.h diff --git a/shared/CMakeLists.txt b/shared/CMakeLists.txt index 6f62b45..4a06397 100644 --- a/shared/CMakeLists.txt +++ b/shared/CMakeLists.txt @@ -52,8 +52,6 @@ if(APPLE) target_sources( bare_worklet PRIVATE - posix/ipc.h - posix/ipc.c apple/ipc.h apple/ipc.m apple/suspension.h @@ -85,8 +83,6 @@ if(ANDROID) target_sources( bare_worklet PRIVATE - posix/ipc.h - posix/ipc.c android/ipc.c android/ipc.h android/suspension.c @@ -113,10 +109,8 @@ if(LINUX) target_sources( bare_worklet PRIVATE - posix/ipc.h - posix/ipc.c - linux/ipc.c - linux/ipc.h + uv/ipc.c + uv/ipc.h linux/suspension.c linux/suspension.h ) @@ -132,8 +126,8 @@ if(WIN32) target_sources( bare_worklet PRIVATE - win32/ipc.c - win32/ipc.h + uv/ipc.c + uv/ipc.h win32/suspension.c win32/suspension.h ) diff --git a/shared/android/ipc.c b/shared/android/ipc.c index 370c937..7e0fbee 100644 --- a/shared/android/ipc.c +++ b/shared/android/ipc.c @@ -1,9 +1,58 @@ #include #include +#include +#include #include +#include +#include #include "ipc.h" +// The looper wakes on a file descriptor, so the signal is an eventfd. The ipc +// owns it because the worklet writes to it; the poll owns the registration, +// which belongs to the host thread whose looper it is. +typedef struct { + int wake; +} bare_ipc_signal_t; + +int +bare_ipc__signal_init(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = malloc(sizeof(bare_ipc_signal_t)); + + if (signal == NULL) return -1; + + // The count stays up until a callback reads it, so a poll that starts after a + // signal still wakes. + signal->wake = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + assert(signal->wake >= 0); + + ipc->signal = signal; + + return 0; +} + +void +bare_ipc__signal(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + uint64_t value = 1; + + ssize_t written = write(signal->wake, &value, sizeof(value)); + assert(written == sizeof(value)); +} + +void +bare_ipc__signal_destroy(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + int err = close(signal->wake); + assert(err == 0); + + free(signal); + + ipc->signal = NULL; +} + int bare_ipc_poll_alloc(bare_ipc_poll_t **result) { bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t)); @@ -20,10 +69,13 @@ bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) { poll->ipc = ipc; poll->events = 0; poll->cb = NULL; + poll->data = NULL; poll->looper = ALooper_forThread(); ALooper_acquire(poll->looper); + ipc->poll = poll; + return 0; } @@ -33,6 +85,8 @@ bare_ipc_poll_destroy(bare_ipc_poll_t *poll) { err = bare_ipc_poll_stop(poll); assert(err == 0); + poll->ipc->poll = NULL; + ALooper_release(poll->looper); } @@ -51,11 +105,22 @@ bare_ipc_poll_get_ipc(bare_ipc_poll_t *poll) { return poll->ipc; } +// Runs on the looper's thread, which is also the thread that started the poll, so +// the poll cannot go away underneath. static int -bare_ipc__on_poll(int fd, int events, void *data) { - bare_ipc_poll_t *poll = data; +bare_ipc__on_wake(int fd, int events, void *data) { + bare_ipc_poll_t *poll = (bare_ipc_poll_t *) data; - if (poll->cb) poll->cb(poll, (events & ALOOPER_EVENT_INPUT ? bare_ipc_readable : 0) | (events & ALOOPER_EVENT_OUTPUT ? bare_ipc_writable : 0)); + uint64_t value; + + ssize_t read_len = read(fd, &value, sizeof(value)); + assert(read_len == sizeof(value) || errno == EAGAIN); + + if (poll->cb != NULL) { + int ready = bare_ipc__ready(poll->ipc, poll->events); + + if (ready != 0) poll->cb(poll, ready); + } return 1; // Don't deregister the file descriptor } @@ -64,43 +129,29 @@ int bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) { if (events == 0) return bare_ipc_poll_stop(poll); - int err; + bare_ipc_t *ipc = poll->ipc; - if ((events & bare_ipc_readable) == 0) { - if ((poll->events & bare_ipc_readable) != 0) { - err = ALooper_removeFd(poll->looper, poll->ipc->incoming); - assert(err == 1); - } - } else { - if ((poll->events & bare_ipc_readable) == 0) { - err = ALooper_addFd(poll->looper, poll->ipc->incoming, ALOOPER_POLL_CALLBACK, ALOOPER_EVENT_INPUT, bare_ipc__on_poll, (void *) poll); - assert(err == 1); - } - } + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; - if ((events & bare_ipc_writable) == 0) { - if ((poll->events & bare_ipc_writable) != 0) { - err = ALooper_removeFd(poll->looper, poll->ipc->outgoing); - assert(err == 1); - } - } else { - if ((poll->events & bare_ipc_writable) == 0) { - err = ALooper_addFd(poll->looper, poll->ipc->outgoing, ALOOPER_POLL_CALLBACK, ALOOPER_EVENT_OUTPUT, bare_ipc__on_poll, (void *) poll); - assert(err == 1); - } + if (poll->events == 0) { + int err = ALooper_addFd(poll->looper, signal->wake, ALOOPER_POLL_CALLBACK, ALOOPER_EVENT_INPUT, bare_ipc__on_wake, (void *) poll); + assert(err == 1); } poll->events = events; poll->cb = cb; + // Level-triggered: re-check in case data or space is already there. + bare_ipc__signal(ipc); + return 0; } int bare_ipc_poll_stop(bare_ipc_poll_t *poll) { - ALooper_removeFd(poll->looper, poll->ipc->incoming); + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) poll->ipc->signal; - ALooper_removeFd(poll->looper, poll->ipc->outgoing); + if (poll->events != 0) ALooper_removeFd(poll->looper, signal->wake); poll->events = 0; poll->cb = NULL; diff --git a/shared/apple/ipc.h b/shared/apple/ipc.h index 742b882..30075bb 100644 --- a/shared/apple/ipc.h +++ b/shared/apple/ipc.h @@ -15,10 +15,6 @@ extern "C" { struct bare_ipc_poll_s { bare_ipc_t *ipc; - dispatch_queue_t queue; - dispatch_source_t reader; - dispatch_source_t writer; - int events; _Atomic bare_ipc_poll_cb cb; diff --git a/shared/apple/ipc.m b/shared/apple/ipc.m index 75c052e..ebca686 100644 --- a/shared/apple/ipc.m +++ b/shared/apple/ipc.m @@ -5,89 +5,130 @@ static const void *const bare_ipc_poll_queue = &bare_ipc_poll_queue; -int -bare_ipc_poll_alloc(bare_ipc_poll_t **result) { - bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t)); +// Waits for the serial queue to reach the current point, unless we are already +// on it, in which case the handler is this call and there is nothing to wait for. +static void +bare_ipc__drain(dispatch_queue_t queue) { + if (dispatch_get_specific(bare_ipc_poll_queue) != NULL) return; - if (poll == NULL) return -1; + dispatch_semaphore_t done = dispatch_semaphore_create(0); - *result = poll; + dispatch_async(queue, ^{ + dispatch_semaphore_signal(done); + }); - return 0; + dispatch_semaphore_wait(done, DISPATCH_TIME_FOREVER); } +typedef struct { + dispatch_queue_t queue; + dispatch_source_t source; +} bare_ipc_signal_t; + int -bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) { +bare_ipc__signal_init(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = malloc(sizeof(bare_ipc_signal_t)); + + if (signal == NULL) return -1; + dispatch_queue_t queue = dispatch_queue_create("to.holepunch.bare.kit.ipc", DISPATCH_QUEUE_SERIAL); dispatch_queue_set_specific(queue, bare_ipc_poll_queue, (void *) bare_ipc_poll_queue, NULL); - dispatch_source_t reader = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, ipc->incoming, 0, queue); - - dispatch_source_t writer = dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE, ipc->outgoing, 0, queue); + // A manual source the queue signal triggers; the handler reports whichever + // requested events are actually ready of whichever poll is subscribed. + dispatch_source_t source = dispatch_source_create(DISPATCH_SOURCE_TYPE_DATA_OR, 0, 0, queue); - dispatch_source_set_cancel_handler(reader, ^{ - dispatch_release(reader); + dispatch_source_set_cancel_handler(source, ^{ + dispatch_release(source); }); - dispatch_source_set_cancel_handler(writer, ^{ - dispatch_release(writer); - }); + dispatch_source_set_event_handler(source, ^{ + bare_ipc_poll_t *poll = ipc->poll; + + if (poll == NULL) return; - dispatch_source_set_event_handler(reader, ^{ bare_ipc_poll_cb cb = atomic_load(&poll->cb); if (cb == NULL) return; + int ready = bare_ipc__ready(ipc, poll->events); + + if (ready == 0) return; + @autoreleasepool { - cb(poll, bare_ipc_readable); + cb(poll, ready); } }); - dispatch_source_set_event_handler(writer, ^{ - bare_ipc_poll_cb cb = atomic_load(&poll->cb); + signal->queue = queue; + signal->source = source; - if (cb == NULL) return; + ipc->signal = signal; - @autoreleasepool { - cb(poll, bare_ipc_writable); - } - }); + dispatch_resume(source); + + return 0; +} + +void +bare_ipc__signal(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + dispatch_source_merge_data(signal->source, 1); +} + +void +bare_ipc__signal_destroy(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + dispatch_source_cancel(signal->source); + + bare_ipc__drain(signal->queue); + + dispatch_release(signal->queue); + + free(signal); + + ipc->signal = NULL; +} + +int +bare_ipc_poll_alloc(bare_ipc_poll_t **result) { + bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t)); + + if (poll == NULL) return -1; + + *result = poll; + + return 0; +} + +int +bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) { poll->ipc = ipc; poll->events = 0; - poll->queue = queue; - poll->reader = reader; - poll->writer = writer; + poll->data = NULL; atomic_init(&poll->cb, NULL); + ipc->poll = poll; + return 0; } void bare_ipc_poll_destroy(bare_ipc_poll_t *poll) { - int err; - err = bare_ipc_poll_stop(poll); - assert(err == 0); + bare_ipc_t *ipc = poll->ipc; - dispatch_resume(poll->reader); - dispatch_source_cancel(poll->reader); - - dispatch_resume(poll->writer); - dispatch_source_cancel(poll->writer); - - if (dispatch_get_specific(bare_ipc_poll_queue) == NULL) { - dispatch_semaphore_t done = dispatch_semaphore_create(0); + atomic_store(&poll->cb, NULL); - dispatch_async(poll->queue, ^{ - dispatch_semaphore_signal(done); - }); + ipc->poll = NULL; - dispatch_semaphore_wait(done, DISPATCH_TIME_FOREVER); - } + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; - dispatch_release(poll->queue); + // Let any handler already running finish before the caller frees the poll. + bare_ipc__drain(signal->queue); } void * @@ -109,30 +150,18 @@ bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) { if (events == 0) return bare_ipc_poll_stop(poll); - if ((events & bare_ipc_readable) == 0) { - if ((poll->events & bare_ipc_readable) != 0) dispatch_suspend(poll->reader); - } else { - if ((poll->events & bare_ipc_readable) == 0) dispatch_resume(poll->reader); - } - - if ((events & bare_ipc_writable) == 0) { - if ((poll->events & bare_ipc_writable) != 0) dispatch_suspend(poll->writer); - } else { - if ((poll->events & bare_ipc_writable) == 0) dispatch_resume(poll->writer); - } - poll->events = events; atomic_store(&poll->cb, cb); + // Level-trigger: re-check immediately in case data or space is already ready. + bare_ipc__signal(poll->ipc); + return 0; } int bare_ipc_poll_stop(bare_ipc_poll_t *poll) { - if ((poll->events & bare_ipc_readable) != 0) dispatch_suspend(poll->reader); - if ((poll->events & bare_ipc_writable) != 0) dispatch_suspend(poll->writer); - poll->events = 0; atomic_store(&poll->cb, NULL); diff --git a/shared/ipc.c b/shared/ipc.c index 04629bf..9ad963c 100644 --- a/shared/ipc.c +++ b/shared/ipc.c @@ -1,5 +1,17 @@ +#include + #include "ipc.h" +// Runs on the worklet thread when the queue changes. +static void +bare_ipc__on_signal(bare_queue_port_t *port) { + bare_ipc_t *ipc = (bare_ipc_t *) port->data; + + if (ipc == NULL) return; + + bare_ipc__signal(ipc); +} + int bare_ipc_alloc(bare_ipc_t **result) { bare_ipc_t *ipc = malloc(sizeof(bare_ipc_t)); @@ -12,11 +24,58 @@ bare_ipc_alloc(bare_ipc_t **result) { } int -bare_ipc_get_incoming(bare_ipc_t *ipc) { - return ipc->incoming; +bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet) { + ipc->poll = NULL; + ipc->signal = NULL; + + int err = bare_ipc__signal_init(ipc); + + if (err < 0) return err; + + ipc->port = bare_queue_open_thread(worklet->queue, bare_ipc__on_signal); + ipc->port->data = ipc; + + return 0; +} + +void +bare_ipc_destroy(bare_ipc_t *ipc) { + bare_queue_t *queue = ipc->port->queue; + + // Order matters: close marks the end shut, retiring clears the callback under + // the queue's lock, and only then is nothing left that can reach the signal. + bare_queue_close(ipc->port); + + bare_queue_shutdown_thread(queue); + + bare_ipc__signal_destroy(ipc); +} + +int +bare_ipc__ready(bare_ipc_t *ipc, int events) { + int ready = 0; + + if ((events & bare_ipc_readable) != 0 && bare_queue_readable(ipc->port)) ready |= bare_ipc_readable; + if ((events & bare_ipc_writable) != 0 && bare_queue_writable(ipc->port)) ready |= bare_ipc_writable; + + return ready; +} + +int +bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len) { + int status = bare_queue_read(ipc->port, data, len); + + if (status == bare_queue_would_block) return bare_ipc_would_block; + + return 0; } int -bare_ipc_get_outgoing(bare_ipc_t *ipc) { - return ipc->outgoing; +bare_ipc_write(bare_ipc_t *ipc, const void *data, size_t len) { + int n = bare_queue_write(ipc->port, data, len); + + if (n == bare_queue_would_block) return bare_ipc_would_block; + if (n == bare_queue_closed) return bare_ipc_error; + + return n; } diff --git a/shared/ipc.h b/shared/ipc.h index 70fffdc..7fde258 100644 --- a/shared/ipc.h +++ b/shared/ipc.h @@ -7,10 +7,9 @@ extern "C" { #include +#include "queue.h" #include "worklet.h" -#define BARE_IPC_READ_BUFFER_SIZE 64 * 1024 - typedef struct bare_ipc_s bare_ipc_t; typedef struct bare_ipc_poll_s bare_ipc_poll_t; @@ -26,6 +25,32 @@ enum { bare_ipc_error = -2, }; +struct bare_ipc_s { + bare_queue_port_t *port; + bare_ipc_poll_t *poll; + + // The primitive that hops a queue change onto the host run loop. Owned here, + // for the lifetime of the port, because the worklet signals through it: a poll + // comes and goes underneath. + void *signal; +}; + +// Per platform. `signal` runs on the worklet thread, so it may only touch state +// owned by the ipc. +int +bare_ipc__signal_init(bare_ipc_t *ipc); + +void +bare_ipc__signal_destroy(bare_ipc_t *ipc); + +void +bare_ipc__signal(bare_ipc_t *ipc); + +// Which of the events a poll asked for are ready right now. A wake says the +// queue changed, not how, so the platform asks after each one. +int +bare_ipc__ready(bare_ipc_t *ipc, int events); + #if defined(BARE_KIT_DARWIN) || defined(BARE_KIT_IOS) #include "apple/ipc.h" #endif @@ -34,14 +59,10 @@ enum { #include "android/ipc.h" #endif -#if defined(BARE_KIT_LINUX) -#include "linux/ipc.h" -#endif - -#if defined(BARE_KIT_WINDOWS) -#include "win32/ipc.h" -#else -#include "posix/ipc.h" +// Neither Linux nor Windows hands us a host run loop to post to, so both wake on +// a loop of their own. +#if defined(BARE_KIT_LINUX) || defined(BARE_KIT_WINDOWS) +#include "uv/ipc.h" #endif int @@ -50,15 +71,12 @@ bare_ipc_alloc(bare_ipc_t **result); int bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet); +// Retires the host end. Must be called before `bare_worklet_destroy`: the queue +// belongs to the worklet, which may free it as soon as the host has signalled it +// to be destroyed. A host that keeps an ipc must therefore keep its worklet. void bare_ipc_destroy(bare_ipc_t *ipc); -int -bare_ipc_get_incoming(bare_ipc_t *ipc); - -int -bare_ipc_get_outgoing(bare_ipc_t *ipc); - int bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len); diff --git a/shared/linux/ipc.c b/shared/linux/ipc.c deleted file mode 100644 index 28d53f3..0000000 --- a/shared/linux/ipc.c +++ /dev/null @@ -1,184 +0,0 @@ -#include -#include -#include -#include -#include -#include - -#include "ipc.h" - -enum { - bare_ipc_poll_close, - bare_ipc_poll_incoming, - bare_ipc_poll_outging, -}; - -void * -bare_ipc_poll__poller(void *data) { - bare_ipc_poll_t *poll = (bare_ipc_poll_t *) data; - - while (true) { - struct epoll_event events[4]; - int n = epoll_wait(poll->fd.poll, events, 4, -1); - assert(n >= 0 || errno == EINTR); - - if (errno == EINTR) continue; - - int flags = 0; - - for (int i = 0; i < n; i++) { - struct epoll_event event = events[i]; - - if (event.data.fd == bare_ipc_poll_close) { - close(poll->fd.poll); - close(poll->fd.close); - - pthread_exit(0); - } - - if (event.data.fd == bare_ipc_poll_incoming && (event.events & EPOLLIN) != 0) { - flags |= bare_ipc_readable; - } - - if (event.data.fd == bare_ipc_poll_outging && (event.events & EPOLLOUT) != 0) { - flags |= bare_ipc_writable; - } - } - - if (poll->cb) poll->cb(poll, flags); - } -} - -int -bare_ipc_poll_alloc(bare_ipc_poll_t **result) { - bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t)); - - if (poll == NULL) return -1; - - *result = poll; - - return 0; -} - -int -bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) { - int err; - - poll->ipc = ipc; - poll->fd.poll = epoll_create(3); - poll->fd.close = eventfd(0, 0); - poll->events = 0; - poll->cb = NULL; - - struct epoll_event signal = { - .events = EPOLLIN, - .data.fd = bare_ipc_poll_close, - }; - - err = epoll_ctl(poll->fd.poll, EPOLL_CTL_ADD, poll->fd.close, &signal); - assert(err == 0); - - pthread_create(&poll->thread, NULL, &bare_ipc_poll__poller, (void *) poll); - - return 0; -} - -void -bare_ipc_poll_destroy(bare_ipc_poll_t *poll) { - int err; - err = bare_ipc_poll_stop(poll); - assert(err == 0); - - uint64_t signal = 1; - - ssize_t written = write(poll->fd.close, (const void *) &signal, 8); - assert(written == 8); - - err = pthread_join(poll->thread, NULL); - assert(err == 0); -} - -void * -bare_ipc_poll_get_data(bare_ipc_poll_t *poll) { - return poll->data; -} - -void -bare_ipc_poll_set_data(bare_ipc_poll_t *poll, void *data) { - poll->data = data; -} - -bare_ipc_t * -bare_ipc_poll_get_ipc(bare_ipc_poll_t *poll) { - return poll->ipc; -} - -int -bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) { - if (events == 0) return bare_ipc_poll_stop(poll); - - int err; - - struct epoll_event incoming = { - .events = EPOLLIN, - .data.fd = bare_ipc_poll_incoming, - }; - - if ((events & bare_ipc_readable) == 0) { - if ((poll->events & bare_ipc_readable) != 0) { - err = epoll_ctl(poll->fd.poll, EPOLL_CTL_DEL, poll->ipc->incoming, &incoming); - assert(err == 0); - } - } else { - if ((poll->events & bare_ipc_readable) == 0) { - err = epoll_ctl(poll->fd.poll, EPOLL_CTL_ADD, poll->ipc->incoming, &incoming); - assert(err == 0); - } - } - - struct epoll_event outgoing = { - .events = EPOLLOUT, - .data.fd = bare_ipc_poll_outging, - }; - - if ((events & bare_ipc_writable) == 0) { - if ((poll->events & bare_ipc_writable) != 0) { - err = epoll_ctl(poll->fd.poll, EPOLL_CTL_DEL, poll->ipc->outgoing, &outgoing); - assert(err == 0); - } - } else { - if ((poll->events & bare_ipc_writable) == 0) { - err = epoll_ctl(poll->fd.poll, EPOLL_CTL_ADD, poll->ipc->outgoing, &outgoing); - assert(err == 0); - } - } - - poll->events = events; - poll->cb = cb; - - return 0; -} - -int -bare_ipc_poll_stop(bare_ipc_poll_t *poll) { - int err; - - struct epoll_event incoming = { - .events = EPOLLIN, - }; - - err = epoll_ctl(poll->fd.poll, EPOLL_CTL_DEL, poll->ipc->incoming, &incoming); - assert(err == 0 || errno == ENOENT); - - struct epoll_event outgoing = { - .events = EPOLLOUT, - }; - - err = epoll_ctl(poll->fd.poll, EPOLL_CTL_DEL, poll->ipc->outgoing, &outgoing); - assert(err == 0 || errno == ENOENT); - - poll->events = 0; - poll->cb = NULL; - - return 0; -} diff --git a/shared/linux/ipc.h b/shared/linux/ipc.h deleted file mode 100644 index 81bf6c0..0000000 --- a/shared/linux/ipc.h +++ /dev/null @@ -1,34 +0,0 @@ -#ifndef BARE_KIT_LINUX_IPC_H -#define BARE_KIT_LINUX_IPC_H - -#ifdef __cplusplus -extern "C" { -#endif - -#include -#include - -#include "../ipc.h" - -struct bare_ipc_poll_s { - bare_ipc_t *ipc; - - struct { - int poll; - int close; - } fd; - - pthread_t thread; - - int events; - - bare_ipc_poll_cb cb; - - void *data; -}; - -#ifdef __cplusplus -} -#endif - -#endif // BARE_KIT_LINUX_IPC_H diff --git a/shared/posix/ipc.c b/shared/posix/ipc.c deleted file mode 100644 index 102999a..0000000 --- a/shared/posix/ipc.c +++ /dev/null @@ -1,42 +0,0 @@ -#include - -#include "ipc.h" - -int -bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet) { - ipc->incoming = dup(worklet->incoming); - ipc->outgoing = dup(worklet->outgoing); - - return 0; -} - -void -bare_ipc_destroy(bare_ipc_t *ipc) { - close(ipc->incoming); - close(ipc->outgoing); -} - -int -bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len) { - ssize_t res = read(ipc->incoming, ipc->data, BARE_IPC_READ_BUFFER_SIZE); - - if (res < 0) { - return (errno == EAGAIN || errno == EWOULDBLOCK) ? bare_ipc_would_block : bare_ipc_error; - } - - *len = res; - *data = ipc->data; - - return 0; -} - -int -bare_ipc_write(bare_ipc_t *ipc, const void *data, size_t len) { - ssize_t res = write(ipc->outgoing, data, len); - - if (res < 0) { - return (errno == EAGAIN || errno == EWOULDBLOCK) ? bare_ipc_would_block : bare_ipc_error; - } - - return res; -} diff --git a/shared/posix/ipc.h b/shared/posix/ipc.h deleted file mode 100644 index 8621f2b..0000000 --- a/shared/posix/ipc.h +++ /dev/null @@ -1,21 +0,0 @@ -#ifndef BARE_KIT_POSIX_IPC_H -#define BARE_KIT_POSIX_IPC_H - -#ifdef __cplusplus -extern "C" { -#endif - -#include "../ipc.h" - -struct bare_ipc_s { - int incoming; - int outgoing; - - char data[BARE_IPC_READ_BUFFER_SIZE]; -}; - -#ifdef __cplusplus -} -#endif - -#endif // BARE_KIT_POSIX_IPC_H diff --git a/shared/uv/ipc.c b/shared/uv/ipc.c new file mode 100644 index 0000000..99b328d --- /dev/null +++ b/shared/uv/ipc.c @@ -0,0 +1,244 @@ +#include +#include +#include +#include + +#include "ipc.h" + +// A loop of our own, on a thread of our own: the callback runs there, and hosts +// marshal to their UI thread themselves. `running` is held across the callback so +// a poll cannot be torn down underneath one, and the key marks the thread that +// holds it, for a poll that stops or destroys itself from inside its callback. +typedef struct { + uv_loop_t loop; + + uv_async_t wake; + uv_async_t close; + + uv_thread_t thread; + uv_key_t key; + uv_mutex_t running; +} bare_ipc_signal_t; + +static inline bool +bare_ipc__reentrant(bare_ipc_signal_t *signal) { + return uv_key_get(&signal->key) == (void *) signal; +} + +static inline void +bare_ipc__lock(bare_ipc_signal_t *signal) { + if (!bare_ipc__reentrant(signal)) uv_mutex_lock(&signal->running); +} + +static inline void +bare_ipc__unlock(bare_ipc_signal_t *signal) { + if (!bare_ipc__reentrant(signal)) uv_mutex_unlock(&signal->running); +} + +static void +bare_ipc__on_wake(uv_async_t *handle) { + bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle); + + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + uv_mutex_lock(&signal->running); + + bare_ipc_poll_t *poll = ipc->poll; + + if (poll != NULL && poll->cb != NULL) { + int ready = bare_ipc__ready(ipc, poll->events); + + if (ready != 0) poll->cb(poll, ready); + } + + uv_mutex_unlock(&signal->running); +} + +static void +bare_ipc__on_close(uv_async_t *handle) { + bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle); + + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + uv_close((uv_handle_t *) &signal->wake, NULL); + uv_close((uv_handle_t *) &signal->close, NULL); +} + +static void +bare_ipc__signal_thread(void *data) { + bare_ipc_t *ipc = (bare_ipc_t *) data; + + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + uv_key_set(&signal->key, (void *) signal); + + int err; + + err = uv_run(&signal->loop, UV_RUN_DEFAULT); + assert(err == 0); + + err = uv_loop_close(&signal->loop); + assert(err == 0); +} + +int +bare_ipc__signal_init(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = malloc(sizeof(bare_ipc_signal_t)); + + if (signal == NULL) return -1; + + int err; + + err = uv_loop_init(&signal->loop); + assert(err == 0); + + err = uv_async_init(&signal->loop, &signal->wake, bare_ipc__on_wake); + assert(err == 0); + + uv_handle_set_data((uv_handle_t *) &signal->wake, (void *) ipc); + + err = uv_async_init(&signal->loop, &signal->close, bare_ipc__on_close); + assert(err == 0); + + uv_handle_set_data((uv_handle_t *) &signal->close, (void *) ipc); + + err = uv_key_create(&signal->key); + assert(err == 0); + + err = uv_mutex_init(&signal->running); + assert(err == 0); + + ipc->signal = signal; + + // The loop and its handles are up, so a signal arriving the moment the port + // opens has somewhere to land. + err = uv_thread_create(&signal->thread, bare_ipc__signal_thread, (void *) ipc); + assert(err == 0); + + return 0; +} + +void +bare_ipc__signal(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + int err = uv_async_send(&signal->wake); + assert(err == 0); +} + +void +bare_ipc__signal_destroy(bare_ipc_t *ipc) { + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + int err; + + err = uv_async_send(&signal->close); + assert(err == 0); + + err = uv_thread_join(&signal->thread); + assert(err == 0); + + uv_mutex_destroy(&signal->running); + + uv_key_delete(&signal->key); + + free(signal); + + ipc->signal = NULL; +} + +int +bare_ipc_poll_alloc(bare_ipc_poll_t **result) { + bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t)); + + if (poll == NULL) return -1; + + *result = poll; + + return 0; +} + +int +bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) { + poll->ipc = ipc; + poll->events = 0; + poll->cb = NULL; + poll->data = NULL; + + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + bare_ipc__lock(signal); + + ipc->poll = poll; + + bare_ipc__unlock(signal); + + return 0; +} + +void +bare_ipc_poll_destroy(bare_ipc_poll_t *poll) { + bare_ipc_t *ipc = poll->ipc; + + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + // Under the lock, so the caller may free the poll as soon as this returns. + bare_ipc__lock(signal); + + poll->events = 0; + poll->cb = NULL; + + ipc->poll = NULL; + + bare_ipc__unlock(signal); +} + +void * +bare_ipc_poll_get_data(bare_ipc_poll_t *poll) { + return poll->data; +} + +void +bare_ipc_poll_set_data(bare_ipc_poll_t *poll, void *data) { + poll->data = data; +} + +bare_ipc_t * +bare_ipc_poll_get_ipc(bare_ipc_poll_t *poll) { + return poll->ipc; +} + +int +bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) { + if (events == 0) return bare_ipc_poll_stop(poll); + + bare_ipc_t *ipc = poll->ipc; + + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; + + bare_ipc__lock(signal); + + poll->events = events; + poll->cb = cb; + + bare_ipc__unlock(signal); + + // Level-triggered: re-check in case data or space is already there. + bare_ipc__signal(ipc); + + return 0; +} + +int +bare_ipc_poll_stop(bare_ipc_poll_t *poll) { + bare_ipc_signal_t *signal = (bare_ipc_signal_t *) poll->ipc->signal; + + bare_ipc__lock(signal); + + poll->events = 0; + poll->cb = NULL; + + bare_ipc__unlock(signal); + + return 0; +} diff --git a/shared/uv/ipc.h b/shared/uv/ipc.h new file mode 100644 index 0000000..1ede71d --- /dev/null +++ b/shared/uv/ipc.h @@ -0,0 +1,24 @@ +#ifndef BARE_KIT_UV_IPC_H +#define BARE_KIT_UV_IPC_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include "../ipc.h" + +struct bare_ipc_poll_s { + bare_ipc_t *ipc; + + int events; + + bare_ipc_poll_cb cb; + + void *data; +}; + +#ifdef __cplusplus +} +#endif + +#endif // BARE_KIT_UV_IPC_H diff --git a/shared/win32/ipc.c b/shared/win32/ipc.c deleted file mode 100644 index 099a517..0000000 --- a/shared/win32/ipc.c +++ /dev/null @@ -1,280 +0,0 @@ -#include -#include - -#include "ipc.h" - -void -bare_ipc__close(uv_async_t *handle) { - bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle); - - uv_close((uv_handle_t *) &ipc->pipe.incoming, NULL); - uv_close((uv_handle_t *) &ipc->pipe.outgoing, NULL); - uv_close((uv_handle_t *) &ipc->read, NULL); - uv_close((uv_handle_t *) &ipc->write, NULL); - uv_close((uv_handle_t *) &ipc->close, NULL); -} - -void -bare_ipc__on_alloc(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) { - bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data(handle); - buf->base = ipc->read_tmp_buffer; - buf->len = BARE_IPC_READ_BUFFER_SIZE; -} - -void -bare_ipc__on_read(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) { - // End-of-stream when the worklet closes its IPC. Stop reading rather than - // aborting; surfacing EOF to the host is not yet implemented for win32. - if (nread == UV_EOF) { - uv_read_stop(stream); - return; - } - - assert(nread >= 0); - - if (nread == 0) { - return; - } - - int err; - - err = uv_read_stop(stream); - assert(err == 0); - - bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) stream); - - uv_mutex_lock(&ipc->reading); - - memcpy(ipc->read_buffer.base, buf->base, nread); - ipc->read_buffer.len = nread; - - uv_mutex_unlock(&ipc->reading); - - if (ipc->poll && ipc->poll->cb) ipc->poll->cb(ipc->poll, bare_ipc_readable); -} - -void -bare_ipc__read(uv_async_t *handle) { - bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle); - - int err = uv_read_start((uv_stream_t *) &ipc->pipe.incoming, bare_ipc__on_alloc, bare_ipc__on_read); - assert(err == 0); -} - -void -bare_ipc__on_write(uv_write_t *req, int status) { - assert(status == 0); - - bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) req); - - free(ipc->write_buffer.base); - - uv_mutex_unlock(&ipc->writing); - - if (ipc->poll && ipc->poll->cb && (ipc->poll->events & bare_ipc_writable) != 0) ipc->poll->cb(ipc->poll, bare_ipc_writable); -} - -void -bare_ipc__write(uv_async_t *handle) { - int err; - - bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle); - - err = uv_write(&ipc->write_request, (uv_stream_t *) &ipc->pipe.outgoing, &ipc->write_buffer, 1, bare_ipc__on_write); - assert(err == 0); -} - -void -bare_ipc__loop(void *arg) { - bare_ipc_t *ipc = (bare_ipc_t *) arg; - - int err; - - err = uv_loop_init(&ipc->loop); - assert(err == 0); - - uv_handle_set_data((uv_handle_t *) &ipc->write_request, (void *) ipc); - - err = uv_async_init(&ipc->loop, &ipc->read, bare_ipc__read); - assert(err == 0); - uv_handle_set_data((uv_handle_t *) &ipc->read, (void *) ipc); - - err = uv_async_init(&ipc->loop, &ipc->write, bare_ipc__write); - assert(err == 0); - uv_handle_set_data((uv_handle_t *) &ipc->write, (void *) ipc); - - err = uv_async_init(&ipc->loop, &ipc->close, bare_ipc__close); - assert(err == 0); - uv_handle_set_data((uv_handle_t *) &ipc->close, (void *) ipc); - - err = uv_pipe_init(&ipc->loop, &ipc->pipe.incoming, 0); - assert(err == 0); - err = uv_pipe_open(&ipc->pipe.incoming, ipc->incoming); - assert(err == 0); - uv_handle_set_data((uv_handle_t *) &ipc->pipe.incoming, (void *) ipc); - - err = uv_pipe_init(&ipc->loop, &ipc->pipe.outgoing, 0); - assert(err == 0); - err = uv_pipe_open(&ipc->pipe.outgoing, ipc->outgoing); - assert(err == 0); - uv_handle_set_data((uv_handle_t *) &ipc->pipe.outgoing, (void *) ipc); - - uv_barrier_wait(&ipc->ready); - - err = uv_run(&ipc->loop, UV_RUN_DEFAULT); - assert(err == 0); - - err = uv_loop_close(&ipc->loop); - assert(err == 0); -} - -int -bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet) { - int err; - - ipc->incoming = _dup(worklet->incoming); - ipc->outgoing = _dup(worklet->outgoing); - - err = uv_mutex_init(&ipc->reading); - assert(err == 0); - - err = uv_mutex_init(&ipc->writing); - assert(err == 0); - - err = uv_barrier_init(&ipc->ready, 2); - assert(err == 0); - - err = uv_thread_create(&ipc->thread, bare_ipc__loop, ipc); - assert(err == 0); - - ipc->read_buffer.len = 0; - ipc->poll = NULL; - - uv_barrier_wait(&ipc->ready); - uv_barrier_destroy(&ipc->ready); - - return 0; -} - -void -bare_ipc_destroy(bare_ipc_t *ipc) { - int err; - - err = uv_async_send(&ipc->close); - assert(err == 0); - - err = uv_thread_join(&ipc->thread); - assert(err == 0); - - uv_mutex_destroy(&ipc->reading); - - uv_mutex_destroy(&ipc->writing); -} - -int -bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len) { - uv_mutex_lock(&ipc->reading); - - if (ipc->read_buffer.len > 0) { - *data = ipc->read_buffer.base; - *len = ipc->read_buffer.len; - ipc->read_buffer.len = 0; - - uv_mutex_unlock(&ipc->reading); - - return 0; - } - - uv_mutex_unlock(&ipc->reading); - - return bare_ipc_would_block; -} - -int -bare_ipc_write(bare_ipc_t *ipc, const void *data, size_t len) { - int err; - - if (uv_mutex_trylock(&ipc->writing) != 0) { - return bare_ipc_would_block; - } - - ipc->write_buffer.base = malloc(len); - memcpy(ipc->write_buffer.base, data, len); - ipc->write_buffer.len = len; - - err = uv_async_send(&ipc->write); - assert(err == 0); - - return len; -} - -int -bare_ipc_poll_alloc(bare_ipc_poll_t **result) { - bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t)); - - if (poll == NULL) return -1; - - *result = poll; - - return 0; -} - -int -bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) { - poll->ipc = ipc; - poll->events = 0; - poll->cb = NULL; - ipc->poll = poll; - - return 0; -} - -void -bare_ipc_poll_destroy(bare_ipc_poll_t *poll) { - int err; - - err = bare_ipc_poll_stop(poll); - assert(err == 0); -} - -void * -bare_ipc_poll_get_data(bare_ipc_poll_t *poll) { - return poll->data; -} - -void -bare_ipc_poll_set_data(bare_ipc_poll_t *poll, void *data) { - poll->data = data; -} - -bare_ipc_t * -bare_ipc_poll_get_ipc(bare_ipc_poll_t *poll) { - return poll->ipc; -} - -int -bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) { - if (events == 0) return bare_ipc_poll_stop(poll); - - int err; - - if ((events & bare_ipc_readable) != 0) { - if ((poll->events & bare_ipc_readable) == 0) { - err = uv_async_send(&poll->ipc->read); - assert(err == 0); - } - } - - poll->events = events; - poll->cb = cb; - - return 0; -} - -int -bare_ipc_poll_stop(bare_ipc_poll_t *poll) { - poll->events = 0; - poll->cb = NULL; - - return 0; -} diff --git a/shared/win32/ipc.h b/shared/win32/ipc.h deleted file mode 100644 index 0b73f26..0000000 --- a/shared/win32/ipc.h +++ /dev/null @@ -1,58 +0,0 @@ -#ifndef BARE_KIT_WIN32_IPC_H -#define BARE_KIT_WIN32_IPC_H - -#ifdef __cplusplus -extern "C" { -#endif - -#include - -#include "../ipc.h" - -struct bare_ipc_s { - int incoming; - int outgoing; - - struct { - char base[BARE_IPC_READ_BUFFER_SIZE]; - size_t len; - } read_buffer; - - char read_tmp_buffer[BARE_IPC_READ_BUFFER_SIZE]; - - uv_write_t write_request; - uv_buf_t write_buffer; - - struct { - uv_pipe_t incoming; - uv_pipe_t outgoing; - } pipe; - - uv_loop_t loop; - uv_async_t close; - uv_async_t read; - uv_async_t write; - - uv_barrier_t ready; - uv_mutex_t reading; - uv_mutex_t writing; - uv_thread_t thread; - - bare_ipc_poll_t *poll; -}; - -struct bare_ipc_poll_s { - bare_ipc_t *ipc; - - int events; - - bare_ipc_poll_cb cb; - - void *data; -}; - -#ifdef __cplusplus -} -#endif - -#endif // BARE_KIT_WIN32_IPC_H diff --git a/shared/worklet.c b/shared/worklet.c index f8bbbf7..4b252c8 100644 --- a/shared/worklet.c +++ b/shared/worklet.c @@ -87,21 +87,29 @@ bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options worklet->options.assets = options->assets == NULL ? NULL : strdup(options->assets); } - worklet->incoming = -1; - worklet->outgoing = -1; + worklet->queue = bare_queue_create(); + + if (worklet->queue == NULL) return -1; return 0; } void bare_worklet_destroy(bare_worklet_t *worklet) { - int err; - - if (worklet->thread != 0) uv_sem_post(worklet->finished); - else free(worklet->state); + if (worklet->thread != 0) { + // Stop the worklet if it is still running and wake its loop so the stop + // takes effect, then unblock it. The thread is detached and tears itself + // down; this function must never block (blocking here causes ANR on + // Android). bare_terminate is a no-op if the worklet already exited. + bare_terminate(worklet->bare); + uv_async_send(&worklet->queue->async); + uv_sem_post(worklet->finished); + } else { + // Nothing ever took ownership of the queue, so it is ours to free. + free(worklet->state); - if (worklet->incoming >= 0) close(worklet->incoming); - if (worklet->outgoing >= 0) close(worklet->outgoing); + bare_queue_destroy(worklet->queue); + } free((char *) worklet->options.assets); } @@ -352,6 +360,132 @@ bare_worklet__on_resume(bare_t *bare, void *data) { if (state->callbacks.resume) state->callbacks.resume(bare, state->callbacks.resume_data); } +// native.read() -> ArrayBuffer (data) | null (EOF) | undefined (would block) +static js_value_t * +bare_worklet__on_ipc_read(js_env_t *env, js_callback_info_t *info) { + int err; + + bare_queue_port_t *port; + err = js_get_callback_info(env, info, NULL, NULL, NULL, (void **) &port); + assert(err == 0); + + void *data; + size_t len; + int status = bare_queue_read(port, &data, &len); + + js_value_t *result; + + if (status == bare_queue_would_block) { + err = js_get_undefined(env, &result); + assert(err == 0); + } else if (len == 0) { + err = js_get_null(env, &result); + assert(err == 0); + } else { + void *copy; + err = js_create_arraybuffer(env, len, ©, &result); + assert(err == 0); + + memcpy(copy, data, len); + } + + return result; +} + +// native.write(buffer) -> bytes written, or a negative queue status +static js_value_t * +bare_worklet__on_ipc_write(js_env_t *env, js_callback_info_t *info) { + int err; + + size_t argc = 1; + js_value_t *argv[1]; + + bare_queue_port_t *port; + err = js_get_callback_info(env, info, &argc, argv, NULL, (void **) &port); + assert(err == 0); + assert(argc == 1); + + void *data; + size_t len; + err = js_get_typedarray_info(env, argv[0], NULL, &data, &len, NULL, NULL); + assert(err == 0); + + int n = bare_queue_write(port, data, len); + + js_value_t *result; + err = js_create_int32(env, n, &result); + assert(err == 0); + + return result; +} + +static js_value_t * +bare_worklet__on_ipc_close(js_env_t *env, js_callback_info_t *info) { + int err; + + bare_queue_port_t *port; + err = js_get_callback_info(env, info, NULL, NULL, NULL, (void **) &port); + assert(err == 0); + + bare_queue_close(port); + + return NULL; +} + +static js_value_t * +bare_worklet__on_ipc_ref(js_env_t *env, js_callback_info_t *info) { + int err; + + bare_queue_port_t *port; + err = js_get_callback_info(env, info, NULL, NULL, NULL, (void **) &port); + assert(err == 0); + + uv_ref((uv_handle_t *) &port->queue->async); + + return NULL; +} + +static js_value_t * +bare_worklet__on_ipc_unref(js_env_t *env, js_callback_info_t *info) { + int err; + + bare_queue_port_t *port; + err = js_get_callback_info(env, info, NULL, NULL, NULL, (void **) &port); + assert(err == 0); + + uv_unref((uv_handle_t *) &port->queue->async); + + return NULL; +} + +// Runs on the worklet loop when the host has touched the queue; drives the JS +// duplex to re-check reads and any pending write. +static void +bare_worklet__on_ipc_signal(bare_queue_port_t *port) { + int err; + + bare_worklet_t *worklet = (bare_worklet_t *) port->data; + + js_env_t *env = worklet->env; + + js_handle_scope_t *scope; + err = js_open_handle_scope(env, &scope); + assert(err == 0); + + js_value_t *fn; + err = js_get_reference_value(env, worklet->ipc_signal, &fn); + assert(err == 0); + + js_value_t *global; + err = js_get_global(env, &global); + assert(err == 0); + + js_call_function(env, global, fn, 0, NULL, NULL); + + err = js_close_handle_scope(env, scope); + assert(err == 0); +} + static void bare_worklet__on_thread(void *opaque) { uv_once(&bare_worklet__platform_guard, bare_worklet__on_platform_init); @@ -416,24 +550,50 @@ bare_worklet__on_thread(void *opaque) { err = js_get_named_property(env, module, "exports", &exports); assert(err == 0); - js_value_t *port; - err = js_get_named_property(env, exports, "port", &port); - assert(err == 0); + worklet->env = env; + + // Held for the rest of this function: the host may destroy its worklet, and + // with it every field read from it, the moment this thread hands back control. + bare_queue_t *queue = worklet->queue; + + bare_queue_port_t *ipc_port = bare_queue_open_uv(queue, &loop, bare_worklet__on_ipc_signal); - js_value_t *incoming; - err = js_get_named_property(env, port, "incoming", &incoming); + ipc_port->data = worklet; + + js_value_t *native; + err = js_create_object(env, &native); assert(err == 0); - err = js_get_value_int32(env, incoming, &worklet->incoming); +#define BARE_WORKLET_IPC_FN(name, cb) \ + do { \ + js_value_t *fn; \ + err = js_create_function(env, name, -1, cb, ipc_port, &fn); \ + assert(err == 0); \ + err = js_set_named_property(env, native, name, fn); \ + assert(err == 0); \ + } while (0) + + BARE_WORKLET_IPC_FN("read", bare_worklet__on_ipc_read); + BARE_WORKLET_IPC_FN("write", bare_worklet__on_ipc_write); + BARE_WORKLET_IPC_FN("close", bare_worklet__on_ipc_close); + BARE_WORKLET_IPC_FN("ref", bare_worklet__on_ipc_ref); + BARE_WORKLET_IPC_FN("unref", bare_worklet__on_ipc_unref); + +#undef BARE_WORKLET_IPC_FN + + js_value_t *open_ipc; + err = js_get_named_property(env, exports, "openIPC", &open_ipc); assert(err == 0); - js_value_t *outgoing; - err = js_get_named_property(env, port, "outgoing", &outgoing); + js_value_t *on_signal; + err = js_call_function(env, exports, open_ipc, 1, &native, &on_signal); assert(err == 0); - err = js_get_value_int32(env, outgoing, &worklet->outgoing); + err = js_create_reference(env, on_signal, 1, &worklet->ipc_signal); assert(err == 0); + js_ref_t *ipc_signal = worklet->ipc_signal; + js_value_t *fn; err = js_get_named_property(env, exports, "push", &fn); assert(err == 0); @@ -489,6 +649,11 @@ bare_worklet__on_thread(void *opaque) { state->finished = true; + // Signal EOF to the host by closing our end of the queue. This is in-process + // (a status flag plus a host wakeup), so it works on every exit path without + // pumping the loop or scheduling async work from the exit event. + bare_queue_close(ipc_port); + uv_sem_wait(&finished); uv_sem_destroy(&finished); @@ -496,6 +661,13 @@ bare_worklet__on_thread(void *opaque) { err = js_release_threadsafe_function(push, js_threadsafe_function_release); assert(err == 0); + // Closing the async here stops further signals; teardown's loop run flushes + // it. Do this before deleting the reference so no late signal dereferences it. + bare_queue_shutdown_uv(queue); + + err = js_delete_reference(env, ipc_signal); + assert(err == 0); + int exit_code; err = bare_teardown(bare, UV_RUN_DEFAULT, &exit_code); assert(err == 0); @@ -503,6 +675,10 @@ bare_worklet__on_thread(void *opaque) { err = uv_loop_close(&loop); assert(err == 0); + // Ours since the host started us, and nothing may reach it now: the host was + // required to retire its end before signalling us to be destroyed. + bare_queue_destroy(queue); + err = bare_suspension_end(&state->suspension); assert(err == 0); @@ -536,9 +712,6 @@ bare_worklet_start(bare_worklet_t *worklet, const char *filename, const uv_buf_t return err; } - err = uv_thread_detach(&worklet->thread); - assert(err == 0); - uv_barrier_wait(&worklet->ready); uv_barrier_destroy(&worklet->ready); diff --git a/shared/worklet.h b/shared/worklet.h index 65d0551..5bfeb66 100644 --- a/shared/worklet.h +++ b/shared/worklet.h @@ -10,6 +10,7 @@ extern "C" { #include #include +#include "queue.h" #include "suspension.h" typedef struct bare_worklet_s bare_worklet_t; @@ -66,8 +67,10 @@ struct bare_worklet_s { uv_barrier_t ready; uv_sem_t *finished; - uv_file incoming; - uv_file outgoing; + bare_queue_t *queue; + + js_env_t *env; + js_ref_t *ipc_signal; js_threadsafe_function_t *push; @@ -97,6 +100,9 @@ bare_worklet_alloc(bare_worklet_t **result); int bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options); +// Signals the worklet to stop and returns immediately; it never blocks, and the +// thread tears itself down. Any ipc opened on this worklet must already have been +// destroyed, because from here the queue is the worklet's alone. void bare_worklet_destroy(bare_worklet_t *worklet); diff --git a/shared/worklet.js b/shared/worklet.js index b68c79c..4843915 100644 --- a/shared/worklet.js +++ b/shared/worklet.js @@ -9,26 +9,97 @@ const Module = require('bare-module') const { startsWithWindowsDriveLetter } = require('bare-module-resolve') const SystemLog = require('bare-system-logger') const Console = require('bare-console') -const IPC = require('bare-ipc') +const { Duplex } = require('bare-stream') const unpack = require('bare-unpack') global.console = new Console(new SystemLog()) -const ports = IPC.open() +let ipc = null -const ipc = new IPC(ports[0]) +// A Duplex over the native in-process queue. The native side provides +// non-blocking read/write plus ref/unref; a single signal (fed back from the +// worklet loop) means re-check both reads and a pending write. +class IPCStream extends Duplex { + constructor(native) { + super() + + this._native = native + this._reading = false + this._pendingWrite = null + } + + _read() { + this._reading = true + this._drain() + } + + _drain() { + if (!this._reading) return + + let chunk + while ((chunk = this._native.read()) !== undefined) { + if (chunk === null) { + this._reading = false + this.push(null) + return + } + + if (this.push(Buffer.from(chunk)) === false) { + this._reading = false + return + } + } + } + + _write(chunk, encoding, cb) { + this._send(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding), cb) + } + + _send(chunk, cb) { + const n = this._native.write(chunk) + + if (n < 0) this._pendingWrite = { chunk, cb } + else if (n < chunk.byteLength) this._pendingWrite = { chunk: chunk.subarray(n), cb } + else cb(null) + } -Bare.IPC = ipc + _retry() { + if (this._pendingWrite === null) return -Bare.on('suspend', () => ipc.unref()) - .on('resume', () => ipc.ref()) - .on('exit', () => ipc.destroy()) + const { chunk, cb } = this._pendingWrite + this._pendingWrite = null + + this._send(chunk, cb) + } + + // end() closes our end of the queue so the host read reaches EOF. streamx + // only calls _final once outstanding writes have drained, so ordering holds. + _final(cb) { + this._native.close() + cb(null) + } + + _onsignal() { + this._retry() + this._drain() + } + + ref() { + this._native.ref() + return this + } + + unref() { + this._native.unref() + return this + } +} class BareKit extends EventEmitter { constructor() { super() - this.IPC = ipc + this.IPC = null } [Symbol.for('bare.inspect')]() { @@ -39,7 +110,6 @@ class BareKit extends EventEmitter { } exports.BareKit = new BareKit() -exports.port = ports[1] Object.defineProperty(global, 'BareKit', { value: exports.BareKit, @@ -48,6 +118,19 @@ Object.defineProperty(global, 'BareKit', { configurable: true }) +// Called by the host runtime with the native queue endpoint before `start`. +// Returns the signal callback the worklet loop invokes when the queue changes. +exports.openIPC = function openIPC(native) { + ipc = new IPCStream(native) + + Bare.IPC = ipc + exports.BareKit.IPC = ipc + + Bare.on('suspend', () => ipc.unref()).on('resume', () => ipc.ref()) + + return () => ipc._onsignal() +} + exports.push = function push(payload, reply) { if (exports.BareKit.emit('push', Buffer.from(payload), replyOnce) === false) { replyOnce(null, null) diff --git a/test/shared/CMakeLists.txt b/test/shared/CMakeLists.txt index 246b817..50a4777 100644 --- a/test/shared/CMakeLists.txt +++ b/test/shared/CMakeLists.txt @@ -4,21 +4,14 @@ list(APPEND tests queue-signal-lock worklet-assets worklet-destroy-before-start + worklet-exit-eof worklet-ipc worklet-ipc-large-write + worklet-jsend-eof worklet-suspend-resume worklet-terminate-after-suspend ) -# The win32 IPC layer does not yet surface end-of-stream, so the EOF tests only -# run on platforms with the posix read path. -if(NOT WIN32) - list(APPEND tests - worklet-exit-eof - worklet-jsend-eof - ) -endif() - execute_process( COMMAND node fixtures/generate.js WORKING_DIRECTORY ${CMAKE_CURRENT_LIST_DIR} diff --git a/test/shared/worklet-ipc-large-write.c b/test/shared/worklet-ipc-large-write.c index 803ba00..5e6862a 100644 --- a/test/shared/worklet-ipc-large-write.c +++ b/test/shared/worklet-ipc-large-write.c @@ -103,14 +103,16 @@ main() { err = uv_run(loop, UV_RUN_DEFAULT); assert(err == 0); + // Before the loop closes: the worklet's exit closes its end of the queue, and + // that wakes the poll - whose callback posts to this loop. + bare_ipc_poll_destroy(&poll); + err = uv_loop_close(loop); assert(err == 0); err = bare_worklet_terminate(&worklet); assert(err == 0); - bare_ipc_poll_destroy(&poll); - bare_ipc_destroy(&ipc); bare_worklet_destroy(&worklet); diff --git a/test/shared/worklet-ipc.c b/test/shared/worklet-ipc.c index 667508d..60b11ee 100644 --- a/test/shared/worklet-ipc.c +++ b/test/shared/worklet-ipc.c @@ -138,14 +138,16 @@ main() { assert(err == 0); } + // Before the loop closes: the worklet's exit closes its end of the queue, and + // that wakes the poll - whose callback posts to this loop. + bare_ipc_poll_destroy(&poll); + err = uv_loop_close(loop); assert(err == 0); err = bare_worklet_terminate(&worklet); assert(err == 0); - bare_ipc_poll_destroy(&poll); - bare_ipc_destroy(&ipc); bare_worklet_destroy(&worklet); diff --git a/test/shared/worklet-jsend-eof.c b/test/shared/worklet-jsend-eof.c index 26fc3ac..622e475 100644 --- a/test/shared/worklet-jsend-eof.c +++ b/test/shared/worklet-jsend-eof.c @@ -3,9 +3,9 @@ #include "../../shared/ipc.h" #include "../../shared/worklet.h" -// Corroborates worklet-exit-no-eof: the host EOF comes from the JS side. When -// the worklet ends its Bare.IPC (closing the b1 write end), the host read -// returns 0 (EOF) - the signal the bindings turn into (nil, nil). +// The host EOF comes from the JS side: when the worklet ends its Bare.IPC, its +// end of the in-process queue closes and the host read returns 0 (EOF) - the +// signal the bindings turn into (nil, nil). int main() { @@ -29,7 +29,7 @@ main() { void *data; size_t len; - // Poll for up to ~2s for EOF. Expect read to return 0 once b1 closes. + // Poll for up to ~2s for EOF. Expect read to return 0 once the worklet ends. bool eof = false; for (int i = 0; i < 200; i++) { e = bare_ipc_read(&ipc, &data, &len); diff --git a/test/shared/worklet-suspend-resume.c b/test/shared/worklet-suspend-resume.c index 6e3ae14..913622e 100644 --- a/test/shared/worklet-suspend-resume.c +++ b/test/shared/worklet-suspend-resume.c @@ -31,7 +31,8 @@ main() { e = bare_worklet_on_resume(&worklet, on_resume, NULL); assert(e == 0); - char *code = "console.log('Hello world')"; + // Keep Bare.IPC open so suspend/resume exercise the queue's ref/unref. + char *code = "BareKit.IPC.on('data', () => {})"; uv_buf_t source = uv_buf_init(code, strlen(code)); diff --git a/win32/ipc.def b/win32/ipc.def index 6085502..db25a59 100644 --- a/win32/ipc.def +++ b/win32/ipc.def @@ -2,8 +2,6 @@ EXPORTS bare_ipc_alloc bare_ipc_init bare_ipc_destroy - bare_ipc_get_incoming - bare_ipc_get_outgoing bare_ipc_read bare_ipc_write bare_ipc_poll_alloc From 2028251a069af9277bb68184dbe7f1968f4d94e9 Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Fri, 11 Sep 2026 12:52:42 +0530 Subject: [PATCH 02/11] Update shared/worklet.c MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Kasper Isager Dalsgarð --- shared/worklet.c | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/shared/worklet.c b/shared/worklet.c index 4b252c8..4df1086 100644 --- a/shared/worklet.c +++ b/shared/worklet.c @@ -564,22 +564,21 @@ bare_worklet__on_thread(void *opaque) { err = js_create_object(env, &native); assert(err == 0); -#define BARE_WORKLET_IPC_FN(name, cb) \ - do { \ +#define V(name, cb) \ + { \ js_value_t *fn; \ err = js_create_function(env, name, -1, cb, ipc_port, &fn); \ assert(err == 0); \ err = js_set_named_property(env, native, name, fn); \ assert(err == 0); \ - } while (0) - - BARE_WORKLET_IPC_FN("read", bare_worklet__on_ipc_read); - BARE_WORKLET_IPC_FN("write", bare_worklet__on_ipc_write); - BARE_WORKLET_IPC_FN("close", bare_worklet__on_ipc_close); - BARE_WORKLET_IPC_FN("ref", bare_worklet__on_ipc_ref); - BARE_WORKLET_IPC_FN("unref", bare_worklet__on_ipc_unref); + } -#undef BARE_WORKLET_IPC_FN + V("read", bare_worklet__on_ipc_read); + V("write", bare_worklet__on_ipc_write); + V("close", bare_worklet__on_ipc_close); + V("ref", bare_worklet__on_ipc_ref); + V("unref", bare_worklet__on_ipc_unref); +#undef V js_value_t *open_ipc; err = js_get_named_property(env, exports, "openIPC", &open_ipc); From 4c445ca1b4126084bc2c71097edb5dedab7db051 Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Thu, 17 Sep 2026 23:11:34 +0530 Subject: [PATCH 03/11] Detach the worklet thread again The thread must stay detached: destroy is never allowed to block, and joining here has historically caused ANR errors on Android. --- shared/worklet.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/shared/worklet.c b/shared/worklet.c index 3102d1d..304ccb6 100644 --- a/shared/worklet.c +++ b/shared/worklet.c @@ -714,6 +714,9 @@ bare_worklet_start(bare_worklet_t *worklet, const char *filename, const uv_buf_t return err; } + err = uv_thread_detach(&worklet->thread); + assert(err == 0); + uv_barrier_wait(&worklet->ready); uv_barrier_destroy(&worklet->ready); From 09097a41ef49e1bcc78dd3d99343e46bdbfb0906 Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Thu, 17 Sep 2026 23:12:39 +0530 Subject: [PATCH 04/11] Keep the worklet's IPC signal state out of the host's worklet The signal callback read env and ipc_signal from the host's worklet through port->data, so a queue wakeup still in flight at teardown dereferenced storage the host had already released, crashing in js_open_handle_scope. The worklet thread now owns those two fields and frees them once the queue can no longer signal. Fixes the intermittent worklet-ipc segfault on darwin-x64 and win32-arm64. --- shared/worklet.c | 32 ++++++++++++++++++++++---------- shared/worklet.h | 3 --- 2 files changed, 22 insertions(+), 13 deletions(-) diff --git a/shared/worklet.c b/shared/worklet.c index 304ccb6..3045432 100644 --- a/shared/worklet.c +++ b/shared/worklet.c @@ -459,22 +459,29 @@ bare_worklet__on_ipc_unref(js_env_t *env, js_callback_info_t *info) { return NULL; } +// Owned by the worklet thread, so the signal callback never dereferences the +// host's worklet, which may be destroyed while the queue is still open. +typedef struct { + js_env_t *env; + js_ref_t *signal; +} bare_worklet__ipc_t; + // Runs on the worklet loop when the host has touched the queue; drives the JS // duplex to re-check reads and any pending write. static void bare_worklet__on_ipc_signal(bare_queue_port_t *port) { int err; - bare_worklet_t *worklet = (bare_worklet_t *) port->data; + bare_worklet__ipc_t *ipc = (bare_worklet__ipc_t *) port->data; - js_env_t *env = worklet->env; + js_env_t *env = ipc->env; js_handle_scope_t *scope; err = js_open_handle_scope(env, &scope); assert(err == 0); js_value_t *fn; - err = js_get_reference_value(env, worklet->ipc_signal, &fn); + err = js_get_reference_value(env, ipc->signal, &fn); assert(err == 0); js_value_t *global; @@ -553,15 +560,20 @@ bare_worklet__on_thread(void *opaque) { err = js_get_named_property(env, module, "exports", &exports); assert(err == 0); - worklet->env = env; - // Held for the rest of this function: the host may destroy its worklet, and // with it every field read from it, the moment this thread hands back control. bare_queue_t *queue = worklet->queue; + // Ours, not the host's: the signal callback runs until we shut the queue down, + // which is after the host is free to have destroyed its worklet. + bare_worklet__ipc_t *ipc = malloc(sizeof(bare_worklet__ipc_t)); + + ipc->env = env; + ipc->signal = NULL; + bare_queue_port_t *ipc_port = bare_queue_open_uv(queue, &loop, bare_worklet__on_ipc_signal); - ipc_port->data = worklet; + ipc_port->data = ipc; js_value_t *native; err = js_create_object(env, &native); @@ -591,11 +603,9 @@ bare_worklet__on_thread(void *opaque) { err = js_call_function(env, exports, open_ipc, 1, &native, &on_signal); assert(err == 0); - err = js_create_reference(env, on_signal, 1, &worklet->ipc_signal); + err = js_create_reference(env, on_signal, 1, &ipc->signal); assert(err == 0); - js_ref_t *ipc_signal = worklet->ipc_signal; - js_value_t *fn; err = js_get_named_property(env, exports, "push", &fn); assert(err == 0); @@ -667,9 +677,11 @@ bare_worklet__on_thread(void *opaque) { // it. Do this before deleting the reference so no late signal dereferences it. bare_queue_shutdown_uv(queue); - err = js_delete_reference(env, ipc_signal); + err = js_delete_reference(env, ipc->signal); assert(err == 0); + free(ipc); + int exit_code; err = bare_teardown(bare, UV_RUN_DEFAULT, &exit_code); assert(err == 0); diff --git a/shared/worklet.h b/shared/worklet.h index 5bfeb66..cc8fe61 100644 --- a/shared/worklet.h +++ b/shared/worklet.h @@ -69,9 +69,6 @@ struct bare_worklet_s { bare_queue_t *queue; - js_env_t *env; - js_ref_t *ipc_signal; - js_threadsafe_function_t *push; void *data; From 1c89970e1bff67a14712006f90860686f7dc99c0 Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Thu, 17 Sep 2026 23:13:00 +0530 Subject: [PATCH 05/11] Error the IPC stream when the host has closed its end A write to a closed peer was treated as a would-block and retried forever, so the write callback never fired, _final never ran, and the duplex wedged while holding a ref'd async. The native write now reports a closed peer as null, the way read already reports EOF, and the stream errors. --- shared/worklet.c | 10 ++- shared/worklet.js | 11 +++- test/shared/CMakeLists.txt | 1 + test/shared/worklet-ipc-write-after-close.c | 70 +++++++++++++++++++++ 4 files changed, 89 insertions(+), 3 deletions(-) create mode 100644 test/shared/worklet-ipc-write-after-close.c diff --git a/shared/worklet.c b/shared/worklet.c index 3045432..bda2e6c 100644 --- a/shared/worklet.c +++ b/shared/worklet.c @@ -393,7 +393,8 @@ bare_worklet__on_ipc_read(js_env_t *env, js_callback_info_t *info) { return result; } -// native.write(buffer) -> bytes written, or a negative queue status +// native.write(buffer) -> bytes written, -1 when the queue is full, or null +// once the peer has closed static js_value_t * bare_worklet__on_ipc_write(js_env_t *env, js_callback_info_t *info) { int err; @@ -414,7 +415,12 @@ bare_worklet__on_ipc_write(js_env_t *env, js_callback_info_t *info) { int n = bare_queue_write(port, data, len); js_value_t *result; - err = js_create_int32(env, n, &result); + + if (n == bare_queue_closed) { + err = js_get_null(env, &result); + } else { + err = js_create_int32(env, n, &result); + } assert(err == 0); return result; diff --git a/shared/worklet.js b/shared/worklet.js index 74dd8cf..23c4140 100644 --- a/shared/worklet.js +++ b/shared/worklet.js @@ -16,6 +16,14 @@ const { protocol } = module global.console = new Console(new SystemLog()) +const errors = { + IPC_CLOSED() { + const err = new Error('IPC stream closed') + err.code = 'IPC_CLOSED' + return err + } +} + // A Duplex over the native in-process queue. The native side provides // non-blocking read/write plus ref/unref; a single signal (fed back from the // worklet loop) means re-check both reads and a pending write. @@ -58,7 +66,8 @@ class IPCStream extends Duplex { _send(chunk, cb) { const n = this._native.write(chunk) - if (n < 0) this._pendingWrite = { chunk, cb } + if (n === null) cb(errors.IPC_CLOSED()) + else if (n < 0) this._pendingWrite = { chunk, cb } else if (n < chunk.byteLength) this._pendingWrite = { chunk: chunk.subarray(n), cb } else cb(null) } diff --git a/test/shared/CMakeLists.txt b/test/shared/CMakeLists.txt index 50a4777..d5451a8 100644 --- a/test/shared/CMakeLists.txt +++ b/test/shared/CMakeLists.txt @@ -7,6 +7,7 @@ list(APPEND tests worklet-exit-eof worklet-ipc worklet-ipc-large-write + worklet-ipc-write-after-close worklet-jsend-eof worklet-suspend-resume worklet-terminate-after-suspend diff --git a/test/shared/worklet-ipc-write-after-close.c b/test/shared/worklet-ipc-write-after-close.c new file mode 100644 index 0000000..6fa719e --- /dev/null +++ b/test/shared/worklet-ipc-write-after-close.c @@ -0,0 +1,70 @@ +#include +#include +#include +#include + +#include "../../shared/ipc.h" +#include "../../shared/worklet.h" + +// The host retires its IPC while the worklet still has a write outstanding. The +// worklet's queue end then reports closed, which must surface as a stream error +// rather than a write that never completes; the worklet reports which it saw. + +static bool replied = false; +static char reply_buf[64]; + +static void +on_push(bare_worklet_push_t *req, const char *error, const uv_buf_t *reply) { + assert(error == NULL); + + memcpy(reply_buf, reply->base, reply->len); + reply_buf[reply->len] = '\0'; + + replied = true; +} + +int +main() { + int e; + + bare_worklet_t worklet; + e = bare_worklet_init(&worklet, NULL); + assert(e == 0); + + char *code = + "let errored = false\n" + "Bare.IPC.on('error', () => { errored = true })\n" + "setInterval(() => Bare.IPC.write(Buffer.alloc(64)), 5)\n" + "BareKit.on('push', (data, reply) => reply(null, errored ? 'errored' : 'wedged'))\n"; + + uv_buf_t source = uv_buf_init(code, strlen(code)); + + e = bare_worklet_start(&worklet, "app.js", &source, 0, NULL); + assert(e == 0); + + bare_ipc_t ipc; + e = bare_ipc_init(&ipc, &worklet); + assert(e == 0); + + // Never read, so the queue fills and the worklet is left with a pending write. + uv_sleep(100); + + bare_ipc_destroy(&ipc); + + // Give the worklet time to retry that write against the closed peer. + uv_sleep(100); + + uv_buf_t payload = uv_buf_init("status", 6); + + bare_worklet_push_t req; + e = bare_worklet_push(&worklet, &req, &payload, on_push); + assert(e == 0); + + for (int i = 0; i < 200 && !replied; i++) + uv_sleep(10); + + assert(replied); + assert(strcmp(reply_buf, "errored") == 0); // not "wedged": the write completed + + bare_worklet_destroy(&worklet); +} From eb07ca1bcd05a3a65a9503a96dc33e9a858e9254 Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Fri, 18 Sep 2026 13:19:55 +0530 Subject: [PATCH 06/11] Assert the IPC signal reference before using it The callback only runs once the reference exists, so make that ordering explicit rather than implicit, and give the closed-peer write longer to be retried so the test does not flake on a loaded runner. Co-authored-by: Tony Gorez <22824417+tony-go@users.noreply.github.com> --- shared/worklet.c | 2 ++ test/shared/worklet-ipc-write-after-close.c | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/shared/worklet.c b/shared/worklet.c index bda2e6c..cac42b2 100644 --- a/shared/worklet.c +++ b/shared/worklet.c @@ -480,6 +480,8 @@ bare_worklet__on_ipc_signal(bare_queue_port_t *port) { bare_worklet__ipc_t *ipc = (bare_worklet__ipc_t *) port->data; + assert(ipc->signal); + js_env_t *env = ipc->env; js_handle_scope_t *scope; diff --git a/test/shared/worklet-ipc-write-after-close.c b/test/shared/worklet-ipc-write-after-close.c index 6fa719e..a20303f 100644 --- a/test/shared/worklet-ipc-write-after-close.c +++ b/test/shared/worklet-ipc-write-after-close.c @@ -52,7 +52,7 @@ main() { bare_ipc_destroy(&ipc); // Give the worklet time to retry that write against the closed peer. - uv_sleep(100); + uv_sleep(300); uv_buf_t payload = uv_buf_init("status", 6); From 63843895de7b8974d765e75343e8f29322be69a9 Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Tue, 22 Sep 2026 16:43:01 +0530 Subject: [PATCH 07/11] Fix the worklet init error paths Nothing unwound when an allocation failed, leaking the state and the assets copy. Move the options copy after the queue, so the only failure path left frees the state. Co-Authored-By: Claude Opus 5 (1M context) --- shared/worklet.c | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/shared/worklet.c b/shared/worklet.c index cac42b2..dd22aa6 100644 --- a/shared/worklet.c +++ b/shared/worklet.c @@ -69,16 +69,24 @@ int bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options) { uv_once(&bare_worklet__init_guard, bare_worklet__on_init); - int err; - worklet->thread = 0; bare_worklet_state_t *state = malloc(sizeof(bare_worklet_state_t)); + if (state == NULL) return -1; + state->finished = false; memset(&state->callbacks, 0, sizeof(state->callbacks)); + worklet->queue = bare_queue_create(); + + if (worklet->queue == NULL) { + free(state); + + return -1; + } + worklet->state = state; memset(&worklet->options, 0, sizeof(worklet->options)); @@ -88,10 +96,6 @@ bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options worklet->options.assets = options->assets == NULL ? NULL : strdup(options->assets); } - worklet->queue = bare_queue_create(); - - if (worklet->queue == NULL) return -1; - return 0; } From 63d2cb4344793bbab7978fff9597173c6698d7ed Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Tue, 22 Sep 2026 16:43:35 +0530 Subject: [PATCH 08/11] Align the IPC poll teardown across platforms Apple left the requested events set and uv cleared the poll state by hand. Both now stop the poll first, as android already did, so the three implementations differ only in how they wait out a callback in flight. Co-Authored-By: Claude Opus 5 (1M context) --- shared/apple/ipc.m | 6 ++++-- shared/uv/ipc.c | 7 ++++--- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/shared/apple/ipc.m b/shared/apple/ipc.m index ebca686..322d54a 100644 --- a/shared/apple/ipc.m +++ b/shared/apple/ipc.m @@ -119,9 +119,11 @@ void bare_ipc_poll_destroy(bare_ipc_poll_t *poll) { - bare_ipc_t *ipc = poll->ipc; + int err; + err = bare_ipc_poll_stop(poll); + assert(err == 0); - atomic_store(&poll->cb, NULL); + bare_ipc_t *ipc = poll->ipc; ipc->poll = NULL; diff --git a/shared/uv/ipc.c b/shared/uv/ipc.c index 99b328d..02b5f73 100644 --- a/shared/uv/ipc.c +++ b/shared/uv/ipc.c @@ -178,6 +178,10 @@ bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) { void bare_ipc_poll_destroy(bare_ipc_poll_t *poll) { + int err; + err = bare_ipc_poll_stop(poll); + assert(err == 0); + bare_ipc_t *ipc = poll->ipc; bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal; @@ -185,9 +189,6 @@ bare_ipc_poll_destroy(bare_ipc_poll_t *poll) { // Under the lock, so the caller may free the poll as soon as this returns. bare_ipc__lock(signal); - poll->events = 0; - poll->cb = NULL; - ipc->poll = NULL; bare_ipc__unlock(signal); From 775b86232cdc33c09d931dc15c56722bba81adce Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Tue, 22 Sep 2026 16:43:36 +0530 Subject: [PATCH 09/11] Swap the bundled IPC addons for bare-stream The worklet bootstrap builds its Duplex on bare-stream, which was never declared, and no longer requires bare-ipc or the bare-pipe it drags in. Breaking: app code in a worklet can no longer require bare-ipc or bare-pipe. Co-Authored-By: Claude Opus 5 (1M context) --- package-lock.json | 20 ++++++++------------ package.json | 2 +- shared/builtins.json | 2 -- 3 files changed, 9 insertions(+), 15 deletions(-) diff --git a/package-lock.json b/package-lock.json index da83fa2..dc38df6 100644 --- a/package-lock.json +++ b/package-lock.json @@ -12,11 +12,11 @@ "bare-crypto": "^1.2.8", "bare-events": "^2.4.2", "bare-fs": "^4.0.1", - "bare-ipc": "^1.0.1", "bare-logger": "^2.0.0", "bare-module": "^7.0.1", "bare-module-resolve": "^1.6.2", "bare-path": "^3.0.0", + "bare-stream": "^2.13.4", "bare-system-logger": "^1.0.1", "bare-unpack": "^1.0.2", "bare-url": "^2.0.1" @@ -101,6 +101,7 @@ "version": "3.7.1", "resolved": "https://registry.npmjs.org/bare-buffer/-/bare-buffer-3.7.1.tgz", "integrity": "sha512-cIjZnSO+y89ykVX9x96OVnA6tGORBn7vGJ/e3PCbBHMEyfMNG/ixzOS0lLa6saMlWcs69bSKjxFyDwjK152OMg==", + "devOptional": true, "license": "Apache-2.0", "engines": { "bare": ">=1.20.0" @@ -302,17 +303,6 @@ "bare": ">=1.18.0" } }, - "node_modules/bare-ipc": { - "version": "1.1.2", - "resolved": "https://registry.npmjs.org/bare-ipc/-/bare-ipc-1.1.2.tgz", - "integrity": "sha512-z2VXX9nHkWTgtiOwDJpuZa3RJeUDqG1VpGJJttVHEdqX9ebX09Bf4K9/RwYeNKerpFJ9tJP0kjfTjk6bJC9wXw==", - "license": "Apache-2.0", - "dependencies": { - "bare-pipe": "^4.0.0", - "bare-stream": "^2.1.3", - "bare-structured-clone": "^1.4.0" - } - }, "node_modules/bare-logger": { "version": "2.0.4", "resolved": "https://registry.npmjs.org/bare-logger/-/bare-logger-2.0.4.tgz", @@ -469,7 +459,9 @@ "version": "4.3.1", "resolved": "https://registry.npmjs.org/bare-pipe/-/bare-pipe-4.3.1.tgz", "integrity": "sha512-3P4MYTgOys1Bbz5dgWzwfroNOOl8x1zWJHTHVlwsojThhV/uJPRf+qFrYYBOnI0vQiJJLJtHlw8a4Osj/kcsJQ==", + "dev": true, "license": "Apache-2.0", + "optional": true, "dependencies": { "bare-events": "^2.5.4", "bare-stream": "^2.6.4" @@ -576,7 +568,9 @@ "version": "1.6.0", "resolved": "https://registry.npmjs.org/bare-structured-clone/-/bare-structured-clone-1.6.0.tgz", "integrity": "sha512-AZjEERyqF7kAuudlmgrweT2KwwWM0LsGG4Gx/bWyFwJrKU7E3wNG8c09z2DIrdw93p3vDtfOX8fyHOcXfy5m+g==", + "dev": true, "license": "Apache-2.0", + "optional": true, "dependencies": { "bare-buffer": "^3.6.0", "bare-type": "^1.1.0", @@ -801,7 +795,9 @@ "version": "3.5.0", "resolved": "https://registry.npmjs.org/compact-encoding/-/compact-encoding-3.5.0.tgz", "integrity": "sha512-X7yaWh0NNMVB2YXAWSTWjYU3WXvDTniKxbtQlisbXOVutIE6awd+zCQdNqScTjIXv9m9orqdlri+oaWWsKqTuA==", + "dev": true, "license": "Apache-2.0", + "optional": true, "dependencies": { "b4a": "^1.3.0" } diff --git a/package.json b/package.json index 419d34e..acac130 100644 --- a/package.json +++ b/package.json @@ -22,11 +22,11 @@ "bare-crypto": "^1.2.8", "bare-events": "^2.4.2", "bare-fs": "^4.0.1", - "bare-ipc": "^1.0.1", "bare-logger": "^2.0.0", "bare-module": "^7.0.1", "bare-module-resolve": "^1.6.2", "bare-path": "^3.0.0", + "bare-stream": "^2.13.4", "bare-system-logger": "^1.0.1", "bare-unpack": "^1.0.2", "bare-url": "^2.0.1" diff --git a/shared/builtins.json b/shared/builtins.json index b8c3aa8..60278f2 100644 --- a/shared/builtins.json +++ b/shared/builtins.json @@ -3,12 +3,10 @@ { "addon": "bare-fs" }, { "addon": "bare-hrtime" }, { "addon": "bare-inspect" }, - { "addon": "bare-ipc" }, { "addon": "bare-logger" }, { "addon": "bare-module" }, { "addon": "bare-module-lexer" }, { "addon": "bare-path" }, - { "addon": "bare-pipe" }, { "addon": "bare-os" }, { "addon": "bare-system-logger" }, { "addon": "bare-type" }, From 351302978b6dbd2174780980ee3e2edc0acbff53 Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Thu, 24 Sep 2026 13:10:31 +0530 Subject: [PATCH 10/11] Assert the host has retired its IPC before destroy The queue is the worklet's from the moment it is signalled to be destroyed, so a host holding an open IPC reads freed memory. Assert the documented order rather than letting it corrupt silently. --- shared/worklet.c | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/shared/worklet.c b/shared/worklet.c index dd22aa6..a3ced9a 100644 --- a/shared/worklet.c +++ b/shared/worklet.c @@ -101,6 +101,10 @@ bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options void bare_worklet_destroy(bare_worklet_t *worklet) { + // The host retires its ipc first: from here the queue is the worklet's and may + // be freed at any moment. + assert(!worklet->queue->thread_open || worklet->queue->thread_closed); + if (worklet->thread != 0) { // Stop the worklet if it is still running and wake its loop so the stop // takes effect, then unblock it. The thread is detached and tears itself From 51f044323ba5840173a7fc3198bc0993516c83da Mon Sep 17 00:00:00 2001 From: Balakrishna Avulapati Date: Thu, 24 Sep 2026 13:10:32 +0530 Subject: [PATCH 11/11] Retire the IPC when a worklet is terminated Both bindings called bare_worklet_destroy with the IPC still open, which is what the headers say not to do. Terminate now closes it first, and a closed IPC reads end-of-stream and refuses writes instead of reaching for a freed queue. --- .../main/java/to/holepunch/bare/kit/IPC.java | 28 +++++++++++++++- .../java/to/holepunch/bare/kit/Worklet.java | 5 +++ apple/BareKit/BareKit.h | 10 +++++- apple/BareKit/BareKit.m | 32 +++++++++++++++++++ 4 files changed, 73 insertions(+), 2 deletions(-) diff --git a/android/src/main/java/to/holepunch/bare/kit/IPC.java b/android/src/main/java/to/holepunch/bare/kit/IPC.java index c847aef..321e2f1 100644 --- a/android/src/main/java/to/holepunch/bare/kit/IPC.java +++ b/android/src/main/java/to/holepunch/bare/kit/IPC.java @@ -1,6 +1,7 @@ package to.holepunch.bare.kit; import java.io.Closeable; +import java.io.IOException; import java.nio.ByteBuffer; import java.nio.charset.Charset; import java.util.concurrent.atomic.AtomicReference; @@ -38,6 +39,8 @@ public IPC(Worklet worklet) { this.worklet = worklet; handle = init(worklet.handle); + + worklet.ipc = this; } private native ByteBuffer @@ -60,6 +63,8 @@ public IPC(Worklet worklet) { private void update() { + if (handle == null) return; + update(handle, readable != null, writable != null); } @@ -90,10 +95,13 @@ public IPC(Worklet worklet) { /** * Reads whatever is available without blocking. Returns null when nothing is * ready yet, and an empty buffer at end-of-stream, once the worklet has - * closed its end. Wait for readable() before retrying a null. + * closed its end or this IPC is closed. Wait for readable() before retrying + * a null. */ public ByteBuffer read() { + if (handle == null) return ByteBuffer.allocateDirect(0); + ByteBuffer buffer = read(handle); if (buffer == null) { @@ -134,6 +142,8 @@ public IPC(Worklet worklet) { public int write(ByteBuffer data) { + if (handle == null) return -1; + ByteBuffer buffer; if (data.isDirect()) { @@ -149,6 +159,12 @@ public IPC(Worklet worklet) { public void write(ByteBuffer data, WriteCallback callback) { + if (handle == null) { + callback.apply(new IOException("IPC closed")); + + return; + } + ByteBuffer buffer; if (data.isDirect()) { @@ -186,8 +202,18 @@ public IPC(Worklet worklet) { } } + /** + * Retires this IPC, which must happen before its worklet is destroyed. + * terminate() does it for you, and closing twice is a no-op. + */ public void close() { + if (handle == null) return; + destroy(handle); + + handle = null; + + worklet.ipc = null; } } diff --git a/android/src/main/java/to/holepunch/bare/kit/Worklet.java b/android/src/main/java/to/holepunch/bare/kit/Worklet.java index 40f1faa..65c3401 100644 --- a/android/src/main/java/to/holepunch/bare/kit/Worklet.java +++ b/android/src/main/java/to/holepunch/bare/kit/Worklet.java @@ -51,6 +51,9 @@ public static class Options { ByteBuffer handle; + // Unowned: the IPC holds this worklet and clears this when it closes. + IPC ipc; + public Worklet(Options options) { if (options == null) options = new Options(); @@ -158,6 +161,8 @@ public Worklet(Options options) { public void terminate() { + if (ipc != null) ipc.close(); + terminate(handle); handle = null; diff --git a/apple/BareKit/BareKit.h b/apple/BareKit/BareKit.h index e8cc82a..c9edcf7 100644 --- a/apple/BareKit/BareKit.h +++ b/apple/BareKit/BareKit.h @@ -85,7 +85,7 @@ /** * Reads whatever is available without blocking. Returns `nil` when nothing is * ready yet, and empty data at end-of-stream, once the worklet has closed its - * end. Wait for `readable` before retrying a `nil`. + * end or this ipc is closed. Wait for `readable` before retrying a `nil`. */ - (NSData *_Nullable)read; @@ -95,11 +95,19 @@ */ - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error))completion; +/** + * Writes whatever fits without blocking, returning the number of bytes taken, + * or -1 once this ipc is closed. + */ - (NSInteger)write:(NSData *_Nonnull)data; - (void)write:(NSData *_Nonnull)data completion:(void (^_Nonnull)(NSError *_Nullable error))completion; +/** + * Retires this ipc, which must happen before its worklet is destroyed. + * `terminate` does it for you, and closing twice is a no-op. + */ - (void)close; @end diff --git a/apple/BareKit/BareKit.m b/apple/BareKit/BareKit.m index e3934bc..b491519 100644 --- a/apple/BareKit/BareKit.m +++ b/apple/BareKit/BareKit.m @@ -103,6 +103,9 @@ - (_Nullable instancetype)init { @implementation BareWorklet { @public bare_worklet_t _worklet; + + // Unowned: the ipc retains this worklet and clears this when it closes. + BareIPC *_ipc; } + (void)optimizeForMemory:(BOOL)enabled { @@ -234,6 +237,8 @@ - (void)resume { } - (void)terminate { + [_ipc close]; + int err; err = bare_worklet_terminate(&_worklet); assert(err == 0); @@ -303,6 +308,8 @@ @implementation BareIPC { bare_ipc_t _ipc; bare_ipc_poll_t _poll; + + bool _closed; } - (_Nullable instancetype)initWithWorklet:(BareWorklet *_Nonnull)worklet { @@ -320,6 +327,8 @@ - (_Nullable instancetype)initWithWorklet:(BareWorklet *_Nonnull)worklet { assert(err == 0); _poll.data = (__bridge void *) self; + + worklet->_ipc = self; } return self; @@ -347,6 +356,8 @@ - (void)setWritable:(void (^)(BareIPC *_Nonnull))writable { } - (void)update { + if (_closed) return; + int events = 0; if (_readable) events |= bare_ipc_readable; @@ -369,6 +380,8 @@ - (void)poll:(int)events { } - (NSData *_Nullable)read { + if (_closed) return [NSData data]; + int err; void *data; @@ -384,6 +397,8 @@ - (NSData *_Nullable)read { } - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error))completion { + if (_closed) return completion(nil, nil); + int err; void *data; @@ -411,6 +426,8 @@ - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error) } - (NSInteger)write:(NSData *_Nonnull)data { + if (_closed) return -1; + int err; err = bare_ipc_write(&_ipc, data.bytes, data.length); assert(err >= 0 || err == bare_ipc_would_block); @@ -424,6 +441,13 @@ - (NSInteger)write:(NSData *_Nonnull)data { - (void)write:(NSData *_Nonnull)data completion:(void (^_Nonnull)(NSError *_Nullable error))completion { + if (_closed) { + return completion([NSError + errorWithDomain:@"to.holepunch.bare.kit" + code:-1 + userInfo:@{NSLocalizedDescriptionKey : @"IPC closed"}]); + } + NSInteger written = [self write:data]; if (written == data.length) { @@ -457,11 +481,19 @@ - (void)write:(NSData *_Nonnull)data } - (void)close { + if (_closed) return; + + _closed = true; + + _worklet->_ipc = nil; + bare_ipc_poll_destroy(&_poll); bare_ipc_destroy(&_ipc); } - (void)dealloc { + [self close]; + [_worklet release]; [super dealloc];