Add C++ wrapper around the proxying.h API (#16468)

tlively · web-flow · commit 240277ee379d · 2022-03-21T09:13:12.000-07:00
Add a thin C++ wrapper API to proxying.h that wraps the underlying C API in the
same file. The C++ wrapper is defined only when using C++ at least as recent as
C++11. The wrapper handles creating and destroying the underlying
em_proxying_queue objects automatically, defines proxying methods that take
`std::function` rather than void*, and wraps `em_proxying_ctx` in a class with a
`finish()` method.
diff --git a/system/lib/pthread/proxying.h b/system/lib/pthread/proxying.h
@@ -43,7 +43,7 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q);
 // of the task.
 typedef struct em_proxying_ctx em_proxying_ctx;
 
-// Signal the end of a proxied task.
+// Signal the end of a task proxied with `emscripten_proxy_sync_with_ctx`.
 void emscripten_proxy_finish(em_proxying_ctx* ctx);
 
 // Enqueue `func` on the given queue and thread and return immediately. Returns
@@ -64,13 +64,103 @@ int emscripten_proxy_sync(em_proxying_queue* q,
 
 // 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. Returns 1 if the task was successfully completed and 0
-// otherwise.
+// 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.
 int emscripten_proxy_sync_with_ctx(em_proxying_queue* q,
                                    pthread_t target_thread,
                                    void (*func)(em_proxying_ctx*, void*),
                                    void* arg);
 
 #ifdef __cplusplus
-}
-#endif
+} // extern "C"
+
+#if __cplusplus < 201103L
+#warning "C++ ProxyingQueue support requires building with -std=c++11 or newer!"
+#else
+
+#include <functional>
+#include <thread>
+#include <utility>
+
+namespace emscripten {
+
+// A thin C++ wrapper around the underlying C API.
+class ProxyingQueue {
+  em_proxying_queue* queue = em_proxying_queue_create();
+
+  static void runAndFree(void* arg) {
+    auto f = (std::function<void()>*)arg;
+    (*f)();
+    delete f;
+  }
+
+  static void run(void* arg) {
+    auto f = *(std::function<void()>*)arg;
+    f();
+  }
+
+  static void runWithCtx(em_proxying_ctx* ctx, void* arg) {
+    auto f = *(std::function<void(ProxyingCtx)>*)arg;
+    f(ProxyingCtx{ctx});
+  }
+
+public:
+  // ProxyingQueue can be moved but not copied. It is not valid to call any
+  // methods on ProxyingQueues that have been moved out of.
+  ProxyingQueue() = default;
+  ProxyingQueue& operator=(const ProxyingQueue&) = delete;
+  ProxyingQueue& operator=(ProxyingQueue&& other) {
+    if (queue) {
+      em_proxying_queue_destroy(queue);
+    }
+    queue = other.queue;
+    other.queue = nullptr;
+    return *this;
+  }
+
+  ProxyingQueue(const ProxyingQueue&) = delete;
+  ProxyingQueue(ProxyingQueue&& other) : queue(nullptr) {
+    *this = std::move(other);
+  }
+
+  ~ProxyingQueue() {
+    if (queue) {
+      em_proxying_queue_destroy(queue);
+    }
+  }
+
+  // Simple wrapper around `em_proxying_ctx*` providing a `finish` method as an
+  // alternative to `emscripten_proxy_finish`.
+  struct ProxyingCtx {
+    em_proxying_ctx* ctx;
+
+    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.
+  // 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 proxySync(const pthread_t target, const std::function<void()>& func) {
+    return emscripten_proxy_sync(queue, target, run, (void*)&func);
+  }
+
+  bool proxySyncWithCtx(const pthread_t target,
+                        const std::function<void(ProxyingCtx)>& func) {
+    return emscripten_proxy_sync_with_ctx(
+      queue, target, runWithCtx, (void*)&func);
+  }
+};
+
+} // namespace emscripten
+
+#endif // __cplusplus < 201103L
+#endif // __cplusplus
diff --git a/tests/pthread/test_pthread_proxying.c b/tests/pthread/test_pthread_proxying.c
@@ -351,5 +351,4 @@ int main(int argc, char* argv[]) {
   test_proxying_queue_growth();
 
   printf("done\n");
-  emscripten_force_exit(0);
 }
