[Proxying] Do not allocate while holding mutex

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 ||

test/pthread/test_pthread_proxy_deadlock.c

@@ -3,38 +3,164 @@
 // University of Illinois/NCSA Open Source License.  Both these licenses can be
 // found in the LICENSE file.
 
+#include <assert.h>
+#include <math.h>
 #include <pthread.h>
+#include <stdatomic.h>
+#include <stdbool.h>
 #include <stdlib.h>
 #include <unistd.h>
 #include <emscripten/console.h>
 #include <emscripten/heap.h>
 #include <emscripten/proxying.h>
 #include <emscripten/threading.h>
 
-// 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");
 }

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):