Allow pthreads on Node.js without a pthread pool

Node.js `worker_threads` are different than browser `Worker`s in that
they start spawning synchronously without waiting for an event loop
tick.

If we can also avoid waiting for the "loaded" event from the worker
before sending pthread messages to it, then we can spawn new pthreads
"synchronously" (prepare everything within the same event loop tick),
and block the current thread, making the behaviour on Node.js a lot
closer to native and avoiding the need for a pthread pool even without
Asyncify or extra helper workers.

That's what I did in this implementation. Instead of waiting for the
worker to tell us it's loaded and ready, I'm sending all commands
immediately to the worker. The worker accepts the first "load" message,
starts initializing the runtime, and meanwhile queues up any further
messages such as "run". Once the runtime is ready, it processes the
queue.

I could limit those changes only to Node.js, but it's easier to do both
together, allows to avoid a custom `worker.runPthread` callback, and
should be in theory even a bit faster in browsers too (by avoiding the
"loaded" roundtrip).

Author: RReverser

emcc.py

@@ -2274,6 +2274,8 @@ def phase_linker_setup(options, state, newargs):
   if settings.USE_PTHREADS:
     setup_pthreads(target)
     settings.JS_LIBRARIES.append((0, 'library_pthread.js'))
+    if settings.PROXY_TO_PTHREAD:
+      settings.PTHREAD_POOL_SIZE_STRICT = 0
   else:
     if settings.PROXY_TO_PTHREAD:
       exit_with_error('-sPROXY_TO_PTHREAD requires -sUSE_PTHREADS to work!')

src/library_pthread.js

