From 6584c1655d499a1ce91df4db7d320b55fafc1f7e Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Thu, 9 Dec 2021 21:55:03 -0800
Subject: [PATCH 01/33] [WIP] New low-level proxying C API

Implement a new proxying API that is meant to be suitable both for proxying in
syscall implementations as well as for proxying arbitrary user work. Since the
system proxying queue is processed from so many locations, it is dangerous to
use it for arbitrary work that might take a lock or otherwise block and the work
sent to it has to be structured more like native signal handlers. To avoid this
limitation, the new API allows users to create their own proxying queues that
are processed only when the target thread returns to the JS event loop or when
the user explicitly requests processing.

In contrast to the existing proxying API, this new API:

 - Never drops tasks (except in the case of allocation failure). It grows the
   task queues as necessary instead.
 - Does not attempt to dynamically type or dispatch queued functions, but rather
   uses statically typed function pointers that take `void*` arguments. This
   simplifies both the API and implementation. Packing of varargs into
   dynamically typed function wrappers could easily be layered on top of this
   API.
 - Is less redundant. There is only one way to proxy work synchronously or
   asynchronously to any thread.
 - Is more general. It allows waiting for a task to be explicitly signaled as
   done in addition to waiting for the proxied function to return.
 - Uses arrays instead of linked lists for better data locality.
 - Has a more uniform naming convention.

A follow-up PR will reimplement the existing proxying API in terms of this new
API.
---
 system/include/emscripten/proxying.h |  68 ++++++
 system/lib/pthread/proxying.c        | 326 +++++++++++++++++++++++++++
 tools/system_libs.py                 |   1 +
 3 files changed, 395 insertions(+)
 create mode 100644 system/include/emscripten/proxying.h
 create mode 100644 system/lib/pthread/proxying.c

diff --git a/system/include/emscripten/proxying.h b/system/include/emscripten/proxying.h
new file mode 100644
index 0000000000000..a3e77112f0d4a
--- /dev/null
+++ b/system/include/emscripten/proxying.h
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2021 The Emscripten Authors.  All rights reserved.
+ * Emscripten is available under two separate licenses, the MIT license and the
+ * University of Illinois/NCSA Open Source License.  Both these licenses can be
+ * found in the LICENSE file.
+ */
+
+#pragma once
+
+#include <pthread.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// Opaque handle to a set of thread-local work queues to which work can be
+// asynchronously or synchronously proxied from other threads. When work is
+// proxied to a queue on a particular thread, that thread is notified to start
+// processing work from that queue if it is not already doing so.
+typedef struct em_proxying_queue em_proxying_queue;
+
+// Create and destroy proxying queues.
+em_proxying_queue* em_proxying_queue_create();
+void em_proxying_queue_destroy(em_proxying_queue* q);
+
+// Get the queue used for proxying low-level runtime work. Work on this queue
+// may be processed at any time inside system functions, so it must be
+// nonblocking and safe to run at any time, similar to a native signal handler.
+em_proxying_queue* emscripten_proxy_get_system_queue();
+
+// Execute all the tasks enqueued for the current thread on the given queue.
+void emscripten_proxy_execute_queue(em_proxying_queue* q);
+
+// Opaque handle to a currently-executing proxied task, used to signal the end
+// of the task.
+typedef struct em_proxying_ctx em_proxying_ctx;
+
+// Signal the end of a proxied task.
+void emscripten_proxy_finish(em_proxying_ctx* ctx);
+
+// Enqueue `func` on the given queue and thread and return immediately. Returns
+// 1 if the work was successfully enqueued and the target thread notified or 0
+// otherwise.
+int emscripten_proxy_async(em_proxying_queue* q,
+                           pthread_t target_thread,
+                           void (*func)(void*),
+                           void* 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.
+int emscripten_proxy_sync(em_proxying_queue* q,
+                          pthread_t target_thread,
+                          void (*func)(void*),
+                          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. 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
diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
new file mode 100644
index 0000000000000..d5ae077506174
--- /dev/null
+++ b/system/lib/pthread/proxying.c
@@ -0,0 +1,326 @@
+/*
+ * Copyright 2021 The Emscripten Authors.  All rights reserved.
+ * Emscripten is available under two separate licenses, the MIT license and the
+ * University of Illinois/NCSA Open Source License.  Both these licenses can be
+ * found in the LICENSE file.
+ */
+
+#include <assert.h>
+#include <emscripten/html5.h>
+#include <emscripten/proxying.h>
+#include <emscripten/threading.h>
+#include <pthread.h>
+#include <stdlib.h>
+#include <string.h>
+
+#define TASK_QUEUE_INITIAL_CAPACITY 128
+
+// TODO: Update this to take a `em_proxying_queue` argument.
+extern int _emscripten_notify_thread_queue(pthread_t target_thread,
+                                           pthread_t main_thread);
+
+typedef struct task {
+  void (*func)(void*);
+  void* arg;
+} task;
+
+// A task queue for a particular thread. Organized into a linked list of
+// task_queues for different threads.
+typedef struct task_queue {
+  // The target thread for this task_queue.
+  pthread_t thread;
+  // Recursion guard.
+  int processing;
+  // Ring buffer of tasks of size `capacity`. New tasks are enqueued at
+  // `tail` and dequeued at `head`.
+  task* tasks;
+  int capacity;
+  int head;
+  int tail;
+} task_queue;
+
+static int task_queue_empty(task_queue* tasks) {
+  return tasks->head == tasks->tail;
+}
+
+static int task_queue_full(task_queue* tasks) {
+  return tasks->head == (tasks->tail + 1) % tasks->capacity;
+}
+
+// Returns 1 on success and 0 on failure.
+static int task_queue_enqueue(task_queue* tasks, task t) {
+  if (task_queue_full(tasks)) {
+    // Allocate a larger task queue.
+    int new_capacity = tasks->capacity * 2;
+    task* new_tasks = malloc(sizeof(task) * new_capacity);
+    if (new_tasks == NULL) {
+      return 0;
+    }
+    // Copy the tasks such that the head of the queue is at the beginning of the
+    // buffer. There are two cases to handle: either the queue wraps around the
+    // end of the old buffer or it does not.
+    int queued_tasks;
+    if (tasks->head <= tasks->tail) {
+      // No wrap. Copy the tasks in one chunk.
+      queued_tasks = tasks->tail - tasks->head;
+      memcpy(
+        new_tasks, &tasks->tasks[tasks->head], sizeof(task) * queued_tasks);
+    } else {
+      // Wrap. Copy `first_queued` tasks up to the end of the old buffer and
+      // `last_queued` tasks at the beginning of the old buffer.
+      int first_queued = tasks->capacity - tasks->head;
+      int last_queued = tasks->tail;
+      queued_tasks = first_queued + last_queued;
+      memcpy(
+        new_tasks, &tasks->tasks[tasks->head], sizeof(task) * first_queued);
+      memcpy(
+        new_tasks + first_queued, tasks->tasks, sizeof(task) * last_queued);
+    }
+    free(tasks->tasks);
+    tasks->tasks = new_tasks;
+    tasks->capacity = new_capacity;
+    tasks->head = 0;
+    tasks->tail = queued_tasks;
+  }
+  // Enqueue the task.
+  tasks->tasks[tasks->tail] = t;
+  tasks->tail = (tasks->tail + 1) % tasks->capacity;
+  return 1;
+}
+
+static task task_queue_dequeue(task_queue* tasks) {
+  task t = tasks->tasks[tasks->head];
+  tasks->head = (tasks->head + 1) % tasks->capacity;
+  return t;
+}
+
+struct em_proxying_queue {
+  // Protects all accesses to all task_queues.
+  pthread_mutex_t mutex;
+  // `size` task queues stored in an array of size `capacity`.
+  task_queue* task_queues;
+  int size;
+  int capacity;
+};
+
+static em_proxying_queue system_proxying_queue = {
+  PTHREAD_MUTEX_INITIALIZER, NULL, 0, 0};
+
+em_proxying_queue* emscripten_proxy_get_system_queue(void) {
+  return &system_proxying_queue;
+}
+
+em_proxying_queue* em_proxying_queue_create(void) {
+  em_proxying_queue* q = malloc(sizeof(em_proxying_queue));
+  if (q == NULL) {
+    return NULL;
+  }
+  pthread_mutex_init(&q->mutex, NULL);
+  q->task_queues = NULL;
+  q->size = 0;
+  q->capacity = 0;
+  return q;
+}
+
+void em_proxying_queue_destroy(em_proxying_queue* q) {
+  if (q == NULL) {
+    return;
+  }
+  assert(q != &system_proxying_queue && "cannot destroy system proxying queue");
+  // No need to acquire the lock; no one should be racing with the destruction
+  // of the queue.
+  pthread_mutex_destroy(&q->mutex);
+  for (int i = 0; i < q->size; i++) {
+    free(q->task_queues[i].tasks);
+  }
+  free(q->task_queues);
+  free(q);
+}
+
+// Not thread safe.
+static task_queue* get_tasks_for_thread(em_proxying_queue* q,
+                                        pthread_t thread) {
+  assert(q != NULL);
+  task_queue* tasks = NULL;
+  for (int i = 0; i < q->size; i++) {
+    if (pthread_equal(q->task_queues[i].thread, thread)) {
+      tasks = &q->task_queues[i];
+      break;
+    }
+  }
+  return tasks;
+}
+
+// Not thread safe.
+static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
+                                               pthread_t thread) {
+  task_queue* tasks = get_tasks_for_thread(q, thread);
+  if (tasks != NULL) {
+    return tasks;
+  }
+  // There were no tasks for the thread; initialize a new task_queue. If there
+  // are not enough queues, allocate more.
+  if (q->size == q->capacity) {
+    int new_capacity = q->capacity == 0 ? 1 : q->capacity * 2;
+    task_queue* new_task_queues =
+      realloc(q->task_queues, sizeof(task_queue) * new_capacity);
+    if (new_task_queues == NULL) {
+      return NULL;
+    }
+    q->task_queues = new_task_queues;
+    q->capacity = new_capacity;
+  }
+  // Initialize the next available task queue.
+  tasks = &q->task_queues[q->size];
+  tasks->thread = thread;
+  tasks->processing = 0;
+  tasks->tasks = malloc(sizeof(task) * TASK_QUEUE_INITIAL_CAPACITY);
+  if (tasks->tasks == NULL) {
+    return NULL;
+  }
+  tasks->head = 0;
+  tasks->tail = 0;
+  q->size++;
+  return tasks;
+}
+
+void emscripten_proxy_execute_queue(em_proxying_queue* q) {
+  if (q == NULL) {
+    return;
+  }
+  pthread_mutex_lock(&q->mutex);
+  task_queue* tasks = get_tasks_for_thread(q, pthread_self());
+  if (tasks == NULL || tasks->processing) {
+    // No tasks for this thread or they are already being processed.
+    pthread_mutex_unlock(&q->mutex);
+    return;
+  }
+  // Found the task queue; process the tasks.
+  tasks->processing = 1;
+  while (!task_queue_empty(tasks)) {
+    task t = task_queue_dequeue(tasks);
+    // Unlock while the task is running to allow more work to be queued in
+    // parallel.
+    pthread_mutex_unlock(&q->mutex);
+    t.func(t.arg);
+    pthread_mutex_lock(&q->mutex);
+  }
+  tasks->processing = 0;
+  pthread_mutex_unlock(&q->mutex);
+}
+
+static pthread_t normalize_thread(pthread_t thread) {
+  if (pthread_equal(thread, EM_CALLBACK_THREAD_CONTEXT_MAIN_BROWSER_THREAD)) {
+    return emscripten_main_browser_thread_id();
+  }
+  if (pthread_equal(thread, EM_CALLBACK_THREAD_CONTEXT_CALLING_THREAD)) {
+    return pthread_self();
+  }
+  return thread;
+}
+
+int emscripten_proxy_async(em_proxying_queue* q,
+                           pthread_t target_thread,
+                           void (*func)(void*),
+                           void* arg) {
+  if (q == NULL) {
+    return 0;
+  }
+  target_thread = normalize_thread(target_thread);
+  pthread_mutex_lock(&q->mutex);
+  task_queue* tasks = get_or_add_tasks_for_thread(q, target_thread);
+  if (tasks == NULL) {
+    pthread_mutex_unlock(&q->mutex);
+    return 0;
+  }
+  int empty = task_queue_empty(tasks);
+  if (!task_queue_enqueue(tasks, (task){func, arg})) {
+    pthread_mutex_unlock(&q->mutex);
+    return 0;
+  }
+  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());
+  }
+  return 1;
+}
+
+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;
+  pthread_mutex_t mutex;
+  pthread_cond_t cond;
+};
+
+static void em_proxying_ctx_init(em_proxying_ctx* ctx,
+                                 void (*func)(em_proxying_ctx*, void*),
+                                 void* arg) {
+  ctx->func = func;
+  ctx->arg = arg;
+  ctx->done = 0;
+  pthread_mutex_init(&ctx->mutex, NULL);
+  pthread_cond_init(&ctx->cond, NULL);
+}
+
+static void em_proxying_ctx_deinit(em_proxying_ctx* ctx) {
+  pthread_mutex_destroy(&ctx->mutex);
+  pthread_cond_destroy(&ctx->cond);
+}
+
+void emscripten_proxy_finish(em_proxying_ctx* ctx) {
+  pthread_mutex_lock(&ctx->mutex);
+  ctx->done = 1;
+  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;
+  ctx->func(ctx, ctx->arg);
+}
+
+int emscripten_proxy_sync_with_ctx(em_proxying_queue* q,
+                                   pthread_t target_thread,
+                                   void (*func)(em_proxying_ctx*, void*),
+                                   void* arg) {
+  assert(!pthread_equal(normalize_thread(target_thread), pthread_self()) &&
+         "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)) {
+    return 0;
+  }
+  pthread_mutex_lock(&ctx.mutex);
+  while (!ctx.done) {
+    pthread_cond_wait(&ctx.cond, &ctx.mutex);
+  }
+  pthread_mutex_unlock(&ctx.mutex);
+  em_proxying_ctx_deinit(&ctx);
+  return 1;
+}
+
+// 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;
+  t->func(t->arg);
+  emscripten_proxy_finish(ctx);
+}
+
+int emscripten_proxy_sync(em_proxying_queue* q,
+                          pthread_t target_thread,
+                          void (*func)(void*),
+                          void* arg) {
+  task t = {func, arg};
+  return emscripten_proxy_sync_with_ctx(q, target_thread, call_then_finish, &t);
+}
diff --git a/tools/system_libs.py b/tools/system_libs.py
index 4c5e9f006a357..18a169d29dac8 100644
--- a/tools/system_libs.py
+++ b/tools/system_libs.py
@@ -822,6 +822,7 @@ def get_files(self):
         path='system/lib/pthread',
         filenames=[
           'library_pthread.c',
+          'proxying.c',
           'pthread_create.c',
           'pthread_join.c',
           'pthread_testcancel.c',

From 99f14f55e6339f29eb6af64fa62cd053878861e6 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 08:39:46 -0800
Subject: [PATCH 02/33] Remove normalize_thread

---
 system/lib/pthread/proxying.c | 14 +-------------
 1 file changed, 1 insertion(+), 13 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index d5ae077506174..02852f45d4c40 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -6,7 +6,6 @@
  */
 
 #include <assert.h>
-#include <emscripten/html5.h>
 #include <emscripten/proxying.h>
 #include <emscripten/threading.h>
 #include <pthread.h>
@@ -209,16 +208,6 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) {
   pthread_mutex_unlock(&q->mutex);
 }
 
