diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index 1342cb2cc924f..d8570a2f79fdb 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -35,8 +35,6 @@ static em_proxying_queue system_proxying_queue = { .capacity = 0, }; -static _Thread_local bool system_queue_in_use = false; - em_proxying_queue* emscripten_proxy_get_system_queue(void) { return &system_proxying_queue; } @@ -80,56 +78,92 @@ static em_task_queue* get_tasks_for_thread(em_proxying_queue* q, return NULL; } -// Not thread safe. static em_task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q, pthread_t thread) { + pthread_mutex_lock(&q->mutex); em_task_queue* tasks = get_tasks_for_thread(q, thread); + pthread_mutex_unlock(&q->mutex); if (tasks != NULL) { return tasks; } - // There were no tasks for the thread; initialize a new em_task_queue. If - // there are not enough queues, allocate more. - if (q->size == q->capacity) { + + // There were no tasks for the thread; initialize a new em_task_queue outside + // the lock so allocations cannot deadlock with queue execution. + em_task_queue* new_tasks = em_task_queue_create(thread); + if (new_tasks == NULL) { + return NULL; + } + + // Holds the newly allocated task_queues array before it is installed, or the + // old task_queues array after they are swapped so it can be freed outside + // the lock. + em_task_queue** queues_to_free = NULL; + pthread_mutex_lock(&q->mutex); + while (true) { + tasks = get_tasks_for_thread(q, thread); + if (tasks != NULL) { + pthread_mutex_unlock(&q->mutex); + goto out; + } + if (q->size < q->capacity) { + // We have free capacity to add a new queue. + break; + } + // There is not capacity for more queues; reallocate outside the lock. int new_capacity = q->capacity == 0 ? 1 : q->capacity * 2; + pthread_mutex_unlock(&q->mutex); em_task_queue** new_task_queues = - realloc(q->task_queues, sizeof(em_task_queue*) * new_capacity); + realloc(queues_to_free, sizeof(em_task_queue*) * new_capacity); if (new_task_queues == NULL) { - return NULL; + goto out; + } + queues_to_free = new_task_queues; + pthread_mutex_lock(&q->mutex); + if (q->capacity < new_capacity) { + // Copy existing queues to the new array. + if (q->task_queues != NULL) { + memcpy( + queues_to_free, q->task_queues, sizeof(em_task_queue*) * q->size); + } + // Install the new array and arrange for the old to be freed. + em_task_queue** old_task_queues = q->task_queues; + q->task_queues = queues_to_free; + q->capacity = new_capacity; + queues_to_free = old_task_queues; } - q->task_queues = new_task_queues; - q->capacity = new_capacity; } - // Initialize the next available task queue. - tasks = em_task_queue_create(thread); - if (tasks == NULL) { - return NULL; + // Add the new task queue. + q->task_queues[q->size++] = new_tasks; + pthread_mutex_unlock(&q->mutex); + tasks = new_tasks; + new_tasks = NULL; + +out: + if (queues_to_free != NULL) { + free(queues_to_free); + } + if (new_tasks != NULL) { + em_task_queue_destroy(new_tasks); } - q->task_queues[q->size++] = tasks; return tasks; } +static _Thread_local bool executing_system_queue = false; + void emscripten_proxy_execute_queue(em_proxying_queue* q) { assert(q != NULL); assert(pthread_self()); - // Below is a recursion and deadlock guard: The recursion guard is to avoid - // infinite recursion when we arrive here from the pthread_lock call below - // that executes the system queue. The per-task_queue recursion lock can't - // catch these recursions because it can only be checked after the lock has - // been acquired. - // - // This also guards against deadlocks when adding to the system queue. When - // the current thread is adding tasks, it locks the queue, but we can - // potentially try to execute the queue during the add (from emscripten_yield - // when malloc takes a lock). This will deadlock the thread, so only try to - // take the lock if the current thread is not using the queue. We then hope - // the queue is executed later when it is unlocked. + // Recursion guard to avoid infinite recursion when we arrive here from the + // pthread_lock call below that executes the system queue. The per-task_queue + // recursion lock can't catch these recursions because it can only be checked + // after the lock has been acquired. bool is_system_queue = q == &system_proxying_queue; if (is_system_queue) { - if (system_queue_in_use) { + if (executing_system_queue) { return; } - system_queue_in_use = true; + executing_system_queue = true; } pthread_mutex_lock(&q->mutex); @@ -142,22 +176,13 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) { } if (is_system_queue) { - system_queue_in_use = false; + executing_system_queue = false; } } static bool do_proxy(em_proxying_queue* q, pthread_t target_thread, task t) { assert(q != NULL); - pthread_mutex_lock(&q->mutex); - bool is_system_queue = q == &system_proxying_queue; - if (is_system_queue) { - system_queue_in_use = true; - } em_task_queue* tasks = get_or_add_tasks_for_thread(q, target_thread); - if (is_system_queue) { - system_queue_in_use = false; - } - pthread_mutex_unlock(&q->mutex); if (tasks == NULL) { return false; } @@ -168,6 +193,7 @@ static bool do_proxy(em_proxying_queue* q, pthread_t target_thread, task t) { // special case in that we need to wake the target thread in case it is in // `emscripten_futex_wait`. Additionally, the _dlopen_proxying_queue also // requires a wakeup of the target thread after enqueuing work. + bool is_system_queue = q == &system_proxying_queue; bool needs_notify = #ifdef EMSCRIPTEN_DYNAMIC_LINKING q == _dlopen_proxying_queue || diff --git a/test/pthread/test_pthread_proxy_deadlock.c b/test/pthread/test_pthread_proxy_deadlock.c index 8e28ad8af6896..ee574af6beb88 100644 --- a/test/pthread/test_pthread_proxy_deadlock.c +++ b/test/pthread/test_pthread_proxy_deadlock.c @@ -3,7 +3,11 @@ // University of Illinois/NCSA Open Source License. Both these licenses can be // found in the LICENSE file. +#include +#include #include +#include +#include #include #include #include @@ -11,30 +15,152 @@ #include #include -// In the actual implementation of malloc the system queue may be executed -// non-deterministically if malloc is waiting on a mutex. This wraps malloc and -// executes the system queue during every allocation to make the behavior -// deterministic. -void *malloc(size_t size) { - if (emscripten_is_main_runtime_thread()) { +em_proxying_queue* user_queue = NULL; +_Thread_local bool intercept_malloc = false; +_Thread_local bool wait_in_malloc = false; +_Thread_local bool wait_for_worker_in_malloc = false; +_Atomic int in_malloc = 0; +_Atomic int worker_proxied = 0; +_Atomic int task_done = 0; + +// In the actual implementation of malloc, the system proxying queue may be +// executed non-deterministically if malloc contends on its mutex and calls +// emscripten_futex_wait -> _emscripten_yield. This wraps malloc and realloc to +// make that behavior deterministic. +void maybe_execute_queues(void) { + if (wait_in_malloc) { + wait_in_malloc = false; + int dummy = 0; + emscripten_futex_wait(&dummy, 0, INFINITY); + } else if (wait_for_worker_in_malloc) { + wait_for_worker_in_malloc = false; + in_malloc = 1; + while (!task_done) { + emscripten_futex_wait(&task_done, 0, INFINITY); + } + } else if (intercept_malloc && emscripten_is_main_runtime_thread()) { emscripten_proxy_execute_queue(emscripten_proxy_get_system_queue()); } - void *ptr = emscripten_builtin_malloc(size); - return ptr; +} + +void* malloc(size_t size) { + maybe_execute_queues(); + return emscripten_builtin_malloc(size); +} + +void* realloc(void* ptr, size_t size) { + maybe_execute_queues(); + return emscripten_builtin_realloc(ptr, size); } void task(void* arg) { emscripten_out("task\n"); + task_done = 1; + emscripten_futex_wake(&task_done, 1); +} + +void nop(void* arg) {} + +void execute_user_queue(void* arg) { + emscripten_proxy_execute_queue(user_queue); +} + +void* user_queue_worker_main(void* arg) { + while (!in_malloc) { + } + emscripten_proxy_execute_queue(user_queue); + task_done = 1; + emscripten_futex_wake(&task_done, 1); + return NULL; +} + +void* lost_wakeup_worker_main(void* arg) { + pthread_t main_thread = *(pthread_t*)arg; + // Enqueue a task for the main thread on the system queue. This also calls + // _emscripten_thread_notify(main_thread), setting NOTIFY_BIT on the main + // thread's wait_addr. + emscripten_proxy_async( + emscripten_proxy_get_system_queue(), main_thread, task, NULL); + worker_proxied = 1; + while (!task_done) { + emscripten_futex_wait(&task_done, 0, INFINITY); + } + return NULL; } int main() { - // Tests for a deadlock scenario that can occur when sending a task. - // The sequence of events is: +#if defined(TEST_SYSTEM_QUEUE) + // Tests for a deadlock scenario that can occur when sending a task on the + // system queue: // 1. Sending a task locks the queue. - // 2. Allocating a new task queue calls malloc. + // 2. Allocating a new task queue calls malloc/realloc. // 3. Malloc then attempts to execute and lock the already-locked queue, // causing a deadlock. - // This test ensures our implementation prevents this re-entrant lock. - emscripten_proxy_async(emscripten_proxy_get_system_queue(), pthread_self(), task, NULL); + intercept_malloc = true; + emscripten_proxy_async( + emscripten_proxy_get_system_queue(), pthread_self(), task, NULL); + intercept_malloc = false; +#elif defined(TEST_USER_QUEUE) + // Tests for a same-thread re-entrant lock on a user queue via a task on the + // system queue: + // 1. A task that executes `user_queue` is pending on the system queue. + // 2. Sending a task on `user_queue` locks `user_queue->mutex` and allocates. + // 3. Malloc executes the system queue, which runs `execute_user_queue` and + // attempts to lock `user_queue->mutex` on the same thread. + user_queue = em_proxying_queue_create(); + assert(user_queue != NULL); + emscripten_proxy_async(emscripten_proxy_get_system_queue(), + pthread_self(), + execute_user_queue, + NULL); + intercept_malloc = true; + emscripten_proxy_async(user_queue, pthread_self(), task, NULL); + intercept_malloc = false; + em_proxying_queue_destroy(user_queue); +#elif defined(TEST_USER_QUEUE_CONTENDED) + // Tests for a cross-thread deadlock when `user_queue->mutex` is held across a + // blocking allocation: + // 1. Main thread sends a task on `user_queue` and blocks inside malloc + // waiting for `worker`. + // 2. Worker calls `emscripten_proxy_execute_queue(user_queue)` before + // unblocking the main thread, which deadlocks if the main thread is + // holding `user_queue->mutex` across malloc. + user_queue = em_proxying_queue_create(); + assert(user_queue != NULL); + pthread_t worker; + int rc = pthread_create(&worker, NULL, user_queue_worker_main, NULL); + assert(rc == 0); + wait_for_worker_in_malloc = true; + emscripten_proxy_async(user_queue, worker, nop, NULL); + pthread_join(worker, NULL); + em_proxying_queue_destroy(user_queue); +#elif defined(TEST_LOST_WAKEUP) + // Tests that a contended allocation while adding a new task queue in do_proxy + // does not consume NOTIFY_BIT while skipping execution of the system queue, + // which would lose the wakeup and cause pthread_join to deadlock: + // 1. Worker proxies `task` to the main thread (setting NOTIFY_BIT on the main + // thread's wait_addr) and waits for `task` to finish before exiting. + // 2. Main thread proxies `nop` to the worker, allocating a new task queue. + // 3. That allocation simulates mutex contention by calling + // emscripten_futex_wait, which consumes NOTIFY_BIT and calls + // _emscripten_yield -> emscripten_proxy_execute_queue. + // 4. If emscripten_proxy_execute_queue is skipped because the system queue is + // being added to, `task` is not executed and NOTIFY_BIT is lost, so + // pthread_join blocks forever waiting for the worker. + pthread_t main_thread = pthread_self(); + pthread_t worker; + int rc = pthread_create(&worker, NULL, lost_wakeup_worker_main, &main_thread); + assert(rc == 0); + // Spin rather than using pthread_barrier_wait or emscripten_futex_wait so we + // do not call _emscripten_yield and execute the system queue (consuming + // NOTIFY_BIT) before calling emscripten_proxy_async below. + while (!worker_proxied) { + } + wait_in_malloc = true; + emscripten_proxy_async(emscripten_proxy_get_system_queue(), worker, nop, NULL); + pthread_join(worker, NULL); +#else +#error "Must define a test case" +#endif emscripten_out("done\n"); } diff --git a/test/test_core.py b/test/test_core.py index 94d51b5273b9e..779c47fbafa1e 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -2559,8 +2559,14 @@ def test_pthread_cancel_async(self): @no_asan('cannot replace malloc/free with ASan') @no_lsan('cannot replace malloc/free with LSan') @requires_pthreads - def test_pthread_proxy_deadlock(self): - self.do_runf('pthread/test_pthread_proxy_deadlock.c') + @parameterized({ + '': (['-DTEST_SYSTEM_QUEUE'],), + 'user_queue': (['-DTEST_USER_QUEUE'],), + 'user_queue_contended': (['-DTEST_USER_QUEUE_CONTENDED'],), + 'lost_wakeup': (['-DTEST_LOST_WAKEUP'],), + }) + def test_pthread_proxy_deadlock(self, args): + self.do_runf('pthread/test_pthread_proxy_deadlock.c', cflags=args) @no_asan('test relies on null pointer reads') def test_pthread_specific(self):