From bee58ad902e0cc863d78c1493045c6c37a7b7873 Mon Sep 17 00:00:00 2001 From: Ingvar Stepanyan Date: Tue, 29 Nov 2022 19:01:58 +0000 Subject: [PATCH 1/4] 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). --- ChangeLog.md | 2 ++ emcc.py | 2 ++ src/library_pthread.js | 57 +++++++++++++++++++++--------------------- src/worker.js | 32 ++++++++++++++++-------- test/test_core.py | 25 +++++++++++++----- 5 files changed, 73 insertions(+), 45 deletions(-) diff --git a/ChangeLog.md b/ChangeLog.md index 4c87c21128882..5109ac69e34ea 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -20,6 +20,8 @@ See docs/process.md for more on how version tagging works. 3.1.29 (in development) ----------------------- +- PThreads can now be safely spawned on-demand in Node.js even without a PThread + pool (`PTHREAD_POOL_SIZE`) or proxying (`PROXY_TO_PTHREAD`) options. (#18305) 3.1.28 - 12/08/22 ----------------- diff --git a/emcc.py b/emcc.py index 010d37a4b3f5a..74e7999558cee 100755 --- a/emcc.py +++ b/emcc.py @@ -2314,6 +2314,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!') diff --git a/src/library_pthread.js b/src/library_pthread.js index cb5c68eecdd8b..01bb1c0ff8d38 100644 --- a/src/library_pthread.js +++ b/src/library_pthread.js @@ -277,7 +277,8 @@ 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. @@ -285,10 +286,6 @@ var LibraryPThread = { } #endif 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') { @@ -479,22 +476,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 @@ -626,21 +631,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; }, diff --git a/src/worker.js b/src/worker.js index ba5dd9d566a50..847b56515e8f8 100644 --- a/src/worker.js +++ b/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; diff --git a/test/test_core.py b/test/test_core.py index 313e26ebb2e84..06d735d4547d0 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -9119,8 +9119,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'): @@ -9130,13 +9135,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 From cb847083f1bfa734152cbf25138bfbf048df8e07 Mon Sep 17 00:00:00 2001 From: Ingvar Stepanyan Date: Mon, 12 Dec 2022 18:08:15 +0000 Subject: [PATCH 2/4] Address review comments --- test/test_core.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/test/test_core.py b/test/test_core.py index 06d735d4547d0..d69ca079a106c 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -9120,13 +9120,12 @@ def test_pthread_create(self): @node_pthreads @parameterized({ - 'unpooled': ([],), + '': ([],), 'pooled': (['-sPTHREAD_POOL_SIZE=1'],), - 'proxied': (['-sPROXY_TO_PTHREAD'],), + 'proxied': (['-sPROXY_TO_PTHREAD', '-sEXIT_RUNTIME'],), }) 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'): self.set_setting('INITIAL_MEMORY', '64mb') From 5dd3d0dbbfb850d883417909b61481d613f3edf6 Mon Sep 17 00:00:00 2001 From: Ingvar Stepanyan Date: Mon, 12 Dec 2022 20:40:44 +0000 Subject: [PATCH 3/4] Update src/library_pthread.js --- src/library_pthread.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/library_pthread.js b/src/library_pthread.js index 01bb1c0ff8d38..6b6d40fc8ef63 100644 --- a/src/library_pthread.js +++ b/src/library_pthread.js @@ -477,7 +477,7 @@ var LibraryPThread = { getNewWorker: function() { if (PThread.unusedWorkers.length == 0) { // 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 +#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 From 8506b3ac00ba6f370aac9ed555069403cbf66208 Mon Sep 17 00:00:00 2001 From: Ingvar Stepanyan Date: Mon, 12 Dec 2022 23:01:48 +0000 Subject: [PATCH 4/4] Remove prefix from few more tests --- test/test_core.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/test_core.py b/test/test_core.py index 96ff0aa75d4db..0e0f834752c36 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -9145,7 +9145,7 @@ def test_pthread_c11_threads(self, args): @node_pthreads @parameterized({ - 'unpooled': (0,), + '': (0,), 'pooled': (1,), }) def test_pthread_cxx_threads(self, pthread_pool_size): @@ -9154,7 +9154,7 @@ def test_pthread_cxx_threads(self, pthread_pool_size): @node_pthreads @parameterized({ - 'unpooled': (0,), + '': (0,), 'pooled': (1,), }) def test_pthread_busy_wait(self, pthread_pool_size):