Implement Pthread Manager Worker for synchronous thread creation

Fixes: #18633

Author: sbc100

src/lib/libpthread.js

@@ -40,6 +40,14 @@ const CMD_CLEANUP_THREAD = 6;
 const CMD_MARK_AS_FINISHED = 7;
 const CMD_UNCAUGHT_EXN = 8;
 const CMD_CALL_HANDLER = 9;
+#if PTHREAD_MANAGER
+const CMD_LOAD_MANAGER = 10;
+const CMD_CREATE_WORKER = 11;
+const CMD_TERMINATE_WORKER = 12;
+const CMD_FORWARD_TO_WORKER = 13;
+const CMD_FORWARD_FROM_WORKER = 14;
+const CMD_ERROR_FROM_WORKER = 15;
+#endif
 
 #if WASM_ESM_INTEGRATION
 const pthreadWorkerScript = TARGET_BASENAME + '.pthread.mjs';
@@ -113,13 +121,48 @@ var LibraryPThread = {
     // the reverse mapping, each worker has a `pthread_ptr` when its running a
     // pthread.
     pthreads: {},
+#if PTHREAD_MANAGER
+    proxyWorkers: {},
+    // A ProxyWorker acts as a main-thread representative for a Web Worker
+    // managed by the manager worker. It implements the standard Worker
+    // interface (postMessage, terminate, onmessage, onerror) so that the main
+    // thread's scheduler can interact with it transparently and pool it just
+    // like a standard worker.
+    ProxyWorker: class {
+      constructor(workerID) {
+        this.workerID = workerID;
+        PThread.proxyWorkers[workerID] = this;
+        PThread.managerWorker.postMessage({
+          cmd: {{{ CMD_CREATE_WORKER }}},
+          workerID: workerID
+        });
+      }
+      postMessage(msg, transfer) {
+        PThread.managerWorker.postMessage({
+          cmd: {{{ CMD_FORWARD_TO_WORKER }}},
+          workerID: this.workerID,
+          msg: msg,
+          transferList: transfer
+        }, transfer);
+      }
+      terminate() {
+        PThread.managerWorker.postMessage({
+          cmd: {{{ CMD_TERMINATE_WORKER }}},
+          workerID: this.workerID
+        });
+        delete PThread.proxyWorkers[this.workerID];
+      }
+    },
+#endif
 #if MAIN_MODULE
     outstandingPromises: {},
     // Finished threads are threads that have finished running but we are not yet
     // joined.
     finishedThreads: new Set(),
 #endif
-#if ASSERTIONS
+#if ASSERTIONS || PTHREAD_MANAGER
+    // Used for debugging/assertions, or functionally by PTHREAD_MANAGER to
+    // route multiplexed messages/errors to the correct ProxyWorker instance.
     nextWorkerID: 1,
 #endif
     init() {
@@ -128,19 +171,46 @@ var LibraryPThread = {
       }
     },
     initMainThread() {
+#if ENVIRONMENT_MAY_BE_NODE
+      if (ENVIRONMENT_IS_NODE) return;
+#endif
+#if PTHREAD_MANAGER
+#if ASSERTIONS
+      dbg('PThread: initializing manager worker');
+#endif
+      var managerReadyResolve;
+      PThread.managerWorkerReady = new Promise((resolve) => { managerReadyResolve = resolve; });
+      PThread.managerWorker = PThread.createRealWorker();
+      addOnPreRun(async () => {
+        var managerReady = PThread.initManagerWorker(PThread.managerWorker);
+        addRunDependency('manager-worker');
+        await managerReady;
 #if PTHREAD_POOL_SIZE
-      var pthreadPoolSize = {{{ PTHREAD_POOL_SIZE }}};
-      // Start loading up the Worker pool, if requested.
-      while (pthreadPoolSize--) {
-        PThread.allocateUnusedWorker();
-      }
+        PThread.spawnPool();
+#endif
+        removeRunDependency('manager-worker');
+        managerReadyResolve();
+      });
+#endif // PTHREAD_MANAGER
+#if PTHREAD_POOL_SIZE
+#if ASSERTIONS
+      dbg('PThread: initializing worker pool');
+#endif
+#if !PTHREAD_MANAGER
+      PThread.spawnPool();
+#endif
 #if !MINIMAL_RUNTIME
       // MINIMAL_RUNTIME takes care of calling loadWasmModuleToAllWorkers
       // in postamble_minimal.js
       addOnPreRun(async () => {
-        var pthreadPoolReady = PThread.loadWasmModuleToAllWorkers();
 #if !PTHREAD_POOL_DELAY_LOAD
         addRunDependency('loading-workers');
+#endif
+#if PTHREAD_MANAGER
+        await PThread.managerWorkerReady;
+#endif
+        var pthreadPoolReady = PThread.loadWasmModuleToAllWorkers();
+#if !PTHREAD_POOL_DELAY_LOAD
         await pthreadPoolReady;
         removeRunDependency('loading-workers');
 #endif // PTHREAD_POOL_DELAY_LOAD
@@ -226,6 +296,7 @@ var LibraryPThread = {
       // Note: worker is intentionally not terminated so the pool can
       // dynamically grow.
       PThread.unusedWorkers.push(worker);
+
       // Not a running Worker anymore
       // Detach the worker from the pthread object, and return it to the
       // worker pool as an unused worker.
@@ -265,6 +336,37 @@ var LibraryPThread = {
       // module loaded.
       PThread.tlsInitFunctions.forEach((f) => f());
     },
+#if PTHREAD_MANAGER
+    initManagerWorker: (worker) => new Promise((onFinishedLoading) => {
+      worker.onmessage = (e) => {
+        var d = e.data;
+        var cmd = d.cmd;
+        switch (cmd) {
+          case {{{ CMD_FORWARD_FROM_WORKER }}}:
+            PThread.proxyWorkers[d.workerID]?.onmessage({ data: d.msg });
+            break;
+          case {{{ CMD_ERROR_FROM_WORKER }}}:
+            PThread.proxyWorkers[d.workerID]?.onerror(d.error);
+            break;
+          case {{{ CMD_LOADED }}}:
+            onFinishedLoading(worker);
+            break;
+          default:
+            if (cmd) err(`manager worker sent an unknown command ${cmd}`);
+        }
+      };
+      worker.onerror = (e) => {
+        err(`Manager worker sent an error! ${e.filename}:${e.lineno}: ${e.message}`);
+        throw e;
+      };
+      worker.postMessage({
+        cmd: {{{ CMD_LOAD_MANAGER }}},
+#if ASSERTIONS
+        workerID: worker.workerID,
+#endif
+      });
+    }),
+#endif
     // Loads the WebAssembly module into the given Worker.
     // onFinishedLoading: A callback function that will be called once all of
     //                    the workers have been initialized and are
@@ -442,7 +544,7 @@ var LibraryPThread = {
     }),
 
 #if PTHREAD_POOL_SIZE
-    async loadWasmModuleToAllWorkers() {
+    loadWasmModuleToAllWorkers() {
       // Instantiation is synchronous in pthreads.
       if (
         ENVIRONMENT_IS_PTHREAD
@@ -468,7 +570,7 @@ var LibraryPThread = {
 #endif // PTHREAD_POOL_SIZE
 
     // Creates a new web Worker and places it in the unused worker pool to wait for its use.
-    allocateUnusedWorker() {
+    createRealWorker() {
       var worker;
 #if EXPORT_ES6
       // If we're using module output, use bundler-friendly pattern.
@@ -541,11 +643,33 @@ var LibraryPThread = {
 #endif
       worker = new Worker(pthreadMainJs, {{{ pthreadWorkerOptions }}});
 #endif // EXPORT_ES6
-#if ASSERTIONS
+#if ASSERTIONS || PTHREAD_MANAGER
       worker.workerID = PThread.nextWorkerID++;
 #endif
+      return worker;
+    },
+
+    allocateUnusedWorker() {
+      var worker;
+#if PTHREAD_MANAGER
+      if (PThread.managerWorker) {
+        var workerID = PThread.nextWorkerID++;
+        worker = new PThread.ProxyWorker(workerID);
+      } else
+#endif
+      {
+        worker = PThread.createRealWorker();
+      }
       PThread.unusedWorkers.push(worker);
     },
+#if PTHREAD_POOL_SIZE
+    spawnPool() {
+      var pthreadPoolSize = {{{ PTHREAD_POOL_SIZE }}};
+      while (pthreadPoolSize--) {
+        PThread.allocateUnusedWorker();
+      }
+    },
+#endif
 
     getNewWorker() {
       if (PThread.unusedWorkers.length == 0) {
@@ -697,6 +821,8 @@ var LibraryPThread = {
     assert(threadParams.pthread_ptr, 'spawnThread called with null pthread ptr');
 #endif
 
+
+
     var worker = PThread.getNewWorker();
     if (!worker) {
       // No available workers in the PThread pool.

src/runtime_pthread.js

@@ -42,15 +42,68 @@ if (ENVIRONMENT_IS_PTHREAD) {
   // notified about them.
   self.onunhandledrejection = (e) => { throw e.reason || e; };
 
+#if PTHREAD_MANAGER
+  const targetWorkers = {};
+
+  function handleManagerMessage(e) {
+    var d = e.data;
+    var cmd = d.cmd;
+    if (cmd === {{{ CMD_CREATE_WORKER }}}) {
+      var workerID = d.workerID;
+      var worker = PThread.createRealWorker();
+      worker.workerID = workerID;
+      targetWorkers[workerID] = worker;
+      worker.onmessage = (e) => {
+        var msg = e.data;
+        var transferList = msg && msg.transferList;
+        postMessage({
+          cmd: {{{ CMD_FORWARD_FROM_WORKER }}},
+          workerID: workerID,
+          msg: msg
+        }, transferList);
+      };
+      worker.onerror = (e) => {
+        postMessage({
+          cmd: {{{ CMD_ERROR_FROM_WORKER }}},
+          workerID: workerID,
+          error: {
+            message: e.message,
+            filename: e.filename,
+            lineno: e.lineno,
+            colno: e.colno
+          }
+        });
+      };
+    } else if (cmd === {{{ CMD_FORWARD_TO_WORKER }}}) {
+      var worker = targetWorkers[d.workerID];
+      if (worker) {
+        worker.postMessage(d.msg, d.transferList);
+      }
+    } else if (cmd === {{{ CMD_TERMINATE_WORKER }}}) {
+      var worker = targetWorkers[d.workerID];
+      if (worker) {
+        worker.terminate();
+        delete targetWorkers[d.workerID];
+      }
+    } else {
+#if ASSERTIONS
+      assert(false, "unknown message in pthread manager: " + cmd);
+#endif
+    }
+  }
+#endif
+
   {{{ asyncIf(ASYNCIFY == 2) }}}function handleMessage(e) {
     try {
       var msgData = e.data;
       //dbg('msgData: ' + Object.keys(msgData));
       var cmd = msgData.cmd;
+
       if (cmd == {{{ CMD_LOAD }}}) { // Preload command that is called once per worker to parse and load the Emscripten code.
 #if ASSERTIONS
         workerID = msgData.workerID;
 #endif
+
 #if PTHREADS_DEBUG
         dbg('worker: loading module')
 #endif
@@ -61,8 +114,12 @@ if (ENVIRONMENT_IS_PTHREAD) {
 
         // And add a callback for when the runtime is initialized.
         startWorker = () => {
+#if PTHREADS_DEBUG
+          dbg('worker: startWorker');
+#endif
           // Notify the main thread that this thread has loaded.
           postMessage({ cmd: {{{ CMD_LOADED }}} });
+
           // Process any messages that were queued before the thread was ready.
           for (let msg of messageQueue) {
             handleMessage(msg);
@@ -127,6 +184,14 @@ if (ENVIRONMENT_IS_PTHREAD) {
         run();
 #endif
 #endif // MINIMAL_RUNTIME
+#endif
+#if PTHREAD_MANAGER
+      } else if (cmd == {{{ CMD_LOAD_MANAGER }}}) {
+#if RUNTIME_DEBUG
+        dbg('creating worker manager');
+#endif
+        postMessage({ cmd: {{{ CMD_LOADED }}} });
+        self.onmessage = handleManagerMessage;
 #endif
       } else if (cmd == {{{ CMD_RUN }}}) {
 #if ASSERTIONS
@@ -178,6 +243,7 @@ if (ENVIRONMENT_IS_PTHREAD) {
         if (initializedJS) {
           checkMailbox();
         }
+
       } else if (cmd) {
         // The received message looks like something that should be handled by this message
         // handler, (since there is a cmd field present), but is not one of the
@@ -190,7 +256,7 @@ if (ENVIRONMENT_IS_PTHREAD) {
       err(`worker: onmessage() captured an uncaught exception: ${ex}`);
       if (ex?.stack) err(ex.stack);
 #endif
-      __emscripten_thread_crashed();
+      if (typeof __emscripten_thread_crashed !== 'undefined') __emscripten_thread_crashed();
       throw ex;
     }
   };

src/settings.js

@@ -1617,6 +1617,11 @@ var WEBAUDIO_DEBUG = 0;
 // repeatedly yield back to the JS event loop in order for the thread to
 // actually start.
 // If your application needs to be able to synchronously create new threads,
+// If true, a dedicated worker is used to manage pthread lifecycles.
+// This allows synchronous thread creation even when the main thread is
+// blocked.
+var PTHREAD_MANAGER = false;
+
 // you can pre-create a pthread pool by specifying -sPTHREAD_POOL_SIZE=x,
 // in which case the specified number of Workers will be preloaded into a pool
 // before the application starts, and that many threads can then be available

test/pthread/test_pthread_manager.c

@@ -0,0 +1,29 @@
+#include <pthread.h>
+#include <stdio.h>
+
+#include <emscripten/em_js.h>
+
+EM_JS(int, get_worker_id, (), {
+  return workerID;
+});
+
+
+void *thread_main(void* arg) {
+  printf("thread_main (workerID=%d)\n", get_worker_id());
+  return NULL;
+}
+
+int main() {
+  printf("main (workerID=%d)\n", get_worker_id());
+  pthread_t t;
+
+  pthread_create(&t, NULL, thread_main, NULL);
+  pthread_join(t, NULL);
+
+  // The second pthread should run on the same worker.
+  pthread_create(&t, NULL, thread_main, NULL);
+  pthread_join(t, NULL);
+
+  printf("done\n");
+  return 0;
+}