[Proxying] Cancel proxied work when the worker thread dies (#18741)

Update the synchronous proxying APIs to return 0 when attempting to proxy to a
thread that has already exited or has been canceled or exits or is canceled
before the proxied work is completed.

To implement cancellation, add an optional `cancel` function pointer to the
tasks queued in `em_task_queue`s, which are the internal thread-specific queues
used in the implementation of `em_proxying_queue`. This `cancel` function
pointer is not controllable directly by users of the API, but is rather used
internally to ensure that synchronous waiters are notified when the thread they
are waiting on dies. This mechanism will scale to other forms of cancellation
notification in the future. For example, it could be used to reject a promise
associated with the proxied work.

When a thread dies, it runs the cancellation handlers for each task in its
mailbox. Those tasks each carry a pointer to some task queue owned by a proxying
queue, so their cancellation handlers in turn run the cancellation handlers for
all the tasks in those pointed-to queues.

Task queue cancellation handlers do not handle the case of proxied work that has
been started but not finished when the worker thread dies because that work has
already been removed from the task queues. To cancel in-progress work as well,
update the implementation of synchronous proxying to place `em_proxying_ctx`
objects associated with active work in a thread-local doubly linked list that is
traversed by a registered thread specific data destructor. The destructor takes
care of canceling the active work.

Implements the design sketched out in #18631.

Author: tlively

ChangeLog.md

@@ -45,6 +45,9 @@ See docs/process.md for more on how version tagging works.
   SjLj, the combination we do not intend to support for the long term.
 - Added support for Wasm-based AudioWorklets for realtime audio processing
   (#16449)
+- Synchronous proxying functions in emscripten/proxying.h now return errors
+  instead of hanging forever when the worker thread dies before the proxied work
+  is finished.
 
 3.1.31 - 01/26/23
 -----------------

site/source/docs/api_reference/proxying.h.rst

@@ -91,7 +91,9 @@ Functions
 
   Enqueue ``func`` to be called with argument ``arg`` on the given queue and
   thread then wait for ``func`` to be executed synchronously before returning.
-  Returns 1 if the ``func`` was successfully completed and 0 otherwise.
+  Returns 1 if the ``func`` was successfully completed and 0 otherwise,
+  including if the target thread is canceled or exits before the work is
+  completed.
 
 .. c:function:: int emscripten_proxy_sync_with_ctx(em_proxying_queue* q, pthread_t target_thread, void (*func)(em_proxying_ctx*, void*), void* arg)
 

system/include/emscripten/proxying.h

@@ -71,7 +71,8 @@ int emscripten_proxy_async_with_callback(em_proxying_queue* q,
 
 // Enqueue `func` on the given queue and thread and wait for it to finish
 // executing before returning. Returns 1 if the task was successfully completed
-// and 0 otherwise.
+// and 0 otherwise, including if the target thread is canceled or exits before
+// the work is completed.
 int emscripten_proxy_sync(em_proxying_queue* q,
                           pthread_t target_thread,
                           void (*func)(void*),
@@ -82,7 +83,8 @@ int emscripten_proxy_sync(em_proxying_queue* q,
 // before returning. `func` need not call `emscripten_proxying_finish` itself;
 // it could instead store the context pointer and call
 // `emscripten_proxying_finish` at an arbitrary later time. Returns 1 if the
-// task was successfully completed and 0 otherwise.
+// task was successfully completed and 0 otherwise, including if the target
+// thread is canceled or exits before the work is completed.
 int emscripten_proxy_sync_with_ctx(em_proxying_queue* q,
                                    pthread_t target_thread,
                                    void (*func)(em_proxying_ctx*, void*),

system/lib/pthread/em_task_queue.c

@@ -180,6 +180,23 @@ void em_task_queue_execute(em_task_queue* queue) {
   queue->processing = 0;
 }
 
+void em_task_queue_cancel(em_task_queue* queue) {
+  pthread_mutex_lock(&queue->mutex);
+  while (!em_task_queue_is_empty(queue)) {
+    task t = em_task_queue_dequeue(queue);
+    if (t.cancel) {
+      t.cancel(t.arg);
+    }
+  }
+  pthread_mutex_unlock(&queue->mutex);
+  // Any subsequent messages to this queue (for example if a pthread struct is
+  // reused for a future thread, potentially on a different worker) will require
+  // a new notification. Clearing the flag is safe here because in both the
+  // proxying queue and mailbox cases, there are no more outstanding references
+  // to the queue after thread shutdown.
+  queue->notification = NOTIFICATION_NONE;
+}
+
 int em_task_queue_enqueue(em_task_queue* queue, task t) {
   if (em_task_queue_is_full(queue) && !em_task_queue_grow(queue)) {
     return 0;
@@ -204,6 +221,11 @@ static void receive_notification(void* arg) {
     &tasks->notification, &expected, NOTIFICATION_NONE);
 }
 
+static void cancel_notification(void* arg) {
+  em_task_queue* tasks = arg;
+  em_task_queue_cancel(tasks);
+}
+
 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.
@@ -228,8 +250,10 @@ int em_task_queue_send(em_task_queue* queue, task t) {
     return 1;
   }
 
-  emscripten_thread_mailbox_send(
-    queue->thread, (task){.func = receive_notification, .arg = queue});
+  emscripten_thread_mailbox_send(queue->thread,
+                                 (task){.func = receive_notification,
+                                        .cancel = cancel_notification,
+                                        .arg = queue});
   emscripten_thread_mailbox_unref(queue->thread);
   return 1;
 }

system/lib/pthread/em_task_queue.h

@@ -14,6 +14,7 @@
 // A task is an arbitrary function combined with some arbitrary state.
 typedef struct task {
   void (*func)(void*);
+  void (*cancel)(void*);
   void* arg;
 } task;
 
@@ -53,6 +54,9 @@ void em_task_queue_destroy(em_task_queue* queue);
 // Execute tasks until an empty queue is observed. Internally locks the queue.
 void em_task_queue_execute(em_task_queue* queue);
 
+// Cancel all tasks in the queue. Internally locks the queue.
+void em_task_queue_cancel(em_task_queue* queue);
+
 // Not thread safe.
 static inline int em_task_queue_is_empty(em_task_queue* queue) {
   return queue->head == queue->tail;

system/lib/pthread/proxying.c

@@ -135,10 +135,7 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) {
   }
 }
 
-int emscripten_proxy_async(em_proxying_queue* q,
-                           pthread_t target_thread,
-                           void (*func)(void*),
-                           void* arg) {
+static int do_proxy(em_proxying_queue* q, pthread_t target_thread, task t) {
   assert(q != NULL);
   pthread_mutex_lock(&q->mutex);
   em_task_queue* tasks = get_or_add_tasks_for_thread(q, target_thread);
@@ -147,26 +144,105 @@ int emscripten_proxy_async(em_proxying_queue* q,
     return 0;
   }
 
-  return em_task_queue_send(tasks, (task){func, arg});
+  return em_task_queue_send(tasks, t);
+}
+
+int emscripten_proxy_async(em_proxying_queue* q,
+                           pthread_t target_thread,
+                           void (*func)(void*),
+                           void* arg) {
+  return do_proxy(q, target_thread, (task){func, NULL, arg});
 }
 
+enum ctx_state { PENDING, DONE, CANCELED };
+
 struct em_proxying_ctx {
   // The user-provided function and argument.
   void (*func)(em_proxying_ctx*, void*);
   void* arg;
-  // Set `done` to 1 and signal the condition variable once the proxied task is
-  // done.
-  int done;
+  // Update `state` and signal the condition variable once the proxied task is
+  // done or canceled.
+  enum ctx_state state;
   pthread_mutex_t mutex;
   pthread_cond_t cond;
+  // A doubly linked list of contexts associated with active work on a single
+  // thread. If the thread is canceled, it will traverse this list to find
+  // contexts that need to be canceled.
+  struct em_proxying_ctx* next;
+  struct em_proxying_ctx* prev;
 };
 
+// The key that `cancel_active_ctxs` is bound to so that it runs when a thread
+// is canceled or exits.
+static pthread_key_t active_ctxs;
+
+static void cancel_ctx(void* arg);
+static void cancel_active_ctxs(void* arg);
+
+static void init_active_ctxs(void) {
+  int ret = pthread_key_create(&active_ctxs, cancel_active_ctxs);
+  assert(ret == 0);
+  (void)ret;
+}
+
+static void add_active_ctx(em_proxying_ctx* ctx) {
+  assert(ctx != NULL);
+  em_proxying_ctx* head = pthread_getspecific(active_ctxs);
+  if (head == NULL) {
+    // This is the only active context; initialize the active contexts list.
+    ctx->next = ctx->prev = ctx;
+    pthread_setspecific(active_ctxs, ctx);
+  } else {
+    // Insert this context at the tail of the list just before `head`.
+    ctx->next = head;
+    ctx->prev = head->prev;
+    ctx->next->prev = ctx;
+    ctx->prev->next = ctx;
+  }
+}
+
+static void remove_active_ctx(em_proxying_ctx* ctx) {
+  assert(ctx != NULL);
+  assert(ctx->next != NULL);
+  assert(ctx->prev != NULL);
+  if (ctx->next == ctx) {
+    // This is the only active context; clear the active contexts list.
+    ctx->next = ctx->prev = NULL;
+    pthread_setspecific(active_ctxs, NULL);
+    return;
+  }
+
+  // Update the list head if we are removing the current head.
+  em_proxying_ctx* head = pthread_getspecific(active_ctxs);
+  if (ctx == head) {
+    pthread_setspecific(active_ctxs, head->next);
+  }
+
+  // Remove the context from the list.
+  ctx->prev->next = ctx->next;
+  ctx->next->prev = ctx->prev;
+  ctx->next = ctx->prev = NULL;
+}
+
+static void cancel_active_ctxs(void* arg) {
+  pthread_setspecific(active_ctxs, NULL);
+  em_proxying_ctx* head = arg;
+  em_proxying_ctx* curr = head;
+  do {
+    em_proxying_ctx* next = curr->next;
+    cancel_ctx(curr);
+    curr = next;
+  } while (curr != head);
+}
+
 static void em_proxying_ctx_init(em_proxying_ctx* ctx,
                                  void (*func)(em_proxying_ctx*, void*),
                                  void* arg) {
+  static pthread_once_t once = PTHREAD_ONCE_INIT;
+  pthread_once(&once, init_active_ctxs);
   *ctx = (em_proxying_ctx){.func = func,
                            .arg = arg,
-                           .done = 0,
+                           .state = PENDING,
                            .mutex = PTHREAD_MUTEX_INITIALIZER,
                            .cond = PTHREAD_COND_INITIALIZER};
 }
@@ -178,14 +254,24 @@ static void em_proxying_ctx_deinit(em_proxying_ctx* ctx) {
 
 void emscripten_proxy_finish(em_proxying_ctx* ctx) {
   pthread_mutex_lock(&ctx->mutex);
-  ctx->done = 1;
+  ctx->state = DONE;
+  remove_active_ctx(ctx);
+  pthread_mutex_unlock(&ctx->mutex);
+  pthread_cond_signal(&ctx->cond);
+}
+
+static void cancel_ctx(void* arg) {
+  em_proxying_ctx* ctx = arg;
+  pthread_mutex_lock(&ctx->mutex);
+  ctx->state = CANCELED;
   pthread_mutex_unlock(&ctx->mutex);
   pthread_cond_signal(&ctx->cond);
 }
 
 // Helper for wrapping the call with ctx as a `void (*)(void*)`.
-static void call_with_ctx(void* p) {
-  em_proxying_ctx* ctx = (em_proxying_ctx*)p;
+static void call_with_ctx(void* arg) {
+  em_proxying_ctx* ctx = arg;
+  add_active_ctx(ctx);
   ctx->func(ctx, ctx->arg);
 }
 
@@ -197,21 +283,23 @@ int emscripten_proxy_sync_with_ctx(em_proxying_queue* q,
          "Cannot synchronously wait for work proxied to the current thread");
   em_proxying_ctx ctx;
   em_proxying_ctx_init(&ctx, func, arg);
-  if (!emscripten_proxy_async(q, target_thread, call_with_ctx, &ctx)) {
+  if (!do_proxy(q, target_thread, (task){call_with_ctx, cancel_ctx, &ctx})) {
+    em_proxying_ctx_deinit(&ctx);
     return 0;
   }
   pthread_mutex_lock(&ctx.mutex);
-  while (!ctx.done) {
+  while (ctx.state == PENDING) {
     pthread_cond_wait(&ctx.cond, &ctx.mutex);
   }
   pthread_mutex_unlock(&ctx.mutex);
+  int ret = ctx.state == DONE;
   em_proxying_ctx_deinit(&ctx);
-  return 1;
+  return ret;
 }
 
 // Helper for signaling the end of the task after the user function returns.
 static void call_then_finish(em_proxying_ctx* ctx, void* arg) {
-  task* t = (task*)arg;
+  task* t = arg;
   t->func(t->arg);
   emscripten_proxy_finish(ctx);
 }
@@ -220,7 +308,7 @@ int emscripten_proxy_sync(em_proxying_queue* q,
                           pthread_t target_thread,
                           void (*func)(void*),
                           void* arg) {
-  task t = {func, arg};
+  task t = {.func = func, .arg = arg};
   return emscripten_proxy_sync_with_ctx(q, target_thread, call_then_finish, &t);
 }
 

system/lib/pthread/pthread_create.c

@@ -270,6 +270,7 @@ int __pthread_create(pthread_t* restrict res,
       self->next,
       self->prev,
       new);
+
   *res = new;
   return 0;
 }

system/lib/pthread/thread_mailbox.c

@@ -58,9 +58,12 @@ void _emscripten_thread_mailbox_shutdown(pthread_t thread) {
     emscripten_futex_wait(&thread->mailbox_refcount, count, INFINITY);
     count = thread->mailbox_refcount;
   }
-  // TODO: Cancel tasks.
 
-  // The mailbox will not be accessed again after this point.
+  // The mailbox is now closed. No more messages will be enqueued. Run the
+  // shutdown handler for any message already in the queue.
+  em_task_queue_cancel(thread->mailbox);
+
+  // The mailbox is now empty and will not be accessed again after this point.
   em_task_queue_destroy(thread->mailbox);
 }
 

test/other/metadce/test_metadce_minimal_pthreads.funcs

@@ -4,7 +4,10 @@ $__memcpy
 $__pthread_mutex_lock
 $__pthread_mutex_trylock
 $__pthread_mutex_unlock
+$__pthread_rwlock_rdlock
+$__pthread_rwlock_timedrdlock
 $__pthread_rwlock_tryrdlock
+$__pthread_rwlock_unlock
 $__pthread_self_internal
 $__pthread_setcancelstate
 $__set_thread_state
@@ -38,12 +41,15 @@ $a_inc
 $a_store
 $a_swap
 $add
+$cancel_notification
 $dispose_chunk
 $dlfree
 $dlmalloc
 $do_dispatch_to_thread
 $em_queued_call_malloc
+$em_task_queue_cancel
 $em_task_queue_create
+$em_task_queue_dequeue
 $em_task_queue_enqueue
 $em_task_queue_execute
 $em_task_queue_free
@@ -58,6 +64,7 @@ $init_file_lock
 $init_mparams
 $main
 $memset
+$nodtor
 $pthread_attr_destroy
 $receive_notification
 $sbrk

test/other/metadce/test_metadce_minimal_pthreads.size

@@ -1 +1 @@
-18494
+18911

test/pthread/test_pthread_proxying.c

@@ -293,7 +293,7 @@ void test_tasks_queue_growth(void) {
   printf("Testing tasks queue growth\n");
 
   em_proxying_queue* queue = em_proxying_queue_create();
-  assert(proxy_queue != NULL);
+  assert(queue != NULL);
 
   int incremented = 0;