@@ -277,18 +277,15 @@ var LibraryPThread = {
         } else if (cmd === 'loaded') {
           worker.loaded = true;
 #if ENVIRONMENT_MAY_BE_NODE
-          if (ENVIRONMENT_IS_NODE) {
+          // Check that this worker doesn't have an associated pthread.
+          if (ENVIRONMENT_IS_NODE && !worker.pthread_ptr) {
             // Once worker is loaded & idle, mark it as weakly referenced,
             // so that mere existence of a Worker in the pool does not prevent
             // Node.js from exiting the app.
             worker.unref();
           }
 #endif
           if (onFinishedLoading) onFinishedLoading(worker);
-          // If this Worker is already pending to start running a thread, launch the thread now
-          if (worker.runPthread) {
-            worker.runPthread();
-          }
         } else if (cmd === 'print') {
           out('Thread ' + d['threadId'] + ': ' + d['text']);
         } else if (cmd === 'printErr') {
@@ -455,22 +452,30 @@ var LibraryPThread = {
 
     getNewWorker: function() {
       if (PThread.unusedWorkers.length == 0) {
-#if !PROXY_TO_PTHREAD && PTHREAD_POOL_SIZE_STRICT
+// PTHREAD_POOL_SIZE_STRICT should show a warning and, if set to level `2`, return from the function.
+#if PTHREAD_POOL_SIZE_STRICT && ASSERTIONS || PTHREAD_POOL_SIZE_STRICT == 2
+// However, if we're in Node.js, then we can create new workers on the fly and PTHREAD_POOL_SIZE_STRICT
+// should be ignored altogether.
+#if ENVIRONMENT_MAY_BE_NODE
+        if (!ENVIRONMENT_IS_NODE) {
+#endif
 #if ASSERTIONS
-        err('Tried to spawn a new thread, but the thread pool is exhausted.\n' +
-        'This might result in a deadlock unless some threads eventually exit or the code explicitly breaks out to the event loop.\n' +
-        'If you want to increase the pool size, use setting `-sPTHREAD_POOL_SIZE=...`.'
+            err('Tried to spawn a new thread, but the thread pool is exhausted.\n' +
+            'This might result in a deadlock unless some threads eventually exit or the code explicitly breaks out to the event loop.\n' +
+            'If you want to increase the pool size, use setting `-sPTHREAD_POOL_SIZE=...`.'
 #if PTHREAD_POOL_SIZE_STRICT == 1
-        + '\nIf you want to throw an explicit error instead of the risk of deadlocking in those cases, use setting `-sPTHREAD_POOL_SIZE_STRICT=2`.'
+              + '\nIf you want to throw an explicit error instead of the risk of deadlocking in those cases, use setting `-sPTHREAD_POOL_SIZE_STRICT=2`.'
 #endif
-        );
+            );
 #endif // ASSERTIONS
-
+#if PTHREAD_POOL_SIZE_STRICT == 2
+            return;
 #endif
-#if !PROXY_TO_PTHREAD && PTHREAD_POOL_SIZE_STRICT == 2
-        // Don't return a Worker, which will translate into an EAGAIN error.
-        return;
-#else
+#if ENVIRONMENT_MAY_BE_NODE
+        }
+#endif
+#endif // PTHREAD_POOL_SIZE_STRICT
+#if PTHREAD_POOL_SIZE_STRICT < 2 || ENVIRONMENT_MAY_BE_NODE
         PThread.allocateUnusedWorker();
         PThread.loadWasmModuleToWorker(PThread.unusedWorkers[0]);
 #endif
@@ -602,21 +607,15 @@ var LibraryPThread = {
     msg.moduleCanvasId = threadParams.moduleCanvasId;
     msg.offscreenCanvases = threadParams.offscreenCanvases;
 #endif
-    worker.runPthread = () => {
-      // Ask the worker to start executing its pthread entry point function.
+    // Ask the worker to start executing its pthread entry point function.
 #if ENVIRONMENT_MAY_BE_NODE
-      if (ENVIRONMENT_IS_NODE) {
-        // Mark worker as strongly referenced once we start executing a pthread,
-        // so that Node.js doesn't exit while the pthread is running.
-        worker.ref();
-      }
-#endif
-      worker.postMessage(msg, threadParams.transferList);
-      delete worker.runPthread;
-    };
-    if (worker.loaded) {
-      worker.runPthread();
+    if (ENVIRONMENT_IS_NODE) {
+      // Mark worker as strongly referenced once we start executing a pthread,
+      // so that Node.js doesn't exit while the pthread is running.
+      worker.ref();
     }
+#endif
+    worker.postMessage(msg, threadParams.transferList);
     return 0;
   },
 

src/worker.js

@@ -114,16 +114,7 @@ self.onunhandledrejection = (e) => {
   throw e.reason ?? e;
 };
 
-// Add a callback for when the runtime is initialized.
-self.startWorker = (instance) => {
-#if MODULARIZE
-  Module = instance;
-#endif
-  // Notify the main thread that this thread has loaded.
-  postMessage({ 'cmd': 'loaded' });
-};
-
-self.onmessage = (e) => {
+function handleMessage(e) {
   try {
     if (e.data.cmd === 'load') { // Preload command that is called once per worker to parse and load the Emscripten code.
 #if PTHREADS_DEBUG
@@ -133,6 +124,25 @@ self.onmessage = (e) => {
       var imports = {};
 #endif
 
+    // Until we initialize the runtime, queue up any further incoming messages.
+    let messageQueue = [];
+    self.onmessage = (e) => messageQueue.push(e);
+
+    // And add a callback for when the runtime is initialized.
+    self.startWorker = (instance) => {
+#if MODULARIZE
+      Module = instance;
+#endif
+      // Notify the main thread that this thread has loaded.
+      postMessage({ 'cmd': 'loaded' });
+      // Process any messages that were queued before the thread was ready.
+      for (let msg of messageQueue) {
+        handleMessage(msg);
+      }
+      // Restore the real message handler.
+      self.onmessage = handleMessage;
+    };
+
       // Module and memory were sent from main thread
 #if MINIMAL_RUNTIME
 #if MODULARIZE
@@ -303,3 +313,5 @@ self.onmessage = (e) => {
     throw ex;
   }
 };
+
+self.onmessage = handleMessage;

test/test_core.py

@@ -9116,8 +9116,13 @@ def test_pthread_create(self):
     self.do_run_in_out_file_test('core/pthread/create.cpp')
 
   @node_pthreads
-  def test_pthread_c11_threads(self):
-    self.set_setting('PROXY_TO_PTHREAD')
+  @parameterized({
+    'unpooled': ([],),
+    'pooled': (['-sPTHREAD_POOL_SIZE=1'],),
+    'proxied': (['-sPROXY_TO_PTHREAD'],),
+  })
+  def test_pthread_c11_threads(self, args):
+    self.emcc_args += args
     self.set_setting('EXIT_RUNTIME')
     self.set_setting('PTHREADS_DEBUG')
     if not self.has_changed_setting('INITIAL_MEMORY'):
@@ -9127,13 +9132,21 @@ def test_pthread_c11_threads(self):
     self.do_run_in_out_file_test('pthread/test_pthread_c11_threads.c')
 
   @node_pthreads
-  def test_pthread_cxx_threads(self):
-    self.set_setting('PTHREAD_POOL_SIZE', 1)
+  @parameterized({
+    'unpooled': (0,),
+    'pooled': (1,),
+  })
+  def test_pthread_cxx_threads(self, pthread_pool_size):
+    self.set_setting('PTHREAD_POOL_SIZE', pthread_pool_size)
     self.do_run_in_out_file_test('pthread/test_pthread_cxx_threads.cpp')
 
   @node_pthreads
-  def test_pthread_busy_wait(self):
-    self.set_setting('PTHREAD_POOL_SIZE', 1)
+  @parameterized({
+    'unpooled': (0,),
+    'pooled': (1,),
+  })
+  def test_pthread_busy_wait(self, pthread_pool_size):
+    self.set_setting('PTHREAD_POOL_SIZE', pthread_pool_size)
     self.do_run_in_out_file_test('pthread/test_pthread_busy_wait.cpp')
 
   @node_pthreads