Don't execute the proxy queue while adding to it from same thread. (#24565)

test_pthread_cancel was intermittently failing with a deadlock. From the
backtrace, I found we were adding to the queue which can sometimes call
malloc->emscripten_yield->emscripten_proxy_execute_queue which also
tries to lock the same queue on the same thread.

Author: brendandahl

system/lib/pthread/proxying.c

@@ -35,6 +35,8 @@ 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;
 }
@@ -106,22 +108,28 @@ static em_task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
   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());
 
-  // 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.
+  // 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.
   bool is_system_queue = q == &system_proxying_queue;
   if (is_system_queue) {
-    if (executing_system_queue) {
+    if (system_queue_in_use) {
       return;
     }
-    executing_system_queue = true;
+    system_queue_in_use = true;
   }
 
   pthread_mutex_lock(&q->mutex);
@@ -134,14 +142,21 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) {
   }
 
   if (is_system_queue) {
-    executing_system_queue = false;
+    system_queue_in_use = false;
   }
 }
 
 static int 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 0;

test/other/codesize/test_codesize_minimal_pthreads.funcs

@@ -72,6 +72,7 @@ $emscripten_futex_wake
 $emscripten_stack_get_current
 $emscripten_stack_set_limits
 $free_ctx
+$get_or_add_tasks_for_thread
 $get_tasks_for_thread
 $init_active_ctxs
 $init_file_lock

test/other/codesize/test_codesize_minimal_pthreads.size

@@ -1 +1 @@
-19540
+19590

test/other/codesize/test_codesize_minimal_pthreads_memgrowth.funcs

@@ -72,6 +72,7 @@ $emscripten_futex_wake
 $emscripten_stack_get_current
 $emscripten_stack_set_limits
 $free_ctx
+$get_or_add_tasks_for_thread
 $get_tasks_for_thread
 $init_active_ctxs
 $init_file_lock

test/other/codesize/test_codesize_minimal_pthreads_memgrowth.size

@@ -1 +1 @@
-19541
+19591

test/pthread/test_pthread_proxy_deadlock.c

@@ -0,0 +1,40 @@
+// Copyright 2025 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 <pthread.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()) {
+    emscripten_proxy_execute_queue(emscripten_proxy_get_system_queue());
+  }
+  void *ptr = emscripten_builtin_malloc(size);
+  return ptr;
+}
+
+void task(void* arg) {
+  emscripten_out("task\n");
+}
+
+int main() {
+  // Tests for a deadlock scenario that can occur when sending a task.
+  // The sequence of events is:
+  // 1. Sending a task locks the queue.
+  // 2. Allocating a new task queue calls malloc.
+  // 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);
+  emscripten_out("done\n");
+}

test/test_core.py

@@ -2564,6 +2564,12 @@ def test_pthread_cancel(self):
   def test_pthread_cancel_async(self):
     self.do_run_in_out_file_test('pthread/test_pthread_cancel_async.c')
 
+  @no_asan('cannot replace malloc/free with ASan')
+  @no_lsan('cannot replace malloc/free with LSan')
+  @node_pthreads
+  def test_pthread_proxy_deadlock(self):
+    self.do_runf('pthread/test_pthread_proxy_deadlock.c')
+
   @no_asan('test relies on null pointer reads')
   def test_pthread_specific(self):
     self.do_run_in_out_file_test('pthread/specific.c')