From 679b8633f7ca319edeb21c316835f8703489e7d8 Mon Sep 17 00:00:00 2001 From: Thomas Lively Date: Fri, 24 Feb 2023 19:40:48 -0800 Subject: [PATCH 1/4] [Proxying] Send messages via in-memory mailbox queues Threads were previously notified of new work via postMessage messages that carried pointers to the task queues to execute. There was no way to synchronously pump or inspect these pending messages however, and there is no central registry of all task queues for a thread, so this mechanism afforded no way to discover or cancel pending work when a thread dies. In preparation for implementing work cancellation, move the pending messages into userspace by giving each thread a "mailbox", which is an `em_task_queue` in the pthread struct. Instead of using `postMessage`, proxying queues now use the thread mailbox API to notify threads of new work. Internally, thread mailboxes still use postMessage to schedule work to be executed when a thread returns to its event loop. Since the only task queues involved in postMessages are now at known locations relative to the pthread struct, there is no longer any need to store pointers to them in the postMessage messages themselves. Removing these pointers works around tricky notification and lifetime management edge cases that would have caused problems such as dropped work or use-after-free bugs in future PRs. When a thread dies because it exits or is canceled, it "closes" its mailbox by decrementing a refcount and waiting to observe a refcount of 0. At this point, the thread mailbox API ensures that no new messages will be enqueued on the mailbox. Because the postMessage messages no longer contain task queue pointers, it is safe to destroy the mailbox immediately after it is closed. A user-visible behavior change this introduces is that proxied work is more frequently completed _before_ a thread's main function begins running, since it no longer gets ordered behind the `run` message in the JS postMessage queue. A few tests are updated accordingly. --- emcc.py | 3 +- src/library_pthread.js | 33 ++--- src/worker.js | 20 +-- .../lib/libc/musl/src/internal/pthread_impl.h | 21 ++- system/lib/pthread/em_task_queue.c | 54 +++++--- system/lib/pthread/em_task_queue.h | 9 +- system/lib/pthread/library_pthread.c | 2 + system/lib/pthread/proxying.c | 23 +--- system/lib/pthread/pthread_create.c | 4 + system/lib/pthread/thread_mailbox.c | 120 ++++++++++++++++++ system/lib/pthread/thread_mailbox.h | 32 +++++ .../test_metadce_minimal_pthreads.funcs | 7 +- .../test_metadce_minimal_pthreads.jssize | 2 +- .../test_metadce_minimal_pthreads.size | 2 +- .../test_pthread_proxying_canceled_work.c | 84 ++++++++++++ .../test_pthread_proxying_canceled_work.out | 3 + .../test_pthread_proxying_dropped_work.c | 6 +- test/pthread/test_pthread_proxying_refcount.c | 13 +- test/reference_struct_info.json | 2 +- test/test_core.py | 8 ++ tools/system_libs.py | 1 + 21 files changed, 360 insertions(+), 89 deletions(-) create mode 100644 system/lib/pthread/thread_mailbox.c create mode 100644 system/lib/pthread/thread_mailbox.h create mode 100644 test/pthread/test_pthread_proxying_canceled_work.c create mode 100644 test/pthread/test_pthread_proxying_canceled_work.out diff --git a/emcc.py b/emcc.py index 5cbf038d37067..f5e040384264d 100755 --- a/emcc.py +++ b/emcc.py @@ -1712,7 +1712,8 @@ def setup_pthreads(target): '__emscripten_thread_crashed', '__emscripten_tls_init', '_pthread_self', - 'executeNotifiedProxyingQueue', + '__emscripten_check_mailbox', + 'checkMailbox', ] settings.EXPORTED_FUNCTIONS += worker_imports building.user_requested_exports.update(worker_imports) diff --git a/src/library_pthread.js b/src/library_pthread.js index af921a8e2a320..01fc0505489ff 100644 --- a/src/library_pthread.js +++ b/src/library_pthread.js @@ -270,8 +270,8 @@ var LibraryPThread = { return; } - if (cmd === 'processProxyingQueue') { - executeNotifiedProxyingQueue(d['queue']); + if (cmd === 'checkMailbox') { + checkMailbox(); } else if (cmd === 'spawnThread') { spawnThread(d); } else if (cmd === 'cleanupThread') { @@ -1212,29 +1212,22 @@ var LibraryPThread = { }, #endif // MAIN_MODULE - $executeNotifiedProxyingQueue__deps: ['$callUserCallback'], - $executeNotifiedProxyingQueue: function(queue) { - // Set the notification state to processing. - Atomics.store(HEAP32, queue >> 2, {{{ cDefine('NOTIFICATION_RECEIVED') }}}); - // Only execute the queue if we have a live pthread runtime. We - // implement pthread_self to return 0 if there is no live runtime. + $checkMailbox__deps: ['$callUserCallback'], + $checkMailbox: function() { + // Only check the mailbox if we have a live pthread runtime. We implement + // pthread_self to return 0 if there is no live runtime. if (_pthread_self()) { - callUserCallback(() => __emscripten_proxy_execute_task_queue(queue)); + callUserCallback(() => __emscripten_check_mailbox()); } - // Set the notification state to none as long as a new notification has not - // been sent while we were processing. - Atomics.compareExchange(HEAP32, queue >> 2, - {{{ cDefine('NOTIFICATION_RECEIVED') }}}, - {{{ cDefine('NOTIFICATION_NONE') }}}); }, - _emscripten_notify_task_queue__deps: ['$executeNotifiedProxyingQueue'], - _emscripten_notify_task_queue__sig: 'vpppp', - _emscripten_notify_task_queue: function(targetThreadId, currThreadId, mainThreadId, queue) { + _emscripten_notify_mailbox__deps: ['$checkMailbox'], + _emscripten_notify_mailbox__sig: 'vppp', + _emscripten_notify_mailbox: function(targetThreadId, currThreadId, mainThreadId) { if (targetThreadId == currThreadId) { - setTimeout(() => executeNotifiedProxyingQueue(queue)); + setTimeout(() => checkMailbox()); } else if (ENVIRONMENT_IS_PTHREAD) { - postMessage({'targetThread' : targetThreadId, 'cmd' : 'processProxyingQueue', 'queue' : queue}); + postMessage({'targetThread' : targetThreadId, 'cmd' : 'checkMailbox'}); } else { var worker = PThread.pthreads[targetThreadId]; if (!worker) { @@ -1243,7 +1236,7 @@ var LibraryPThread = { #endif return /*0*/; } - worker.postMessage({'cmd' : 'processProxyingQueue', 'queue': queue}); + worker.postMessage({'cmd' : 'checkMailbox'}); } } }; diff --git a/src/worker.js b/src/worker.js index c7a911206b2c0..38883ac60de21 100644 --- a/src/worker.js +++ b/src/worker.js @@ -52,10 +52,6 @@ if (ENVIRONMENT_IS_NODE) { // Thread-local guard variable for one-time init of the JS state var initializedJS = false; -// Proxying queues that were notified before the thread started and need to be -// executed as part of startup. -var pendingNotifiedProxyingQueues = []; - #if ASSERTIONS function assert(condition, text) { if (!condition) abort('Assertion failed: ' + text); @@ -240,12 +236,9 @@ function handleMessage(e) { // Execute any proxied work that came in before the thread was // initialized. Only do this once because it is only possible for - // proxying notifications to arrive before thread initialization on - // fresh workers. - pendingNotifiedProxyingQueues.forEach(queue => { - Module['executeNotifiedProxyingQueue'](queue); - }); - pendingNotifiedProxyingQueues = []; + // notifications to arrive before thread initialization on fresh + // workers. + Module['__emscripten_check_mailbox'](); initializedJS = true; } @@ -268,12 +261,9 @@ function handleMessage(e) { } } else if (e.data.target === 'setimmediate') { // no-op - } else if (e.data.cmd === 'processProxyingQueue') { + } else if (e.data.cmd === 'checkMailbox') { if (initializedJS) { - Module['executeNotifiedProxyingQueue'](e.data.queue); - } else { - // Defer executing this queue until the runtime is initialized. - pendingNotifiedProxyingQueues.push(e.data.queue); + Module['checkMailbox'](); } } else if (e.data.cmd) { // The received message looks like something that should be handled by this message diff --git a/system/lib/libc/musl/src/internal/pthread_impl.h b/system/lib/libc/musl/src/internal/pthread_impl.h index bb9df2fdf55f9..a34cd6210d086 100644 --- a/system/lib/libc/musl/src/internal/pthread_impl.h +++ b/system/lib/libc/musl/src/internal/pthread_impl.h @@ -10,8 +10,10 @@ #include "syscall.h" #include "atomic.h" #ifdef __EMSCRIPTEN__ -#include +#include "em_task_queue.h" +#include "thread_mailbox.h" #include "threading_internal.h" +#include #endif #include "futex.h" @@ -78,6 +80,23 @@ struct pthread { // The TLS base to use the main module TLS data. Secondary modules // still require dynamic allocation. void* tls_base; + // The lowest level of the proxying system. Other threads can enqueue + // messages on the mailbox and notify this thread to asynchronously + // process them once it returns to its event loop. When this thread is + // shut down, the mailbox is closed (see below) to prevent further + // messages from being enqueued and all the remaining queued messages + // are dequeued and their shutdown handlers are executed. This allows + // other threads waiting for their messages to be processed to be + // notified that their messages will not be processed after all. + em_task_queue* mailbox; + // To ensure that no other thread is concurrently enqueueing a message + // when this thread shuts down, maintain an atomic refcount. Enqueueing + // threads atomically increment the count from a nonzero number to + // acquire the mailbox and decrement the count when they finish. When + // this thread shuts down it will atomically decrement the count and + // wait until it reaches 0, at which point the mailbox is considered + // closed and no further messages will be enqueued. + _Atomic int mailbox_refcount; #endif #if _REENTRANT _Atomic char sleeping; diff --git a/system/lib/pthread/em_task_queue.c b/system/lib/pthread/em_task_queue.c index 80acf64d24d50..dd79c3bee65d3 100644 --- a/system/lib/pthread/em_task_queue.c +++ b/system/lib/pthread/em_task_queue.c @@ -13,6 +13,7 @@ #include "em_task_queue.h" #include "proxying_notification_state.h" +#include "thread_mailbox.h" #define EM_TASK_QUEUE_INITIAL_CAPACITY 128 @@ -194,26 +195,41 @@ task em_task_queue_dequeue(em_task_queue* queue) { return t; } -// Send a postMessage notification containing the em_task_queue pointer to the -// target thread so it will execute the queue when it returns to the event loop. -// Also pass in the current thread and main thread ids to minimize calls back -// into Wasm. -void _emscripten_notify_task_queue(pthread_t target_thread, - pthread_t curr_thread, - pthread_t main_thread, - em_task_queue* queue); - -void em_task_queue_notify(em_task_queue* queue) { - // If there is no pending notification for this queue, create one. If an old - // notification is currently being processed, it may or may not execute this - // work. In case it does not, the new notification will ensure the work is - // still executed. +static void receive_notification(void* arg) { + em_task_queue* tasks = arg; + tasks->notification = NOTIFICATION_RECEIVED; + em_task_queue_execute(tasks); + notification_state expected = NOTIFICATION_RECEIVED; + atomic_compare_exchange_strong( + &tasks->notification, &expected, NOTIFICATION_NONE); +} + +int em_task_queue_send(em_task_queue* queue, task t) { + // Ensure the target mailbox will remain open or detect that it is already + // closed. + if (!emscripten_thread_mailbox_ref(queue->thread)) { + return 0; + } + + pthread_mutex_lock(&queue->mutex); + int enqueued = em_task_queue_enqueue(queue, t); + pthread_mutex_unlock(&queue->mutex); + if (!enqueued) { + emscripten_thread_mailbox_unref(queue->thread); + return 0; + } + + // We're done if there is already a pending notification for this task queue. + // Otherwise, we will send one. notification_state previous = atomic_exchange(&queue->notification, NOTIFICATION_PENDING); - if (previous != NOTIFICATION_PENDING) { - _emscripten_notify_task_queue(queue->thread, - pthread_self(), - emscripten_main_runtime_thread_id(), - queue); + if (previous == NOTIFICATION_PENDING) { + emscripten_thread_mailbox_unref(queue->thread); + return 1; } + + emscripten_thread_mailbox_send( + queue->thread, (task){.func = receive_notification, .arg = queue}); + emscripten_thread_mailbox_unref(queue->thread); + return 1; } diff --git a/system/lib/pthread/em_task_queue.h b/system/lib/pthread/em_task_queue.h index bb1b6d5fc605e..55f513b90332a 100644 --- a/system/lib/pthread/em_task_queue.h +++ b/system/lib/pthread/em_task_queue.h @@ -50,7 +50,7 @@ em_task_queue* em_task_queue_create(pthread_t thread); void em_task_queue_destroy(em_task_queue* queue); -// Execute tasks until an empty queue is observed. +// Execute tasks until an empty queue is observed. Internally locks the queue. void em_task_queue_execute(em_task_queue* queue); // Not thread safe. @@ -69,6 +69,7 @@ int em_task_queue_enqueue(em_task_queue* queue, task t); // Not thread safe. Assumes the queue is not empty. task em_task_queue_dequeue(em_task_queue* queue); -// Schedule the queue to be executed next time its owning thread returns to its -// event loop. -void em_task_queue_notify(em_task_queue* queue); +// Atomically enqueue the task and schedule the queue to be executed next time +// its owning thread returns to its event loop. Returns 1 on success and 0 +// otherwise. Internally locks the queue. +int em_task_queue_send(em_task_queue* queue, task t); diff --git a/system/lib/pthread/library_pthread.c b/system/lib/pthread/library_pthread.c index 8338096c050fa..7f8ed21d0b82a 100644 --- a/system/lib/pthread/library_pthread.c +++ b/system/lib/pthread/library_pthread.c @@ -595,4 +595,6 @@ void __emscripten_init_main_thread(void) { // this is used by pthread_key_delete for deleting thread-specific data. __main_pthread.next = __main_pthread.prev = &__main_pthread; __main_pthread.tsd = (void **)__pthread_tsd_main; + + _emscripten_thread_mailbox_init(&__main_pthread); } diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index b1e9e234429bd..90001632362fa 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -9,11 +9,12 @@ #include #include #include +#include #include #include #include "em_task_queue.h" -#include "proxying_notification_state.h" +#include "thread_mailbox.h" struct em_proxying_queue { // Protects all accesses to em_task_queues, size, and capacity. @@ -103,17 +104,6 @@ static em_task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q, return tasks; } -// Exported for use in worker.js, but otherwise an internal function. -EMSCRIPTEN_KEEPALIVE -void _emscripten_proxy_execute_task_queue(em_task_queue* tasks) { - // Before we attempt to execute a request from another thread make sure we - // are in sync with all the loaded code. - // For example, in PROXY_TO_PTHREAD the atexit functions are called via - // a proxied call, and without this call to syncronize we would crash if - // any atexit functions were registered from a side module. - em_task_queue_execute(tasks); -} - void emscripten_proxy_execute_queue(em_proxying_queue* q) { assert(q != NULL); assert(pthread_self()); @@ -156,15 +146,8 @@ int emscripten_proxy_async(em_proxying_queue* q, if (tasks == NULL) { return 0; } - pthread_mutex_lock(&tasks->mutex); - int enqueued = em_task_queue_enqueue(tasks, (task){func, arg}); - pthread_mutex_unlock(&tasks->mutex); - if (!enqueued) { - return 0; - } - em_task_queue_notify(tasks); - return 1; + return em_task_queue_send(tasks, (task){func, arg}); } struct em_proxying_ctx { diff --git a/system/lib/pthread/pthread_create.c b/system/lib/pthread/pthread_create.c index b61158052584b..c3fdd046a3764 100644 --- a/system/lib/pthread/pthread_create.c +++ b/system/lib/pthread/pthread_create.c @@ -223,6 +223,8 @@ int __pthread_create(pthread_t* restrict res, _emscripten_thread_profiler_init(new); #endif + _emscripten_thread_mailbox_init(new); + struct pthread *self = __pthread_self(); dbg("start __pthread_create: new=%p new_end=%p stack=%p->%p " "stack_size=%zu tls_base=%p", @@ -303,6 +305,8 @@ void _emscripten_thread_exit(void* result) { self->cancelasync = PTHREAD_CANCEL_DEFERRED; self->result = result; + _emscripten_thread_mailbox_shutdown(self); + // Run any handlers registered with pthread_cleanup_push __run_cleanup_handlers(); diff --git a/system/lib/pthread/thread_mailbox.c b/system/lib/pthread/thread_mailbox.c new file mode 100644 index 0000000000000..16b8abeded335 --- /dev/null +++ b/system/lib/pthread/thread_mailbox.c @@ -0,0 +1,120 @@ +/* + * Copyright 2023 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. + */ + +#include +#include +#include + +#include "em_task_queue.h" +#include "pthread_impl.h" +#include "thread_mailbox.h" +#include "threading_internal.h" + +int emscripten_thread_mailbox_ref(pthread_t thread) { + // Attempt to increment the refcount, being careful not to increment it if we + // ever observe a 0. + int prev_count = thread->mailbox_refcount; + while (1) { + if (prev_count == 0) { + // The mailbox is already closed! + return 0; + } + int desired_count = prev_count + 1; + if (atomic_compare_exchange_weak( + &thread->mailbox_refcount, &prev_count, desired_count)) { + return 1; + } + } +} + +// Decrement and return the refcount. +void emscripten_thread_mailbox_unref(pthread_t thread) { + int new_count = atomic_fetch_sub(&thread->mailbox_refcount, 1) - 1; + assert(new_count >= 0); + if (new_count == 0) { + // The count is now zero. The thread that owns this queue may be waiting to + // shut down. Notify the thread that it is safe to proceed now that the + // mailbox is closed. + __builtin_wasm_memory_atomic_notify((int*)&thread->mailbox_refcount, -1); + } +} + +// Defined in emscripten_thread_state.S. +int _emscripten_thread_supports_atomics_wait(void); + +void _emscripten_thread_mailbox_shutdown(pthread_t thread) { + assert(thread == pthread_self()); + + // Decrement the refcount and wait for it to reach zero. + assert(thread->mailbox_refcount > 0); + int count = atomic_fetch_sub(&thread->mailbox_refcount, 1) - 1; + + while (count != 0) { + // Wait if possible and otherwise spin. + if (_emscripten_thread_supports_atomics_wait() && + __builtin_wasm_memory_atomic_wait32( + (int*)&thread->mailbox_refcount, count, -1) == 0) { + break; + } + count = thread->mailbox_refcount; + } + // TODO: Cancel tasks. + + // The mailbox will not be accessed again after this point. + em_task_queue_destroy(thread->mailbox); +} + +void _emscripten_thread_mailbox_init(pthread_t thread) { + thread->mailbox = em_task_queue_create(thread); + thread->mailbox_refcount = 1; +} + +// Exported for use in worker.js, but otherwise an internal function. +EMSCRIPTEN_KEEPALIVE +void _emscripten_check_mailbox() { + // Before we attempt to execute a request from another thread make sure we + // are in sync with all the loaded code. + // For example, in PROXY_TO_PTHREAD the atexit functions are called via + // a proxied call, and without this call to syncronize we would crash if + // any atexit functions were registered from a side module. + assert(pthread_self()); + em_task_queue* mailbox = pthread_self()->mailbox; + mailbox->notification = NOTIFICATION_RECEIVED; + em_task_queue_execute(pthread_self()->mailbox); + notification_state expected = NOTIFICATION_RECEIVED; + atomic_compare_exchange_strong( + &mailbox->notification, &expected, NOTIFICATION_NONE); +} + +// Send a postMessage notification telling the target thread to check its +// mailbox when it returns to its event loop. Pass in the current thread and +// main thread ids to minimize calls back into Wasm. +void _emscripten_notify_mailbox(pthread_t target_thread, + pthread_t curr_thread, + pthread_t main_thread); + +void emscripten_thread_mailbox_send(pthread_t thread, task t) { + assert(thread->mailbox_refcount > 0); + + pthread_mutex_lock(&thread->mailbox->mutex); + if (!em_task_queue_enqueue(thread->mailbox, t)) { + assert(0 && "No way to correctly recover from allocation failure"); + } + pthread_mutex_unlock(&thread->mailbox->mutex); + + // If there is no pending notification for this mailbox, create one. If an old + // notification is currently being processed, it may or may not execute the + // new work. In case it does not, the new notification will ensure the work is + // still executed. + notification_state previous = + atomic_exchange(&thread->mailbox->notification, NOTIFICATION_PENDING); + if (previous != NOTIFICATION_PENDING) { + _emscripten_notify_mailbox(thread, + pthread_self(), + emscripten_main_runtime_thread_id()); + } +} diff --git a/system/lib/pthread/thread_mailbox.h b/system/lib/pthread/thread_mailbox.h new file mode 100644 index 0000000000000..1c0e0c1893af9 --- /dev/null +++ b/system/lib/pthread/thread_mailbox.h @@ -0,0 +1,32 @@ +/* + * Copyright 2023 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. + */ + +#pragma once + +#include +#include + +// Try to increment the refcount of the mailbox, ensuring it will stay open +// until the refcount is decremented again. Returns 1 on success or 0 if the +// mailbox is already closed. +int emscripten_thread_mailbox_ref(pthread_t thread); + +// Decrement the mailbox's refcount. +void emscripten_thread_mailbox_unref(pthread_t thread); + +// Send a message to the given `thread`. This should only be called after +// incrementing the mailbox refcount to ensure it stays open. The receiving +// thread will receive the message the next time it returns to its event loop, +// or if the target thread shuts down before then, the message's shutdown +// handler will be called instead. +void emscripten_thread_mailbox_send(pthread_t thread, task t); + +// Initialize the mailbox on a pthread struct. Called during `pthread_create`. +void _emscripten_thread_mailbox_init(pthread_t thread); + +// Close the mailbox and cancel any pending messages. +void _emscripten_thread_mailbox_shutdown(pthread_t thread); diff --git a/test/other/metadce/test_metadce_minimal_pthreads.funcs b/test/other/metadce/test_metadce_minimal_pthreads.funcs index 63004939b6d50..a496ac83bcfc9 100644 --- a/test/other/metadce/test_metadce_minimal_pthreads.funcs +++ b/test/other/metadce/test_metadce_minimal_pthreads.funcs @@ -19,13 +19,14 @@ $__wasm_call_ctors $__wasm_init_memory $__wasm_init_tls $_do_call -$_emscripten_proxy_execute_task_queue +$_emscripten_check_mailbox $_emscripten_proxy_main $_emscripten_run_in_main_runtime_thread_js $_emscripten_thread_crashed $_emscripten_thread_exit $_emscripten_thread_free_data $_emscripten_thread_init +$_emscripten_thread_mailbox_init $_emscripten_tls_init $_emscripten_yield $_main_thread @@ -42,7 +43,10 @@ $dlfree $dlmalloc $do_dispatch_to_thread $em_queued_call_malloc +$em_task_queue_create +$em_task_queue_enqueue $em_task_queue_execute +$em_task_queue_free $em_task_queue_is_empty $emscripten_async_run_in_main_thread $emscripten_current_thread_process_queued_calls @@ -55,6 +59,7 @@ $init_mparams $main $memset $pthread_attr_destroy +$receive_notification $sbrk $stackAlloc $stackRestore diff --git a/test/other/metadce/test_metadce_minimal_pthreads.jssize b/test/other/metadce/test_metadce_minimal_pthreads.jssize index 542b832fde0c5..ffec19164095f 100644 --- a/test/other/metadce/test_metadce_minimal_pthreads.jssize +++ b/test/other/metadce/test_metadce_minimal_pthreads.jssize @@ -1 +1 @@ -15730 +15359 diff --git a/test/other/metadce/test_metadce_minimal_pthreads.size b/test/other/metadce/test_metadce_minimal_pthreads.size index 0555a7613f23e..2509094d1ebca 100644 --- a/test/other/metadce/test_metadce_minimal_pthreads.size +++ b/test/other/metadce/test_metadce_minimal_pthreads.size @@ -1 +1 @@ -17820 +18487 diff --git a/test/pthread/test_pthread_proxying_canceled_work.c b/test/pthread/test_pthread_proxying_canceled_work.c new file mode 100644 index 0000000000000..ba2a4ffe507b2 --- /dev/null +++ b/test/pthread/test_pthread_proxying_canceled_work.c @@ -0,0 +1,84 @@ +#include +#include +#include +#include + +em_proxying_queue* queue; + +void explode(void* arg) { assert(0 && "the work should not be run!"); } + +void set_flag(void* flag) { + // Schedule the flag to be set on the next turn of the event loop so that we + // can be sure cleanup has finished first. We need to use EM_ASM and JS here + // because this code needs to run after the thread runtime has exited. + + // clang-format off + EM_ASM({setTimeout(() => Atomics.store(HEAP32, $0 >> 2, 1))}, flag); + // clang-format on +} + +// Used to call `set_flag` on thread exit or cancellation. +pthread_key_t dtor_key; + +void* cancel_self(void* canceled) { + pthread_setspecific(dtor_key, canceled); + pthread_cancel(pthread_self()); + pthread_testcancel(); + assert(0 && "thread should have been canceled!"); + return NULL; +} + +void* exit_self(void* exited) { + pthread_setspecific(dtor_key, exited); + pthread_exit(NULL); + assert(0 && "thread should have exited!"); + return NULL; +} + +void test_cancel_then_proxy() { + printf("testing cancel followed by proxy\n"); + + pthread_t thread; + _Atomic int canceled = 0; + pthread_create(&thread, NULL, cancel_self, &canceled); + + // Wait for the thread to be canceled. + while (!canceled) { + } + + // Proxying work to the thread should return an error. + int ret = emscripten_proxy_sync(queue, thread, explode, NULL); + assert(ret == 0); + + pthread_join(thread, NULL); +} + +void test_exit_then_proxy() { + printf("testing exit followed by proxy\n"); + + pthread_t thread; + _Atomic int exited = 0; + pthread_create(&thread, NULL, exit_self, &exited); + + // Wait for the thread to exit. + while (!exited) { + } + + // Proxying work to the thread should return an error. + int ret = emscripten_proxy_sync(queue, thread, explode, NULL); + assert(ret == 0); + + pthread_join(thread, NULL); +} + +int main() { + queue = em_proxying_queue_create(); + pthread_key_create(&dtor_key, set_flag); + + test_cancel_then_proxy(); + test_exit_then_proxy(); + + em_proxying_queue_destroy(queue); + + printf("done\n"); +} diff --git a/test/pthread/test_pthread_proxying_canceled_work.out b/test/pthread/test_pthread_proxying_canceled_work.out new file mode 100644 index 0000000000000..a58137e245219 --- /dev/null +++ b/test/pthread/test_pthread_proxying_canceled_work.out @@ -0,0 +1,3 @@ +testing cancel followed by proxy +testing exit followed by proxy +done diff --git a/test/pthread/test_pthread_proxying_dropped_work.c b/test/pthread/test_pthread_proxying_dropped_work.c index 5c9dff9f5642d..ab248b5b04383 100644 --- a/test/pthread/test_pthread_proxying_dropped_work.c +++ b/test/pthread/test_pthread_proxying_dropped_work.c @@ -20,6 +20,7 @@ void* proxy_to_self(void* arg) { } void* do_nothing(void* arg) { + *((_Atomic int*)arg) = 1; return NULL; } @@ -31,7 +32,10 @@ int main() { // Check that proxying to a thread that exits without a live runtime causes // the work to be dropped without other errors. pthread_t worker; - pthread_create(&worker, NULL, do_nothing, NULL); + _Atomic int running = 0; + pthread_create(&worker, NULL, do_nothing, &running); + while (!running) { + } emscripten_proxy_async(queue, worker, explode, NULL); // Check that a thread proxying to itself but exiting without a live runtime diff --git a/test/pthread/test_pthread_proxying_refcount.c b/test/pthread/test_pthread_proxying_refcount.c index 454628868158a..99d0ee9c07879 100644 --- a/test/pthread/test_pthread_proxying_refcount.c +++ b/test/pthread/test_pthread_proxying_refcount.c @@ -40,6 +40,8 @@ void register_processed(void) { void task(void* arg) { *(_Atomic int*)arg = 1; } void* execute_and_free_queue(void* arg) { + *((_Atomic int*)arg) = 1; + // Wait until we are signaled to execute the queue. while (!should_execute) { } @@ -56,7 +58,7 @@ void* execute_and_free_queue(void* arg) { var oldOnMessage = onmessage; onmessage = (e) => { oldOnMessage(e); - if (e.data.cmd == 'processProxyingQueue') { + if (e.data.cmd == 'checkMailbox') { _register_processed(); } }; @@ -80,7 +82,10 @@ int main() { // Create the worker and send it tasks. pthread_t worker; - pthread_create(&worker, NULL, execute_and_free_queue, NULL); + _Atomic int running = 0; + pthread_create(&worker, NULL, execute_and_free_queue, &running); + while (!running) { + } for (int i = 0; i < 2; i++) { emscripten_proxy_async(queues[i], worker, task, &executed[i]); } @@ -90,8 +95,8 @@ int main() { while (!executed[0] || !executed[1]) { } - // Wait for the notifications to be received. - while (processed < 2) { + // Wait for the postMessage notification to be received. + while (processed < 1) { } #ifndef SANITIZER diff --git a/test/reference_struct_info.json b/test/reference_struct_info.json index ba45652953bc6..c4e4f38c4599b 100644 --- a/test/reference_struct_info.json +++ b/test/reference_struct_info.json @@ -1351,7 +1351,7 @@ "p_proto": 8 }, "pthread": { - "__size__": 120, + "__size__": 128, "profilerBlock": 112, "stack": 52, "stack_size": 56 diff --git a/test/test_core.py b/test/test_core.py index f52fe170aa229..049fa0b0590cc 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -2838,6 +2838,14 @@ def test_pthread_proxying_dropped_work(self): self.set_setting('PTHREAD_POOL_SIZE=2') self.do_run_in_out_file_test('pthread/test_pthread_proxying_dropped_work.c') + @node_pthreads + def test_pthread_proxying_canceled_work(self): + self.set_setting('EXIT_RUNTIME') + self.set_setting('PROXY_TO_PTHREAD') + self.do_run_in_out_file_test( + 'pthread/test_pthread_proxying_canceled_work.c', + interleaved_output=False) + @node_pthreads def test_pthread_proxying_refcount(self): self.set_setting('EXIT_RUNTIME') diff --git a/tools/system_libs.py b/tools/system_libs.py index 2bcfeb0e154c5..41bb2c402e0c8 100644 --- a/tools/system_libs.py +++ b/tools/system_libs.py @@ -999,6 +999,7 @@ def get_files(self): 'library_pthread.c', 'em_task_queue.c', 'proxying.c', + 'thread_mailbox.c', 'pthread_create.c', 'pthread_kill.c', 'emscripten_thread_init.c', From c0eee506240c820c6d4df60e7168d428641da5a4 Mon Sep 17 00:00:00 2001 From: Thomas Lively Date: Mon, 27 Feb 2023 07:34:54 -0800 Subject: [PATCH 2/4] Remove mailbox execution before thread start --- emcc.py | 1 - src/worker.js | 6 ------ test/pthread/test_pthread_proxying_dropped_work.c | 6 +----- test/pthread/test_pthread_proxying_refcount.c | 7 +------ 4 files changed, 2 insertions(+), 18 deletions(-) diff --git a/emcc.py b/emcc.py index f5e040384264d..3e54ad487dfcb 100755 --- a/emcc.py +++ b/emcc.py @@ -1712,7 +1712,6 @@ def setup_pthreads(target): '__emscripten_thread_crashed', '__emscripten_tls_init', '_pthread_self', - '__emscripten_check_mailbox', 'checkMailbox', ] settings.EXPORTED_FUNCTIONS += worker_imports diff --git a/src/worker.js b/src/worker.js index 38883ac60de21..77ddee2b749c5 100644 --- a/src/worker.js +++ b/src/worker.js @@ -233,12 +233,6 @@ function handleMessage(e) { // We only do this once per worker since they get reused Module['__embind_initialize_bindings'](); #endif // EMBIND - - // Execute any proxied work that came in before the thread was - // initialized. Only do this once because it is only possible for - // notifications to arrive before thread initialization on fresh - // workers. - Module['__emscripten_check_mailbox'](); initializedJS = true; } diff --git a/test/pthread/test_pthread_proxying_dropped_work.c b/test/pthread/test_pthread_proxying_dropped_work.c index ab248b5b04383..5c9dff9f5642d 100644 --- a/test/pthread/test_pthread_proxying_dropped_work.c +++ b/test/pthread/test_pthread_proxying_dropped_work.c @@ -20,7 +20,6 @@ void* proxy_to_self(void* arg) { } void* do_nothing(void* arg) { - *((_Atomic int*)arg) = 1; return NULL; } @@ -32,10 +31,7 @@ int main() { // Check that proxying to a thread that exits without a live runtime causes // the work to be dropped without other errors. pthread_t worker; - _Atomic int running = 0; - pthread_create(&worker, NULL, do_nothing, &running); - while (!running) { - } + pthread_create(&worker, NULL, do_nothing, NULL); emscripten_proxy_async(queue, worker, explode, NULL); // Check that a thread proxying to itself but exiting without a live runtime diff --git a/test/pthread/test_pthread_proxying_refcount.c b/test/pthread/test_pthread_proxying_refcount.c index 99d0ee9c07879..09d3f031df562 100644 --- a/test/pthread/test_pthread_proxying_refcount.c +++ b/test/pthread/test_pthread_proxying_refcount.c @@ -40,8 +40,6 @@ void register_processed(void) { void task(void* arg) { *(_Atomic int*)arg = 1; } void* execute_and_free_queue(void* arg) { - *((_Atomic int*)arg) = 1; - // Wait until we are signaled to execute the queue. while (!should_execute) { } @@ -82,10 +80,7 @@ int main() { // Create the worker and send it tasks. pthread_t worker; - _Atomic int running = 0; - pthread_create(&worker, NULL, execute_and_free_queue, &running); - while (!running) { - } + pthread_create(&worker, NULL, execute_and_free_queue, NULL); for (int i = 0; i < 2; i++) { emscripten_proxy_async(queues[i], worker, task, &executed[i]); } From b312d8a3d16ed1ef531fa1c8e7a354350c248473 Mon Sep 17 00:00:00 2001 From: Thomas Lively Date: Tue, 28 Feb 2023 15:17:27 -0800 Subject: [PATCH 3/4] futex instead of builtins --- system/lib/pthread/thread_mailbox.c | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/system/lib/pthread/thread_mailbox.c b/system/lib/pthread/thread_mailbox.c index 16b8abeded335..3b3c18ecb37f4 100644 --- a/system/lib/pthread/thread_mailbox.c +++ b/system/lib/pthread/thread_mailbox.c @@ -6,6 +6,7 @@ */ #include +#include #include #include @@ -39,7 +40,7 @@ void emscripten_thread_mailbox_unref(pthread_t thread) { // The count is now zero. The thread that owns this queue may be waiting to // shut down. Notify the thread that it is safe to proceed now that the // mailbox is closed. - __builtin_wasm_memory_atomic_notify((int*)&thread->mailbox_refcount, -1); + emscripten_futex_wake(&thread->mailbox_refcount, INT_MAX); } } @@ -54,12 +55,7 @@ void _emscripten_thread_mailbox_shutdown(pthread_t thread) { int count = atomic_fetch_sub(&thread->mailbox_refcount, 1) - 1; while (count != 0) { - // Wait if possible and otherwise spin. - if (_emscripten_thread_supports_atomics_wait() && - __builtin_wasm_memory_atomic_wait32( - (int*)&thread->mailbox_refcount, count, -1) == 0) { - break; - } + emscripten_futex_wait(&thread->mailbox_refcount, count, INFINITY); count = thread->mailbox_refcount; } // TODO: Cancel tasks. From bc9b5addbfe110ff3182a02c7a884b0d9c9ab22a Mon Sep 17 00:00:00 2001 From: Thomas Lively Date: Tue, 28 Feb 2023 16:52:45 -0800 Subject: [PATCH 4/4] skip asan dlopen test --- test/test_core.py | 1 + 1 file changed, 1 insertion(+) diff --git a/test/test_core.py b/test/test_core.py index 049fa0b0590cc..d5cf6643259c5 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -9428,6 +9428,7 @@ def test_pthread_dlopen(self): @needs_dylink @node_pthreads + @no_asan("Transient memory leaks to be solved by #18776") def test_pthread_dlopen_many(self): nthreads = 10 self.set_setting('USE_PTHREADS')