[Proxying] Update proxying with callback to support cancellation (#18776)

Completely overhaul the interface for proxying with callbacks:

 - Rename `emscripten_proxy_async_with_callback` to the simpler
   `emscripten_proxy_callback`.

 - Add an optional `cancel` callback argument that will be scheduled on the
   proxying thread if the worker thread dies before the work is started. A
   future PR will change this so that the `cancel` callback will be called if a
   worker thread dies before the work is finished to be more consistent with the
   `proxy_sync` APIs.

 - Remove the ability to receive the result of the proxied function as an
   argument to the callback function since the extra expressiveness was not
   general enough to warrant the added complexity.

 - Use only a single context argument rather than a separate context for the
   proxied function and the callback. This is just as expressive, but it is
   simpler avoids the question of what context object should be passed to the
   cancellation callback.

Also update the only user of the callback API, dynlink.c, to use the new API. A
future PR will implement a promise-based proxying API and update dynlink.c to
use that instead, so this PR just tries to make the minimal necessary changes to
dynlink.c.

Author: tlively

ChangeLog.md

@@ -48,6 +48,9 @@ See docs/process.md for more on how version tagging works.
 - 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.
+- The `emscripten_proxy_async_with_callback` API was replaced with a simpler
+  `emscripten_proxy_callback` API that takes a second callback to be called if
+  the worker thread dies before completing the proxied work.
 
 3.1.31 - 01/26/23
 -----------------

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

@@ -78,15 +78,6 @@ Functions
   thread then return immediately without waiting for ``func`` to be executed.
   Returns 1 if the work was successfully enqueued or 0 otherwise.
 
-.. c:function:: int emscripten_proxy_async_with_callback(em_proxying_queue* q, pthread_t target_thread, void* (*func)(void*), void* arg, void (*callback)(void* arg, void* result), void* callback_arg)
-
-  Enqueue `func` on the given queue and thread. Once (and if) it finishes
-  executing, it will asynchronously proxy `callback` back to the current thread
-  on the same queue. The result of the proxied function will be passed as the
-  second argument to the callback. Returns 1 if the initial work was
-  successfully enqueued and the target thread notified or 0 otherwise. If the
-  callback cannot be scheduled (for example due to OOM), the program is aborted.
-
 .. c:function:: int emscripten_proxy_sync(em_proxying_queue* q, pthread_t target_thread, void (*func)(void*), void* arg)
 
   Enqueue ``func`` to be called with argument ``arg`` on the given queue and
@@ -103,6 +94,15 @@ Functions
   ``emscripten_proxying_finish`` itself; it could instead store the context
   pointer and call ``emscripten_proxying_finish`` at an arbitrary later time.
 
+.. c:function:: int emscripten_proxy_callback(em_proxying_queue* q, pthread_t target_thread, void (*func)(void*), void (*callback)(void*), void (*cancel)(void*), void* arg)
+
+  Enqueue ``func`` on the given queue and thread. Once (and if) it finishes
+  executing, it will asynchronously proxy ``callback`` back to the current
+  thread on the same queue, or if the target thread dies before the work can be
+  completed, ``cancel`` will be proxied back instead. All three function will
+  receive the same argument, ``arg``. Returns 1 if ``func`` was successfully
+  enqueued and the target thread notified or 0 otherwise.
+
 C++ API
 -------
 
@@ -134,11 +134,11 @@ defined within namespace ``emscripten``.
     Calls ``emscripten_proxy_async`` to execute ``func``, returning ``true`` if the
     function was successfully enqueued and ``false`` otherwise.
 
-  .. cpp:member:: bool proxyAsyncWithCallback(pthread_t target, std::function<void()>&& func, std::function<void()>&& callback)
+  .. cpp:member:: bool proxyCallback(pthread_t target, std::function<void()>&& func, std::function<void()>&& callback, std::function<void()>&& cancel)
 
-    Calls ``emscripten_proxy_async_with_callback`` to execute ``func`` and
-    schedule ``callback``, returning ``true`` if the function was successfully
-    enqueued and ``false`` otherwise.
+    Calls ``emscripten_proxy_callback`` to execute ``func`` and schedule either
+    ``callback`` or ``cancel``, returning ``true`` if the function was
+    successfully enqueued and ``false`` otherwise.
 
   .. cpp:member:: bool proxySync(const pthread_t target, const std::function<void()>& func)
 

src/library_pthread.js

@@ -1085,6 +1085,13 @@ var LibraryPThread = {
 #if PTHREADS_DEBUG
     dbg('invokeEntryPoint: ' + ptrToString(ptr));
 #endif
+#if EXIT_RUNTIME && !MINIMAL_RUNTIME
+    // An old thread on this worker may have been canceled without returning the
+    // `runtimeKeepaliveCounter` to zero. Reset it now so the new thread won't
+    // be affected.
+    runtimeKeepaliveCounter = 0;
+#endif
+
 #if MAIN_MODULE
     // Before we call the thread entry point, make sure any shared libraries
     // have been loaded on this there.  Otherwise our table migth be not be

system/include/emscripten/proxying.h

@@ -54,21 +54,6 @@ int emscripten_proxy_async(em_proxying_queue* q,
                            void (*func)(void*),
                            void* arg);
 
-// Enqueue `func` on the given queue and thread. Once (and if) it finishes
-// executing, it will asynchronously proxy `callback` back to the current thread
-// on the same queue. The result of the proxied function will be passed as the
-// second argument to the callback. Returns 1 if the initial work was
-// successfully enqueued and the target thread notified or 0 otherwise. If the
-// callback cannot be scheduled (for example due to OOM), the program is
-// aborted.
-int emscripten_proxy_async_with_callback(em_proxying_queue* q,
-                                         pthread_t target_thread,
-                                         void* (*func)(void*),
-                                         void* arg,
-                                         void (*callback)(void* arg,
-                                                          void* result),
-                                         void* callback_arg);
-
 // 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, including if the target thread is canceled or exits before
@@ -90,6 +75,19 @@ int emscripten_proxy_sync_with_ctx(em_proxying_queue* q,
                                    void (*func)(em_proxying_ctx*, void*),
                                    void* arg);
 
+// Enqueue `func` on the given queue and thread. Once (and if) it finishes
+// executing, it will asynchronously proxy `callback` back to the current thread
+// on the same queue, or if the target thread dies before the work can be
+// completed, `cancel` will be proxied back instead. All three function will
+// receive the same argument, `arg`. Returns 1 if `func` was successfully
+// enqueued and the target thread notified or 0 otherwise.
+int emscripten_proxy_callback(em_proxying_queue* q,
+                              pthread_t target_thread,
+                              void (*func)(void*),
+                              void (*callback)(void*),
+                              void (*cancel)(void*),
+                              void* arg);
+
 #ifdef __cplusplus
 } // extern "C"
 
@@ -111,19 +109,6 @@ class ProxyingQueue {
     delete f;
   }
 
-  static void* runAndFreeWithResult(void* arg) {
-    auto* f = (std::function<void*()>*)arg;
-    void* result = (*f)();
-    delete f;
-    return result;
-  }
-
-  static void runAndFreeCallback(void* arg, void* result) {
-    auto* f = (std::function<void(void*)>*)arg;
-    (*f)(result);
-    delete f;
-  }
-
   static void run(void* arg) {
     auto* f = (std::function<void()>*)arg;
     (*f)();
@@ -134,6 +119,37 @@ class ProxyingQueue {
     (*f)(ProxyingCtx{ctx});
   }
 
+  struct CallbackFuncs {
+    std::function<void()> func;
+    std::function<void()> callback;
+    std::function<void()> cancel;
+
+    CallbackFuncs(std::function<void()>&& func,
+                  std::function<void()>&& callback,
+                  std::function<void()>&& cancel)
+      : func(std::move(func)), callback(std::move(callback)),
+        cancel(std::move(cancel)) {}
+  };
+
+  static void runFunc(void* arg) {
+    auto* info = (CallbackFuncs*)arg;
+    info->func();
+  }
+
+  static void runCallback(void* arg) {
+    auto* info = (CallbackFuncs*)arg;
+    info->callback();
+    delete info;
+  }
+
+  static void runCancel(void* arg) {
+    auto* info = (CallbackFuncs*)arg;
+    if (info->cancel) {
+      info->cancel();
+    }
+    delete info;
+  }
+
 public:
   em_proxying_queue* queue = em_proxying_queue_create();
 
@@ -142,11 +158,13 @@ class ProxyingQueue {
   ProxyingQueue() = default;
   ProxyingQueue& operator=(const ProxyingQueue&) = delete;
   ProxyingQueue& operator=(ProxyingQueue&& other) {
-    if (queue) {
-      em_proxying_queue_destroy(queue);
+    if (this != &other) {
+      if (queue) {
+        em_proxying_queue_destroy(queue);
+      }
+      queue = other.queue;
+      other.queue = nullptr;
     }
-    queue = other.queue;
-    other.queue = nullptr;
     return *this;
   }
 
@@ -177,21 +195,11 @@ class ProxyingQueue {
   // Refer to the corresponding C API documentation.
   bool proxyAsync(pthread_t target, std::function<void()>&& func) {
     std::function<void()>* arg = new std::function<void()>(std::move(func));
-    return emscripten_proxy_async(queue, target, runAndFree, (void*)arg);
-  }
-
-  bool proxyAsyncWithCallback(pthread_t target,
-                              std::function<void*()>&& func,
-                              std::function<void(void*)>&& callback) {
-    std::function<void*()>* arg = new std::function<void*()>(std::move(func));
-    std::function<void(void*)>* callback_arg =
-      new std::function<void(void*)>(std::move(callback));
-    return emscripten_proxy_async_with_callback(queue,
-                                                target,
-                                                runAndFreeWithResult,
-                                                (void*)arg,
-                                                runAndFreeCallback,
-                                                (void*)callback_arg);
+    if (!emscripten_proxy_async(queue, target, runAndFree, (void*)arg)) {
+      delete arg;
+      return false;
+    }
+    return true;
   }
 
   bool proxySync(const pthread_t target, const std::function<void()>& func) {
@@ -203,6 +211,20 @@ class ProxyingQueue {
     return emscripten_proxy_sync_with_ctx(
       queue, target, runWithCtx, (void*)&func);
   }
+
+  bool proxyCallback(pthread_t target,
+                     std::function<void()>&& func,
+                     std::function<void()>&& callback,
+                     std::function<void()>&& cancel) {
+    CallbackFuncs* info = new CallbackFuncs(
+      std::move(func), std::move(callback), std::move(cancel));
+    if (!emscripten_proxy_callback(
+          queue, target, runFunc, runCallback, runCancel, info)) {
+      delete info;
+      return false;
+    }
+    return true;
+  }
 };
 
 } // namespace emscripten

system/lib/libc/dynlink.c

@@ -303,9 +303,15 @@ bool _emscripten_dlsync_self() {
   return true;
 }
 
-static void* do_thread_sync(void* arg) {
+struct promise_result {
+  em_promise_t promise;
+  bool result;
+};
+
+static void do_thread_sync(void* arg) {
   dbg("do_thread_sync");
-  return (void*)_emscripten_dlsync_self();
+  struct promise_result* info = arg;
+  info->result = _emscripten_dlsync_self();
 }
 
 static void do_thread_sync_out(void* arg) {
@@ -315,10 +321,11 @@ static void do_thread_sync_out(void* arg) {
 }
 
 // Called once _emscripten_proxy_dlsync completes
-static void done_thread_sync(void* arg, void* result) {
-  em_promise_t promise = (em_promise_t)arg;
-  dbg("done_thread_sync: promise=%p result=%p", promise, result);
-  if (result) {
+static void done_thread_sync(void* arg) {
+  struct promise_result* info = arg;
+  em_promise_t promise = info->promise;
+  dbg("done_thread_sync: promise=%p result=%i", promise, info->result);
+  if (info->result) {
     emscripten_promise_resolve(promise, EM_PROMISE_FULFILL, NULL);
   } else {
 #if ABORT_ON_SYNC_FAILURE
@@ -328,6 +335,7 @@ static void done_thread_sync(void* arg, void* result) {
 #endif
   }
   emscripten_promise_destroy(promise);
+  free(info);
 }
 
 // Proxying queue specically for handling code loading (dlopen) events.
@@ -356,13 +364,28 @@ int _emscripten_proxy_dlsync_async(pthread_t target_thread, em_promise_t promise
   if (!dlopen_proxying_queue) {
     dlopen_proxying_queue = em_proxying_queue_create();
   }
-  int rtn = emscripten_proxy_async_with_callback(dlopen_proxying_queue,
-                                                 target_thread,
-                                                 do_thread_sync,
-                                                 NULL,
-                                                 done_thread_sync,
-                                                 promise);
-  if (target_thread->sleeping) {
+
+  struct promise_result* info = malloc(sizeof(struct promise_result));
+  if (!info) {
+    return false;
+  }
+  *info = (struct promise_result){
+    .promise = promise,
+    .result = false,
+  };
+  int rtn = emscripten_proxy_callback(dlopen_proxying_queue,
+                                      target_thread,
+                                      do_thread_sync,
+                                      done_thread_sync,
+                                      done_thread_sync,
+                                      info);
+  if (!rtn) {
+    // If we failed to proxy, then the target thread is no longer alive and no
+    // longer needs to be caught up, so we can resolve the promise early.
+    emscripten_promise_resolve(promise, EM_PROMISE_FULFILL, NULL);
+    emscripten_promise_destroy(promise);
+    free(info);
+  } else if (target_thread->sleeping) {
     // If the target thread is in the sleeping state (and this check is
     // performed after the enqueuing of the async work) then we know its safe to
     // resolve the promise early, since the thread will process our event as