[Proxying] Add emscripten_proxy_callback_with_ctx (#18810)

This new proxying function is like `emscripten_proxy_callback`, but supports
proxying of asynchronous work just like `emscripten_proxy_sync_with_ctx`. It
uses the existing `em_proxying_ctx` type and `emscripten_proxy_finish` function
to mark the work done, and that type and function are internally augmented to
handle both synchronous and callback-based proxying. `emscripten_proxy_callback`
is reimplemented to share most of its implementation with
`emscripten_proxy_callback_with_ctx`, but it does break the abstraction slightly
to avoid having to make an extra allocation.

Author: tlively

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

@@ -90,9 +90,9 @@ Functions
 
   The same as ``emscripten_proxy_sync`` except that instead of waiting for the
   proxied function to return, it waits for the proxied task to be explicitly
-  marked finished with ``emscripten_proxying_finish``. ``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.
+  marked finished with ``emscripten_proxy_finish``. ``func`` need not call
+  ``emscripten_proxy_finish`` itself; it could instead store the context pointer
+  and call ``emscripten_proxy_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)
 
@@ -103,6 +103,16 @@ Functions
   receive the same argument, ``arg``. Returns 1 if ``func`` was successfully
   enqueued and the target thread notified or 0 otherwise.
 
+.. c:function:: int emscripten_proxy_callback_with_ctx(em_proxying_queue* q, pthread_t target_thread, void (*func)(em_proxying_ctx*, void*), void (*callback)(void*), void (*cancel)(void*), void* arg)
+
+  Enqueue ``func`` on the given queue and thread. Once (and if) it finishes the
+  task by calling ``emscripten_proxy_finish`` on the given ``em_proxying_ctx``,
+  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,22 +144,29 @@ 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 proxyCallback(pthread_t target, std::function<void()>&& func, std::function<void()>&& callback, std::function<void()>&& cancel)
-
-    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)
 
     Calls ``emscripten_proxy_sync`` to execute ``func``, returning ``true`` if the
     function was successfully completed or ``false`` otherwise.
 
   .. cpp:member:: bool proxySyncWithCtx(const pthread_t target, const std::function<void(ProxyingCtx)>& func)
 
-    Calls ``emscripten_proxy_sync_with_ctx`` to execute ``func``, returning ``true``
-    if the function was successfully marked done with
-    ``emscripten_proxying_finish`` or ``ProxyingCtx::finish`` and ``false`` otherwise.
+    Calls ``emscripten_proxy_sync_with_ctx`` to execute ``func``, returning
+    ``true`` if the function was successfully marked done with
+    ``emscripten_proxy_finish`` or ``ProxyingCtx::finish`` and ``false``
+    otherwise.
+
+  .. cpp:member:: bool proxyCallback(pthread_t target, std::function<void()>&& func, std::function<void()>&& callback, std::function<void()>&& cancel)
+
+    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 proxyCallbackWithCtx(pthread_t target, std::function<void(ProxyingCtx)>&& func, std::function<void()>&& callback, std::function<void()>&& cancel)
+
+    Calls ``emscripten_proxy_callback_with_ctx`` to execute ``func`` and
+    schedule either ``callback`` or ``cancel``, returning ``true`` if the
+    function was successfully enqueued and ``false`` otherwise.
 
 .. _proxying.h: https://github.com/emscripten-core/emscripten/blob/main/system/include/emscripten/proxying.h
 .. _test_pthread_proxying.c: https://github.com/emscripten-core/emscripten/blob/main/test/pthread/test_pthread_proxying.c

system/include/emscripten/proxying.h

@@ -64,12 +64,12 @@ int emscripten_proxy_sync(em_proxying_queue* q,
                           void* arg);
 
 // Enqueue `func` on the given queue and thread and wait for it to be executed
-// and for the task to be marked finished with `emscripten_proxying_finish`
-// 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, including if the target
-// thread is canceled or exits before the work is completed.
+// and for the task to be marked finished with `emscripten_proxy_finish` before
+// returning. `func` need not call `emscripten_proxy_finish` itself; it could
+// instead store the context pointer and call `emscripten_proxy_finish` at an
+// arbitrary later time. Returns 1 if the 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*),
@@ -88,6 +88,20 @@ int emscripten_proxy_callback(em_proxying_queue* q,
                               void (*cancel)(void*),
                               void* arg);
 