diff --git a/tests/pthread/test_pthread_proxying_cpp.cpp b/tests/pthread/test_pthread_proxying_cpp.cpp
@@ -0,0 +1,158 @@
+#include <cassert>
+#include <condition_variable>
+#include <iostream>
+#include <sched.h>
+
+#include "proxying.h"
+
+using namespace emscripten;
+
+// The worker threads we will use. `looper` sits in a loop, continuously
+// processing work as it becomes available, while `returner` returns to the JS
+// event loop each time it processes work.
+std::thread looper;
+std::thread returner;
+
+// The queue used to send work to both `looper` and `returner`.
+ProxyingQueue queue;
+
+// Whether `looper` should exit.
+std::atomic<bool> should_quit{false};
+
+// Whether `returner` has spun up.
+std::atomic<bool> has_begun{false};
+
+void looper_main() {
+  while (!should_quit) {
+    queue.execute();
+    sched_yield();
+  }
+}
+
+void returner_main() {
+  has_begun = 1;
+  emscripten_exit_with_live_runtime();
+}
+
+void test_proxy_async() {
+  std::cout << "Testing async proxying\n";
+
+  int i = 0;
+
+  std::mutex mutex;
+  std::condition_variable cond;
+  std::thread::id executor;
+
+  // Proxy to ourselves.
+  queue.proxyAsync(pthread_self(), [&]() {
+    i = 1;
+    executor = std::this_thread::get_id();
+  });
+  assert(i == 0);
+  queue.execute();
+  assert(i == 1);
+  assert(executor == std::this_thread::get_id());
+
+  // Proxy to looper.
+  {
+    queue.proxyAsync(looper.native_handle(), [&]() {
+      i = 2;
+      executor = std::this_thread::get_id();
+      cond.notify_one();
+    });
+    std::unique_lock<std::mutex> lock(mutex);
+    cond.wait(lock, [&]() { return i == 2; });
+    assert(executor == looper.get_id());
+  }
+
+  // Proxy to returner.
+  {
+    queue.proxyAsync(returner.native_handle(), [&]() {
+      i = 3;
+      executor = std::this_thread::get_id();
+      cond.notify_one();
+    });
+    std::unique_lock<std::mutex> lock(mutex);
+    cond.wait(lock, [&]() { return i == 3; });
+    assert(executor == returner.get_id());
+  }
+}
+
+void test_proxy_sync() {
+  std::cout << "Testing sync proxying\n";
+
+  int i = 0;
+  std::thread::id executor;
+
+  // Proxy to looper.
+  {
+    queue.proxySync(looper.native_handle(), [&]() {
+      i = 2;
+      executor = std::this_thread::get_id();
+    });
+    assert(i == 2);
+    assert(executor == looper.get_id());
+  }
+
+  // Proxy to returner.
+  {
+    queue.proxySync(returner.native_handle(), [&]() {
+      i = 3;
+      executor = std::this_thread::get_id();
+    });
+    assert(i == 3);
+    assert(executor == returner.get_id());
+  }
+}
+
+void test_proxy_sync_with_ctx(void) {
+  std::cout << "Testing sync proxying\n";
+
+  int i = 0;
+  std::thread::id executor;
+
+  // Proxy to looper.
+  {
+    queue.proxySyncWithCtx(looper.native_handle(), [&](auto ctx) {
+      i = 2;
+      executor = std::this_thread::get_id();
+      ctx.finish();
+    });
+    assert(i == 2);
+    assert(executor == looper.get_id());
+  }
+
+  // Proxy to returner.
+  {
+    queue.proxySyncWithCtx(returner.native_handle(), [&](auto ctx) {
+      i = 3;
+      executor = std::this_thread::get_id();
+      auto finish = (void(*)(void*))emscripten_proxy_finish;
+      emscripten_async_call(finish, ctx.ctx, 0);
+    });
+    assert(i == 3);
+    assert(executor == returner.get_id());
+  }
+}
+
+int main(int argc, char* argv[]) {
+  looper = std::thread(looper_main);
+  returner = std::thread(returner_main);
+
+  // `returner` can't process its queue until it starts up.
+  while (!has_begun) {
+    sched_yield();
+  }
+
+  test_proxy_async();
+  test_proxy_sync();
+  test_proxy_sync_with_ctx();
+
+  should_quit = true;
+  looper.join();
+
+  pthread_cancel(returner.native_handle());
+  returner.join();
+
+  std::cout << "done\n";
+}
diff --git a/tests/pthread/test_pthread_proxying_cpp.out b/tests/pthread/test_pthread_proxying_cpp.out
@@ -0,0 +1,4 @@
+Testing async proxying
+Testing sync proxying
+Testing sync proxying
+done
diff --git a/tests/test_core.py b/tests/test_core.py
@@ -2578,6 +2578,15 @@ def test_pthread_proxying(self):
     self.do_run_in_out_file_test('pthread/test_pthread_proxying.c',
                                  emcc_args=args, interleaved_output=False)
 
+  @node_pthreads
+  def test_pthread_proxying_cpp(self):
+    self.set_setting('EXIT_RUNTIME')
+    self.set_setting('PROXY_TO_PTHREAD')
+    self.set_setting('INITIAL_MEMORY=32mb')
+    args = [f'-I{path_from_root("system/lib/pthread")}']
+    self.do_run_in_out_file_test('pthread/test_pthread_proxying_cpp.cpp',
+                                 emcc_args=args, interleaved_output=False)
+
   @node_pthreads
   def test_pthread_proxying_dropped_work(self):
     self.set_setting('EXIT_RUNTIME')

Original file line number	Diff line number	Diff line change
`@@ -351,5 +351,4 @@ int main(int argc, char* argv[]) {`
`351`	`351`	`test_proxying_queue_growth();`
`352`	`352`
`353`	`353`	`printf("done\n");`
`354`		`- emscripten_force_exit(0);`
`355`	`354`	`}`