-static pthread_t normalize_thread(pthread_t thread) {
-  if (pthread_equal(thread, EM_CALLBACK_THREAD_CONTEXT_MAIN_BROWSER_THREAD)) {
-    return emscripten_main_browser_thread_id();
-  }
-  if (pthread_equal(thread, EM_CALLBACK_THREAD_CONTEXT_CALLING_THREAD)) {
-    return pthread_self();
-  }
-  return thread;
-}
-
 int emscripten_proxy_async(em_proxying_queue* q,
                            pthread_t target_thread,
                            void (*func)(void*),
@@ -226,7 +215,6 @@ int emscripten_proxy_async(em_proxying_queue* q,
   if (q == NULL) {
     return 0;
   }
-  target_thread = normalize_thread(target_thread);
   pthread_mutex_lock(&q->mutex);
   task_queue* tasks = get_or_add_tasks_for_thread(q, target_thread);
   if (tasks == NULL) {
@@ -294,7 +282,7 @@ int emscripten_proxy_sync_with_ctx(em_proxying_queue* q,
                                    pthread_t target_thread,
                                    void (*func)(em_proxying_ctx*, void*),
                                    void* arg) {
-  assert(!pthread_equal(normalize_thread(target_thread), pthread_self()) &&
+  assert(!pthread_equal(target_thread, pthread_self()) &&
          "Cannot synchronously wait for work proxied to the current thread");
   em_proxying_ctx ctx;
   em_proxying_ctx_init(&ctx, func, arg);

From 59a7bb31fcb6b040e9839032f32396815a5f2b77 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 08:43:18 -0800
Subject: [PATCH 03/33] Make header internal

---
 system/lib/pthread/proxying.c                         | 3 ++-
 system/{include/emscripten => lib/pthread}/proxying.h | 0
 2 files changed, 2 insertions(+), 1 deletion(-)
 rename system/{include/emscripten => lib/pthread}/proxying.h (100%)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 02852f45d4c40..321c837f3ad7b 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -6,12 +6,13 @@
  */
 
 #include <assert.h>
-#include <emscripten/proxying.h>
 #include <emscripten/threading.h>
 #include <pthread.h>
 #include <stdlib.h>
 #include <string.h>
 
+#include "proxying.h"
+
 #define TASK_QUEUE_INITIAL_CAPACITY 128
 
 // TODO: Update this to take a `em_proxying_queue` argument.
diff --git a/system/include/emscripten/proxying.h b/system/lib/pthread/proxying.h
similarity index 100%
rename from system/include/emscripten/proxying.h
rename to system/lib/pthread/proxying.h

From faa642324ed305ca3fc24d184f0b601713752eb1 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 08:45:45 -0800
Subject: [PATCH 04/33] Assert that q is not null

---
 system/lib/pthread/proxying.c | 12 +++---------
 1 file changed, 3 insertions(+), 9 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 321c837f3ad7b..558e495b64605 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -123,9 +123,7 @@ em_proxying_queue* em_proxying_queue_create(void) {
 }
 
 void em_proxying_queue_destroy(em_proxying_queue* q) {
-  if (q == NULL) {
-    return;
-  }
+  assert(q != NULL);
   assert(q != &system_proxying_queue && "cannot destroy system proxying queue");
   // No need to acquire the lock; no one should be racing with the destruction
   // of the queue.
@@ -185,9 +183,7 @@ static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
 }
 
 void emscripten_proxy_execute_queue(em_proxying_queue* q) {
-  if (q == NULL) {
-    return;
-  }
+  assert(q != NULL);
   pthread_mutex_lock(&q->mutex);
   task_queue* tasks = get_tasks_for_thread(q, pthread_self());
   if (tasks == NULL || tasks->processing) {
@@ -213,9 +209,7 @@ int emscripten_proxy_async(em_proxying_queue* q,
                            pthread_t target_thread,
                            void (*func)(void*),
                            void* arg) {
-  if (q == NULL) {
-    return 0;
-  }
+  assert(q != NULL);
   pthread_mutex_lock(&q->mutex);
   task_queue* tasks = get_or_add_tasks_for_thread(q, target_thread);
   if (tasks == NULL) {

From f241353d834477f436a27409d31cde791e0b80d9 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 08:53:36 -0800
Subject: [PATCH 05/33] Simplify em_proyxing_queue_create

---
 system/lib/pthread/proxying.c | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 558e495b64605..716032a0a1dd4 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -115,10 +115,7 @@ em_proxying_queue* em_proxying_queue_create(void) {
   if (q == NULL) {
     return NULL;
   }
-  pthread_mutex_init(&q->mutex, NULL);
-  q->task_queues = NULL;
-  q->size = 0;
-  q->capacity = 0;
+  *q = (em_proxying_queue){PTHREAD_MUTEX_INITIALIZER, NULL, 0, 0};
   return q;
 }
 

From c3732d5567ecacae345521e404849a4cab2ac425 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 09:03:10 -0800
Subject: [PATCH 06/33] goto failed

---
 system/lib/pthread/proxying.c | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 716032a0a1dd4..848bc2af3eb17 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -210,13 +210,11 @@ int emscripten_proxy_async(em_proxying_queue* q,
   pthread_mutex_lock(&q->mutex);
   task_queue* tasks = get_or_add_tasks_for_thread(q, target_thread);
   if (tasks == NULL) {
-    pthread_mutex_unlock(&q->mutex);
-    return 0;
+    goto failed;
   }
   int empty = task_queue_empty(tasks);
   if (!task_queue_enqueue(tasks, (task){func, arg})) {
-    pthread_mutex_unlock(&q->mutex);
-    return 0;
+    goto failed;
   }
   pthread_mutex_unlock(&q->mutex);
   // If the queue was previously empty, notify the target thread to process it.
@@ -229,6 +227,10 @@ int emscripten_proxy_async(em_proxying_queue* q,
                                     emscripten_main_browser_thread_id());
   }
   return 1;
+
+failed:
+  pthread_mutex_unlock(&q->mutex);
+  return 0;
 }
 
 struct em_proxying_ctx {

From be9db82f17b400739419aaa7669523adde6919ee Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 09:09:04 -0800
Subject: [PATCH 07/33] Nicer struct initialization syntax

---
 system/lib/pthread/proxying.c | 22 ++++++++++++++--------
 1 file changed, 14 insertions(+), 8 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 848bc2af3eb17..754fd016d5b2a 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -103,8 +103,11 @@ struct em_proxying_queue {
   int capacity;
 };
 
-static em_proxying_queue system_proxying_queue = {
-  PTHREAD_MUTEX_INITIALIZER, NULL, 0, 0};
+static em_proxying_queue system_proxying_queue = {.mutex =
+                                                    PTHREAD_MUTEX_INITIALIZER,
+                                                  .task_queues = NULL,
+                                                  .size = 0,
+                                                  .capacity = 0};
 
 em_proxying_queue* emscripten_proxy_get_system_queue(void) {
   return &system_proxying_queue;
@@ -115,7 +118,10 @@ em_proxying_queue* em_proxying_queue_create(void) {
   if (q == NULL) {
     return NULL;
   }
-  *q = (em_proxying_queue){PTHREAD_MUTEX_INITIALIZER, NULL, 0, 0};
+  *q = (em_proxying_queue){.mutex = PTHREAD_MUTEX_INITIALIZER,
+                           .task_queues = NULL,
+                           .size = 0,
+                           .capacity = 0};
   return q;
 }
 
@@ -247,11 +253,11 @@ struct em_proxying_ctx {
 static void em_proxying_ctx_init(em_proxying_ctx* ctx,
                                  void (*func)(em_proxying_ctx*, void*),
                                  void* arg) {
-  ctx->func = func;
-  ctx->arg = arg;
-  ctx->done = 0;
-  pthread_mutex_init(&ctx->mutex, NULL);
-  pthread_cond_init(&ctx->cond, NULL);
+  *ctx = (em_proxying_ctx){.func = func,
+                           .arg = arg,
+                           .done = 0,
+                           .mutex = PTHREAD_MUTEX_INITIALIZER,
+                           .cond = PTHREAD_COND_INITIALIZER};
 }
 
 static void em_proxying_ctx_deinit(em_proxying_ctx* ctx) {

From 5b6294e2ea259223763a164d9f98c76acd293d04 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 09:35:23 -0800
Subject: [PATCH 08/33] Split out task_queue_grow

---
 system/lib/pthread/proxying.c | 68 ++++++++++++++++++-----------------
 1 file changed, 35 insertions(+), 33 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 754fd016d5b2a..bacff65f40b1e 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -47,42 +47,44 @@ static int task_queue_full(task_queue* tasks) {
   return tasks->head == (tasks->tail + 1) % tasks->capacity;
 }
 
+// Returns 1 on success and 0 on failure.
+static int task_queue_grow(task_queue* tasks) {
+  // Allocate a larger task queue.
+  int new_capacity = tasks->capacity * 2;
+  task* new_tasks = malloc(sizeof(task) * new_capacity);
+  if (new_tasks == NULL) {
+    return 0;
+  }
+  // Copy the tasks such that the head of the queue is at the beginning of the
+  // buffer. There are two cases to handle: either the queue wraps around the
+  // end of the old buffer or it does not.
+  int queued_tasks;
+  if (tasks->head <= tasks->tail) {
+    // No wrap. Copy the tasks in one chunk.
+    queued_tasks = tasks->tail - tasks->head;
+    memcpy(new_tasks, &tasks->tasks[tasks->head], sizeof(task) * queued_tasks);
+  } else {
+    // Wrap. Copy `first_queued` tasks up to the end of the old buffer and
+    // `last_queued` tasks at the beginning of the old buffer.
+    int first_queued = tasks->capacity - tasks->head;
+    int last_queued = tasks->tail;
+    queued_tasks = first_queued + last_queued;
+    memcpy(new_tasks, &tasks->tasks[tasks->head], sizeof(task) * first_queued);
+    memcpy(new_tasks + first_queued, tasks->tasks, sizeof(task) * last_queued);
+  }
+  free(tasks->tasks);
+  tasks->tasks = new_tasks;
+  tasks->capacity = new_capacity;
+  tasks->head = 0;
+  tasks->tail = queued_tasks;
+  return 1;
+}
+
 // Returns 1 on success and 0 on failure.
 static int task_queue_enqueue(task_queue* tasks, task t) {
-  if (task_queue_full(tasks)) {
-    // Allocate a larger task queue.
-    int new_capacity = tasks->capacity * 2;
-    task* new_tasks = malloc(sizeof(task) * new_capacity);
-    if (new_tasks == NULL) {
-      return 0;
-    }
-    // Copy the tasks such that the head of the queue is at the beginning of the
-    // buffer. There are two cases to handle: either the queue wraps around the
-    // end of the old buffer or it does not.
-    int queued_tasks;
-    if (tasks->head <= tasks->tail) {
-      // No wrap. Copy the tasks in one chunk.
-      queued_tasks = tasks->tail - tasks->head;
-      memcpy(
-        new_tasks, &tasks->tasks[tasks->head], sizeof(task) * queued_tasks);
-    } else {
-      // Wrap. Copy `first_queued` tasks up to the end of the old buffer and
-      // `last_queued` tasks at the beginning of the old buffer.
-      int first_queued = tasks->capacity - tasks->head;
-      int last_queued = tasks->tail;
-      queued_tasks = first_queued + last_queued;
-      memcpy(
-        new_tasks, &tasks->tasks[tasks->head], sizeof(task) * first_queued);
-      memcpy(
-        new_tasks + first_queued, tasks->tasks, sizeof(task) * last_queued);
-    }
-    free(tasks->tasks);
-    tasks->tasks = new_tasks;
-    tasks->capacity = new_capacity;
-    tasks->head = 0;
-    tasks->tail = queued_tasks;
+  if (task_queue_full(tasks) && !task_queue_grow(tasks)) {
+    return 0;
   }
-  // Enqueue the task.
   tasks->tasks[tasks->tail] = t;
   tasks->tail = (tasks->tail + 1) % tasks->capacity;
   return 1;

From e5ed1ba3a978607112d7fdb61a676a3b293a1a22 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 09:37:04 -0800
Subject: [PATCH 09/33] More "Not thread safe"

---
 system/lib/pthread/proxying.c | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index bacff65f40b1e..49171b72799b2 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -39,15 +39,17 @@ typedef struct task_queue {
   int tail;
 } task_queue;
 
+// Not thread safe.
 static int task_queue_empty(task_queue* tasks) {
   return tasks->head == tasks->tail;
 }
 
+// Not thread safe.
 static int task_queue_full(task_queue* tasks) {
   return tasks->head == (tasks->tail + 1) % tasks->capacity;
 }
 
-// Returns 1 on success and 0 on failure.
+// // Not thread safe. Returns 1 on success and 0 on failure.
 static int task_queue_grow(task_queue* tasks) {
   // Allocate a larger task queue.
   int new_capacity = tasks->capacity * 2;
@@ -80,7 +82,7 @@ static int task_queue_grow(task_queue* tasks) {
   return 1;
 }
 
-// Returns 1 on success and 0 on failure.
+// Not thread safe. Returns 1 on success and 0 on failure.
 static int task_queue_enqueue(task_queue* tasks, task t) {
   if (task_queue_full(tasks) && !task_queue_grow(tasks)) {
     return 0;
@@ -90,6 +92,7 @@ static int task_queue_enqueue(task_queue* tasks, task t) {
   return 1;
 }
 
+// Not thread safe.
 static task task_queue_dequeue(task_queue* tasks) {
   task t = tasks->tasks[tasks->head];
   tasks->head = (tasks->head + 1) % tasks->capacity;

From 8e17f9141e5350bc5cb9811245e781202d96d9fa Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Sat, 11 Dec 2021 09:48:55 -0800
Subject: [PATCH 10/33] Split out task_queue_{de}init

---
 system/lib/pthread/proxying.c | 25 ++++++++++++++++++-------
 1 file changed, 18 insertions(+), 7 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 49171b72799b2..c02d9d73f0125 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -39,6 +39,22 @@ typedef struct task_queue {
   int tail;
 } task_queue;
 
+static int task_queue_init(task_queue* tasks, pthread_t thread) {
+  task* task_buffer = malloc(sizeof(task) * TASK_QUEUE_INITIAL_CAPACITY);
+  if (task_buffer == NULL) {
+    return 0;
+  }
+  *tasks = (task_queue){.thread = thread,
+                        .processing = 0,
+                        .tasks = task_buffer,
+                        .capacity = TASK_QUEUE_INITIAL_CAPACITY,
+                        .head = 0,
+                        .tail = 0};
+  return 1;
+}
+
+static void task_queue_deinit(task_queue* tasks) { free(tasks->tasks); }
+
 // Not thread safe.
 static int task_queue_empty(task_queue* tasks) {
   return tasks->head == tasks->tail;
@@ -137,7 +153,7 @@ void em_proxying_queue_destroy(em_proxying_queue* q) {
   // of the queue.
   pthread_mutex_destroy(&q->mutex);
   for (int i = 0; i < q->size; i++) {
-    free(q->task_queues[i].tasks);
+    task_queue_deinit(&q->task_queues[i]);
   }
   free(q->task_queues);
   free(q);
@@ -178,14 +194,9 @@ static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
   }
   // Initialize the next available task queue.
   tasks = &q->task_queues[q->size];
-  tasks->thread = thread;
-  tasks->processing = 0;
-  tasks->tasks = malloc(sizeof(task) * TASK_QUEUE_INITIAL_CAPACITY);
-  if (tasks->tasks == NULL) {
+  if (!task_queue_init(tasks, thread)) {
     return NULL;
   }
-  tasks->head = 0;
-  tasks->tail = 0;
   q->size++;
   return tasks;
 }

From be3de8fb0164f4ce0ba8f15362707dc3995114a3 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Mon, 13 Dec 2021 21:57:27 -0800
Subject: [PATCH 11/33] Add test (excluding postMessage notification)

---
 system/lib/pthread/proxying.c           |  21 +-
 tests/pthread/test_pthread_proxying.c   | 270 ++++++++++++++++++++++++
 tests/pthread/test_pthread_proxying.out |   9 +
 tests/test_core.py                      |   7 +
 4 files changed, 298 insertions(+), 9 deletions(-)
 create mode 100644 tests/pthread/test_pthread_proxying.c
 create mode 100644 tests/pthread/test_pthread_proxying.out

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index c02d9d73f0125..c0d3e8a4d086a 100644
--- a/system/lib/pthread/proxying.c
+++ b/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:
diff --git a/tests/pthread/test_pthread_proxying.c b/tests/pthread/test_pthread_proxying.c
new file mode 100644
index 0000000000000..7099176ad5652
--- /dev/null
+++ b/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");
+}
diff --git a/tests/pthread/test_pthread_proxying.out b/tests/pthread/test_pthread_proxying.out
new file mode 100644
index 0000000000000..2ea7dcf6664c5
--- /dev/null
+++ b/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
diff --git a/tests/test_core.py b/tests/test_core.py
index 0340b6abc5eef..cede8a2c2c489 100644
--- a/tests/test_core.py
+++ b/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)

From f274a3cdd08a6e0055c6953f88cf9ac679f3858d Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 14 Dec 2021 13:06:44 -0800
Subject: [PATCH 12/33] Test the recursion guard logic

---
 tests/pthread/test_pthread_proxying.c | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/tests/pthread/test_pthread_proxying.c b/tests/pthread/test_pthread_proxying.c
index 7099176ad5652..1c2e84a8ba491 100644
--- a/tests/pthread/test_pthread_proxying.c
+++ b/tests/pthread/test_pthread_proxying.c
@@ -196,12 +196,19 @@ void test_proxy_sync_with_ctx(void) {
 }
 
 typedef struct increment_to_arg {
+  em_proxying_queue* queue;
   int* ip;
   int i;
 } increment_to_arg;
 
 void increment_to(void* arg_p) {
   increment_to_arg* arg = (increment_to_arg*)arg_p;
+
+  // Try executing the queue; since the queue is already being executed, this
+  // shouldn't do anything and *arg->ip should still be one less than arg->i
+  // afterward.
+  emscripten_proxy_execute_queue(arg->queue);
+
   assert(*arg->ip == arg->i - 1);
   *arg->ip = arg->i;
   free(arg);
@@ -219,7 +226,7 @@ void test_queue_growth(void) {
   // 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};
+    *arg = (increment_to_arg){queue, &incremented, i};
     int res = emscripten_proxy_async(queue, pthread_self(), increment_to, arg);
     assert(res == 1);
   }
@@ -232,7 +239,7 @@ void test_queue_growth(void) {
   // 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};
+    *arg = (increment_to_arg){queue, &incremented, i};
     int res = emscripten_proxy_async(queue, pthread_self(), increment_to, arg);
     assert(res == 1);
   }
@@ -244,8 +251,6 @@ void test_queue_growth(void) {
   em_proxying_queue_destroy(queue);
 }
 
-void force_exit(void* arg) { emscripten_force_exit(0); }
-
 int main(int argc, char* argv[]) {
   main_thread = pthread_self();
 

From d147a5a7d282ca3b1b7351ed40e61f2b472cd20c Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Mon, 20 Dec 2021 13:13:22 -0800
Subject: [PATCH 13/33] Fix dangling tasks pointer

---
 system/lib/pthread/proxying.c | 24 ++++++++++++------------
 1 file changed, 12 insertions(+), 12 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index c0d3e8a4d086a..b471690858eaa 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -159,26 +159,23 @@ void em_proxying_queue_destroy(em_proxying_queue* q) {
   free(q);
 }
 
-// Not thread safe.
-static task_queue* get_tasks_for_thread(em_proxying_queue* q,
-                                        pthread_t thread) {
+// Not thread safe. Returns -1 if there are no tasks for the thread.
+static int get_tasks_index_for_thread(em_proxying_queue* q, pthread_t thread) {
   assert(q != NULL);
-  task_queue* tasks = NULL;
   for (int i = 0; i < q->size; i++) {
     if (pthread_equal(q->task_queues[i].thread, thread)) {
-      tasks = &q->task_queues[i];
-      break;
+      return i;
     }
   }
-  return tasks;
+  return -1;
 }
 
 // Not thread safe.
 static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
                                                pthread_t thread) {
-  task_queue* tasks = get_tasks_for_thread(q, thread);
-  if (tasks != NULL) {
-    return tasks;
+  int tasks_index = get_tasks_index_for_thread(q, thread);
+  if (tasks_index != -1) {
+    return &q->task_queues[tasks_index];
   }
   // There were no tasks for the thread; initialize a new task_queue. If there
   // are not enough queues, allocate more.
@@ -193,7 +190,7 @@ static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
     q->capacity = new_capacity;
   }
   // Initialize the next available task queue.
-  tasks = &q->task_queues[q->size];
+  task_queue* tasks = &q->task_queues[q->size];
   if (!task_queue_init(tasks, thread)) {
     return NULL;
   }
@@ -204,7 +201,8 @@ static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
 void emscripten_proxy_execute_queue(em_proxying_queue* q) {
   assert(q != NULL);
   pthread_mutex_lock(&q->mutex);
-  task_queue* tasks = get_tasks_for_thread(q, pthread_self());
+  int tasks_index = get_tasks_index_for_thread(q, pthread_self());
+  task_queue* tasks = tasks_index == -1 ? NULL : &q->task_queues[tasks_index];
   if (tasks == NULL || tasks->processing) {
     // No tasks for this thread or they are already being processed.
     pthread_mutex_unlock(&q->mutex);
@@ -219,6 +217,8 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) {
     pthread_mutex_unlock(&q->mutex);
     t.func(t.arg);
     pthread_mutex_lock(&q->mutex);
+    // The tasks might have been reallocated, so recalculate the pointer.
+    tasks = &q->task_queues[tasks_index];
   }
   tasks->processing = 0;
   pthread_mutex_unlock(&q->mutex);

From 64c1d426dc8e8d7c2b2ce2d46720e6e85a72521f Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Mon, 20 Dec 2021 13:33:59 -0800
Subject: [PATCH 14/33] new postMessage notification

---
 src/library_pthread.js        | 19 +++++++++++++++++++
 src/worker.js                 |  4 ++++
 system/lib/pthread/proxying.c | 26 +++++++++++---------------
 3 files changed, 34 insertions(+), 15 deletions(-)

diff --git a/src/library_pthread.js b/src/library_pthread.js
index fb709b01ad82c..a54512e145a31 100644
--- a/src/library_pthread.js
+++ b/src/library_pthread.js
@@ -1197,6 +1197,25 @@ var LibraryPThread = {
       worker.postMessage({'cmd' : 'processThreadQueue'});
     }
     return 1;
+  },
+
+  _emscripten_notify_proxying_queue: function(targetThreadId, currThreadId, mainThreadId, queue) {
+    if (targetThreadId == currThreadId) {
+      setTimeout(function() { _emscripten_proxy_execute_queue(queue); });
+    } else if (ENVIRONMENT_IS_PTHREAD) {
+      postMessage({'targetThread' : targetThreadId, 'cmd' : 'processProxyingQueue', 'queue' : queue});
+    } else {
+      var pthread = PThread.pthreads[targetThreadId];
+      var worker = pthread && pthread.worker;
+      if (!worker) {
+#if ASSERTIONS
+        err('Cannot send message to thread with ID ' + targetThreadId + ', unknown thread ID!');
+#endif
+        return /*0*/;
+      }
+      worker.postMessage({'cmd' : 'processProxyingQueue', 'queue': queue});
+    }
+    return 1;
   }
 };
 
diff --git a/src/worker.js b/src/worker.js
index 87040c2098d66..ecb7bfbc3b1b3 100644
--- a/src/worker.js
+++ b/src/worker.js
@@ -286,6 +286,10 @@ self.onmessage = function(e) {
       if (Module['_pthread_self']()) { // If this thread is actually running?
         Module['_emscripten_current_thread_process_queued_calls']();
       }
+    } else if (e.data.cmd === 'processProxyingQueue') {
+      if (Module['_pthread_self']()) { // If this thread is actually running?
+        Module['_emscripten_proxy_execute_queue'](e.data.queue);
+      }
     } else {
       err('worker.js received unknown command ' + e.data.cmd);
       err(e.data);
diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index b471690858eaa..da3acbac05aed 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -15,9 +15,10 @@
 
 #define TASK_QUEUE_INITIAL_CAPACITY 128
 
-// TODO: Update this to take a `em_proxying_queue` argument.
-extern int _emscripten_notify_thread_queue(pthread_t target_thread,
-                                           pthread_t main_thread);
+extern int _emscripten_notify_proxying_queue(pthread_t target_thread,
+                                             pthread_t curr_thread,
+                                             pthread_t main_thread,
+                                             em_proxying_queue* queue);
 
 typedef struct task {
   void (*func)(void*);
@@ -239,18 +240,13 @@ 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) {
+    _emscripten_notify_proxying_queue(
+      target_thread, pthread_self(), emscripten_main_browser_thread_id(), q);
+  }
   return 1;
 
 failed:

From 4d5aa0eaf72c11e8b8df58e1bfe4866018e29f22 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Mon, 20 Dec 2021 15:26:58 -0800
Subject: [PATCH 15/33] Test returner thread

---
 tests/pthread/test_pthread_proxying.c   | 40 ++++++++++++++-----------
 tests/pthread/test_pthread_proxying.out |  3 ++
 tests/test_core.py                      |  4 ++-
 3 files changed, 28 insertions(+), 19 deletions(-)

diff --git a/tests/pthread/test_pthread_proxying.c b/tests/pthread/test_pthread_proxying.c
index 1c2e84a8ba491..bff38b585ad23 100644
--- a/tests/pthread/test_pthread_proxying.c
+++ b/tests/pthread/test_pthread_proxying.c
@@ -27,7 +27,7 @@ void* looper_main(void* arg) {
   return NULL;
 }
 
-void* returner_main(void* arg) { return NULL; }
+void* returner_main(void* arg) { emscripten_exit_with_live_runtime(); }
 
 typedef struct widget {
   // `val` will be stored to `out` and the current thread will be stored to
@@ -133,12 +133,12 @@ void test_proxy_async(void) {
   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)); */
+  // 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);
@@ -159,11 +159,11 @@ void test_proxy_sync(void) {
   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)); */
+  // 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);
@@ -184,12 +184,12 @@ void test_proxy_sync_with_ctx(void) {
   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)); */
+  // 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);
@@ -266,10 +266,14 @@ int main(int argc, char* argv[]) {
 
   should_quit = 1;
   pthread_join(looper, NULL);
+
+  pthread_cancel(returner);
   pthread_join(returner, NULL);
+
   em_proxying_queue_destroy(proxy_queue);
 
   test_queue_growth();
 
   printf("done\n");
+  emscripten_force_exit(0);
 }
diff --git a/tests/pthread/test_pthread_proxying.out b/tests/pthread/test_pthread_proxying.out
index 2ea7dcf6664c5..a2be8782ecd6a 100644
--- a/tests/pthread/test_pthread_proxying.out
+++ b/tests/pthread/test_pthread_proxying.out
@@ -1,9 +1,12 @@
 Testing async proxying
 running widget 1 on main
 running widget 2 on looper
+running widget 3 on returner
 Testing sync proxying
 running widget 4 on looper
+running widget 5 on returner
 Testing sync_with_ctx proxying
 running widget 6 on looper
+running widget 7 on returner
 Testing queue growth
 done
diff --git a/tests/test_core.py b/tests/test_core.py
index cede8a2c2c489..7597e53cdc9ea 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -2422,7 +2422,9 @@ def test_pthread_equal(self):
   @node_pthreads
   def test_pthread_proxying(self):
     self.set_setting('EXIT_RUNTIME')
-    self.set_setting('PTHREAD_POOL_SIZE=2')
+    self.set_setting('PROXY_TO_PTHREAD')
+    self.set_setting('PTHREAD_POOL_SIZE=3')
+    self.set_setting('EXPORTED_FUNCTIONS=_emscripten_proxy_execute_queue,_main')
     args = [f'-I{path_from_root("system/lib/pthread")}']
     self.do_run_in_out_file_test('pthread/test_pthread_proxying.c', emcc_args=args)
 

From b421b3c1e35ddfbb4fd876b072c6ab7d2124be92 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Mon, 20 Dec 2021 19:58:25 -0800
Subject: [PATCH 16/33] test tasks_queue growth during processing

---
 tests/pthread/test_pthread_proxying.c   | 69 +++++++++++++++++++++++--
 tests/pthread/test_pthread_proxying.out |  7 ++-
 tests/test_core.py                      |  1 +
 3 files changed, 73 insertions(+), 4 deletions(-)

diff --git a/tests/pthread/test_pthread_proxying.c b/tests/pthread/test_pthread_proxying.c
index bff38b585ad23..32a436f788155 100644
--- a/tests/pthread/test_pthread_proxying.c
+++ b/tests/pthread/test_pthread_proxying.c
@@ -214,8 +214,8 @@ void increment_to(void* arg_p) {
   free(arg);
 }
 
-void test_queue_growth(void) {
-  printf("Testing queue growth\n");
+void test_tasks_queue_growth(void) {
+  printf("Testing tasks queue growth\n");
 
   em_proxying_queue* queue = em_proxying_queue_create();
   assert(proxy_queue != NULL);
@@ -251,6 +251,68 @@ void test_queue_growth(void) {
   em_proxying_queue_destroy(queue);
 }
 
+typedef struct proxying_queue_growth_arg {
+  em_proxying_queue* queue;
+  pthread_t a;
+  pthread_t b;
+  _Atomic int work_count;
+} proxying_queue_growth_arg;
+
+void trivial_work(void* arg) {
+  printf("work\n");
+  (*(_Atomic int*)arg)++;
+}
+
+void grow_proxying_queue(void* arg_p) {
+  // Add task_queues for two new threads, causing a reallocation of the
+  // `em_proxying_queue`'s task_queues array the first time this is called.
+  proxying_queue_growth_arg* arg = (proxying_queue_growth_arg*)arg_p;
+  emscripten_proxy_async(arg->queue, arg->a, trivial_work, &arg->work_count);
+  emscripten_proxy_async(arg->queue, arg->b, trivial_work, &arg->work_count);
+}
+
+void test_proxying_queue_growth(void) {
+  printf("Testing proxying queue growth\n");
+
+  proxying_queue_growth_arg arg;
+  arg.queue = em_proxying_queue_create();
+  assert(arg.queue != NULL);
+
+  pthread_create(&arg.a, NULL, returner_main, NULL);
+  pthread_create(&arg.b, NULL, returner_main, NULL);
+
+  arg.work_count = 0;
+
+  // Queue a task for the current thread, allocating an array of one task_queue.
+  // Then when the task is executed, work is queued on two new threads, bumping
+  // up the array size to 4 and causing it to be reallocated elsewhere. Make
+  // sure we correctly handle this reallocation in the middle of executing the
+  // queue.
+  emscripten_proxy_async(arg.queue, pthread_self(), grow_proxying_queue, &arg);
+  emscripten_proxy_execute_queue(arg.queue);
+
+  while (arg.work_count < 2) {
+    sched_yield();
+  }
+
+  // Do it again to make sure the queue was left in a valid state. Specifically,
+  // if the reallocation is not handled correctly, the recursion guard might not
+  // have been updated correctly, so the work will not be completed.
+  emscripten_proxy_async(arg.queue, pthread_self(), grow_proxying_queue, &arg);
+  emscripten_proxy_execute_queue(arg.queue);
+
+  while (arg.work_count < 4) {
+    sched_yield();
+  }
+
+  // Clean up.
+  pthread_cancel(arg.a);
+  pthread_cancel(arg.b);
+  pthread_join(arg.a, NULL);
+  pthread_join(arg.b, NULL);
+  em_proxying_queue_destroy(arg.queue);
+}
+
 int main(int argc, char* argv[]) {
   main_thread = pthread_self();
 
@@ -272,7 +334,8 @@ int main(int argc, char* argv[]) {
 
   em_proxying_queue_destroy(proxy_queue);
 
-  test_queue_growth();
+  test_tasks_queue_growth();
+  test_proxying_queue_growth();
 
   printf("done\n");
   emscripten_force_exit(0);
diff --git a/tests/pthread/test_pthread_proxying.out b/tests/pthread/test_pthread_proxying.out
index a2be8782ecd6a..ff981341e5f63 100644
--- a/tests/pthread/test_pthread_proxying.out
+++ b/tests/pthread/test_pthread_proxying.out
@@ -8,5 +8,10 @@ running widget 5 on returner
 Testing sync_with_ctx proxying
 running widget 6 on looper
 running widget 7 on returner
-Testing queue growth
+Testing tasks queue growth
+Testing proxying queue growth
+work
+work
+work
+work
 done
diff --git a/tests/test_core.py b/tests/test_core.py
index 7597e53cdc9ea..b442ddf5f48cd 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -2424,6 +2424,7 @@ def test_pthread_proxying(self):
     self.set_setting('EXIT_RUNTIME')
     self.set_setting('PROXY_TO_PTHREAD')
     self.set_setting('PTHREAD_POOL_SIZE=3')
+    self.set_setting('INITIAL_MEMORY=32mb')
     self.set_setting('EXPORTED_FUNCTIONS=_emscripten_proxy_execute_queue,_main')
     args = [f'-I{path_from_root("system/lib/pthread")}']
     self.do_run_in_out_file_test('pthread/test_pthread_proxying.c', emcc_args=args)

From 2a21b1209876d0a3b2e385a4a61eed897746074e Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 21 Dec 2021 09:44:08 -0800
Subject: [PATCH 17/33] Fix test output

---
 tests/test_core.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/test_core.py b/tests/test_core.py
index b442ddf5f48cd..6456a70f69c96 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -2427,7 +2427,8 @@ def test_pthread_proxying(self):
     self.set_setting('INITIAL_MEMORY=32mb')
     self.set_setting('EXPORTED_FUNCTIONS=_emscripten_proxy_execute_queue,_main')
     args = [f'-I{path_from_root("system/lib/pthread")}']
-    self.do_run_in_out_file_test('pthread/test_pthread_proxying.c', emcc_args=args)
+    self.do_run_in_out_file_test('pthread/test_pthread_proxying.c',
+                                 emcc_args=args, interleaved_output=False)
 
   @node_pthreads
   def test_pthread_dispatch_after_exit(self):

From 82ea908694cbc15813a6cf926995f4d50ccadb6c Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 21 Dec 2021 11:05:13 -0800
Subject: [PATCH 18/33] Address comments

---
 system/lib/pthread/proxying.c         | 5 +++--
 system/lib/pthread/proxying.h         | 3 +++
 tests/pthread/test_pthread_proxying.c | 2 +-
 tests/test_core.py                    | 2 --
 4 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index da3acbac05aed..22ef00b01ad5a 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -30,7 +30,8 @@ typedef struct task {
 typedef struct task_queue {
   // The target thread for this task_queue.
   pthread_t thread;
-  // Recursion guard.
+  // Recursion guard. TODO: We disallow recursive processing because that's what
+  // the old proxying API does. Experiment with relaxing this restriction.
   int processing;
   // Ring buffer of tasks of size `capacity`. New tasks are enqueued at
   // `tail` and dequeued at `head`.
@@ -109,7 +110,7 @@ static int task_queue_enqueue(task_queue* tasks, task t) {
   return 1;
 }
 
-// Not thread safe.
+// Not thread safe. Assumes the queue is not empty.
 static task task_queue_dequeue(task_queue* tasks) {
   task t = tasks->tasks[tasks->head];
   tasks->head = (tasks->head + 1) % tasks->capacity;
diff --git a/system/lib/pthread/proxying.h b/system/lib/pthread/proxying.h
index a3e77112f0d4a..bf94e76deed6b 100644
--- a/system/lib/pthread/proxying.h
+++ b/system/lib/pthread/proxying.h
@@ -7,6 +7,7 @@
 
 #pragma once
 
+#include <emscripten/emscripten.h>
 #include <pthread.h>
 
 #ifdef __cplusplus
@@ -29,6 +30,8 @@ void em_proxying_queue_destroy(em_proxying_queue* q);
 em_proxying_queue* emscripten_proxy_get_system_queue();
 
 // Execute all the tasks enqueued for the current thread on the given queue.
+// Exported for use in worker.js.
+EMSCRIPTEN_KEEPALIVE
 void emscripten_proxy_execute_queue(em_proxying_queue* q);
 
 // Opaque handle to a currently-executing proxied task, used to signal the end
diff --git a/tests/pthread/test_pthread_proxying.c b/tests/pthread/test_pthread_proxying.c
index 32a436f788155..43d3009ed9f31 100644
--- a/tests/pthread/test_pthread_proxying.c
+++ b/tests/pthread/test_pthread_proxying.c
@@ -338,5 +338,5 @@ int main(int argc, char* argv[]) {
   test_proxying_queue_growth();
 
   printf("done\n");
-  emscripten_force_exit(0);
+  return 0;
 }
diff --git a/tests/test_core.py b/tests/test_core.py
index 6456a70f69c96..160a5a75ba94f 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -2423,9 +2423,7 @@ def test_pthread_equal(self):
   def test_pthread_proxying(self):
     self.set_setting('EXIT_RUNTIME')
     self.set_setting('PROXY_TO_PTHREAD')
-    self.set_setting('PTHREAD_POOL_SIZE=3')
     self.set_setting('INITIAL_MEMORY=32mb')
-    self.set_setting('EXPORTED_FUNCTIONS=_emscripten_proxy_execute_queue,_main')
     args = [f'-I{path_from_root("system/lib/pthread")}']
     self.do_run_in_out_file_test('pthread/test_pthread_proxying.c',
                                  emcc_args=args, interleaved_output=False)

From 45f7b979dba15e7c9a7c89355cfe891423a19da2 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 21 Dec 2021 13:12:17 -0800
Subject: [PATCH 19/33] Put the emscripten_force_exit back

---
 tests/pthread/test_pthread_proxying.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/pthread/test_pthread_proxying.c b/tests/pthread/test_pthread_proxying.c
index 43d3009ed9f31..32a436f788155 100644
--- a/tests/pthread/test_pthread_proxying.c
+++ b/tests/pthread/test_pthread_proxying.c
@@ -338,5 +338,5 @@ int main(int argc, char* argv[]) {
   test_proxying_queue_growth();
 
   printf("done\n");
-  return 0;
+  emscripten_force_exit(0);
 }

From 3b9cbd8df6e3df0c5e130ad18c582889b7bfae5c Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 4 Jan 2022 09:31:11 -0800
Subject: [PATCH 20/33] Expand comment and move EMSCRIPTEN_KEEPALIVE

---
 system/lib/pthread/proxying.c | 2 ++
 system/lib/pthread/proxying.h | 6 +++---
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 22ef00b01ad5a..1be6845361247 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -200,6 +200,8 @@ static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
   return tasks;
 }
 
+// Exported for use in worker.js.
+EMSCRIPTEN_KEEPALIVE
 void emscripten_proxy_execute_queue(em_proxying_queue* q) {
   assert(q != NULL);
   pthread_mutex_lock(&q->mutex);
diff --git a/system/lib/pthread/proxying.h b/system/lib/pthread/proxying.h
index bf94e76deed6b..de08d78cef0c9 100644
--- a/system/lib/pthread/proxying.h
+++ b/system/lib/pthread/proxying.h
@@ -29,9 +29,9 @@ void em_proxying_queue_destroy(em_proxying_queue* q);
 // nonblocking and safe to run at any time, similar to a native signal handler.
 em_proxying_queue* emscripten_proxy_get_system_queue();
 
-// Execute all the tasks enqueued for the current thread on the given queue.
-// Exported for use in worker.js.
-EMSCRIPTEN_KEEPALIVE
+// Execute all the tasks enqueued for the current thread on the given queue. New
+// tasks that are enqueued concurrently with this execution will be executed as
+// well. This function returns once it observes an empty queue.
 void emscripten_proxy_execute_queue(em_proxying_queue* q);
 
 // Opaque handle to a currently-executing proxied task, used to signal the end

From e9af913d5ee4514a176aa8cf82abd211811097fa Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 4 Jan 2022 09:32:27 -0800
Subject: [PATCH 21/33] rename to task_queue_is_empty

---
 system/lib/pthread/proxying.c | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 1be6845361247..0c54166e14b1b 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -58,7 +58,7 @@ static int task_queue_init(task_queue* tasks, pthread_t thread) {
 static void task_queue_deinit(task_queue* tasks) { free(tasks->tasks); }
 
 // Not thread safe.
-static int task_queue_empty(task_queue* tasks) {
+static int task_queue_is_empty(task_queue* tasks) {
   return tasks->head == tasks->tail;
 }
 
@@ -214,7 +214,7 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) {
   }
   // Found the task queue; process the tasks.
   tasks->processing = 1;
-  while (!task_queue_empty(tasks)) {
+  while (!task_queue_is_empty(tasks)) {
     task t = task_queue_dequeue(tasks);
     // Unlock while the task is running to allow more work to be queued in
     // parallel.
@@ -238,7 +238,7 @@ int emscripten_proxy_async(em_proxying_queue* q,
   if (tasks == NULL) {
     goto failed;
   }
-  int empty = task_queue_empty(tasks);
+  int empty = task_queue_is_empty(tasks);
   if (!task_queue_enqueue(tasks, (task){func, arg})) {
     goto failed;
   }

From fa0d19874e24cd58873e71798a38291b40c6e25d Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 4 Jan 2022 09:54:01 -0800
Subject: [PATCH 22/33] Expand recursion guard comment

---
 system/lib/pthread/proxying.c | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 0c54166e14b1b..6c58a813c4ce3 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -31,7 +31,9 @@ typedef struct task_queue {
   // The target thread for this task_queue.
   pthread_t thread;
   // Recursion guard. TODO: We disallow recursive processing because that's what
-  // the old proxying API does. Experiment with relaxing this restriction.
+  // the old proxying API does, so it is safer to start with the same behavior.
+  // Experiment with relaxing this restriction once the old API uses these
+  // queues as well.
   int processing;
   // Ring buffer of tasks of size `capacity`. New tasks are enqueued at
   // `tail` and dequeued at `head`.

From e4b05ebcc46ce95932974dae8928e59c5fa44cb1 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 4 Jan 2022 14:59:29 -0800
Subject: [PATCH 23/33] Wait for `returner` to start up

---
 tests/pthread/test_pthread_proxying.c | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/tests/pthread/test_pthread_proxying.c b/tests/pthread/test_pthread_proxying.c
index 32a436f788155..e993e17fd5952 100644
--- a/tests/pthread/test_pthread_proxying.c
+++ b/tests/pthread/test_pthread_proxying.c
@@ -17,8 +17,13 @@ pthread_t returner;
 
 // The queue used to send work to both `looper` and `returner`.
 em_proxying_queue* proxy_queue = NULL;
+
+// Whether `looper` should exit.
 _Atomic int should_quit = 0;
 
+// Whether `returner` has spun up.
+_Atomic int has_begun = 0;
+
 void* looper_main(void* arg) {
   while (!should_quit) {
     emscripten_proxy_execute_queue(proxy_queue);
@@ -27,7 +32,10 @@ void* looper_main(void* arg) {
   return NULL;
 }
 
-void* returner_main(void* arg) { emscripten_exit_with_live_runtime(); }
+void* returner_main(void* arg) {
+  has_begun = 1;
+  emscripten_exit_with_live_runtime();
+}
 
 typedef struct widget {
   // `val` will be stored to `out` and the current thread will be stored to
@@ -322,6 +330,11 @@ int main(int argc, char* argv[]) {
   pthread_create(&looper, NULL, looper_main, NULL);
   pthread_create(&returner, NULL, returner_main, NULL);
 
+  // `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();

From 28be33ea3568537dd752aa9156eaa40db97fdb7b Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 21 Dec 2021 12:04:38 -0800
Subject: [PATCH 24/33] Rewrite the old proxying API in terms of the new API

Rewrite the old threading.h proxying API used for internal system call
implementations in terms of the new proxying API introduced in #15737.
---
 src/library_pthread.js               |  27 +---
 src/worker.js                        |   4 -
 system/lib/pthread/library_pthread.c | 213 ++++-----------------------
 tools/system_libs.py                 |   3 +
 4 files changed, 35 insertions(+), 212 deletions(-)

diff --git a/src/library_pthread.js b/src/library_pthread.js
index 9e5a9dfe72432..786d0684112dd 100644
--- a/src/library_pthread.js
+++ b/src/library_pthread.js
@@ -250,9 +250,9 @@ var LibraryPThread = {
           return;
         }
 
-        if (cmd === 'processQueuedMainThreadWork') {
+        if (cmd === 'processProxyingQueue') {
           // TODO: Must post message to main Emscripten thread in PROXY_TO_WORKER mode.
-          _emscripten_main_thread_process_queued_calls();
+          _emscripten_proxy_execute_queue(d['queue']);
         } else if (cmd === 'spawnThread') {
           spawnThread(d);
         } else if (cmd === 'cleanupThread') {
@@ -1149,29 +1149,6 @@ var LibraryPThread = {
     return {{{ makeDynCall('ii', 'ptr') }}}(arg);
   },
 
-  // This function is called internally to notify target thread ID that it has messages it needs to
-  // process in its message queue inside the Wasm heap. As a helper, the caller must also pass the
-  // ID of the main browser thread to this function, to avoid needlessly ping-ponging between JS and
-  // Wasm boundaries.
-  _emscripten_notify_thread_queue: function(targetThreadId, mainThreadId) {
-    if (targetThreadId == mainThreadId) {
-      postMessage({'cmd' : 'processQueuedMainThreadWork'});
-    } else if (ENVIRONMENT_IS_PTHREAD) {
-      postMessage({'targetThread': targetThreadId, 'cmd': 'processThreadQueue'});
-    } else {
-      var pthread = PThread.pthreads[targetThreadId];
-      var worker = pthread && pthread.worker;
-      if (!worker) {
-#if ASSERTIONS
-        err('Cannot send message to thread with ID ' + targetThreadId + ', unknown thread ID!');
-#endif
-        return /*0*/;
-      }
-      worker.postMessage({'cmd' : 'processThreadQueue'});
-    }
-    return 1;
-  },
-
   _emscripten_notify_proxying_queue: function(targetThreadId, currThreadId, mainThreadId, queue) {
     if (targetThreadId == currThreadId) {
       setTimeout(function() { _emscripten_proxy_execute_queue(queue); });
diff --git a/src/worker.js b/src/worker.js
index 44cc198985cc6..fa0b403f82d77 100644
--- a/src/worker.js
+++ b/src/worker.js
@@ -272,10 +272,6 @@ self.onmessage = function(e) {
       }
     } else if (e.data.target === 'setimmediate') {
       // no-op
-    } else if (e.data.cmd === 'processThreadQueue') {
-      if (Module['_pthread_self']()) { // If this thread is actually running?
-        Module['_emscripten_current_thread_process_queued_calls']();
-      }
     } else if (e.data.cmd === 'processProxyingQueue') {
       if (Module['_pthread_self']()) { // If this thread is actually running?
         Module['_emscripten_proxy_execute_queue'](e.data.queue);
diff --git a/system/lib/pthread/library_pthread.c b/system/lib/pthread/library_pthread.c
index 612e697d7690f..0b6707b54bc23 100644
--- a/system/lib/pthread/library_pthread.c
+++ b/system/lib/pthread/library_pthread.c
@@ -35,6 +35,7 @@
 #include <emscripten/stack.h>
 
 #include "threading_internal.h"
+#include "proxying.h"
 
 void __pthread_testcancel();
 
@@ -183,7 +184,6 @@ void emscripten_async_waitable_close(em_queued_call* call) {
 }
 
 extern double emscripten_receive_on_main_thread_js(int functionIndex, int numCallArgs, double* args);
-extern int _emscripten_notify_thread_queue(pthread_t targetThreadId, pthread_t mainThreadId);
 extern int __pthread_create_js(struct pthread *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);
 
 static void _do_call(void* arg) {
@@ -335,60 +335,6 @@ static void _do_call(void* arg) {
   }
 }
 
-#define CALL_QUEUE_SIZE 128
-
-// Shared data synchronized by call_queue_lock.
-typedef struct CallQueueEntry {
-  void (*func)(void*);
-  void* arg;
-} CallQueueEntry;
-
-typedef struct CallQueue {
-  void* target_thread;
-  CallQueueEntry* call_queue;
-  int call_queue_head;
-  int call_queue_tail;
-  struct CallQueue* next;
-} CallQueue;
-
-// Currently global to the queue, but this can be improved to be per-queue specific. (TODO: with
-// lockfree list operations on callQueue_head, or removing the list by moving this data to
-// pthread_t)
-static pthread_mutex_t call_queue_lock = PTHREAD_MUTEX_INITIALIZER;
-static CallQueue* callQueue_head = 0;
-
-// Not thread safe, call while having call_queue_lock obtained.
-static CallQueue* GetQueue(void* target) {
-  assert(target);
-  CallQueue* q = callQueue_head;
-  while (q && q->target_thread != target)
-    q = q->next;
-  return q;
-}
-
-// Not thread safe, call while having call_queue_lock obtained.
-static CallQueue* GetOrAllocateQueue(void* target) {
-  CallQueue* q = GetQueue(target);
-  if (q)
-    return q;
-
-  q = (CallQueue*)malloc(sizeof(CallQueue));
-  q->target_thread = target;
-  q->call_queue = 0;
-  q->call_queue_head = 0;
-  q->call_queue_tail = 0;
-  q->next = 0;
-  if (callQueue_head) {
-    CallQueue* last = callQueue_head;
-    while (last->next)
-      last = last->next;
-    last->next = q;
-  } else {
-    callQueue_head = q;
-  }
-  return q;
-}
-
 EMSCRIPTEN_RESULT emscripten_wait_for_call_v(em_queued_call* call, double timeoutMSecs) {
   int r;
 
@@ -424,85 +370,43 @@ pthread_t emscripten_main_browser_thread_id() {
   return &__main_pthread;
 }
 
-int _emscripten_do_dispatch_to_thread(pthread_t target_thread, em_queued_call* call) {
-  assert(call);
-
-  // #if PTHREADS_DEBUG // TODO: Create a debug version of pthreads library
-  //	EM_ASM_INT({dump('thread ' + _pthread_self() + ' (ENVIRONMENT_IS_WORKER: ' +
-  //ENVIRONMENT_IS_WORKER + '), queueing call of function enum=' + $0 + '/ptr=' + $1 + ' on thread '
-  //+ $2 + '\n' + new Error().stack)}, call->functionEnum, call->functionPtr, target_thread);
-  // #endif
-
-  // Can't be a null pointer here, and can't be
-  // EM_CALLBACK_THREAD_CONTEXT_MAIN_BROWSER_THREAD either.
+static pthread_t normalize_thread(pthread_t target_thread) {
   assert(target_thread);
-  if (target_thread == EM_CALLBACK_THREAD_CONTEXT_MAIN_BROWSER_THREAD)
-    target_thread = emscripten_main_browser_thread_id();
-
-  // If we are the target recipient of this message, we can just call the operation directly.
-  if (target_thread == EM_CALLBACK_THREAD_CONTEXT_CALLING_THREAD ||
-      target_thread == pthread_self()) {
-    _do_call(call);
-    return 1;
+  if (target_thread == EM_CALLBACK_THREAD_CONTEXT_MAIN_BROWSER_THREAD) {
+    return emscripten_main_browser_thread_id();
   }
-
-  // Add the operation to the call queue of the main runtime thread.
-  pthread_mutex_lock(&call_queue_lock);
-  CallQueue* q = GetOrAllocateQueue(target_thread);
-  if (!q->call_queue) {
-    // Shared data synchronized by call_queue_lock.
-    q->call_queue = malloc(sizeof(CallQueueEntry) * CALL_QUEUE_SIZE);
+  if (target_thread == EM_CALLBACK_THREAD_CONTEXT_CALLING_THREAD) {
+    return pthread_self();
   }
+  return target_thread;
+}
 
-  int head = q->call_queue_head;
-  int tail = q->call_queue_tail;
-  int new_tail = (tail + 1) % CALL_QUEUE_SIZE;
-
-  while (new_tail == head) { // Queue is full?
-    pthread_mutex_unlock(&call_queue_lock);
-
-    // If queue of the main browser thread is full, then we wait. (never drop messages for the main
-    // browser thread)
-    if (target_thread == emscripten_main_browser_thread_id()) {
-      emscripten_futex_wait((void*)&q->call_queue_head, head, INFINITY);
-      pthread_mutex_lock(&call_queue_lock);
-      head = q->call_queue_head;
-      tail = q->call_queue_tail;
-      new_tail = (tail + 1) % CALL_QUEUE_SIZE;
-    } else {
-      // For the queues of other threads, just drop the message.
-      // #if DEBUG TODO: a debug build of pthreads library?
-      //			EM_ASM(console.error('Pthread queue overflowed, dropping queued
-      //message to thread. ' + new Error().stack));
-      // #endif
-      em_queued_call_free(call);
-      return 0;
-    }
+// Execute `call` and return 1 only if already on the `target_thread`. Otherwise
+// return 0.
+static int maybe_call_on_current_thread(pthread_t target_thread,
+                                        em_queued_call* call) {
+  if (pthread_equal(target_thread, pthread_self())) {
+    _do_call(call);
+    return 1;
   }
+  return 0;
+}
 
-  q->call_queue[tail].func = _do_call;
-  q->call_queue[tail].arg = call;
-
-  // If the call queue was empty, the main runtime thread is likely idle in the browser event loop,
-  // so send a message to it to ensure that it wakes up to start processing the command we have
-  // posted.
-  if (head == tail) {
-    int success = _emscripten_notify_thread_queue(target_thread, emscripten_main_browser_thread_id());
-    // Failed to dispatch the thread, delete the crafted message.
-    if (!success) {
-      em_queued_call_free(call);
-      pthread_mutex_unlock(&call_queue_lock);
-      return 0;
-    }
+// Execute or proxy `call`. Return 1 if the work was executed or otherwise
+// return 0.
+static int do_dispatch_to_thread(pthread_t target_thread,
+                                 em_queued_call* call) {
+  target_thread = normalize_thread(target_thread);
+  if (maybe_call_on_current_thread(target_thread, call)) {
+    return 1;
   }
-
-  q->call_queue_tail = new_tail;
-  pthread_mutex_unlock(&call_queue_lock);
+  emscripten_proxy_async(
+    emscripten_proxy_get_system_queue(), target_thread, _do_call, call);
   return 0;
 }
 
 void emscripten_async_run_in_main_thread(em_queued_call* call) {
-  _emscripten_do_dispatch_to_thread(emscripten_main_browser_thread_id(), call);
+  do_dispatch_to_thread(emscripten_main_browser_thread_id(), call);
 }
 
 void emscripten_sync_run_in_main_thread(em_queued_call* call) {
@@ -603,50 +507,7 @@ void* emscripten_sync_run_in_main_thread_7(int function, void* arg1,
 }
 
 void emscripten_current_thread_process_queued_calls() {
-  // #if PTHREADS_DEBUG == 2
-  //  EM_ASM(console.error('thread ' + _pthread_self() + ':
-  //emscripten_current_thread_process_queued_calls(), ' + new Error().stack));
-  // #endif
-
-  static thread_local bool thread_is_processing_queued_calls = false;
-
-  // It is possible that when processing a queued call, the control flow leads back to calling this
-  // function in a nested fashion! Therefore this scenario must explicitly be detected, and
-  // processing the queue must be avoided if we are nesting, or otherwise the same queued calls
-  // would be processed again and again.
-  if (thread_is_processing_queued_calls)
-    return;
-  // This must be before pthread_mutex_lock(), since pthread_mutex_lock() can call back to this
-  // function.
-  thread_is_processing_queued_calls = true;
-
-  pthread_mutex_lock(&call_queue_lock);
-  CallQueue* q = GetQueue(pthread_self());
-  if (!q) {
-    pthread_mutex_unlock(&call_queue_lock);
-    thread_is_processing_queued_calls = false;
-    return;
-  }
-
-  int head = q->call_queue_head;
-  int tail = q->call_queue_tail;
-  while (head != tail) {
-    // Assume that the call is heavy, so unlock access to the call queue while it is being
-    // performed.
-    pthread_mutex_unlock(&call_queue_lock);
-    q->call_queue[head].func(q->call_queue[head].arg);
-    pthread_mutex_lock(&call_queue_lock);
-
-    head = (head + 1) % CALL_QUEUE_SIZE;
-    q->call_queue_head = head;
-    tail = q->call_queue_tail;
-  }
-  pthread_mutex_unlock(&call_queue_lock);
-
-  // If the queue was full and we had waiters pending to get to put data to queue, wake them up.
-  emscripten_futex_wake((void*)&q->call_queue_head, INT_MAX);
-
-  thread_is_processing_queued_calls = false;
+  emscripten_proxy_execute_queue(emscripten_proxy_get_system_queue());
 }
 
 // At times when we disallow the main thread to process queued calls, this will
@@ -747,17 +608,6 @@ em_queued_call* emscripten_async_waitable_run_in_main_runtime_thread_(
   return q;
 }
 
-typedef struct DispatchToThreadArgs {
-  pthread_t target_thread;
-  em_queued_call* q;
-} DispatchToThreadArgs;
-
-static void dispatch_to_thread_helper(void* user_data) {
-  DispatchToThreadArgs* args = (DispatchToThreadArgs*)user_data;
-  _emscripten_do_dispatch_to_thread(args->target_thread, args->q);
-  free(user_data);
-}
-
 int emscripten_dispatch_to_thread_args(pthread_t target_thread,
                                        EM_FUNC_SIGNATURE sig,
                                        void* func_ptr,
@@ -775,7 +625,7 @@ int emscripten_dispatch_to_thread_args(pthread_t target_thread,
 
   // `q` will not be used after it is called, so let the call clean it up.
   q->calleeDelete = 1;
-  return _emscripten_do_dispatch_to_thread(target_thread, q);
+  return do_dispatch_to_thread(target_thread, q);
 }
 
 int emscripten_dispatch_to_thread_(pthread_t target_thread,
@@ -806,10 +656,7 @@ int emscripten_dispatch_to_thread_async_args(pthread_t target_thread,
     q->calleeDelete = 1;
 
     // Schedule the call to run later on this thread.
-    DispatchToThreadArgs* args = malloc(sizeof(DispatchToThreadArgs));
-    args->target_thread = target_thread;
-    args->q = q;
-    emscripten_set_timeout(dispatch_to_thread_helper, 0, args);
+    emscripten_set_timeout(_do_call, 0, args);
     return 0;
   }
 
diff --git a/tools/system_libs.py b/tools/system_libs.py
index 96be0e1f2688c..c101dedadbfbe 100644
--- a/tools/system_libs.py
+++ b/tools/system_libs.py
@@ -776,6 +776,9 @@ class libc(DebugLibrary, AsanInstrumentedLibrary, MuslInternalLibrary, MTLibrary
              '-Wno-string-plus-int',
              '-Wno-pointer-sign']
 
+  # Include internal proxying header.
+  cflags += [f'-I{utils.path_from_root("system/lib/pthread")}']
+
   def get_files(self):
     libc_files = []
     musl_srcdir = utils.path_from_root('system/lib/libc/musl/src')

From af9d2ce396fd93f10b141b1bda49462ca7f9a2cb Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 8 Mar 2022 17:38:35 -0800
Subject: [PATCH 25/33] Remove pthread_create from deps_info.py to fix test

---
 tools/deps_info.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tools/deps_info.py b/tools/deps_info.py
index fea8a36ad8d9c..de1bf50916de4 100644
--- a/tools/deps_info.py
+++ b/tools/deps_info.py
@@ -167,7 +167,6 @@
   'localtime': ['malloc'],
   'localtime_r': ['malloc'],
   'mktime': ['malloc'],
-  'pthread_create': ['emscripten_main_thread_process_queued_calls'],
   'recv': ['htons'],
   'recvmsg': ['htons'],
   'accept': ['htons'],

From d03c3da307d2f4c3382c968befb169e869ac4f73 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 8 Mar 2022 17:46:35 -0800
Subject: [PATCH 26/33] Rebaseline size test

---
 ...nimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.funcs | 12 +++++++-----
 ...imal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.jssize |  2 +-
 ...inimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.size |  2 +-
 3 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.funcs b/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.funcs
index 68ca03c8e10c6..32c2fc020b981 100644
--- a/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.funcs
+++ b/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.funcs
@@ -1,6 +1,6 @@
-$GetQueue
 $__emscripten_stdout_seek
 $__errno_location
+$__memcpy
 $__pthread_mutex_lock
 $__pthread_mutex_trylock
 $__pthread_mutex_unlock
@@ -10,12 +10,11 @@ $__pthread_setcancelstate
 $__set_thread_state
 $__stdio_write
 $__timedwait
-$__wake.2
+$__wake
 $__wasi_syscall_ret
 $__wasm_call_ctors
 $__wasm_init_memory
 $_do_call
-$_emscripten_do_dispatch_to_thread
 $_emscripten_thread_crashed
 $_emscripten_thread_exit
 $_emscripten_thread_free_data
@@ -24,7 +23,7 @@ $_main_thread
 $a_cas
 $a_cas_p.1
 $a_dec
-$a_fetch_add.1
+$a_fetch_add
 $a_inc
 $a_swap
 $add
@@ -32,17 +31,19 @@ $dispose_chunk
 $dlfree
 $dlmalloc
 $dlmemalign
-$em_queued_call_free
+$do_dispatch_to_thread
 $em_queued_call_malloc
 $emscripten_async_run_in_main_thread
 $emscripten_current_thread_process_queued_calls
 $emscripten_futex_wait
 $emscripten_futex_wake
 $emscripten_main_thread_process_queued_calls
+$emscripten_proxy_execute_queue
 $emscripten_proxy_main
 $emscripten_run_in_main_runtime_thread_js
 $emscripten_stack_set_limits
 $emscripten_tls_init
+$get_tasks_index_for_thread
 $init_file_lock
 $init_mparams
 $main
@@ -52,3 +53,4 @@ $sbrk
 $stackAlloc
 $stackRestore
 $stackSave
+$task_queue_is_empty
diff --git a/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.jssize b/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.jssize
index 0092864eec294..ec3d19f448375 100644
--- a/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.jssize
+++ b/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.jssize
@@ -1 +1 @@
-32358
+32392
diff --git a/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.size b/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.size
index 5ffc3a6c69f94..e0e5882c3069d 100644
--- a/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.size
+++ b/tests/other/metadce/minimal_main_Oz_USE_PTHREADS_PROXY_TO_PTHREAD.size
@@ -1 +1 @@
-17053
+18512

From 9a89e69e9d069291bd2e2ebe669e29b4ee115a01 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Tue, 8 Mar 2022 18:01:42 -0800
Subject: [PATCH 27/33] Recursion guard on
 emscripten_main_thread_process_queued_calls

---
 system/lib/pthread/library_pthread.c | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/system/lib/pthread/library_pthread.c b/system/lib/pthread/library_pthread.c
index 905a3424b5fc8..c8f0a00fdb6ef 100644
--- a/system/lib/pthread/library_pthread.c
+++ b/system/lib/pthread/library_pthread.c
@@ -505,7 +505,15 @@ void emscripten_main_thread_process_queued_calls() {
   if (!_emscripten_allow_main_runtime_queued_calls)
     return;
 
+  // Recursion guard to avoid infinite recursion when we arrive here from the
+  // pthread calls inside `emscripten_proxy_execute_queue`.
+  static bool processing = 0;
+  if (processing) {
+    return;
+  }
+  processing = 1;
   emscripten_current_thread_process_queued_calls();
+  processing = 0;
 }
 
 int emscripten_sync_run_in_main_runtime_thread_(EM_FUNC_SIGNATURE sig, void* func_ptr, ...) {

From f67a82e26c78ae74b40056d338e43fab7bc66603 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Wed, 9 Mar 2022 11:46:02 -0800
Subject: [PATCH 28/33] [ci skip] improve comment

---
 system/lib/pthread/library_pthread.c | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/system/lib/pthread/library_pthread.c b/system/lib/pthread/library_pthread.c
index c8f0a00fdb6ef..96cedc8783ce9 100644
--- a/system/lib/pthread/library_pthread.c
+++ b/system/lib/pthread/library_pthread.c
@@ -506,7 +506,10 @@ void emscripten_main_thread_process_queued_calls() {
     return;
 
   // Recursion guard to avoid infinite recursion when we arrive here from the
-  // pthread calls inside `emscripten_proxy_execute_queue`.
+  // pthread_lock calls inside `emscripten_proxy_execute_queue`. This isn't
+  // caught by the queue's own recursion guard because the lock has to be
+  // acquired before that recursion guard can be checked. `static` rather than
+  // thread_local because this function is only ever called on the main thread.
   static bool processing = 0;
   if (processing) {
     return;

From 18eebf5086cab8ef3b29c938780e65e44dc5d0a4 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Wed, 9 Mar 2022 12:59:27 -0800
Subject: [PATCH 29/33] Remove duplicate function

---
 src/library_pthread.js | 19 -------------------
 1 file changed, 19 deletions(-)

diff --git a/src/library_pthread.js b/src/library_pthread.js
index a893b4d0fe0b3..9131fbd14b98b 100644
--- a/src/library_pthread.js
+++ b/src/library_pthread.js
@@ -1037,25 +1037,6 @@ var LibraryPThread = {
     return {{{ makeDynCall('ii', 'ptr') }}}(arg);
   },
 
-  _emscripten_notify_proxying_queue: function(targetThreadId, currThreadId, mainThreadId, queue) {
-    if (targetThreadId == currThreadId) {
-      setTimeout(function() { _emscripten_proxy_execute_queue(queue); });
-    } else if (ENVIRONMENT_IS_PTHREAD) {
-      postMessage({'targetThread' : targetThreadId, 'cmd' : 'processProxyingQueue', 'queue' : queue});
-    } else {
-      var pthread = PThread.pthreads[targetThreadId];
-      var worker = pthread && pthread.worker;
-      if (!worker) {
-#if ASSERTIONS
-        err('Cannot send message to thread with ID ' + targetThreadId + ', unknown thread ID!');
-#endif
-        return /*0*/;
-      }
-      worker.postMessage({'cmd' : 'processProxyingQueue', 'queue': queue});
-    }
-    return 1;
-  },
-
   _emscripten_notify_proxying_queue: function(targetThreadId, currThreadId, mainThreadId, queue) {
     if (targetThreadId == currThreadId) {
       setTimeout(function() { _emscripten_proxy_execute_queue(queue); });

From 67aa5ef8143b682e0cbf0da2db4a2c179704b9df Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Wed, 9 Mar 2022 14:09:38 -0800
Subject: [PATCH 30/33] Feedback

---
 system/lib/pthread/library_pthread.c | 2 +-
 tools/system_libs.py                 | 3 ---
 2 files changed, 1 insertion(+), 4 deletions(-)

diff --git a/system/lib/pthread/library_pthread.c b/system/lib/pthread/library_pthread.c
index 96cedc8783ce9..b779cb28899d1 100644
--- a/system/lib/pthread/library_pthread.c
+++ b/system/lib/pthread/library_pthread.c
@@ -653,7 +653,7 @@ int emscripten_dispatch_to_thread_async_args(pthread_t target_thread,
     q->calleeDelete = 1;
 
     // Schedule the call to run later on this thread.
-    emscripten_set_timeout(_do_call, 0, args);
+    emscripten_set_timeout(_do_call, 0, q);
     return 0;
   }
 
diff --git a/tools/system_libs.py b/tools/system_libs.py
index 80e03a912e717..ab50a52eada00 100644
--- a/tools/system_libs.py
+++ b/tools/system_libs.py
@@ -739,9 +739,6 @@ class libc(MuslInternalLibrary,
              '-Wno-string-plus-int',
              '-Wno-pointer-sign']
 
-  # Include internal proxying header.
-  cflags += [f'-I{utils.path_from_root("system/lib/pthread")}']
-
   def __init__(self, **kwargs):
     self.non_lto_files = self.get_non_lto_files()
     super().__init__(**kwargs)

From 72d7457f256d7c095243e3949e6000588cc1a780 Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Thu, 10 Mar 2022 13:56:09 -0800
Subject: [PATCH 31/33] Move system recursion guard to
 emscripten_proxy_execute_queue

---
 system/lib/pthread/library_pthread.c | 11 -----------
 system/lib/pthread/proxying.c        | 22 ++++++++++++++++++++--
 2 files changed, 20 insertions(+), 13 deletions(-)

diff --git a/system/lib/pthread/library_pthread.c b/system/lib/pthread/library_pthread.c
index b779cb28899d1..9246bacfb2c1d 100644
--- a/system/lib/pthread/library_pthread.c
+++ b/system/lib/pthread/library_pthread.c
@@ -505,18 +505,7 @@ void emscripten_main_thread_process_queued_calls() {
   if (!_emscripten_allow_main_runtime_queued_calls)
     return;
 
-  // Recursion guard to avoid infinite recursion when we arrive here from the
-  // pthread_lock calls inside `emscripten_proxy_execute_queue`. This isn't
-  // caught by the queue's own recursion guard because the lock has to be
-  // acquired before that recursion guard can be checked. `static` rather than
-  // thread_local because this function is only ever called on the main thread.
-  static bool processing = 0;
-  if (processing) {
-    return;
-  }
-  processing = 1;
   emscripten_current_thread_process_queued_calls();
-  processing = 0;
 }
 
 int emscripten_sync_run_in_main_runtime_thread_(EM_FUNC_SIGNATURE sig, void* func_ptr, ...) {
diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 6c58a813c4ce3..0aefa60c582ef 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -206,13 +206,26 @@ static task_queue* get_or_add_tasks_for_thread(em_proxying_queue* q,
 EMSCRIPTEN_KEEPALIVE
 void emscripten_proxy_execute_queue(em_proxying_queue* q) {
   assert(q != NULL);
+
+  // Recursion guard to avoid infinite recursion when we arrive here from the
+  // pthread_lock calls below that execute the system queue. The per-task_queue
+  // recursion lock below can't catch these recursions because it can only be
+  // checked after the lock has been acquired.
+  thread_local bool executing_system_queue = false;
+  bool is_system_queue = q == &system_proxying_queue;
+  if (is_system_queue) {
+    if (executing_system_queue) {
+      return;
+    }
+    executing_system_queue = true;
+  }
+
   pthread_mutex_lock(&q->mutex);
   int tasks_index = get_tasks_index_for_thread(q, pthread_self());
   task_queue* tasks = tasks_index == -1 ? NULL : &q->task_queues[tasks_index];
   if (tasks == NULL || tasks->processing) {
     // No tasks for this thread or they are already being processed.
-    pthread_mutex_unlock(&q->mutex);
-    return;
+    goto end;
   }
   // Found the task queue; process the tasks.
   tasks->processing = 1;
@@ -227,7 +240,12 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) {
     tasks = &q->task_queues[tasks_index];
   }
   tasks->processing = 0;
+
+end:
   pthread_mutex_unlock(&q->mutex);
+  if (is_system_queue) {
+    executing_system_queue = false;
+  }
 }
 
 int emscripten_proxy_async(em_proxying_queue* q,

From 47fbd08e1201781dfea6f2ff3b48b437eb672f9d Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Thu, 10 Mar 2022 14:04:43 -0800
Subject: [PATCH 32/33] Fix build after previous commit

---
 system/lib/pthread/proxying.c | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c
index 0aefa60c582ef..27c02fecface7 100644
--- a/system/lib/pthread/proxying.c
+++ b/system/lib/pthread/proxying.c
@@ -208,16 +208,16 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) {
   assert(q != NULL);
 
   // Recursion guard to avoid infinite recursion when we arrive here from the
-  // pthread_lock calls below that execute the system queue. The per-task_queue
-  // recursion lock below can't catch these recursions because it can only be
-  // checked after the lock has been acquired.
-  thread_local bool executing_system_queue = false;
-  bool is_system_queue = q == &system_proxying_queue;
+  // pthread_lock call below that executes the system queue. The per-task_queue
+  // recursion lock can't catch these recursions because it can only be checked
+  // after the lock has been acquired.
+  static _Thread_local int executing_system_queue = 0;
+  int is_system_queue = q == &system_proxying_queue;
   if (is_system_queue) {
     if (executing_system_queue) {
       return;
     }
-    executing_system_queue = true;
+    executing_system_queue = 1;
   }
 
   pthread_mutex_lock(&q->mutex);
@@ -244,7 +244,7 @@ void emscripten_proxy_execute_queue(em_proxying_queue* q) {
 end:
   pthread_mutex_unlock(&q->mutex);
   if (is_system_queue) {
-    executing_system_queue = false;
+    executing_system_queue = 0;
   }
 }
 

From a5a151bc10be6c7860022192482734cb7ee3395e Mon Sep 17 00:00:00 2001
From: Thomas Lively <tlively@google.com>
Date: Thu, 10 Mar 2022 14:23:58 -0800
Subject: [PATCH 33/33] Remove duplicated worker JS

---
 src/worker.js | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/src/worker.js b/src/worker.js
index e0f3949ad058a..a472d0c66b009 100644
--- a/src/worker.js
+++ b/src/worker.js
@@ -289,10 +289,6 @@ self.onmessage = (e) => {
       if (Module['_pthread_self']()) { // If this thread is actually running?
         Module['_emscripten_proxy_execute_queue'](e.data.queue);
       }
-    } else if (e.data.cmd === 'processProxyingQueue') {
-      if (Module['_pthread_self']()) { // If this thread is actually running?
-        Module['_emscripten_proxy_execute_queue'](e.data.queue);
-      }
     } else {
       err('worker.js received unknown command ' + e.data.cmd);
       err(e.data);