From 1223510f70679351badcfabbdd779180458b2a9d Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Sun, 23 Aug 2026 11:22:33 -0700 Subject: [PATCH 1/8] Make sync proxied waits safe cancellation points A thread blocked in emscripten_proxy_sync[_with_ctx] already woke on pthread_cancel (the cond wait's futex honors it), but unwound with the em_proxying_ctx and the caller-owned argument still on its stack while the target thread kept referencing both, and for PROXY_SYNC_ASYNC JS imports a later promise resolution wrote the result into the freed stack. Any blocking proxied wait could hit this, e.g. poll()/epoll_wait() with an infinite timeout. The sync ctx is now heap allocated and refcounted between caller and target. A cancellation cleanup handler on the caller waits until the target has released the argument (immediately after a JS import has been dispatched; on completion for generic callers), marks the ctx orphaned and drops its reference, so a late completion on the target is harmless. The PROXY_SYNC JS path routes through the same guarded result write. --- system/include/emscripten/proxying.h | 8 +- system/lib/pthread/proxying.c | 131 ++++++++++++++---- .../test_pthread_proxying_canceled_caller.c | 114 +++++++++++++++ test/sockets/test_epoll_cancel.c | 85 ++++++++++++ test/test_core.py | 8 ++ test/test_sockets_node.py | 55 ++++---- 6 files changed, 350 insertions(+), 51 deletions(-) create mode 100644 test/pthread/test_pthread_proxying_canceled_caller.c create mode 100644 test/sockets/test_epoll_cancel.c diff --git a/system/include/emscripten/proxying.h b/system/include/emscripten/proxying.h index 380adcb5cb932..967f30aecc95d 100644 --- a/system/include/emscripten/proxying.h +++ b/system/include/emscripten/proxying.h @@ -60,7 +60,9 @@ bool emscripten_proxy_async(em_proxying_queue* q, // Enqueue `func` on the given queue and thread and wait for it to finish // executing before returning. Returns true if the task was successfully // completed and false otherwise, including if the target thread is canceled or -// exits before the work is completed. +// exits before the work is completed. The wait is a cancellation point: a +// canceled caller exits with PTHREAD_CANCELED once `func` has completed, since +// `arg` may be on the caller's stack. bool emscripten_proxy_sync(em_proxying_queue* q, pthread_t target_thread, void (*func)(void*), @@ -72,7 +74,9 @@ bool emscripten_proxy_sync(em_proxying_queue* q, // instead store the context pointer and call `emscripten_proxy_finish` at an // arbitrary later time. Returns true if the task was successfully completed and // false otherwise, including if the target thread is canceled or exits before -// the work is completed. +// the work is completed. The wait is a cancellation point: a canceled caller +// exits with PTHREAD_CANCELED once the task is finished (or canceled), since +// `arg` may be on the caller's stack. bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, pthread_t target_thread, void (*func)(em_proxying_ctx*, void*), diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index 1342cb2cc924f..2ea9dd5bfcca3 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -199,13 +199,23 @@ struct em_proxying_ctx { enum ctx_kind kind; union { - // Context for synchronous proxying. + // Context for synchronous proxying. Heap allocated and shared between the + // waiting caller and the target thread, each holding one reference, so + // that a caller canceled or exiting mid-wait leaves the target a valid + // ctx to finish (or cancel) later. struct { // Update `state` and signal the condition variable once the proxied task // is done or canceled. enum ctx_state state; pthread_mutex_t mutex; pthread_cond_t cond; + // `arg` is caller-owned (typically on its stack). The target sets + // `arg_released` under `mutex` once it will no longer read it (implied by + // completion); a canceled caller waits for that before marking + // `caller_gone`, after which `arg` must not be touched at all. + bool arg_released; + bool caller_gone; + _Atomic int refs; } sync; // Context for proxying with callbacks. @@ -301,6 +311,9 @@ static void em_proxying_ctx_init_sync(em_proxying_ctx* ctx, .state = PENDING, .mutex = PTHREAD_MUTEX_INITIALIZER, .cond = PTHREAD_COND_INITIALIZER, + .arg_released = false, + .caller_gone = false, + .refs = 2, }, }; } @@ -342,6 +355,28 @@ static void free_ctx(void* arg) { free(ctx); } +static void sync_ctx_unref(em_proxying_ctx* ctx) { + assert(ctx->kind == SYNC); + if (atomic_fetch_sub(&ctx->sync.refs, 1) == 1) { + free_ctx(ctx); + } +} + +// Complete a sync ctx with `mutex` held: publish the state, drop it from the +// target thread's active list and wake the caller. Unlocks and drops the +// target's reference. +static void sync_ctx_complete_locked(em_proxying_ctx* ctx, + enum ctx_state state) { + ctx->sync.state = state; + ctx->sync.arg_released = true; + // Signal must come before unlock to avoid emscripten_proxy_sync_with_ctx + // seeing the state as DONE and freeing the ctx before we call unlock. + // See https://github.com/emscripten-core/emscripten/pull/26582 + pthread_cond_signal(&ctx->sync.cond); + pthread_mutex_unlock(&ctx->sync.mutex); + sync_ctx_unref(ctx); +} + // Free the callback info on the same thread it was originally allocated on. // This may be more efficient. static void call_callback_then_free_ctx(void* arg) { @@ -353,13 +388,8 @@ static void call_callback_then_free_ctx(void* arg) { void emscripten_proxy_finish(em_proxying_ctx* ctx) { if (ctx->kind == SYNC) { pthread_mutex_lock(&ctx->sync.mutex); - ctx->sync.state = DONE; remove_active_ctx(ctx); - // Signal must come before unlock to avoid emscripten_proxy_sync_with ctx - // seeing the state as DONE and freeing the ctx before we call unlock. - // See https://github.com/emscripten-core/emscripten/pull/26582 - pthread_cond_signal(&ctx->sync.cond); - pthread_mutex_unlock(&ctx->sync.mutex); + sync_ctx_complete_locked(ctx, DONE); } else { // Schedule the callback on the caller thread. If the caller thread has // already died or dies before the callback is executed, then at least make @@ -383,10 +413,7 @@ static void cancel_ctx(void* arg) { em_proxying_ctx* ctx = arg; if (ctx->kind == SYNC) { pthread_mutex_lock(&ctx->sync.mutex); - ctx->sync.state = CANCELED; - // Signal must be first, see comment in emscripten_proxy_finish. - pthread_cond_signal(&ctx->sync.cond); - pthread_mutex_unlock(&ctx->sync.mutex); + sync_ctx_complete_locked(ctx, CANCELED); } else { if (ctx->cb.cancel == NULL || !do_proxy(ctx->cb.queue, @@ -404,25 +431,47 @@ static void call_with_ctx(void* arg) { ctx->func(ctx, ctx->arg); } +// Cancellation cleanup for a caller unwound out of the wait below. Runs with +// `mutex` held (the cond wait re-acquires it before acting on cancellation) +// and with cancellation disabled, so waiting here cannot recurse. Hold the +// caller's stack alive until the target no longer reads `arg`, then hand the +// ctx over to the target. +static void orphan_sync_ctx(void* arg) { + em_proxying_ctx* ctx = arg; + while (!ctx->sync.arg_released) { + pthread_cond_wait(&ctx->sync.cond, &ctx->sync.mutex); + } + ctx->sync.caller_gone = true; + pthread_mutex_unlock(&ctx->sync.mutex); + sync_ctx_unref(ctx); +} + bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, pthread_t target_thread, void (*func)(em_proxying_ctx*, void*), void* arg) { assert(!pthread_equal(target_thread, pthread_self()) && "Cannot synchronously wait for work proxied to the current thread"); - em_proxying_ctx ctx; - em_proxying_ctx_init_sync(&ctx, func, arg); - if (!do_proxy(q, target_thread, (task){call_with_ctx, cancel_ctx, &ctx})) { - em_proxying_ctx_deinit(&ctx); + em_proxying_ctx* ctx = malloc(sizeof(em_proxying_ctx)); + if (!ctx) { return false; } - pthread_mutex_lock(&ctx.sync.mutex); - while (ctx.sync.state == PENDING) { - pthread_cond_wait(&ctx.sync.cond, &ctx.sync.mutex); + em_proxying_ctx_init_sync(ctx, func, arg); + if (!do_proxy(q, target_thread, (task){call_with_ctx, cancel_ctx, ctx})) { + free_ctx(ctx); + return false; } - pthread_mutex_unlock(&ctx.sync.mutex); - int ret = ctx.sync.state == DONE; - em_proxying_ctx_deinit(&ctx); + pthread_mutex_lock(&ctx->sync.mutex); + // The wait is a cancellation point: a canceled caller exits from inside it, + // so hand the ctx over to the target rather than leaving it dangling. + pthread_cleanup_push(orphan_sync_ctx, ctx); + while (ctx->sync.state == PENDING) { + pthread_cond_wait(&ctx->sync.cond, &ctx->sync.mutex); + } + pthread_cleanup_pop(0); + pthread_mutex_unlock(&ctx->sync.mutex); + int ret = ctx->sync.state == DONE; + sync_ctx_unref(ctx); return ret; } @@ -648,12 +697,43 @@ static void run_js_func_with_ctx(em_proxying_ctx* ctx, void* arg) { // should never be owned on the main thread (i.e. the argument here always // exists on the stack of the calling thread, it's never copied/malloced). assert(!f->owned); + + // The arguments have been deserialized; the only later access to `f` is the + // guarded result write in _emscripten_run_js_on_main_thread_done, so a + // canceled caller may now leave. + pthread_mutex_lock(&ctx->sync.mutex); + ctx->sync.arg_released = true; + pthread_cond_signal(&ctx->sync.cond); + pthread_mutex_unlock(&ctx->sync.mutex); } -void _emscripten_run_js_on_main_thread_done(void* ctx, void* arg, double result) { +void _emscripten_run_js_on_main_thread_done(void* arg_ctx, + void* arg, + double result) { + em_proxying_ctx* ctx = arg_ctx; proxied_js_func_t* f = (proxied_js_func_t*)arg; - f->result = result; - emscripten_proxy_finish(ctx); + // `f` lives on the caller's stack; it is gone if the caller was canceled + // while waiting (e.g. a pthread_cancel of a thread blocked in recv/poll). + pthread_mutex_lock(&ctx->sync.mutex); + if (!ctx->sync.caller_gone) { + f->result = result; + } + remove_active_ctx(ctx); + sync_ctx_complete_locked(ctx, DONE); +} + +// PROXY_SYNC: run the JS function to completion on the target thread, then +// hand the result back through the same caller-gone guard as the async case. +static void run_js_func_sync_with_ctx(em_proxying_ctx* ctx, void* arg) { + proxied_js_func_t* f = (proxied_js_func_t*)arg; + double result = _emscripten_receive_on_main_thread_js(f->funcIndex, + f->emAsmAddr, + f->callingThread, + f->bufSize, + f->argBuffer, + NULL, + NULL); + _emscripten_run_js_on_main_thread_done(ctx, arg, result); } /* @@ -694,7 +774,8 @@ double _emscripten_run_js_on_main_thread(int func_index, if (proxyMode == PROXY_SYNC_ASYNC) { rtn = emscripten_proxy_sync_with_ctx(q, target, run_js_func_with_ctx, &f); } else { - rtn = emscripten_proxy_sync(q, target, run_js_func, &f); + rtn = emscripten_proxy_sync_with_ctx( + q, target, run_js_func_sync_with_ctx, &f); } if (!rtn) { assert(false && "emscripten_proxy_sync_with_ctx failed"); diff --git a/test/pthread/test_pthread_proxying_canceled_caller.c b/test/pthread/test_pthread_proxying_canceled_caller.c new file mode 100644 index 0000000000000..765a8aaf939da --- /dev/null +++ b/test/pthread/test_pthread_proxying_canceled_caller.c @@ -0,0 +1,114 @@ +/* + * Copyright 2026 The Emscripten Authors. All rights reserved. + * Emscripten is available under two separate licenses, the MIT license and the + * University of Illinois/NCSA Open Source License. Both these licenses can be + * found in the LICENSE file. + * + * A thread blocked in emscripten_proxy_sync[_with_ctx] is canceled while its + * work is (a) in progress on the target and (b) handed a ctx the target has + * not finished yet. The caller must exit with PTHREAD_CANCELED, but only once + * the target is done with the caller-owned argument (its stack), i.e. when the + * work completes or the ctx is finished. + */ + +#include +#include +#include +#include +#include +#include +#include + +static em_proxying_queue* q; +static pthread_t target; +static _Atomic bool stop_target; + +static void* target_main(void* arg) { + while (!stop_target) { + emscripten_proxy_execute_queue(q); + usleep(1000); + } + return NULL; +} + +// (a) Work in progress on the target when the caller is canceled. The target +// keeps reading the caller's stack argument until released, so the canceled +// caller must not have unwound yet. +static _Atomic bool slow_started, slow_release, slow_done; + +static void slow(void* arg) { + slow_started = true; + while (!slow_release) { + usleep(1000); + } + assert(*(int*)arg == 42 && "caller stack freed while target still reads it"); + slow_done = true; +} + +static void* sync_caller(void* arg) { + int local = 42; + emscripten_proxy_sync(q, target, slow, &local); + assert(false && "should have been canceled"); + return NULL; +} + +// (b) The target returns from the proxied function without finishing the ctx; +// the canceled caller is held until the ctx is finished. +static em_proxying_ctx* _Atomic stashed; +static _Atomic bool ctx_caller_exiting; + +static void stash(em_proxying_ctx* ctx, void* arg) { stashed = ctx; } + +static void note_exit(void* arg) { ctx_caller_exiting = true; } + +static void* ctx_caller(void* arg) { + int local = 42; + pthread_cleanup_push(note_exit, NULL); + emscripten_proxy_sync_with_ctx(q, target, stash, &local); + pthread_cleanup_pop(0); + assert(false && "should have been canceled"); + return NULL; +} + +static void finish_stashed(void* arg) { emscripten_proxy_finish(stashed); } + +static void noop(void* arg) {} + +int main(void) { + q = em_proxying_queue_create(); + assert(pthread_create(&target, NULL, target_main, NULL) == 0); + + pthread_t caller; + void* ret; + + assert(pthread_create(&caller, NULL, sync_caller, NULL) == 0); + while (!slow_started) + usleep(1000); + assert(pthread_cancel(caller) == 0); + usleep(20000); // the cancel must not complete while `slow` still runs + assert(!slow_done); + slow_release = true; + assert(pthread_join(caller, &ret) == 0); + assert(ret == PTHREAD_CANCELED); + assert(slow_done); + + assert(pthread_create(&caller, NULL, ctx_caller, NULL) == 0); + while (!stashed) + usleep(1000); + assert(pthread_cancel(caller) == 0); + usleep(20000); // the cancel must not complete while the ctx is unfinished + assert(!ctx_caller_exiting); + assert(emscripten_proxy_async(q, target, finish_stashed, NULL)); + assert(pthread_join(caller, &ret) == 0); + assert(ret == PTHREAD_CANCELED); + assert(ctx_caller_exiting); + + // The queue is still healthy for ordinary work. + assert(emscripten_proxy_sync(q, target, noop, NULL)); + + stop_target = true; + assert(pthread_join(target, NULL) == 0); + em_proxying_queue_destroy(q); + printf("done\n"); + return 0; +} diff --git a/test/sockets/test_epoll_cancel.c b/test/sockets/test_epoll_cancel.c new file mode 100644 index 0000000000000..65b9dbc0d0bdf --- /dev/null +++ b/test/sockets/test_epoll_cancel.c @@ -0,0 +1,85 @@ +/* + * Copyright 2026 The Emscripten Authors. All rights reserved. + * Emscripten is available under two separate licenses, the MIT license and the + * University of Illinois/NCSA Open Source License. Both these licenses can be + * found in the LICENSE file. + * + * Threads blocked indefinitely in epoll_wait() and poll() on a socket are + * canceled with pthread_cancel. Each must exit with PTHREAD_CANCELED, leaving + * its readiness listener registered on the main thread. A datagram then fires + * those orphaned listeners - resolving proxied waits whose callers are gone - + * which must be harmless, and the socket must still be usable afterwards. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static int rx, tx, ep; +static struct sockaddr_in addr; + +static void* epoll_waiter(void* arg) { + struct epoll_event out; + epoll_wait(ep, &out, 1, -1); + assert(0 && "epoll_wait should have been canceled"); + return NULL; +} + +static void* poll_waiter(void* arg) { + struct pollfd pfd = {.fd = rx, .events = POLLIN}; + poll(&pfd, 1, -1); + assert(0 && "poll should have been canceled"); + return NULL; +} + +int main(void) { + rx = socket(AF_INET, SOCK_DGRAM, 0); + tx = socket(AF_INET, SOCK_DGRAM, 0); + assert(rx >= 0 && tx >= 0); + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr); + assert(bind(rx, (struct sockaddr*)&addr, sizeof(addr)) == 0); + socklen_t l = sizeof(addr); + assert(getsockname(rx, (struct sockaddr*)&addr, &l) == 0); + + ep = epoll_create1(0); + assert(ep >= 0); + struct epoll_event ev = {.events = EPOLLIN}; + ev.data.fd = rx; + assert(epoll_ctl(ep, EPOLL_CTL_ADD, rx, &ev) == 0); + + pthread_t t1, t2; + assert(pthread_create(&t1, NULL, epoll_waiter, NULL) == 0); + assert(pthread_create(&t2, NULL, poll_waiter, NULL) == 0); + usleep(100000); // let both park in their waits first + + void* ret; + assert(pthread_cancel(t1) == 0); + assert(pthread_join(t1, &ret) == 0 && ret == PTHREAD_CANCELED); + assert(pthread_cancel(t2) == 0); + assert(pthread_join(t2, &ret) == 0 && ret == PTHREAD_CANCELED); + + // Fire the listeners the canceled waits left behind. + assert(sendto(tx, "ping", 4, 0, (struct sockaddr*)&addr, sizeof(addr)) == 4); + usleep(100000); // let the late completions run on the main thread + + struct epoll_event out; + assert(epoll_wait(ep, &out, 1, -1) == 1 && out.data.fd == rx); + char buf[4]; + assert(recvfrom(rx, buf, sizeof(buf), 0, NULL, NULL) == 4); + assert(memcmp(buf, "ping", 4) == 0); + + close(ep); + close(rx); + close(tx); + printf("done\n"); + return 0; +} diff --git a/test/test_core.py b/test/test_core.py index 110dacf458821..08dc94e502ce3 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -2616,6 +2616,14 @@ def test_pthread_proxying_canceled_work(self): 'pthread/test_pthread_proxying_canceled_work.c', interleaved_output=False) + @requires_pthreads + def test_pthread_proxying_canceled_caller(self): + # pthread_cancel of a thread blocked in a sync proxy call, both while the + # target still runs the work and after the target has been handed the ctx. + self.set_setting('PROXY_TO_PTHREAD') + self.set_setting('EXIT_RUNTIME') + self.do_runf('pthread/test_pthread_proxying_canceled_caller.c', 'done\n') + @requires_pthreads @flaky('https://github.com/emscripten-core/emscripten/issues/19795') def test_pthread_proxying_refcount(self): diff --git a/test/test_sockets_node.py b/test/test_sockets_node.py index 16ec7db2815aa..09dac7f6c9327 100644 --- a/test/test_sockets_node.py +++ b/test/test_sockets_node.py @@ -100,12 +100,12 @@ def _run_against_echo_server(self, src): # are proxied to the main thread: with PROXY_TO_PTHREAD, main() runs on a # worker and every socket call funnels to the main thread where node:net lives. @also_with_proxy_to_pthread - def test_noderawsockets_echo(self): + def test_echo(self): # With -sNODERAWSOCKETS the client does a non-blocking connect, send and # recv over a real OS socket against a loopback echo server we run here. self._run_against_echo_server('sockets/test_tcp_echo.c') - def test_noderawsockets_client_bind(self): + def test_client_bind(self): # A client that bind()s an explicit source port has it honored by connect(), # and the plain client path never realizes a private tcp_wrap handle. We # allocate a free source port here and pass it alongside the echo server's. @@ -126,22 +126,22 @@ def test_noderawsockets_client_bind(self): server.server_close() thread.join() - def test_noderawsockets_connect_getsockname(self): + def test_connect_getsockname(self): # getsockname() immediately after a non-blocking connect() on an unbound # client reports the ephemeral source port synchronously (kernel semantics: # the port is assigned at connect(), not when the connection completes). self.do_runf('sockets/test_tcp_connect_getsockname.c', 'done\n', cflags=['-sNODERAWSOCKETS']) - def test_noderawsockets_client_semantics(self): + def test_client_semantics(self): # EISCONN on a second connect, shutdown(SHUT_WR) leaving reads working, # EPIPE on a write after that, and POLLHUP after a full shutdown(SHUT_RDWR). self._run_against_echo_server('sockets/test_tcp_client_semantics.c') - def test_noderawsockets_refused(self): + def test_refused(self): # A connect to a loopback port with nothing listening reports ECONNREFUSED. self.do_runf('sockets/test_tcp_refused.c', 'done\n', cflags=['-sNODERAWSOCKETS']) - def test_noderawsockets_backpressure(self): + def test_backpressure(self): # A sink server that accepts but never reads, so the client's writes fill # the buffers and send() reports EAGAIN rather than buffering unboundedly. done = threading.Event() @@ -163,14 +163,14 @@ def handle(self): thread.join() @also_with_proxy_to_pthread - def test_noderawsockets_server(self): + def test_server(self): # Self-contained loopback accept+echo, exercising bind(:0)+getsockname # (synchronous ephemeral port), listen, accept, non-blocking connect, send # and recv over real OS sockets via the tcp_wrap server path. self.do_runf('sockets/test_tcp_server.c', 'done\n', cflags=['-sNODERAWSOCKETS']) @also_with_proxy_to_pthread - def test_noderawsockets_peek(self): + def test_peek(self): # recv(MSG_PEEK) must leave the data buffered: a peek returns the bytes, the # socket stays readable, and the following plain recv returns them again. self.do_runf('sockets/test_tcp_peek.c', 'done\n', cflags=['-sNODERAWSOCKETS']) @@ -179,39 +179,39 @@ def test_noderawsockets_peek(self): # filesystem, which only stays coherent with the program's own file syscalls # (bind's parent dir, getsockname, unlink) when the FS is the host FS. @also_with_proxy_to_pthread - def test_noderawsockets_unix_server(self): + def test_unix_server(self): # Self-contained named AF_UNIX (pathname) loopback accept+echo: bind(path), # listen, getsockname (the bound path), accept, getpeername, non-blocking # connect-by-path, send and recv over a real node pipe. self.do_runf('sockets/test_unix_server.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sNODERAWFS']) - def test_noderawsockets_unix_refused(self): + def test_unix_refused(self): # A connect to an AF_UNIX path with no socket file reports ENOENT. self.do_runf('sockets/test_unix_refused.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sNODERAWFS']) - def test_noderawsockets_unix_bind_inuse(self): + def test_unix_bind_inuse(self): # Binding an already-bound AF_UNIX path fails synchronously with EADDRINUSE. self.do_runf('sockets/test_unix_bind_inuse.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sNODERAWFS']) - def test_noderawsockets_server_autobind(self): + def test_server_autobind(self): # listen() without a prior bind() must auto-bind an ephemeral port and # getsockname() must report it (POSIX), then accept+echo as usual. self.do_runf('sockets/test_tcp_server.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-DNO_EXPLICIT_BIND']) - def test_noderawsockets_tcp_ipv6(self): + def test_tcp_ipv6(self): # Self-contained IPv6 TCP loopback accept+echo over ::1: bind(:0)+getsockname, # listen, accept, non-blocking connect, send/recv on AF_INET6 sockets. if not HAS_IPV6_LOOPBACK: self.skipTest('no IPv6 loopback available') self.do_runf('sockets/test_tcp_ipv6.c', 'done\n', cflags=['-sNODERAWSOCKETS']) - def test_noderawsockets_udp_ipv6(self): + def test_udp_ipv6(self): # Self-contained IPv6 UDP loopback echo over ::1 on AF_INET6 sockets. if not HAS_IPV6_LOOPBACK: self.skipTest('no IPv6 loopback available') self.do_runf('sockets/test_udp_ipv6.c', 'done\n', cflags=['-sNODERAWSOCKETS']) - def test_noderawsockets_epoll_socket_blocking(self): + def test_epoll_socket_blocking(self): # A blocking epoll_wait() on a socket is woken by an incoming datagram # through the unified readiness wait-queue (the SOCKFS.emit bridge), with # main() proxied to a worker so the wait can suspend. @@ -219,47 +219,54 @@ def test_noderawsockets_epoll_socket_blocking(self): cflags=['-sNODERAWSOCKETS', '-pthread', '-sPROXY_TO_PTHREAD', '-sEXIT_RUNTIME']) @requires_jspi_node - def test_noderawsockets_epoll_socket_blocking_jspi(self): + def test_epoll_socket_blocking_jspi(self): # Same, but the blocking epoll_wait() suspends the wasm stack under JSPI. self.do_runf('sockets/test_epoll_socket_blocking.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME']) - def test_noderawsockets_epoll_rdhup(self): + def test_epoll_cancel(self): + # pthread_cancel of threads blocked in epoll_wait(-1) and poll(-1) exits + # them with PTHREAD_CANCELED; the readiness listeners they left behind are + # then fired by a real datagram and must complete harmlessly. + self.do_runf('sockets/test_epoll_cancel.c', 'done\n', + cflags=['-sNODERAWSOCKETS', '-pthread', '-sPROXY_TO_PTHREAD', '-sEXIT_RUNTIME']) + + def test_epoll_rdhup(self): # A blocking epoll_wait reports EPOLLRDHUP when the TCP peer half-closes its # write side (FIN), distinct from a full EPOLLHUP, and only when requested. self.do_runf('sockets/test_epoll_rdhup.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-pthread', '-sPROXY_TO_PTHREAD', '-sEXIT_RUNTIME']) @requires_jspi_node - def test_noderawsockets_epoll_rdhup_jspi(self): + def test_epoll_rdhup_jspi(self): # Same, but the blocking calls suspend the wasm stack under JSPI. self.do_runf('sockets/test_epoll_rdhup.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME']) @also_with_proxy_to_pthread - def test_noderawsockets_udp(self): + def test_udp(self): # Self-contained loopback UDP echo: the server binds(:0)+getsockname for its # ephemeral port, the client sends a datagram, the server echoes it back. self.do_runf('sockets/test_udp_echo.c', 'done\n', cflags=['-sNODERAWSOCKETS']) - def test_noderawsockets_udp_recvmsg(self): + def test_udp_recvmsg(self): # recvmsg scatters a datagram across multiple iovecs at the right offsets # and updates msg_namelen/msg_controllen/msg_flags in the caller's msghdr. self.do_runf('sockets/test_udp_recvmsg.c', 'done\n', cflags=['-sNODERAWSOCKETS']) - def test_noderawsockets_mmsg(self): + def test_mmsg(self): # sendmmsg batches two datagrams out, recvmmsg receives them back in one # call, updating msg_len per message. self.do_runf('sockets/test_udp_mmsg.c', 'done\n', cflags=['-sNODERAWSOCKETS']) @also_with_proxy_to_pthread - def test_noderawsockets_udp_connect(self): + def test_udp_connect(self): # Connected UDP: sendto() with an address gives EISCONN, send() reaches the # peer, and datagrams from a non-peer socket are filtered out. self.do_runf('sockets/test_udp_connect.c', 'done\n', cflags=['-sNODERAWSOCKETS']) @also_with_proxy_to_pthread - def test_noderawsockets_udp_sockopts(self): + def test_udp_sockopts(self): # UDP multicast socket options: IP_MULTICAST_TTL/LOOP and their IPv6 # counterparts round-trip through set/getsockopt, with POSIX defaults # readable before any set. EXIT_RUNTIME so the plain synchronous main() @@ -268,7 +275,7 @@ def test_noderawsockets_udp_sockopts(self): self.do_runf('sockets/test_udp_sockopts.c', 'done\n', cflags=['-sNODERAWSOCKETS', '-sEXIT_RUNTIME']) @also_with_proxy_to_pthread - def test_noderawsockets_socket_options(self): + def test_socket_options(self): # Socket metadata/options on a fresh socket: fstat reports S_ISSOCK, SO_TYPE # reports the socket type, and SO_LINGER round-trips a struct linger. self.do_runf('sockets/test_socket_options.c', 'done\n', From 743ff32b4b67ff7b4d940b570bd8f138745660d5 Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Fri, 28 Aug 2026 16:49:17 -0700 Subject: [PATCH 2/8] Review feedback: doc wording, event-loop target in cancel test --- system/include/emscripten/proxying.h | 8 ++-- .../test_pthread_proxying_canceled_caller.c | 40 +++++++++---------- 2 files changed, 23 insertions(+), 25 deletions(-) diff --git a/system/include/emscripten/proxying.h b/system/include/emscripten/proxying.h index 967f30aecc95d..16e7a94c2c5b7 100644 --- a/system/include/emscripten/proxying.h +++ b/system/include/emscripten/proxying.h @@ -61,8 +61,8 @@ bool emscripten_proxy_async(em_proxying_queue* q, // executing before returning. Returns true if the task was successfully // completed and false otherwise, including if the target thread is canceled or // exits before the work is completed. The wait is a cancellation point: a -// canceled caller exits with PTHREAD_CANCELED once `func` has completed, since -// `arg` may be on the caller's stack. +// canceled caller exits with PTHREAD_CANCELED, but only after `func` has +// completed, since `arg` may be on the caller's stack. bool emscripten_proxy_sync(em_proxying_queue* q, pthread_t target_thread, void (*func)(void*), @@ -75,8 +75,8 @@ bool emscripten_proxy_sync(em_proxying_queue* q, // arbitrary later time. Returns true if the task was successfully completed and // false otherwise, including if the target thread is canceled or exits before // the work is completed. The wait is a cancellation point: a canceled caller -// exits with PTHREAD_CANCELED once the task is finished (or canceled), since -// `arg` may be on the caller's stack. +// exits with PTHREAD_CANCELED, but only after the task is finished (or +// canceled), since `arg` may be on the caller's stack. bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, pthread_t target_thread, void (*func)(em_proxying_ctx*, void*), diff --git a/test/pthread/test_pthread_proxying_canceled_caller.c b/test/pthread/test_pthread_proxying_canceled_caller.c index 765a8aaf939da..5904a47ab945e 100644 --- a/test/pthread/test_pthread_proxying_canceled_caller.c +++ b/test/pthread/test_pthread_proxying_canceled_caller.c @@ -12,6 +12,8 @@ */ #include +#include +#include #include #include #include @@ -19,24 +21,20 @@ #include #include -static em_proxying_queue* q; -static pthread_t target; -static _Atomic bool stop_target; +em_proxying_queue* q; +pthread_t target; -static void* target_main(void* arg) { - while (!stop_target) { - emscripten_proxy_execute_queue(q); - usleep(1000); - } - return NULL; -} +// The target runs its event loop, which is where proxied work is dispatched. +void* target_main(void* arg) { emscripten_exit_with_live_runtime(); } + +void stop_target(void* arg) { emscripten_runtime_keepalive_pop(); } // (a) Work in progress on the target when the caller is canceled. The target // keeps reading the caller's stack argument until released, so the canceled // caller must not have unwound yet. -static _Atomic bool slow_started, slow_release, slow_done; +_Atomic bool slow_started, slow_release, slow_done; -static void slow(void* arg) { +void slow(void* arg) { slow_started = true; while (!slow_release) { usleep(1000); @@ -45,7 +43,7 @@ static void slow(void* arg) { slow_done = true; } -static void* sync_caller(void* arg) { +void* sync_caller(void* arg) { int local = 42; emscripten_proxy_sync(q, target, slow, &local); assert(false && "should have been canceled"); @@ -54,14 +52,14 @@ static void* sync_caller(void* arg) { // (b) The target returns from the proxied function without finishing the ctx; // the canceled caller is held until the ctx is finished. -static em_proxying_ctx* _Atomic stashed; -static _Atomic bool ctx_caller_exiting; +em_proxying_ctx* _Atomic stashed; +_Atomic bool ctx_caller_exiting; -static void stash(em_proxying_ctx* ctx, void* arg) { stashed = ctx; } +void stash(em_proxying_ctx* ctx, void* arg) { stashed = ctx; } -static void note_exit(void* arg) { ctx_caller_exiting = true; } +void note_exit(void* arg) { ctx_caller_exiting = true; } -static void* ctx_caller(void* arg) { +void* ctx_caller(void* arg) { int local = 42; pthread_cleanup_push(note_exit, NULL); emscripten_proxy_sync_with_ctx(q, target, stash, &local); @@ -70,9 +68,9 @@ static void* ctx_caller(void* arg) { return NULL; } -static void finish_stashed(void* arg) { emscripten_proxy_finish(stashed); } +void finish_stashed(void* arg) { emscripten_proxy_finish(stashed); } -static void noop(void* arg) {} +void noop(void* arg) {} int main(void) { q = em_proxying_queue_create(); @@ -106,7 +104,7 @@ int main(void) { // The queue is still healthy for ordinary work. assert(emscripten_proxy_sync(q, target, noop, NULL)); - stop_target = true; + assert(emscripten_proxy_async(q, target, stop_target, NULL)); assert(pthread_join(target, NULL) == 0); em_proxying_queue_destroy(q); printf("done\n"); From e370453a01d404fae368aec43b3ee0d7f3a48de2 Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Fri, 28 Aug 2026 17:41:25 -0700 Subject: [PATCH 3/8] Initialize thread result when a pthread exits after unwinding --- src/lib/libcore.js | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/lib/libcore.js b/src/lib/libcore.js index 5247fbfd06284..6c998e7a65e5f 100644 --- a/src/lib/libcore.js +++ b/src/lib/libcore.js @@ -2138,7 +2138,9 @@ addToLibrary({ // exit the current thread, but only if there is one active. // TODO(https://github.com/emscripten-core/emscripten/issues/25076): // Unify this check with the runtimeExited check above - if (_pthread_self()) __emscripten_thread_exit(EXITSTATUS); + // `EXITSTATUS` holds the thread result if the entry point has already + // returned, but is unset if the thread unwound to the event loop. + if (_pthread_self()) __emscripten_thread_exit(EXITSTATUS ?? 0); return; } #endif From 810d902d0a68107e3b624a939876c37ebc7d7ec2 Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Fri, 28 Aug 2026 18:00:45 -0700 Subject: [PATCH 4/8] rebaseline --- .../test_codesize_minimal_pthreads.json | 41 ++++++++++--------- ...t_codesize_minimal_pthreads_memgrowth.json | 41 ++++++++++--------- 2 files changed, 42 insertions(+), 40 deletions(-) diff --git a/test/codesize/test_codesize_minimal_pthreads.json b/test/codesize/test_codesize_minimal_pthreads.json index d0651f8052264..d546c205a609e 100644 --- a/test/codesize/test_codesize_minimal_pthreads.json +++ b/test/codesize/test_codesize_minimal_pthreads.json @@ -1,16 +1,16 @@ { - "a.out.js": 6883, - "a.out.js.gz": 3422, - "a.out.nodebug.wasm": 19132, - "a.out.nodebug.wasm.gz": 8845, - "total": 26015, - "total_gz": 12267, + "a.out.js": 6886, + "a.out.js.gz": 3424, + "a.out.nodebug.wasm": 19223, + "a.out.nodebug.wasm.gz": 8923, + "total": 26109, + "total_gz": 12347, "sent": [ "a (memory)", - "b (exit)", - "c (emscripten_get_now)", - "d (_emscripten_thread_set_strongref)", - "e (_emscripten_receive_on_main_thread_js)", + "b (_emscripten_receive_on_main_thread_js)", + "c (exit)", + "d (emscripten_get_now)", + "e (_emscripten_thread_set_strongref)", "f (emscripten_runtime_keepalive_check)", "g (emscripten_resize_heap)", "h (emscripten_exit_with_live_runtime)", @@ -23,10 +23,10 @@ ], "imports": [ "a (memory)", - "b (exit)", - "c (emscripten_get_now)", - "d (_emscripten_thread_set_strongref)", - "e (_emscripten_receive_on_main_thread_js)", + "b (_emscripten_receive_on_main_thread_js)", + "c (exit)", + "d (emscripten_get_now)", + "e (_emscripten_thread_set_strongref)", "f (emscripten_runtime_keepalive_check)", "g (emscripten_resize_heap)", "h (emscripten_exit_with_live_runtime)", @@ -60,8 +60,10 @@ "z (_emscripten_run_js_on_main_thread_done)" ], "funcs": [ + "$__do_cleanup_pop", "$__errno_location", "$__memcpy", + "$__pthread_cond_timedwait", "$__pthread_getspecific", "$__pthread_mutex_lock", "$__pthread_mutex_trylock", @@ -100,6 +102,7 @@ "$_emscripten_tls_init", "$_emscripten_yield", "$_main_thread", + "$_pthread_cleanup_push", "$a_cas", "$a_cas_p", "$a_dec", @@ -108,16 +111,13 @@ "$a_inc", "$a_store", "$add", - "$call_callback_then_free_ctx", "$call_cancel_then_free_ctx", - "$call_then_finish_task", "$call_with_ctx", "$cancel_active_ctxs", "$cancel_ctx", "$cancel_notification", "$dispose_chunk", "$do_proxy", - "$em_proxying_ctx_deinit", "$em_task_queue_cancel", "$em_task_queue_create", "$em_task_queue_dequeue", @@ -129,8 +129,6 @@ "$emscripten_builtin_malloc", "$emscripten_futex_wait", "$emscripten_futex_wake", - "$emscripten_proxy_finish", - "$emscripten_proxy_sync_with_ctx", "$emscripten_stack_get_current", "$emscripten_stack_set_limits", "$free_ctx", @@ -142,14 +140,17 @@ "$lock", "$main", "$nodtor", + "$orphan_sync_ctx", "$pthread_cond_signal", "$pthread_mutex_destroy", "$pthread_setspecific", "$receive_notification", - "$remove_active_ctx", "$run_js_func", + "$run_js_func_sync_with_ctx", "$run_js_func_with_ctx", "$sbrk", + "$sync_ctx_complete_locked", + "$sync_ctx_unref", "$undo", "$unlock" ] diff --git a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json index c258456676ca6..aa3746254dec2 100644 --- a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json +++ b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json @@ -1,16 +1,16 @@ { - "a.out.js": 7341, - "a.out.js.gz": 3639, - "a.out.nodebug.wasm": 19133, - "a.out.nodebug.wasm.gz": 8846, - "total": 26474, - "total_gz": 12485, + "a.out.js": 7344, + "a.out.js.gz": 3641, + "a.out.nodebug.wasm": 19224, + "a.out.nodebug.wasm.gz": 8923, + "total": 26568, + "total_gz": 12564, "sent": [ "a (memory)", - "b (exit)", - "c (emscripten_get_now)", - "d (_emscripten_thread_set_strongref)", - "e (_emscripten_receive_on_main_thread_js)", + "b (_emscripten_receive_on_main_thread_js)", + "c (exit)", + "d (emscripten_get_now)", + "e (_emscripten_thread_set_strongref)", "f (emscripten_runtime_keepalive_check)", "g (emscripten_resize_heap)", "h (emscripten_exit_with_live_runtime)", @@ -23,10 +23,10 @@ ], "imports": [ "a (memory)", - "b (exit)", - "c (emscripten_get_now)", - "d (_emscripten_thread_set_strongref)", - "e (_emscripten_receive_on_main_thread_js)", + "b (_emscripten_receive_on_main_thread_js)", + "c (exit)", + "d (emscripten_get_now)", + "e (_emscripten_thread_set_strongref)", "f (emscripten_runtime_keepalive_check)", "g (emscripten_resize_heap)", "h (emscripten_exit_with_live_runtime)", @@ -60,8 +60,10 @@ "z (_emscripten_run_js_on_main_thread_done)" ], "funcs": [ + "$__do_cleanup_pop", "$__errno_location", "$__memcpy", + "$__pthread_cond_timedwait", "$__pthread_getspecific", "$__pthread_mutex_lock", "$__pthread_mutex_trylock", @@ -100,6 +102,7 @@ "$_emscripten_tls_init", "$_emscripten_yield", "$_main_thread", + "$_pthread_cleanup_push", "$a_cas", "$a_cas_p", "$a_dec", @@ -108,16 +111,13 @@ "$a_inc", "$a_store", "$add", - "$call_callback_then_free_ctx", "$call_cancel_then_free_ctx", - "$call_then_finish_task", "$call_with_ctx", "$cancel_active_ctxs", "$cancel_ctx", "$cancel_notification", "$dispose_chunk", "$do_proxy", - "$em_proxying_ctx_deinit", "$em_task_queue_cancel", "$em_task_queue_create", "$em_task_queue_dequeue", @@ -129,8 +129,6 @@ "$emscripten_builtin_malloc", "$emscripten_futex_wait", "$emscripten_futex_wake", - "$emscripten_proxy_finish", - "$emscripten_proxy_sync_with_ctx", "$emscripten_stack_get_current", "$emscripten_stack_set_limits", "$free_ctx", @@ -142,14 +140,17 @@ "$lock", "$main", "$nodtor", + "$orphan_sync_ctx", "$pthread_cond_signal", "$pthread_mutex_destroy", "$pthread_setspecific", "$receive_notification", - "$remove_active_ctx", "$run_js_func", + "$run_js_func_sync_with_ctx", "$run_js_func_with_ctx", "$sbrk", + "$sync_ctx_complete_locked", + "$sync_ctx_unref", "$undo", "$unlock" ] From 4b731fef3fe95b5777380d4d9c8822407da7ae2d Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Fri, 28 Aug 2026 18:59:22 -0700 Subject: [PATCH 5/8] Expose emscripten_proxy_release_arg/acquire_arg as public API --- system/include/emscripten/proxying.h | 14 +++++++ system/lib/pthread/proxying.c | 41 +++++++++++++------ .../test_pthread_proxying_canceled_caller.c | 38 ++++++++++++++++- 3 files changed, 79 insertions(+), 14 deletions(-) diff --git a/system/include/emscripten/proxying.h b/system/include/emscripten/proxying.h index 16e7a94c2c5b7..d31c09b1653a9 100644 --- a/system/include/emscripten/proxying.h +++ b/system/include/emscripten/proxying.h @@ -49,6 +49,20 @@ typedef struct em_proxying_ctx em_proxying_ctx; // Signal the end of a task proxied with `emscripten_proxy_sync_with_ctx`. void emscripten_proxy_finish(em_proxying_ctx* ctx); +// Declare that a task proxied with `emscripten_proxy_sync_with_ctx` will no +// longer access `arg`. By default a canceled caller cannot exit until the task +// is finished, since `arg` may be on the caller's stack; after this call it +// may exit as soon as it is canceled, even though the task is still running. +// `emscripten_proxy_finish` implies this release. +void emscripten_proxy_release_arg(em_proxying_ctx* ctx); + +// Attempt to reverse `emscripten_proxy_release_arg`, re-pinning the caller so +// the task may access `arg` again (typically to write results back before +// finishing). Returns false if the caller has already been canceled and its +// `arg` is gone, in which case `arg` must not be accessed. After a successful +// acquire, a canceled caller once again waits for a release or finish. +bool emscripten_proxy_acquire_arg(em_proxying_ctx* ctx); + // Enqueue `func` on the given queue and thread and return immediately. Returns // true if the work was successfully enqueued and the target thread notified or // false otherwise. diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index 2ea9dd5bfcca3..6a5f9dabecc7b 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -209,10 +209,11 @@ struct em_proxying_ctx { enum ctx_state state; pthread_mutex_t mutex; pthread_cond_t cond; - // `arg` is caller-owned (typically on its stack). The target sets - // `arg_released` under `mutex` once it will no longer read it (implied by - // completion); a canceled caller waits for that before marking - // `caller_gone`, after which `arg` must not be touched at all. + // `arg` is caller-owned (typically on its stack). The task sets + // `arg_released` under `mutex` via `emscripten_proxy_release_arg` once + // it will no longer access it (implied by completion); a canceled caller + // waits for that before marking `caller_gone`, after which the task can + // no longer reacquire `arg` with `emscripten_proxy_acquire_arg`. bool arg_released; bool caller_gone; _Atomic int refs; @@ -385,6 +386,25 @@ static void call_callback_then_free_ctx(void* arg) { free_ctx(ctx); } +void emscripten_proxy_release_arg(em_proxying_ctx* ctx) { + assert(ctx->kind == SYNC); + pthread_mutex_lock(&ctx->sync.mutex); + ctx->sync.arg_released = true; + pthread_cond_signal(&ctx->sync.cond); + pthread_mutex_unlock(&ctx->sync.mutex); +} + +bool emscripten_proxy_acquire_arg(em_proxying_ctx* ctx) { + assert(ctx->kind == SYNC); + pthread_mutex_lock(&ctx->sync.mutex); + bool acquired = !ctx->sync.caller_gone; + if (acquired) { + ctx->sync.arg_released = false; + } + pthread_mutex_unlock(&ctx->sync.mutex); + return acquired; +} + void emscripten_proxy_finish(em_proxying_ctx* ctx) { if (ctx->kind == SYNC) { pthread_mutex_lock(&ctx->sync.mutex); @@ -470,7 +490,7 @@ bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, } pthread_cleanup_pop(0); pthread_mutex_unlock(&ctx->sync.mutex); - int ret = ctx->sync.state == DONE; + bool ret = ctx->sync.state == DONE; sync_ctx_unref(ctx); return ret; } @@ -701,10 +721,7 @@ static void run_js_func_with_ctx(em_proxying_ctx* ctx, void* arg) { // The arguments have been deserialized; the only later access to `f` is the // guarded result write in _emscripten_run_js_on_main_thread_done, so a // canceled caller may now leave. - pthread_mutex_lock(&ctx->sync.mutex); - ctx->sync.arg_released = true; - pthread_cond_signal(&ctx->sync.cond); - pthread_mutex_unlock(&ctx->sync.mutex); + emscripten_proxy_release_arg(ctx); } void _emscripten_run_js_on_main_thread_done(void* arg_ctx, @@ -714,12 +731,10 @@ void _emscripten_run_js_on_main_thread_done(void* arg_ctx, proxied_js_func_t* f = (proxied_js_func_t*)arg; // `f` lives on the caller's stack; it is gone if the caller was canceled // while waiting (e.g. a pthread_cancel of a thread blocked in recv/poll). - pthread_mutex_lock(&ctx->sync.mutex); - if (!ctx->sync.caller_gone) { + if (emscripten_proxy_acquire_arg(ctx)) { f->result = result; } - remove_active_ctx(ctx); - sync_ctx_complete_locked(ctx, DONE); + emscripten_proxy_finish(ctx); } // PROXY_SYNC: run the JS function to completion on the target thread, then diff --git a/test/pthread/test_pthread_proxying_canceled_caller.c b/test/pthread/test_pthread_proxying_canceled_caller.c index 5904a47ab945e..f09398176a90f 100644 --- a/test/pthread/test_pthread_proxying_canceled_caller.c +++ b/test/pthread/test_pthread_proxying_canceled_caller.c @@ -8,7 +8,9 @@ * work is (a) in progress on the target and (b) handed a ctx the target has * not finished yet. The caller must exit with PTHREAD_CANCELED, but only once * the target is done with the caller-owned argument (its stack), i.e. when the - * work completes or the ctx is finished. + * work completes or the ctx is finished, or (c) the task releases the argument + * with emscripten_proxy_release_arg, letting the canceled caller exit early, + * after which emscripten_proxy_acquire_arg must fail. */ #include @@ -70,6 +72,31 @@ void* ctx_caller(void* arg) { void finish_stashed(void* arg) { emscripten_proxy_finish(stashed); } +// (c) The task releases `arg` early: a canceled caller may then exit before +// the task is finished, after which `arg` can no longer be reacquired. +em_proxying_ctx* _Atomic released; + +void release_early(em_proxying_ctx* ctx, void* arg) { + // The caller is still waiting, so the arg can be acquired and accessed. + assert(emscripten_proxy_acquire_arg(ctx)); + assert(*(int*)arg == 42); + emscripten_proxy_release_arg(ctx); + released = ctx; +} + +void* release_caller(void* arg) { + int local = 42; + emscripten_proxy_sync_with_ctx(q, target, release_early, &local); + assert(false && "should have been canceled"); + return NULL; +} + +void finish_released(void* arg) { + // The canceled caller has exited, so the arg can no longer be acquired. + assert(!emscripten_proxy_acquire_arg(released)); + emscripten_proxy_finish(released); +} + void noop(void* arg) {} int main(void) { @@ -101,6 +128,15 @@ int main(void) { assert(ret == PTHREAD_CANCELED); assert(ctx_caller_exiting); + assert(pthread_create(&caller, NULL, release_caller, NULL) == 0); + while (!released) + usleep(1000); + assert(pthread_cancel(caller) == 0); + // The caller can exit even though the ctx is not yet finished. + assert(pthread_join(caller, &ret) == 0); + assert(ret == PTHREAD_CANCELED); + assert(emscripten_proxy_async(q, target, finish_released, NULL)); + // The queue is still healthy for ordinary work. assert(emscripten_proxy_sync(q, target, noop, NULL)); From fddfea008def5d760ca8023143ad4d65d9c82a55 Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Mon, 31 Aug 2026 16:31:37 -0700 Subject: [PATCH 6/8] Replace sync ctx mutex/cond/refcount with a pooled atomic state word Sync proxying ctxs are now allocated from a per-thread free list and coordinate through a single atomic lifecycle word that the caller futex-waits on directly, removing the per-call heap allocation and mutex/cond operations. emscripten_proxy_release_arg and emscripten_proxy_acquire_arg are exposed as public API implemented as transitions on that word, and a caller canceled before its task starts now exits immediately with the task dropped. --- system/include/emscripten/proxying.h | 17 +- system/lib/pthread/proxying.c | 228 ++++++++++-------- system/lib/pthread/proxying_stub.c | 4 + .../test_codesize_minimal_pthreads.json | 10 +- ...t_codesize_minimal_pthreads_memgrowth.json | 10 +- 5 files changed, 156 insertions(+), 113 deletions(-) diff --git a/system/include/emscripten/proxying.h b/system/include/emscripten/proxying.h index d31c09b1653a9..29d2b99d057a0 100644 --- a/system/include/emscripten/proxying.h +++ b/system/include/emscripten/proxying.h @@ -50,9 +50,9 @@ typedef struct em_proxying_ctx em_proxying_ctx; void emscripten_proxy_finish(em_proxying_ctx* ctx); // Declare that a task proxied with `emscripten_proxy_sync_with_ctx` will no -// longer access `arg`. By default a canceled caller cannot exit until the task -// is finished, since `arg` may be on the caller's stack; after this call it -// may exit as soon as it is canceled, even though the task is still running. +// longer access `arg`. By default a canceled caller cannot exit while the task +// is running, since `arg` may be on the caller's stack; after this call it may +// exit as soon as it is canceled, even though the task has not finished. // `emscripten_proxy_finish` implies this release. void emscripten_proxy_release_arg(em_proxying_ctx* ctx); @@ -75,8 +75,9 @@ bool emscripten_proxy_async(em_proxying_queue* q, // executing before returning. Returns true if the task was successfully // completed and false otherwise, including if the target thread is canceled or // exits before the work is completed. The wait is a cancellation point: a -// canceled caller exits with PTHREAD_CANCELED, but only after `func` has -// completed, since `arg` may be on the caller's stack. +// canceled caller exits with PTHREAD_CANCELED once `func` has completed (or +// immediately if `func` has not yet started, in which case it is dropped), +// since `arg` may be on the caller's stack. bool emscripten_proxy_sync(em_proxying_queue* q, pthread_t target_thread, void (*func)(void*), @@ -89,8 +90,10 @@ bool emscripten_proxy_sync(em_proxying_queue* q, // arbitrary later time. Returns true if the task was successfully completed and // false otherwise, including if the target thread is canceled or exits before // the work is completed. The wait is a cancellation point: a canceled caller -// exits with PTHREAD_CANCELED, but only after the task is finished (or -// canceled), since `arg` may be on the caller's stack. +// exits with PTHREAD_CANCELED once the task is finished (or canceled), or +// earlier if the task has not started or has called +// `emscripten_proxy_release_arg`, since `arg` may be on the caller's stack. If +// the caller is canceled before the task starts, the task is dropped. bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, pthread_t target_thread, void (*func)(em_proxying_ctx*, void*), diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index 6a5f9dabecc7b..5cf8474c01cca 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -8,8 +8,10 @@ #include #include #include +#include #include #include +#include #include #include #include @@ -190,7 +192,21 @@ bool emscripten_proxy_async(em_proxying_queue* q, enum ctx_kind { SYNC, CALLBACK }; -enum ctx_state { PENDING, DONE, CANCELED }; +// Phases of the sync ctx state word. `arg` is caller-owned (typically on its +// stack), so the task may only access it in the ACTIVE phase, when the caller +// is pinned in its wait even if canceled. In the PENDING and RELEASED phases a +// canceled caller instead sets CTX_ORPHANED and exits immediately, handing +// ownership of the ctx to the target, which then recycles it when it reaches a +// terminal phase (and can no longer reach ACTIVE via +// `emscripten_proxy_acquire_arg`). +#define CTX_PENDING 0u // Enqueued; the task has not started. +#define CTX_ACTIVE 1u // The task may access `arg`; the caller is pinned. +#define CTX_RELEASED 2u // The task is running but may not access `arg`. +#define CTX_DONE 3u // Terminal: finished. +#define CTX_CANCELED 4u // Terminal: the target died before finishing. + +#define CTX_ORPHANED 0x100u +#define CTX_PHASE(s) ((s) & 0xffu) struct em_proxying_ctx { // The user-provided function and argument. @@ -199,24 +215,13 @@ struct em_proxying_ctx { enum ctx_kind kind; union { - // Context for synchronous proxying. Heap allocated and shared between the - // waiting caller and the target thread, each holding one reference, so - // that a caller canceled or exiting mid-wait leaves the target a valid - // ctx to finish (or cancel) later. + // Context for synchronous proxying. Allocated from a per-thread pool so + // that a caller canceled and exiting mid-wait leaves the target a valid + // ctx to finish (or cancel) and recycle later. struct { - // Update `state` and signal the condition variable once the proxied task - // is done or canceled. - enum ctx_state state; - pthread_mutex_t mutex; - pthread_cond_t cond; - // `arg` is caller-owned (typically on its stack). The task sets - // `arg_released` under `mutex` via `emscripten_proxy_release_arg` once - // it will no longer access it (implied by completion); a canceled caller - // waits for that before marking `caller_gone`, after which the task can - // no longer reacquire `arg` with `emscripten_proxy_acquire_arg`. - bool arg_released; - bool caller_gone; - _Atomic int refs; + // Single-word lifecycle for the sync handshake; the caller futex-waits + // on it directly. See the CTX_* phases above. + _Atomic uint32_t state; } sync; // Context for proxying with callbacks. @@ -240,15 +245,46 @@ struct em_proxying_ctx { static pthread_key_t active_ctxs; static pthread_once_t active_ctxs_once = PTHREAD_ONCE_INIT; +// A per-thread free list of sync ctxs, freed to the heap on thread exit. Sync +// ctxs cannot live on the caller's stack since a canceled caller may exit +// while the target still holds its ctx, and pooling avoids a heap allocation +// per proxied call. +static pthread_key_t ctx_pool; + static void cancel_ctx(void* arg); static void cancel_active_ctxs(void* arg); +static void free_ctx_pool(void* head) { + em_proxying_ctx* ctx = head; + while (ctx) { + em_proxying_ctx* next = ctx->next; + free(ctx); + ctx = next; + } +} + static void init_active_ctxs(void) { int ret = pthread_key_create(&active_ctxs, cancel_active_ctxs); assert(ret == 0); + ret = pthread_key_create(&ctx_pool, free_ctx_pool); + assert(ret == 0); (void)ret; } +static em_proxying_ctx* sync_ctx_alloc(void) { + em_proxying_ctx* ctx = pthread_getspecific(ctx_pool); + if (ctx) { + pthread_setspecific(ctx_pool, ctx->next); + return ctx; + } + return malloc(sizeof(em_proxying_ctx)); +} + +static void sync_ctx_free(em_proxying_ctx* ctx) { + ctx->next = pthread_getspecific(ctx_pool); + pthread_setspecific(ctx_pool, ctx); +} + static void add_active_ctx(em_proxying_ctx* ctx) { assert(ctx != NULL); em_proxying_ctx* head = pthread_getspecific(active_ctxs); @@ -302,21 +338,11 @@ static void cancel_active_ctxs(void* arg) { static void em_proxying_ctx_init_sync(em_proxying_ctx* ctx, void (*func)(em_proxying_ctx*, void*), void* arg) { - pthread_once(&active_ctxs_once, init_active_ctxs); - *ctx = (em_proxying_ctx){ - .func = func, - .arg = arg, - .kind = SYNC, - .sync = - { - .state = PENDING, - .mutex = PTHREAD_MUTEX_INITIALIZER, - .cond = PTHREAD_COND_INITIALIZER, - .arg_released = false, - .caller_gone = false, - .refs = 2, - }, - }; + ctx->func = func; + ctx->arg = arg; + ctx->kind = SYNC; + ctx->next = ctx->prev = NULL; + atomic_store(&ctx->sync.state, CTX_PENDING); } static void em_proxying_ctx_init_callback(em_proxying_ctx* ctx, @@ -341,41 +367,24 @@ static void em_proxying_ctx_init_callback(em_proxying_ctx* ctx, }; } -static void em_proxying_ctx_deinit(em_proxying_ctx* ctx) { - if (ctx->kind == SYNC) { - pthread_mutex_destroy(&ctx->sync.mutex); - pthread_cond_destroy(&ctx->sync.cond); - } - // TODO: We should probably have some kind of refcounting scheme to keep - // `queue` alive for callback ctxs. -} - -static void free_ctx(void* arg) { - em_proxying_ctx* ctx = arg; - em_proxying_ctx_deinit(ctx); - free(ctx); -} +// TODO: We should probably have some kind of refcounting scheme to keep +// `queue` alive for callback ctxs. +static void free_ctx(void* arg) { free(arg); } -static void sync_ctx_unref(em_proxying_ctx* ctx) { - assert(ctx->kind == SYNC); - if (atomic_fetch_sub(&ctx->sync.refs, 1) == 1) { - free_ctx(ctx); +// Publish a terminal phase and either recycle the ctx of a canceled caller or +// wake the waiting one. After this the target must not touch the ctx: waking +// a possibly recycled address is benign (futex waiters recheck), but nothing +// else would be. +static void sync_ctx_complete(em_proxying_ctx* ctx, uint32_t phase) { + uint32_t s = atomic_load(&ctx->sync.state); + while (!atomic_compare_exchange_weak( + &ctx->sync.state, &s, (s & CTX_ORPHANED) | phase)) { + } + if (s & CTX_ORPHANED) { + sync_ctx_free(ctx); + } else { + emscripten_futex_wake(&ctx->sync.state, 1); } -} - -// Complete a sync ctx with `mutex` held: publish the state, drop it from the -// target thread's active list and wake the caller. Unlocks and drops the -// target's reference. -static void sync_ctx_complete_locked(em_proxying_ctx* ctx, - enum ctx_state state) { - ctx->sync.state = state; - ctx->sync.arg_released = true; - // Signal must come before unlock to avoid emscripten_proxy_sync_with_ctx - // seeing the state as DONE and freeing the ctx before we call unlock. - // See https://github.com/emscripten-core/emscripten/pull/26582 - pthread_cond_signal(&ctx->sync.cond); - pthread_mutex_unlock(&ctx->sync.mutex); - sync_ctx_unref(ctx); } // Free the callback info on the same thread it was originally allocated on. @@ -388,28 +397,34 @@ static void call_callback_then_free_ctx(void* arg) { void emscripten_proxy_release_arg(em_proxying_ctx* ctx) { assert(ctx->kind == SYNC); - pthread_mutex_lock(&ctx->sync.mutex); - ctx->sync.arg_released = true; - pthread_cond_signal(&ctx->sync.cond); - pthread_mutex_unlock(&ctx->sync.mutex); + // The orphaned bit cannot be set in the ACTIVE phase, so a plain store + // cannot lose it. + assert(CTX_PHASE(atomic_load(&ctx->sync.state)) == CTX_ACTIVE); + atomic_store(&ctx->sync.state, CTX_RELEASED); + emscripten_futex_wake(&ctx->sync.state, 1); } bool emscripten_proxy_acquire_arg(em_proxying_ctx* ctx) { assert(ctx->kind == SYNC); - pthread_mutex_lock(&ctx->sync.mutex); - bool acquired = !ctx->sync.caller_gone; - if (acquired) { - ctx->sync.arg_released = false; + uint32_t s = atomic_load(&ctx->sync.state); + while (1) { + if (s & CTX_ORPHANED) { + return false; + } + if (CTX_PHASE(s) == CTX_ACTIVE) { + return true; + } + assert(CTX_PHASE(s) == CTX_RELEASED); + if (atomic_compare_exchange_weak(&ctx->sync.state, &s, CTX_ACTIVE)) { + return true; + } } - pthread_mutex_unlock(&ctx->sync.mutex); - return acquired; } void emscripten_proxy_finish(em_proxying_ctx* ctx) { if (ctx->kind == SYNC) { - pthread_mutex_lock(&ctx->sync.mutex); remove_active_ctx(ctx); - sync_ctx_complete_locked(ctx, DONE); + sync_ctx_complete(ctx, CTX_DONE); } else { // Schedule the callback on the caller thread. If the caller thread has // already died or dies before the callback is executed, then at least make @@ -432,8 +447,7 @@ static void call_cancel_then_free_ctx(void* arg) { static void cancel_ctx(void* arg) { em_proxying_ctx* ctx = arg; if (ctx->kind == SYNC) { - pthread_mutex_lock(&ctx->sync.mutex); - sync_ctx_complete_locked(ctx, CANCELED); + sync_ctx_complete(ctx, CTX_CANCELED); } else { if (ctx->cb.cancel == NULL || !do_proxy(ctx->cb.queue, @@ -447,23 +461,41 @@ static void cancel_ctx(void* arg) { // Helper for wrapping the call with ctx as a `void (*)(void*)`. static void call_with_ctx(void* arg) { em_proxying_ctx* ctx = arg; + if (ctx->kind == SYNC) { + uint32_t expected = CTX_PENDING; + if (!atomic_compare_exchange_strong( + &ctx->sync.state, &expected, CTX_ACTIVE)) { + // The caller was canceled before the task started, so its `arg` is gone + // and no one wants the result; drop the work. + assert(expected == (CTX_PENDING | CTX_ORPHANED)); + sync_ctx_free(ctx); + return; + } + } add_active_ctx(ctx); ctx->func(ctx, ctx->arg); } -// Cancellation cleanup for a caller unwound out of the wait below. Runs with -// `mutex` held (the cond wait re-acquires it before acting on cancellation) -// and with cancellation disabled, so waiting here cannot recurse. Hold the -// caller's stack alive until the target no longer reads `arg`, then hand the -// ctx over to the target. +// Cancellation cleanup for a caller unwound out of the wait below. Hold the +// caller's stack alive while the task may access `arg` (the ACTIVE phase), +// then hand the ctx over to the target, or recycle it if the task already +// reached a terminal phase. Cancellation is disabled during exit, so waiting +// here cannot recurse. static void orphan_sync_ctx(void* arg) { em_proxying_ctx* ctx = arg; - while (!ctx->sync.arg_released) { - pthread_cond_wait(&ctx->sync.cond, &ctx->sync.mutex); + uint32_t s = atomic_load(&ctx->sync.state); + while (1) { + if (CTX_PHASE(s) == CTX_ACTIVE) { + emscripten_futex_wait(&ctx->sync.state, s, INFINITY); + s = atomic_load(&ctx->sync.state); + } else if (CTX_PHASE(s) >= CTX_DONE) { + sync_ctx_free(ctx); + return; + } else if (atomic_compare_exchange_weak( + &ctx->sync.state, &s, s | CTX_ORPHANED)) { + return; + } } - ctx->sync.caller_gone = true; - pthread_mutex_unlock(&ctx->sync.mutex); - sync_ctx_unref(ctx); } bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, @@ -472,26 +504,26 @@ bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, void* arg) { assert(!pthread_equal(target_thread, pthread_self()) && "Cannot synchronously wait for work proxied to the current thread"); - em_proxying_ctx* ctx = malloc(sizeof(em_proxying_ctx)); + pthread_once(&active_ctxs_once, init_active_ctxs); + em_proxying_ctx* ctx = sync_ctx_alloc(); if (!ctx) { return false; } em_proxying_ctx_init_sync(ctx, func, arg); if (!do_proxy(q, target_thread, (task){call_with_ctx, cancel_ctx, ctx})) { - free_ctx(ctx); + sync_ctx_free(ctx); return false; } - pthread_mutex_lock(&ctx->sync.mutex); // The wait is a cancellation point: a canceled caller exits from inside it, // so hand the ctx over to the target rather than leaving it dangling. + uint32_t s; pthread_cleanup_push(orphan_sync_ctx, ctx); - while (ctx->sync.state == PENDING) { - pthread_cond_wait(&ctx->sync.cond, &ctx->sync.mutex); + while (CTX_PHASE(s = atomic_load(&ctx->sync.state)) < CTX_DONE) { + emscripten_futex_wait(&ctx->sync.state, s, INFINITY); } pthread_cleanup_pop(0); - pthread_mutex_unlock(&ctx->sync.mutex); - bool ret = ctx->sync.state == DONE; - sync_ctx_unref(ctx); + bool ret = CTX_PHASE(s) == CTX_DONE; + sync_ctx_free(ctx); return ret; } diff --git a/system/lib/pthread/proxying_stub.c b/system/lib/pthread/proxying_stub.c index 0ebb97c2375d9..c97c3863d940d 100644 --- a/system/lib/pthread/proxying_stub.c +++ b/system/lib/pthread/proxying_stub.c @@ -25,6 +25,10 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) { abort(); } void emscripten_proxy_finish(em_proxying_ctx* ctx) { abort(); } +void emscripten_proxy_release_arg(em_proxying_ctx* ctx) { abort(); } + +bool emscripten_proxy_acquire_arg(em_proxying_ctx* ctx) { abort(); } + bool emscripten_proxy_async(em_proxying_queue* q, pthread_t target_thread, void (*func)(void*), diff --git a/test/codesize/test_codesize_minimal_pthreads.json b/test/codesize/test_codesize_minimal_pthreads.json index d546c205a609e..3eb38e7897de1 100644 --- a/test/codesize/test_codesize_minimal_pthreads.json +++ b/test/codesize/test_codesize_minimal_pthreads.json @@ -1,10 +1,10 @@ { "a.out.js": 6886, "a.out.js.gz": 3424, - "a.out.nodebug.wasm": 19223, - "a.out.nodebug.wasm.gz": 8923, - "total": 26109, - "total_gz": 12347, + "a.out.nodebug.wasm": 19368, + "a.out.nodebug.wasm.gz": 8964, + "total": 26254, + "total_gz": 12388, "sent": [ "a (memory)", "b (_emscripten_receive_on_main_thread_js)", @@ -111,6 +111,7 @@ "$a_inc", "$a_store", "$add", + "$call_callback_then_free_ctx", "$call_cancel_then_free_ctx", "$call_with_ctx", "$cancel_active_ctxs", @@ -145,6 +146,7 @@ "$pthread_mutex_destroy", "$pthread_setspecific", "$receive_notification", + "$remove_active_ctx", "$run_js_func", "$run_js_func_sync_with_ctx", "$run_js_func_with_ctx", diff --git a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json index aa3746254dec2..59fd33877388d 100644 --- a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json +++ b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json @@ -1,10 +1,10 @@ { "a.out.js": 7344, "a.out.js.gz": 3641, - "a.out.nodebug.wasm": 19224, - "a.out.nodebug.wasm.gz": 8923, - "total": 26568, - "total_gz": 12564, + "a.out.nodebug.wasm": 19369, + "a.out.nodebug.wasm.gz": 8964, + "total": 26713, + "total_gz": 12605, "sent": [ "a (memory)", "b (_emscripten_receive_on_main_thread_js)", @@ -111,6 +111,7 @@ "$a_inc", "$a_store", "$add", + "$call_callback_then_free_ctx", "$call_cancel_then_free_ctx", "$call_with_ctx", "$cancel_active_ctxs", @@ -145,6 +146,7 @@ "$pthread_mutex_destroy", "$pthread_setspecific", "$receive_notification", + "$remove_active_ctx", "$run_js_func", "$run_js_func_sync_with_ctx", "$run_js_func_with_ctx", From c5075962701412a180cea8c632e515bb61409c05 Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Mon, 31 Aug 2026 16:58:34 -0700 Subject: [PATCH 7/8] Simplify sync ctx state to ownership flags Collapse the five-phase lifecycle into three ownership flags (LOANED, DONE+OK, ORPHANED): task start and emscripten_proxy_acquire_arg become the same loan-taking transition, unifying the dropped-task and orphaned paths. Also removes the pthread cond/mutex machinery from minimal pthread builds entirely, coming out ~1KB smaller than main. --- system/lib/pthread/proxying.c | 100 ++++++++---------- .../test_codesize_minimal_pthreads.json | 91 ++++++++-------- ...t_codesize_minimal_pthreads_memgrowth.json | 91 ++++++++-------- 3 files changed, 130 insertions(+), 152 deletions(-) diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index 5cf8474c01cca..8b14caa0fc6a0 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -192,21 +192,19 @@ bool emscripten_proxy_async(em_proxying_queue* q, enum ctx_kind { SYNC, CALLBACK }; -// Phases of the sync ctx state word. `arg` is caller-owned (typically on its -// stack), so the task may only access it in the ACTIVE phase, when the caller -// is pinned in its wait even if canceled. In the PENDING and RELEASED phases a -// canceled caller instead sets CTX_ORPHANED and exits immediately, handing -// ownership of the ctx to the target, which then recycles it when it reaches a -// terminal phase (and can no longer reach ACTIVE via -// `emscripten_proxy_acquire_arg`). -#define CTX_PENDING 0u // Enqueued; the task has not started. -#define CTX_ACTIVE 1u // The task may access `arg`; the caller is pinned. -#define CTX_RELEASED 2u // The task is running but may not access `arg`. -#define CTX_DONE 3u // Terminal: finished. -#define CTX_CANCELED 4u // Terminal: the target died before finishing. - -#define CTX_ORPHANED 0x100u -#define CTX_PHASE(s) ((s) & 0xffu) +// The sync ctx state word tracks ownership with three flags. `arg` is +// caller-owned (typically on its stack) and loaned to the task by +// `emscripten_proxy_acquire_arg` (which is also how the task starts) until +// `emscripten_proxy_release_arg` or completion. While the loan is outstanding +// a canceled caller is pinned in its wait; otherwise it sets CTX_ORPHANED and +// exits immediately, handing ownership of the ctx to the target to recycle at +// completion (and voiding all future loans, including a task not yet started, +// which is dropped). CTX_LOANED, CTX_DONE, and CTX_ORPHANED are mutually +// exclusive. +#define CTX_LOANED 1u // The task may access `arg`; the caller is pinned. +#define CTX_DONE 2u // The call is over; with CTX_OK unless the target died. +#define CTX_OK 4u +#define CTX_ORPHANED 8u // The canceled caller is gone; `arg` with it. struct em_proxying_ctx { // The user-provided function and argument. @@ -220,7 +218,7 @@ struct em_proxying_ctx { // ctx to finish (or cancel) and recycle later. struct { // Single-word lifecycle for the sync handshake; the caller futex-waits - // on it directly. See the CTX_* phases above. + // on it directly. See the CTX_* flags above. _Atomic uint32_t state; } sync; @@ -342,7 +340,7 @@ static void em_proxying_ctx_init_sync(em_proxying_ctx* ctx, ctx->arg = arg; ctx->kind = SYNC; ctx->next = ctx->prev = NULL; - atomic_store(&ctx->sync.state, CTX_PENDING); + atomic_store(&ctx->sync.state, 0); } static void em_proxying_ctx_init_callback(em_proxying_ctx* ctx, @@ -371,15 +369,16 @@ static void em_proxying_ctx_init_callback(em_proxying_ctx* ctx, // `queue` alive for callback ctxs. static void free_ctx(void* arg) { free(arg); } -// Publish a terminal phase and either recycle the ctx of a canceled caller or -// wake the waiting one. After this the target must not touch the ctx: waking -// a possibly recycled address is benign (futex waiters recheck), but nothing -// else would be. -static void sync_ctx_complete(em_proxying_ctx* ctx, uint32_t phase) { +// Publish completion (releasing any outstanding loan) and either recycle the +// ctx of a canceled caller or wake the waiting one. After this the target must +// not touch the ctx: waking a possibly recycled address is benign (futex +// waiters recheck), but nothing else would be. +static void sync_ctx_complete(em_proxying_ctx* ctx, uint32_t flags) { uint32_t s = atomic_load(&ctx->sync.state); while (!atomic_compare_exchange_weak( - &ctx->sync.state, &s, (s & CTX_ORPHANED) | phase)) { + &ctx->sync.state, &s, (s & CTX_ORPHANED) | flags)) { } + assert(!(s & CTX_DONE)); if (s & CTX_ORPHANED) { sync_ctx_free(ctx); } else { @@ -397,34 +396,33 @@ static void call_callback_then_free_ctx(void* arg) { void emscripten_proxy_release_arg(em_proxying_ctx* ctx) { assert(ctx->kind == SYNC); - // The orphaned bit cannot be set in the ACTIVE phase, so a plain store - // cannot lose it. - assert(CTX_PHASE(atomic_load(&ctx->sync.state)) == CTX_ACTIVE); - atomic_store(&ctx->sync.state, CTX_RELEASED); + // No other flag can be set while the loan is outstanding, so a plain store + // cannot lose one. + assert(atomic_load(&ctx->sync.state) == CTX_LOANED); + atomic_store(&ctx->sync.state, 0); emscripten_futex_wake(&ctx->sync.state, 1); } bool emscripten_proxy_acquire_arg(em_proxying_ctx* ctx) { assert(ctx->kind == SYNC); uint32_t s = atomic_load(&ctx->sync.state); - while (1) { + do { + assert(!(s & CTX_DONE)); if (s & CTX_ORPHANED) { return false; } - if (CTX_PHASE(s) == CTX_ACTIVE) { - return true; - } - assert(CTX_PHASE(s) == CTX_RELEASED); - if (atomic_compare_exchange_weak(&ctx->sync.state, &s, CTX_ACTIVE)) { + if (s & CTX_LOANED) { return true; } - } + } while ( + !atomic_compare_exchange_weak(&ctx->sync.state, &s, s | CTX_LOANED)); + return true; } void emscripten_proxy_finish(em_proxying_ctx* ctx) { if (ctx->kind == SYNC) { remove_active_ctx(ctx); - sync_ctx_complete(ctx, CTX_DONE); + sync_ctx_complete(ctx, CTX_DONE | CTX_OK); } else { // Schedule the callback on the caller thread. If the caller thread has // already died or dies before the callback is executed, then at least make @@ -447,7 +445,7 @@ static void call_cancel_then_free_ctx(void* arg) { static void cancel_ctx(void* arg) { em_proxying_ctx* ctx = arg; if (ctx->kind == SYNC) { - sync_ctx_complete(ctx, CTX_CANCELED); + sync_ctx_complete(ctx, CTX_DONE); } else { if (ctx->cb.cancel == NULL || !do_proxy(ctx->cb.queue, @@ -461,34 +459,28 @@ static void cancel_ctx(void* arg) { // Helper for wrapping the call with ctx as a `void (*)(void*)`. static void call_with_ctx(void* arg) { em_proxying_ctx* ctx = arg; - if (ctx->kind == SYNC) { - uint32_t expected = CTX_PENDING; - if (!atomic_compare_exchange_strong( - &ctx->sync.state, &expected, CTX_ACTIVE)) { - // The caller was canceled before the task started, so its `arg` is gone - // and no one wants the result; drop the work. - assert(expected == (CTX_PENDING | CTX_ORPHANED)); - sync_ctx_free(ctx); - return; - } + // A sync task starts by taking the loan of `arg`; if the caller was already + // canceled its `arg` is gone and no one wants the result, so drop the work. + if (ctx->kind == SYNC && !emscripten_proxy_acquire_arg(ctx)) { + sync_ctx_free(ctx); + return; } add_active_ctx(ctx); ctx->func(ctx, ctx->arg); } // Cancellation cleanup for a caller unwound out of the wait below. Hold the -// caller's stack alive while the task may access `arg` (the ACTIVE phase), -// then hand the ctx over to the target, or recycle it if the task already -// reached a terminal phase. Cancellation is disabled during exit, so waiting -// here cannot recurse. +// caller's stack alive while the loan of `arg` is outstanding, then hand the +// ctx over to the target, or recycle it if the call already completed. +// Cancellation is disabled during exit, so waiting here cannot recurse. static void orphan_sync_ctx(void* arg) { em_proxying_ctx* ctx = arg; uint32_t s = atomic_load(&ctx->sync.state); while (1) { - if (CTX_PHASE(s) == CTX_ACTIVE) { + if (s & CTX_LOANED) { emscripten_futex_wait(&ctx->sync.state, s, INFINITY); s = atomic_load(&ctx->sync.state); - } else if (CTX_PHASE(s) >= CTX_DONE) { + } else if (s & CTX_DONE) { sync_ctx_free(ctx); return; } else if (atomic_compare_exchange_weak( @@ -518,11 +510,11 @@ bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, // so hand the ctx over to the target rather than leaving it dangling. uint32_t s; pthread_cleanup_push(orphan_sync_ctx, ctx); - while (CTX_PHASE(s = atomic_load(&ctx->sync.state)) < CTX_DONE) { + while (!((s = atomic_load(&ctx->sync.state)) & CTX_DONE)) { emscripten_futex_wait(&ctx->sync.state, s, INFINITY); } pthread_cleanup_pop(0); - bool ret = CTX_PHASE(s) == CTX_DONE; + bool ret = s & CTX_OK; sync_ctx_free(ctx); return ret; } diff --git a/test/codesize/test_codesize_minimal_pthreads.json b/test/codesize/test_codesize_minimal_pthreads.json index 3eb38e7897de1..0c0c7ad615d09 100644 --- a/test/codesize/test_codesize_minimal_pthreads.json +++ b/test/codesize/test_codesize_minimal_pthreads.json @@ -1,10 +1,10 @@ { - "a.out.js": 6886, - "a.out.js.gz": 3424, - "a.out.nodebug.wasm": 19368, - "a.out.nodebug.wasm.gz": 8964, - "total": 26254, - "total_gz": 12388, + "a.out.js": 6870, + "a.out.js.gz": 3417, + "a.out.nodebug.wasm": 18128, + "a.out.nodebug.wasm.gz": 8400, + "total": 24998, + "total_gz": 11817, "sent": [ "a (memory)", "b (_emscripten_receive_on_main_thread_js)", @@ -14,12 +14,11 @@ "f (emscripten_runtime_keepalive_check)", "g (emscripten_resize_heap)", "h (emscripten_exit_with_live_runtime)", - "i (emscripten_check_blocking_allowed)", - "j (_emscripten_thread_mailbox_await)", - "k (_emscripten_thread_cleanup)", - "l (_emscripten_notify_mailbox_postmessage)", - "m (_emscripten_init_main_thread_js)", - "n (__pthread_create_js)" + "i (_emscripten_thread_mailbox_await)", + "j (_emscripten_thread_cleanup)", + "k (_emscripten_notify_mailbox_postmessage)", + "l (_emscripten_init_main_thread_js)", + "m (__pthread_create_js)" ], "imports": [ "a (memory)", @@ -30,41 +29,40 @@ "f (emscripten_runtime_keepalive_check)", "g (emscripten_resize_heap)", "h (emscripten_exit_with_live_runtime)", - "i (emscripten_check_blocking_allowed)", - "j (_emscripten_thread_mailbox_await)", - "k (_emscripten_thread_cleanup)", - "l (_emscripten_notify_mailbox_postmessage)", - "m (_emscripten_init_main_thread_js)", - "n (__pthread_create_js)" + "i (_emscripten_thread_mailbox_await)", + "j (_emscripten_thread_cleanup)", + "k (_emscripten_notify_mailbox_postmessage)", + "l (_emscripten_init_main_thread_js)", + "m (__pthread_create_js)" ], "exports": [ - "A (_emscripten_run_js_on_main_thread)", - "B (_emscripten_thread_free_data)", - "C (_emscripten_thread_exit)", - "D (_emscripten_check_mailbox)", - "E (emscripten_stack_set_limits)", - "F (_emscripten_stack_restore)", - "G (_emscripten_stack_alloc)", - "H (emscripten_stack_get_current)", - "o (__wasm_call_ctors)", - "p (add)", - "q (main)", - "r (global_val)", - "s (__indirect_function_table)", - "t (_emscripten_tls_init)", - "u (pthread_self)", - "v (_emscripten_proxy_main)", - "w (_emscripten_thread_init)", - "x (__set_thread_state)", - "y (_emscripten_thread_crashed)", - "z (_emscripten_run_js_on_main_thread_done)" + "A (_emscripten_thread_free_data)", + "B (_emscripten_thread_exit)", + "C (_emscripten_check_mailbox)", + "D (emscripten_stack_set_limits)", + "E (_emscripten_stack_restore)", + "F (_emscripten_stack_alloc)", + "G (emscripten_stack_get_current)", + "n (__wasm_call_ctors)", + "o (add)", + "p (main)", + "q (global_val)", + "r (__indirect_function_table)", + "s (_emscripten_tls_init)", + "t (pthread_self)", + "u (_emscripten_proxy_main)", + "v (_emscripten_thread_init)", + "w (__set_thread_state)", + "x (_emscripten_thread_crashed)", + "y (_emscripten_run_js_on_main_thread_done)", + "z (_emscripten_run_js_on_main_thread)" ], "funcs": [ "$__do_cleanup_pop", "$__errno_location", "$__memcpy", - "$__pthread_cond_timedwait", "$__pthread_getspecific", + "$__pthread_key_create", "$__pthread_mutex_lock", "$__pthread_mutex_trylock", "$__pthread_mutex_unlock", @@ -76,7 +74,6 @@ "$__pthread_setcancelstate", "$__set_thread_state", "$__timedwait", - "$__timedwait_cp", "$__tl_lock", "$__tl_unlock", "$__vm_lock", @@ -107,7 +104,6 @@ "$a_cas_p", "$a_dec", "$a_fetch_add", - "$a_fetch_add", "$a_inc", "$a_store", "$add", @@ -130,30 +126,27 @@ "$emscripten_builtin_malloc", "$emscripten_futex_wait", "$emscripten_futex_wake", + "$emscripten_proxy_acquire_arg", "$emscripten_stack_get_current", "$emscripten_stack_set_limits", "$free_ctx", + "$free_ctx_pool", "$get_or_add_tasks_for_thread", "$get_tasks_for_thread", "$init_active_ctxs", "$init_file_lock", "$init_mparams", - "$lock", "$main", "$nodtor", "$orphan_sync_ctx", - "$pthread_cond_signal", - "$pthread_mutex_destroy", "$pthread_setspecific", "$receive_notification", - "$remove_active_ctx", "$run_js_func", "$run_js_func_sync_with_ctx", "$run_js_func_with_ctx", "$sbrk", - "$sync_ctx_complete_locked", - "$sync_ctx_unref", - "$undo", - "$unlock" + "$sync_ctx_complete", + "$sync_ctx_free", + "$undo" ] } diff --git a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json index 59fd33877388d..3902ca822ab42 100644 --- a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json +++ b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json @@ -1,10 +1,10 @@ { - "a.out.js": 7344, - "a.out.js.gz": 3641, - "a.out.nodebug.wasm": 19369, - "a.out.nodebug.wasm.gz": 8964, - "total": 26713, - "total_gz": 12605, + "a.out.js": 7329, + "a.out.js.gz": 3634, + "a.out.nodebug.wasm": 18129, + "a.out.nodebug.wasm.gz": 8401, + "total": 25458, + "total_gz": 12035, "sent": [ "a (memory)", "b (_emscripten_receive_on_main_thread_js)", @@ -14,12 +14,11 @@ "f (emscripten_runtime_keepalive_check)", "g (emscripten_resize_heap)", "h (emscripten_exit_with_live_runtime)", - "i (emscripten_check_blocking_allowed)", - "j (_emscripten_thread_mailbox_await)", - "k (_emscripten_thread_cleanup)", - "l (_emscripten_notify_mailbox_postmessage)", - "m (_emscripten_init_main_thread_js)", - "n (__pthread_create_js)" + "i (_emscripten_thread_mailbox_await)", + "j (_emscripten_thread_cleanup)", + "k (_emscripten_notify_mailbox_postmessage)", + "l (_emscripten_init_main_thread_js)", + "m (__pthread_create_js)" ], "imports": [ "a (memory)", @@ -30,41 +29,40 @@ "f (emscripten_runtime_keepalive_check)", "g (emscripten_resize_heap)", "h (emscripten_exit_with_live_runtime)", - "i (emscripten_check_blocking_allowed)", - "j (_emscripten_thread_mailbox_await)", - "k (_emscripten_thread_cleanup)", - "l (_emscripten_notify_mailbox_postmessage)", - "m (_emscripten_init_main_thread_js)", - "n (__pthread_create_js)" + "i (_emscripten_thread_mailbox_await)", + "j (_emscripten_thread_cleanup)", + "k (_emscripten_notify_mailbox_postmessage)", + "l (_emscripten_init_main_thread_js)", + "m (__pthread_create_js)" ], "exports": [ - "A (_emscripten_run_js_on_main_thread)", - "B (_emscripten_thread_free_data)", - "C (_emscripten_thread_exit)", - "D (_emscripten_check_mailbox)", - "E (emscripten_stack_set_limits)", - "F (_emscripten_stack_restore)", - "G (_emscripten_stack_alloc)", - "H (emscripten_stack_get_current)", - "o (__wasm_call_ctors)", - "p (add)", - "q (main)", - "r (global_val)", - "s (__indirect_function_table)", - "t (_emscripten_tls_init)", - "u (pthread_self)", - "v (_emscripten_proxy_main)", - "w (_emscripten_thread_init)", - "x (__set_thread_state)", - "y (_emscripten_thread_crashed)", - "z (_emscripten_run_js_on_main_thread_done)" + "A (_emscripten_thread_free_data)", + "B (_emscripten_thread_exit)", + "C (_emscripten_check_mailbox)", + "D (emscripten_stack_set_limits)", + "E (_emscripten_stack_restore)", + "F (_emscripten_stack_alloc)", + "G (emscripten_stack_get_current)", + "n (__wasm_call_ctors)", + "o (add)", + "p (main)", + "q (global_val)", + "r (__indirect_function_table)", + "s (_emscripten_tls_init)", + "t (pthread_self)", + "u (_emscripten_proxy_main)", + "v (_emscripten_thread_init)", + "w (__set_thread_state)", + "x (_emscripten_thread_crashed)", + "y (_emscripten_run_js_on_main_thread_done)", + "z (_emscripten_run_js_on_main_thread)" ], "funcs": [ "$__do_cleanup_pop", "$__errno_location", "$__memcpy", - "$__pthread_cond_timedwait", "$__pthread_getspecific", + "$__pthread_key_create", "$__pthread_mutex_lock", "$__pthread_mutex_trylock", "$__pthread_mutex_unlock", @@ -76,7 +74,6 @@ "$__pthread_setcancelstate", "$__set_thread_state", "$__timedwait", - "$__timedwait_cp", "$__tl_lock", "$__tl_unlock", "$__vm_lock", @@ -107,7 +104,6 @@ "$a_cas_p", "$a_dec", "$a_fetch_add", - "$a_fetch_add", "$a_inc", "$a_store", "$add", @@ -130,30 +126,27 @@ "$emscripten_builtin_malloc", "$emscripten_futex_wait", "$emscripten_futex_wake", + "$emscripten_proxy_acquire_arg", "$emscripten_stack_get_current", "$emscripten_stack_set_limits", "$free_ctx", + "$free_ctx_pool", "$get_or_add_tasks_for_thread", "$get_tasks_for_thread", "$init_active_ctxs", "$init_file_lock", "$init_mparams", - "$lock", "$main", "$nodtor", "$orphan_sync_ctx", - "$pthread_cond_signal", - "$pthread_mutex_destroy", "$pthread_setspecific", "$receive_notification", - "$remove_active_ctx", "$run_js_func", "$run_js_func_sync_with_ctx", "$run_js_func_with_ctx", "$sbrk", - "$sync_ctx_complete_locked", - "$sync_ctx_unref", - "$undo", - "$unlock" + "$sync_ctx_complete", + "$sync_ctx_free", + "$undo" ] } From d38b268b3b7874ec0d2fd830ef028d937072ece7 Mon Sep 17 00:00:00 2001 From: Guy Bedford Date: Mon, 31 Aug 2026 17:27:25 -0700 Subject: [PATCH 8/8] Use PTHREAD_CANCEL_MASKED for the sync proxy wait Mask cancellation around the wait as pthread_cond_timedwait does, so a cancel surfaces as ECANCELED and the ctx handover runs as straight-line code before exiting, rather than unwinding out of the wait through a cleanup handler. --- system/lib/pthread/proxying.c | 33 ++++++++++++------- .../test_codesize_minimal_pthreads.json | 11 +++---- ...t_codesize_minimal_pthreads_memgrowth.json | 11 +++---- 3 files changed, 30 insertions(+), 25 deletions(-) diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index 8b14caa0fc6a0..9d9784289fa1c 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -469,12 +470,12 @@ static void call_with_ctx(void* arg) { ctx->func(ctx, ctx->arg); } -// Cancellation cleanup for a caller unwound out of the wait below. Hold the -// caller's stack alive while the loan of `arg` is outstanding, then hand the -// ctx over to the target, or recycle it if the call already completed. -// Cancellation is disabled during exit, so waiting here cannot recurse. -static void orphan_sync_ctx(void* arg) { - em_proxying_ctx* ctx = arg; +// Handle cancellation of a caller waiting below. Hold the caller's stack +// alive while the loan of `arg` is outstanding, then hand the ctx over to the +// target, or recycle it if the call already completed. Runs after the masked +// cancel has set the cancel state to disabled, so waiting here cannot itself +// be canceled. +static void orphan_sync_ctx(em_proxying_ctx* ctx) { uint32_t s = atomic_load(&ctx->sync.state); while (1) { if (s & CTX_LOANED) { @@ -506,14 +507,24 @@ bool emscripten_proxy_sync_with_ctx(em_proxying_queue* q, sync_ctx_free(ctx); return false; } - // The wait is a cancellation point: a canceled caller exits from inside it, - // so hand the ctx over to the target rather than leaving it dangling. + // The wait is a cancellation point. Mask cancellation (as in + // pthread_cond_timedwait) so it surfaces as ECANCELED rather than unwinding + // the thread from inside the wait, hand the ctx over to the target, then + // exit. + int cs; + __pthread_setcancelstate(PTHREAD_CANCEL_MASKED, &cs); + if (cs == PTHREAD_CANCEL_DISABLE) { + __pthread_setcancelstate(cs, 0); + } uint32_t s; - pthread_cleanup_push(orphan_sync_ctx, ctx); while (!((s = atomic_load(&ctx->sync.state)) & CTX_DONE)) { - emscripten_futex_wait(&ctx->sync.state, s, INFINITY); + if (emscripten_futex_wait(&ctx->sync.state, s, INFINITY) == -ECANCELED) { + orphan_sync_ctx(ctx); + __pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, 0); + __pthread_testcancel(); + } } - pthread_cleanup_pop(0); + __pthread_setcancelstate(cs, 0); bool ret = s & CTX_OK; sync_ctx_free(ctx); return ret; diff --git a/test/codesize/test_codesize_minimal_pthreads.json b/test/codesize/test_codesize_minimal_pthreads.json index 0c0c7ad615d09..a25e9bcb58c35 100644 --- a/test/codesize/test_codesize_minimal_pthreads.json +++ b/test/codesize/test_codesize_minimal_pthreads.json @@ -1,10 +1,10 @@ { "a.out.js": 6870, "a.out.js.gz": 3417, - "a.out.nodebug.wasm": 18128, - "a.out.nodebug.wasm.gz": 8400, - "total": 24998, - "total_gz": 11817, + "a.out.nodebug.wasm": 18130, + "a.out.nodebug.wasm.gz": 8407, + "total": 25000, + "total_gz": 11824, "sent": [ "a (memory)", "b (_emscripten_receive_on_main_thread_js)", @@ -58,7 +58,6 @@ "z (_emscripten_run_js_on_main_thread)" ], "funcs": [ - "$__do_cleanup_pop", "$__errno_location", "$__memcpy", "$__pthread_getspecific", @@ -99,7 +98,6 @@ "$_emscripten_tls_init", "$_emscripten_yield", "$_main_thread", - "$_pthread_cleanup_push", "$a_cas", "$a_cas_p", "$a_dec", @@ -138,7 +136,6 @@ "$init_mparams", "$main", "$nodtor", - "$orphan_sync_ctx", "$pthread_setspecific", "$receive_notification", "$run_js_func", diff --git a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json index 3902ca822ab42..6ce303361bd01 100644 --- a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json +++ b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json @@ -1,10 +1,10 @@ { "a.out.js": 7329, "a.out.js.gz": 3634, - "a.out.nodebug.wasm": 18129, - "a.out.nodebug.wasm.gz": 8401, - "total": 25458, - "total_gz": 12035, + "a.out.nodebug.wasm": 18131, + "a.out.nodebug.wasm.gz": 8410, + "total": 25460, + "total_gz": 12044, "sent": [ "a (memory)", "b (_emscripten_receive_on_main_thread_js)", @@ -58,7 +58,6 @@ "z (_emscripten_run_js_on_main_thread)" ], "funcs": [ - "$__do_cleanup_pop", "$__errno_location", "$__memcpy", "$__pthread_getspecific", @@ -99,7 +98,6 @@ "$_emscripten_tls_init", "$_emscripten_yield", "$_main_thread", - "$_pthread_cleanup_push", "$a_cas", "$a_cas_p", "$a_dec", @@ -138,7 +136,6 @@ "$init_mparams", "$main", "$nodtor", - "$orphan_sync_ctx", "$pthread_setspecific", "$receive_notification", "$run_js_func",