Add test (excluding postMessage notification)

Author: tlively

system/lib/pthread/proxying.c

@@ -239,15 +239,18 @@ int emscripten_proxy_async(em_proxying_queue* q,
     goto failed;
   }
   pthread_mutex_unlock(&q->mutex);
-  // If the queue was previously empty, notify the target thread to process it.
-  // Otherwise, the target thread was already notified when the existing work
-  // was enqueued so we don't need to notify it again.
-  if (empty) {
-    // TODO: Add `q` to this notification so the target thread knows which queue
-    // to process.
-    _emscripten_notify_thread_queue(target_thread,
-                                    emscripten_main_browser_thread_id());
-  }
+  /* // If the queue was previously empty, notify the target thread to process
+   * it. */
+  /* // Otherwise, the target thread was already notified when the existing work
+   */
+  /* // was enqueued so we don't need to notify it again. */
+  /* if (empty) { */
+  /*   // TODO: Add `q` to this notification so the target thread knows which
+   * queue */
+  /*   // to process. */
+  /*   _emscripten_notify_thread_queue(target_thread, */
+  /*                                   emscripten_main_browser_thread_id()); */
+  /* } */
   return 1;
 
 failed:

tests/pthread/test_pthread_proxying.c

@@ -0,0 +1,270 @@
+#define _GNU_SOURCE
+#include <assert.h>
+#include <emscripten.h>
+#include <pthread.h>
+#include <sched.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "proxying.h"
+
+// 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.
+pthread_t main_thread;
+pthread_t looper;
+pthread_t returner;
+
+// The queue used to send work to both `looper` and `returner`.
+em_proxying_queue* proxy_queue = NULL;
+_Atomic int should_quit = 0;
+
+void* looper_main(void* arg) {
+  while (!should_quit) {
+    emscripten_proxy_execute_queue(proxy_queue);
+    sched_yield();
+  }
+  return NULL;
+}
+
+void* returner_main(void* arg) { return NULL; }
+
+typedef struct widget {
+  // `val` will be stored to `out` and the current thread will be stored to
+  // `thread` when the widget is run.
+  int* out;
+  int val;
+  pthread_t thread;
+
+  // Synchronization to allow waiting on a widget to run.
+  pthread_mutex_t mutex;
+  pthread_cond_t cond;
+
+  // Nonzero iff the widget has been run.
+  int done;
+
+  // Only used for async_as_sync tests.
+  em_proxying_ctx* ctx;
+} widget;
+
+void init_widget(widget* w, int* out, int val) {
+  *w = (widget){.out = out,
+                .val = val,
+                // .thread will be set in `run_widget`.
+                .mutex = PTHREAD_MUTEX_INITIALIZER,
+                .cond = PTHREAD_COND_INITIALIZER,
+                .done = 0,
+                .ctx = NULL};
+}
+
+void destroy_widget(widget* w) {
+  pthread_mutex_destroy(&w->mutex);
+  pthread_cond_destroy(&w->cond);
+}
+
+void run_widget(widget* w) {
+  pthread_t self = pthread_self();
+  const char* name = pthread_equal(self, main_thread) ? "main"
+                     : pthread_equal(self, looper)    ? "looper"
+                     : pthread_equal(self, returner)  ? "returner"
+                                                      : "unknown";
+  printf("running widget %d on %s\n", w->val, name);
+  pthread_mutex_lock(&w->mutex);
+  if (w->out) {
+    *w->out = w->val;
+  }
+  w->thread = pthread_self();
+  w->done = 1;
+  pthread_mutex_unlock(&w->mutex);
+  pthread_cond_broadcast(&w->cond);
+}
+
+void await_widget(widget* w) {
+  pthread_mutex_lock(&w->mutex);
+  while (!w->done) {
+    pthread_cond_wait(&w->cond, &w->mutex);
+  }
+  pthread_mutex_unlock(&w->mutex);
+}
+
+// Helper functions we will proxy to perform our work.
+
+void do_run_widget(void* arg) { run_widget((widget*)arg); }
+
+void finish_running_widget(void* arg) {
+  widget* w = (widget*)arg;
+  run_widget(w);
+  emscripten_proxy_finish(w->ctx);
+}
+
+void start_running_widget(em_proxying_ctx* ctx, void* arg) {
+  ((widget*)arg)->ctx = ctx;
+  emscripten_async_call(finish_running_widget, arg, 0);
+}
+
+void start_and_finish_running_widget(em_proxying_ctx* ctx, void* arg) {
+  ((widget*)arg)->ctx = ctx;
+  finish_running_widget(arg);
+}
+
+// Main test functions
+
+void test_proxy_async(void) {
+  printf("Testing async proxying\n");
+
+  int i = 0;
+  widget w1, w2, w3;
+  init_widget(&w1, &i, 1);
+  init_widget(&w2, &i, 2);
+  init_widget(&w3, &i, 3);
+
+  // Proxy to ourselves.
+  emscripten_proxy_async(proxy_queue, pthread_self(), do_run_widget, &w1);
+  assert(!w1.done);
+  emscripten_proxy_execute_queue(proxy_queue);
+  assert(i == 1);
+  assert(w1.done);
+  assert(pthread_equal(w1.thread, pthread_self()));
+
+  // Proxy to looper.
+  emscripten_proxy_async(proxy_queue, looper, do_run_widget, &w2);
+  await_widget(&w2);
+  assert(i == 2);
+  assert(w2.done);
+  assert(pthread_equal(w2.thread, looper));
+
+  /* // Proxy to returner. */
+  /* emscripten_proxy_async(proxy_queue, returner, do_run_widget, &w3); */
+  /* await_widget(&w3); */
+  /* assert(i == 3); */
+  /* assert(w3.done); */
+  /* assert(pthread_equal(w3.thread, returner)); */
+
+  destroy_widget(&w1);
+  destroy_widget(&w2);
+  destroy_widget(&w3);
+}
+
+void test_proxy_sync(void) {
+  printf("Testing sync proxying\n");
+
+  int i = 0;
+  widget w4, w5;
+  init_widget(&w4, &i, 4);
+  init_widget(&w5, &i, 5);
+
+  // Proxy to looper.
+  emscripten_proxy_sync(proxy_queue, looper, do_run_widget, &w4);
+  assert(i == 4);
+  assert(w4.done);
+  assert(pthread_equal(w4.thread, looper));
+
+  /* // Proxy to returner. */
+  /* emscripten_proxy_sync(proxy_queue, returner, do_run_widget, &w5); */
+  /* assert(i == 5); */
+  /* assert(w5.done); */
+  /* assert(pthread_equal(w5.thread, returner)); */
+
+  destroy_widget(&w4);
+  destroy_widget(&w5);
+}
+
+void test_proxy_sync_with_ctx(void) {
+  printf("Testing sync_with_ctx proxying\n");
+
+  int i = 0;
+  widget w6, w7;
+  init_widget(&w6, &i, 6);
+  init_widget(&w7, &i, 7);
+
+  // Proxy to looper.
+  emscripten_proxy_sync_with_ctx(
+    proxy_queue, looper, start_and_finish_running_widget, &w6);
+  assert(i == 6);
+  assert(w6.done);
+  assert(pthread_equal(w6.thread, looper));
+
+  /* // Proxy to returner. */
+  /* emscripten_proxy_sync_with_ctx(proxy_queue, returner, start_running_widget,
+   * &w7); */
+  /* assert(i == 7); */
+  /* assert(w7.done); */
+  /* assert(pthread_equal(w7.thread, returner)); */
+
+  destroy_widget(&w6);
+  destroy_widget(&w7);
+}
+
+typedef struct increment_to_arg {
+  int* ip;
+  int i;
+} increment_to_arg;
+
+void increment_to(void* arg_p) {
+  increment_to_arg* arg = (increment_to_arg*)arg_p;
+  assert(*arg->ip == arg->i - 1);
+  *arg->ip = arg->i;
+  free(arg);
+}
+
+void test_queue_growth(void) {
+  printf("Testing queue growth\n");
+
+  em_proxying_queue* queue = em_proxying_queue_create();
+  assert(proxy_queue != NULL);
+
+  int incremented = 0;
+
+  // Initial queue capacity is 128. Force that to double twice with the head at
+  // index 0 by inserting more than 256 items.
+  for (int i = 1; i <= 300; i++) {
+    increment_to_arg* arg = malloc(sizeof(increment_to_arg));
+    *arg = (increment_to_arg){&incremented, i};
+    int res = emscripten_proxy_async(queue, pthread_self(), increment_to, arg);
+    assert(res == 1);
+  }
+
+  // Drain the queue, moving the head somewhere into the middle of the buffer of
+  // capacity 512.
+  emscripten_proxy_execute_queue(queue);
+  assert(incremented == 300);
+
+  // Double the queue size twice more by inserting more than 1024 items.
+  for (int i = 301; i <= 1500; i++) {
+    increment_to_arg* arg = malloc(sizeof(increment_to_arg));
+    *arg = (increment_to_arg){&incremented, i};
+    int res = emscripten_proxy_async(queue, pthread_self(), increment_to, arg);
+    assert(res == 1);
+  }
+
+  // Drain the queue again.
+  emscripten_proxy_execute_queue(queue);
+  assert(incremented == 1500);
+
+  em_proxying_queue_destroy(queue);
+}
+
+void force_exit(void* arg) { emscripten_force_exit(0); }
+
+int main(int argc, char* argv[]) {
+  main_thread = pthread_self();
+
+  proxy_queue = em_proxying_queue_create();
+  assert(proxy_queue != NULL);
+
+  pthread_create(&looper, NULL, looper_main, NULL);
+  pthread_create(&returner, NULL, returner_main, NULL);
+
+  test_proxy_async();
+  test_proxy_sync();
+  test_proxy_sync_with_ctx();
+
+  should_quit = 1;
+  pthread_join(looper, NULL);
+  pthread_join(returner, NULL);
+  em_proxying_queue_destroy(proxy_queue);
+
+  test_queue_growth();
+
+  printf("done\n");
+}

tests/pthread/test_pthread_proxying.out

@@ -0,0 +1,9 @@
+Testing async proxying
+running widget 1 on main
+running widget 2 on looper
+Testing sync proxying
+running widget 4 on looper
+Testing sync_with_ctx proxying
+running widget 6 on looper
+Testing queue growth
+done

tests/test_core.py

@@ -2419,6 +2419,13 @@ def test_pthread_specific(self):
   def test_pthread_equal(self):
     self.do_run_in_out_file_test('pthread/test_pthread_equal.cpp')
 
+  @node_pthreads
+  def test_pthread_proxying(self):
+    self.set_setting('EXIT_RUNTIME')
+    self.set_setting('PTHREAD_POOL_SIZE=2')
+    args = [f'-I{path_from_root("system/lib/pthread")}']
+    self.do_run_in_out_file_test('pthread/test_pthread_proxying.c', emcc_args=args)
+
   @node_pthreads
   def test_pthread_dispatch_after_exit(self):
     self.do_run_in_out_file_test('pthread/test_pthread_dispatch_after_exit.c', interleaved_output=False)