+// Enqueue `func` on the given queue and thread. Once (and if) it finishes the
+// task by calling `emscripten_proxy_finish` on the given `em_proxying_ctx`, 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_with_ctx(em_proxying_queue* q,
+                                       pthread_t target_thread,
+                                       void (*func)(em_proxying_ctx*, void*),
+                                       void (*callback)(void*),
+                                       void (*cancel)(void*),
+                                       void* arg);
+
 #ifdef __cplusplus
 } // extern "C"
 
@@ -103,6 +117,18 @@ namespace emscripten {
 
 // A thin C++ wrapper around the underlying C API.
 class ProxyingQueue {
+public:
+  // Simple wrapper around `em_proxying_ctx*` providing a `finish` method as an
+  // alternative to `emscripten_proxy_finish`.
+  struct ProxyingCtx {
+    em_proxying_ctx* ctx;
+
+    ProxyingCtx() = default;
+    ProxyingCtx(em_proxying_ctx* ctx) : ctx(ctx) {}
+    void finish() { emscripten_proxy_finish(ctx); }
+  };
+
+private:
   static void runAndFree(void* arg) {
     auto* f = (std::function<void()>*)arg;
     (*f)();
@@ -150,6 +176,37 @@ class ProxyingQueue {
     delete info;
   }
 
+  struct CallbackWithCtxFuncs {
+    std::function<void(ProxyingCtx)> func;
+    std::function<void()> callback;
+    std::function<void()> cancel;
+
+    CallbackWithCtxFuncs(std::function<void(ProxyingCtx)>&& func,
+                         std::function<void()>&& callback,
+                         std::function<void()>&& cancel)
+      : func(std::move(func)), callback(std::move(callback)),
+        cancel(std::move(cancel)) {}
+  };
+
+  static void runFuncWithCtx(em_proxying_ctx* ctx, void* arg) {
+    auto* info = (CallbackWithCtxFuncs*)arg;
+    info->func(ProxyingCtx{ctx});
+  }
+
+  static void runCallbackWithCtx(void* arg) {
+    auto* info = (CallbackWithCtxFuncs*)arg;
+    info->callback();
+    delete info;
+  }
+
+  static void runCancelWithCtx(void* arg) {
+    auto* info = (CallbackWithCtxFuncs*)arg;
+    if (info->cancel) {
+      info->cancel();
+    }
+    delete info;
+  }
+
 public:
   em_proxying_queue* queue = em_proxying_queue_create();
 
@@ -179,16 +236,6 @@ class ProxyingQueue {
     }
   }
 
-  // Simple wrapper around `em_proxying_ctx*` providing a `finish` method as an
-  // alternative to `emscripten_proxy_finish`.
-  struct ProxyingCtx {
-    em_proxying_ctx* ctx;
-
-    ProxyingCtx() = default;
-    ProxyingCtx(em_proxying_ctx* ctx) : ctx(ctx) {}
-    void finish() { emscripten_proxy_finish(ctx); }
-  };
-
   void execute() { emscripten_proxy_execute_queue(queue); }
 
   // Return true if the work was successfully enqueued and false otherwise.
@@ -225,6 +272,24 @@ class ProxyingQueue {
     }
     return true;
   }
+
+  bool proxyCallbackWithCtx(pthread_t target,
+                            std::function<void(ProxyingCtx)>&& func,
+                            std::function<void()>&& callback,
+                            std::function<void()>&& cancel) {
+    CallbackWithCtxFuncs* info = new CallbackWithCtxFuncs(
+      std::move(func), std::move(callback), std::move(cancel));
+    if (!emscripten_proxy_callback_with_ctx(queue,
+                                            target,
+                                            runFuncWithCtx,
+                                            runCallbackWithCtx,
+                                            runCancelWithCtx,
+                                            info)) {
+      delete info;
+      return false;
+    }
+    return true;
+  }
 };
 
 } // namespace emscripten