Add refcounting and a zombie list to em_proxying_queue (#16524)

When work is added to a proxying queue, it sends a message to the target
thread's JS event loop. Those messages contain references to the queues that
sent them, so to prevent use-after-free bugs, we need to defer freeing the
queues until after all their outstanding messages have been processed. Add a
reference count to each queue to track how many outstanding messages there are
and prevent the queues from being freed too early.

Furthermore, a thread may not return to its event loop to process messages until
after the thread has died. In that case, the thread cannot free the queue
because it has no stack to call `free` on. To safely support this use case, also
add a list of zombie queues, i.e. queues that have been destroyed but still have
nonzero reference counts. Make the queue constructor traverse the zombie list
and free any zombie queues whose reference counts have gone to zero.

For more details, see the new documentation comment in proxying.c.

Author: tlively

emcc.py

@@ -1519,7 +1519,6 @@ def setup_pthreads(target):
     '__emscripten_thread_exit',
     '__emscripten_thread_crashed',
     '_emscripten_tls_init',
-    '_emscripten_current_thread_process_queued_calls',
     '_pthread_self',
   ]
   settings.EXPORTED_FUNCTIONS += worker_imports

src/library_pthread.js

@@ -270,6 +270,8 @@ var LibraryPThread = {
         if (cmd === 'processProxyingQueue') {
           // TODO: Must post message to main Emscripten thread in PROXY_TO_WORKER mode.
           _emscripten_proxy_execute_queue(d['queue']);
+          // Decrement the ref count
+          Atomics.sub(HEAP32, d['queue'] >> 2, 1);
         } else if (cmd === 'spawnThread') {
           spawnThread(d);
         } else if (cmd === 'cleanupThread') {
@@ -1054,9 +1056,13 @@ var LibraryPThread = {
       setTimeout(() => {
         // Only execute the queue if we have a live pthread runtime. We
         // implement pthread_self to return 0 if there is no live runtime.
+        // TODO: Use `callUserCallback` to correctly handle unwinds, etc. once
+        //       `runtimeExited` is correctly unset on workers.
         if (_pthread_self()) {
           _emscripten_proxy_execute_queue(queue);
         }
+        // Decrement the ref count
+        Atomics.sub(HEAP32, queue >> 2, 1);
       });
     } else if (ENVIRONMENT_IS_PTHREAD) {
       postMessage({'targetThread' : targetThreadId, 'cmd' : 'processProxyingQueue', 'queue' : queue});

src/worker.js

@@ -289,6 +289,8 @@ self.onmessage = (e) => {
       if (Module['_pthread_self']()) { // If this thread is actually running?
         Module['_emscripten_proxy_execute_queue'](e.data.queue);
       }
+      // Decrement the ref count
+      Atomics.sub(HEAP32, e.data.queue >> 2, 1);
     } else {
       err('worker.js received unknown command ' + e.data.cmd);
       err(e.data);

system/lib/pthread/proxying.c

@@ -15,6 +15,49 @@
 
 #define TASK_QUEUE_INITIAL_CAPACITY 128
 
+// Proxy Queue Lifetime Management
+// -------------------------------
+//
+// Proxied tasks are executed either when the user manually calls
+// `emscripten_proxy_execute_queue` on the target thread or when the target
+// thread returns to the event loop. The queue does not know which execution
+// path will be used ahead of time when the work is proxied, so it must
+// conservatively send a message to the target thread's event loop in case the
+// user expects the event loop to drive the execution. These notifications
+// contain references to the queue that will be dereferenced when the target
+// thread returns to its event loop and receives the notification, even if the
+// user manages the execution of the queue themselves.
+//
+// To avoid use-after-free bugs, we cannot free a queue immediately when a user
+// calls `em_proxying_queue_destroy`; instead, we have to defer freeing the
+// queue until all of its outstanding notifications have been processed. We
+// defer freeing the queue using a reference counting scheme. Each time a
+// notification containing a reference to the queue is generated, we increase
+// the reference count and each time one of the notifications is received and
+// processed, we decrease the reference count. The queue can only be freed once
+// `em_proxying_queue_destroy` has been called and the reference count has
+// reached zero.
+//
+// But an extra complication is that the target thread may have died by the time
+// it gets back to its event loop to process its notifications. This can happen
+// when a user proxies some work to a thread, then calls
+// `emscripten_proxy_execute_queue` on that thread, then destroys the queue and
+// exits the thread. In that situation no work will be dropped, but the thread's
+// worker will still receive a notification and have to decrease the reference
+// count without a live runtime. Without a live runtime, there is no stack, so
+// the worker cannot safely free the queue at this point even if the refcount
+// goes to zero. We need a separate thread with a live runtime to perform the
+// free.
+//
+// To ensure that queues are eventually freed, we place destroyed queues in a
+// global "zombie list" where they wait for their refcounts to fall to zero. The
+// zombie list is scanned whenever a new queue is constructed and any of the
+// zombie queues with zero refcounts are freed. In principle the zombie list
+// could be scanned at any time, but the queue constructor is a nice place to do
+// it because scanning there is sufficient to keep the number of zombie queues
+// from growing without bound; creating a new zombie ultimately requires
+// creating a new queue.
+
 extern int _emscripten_notify_proxying_queue(pthread_t target_thread,
                                              pthread_t curr_thread,
                                              pthread_t main_thread,
@@ -120,15 +163,25 @@ static task task_queue_dequeue(task_queue* tasks) {
 }
 
 struct em_proxying_queue {
-  // Protects all accesses to all task_queues.
+  // The number of references to this queue that exist in JS event queues.
+  // Decremented directly from JS, so this must be the first field.
+  _Atomic int js_refcount;
+  // Doubly linked list pointers for the zombie list.
+  em_proxying_queue* zombie_prev;
+  em_proxying_queue* zombie_next;
+  // Protects all accesses to task_queues, size, and capacity.
   pthread_mutex_t mutex;
   // `size` task queues stored in an array of size `capacity`.
   task_queue* task_queues;
   int size;
   int capacity;
 };
 
-static em_proxying_queue system_proxying_queue = {.mutex =
+// The system proxying queue.
+static em_proxying_queue system_proxying_queue = {.js_refcount = 0,
+                                                  .zombie_prev = NULL,
+                                                  .zombie_next = NULL,
+                                                  .mutex =
                                                     PTHREAD_MUTEX_INITIALIZER,
                                                   .task_queues = NULL,
                                                   .size = 0,
@@ -138,12 +191,51 @@ em_proxying_queue* emscripten_proxy_get_system_queue(void) {
   return &system_proxying_queue;
 }
 
+// The head of the zombie list. Its mutex protects access to the list and its
+// other fields are not used..
+static em_proxying_queue zombie_list_head = {.zombie_prev = &zombie_list_head,
+                                             .zombie_next = &zombie_list_head,
+                                             .mutex =
+                                               PTHREAD_MUTEX_INITIALIZER};
+
+static void em_proxying_queue_free(em_proxying_queue* q) {
+  pthread_mutex_destroy(&q->mutex);
+  for (int i = 0; i < q->size; i++) {
+    task_queue_deinit(&q->task_queues[i]);
+  }
+  free(q->task_queues);
+  free(q);
+}
+
+static void cull_zombies() {
+  pthread_mutex_lock(&zombie_list_head.mutex);
+  em_proxying_queue* curr = zombie_list_head.zombie_next;
+  while (curr != &zombie_list_head) {
+    em_proxying_queue* next = curr->zombie_next;
+    if (curr->js_refcount == 0) {
+      // Remove the zombie from the list and free it.
+      curr->zombie_prev->zombie_next = curr->zombie_next;
+      curr->zombie_next->zombie_prev = curr->zombie_prev;
+      em_proxying_queue_free(curr);
+    }
+    curr = next;
+  }
+  pthread_mutex_unlock(&zombie_list_head.mutex);
+}
+
 em_proxying_queue* em_proxying_queue_create(void) {
+  // Free any queue that has been destroyed and is safe to free.
+  cull_zombies();
+
+  // Allocate the new queue.
   em_proxying_queue* q = malloc(sizeof(em_proxying_queue));
   if (q == NULL) {
     return NULL;
   }
-  *q = (em_proxying_queue){.mutex = PTHREAD_MUTEX_INITIALIZER,
+  *q = (em_proxying_queue){.js_refcount = 0,
+                           .zombie_prev = NULL,
+                           .zombie_next = NULL,
+                           .mutex = PTHREAD_MUTEX_INITIALIZER,
                            .task_queues = NULL,
                            .size = 0,
                            .capacity = 0};
@@ -153,14 +245,21 @@ em_proxying_queue* em_proxying_queue_create(void) {
 void em_proxying_queue_destroy(em_proxying_queue* q) {
   assert(q != NULL);
   assert(q != &system_proxying_queue && "cannot destroy system proxying queue");
-  // No need to acquire the lock; no one should be racing with the destruction
-  // of the queue.
-  pthread_mutex_destroy(&q->mutex);
-  for (int i = 0; i < q->size; i++) {
-    task_queue_deinit(&q->task_queues[i]);
+  assert(!q->zombie_next && !q->zombie_prev &&
+         "double freeing em_proxying_queue!");
+  if (q->js_refcount == 0) {
+    // No outstanding references to the queue, so we can go ahead and free it.
+    em_proxying_queue_free(q);
+    return;
   }
-  free(q->task_queues);
-  free(q);
+  // Otherwise add the queue to the zombie list so that it will eventually be
+  // freed safely.
+  pthread_mutex_lock(&zombie_list_head.mutex);
+  q->zombie_next = zombie_list_head.zombie_next;
+  q->zombie_prev = &zombie_list_head;
+  q->zombie_next->zombie_prev = q;
+  q->zombie_prev->zombie_next = q;
+  pthread_mutex_unlock(&zombie_list_head.mutex);
 }
 
 // Not thread safe. Returns -1 if there are no tasks for the thread.
@@ -268,6 +367,7 @@ int emscripten_proxy_async(em_proxying_queue* q,
   // Otherwise, the target thread was already notified when the existing work
   // was enqueued so we don't need to notify it again.
   if (empty) {
+    q->js_refcount++;
     _emscripten_notify_proxying_queue(
       target_thread, pthread_self(), emscripten_main_browser_thread_id(), q);
   }

tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.exports

@@ -3,7 +3,6 @@ B
 C
 D
 E
-F
 o
 p
 q

tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.funcs

@@ -10,7 +10,7 @@ $__pthread_setcancelstate
 $__set_thread_state
 $__stdio_write
 $__timedwait
-$__wake
+$__wake.2
 $__wasi_syscall_ret
 $__wasm_call_ctors
 $__wasm_init_memory
@@ -24,7 +24,7 @@ $_main_thread
 $a_cas
 $a_cas_p.1
 $a_dec
-$a_fetch_add
+$a_fetch_add.1
 $a_inc
 $a_swap
 $add
@@ -35,7 +35,6 @@ $dlmemalign
 $do_dispatch_to_thread
 $em_queued_call_malloc
 $emscripten_async_run_in_main_thread
-$emscripten_current_thread_process_queued_calls
 $emscripten_futex_wait
 $emscripten_futex_wake
 $emscripten_main_thread_process_queued_calls

tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.jssize

@@ -1 +1 @@
-31848
+31638

tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.size

@@ -1 +1 @@
-18582
+18584

tests/pthread/test_pthread_proxying_refcount.c

@@ -0,0 +1,112 @@
+#include <assert.h>
+#include <emscripten/console.h>
+#include <emscripten/emscripten.h>
+#include <pthread.h>
+#include <stdbool.h>
+#include <unistd.h>
+
+#include "proxying.h"
+
+// The first two queues will be zombies and the next two will be created just to
+// cull the zombies.
+em_proxying_queue* queues[4];
+
+#ifndef SANITIZER
+
+// If we are not using sanitizers (which need to use their own allocators),
+// override free so we can track when queues are actually freed.
+
+int queues_freed[4] = {};
+
+extern void emscripten_builtin_free(void* mem);
+void __attribute__((noinline)) free(void* ptr) {
+  for (int i = 0; i < 4; i++) {
+    if (ptr && queues[i] == ptr) {
+      queues_freed[i] = 1;
+      queues[i] = NULL;
+      break;
+    }
+  }
+  emscripten_builtin_free(ptr);
+}
+
+#endif // SANITIZER
+
+_Atomic int should_execute = 0;
+_Atomic int executed[2] = {};
+
+void task(void* arg) { *(_Atomic int*)arg = 1; }
+
+void* execute_and_free_queue(void* arg) {
+  // Wait until we are signaled to execute the queue.
+  while (!should_execute) {
+  }
+
+  // Execute the proxied work then free the empty queues.
+  for (int i = 0; i < 2; i++) {
+    emscripten_proxy_execute_queue(queues[i]);
+    em_proxying_queue_destroy(queues[i]);
+  }
+
+  // Exit with a live runtime so the queued work notification is received and we
+  // try to execute the queue again, even though it has been destroyed.
+  emscripten_exit_with_live_runtime();
+}
+
+int main() {
+  emscripten_console_log("start");
+  for (int i = 0; i < 2; i++) {
+    queues[i] = em_proxying_queue_create();
+    assert(queues[i]);
+  }
+
+  // Create the worker and send it tasks.
+  pthread_t worker;
+  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]);
+  }
+  should_execute = 1;
+
+  // Wait for the tasks to be executed.
+  while (!executed[0] || !executed[1]) {
+  }
+
+  // Break the queue abstraction to wait for the refcounts to be decreased.
+  while (*(_Atomic int*)queues[0] || *(_Atomic int*)queues[1]) {
+  }
+
+#ifndef SANITIZER
+  // Our zombies should not have been freed yet.
+  assert(!queues_freed[0]);
+  assert(!queues_freed[1]);
+#endif // SANITIZER
+
+  // Cull the zombies!
+  queues[2] = em_proxying_queue_create();
+
+#ifndef SANITIZER
+  // Now they should be free.
+  assert(queues_freed[0]);
+  assert(queues_freed[1]);
+  assert(!queues_freed[2]);
+#endif // SANITIZER
+
+  em_proxying_queue_destroy(queues[2]);
+
+#ifndef SANITIZER
+  // The new queue should have been immediately freed.
+  assert(queues_freed[2]);
+#endif // SANITIZER
+
+  // Cull again, but this time there should be nothing to cull.
+  queues[3] = em_proxying_queue_create();
+  em_proxying_queue_destroy(queues[3]);
+
+#ifndef SANITIZER
+  // The new queue should have been immediately freed.
+  assert(queues_freed[3]);
+#endif // SANITIZER
+
+  emscripten_console_log("done");
+}

tests/pthread/test_pthread_proxying_refcount.out

@@ -0,0 +1,2 @@
+start
+done