From 76add286d27601ffa0e8631dfd7692e31960f060 Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Sat, 13 Apr 2024 21:41:41 +0200 Subject: [PATCH 01/13] module: have a single hooks thread for all workers --- lib/internal/main/worker_thread.js | 2 + lib/internal/modules/esm/hooks.js | 107 +++++++++++------- lib/internal/modules/esm/loader.js | 21 +++- lib/internal/modules/esm/worker.js | 46 ++++++-- lib/internal/worker.js | 15 +++ test/es-module/test-esm-loader-threads.mjs | 38 +++++++ test/fixtures/es-module-loaders/hooks-log.mjs | 19 ++++ .../es-module-loaders/worker-log-again.mjs | 3 + .../fixtures/es-module-loaders/worker-log.mjs | 9 ++ .../es-module-loaders/workers-spawned.mjs | 7 ++ 10 files changed, 214 insertions(+), 53 deletions(-) create mode 100644 test/es-module/test-esm-loader-threads.mjs create mode 100644 test/fixtures/es-module-loaders/hooks-log.mjs create mode 100644 test/fixtures/es-module-loaders/worker-log-again.mjs create mode 100644 test/fixtures/es-module-loaders/worker-log.mjs create mode 100644 test/fixtures/es-module-loaders/workers-spawned.mjs diff --git a/lib/internal/main/worker_thread.js b/lib/internal/main/worker_thread.js index aa329b9fe04f15..32bccca9b53a72 100644 --- a/lib/internal/main/worker_thread.js +++ b/lib/internal/main/worker_thread.js @@ -95,6 +95,7 @@ port.on('message', (message) => { filename, hasStdin, publicPort, + hooksPort, workerData, } = message; @@ -109,6 +110,7 @@ port.on('message', (message) => { } require('internal/worker').assignEnvironmentData(environmentData); + require('internal/worker').hooksPort = hooksPort; if (SharedArrayBuffer !== undefined) { // The counter is only passed to the workers created by the main thread, diff --git a/lib/internal/modules/esm/hooks.js b/lib/internal/modules/esm/hooks.js index ba655116a0bb57..2b7909b1ac96c0 100644 --- a/lib/internal/modules/esm/hooks.js +++ b/lib/internal/modules/esm/hooks.js @@ -35,7 +35,7 @@ const { const { exitCodes: { kUnsettledTopLevelAwait } } = internalBinding('errors'); const { URL } = require('internal/url'); const { canParse: URLCanParse } = internalBinding('url'); -const { receiveMessageOnPort } = require('worker_threads'); +const { receiveMessageOnPort, isMainThread } = require('worker_threads'); const { isAnyArrayBuffer, isArrayBufferView, @@ -481,6 +481,7 @@ class HooksProxy { * The InternalWorker instance, which lets us communicate with the loader thread. */ #worker; + #portToHooksThread; /** * The last notification ID received from the worker. This is used to detect @@ -499,26 +500,40 @@ class HooksProxy { #isReady = false; constructor() { - const { InternalWorker } = require('internal/worker'); + const { InternalWorker, hooksPort } = require('internal/worker'); MessageChannel ??= require('internal/worker/io').MessageChannel; const lock = new SharedArrayBuffer(SHARED_MEMORY_BYTE_LENGTH); this.#lock = new Int32Array(lock); - this.#worker = new InternalWorker(loaderWorkerId, { - stderr: false, - stdin: false, - stdout: false, - trackUnmanagedFds: false, - workerData: { - lock, - }, - }); - this.#worker.unref(); // ! Allows the process to eventually exit. - this.#worker.on('exit', process.exit); + if (isMainThread) { + // Main thread is the only one that creates the internal single hooks worker + this.#worker = new InternalWorker(loaderWorkerId, { + stderr: false, + stdin: false, + stdout: false, + trackUnmanagedFds: false, + workerData: { + lock, + }, + }); + this.#worker.unref(); // ! Allows the process to eventually exit. + this.#worker.on('exit', process.exit); + this.#portToHooksThread = this.#worker; + } else { + this.#portToHooksThread = hooksPort; + } } waitForWorker() { + // There is one Hooks instance for each worker thread. But only one of these Hooks instances + // has an InternalWorker. That was the Hooks instance created for the main thread. + // It means for all Hooks instances that are not on the main thread => they are ready because they + // delegate to the single InternalWorker anyway. + if (!isMainThread) { + return; + } + if (!this.#isReady) { const { kIsOnline } = require('internal/worker'); if (!this.#worker[kIsOnline]) { @@ -535,6 +550,23 @@ class HooksProxy { } } + #postMessageToWorker(method, type, transferList, ...args) { + this.waitForWorker(); + MessageChannel ??= require('internal/worker/io').MessageChannel; + const { port1: fromHooksThread, port2: toHooksThread } = new MessageChannel(); + + // Pass work to the worker. + debug(`post ${type} message to worker`, { method, args, transferList }); + const usedTransferList = [toHooksThread]; + if (transferList) { + ArrayPrototypePushApply(usedTransferList, transferList); + } + this.#portToHooksThread.postMessage( + { __proto__: null, method, args, lock: this.#lock, port: toHooksThread }, usedTransferList); + + return fromHooksThread; + } + /** * Invoke a remote method asynchronously. * @param {string} method Method to invoke @@ -543,22 +575,7 @@ class HooksProxy { * @returns {Promise} */ async makeAsyncRequest(method, transferList, ...args) { - this.waitForWorker(); - - MessageChannel ??= require('internal/worker/io').MessageChannel; - const asyncCommChannel = new MessageChannel(); - - // Pass work to the worker. - debug('post async message to worker', { method, args, transferList }); - const finalTransferList = [asyncCommChannel.port2]; - if (transferList) { - ArrayPrototypePushApply(finalTransferList, transferList); - } - this.#worker.postMessage({ - __proto__: null, - method, args, - port: asyncCommChannel.port2, - }, finalTransferList); + const fromHooksThread = this.#postMessageToWorker(method, 'Async', transferList, ...args); if (this.#numberOfPendingAsyncResponses++ === 0) { // On the next lines, the main thread will await a response from the worker thread that might @@ -567,7 +584,11 @@ class HooksProxy { // However we want to keep the process alive until the worker thread responds (or until the // event loop of the worker thread is also empty), so we ref the worker until we get all the // responses back. - this.#worker.ref(); + if (this.#worker) { + this.#worker.ref(); + } else { + this.#portToHooksThread.ref(); + } } let response; @@ -576,18 +597,25 @@ class HooksProxy { await AtomicsWaitAsync(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, this.#workerNotificationLastId).value; this.#workerNotificationLastId = AtomicsLoad(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION); - response = receiveMessageOnPort(asyncCommChannel.port1); + response = receiveMessageOnPort(fromHooksThread); } while (response == null); debug('got async response from worker', { method, args }, this.#lock); if (--this.#numberOfPendingAsyncResponses === 0) { // We got all the responses from the worker, its job is done (until next time). - this.#worker.unref(); + if (this.#worker) { + this.#worker.unref(); + } else { + this.#portToHooksThread.unref(); + } } - const body = this.#unwrapMessage(response); - asyncCommChannel.port1.close(); - return body; + if (response.message.status === 'exit') { + process.exit(response.message.body); + } + + fromHooksThread.close(); + return this.#unwrapMessage(response); } /** @@ -598,11 +626,7 @@ class HooksProxy { * @returns {any} */ makeSyncRequest(method, transferList, ...args) { - this.waitForWorker(); - - // Pass work to the worker. - debug('post sync message to worker', { method, args, transferList }); - this.#worker.postMessage({ __proto__: null, method, args }, transferList); + const fromHooksThread = this.#postMessageToWorker(method, 'Sync', transferList, ...args); let response; do { @@ -611,7 +635,7 @@ class HooksProxy { AtomicsWait(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, this.#workerNotificationLastId); this.#workerNotificationLastId = AtomicsLoad(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION); - response = this.#worker.receiveMessageSync(); + response = receiveMessageOnPort(fromHooksThread); } while (response == null); debug('got sync response from worker', { method, args }); if (response.message.status === 'never-settle') { @@ -619,6 +643,7 @@ class HooksProxy { } else if (response.message.status === 'exit') { process.exit(response.message.body); } + fromHooksThread.close(); return this.#unwrapMessage(response); } diff --git a/lib/internal/modules/esm/loader.js b/lib/internal/modules/esm/loader.js index 8fbcb56aa65287..7fa121ea592cee 100644 --- a/lib/internal/modules/esm/loader.js +++ b/lib/internal/modules/esm/loader.js @@ -41,6 +41,7 @@ const { ModuleWrap, kEvaluating, kEvaluated } = internalBinding('module_wrap'); const { urlToFilename, } = require('internal/modules/helpers'); +const { isMainThread } = require('worker_threads'); let defaultResolve, defaultLoad, defaultLoadSync, importMetaInitializer; /** @@ -607,10 +608,11 @@ class CustomizedModuleLoader { */ constructor() { getHooksProxy(); + _hasCustomizations = true; } /** - * Register some loader specifier. + * Register a loader specifier. * @param {string} originalSpecifier The specified URL path of the loader to * be registered. * @param {string} parentURL The parent URL from where the loader will be @@ -618,10 +620,14 @@ class CustomizedModuleLoader { * @param {any} [data] Arbitrary data to be passed from the custom loader * (user-land) to the worker. * @param {any[]} [transferList] Objects in `data` that are changing ownership - * @returns {{ format: string, url: URL['href'] }} + * @returns {{ format: string, url: URL['href'] } | undefined} */ register(originalSpecifier, parentURL, data, transferList) { - return hooksProxy.makeSyncRequest('register', transferList, originalSpecifier, parentURL, data); + if (isMainThread) { + // Only the main thread has a Hooks instance with worker thread. All other Worker threads + // delegate thier hooks to the HooksThread of the main thread. + return hooksProxy.makeSyncRequest('register', transferList, originalSpecifier, parentURL, data); + } } /** @@ -630,7 +636,7 @@ class CustomizedModuleLoader { * be resolved. * @param {string} [parentURL] The URL path of the module's parent. * @param {ImportAttributes} importAttributes Attributes from the import - * statement or expression. + * statement or exp-ression. * @returns {{ format: string, url: URL['href'] }} */ resolve(originalSpecifier, parentURL, importAttributes) { @@ -719,6 +725,12 @@ function getHooksProxy() { return hooksProxy; } +let _hasCustomizations = false; +function hasCustomizations() { + return _hasCustomizations; +} + + let cascadedLoader; /** @@ -780,6 +792,7 @@ function register(specifier, parentURL = undefined, options) { module.exports = { createModuleLoader, + hasCustomizations, getHooksProxy, getOrInitializeCascadedLoader, register, diff --git a/lib/internal/modules/esm/worker.js b/lib/internal/modules/esm/worker.js index 311d77fb099384..2078ae46f72060 100644 --- a/lib/internal/modules/esm/worker.js +++ b/lib/internal/modules/esm/worker.js @@ -97,6 +97,17 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { // so it can detect the exit event. const { exit } = process; process.exit = function(code) { + if (hooks) { + for (const registeredPort of allThreadRegisteredHandlerPorts) { + registeredPort.postMessage(wrapMessage('exit', code ?? process.exitCode)); + } + for (const { port, lock: operationLock } of unsettledResponsePorts) { + port.postMessage(wrapMessage('exit', code ?? process.exitCode)); + // Wake all threads that have pending operations. Is that needed??? + AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); + AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION); + } + } syncCommPort.postMessage(wrapMessage('exit', code ?? process.exitCode)); AtomicsAdd(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); AtomicsNotify(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION); @@ -145,8 +156,11 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { const unsettledResponsePorts = new SafeSet(); process.on('beforeExit', () => { - for (const port of unsettledResponsePorts) { + for (const { port, lock: operationLock } of unsettledResponsePorts) { port.postMessage(wrapMessage('never-settle')); + // Wake all threads that have pending operations. Is that needed??? + AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); + AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION); } unsettledResponsePorts.clear(); @@ -164,24 +178,39 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { setImmediate(() => {}); }); + const allThreadRegisteredHandlerPorts = []; + function registerHandler(toWorkerThread) { + toWorkerThread.on('message', handleMessage); + allThreadRegisteredHandlerPorts.push(toWorkerThread); + } + + function getMessageHandler(method) { + if (method === '#registerWorkerClient') { + return registerHandler; + } + return hooks[method]; + } + /** * Handles incoming messages from the main thread or other workers. * @param {object} options - The options object. * @param {string} options.method - The name of the hook. * @param {Array} options.args - The arguments to pass to the method. * @param {MessagePort} options.port - The message port to use for communication. + * @param {Int32Array} options.lock - The shared memory where the caller expects to get awaken. */ - async function handleMessage({ method, args, port }) { + async function handleMessage({ method, args, port, lock: msgLock }) { // Each potential exception needs to be caught individually so that the correct error is sent to // the main thread. let hasError = false; let shouldRemoveGlobalErrorHandler = false; - assert(typeof hooks[method] === 'function'); + assert(typeof getMessageHandler(method) === 'function'); if (port == null && !hasUncaughtExceptionCaptureCallback()) { // When receiving sync messages, we want to unlock the main thread when there's an exception. process.on('uncaughtException', errorHandler); shouldRemoveGlobalErrorHandler = true; } + const usedLock = msgLock ?? lock; // We are about to yield the execution with `await ReflectApply` below. In case the code // following the `await` never runs, we remove the message handler so the `beforeExit` event @@ -192,17 +221,18 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { clearImmediate(immediate); immediate = setImmediate(checkForMessages).unref(); - unsettledResponsePorts.add(port ?? syncCommPort); + const unsettledActionData = { port: port ?? syncCommPort, lock: usedLock }; + unsettledResponsePorts.add(unsettledActionData); let response; try { - response = await ReflectApply(hooks[method], hooks, args); + response = await ReflectApply(getMessageHandler(method), hooks, args); } catch (exception) { hasError = true; response = exception; } - unsettledResponsePorts.delete(port ?? syncCommPort); + unsettledResponsePorts.delete(unsettledActionData); // Send the method response (or exception) to the main thread. try { @@ -215,8 +245,8 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { (port ?? syncCommPort).postMessage(wrapMessage('error', exception)); } - AtomicsAdd(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); - AtomicsNotify(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION); + AtomicsAdd(usedLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); + AtomicsNotify(usedLock, WORKER_TO_MAIN_THREAD_NOTIFICATION); if (shouldRemoveGlobalErrorHandler) { process.off('uncaughtException', errorHandler); } diff --git a/lib/internal/worker.js b/lib/internal/worker.js index 9eefaa97021756..59faa640abd411 100644 --- a/lib/internal/worker.js +++ b/lib/internal/worker.js @@ -258,6 +258,12 @@ class Worker extends EventEmitter { ...new SafeArrayIterator(options.transferList)); this[kPublicPort] = port1; + const { port1: toWorkerThread, port2: toHooksThread } = new MessageChannel(); + if (!isInternal) { + // This is not an internal hooks thread => it needs a channel to the hooks thread: + // - send it one side of a channel here + transferList.push(toHooksThread); + } ArrayPrototypeForEach(['message', 'messageerror'], (event) => { this[kPublicPort].on(event, (message) => this.emit(event, message)); }); @@ -272,8 +278,16 @@ class Worker extends EventEmitter { workerData: options.workerData, environmentData, publicPort: port2, + hooksPort: !isInternal ? toHooksThread : undefined, hasStdin: !!options.stdin, }, transferList); + const loaderModule = require('internal/modules/esm/loader'); + const hasCustomizations = loaderModule.hasCustomizations(); + if (!isInternal && hasCustomizations) { + // - send the second side of the channel to the hooks thread + loaderModule.getHooksProxy().makeAsyncRequest( + '#registerWorkerClient', [toWorkerThread], toWorkerThread); + } // Use this to cache the Worker's loopStart value once available. this[kLoopStartTime] = -1; this[kIsOnline] = false; @@ -532,6 +546,7 @@ module.exports = { kIsOnline, isMainThread, SHARE_ENV, + hooksPort: undefined, resourceLimits: !isMainThread ? makeResourceLimits(resourceLimitsRaw) : {}, setEnvironmentData, diff --git a/test/es-module/test-esm-loader-threads.mjs b/test/es-module/test-esm-loader-threads.mjs new file mode 100644 index 00000000000000..f99493e124a32c --- /dev/null +++ b/test/es-module/test-esm-loader-threads.mjs @@ -0,0 +1,38 @@ +import { spawnPromisified } from '../common/index.mjs'; +import * as fixtures from '../common/fixtures.mjs'; +import { strictEqual } from 'node:assert'; +import { execPath } from 'node:process'; +import { describe, it } from 'node:test'; + +describe('off-thread hooks', { concurrency: true }, () => { + it('uses only one hooks thread to support multiple application threads', async () => { + const { code, signal, stdout, stderr } = await spawnPromisified(execPath, [ + '--no-warnings', + '--import', + `data:text/javascript,${encodeURIComponent(` + import { register } from 'node:module'; + register(${JSON.stringify(fixtures.fileURL('es-module-loaders/hooks-log.mjs'))}); + `)}`, + fixtures.path('es-module-loaders/workers-spawned.mjs'), + ]); + + strictEqual(stderr, ''); + strictEqual(stdout.split('\n').filter((line) => line.startsWith('initialize')).length, 1); + strictEqual(stdout.split('\n').filter((line) => line === 'foo').length, 2); + strictEqual(stdout.split('\n').filter((line) => line === 'bar').length, 4); + // Calls to resolve/load: + // 1x main script: test/fixtures/es-module-loaders/workers-spawned.mjs + // 3x worker_threads + // => 1x test/fixtures/es-module-loaders/worker-log.mjs + // 2x test/fixtures/es-module-loaders/worker-log-again.mjs => once per worker-log.mjs Worker instance + // 2x test/fixtures/es-module-loaders/worker-log.mjs => once per worker-log.mjs Worker instance + // 4x test/fixtures/es-module-loaders/worker-log-again.mjs => 2x for each worker-log + // 6x module-named-exports.mjs => 2x worker-log.mjs + 4x worker-log-again.mjs + // =========================== + // 16 calls to resolve + 16 calls to load hook for the registered custom loader + strictEqual(stdout.split('\n').filter((line) => line.startsWith('hooked resolve')).length, 16); + strictEqual(stdout.split('\n').filter((line) => line.startsWith('hooked load')).length, 16); + strictEqual(code, 0); + strictEqual(signal, null); + }); +}); diff --git a/test/fixtures/es-module-loaders/hooks-log.mjs b/test/fixtures/es-module-loaders/hooks-log.mjs new file mode 100644 index 00000000000000..2d2512281e8bd5 --- /dev/null +++ b/test/fixtures/es-module-loaders/hooks-log.mjs @@ -0,0 +1,19 @@ +import { writeFileSync } from 'node:fs'; + +let initializeCount = 0; +let resolveCount = 0; +let loadCount = 0; + +export function initialize() { + writeFileSync(1, `initialize ${++initializeCount}\n`); +} + +export function resolve(specifier, context, next) { + writeFileSync(1, `hooked resolve ${++resolveCount} ${specifier}\n`); + return next(specifier, context); +} + +export function load(url, context, next) { + writeFileSync(1, `hooked load ${++loadCount} ${url}\n`); + return next(url, context); +} diff --git a/test/fixtures/es-module-loaders/worker-log-again.mjs b/test/fixtures/es-module-loaders/worker-log-again.mjs new file mode 100644 index 00000000000000..2969edc8dac382 --- /dev/null +++ b/test/fixtures/es-module-loaders/worker-log-again.mjs @@ -0,0 +1,3 @@ +import { bar } from './module-named-exports.mjs'; + +console.log(bar); diff --git a/test/fixtures/es-module-loaders/worker-log.mjs b/test/fixtures/es-module-loaders/worker-log.mjs new file mode 100644 index 00000000000000..13290c37d07104 --- /dev/null +++ b/test/fixtures/es-module-loaders/worker-log.mjs @@ -0,0 +1,9 @@ +import { Worker } from 'worker_threads'; +import { foo } from './module-named-exports.mjs'; + +const workerURL = new URL('./worker-log-again.mjs', import.meta.url); +console.log(foo); + +// Spawn two workers +new Worker(workerURL); +new Worker(workerURL); diff --git a/test/fixtures/es-module-loaders/workers-spawned.mjs b/test/fixtures/es-module-loaders/workers-spawned.mjs new file mode 100644 index 00000000000000..439847656fe13e --- /dev/null +++ b/test/fixtures/es-module-loaders/workers-spawned.mjs @@ -0,0 +1,7 @@ +import { Worker } from 'worker_threads'; + +const workerURL = new URL('./worker-log.mjs', import.meta.url); + +// Spawn two workers +new Worker(workerURL); +new Worker(workerURL); From 113b41c3086e8ceb8ef2d9d6b895bc92425df60d Mon Sep 17 00:00:00 2001 From: Gabriel Bota <94833492+dygabo@users.noreply.github.com> Date: Sat, 27 Apr 2024 06:42:45 +0200 Subject: [PATCH 02/13] Update lib/internal/modules/esm/loader.js Co-authored-by: Geoffrey Booth --- lib/internal/modules/esm/loader.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/internal/modules/esm/loader.js b/lib/internal/modules/esm/loader.js index 7fa121ea592cee..c3a4fe9893bb81 100644 --- a/lib/internal/modules/esm/loader.js +++ b/lib/internal/modules/esm/loader.js @@ -625,7 +625,7 @@ class CustomizedModuleLoader { register(originalSpecifier, parentURL, data, transferList) { if (isMainThread) { // Only the main thread has a Hooks instance with worker thread. All other Worker threads - // delegate thier hooks to the HooksThread of the main thread. + // delegate their hooks to the HooksThread of the main thread. return hooksProxy.makeSyncRequest('register', transferList, originalSpecifier, parentURL, data); } } From e19528f022ebbdb794502455f73e0913a141a2e1 Mon Sep 17 00:00:00 2001 From: Gabriel Bota <94833492+dygabo@users.noreply.github.com> Date: Sat, 27 Apr 2024 22:38:04 +0200 Subject: [PATCH 03/13] Apply suggestions from code review Co-authored-by: Jacob Smith <3012099+JakobJingleheimer@users.noreply.github.com> Co-authored-by: Antoine du Hamel --- lib/internal/modules/esm/hooks.js | 22 ++++++++++++++++++++-- lib/internal/modules/esm/loader.js | 2 +- lib/internal/modules/esm/worker.js | 7 +++++-- lib/internal/worker.js | 9 +++++++-- 4 files changed, 33 insertions(+), 7 deletions(-) diff --git a/lib/internal/modules/esm/hooks.js b/lib/internal/modules/esm/hooks.js index 2b7909b1ac96c0..6053f237c21e3f 100644 --- a/lib/internal/modules/esm/hooks.js +++ b/lib/internal/modules/esm/hooks.js @@ -481,6 +481,7 @@ class HooksProxy { * The InternalWorker instance, which lets us communicate with the loader thread. */ #worker; + #portToHooksThread; /** @@ -552,8 +553,13 @@ class HooksProxy { #postMessageToWorker(method, type, transferList, ...args) { this.waitForWorker(); + MessageChannel ??= require('internal/worker/io').MessageChannel; - const { port1: fromHooksThread, port2: toHooksThread } = new MessageChannel(); + + const { + port1: fromHooksThread, + port2: toHooksThread, + } = new MessageChannel(); // Pass work to the worker. debug(`post ${type} message to worker`, { method, args, transferList }); @@ -561,8 +567,17 @@ class HooksProxy { if (transferList) { ArrayPrototypePushApply(usedTransferList, transferList); } + this.#portToHooksThread.postMessage( - { __proto__: null, method, args, lock: this.#lock, port: toHooksThread }, usedTransferList); + { + __proto__: null, + args, + lock: this.#lock, + method, + port: toHooksThread, + }, + usedTransferList, + ); return fromHooksThread; } @@ -615,6 +630,7 @@ class HooksProxy { } fromHooksThread.close(); + return this.#unwrapMessage(response); } @@ -643,7 +659,9 @@ class HooksProxy { } else if (response.message.status === 'exit') { process.exit(response.message.body); } + fromHooksThread.close(); + return this.#unwrapMessage(response); } diff --git a/lib/internal/modules/esm/loader.js b/lib/internal/modules/esm/loader.js index c3a4fe9893bb81..5eb7cd17d02fad 100644 --- a/lib/internal/modules/esm/loader.js +++ b/lib/internal/modules/esm/loader.js @@ -636,7 +636,7 @@ class CustomizedModuleLoader { * be resolved. * @param {string} [parentURL] The URL path of the module's parent. * @param {ImportAttributes} importAttributes Attributes from the import - * statement or exp-ression. + * statement or expression. * @returns {{ format: string, url: URL['href'] }} */ resolve(originalSpecifier, parentURL, importAttributes) { diff --git a/lib/internal/modules/esm/worker.js b/lib/internal/modules/esm/worker.js index 2078ae46f72060..9bff6a73839928 100644 --- a/lib/internal/modules/esm/worker.js +++ b/lib/internal/modules/esm/worker.js @@ -98,9 +98,11 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { const { exit } = process; process.exit = function(code) { if (hooks) { - for (const registeredPort of allThreadRegisteredHandlerPorts) { + for (let i = 0; i < allThreadRegisteredHandlerPorts.length; i++) { + const registeredPort = allThreadRegisteredHandlerPorts[i]; registeredPort.postMessage(wrapMessage('exit', code ?? process.exitCode)); } + for (const { port, lock: operationLock } of unsettledResponsePorts) { port.postMessage(wrapMessage('exit', code ?? process.exitCode)); // Wake all threads that have pending operations. Is that needed??? @@ -181,7 +183,7 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { const allThreadRegisteredHandlerPorts = []; function registerHandler(toWorkerThread) { toWorkerThread.on('message', handleMessage); - allThreadRegisteredHandlerPorts.push(toWorkerThread); + ArrayPrototypePush(allThreadRegisteredHandlerPorts, toWorkerThread); } function getMessageHandler(method) { @@ -222,6 +224,7 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { immediate = setImmediate(checkForMessages).unref(); const unsettledActionData = { port: port ?? syncCommPort, lock: usedLock }; + unsettledResponsePorts.add(unsettledActionData); let response; diff --git a/lib/internal/worker.js b/lib/internal/worker.js index 59faa640abd411..cb1ba66e97469a 100644 --- a/lib/internal/worker.js +++ b/lib/internal/worker.js @@ -258,11 +258,14 @@ class Worker extends EventEmitter { ...new SafeArrayIterator(options.transferList)); this[kPublicPort] = port1; - const { port1: toWorkerThread, port2: toHooksThread } = new MessageChannel(); + const { + port1: toWorkerThread, + port2: toHooksThread, + } = new MessageChannel(); if (!isInternal) { // This is not an internal hooks thread => it needs a channel to the hooks thread: // - send it one side of a channel here - transferList.push(toHooksThread); + ArrayPrototypePush(transferList, toHooksThread); } ArrayPrototypeForEach(['message', 'messageerror'], (event) => { this[kPublicPort].on(event, (message) => this.emit(event, message)); @@ -281,8 +284,10 @@ class Worker extends EventEmitter { hooksPort: !isInternal ? toHooksThread : undefined, hasStdin: !!options.stdin, }, transferList); + const loaderModule = require('internal/modules/esm/loader'); const hasCustomizations = loaderModule.hasCustomizations(); + if (!isInternal && hasCustomizations) { // - send the second side of the channel to the hooks thread loaderModule.getHooksProxy().makeAsyncRequest( From 101dd4f1c3fae33b2efd13860e818c05d7db504c Mon Sep 17 00:00:00 2001 From: Gabriel Bota <94833492+dygabo@users.noreply.github.com> Date: Sat, 27 Apr 2024 22:43:04 +0200 Subject: [PATCH 04/13] Apply suggestions from code review Co-authored-by: Antoine du Hamel --- lib/internal/modules/esm/hooks.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/internal/modules/esm/hooks.js b/lib/internal/modules/esm/hooks.js index 6053f237c21e3f..ee3d15c93e76f9 100644 --- a/lib/internal/modules/esm/hooks.js +++ b/lib/internal/modules/esm/hooks.js @@ -551,7 +551,7 @@ class HooksProxy { } } - #postMessageToWorker(method, type, transferList, ...args) { + #postMessageToWorker(method, type, transferList, args) { this.waitForWorker(); MessageChannel ??= require('internal/worker/io').MessageChannel; @@ -590,7 +590,7 @@ class HooksProxy { * @returns {Promise} */ async makeAsyncRequest(method, transferList, ...args) { - const fromHooksThread = this.#postMessageToWorker(method, 'Async', transferList, ...args); + const fromHooksThread = this.#postMessageToWorker(method, 'Async', transferList, args); if (this.#numberOfPendingAsyncResponses++ === 0) { // On the next lines, the main thread will await a response from the worker thread that might From 7cc640083e8810384ea71df86ff04ee323606733 Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Sat, 27 Apr 2024 23:16:59 +0200 Subject: [PATCH 05/13] code review suggestions addressed --- lib/internal/modules/esm/hooks.js | 2 +- lib/internal/modules/esm/worker.js | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/lib/internal/modules/esm/hooks.js b/lib/internal/modules/esm/hooks.js index ee3d15c93e76f9..82e94d583c8688 100644 --- a/lib/internal/modules/esm/hooks.js +++ b/lib/internal/modules/esm/hooks.js @@ -642,7 +642,7 @@ class HooksProxy { * @returns {any} */ makeSyncRequest(method, transferList, ...args) { - const fromHooksThread = this.#postMessageToWorker(method, 'Sync', transferList, ...args); + const fromHooksThread = this.#postMessageToWorker(method, 'Sync', transferList, args); let response; do { diff --git a/lib/internal/modules/esm/worker.js b/lib/internal/modules/esm/worker.js index 9bff6a73839928..11ae730403d1bb 100644 --- a/lib/internal/modules/esm/worker.js +++ b/lib/internal/modules/esm/worker.js @@ -1,6 +1,7 @@ 'use strict'; const { + ArrayPrototypePush, AtomicsAdd, AtomicsNotify, DataViewPrototypeGetBuffer, @@ -181,6 +182,12 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { }); const allThreadRegisteredHandlerPorts = []; + /** + * @callback registerHandler + * @param {MessagePort} toWorkerThread - Upon Worker creation a message channel between the new Worker + * and the Hooks thread is bein initialized. This is the message part that the Hooks thread will use post + * messages to the worker. The other MessagePort is passed to the new Worker itself via LOAD_SCRIPT message + */ function registerHandler(toWorkerThread) { toWorkerThread.on('message', handleMessage); ArrayPrototypePush(allThreadRegisteredHandlerPorts, toWorkerThread); From a2abef08ccb871d6636b4087159950a9d891ce89 Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Mon, 29 Apr 2024 09:53:54 +0200 Subject: [PATCH 06/13] cache response message on exit --- lib/internal/modules/esm/worker.js | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/lib/internal/modules/esm/worker.js b/lib/internal/modules/esm/worker.js index 11ae730403d1bb..e0d1317c0f1c10 100644 --- a/lib/internal/modules/esm/worker.js +++ b/lib/internal/modules/esm/worker.js @@ -98,20 +98,21 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { // so it can detect the exit event. const { exit } = process; process.exit = function(code) { + const exitMsg = wrapMessage('exit', code ?? process.exitCode); if (hooks) { for (let i = 0; i < allThreadRegisteredHandlerPorts.length; i++) { const registeredPort = allThreadRegisteredHandlerPorts[i]; - registeredPort.postMessage(wrapMessage('exit', code ?? process.exitCode)); + registeredPort.postMessage(exitMsg); } for (const { port, lock: operationLock } of unsettledResponsePorts) { - port.postMessage(wrapMessage('exit', code ?? process.exitCode)); + port.postMessage(exitMsg); // Wake all threads that have pending operations. Is that needed??? AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION); } } - syncCommPort.postMessage(wrapMessage('exit', code ?? process.exitCode)); + syncCommPort.postMessage(exitMsg); AtomicsAdd(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); AtomicsNotify(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION); return ReflectApply(exit, this, arguments); @@ -185,7 +186,7 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { /** * @callback registerHandler * @param {MessagePort} toWorkerThread - Upon Worker creation a message channel between the new Worker - * and the Hooks thread is bein initialized. This is the message part that the Hooks thread will use post + * and the Hooks thread is being initialized. This is the MessagePort that the Hooks thread will use post * messages to the worker. The other MessagePort is passed to the new Worker itself via LOAD_SCRIPT message */ function registerHandler(toWorkerThread) { From 0a4c11fbff1963d52506f442f8e5d8861b3e3ce8 Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Mon, 29 Apr 2024 13:16:13 +0200 Subject: [PATCH 07/13] add test for process.exit for import running for worker thread --- lib/internal/modules/esm/hooks.js | 2 -- lib/internal/worker.js | 2 +- test/es-module/test-esm-loader-threads.mjs | 36 +++++++++++++++++++ .../es-module-loaders/hooks-exit-worker.mjs | 21 +++++++++++ .../es-module-loaders/worker-fail-on-load.mjs | 1 + .../worker-fail-on-resolve.mjs | 1 + .../worker-log-fail-worker-load.mjs | 12 +++++++ .../worker-log-fail-worker-resolve.mjs | 12 +++++++ 8 files changed, 84 insertions(+), 3 deletions(-) create mode 100644 test/fixtures/es-module-loaders/hooks-exit-worker.mjs create mode 100644 test/fixtures/es-module-loaders/worker-fail-on-load.mjs create mode 100644 test/fixtures/es-module-loaders/worker-fail-on-resolve.mjs create mode 100644 test/fixtures/es-module-loaders/worker-log-fail-worker-load.mjs create mode 100644 test/fixtures/es-module-loaders/worker-log-fail-worker-resolve.mjs diff --git a/lib/internal/modules/esm/hooks.js b/lib/internal/modules/esm/hooks.js index 82e94d583c8688..f5833ad61cdb75 100644 --- a/lib/internal/modules/esm/hooks.js +++ b/lib/internal/modules/esm/hooks.js @@ -502,8 +502,6 @@ class HooksProxy { constructor() { const { InternalWorker, hooksPort } = require('internal/worker'); - MessageChannel ??= require('internal/worker/io').MessageChannel; - const lock = new SharedArrayBuffer(SHARED_MEMORY_BYTE_LENGTH); this.#lock = new Int32Array(lock); diff --git a/lib/internal/worker.js b/lib/internal/worker.js index cb1ba66e97469a..90c527aaf73947 100644 --- a/lib/internal/worker.js +++ b/lib/internal/worker.js @@ -290,7 +290,7 @@ class Worker extends EventEmitter { if (!isInternal && hasCustomizations) { // - send the second side of the channel to the hooks thread - loaderModule.getHooksProxy().makeAsyncRequest( + loaderModule.getHooksProxy().makeSyncRequest( '#registerWorkerClient', [toWorkerThread], toWorkerThread); } // Use this to cache the Worker's loopStart value once available. diff --git a/test/es-module/test-esm-loader-threads.mjs b/test/es-module/test-esm-loader-threads.mjs index f99493e124a32c..7310a9ac5b54ac 100644 --- a/test/es-module/test-esm-loader-threads.mjs +++ b/test/es-module/test-esm-loader-threads.mjs @@ -35,4 +35,40 @@ describe('off-thread hooks', { concurrency: true }, () => { strictEqual(code, 0); strictEqual(signal, null); }); + + it('propagates the exit code from worker thread import exiting from resolve hook', async () => { + const { code, signal, stdout, stderr } = await spawnPromisified(execPath, [ + '--no-warnings', + '--import', + `data:text/javascript,${encodeURIComponent(` + import { register } from 'node:module'; + register(${JSON.stringify(fixtures.fileURL('es-module-loaders/hooks-exit-worker.mjs'))}); + `)}`, + fixtures.path('es-module-loaders/worker-log-fail-worker-resolve.mjs'), + ]); + + strictEqual(stderr, ''); + strictEqual(stdout.split('\n').filter((line) => line.startsWith('resolve process-exit-module-resolve')).length, 1); + strictEqual(code, 42); + strictEqual(signal, null); + }); + + it('propagates the exit code from worker thread import exiting from load hook', async () => { + const { code, signal, stdout, stderr } = await spawnPromisified(execPath, [ + '--no-warnings', + '--import', + `data:text/javascript,${encodeURIComponent(` + import { register } from 'node:module'; + register(${JSON.stringify(fixtures.fileURL('es-module-loaders/hooks-exit-worker.mjs'))}); + `)}`, + fixtures.path('es-module-loaders/worker-log-fail-worker-load.mjs'), + ]); + + strictEqual(stderr, ''); + strictEqual(stdout.split('\n').filter((line) => line.startsWith('resolve process-exit-module-load')).length, 1); + strictEqual(stdout.split('\n').filter((line) => line.startsWith('load process-exit-on-load:///')).length, 1); + strictEqual(code, 43); + strictEqual(signal, null); + }); + }); diff --git a/test/fixtures/es-module-loaders/hooks-exit-worker.mjs b/test/fixtures/es-module-loaders/hooks-exit-worker.mjs new file mode 100644 index 00000000000000..d499a835e6456c --- /dev/null +++ b/test/fixtures/es-module-loaders/hooks-exit-worker.mjs @@ -0,0 +1,21 @@ +import { writeFileSync } from 'node:fs'; + +export function resolve(specifier, context, next) { + writeFileSync(1, `resolve ${specifier}\n`); + if (specifier === 'process-exit-module-resolve') { + process.exit(42); + } + + if (specifier === 'process-exit-module-load') { + return { __proto__: null, shortCircuit: true, url: 'process-exit-on-load:///' } + } + return next(specifier, context); +} + +export function load(url, context, next) { + writeFileSync(1, `load ${url}\n`); + if (url === 'process-exit-on-load:///') { + process.exit(43); + } + return next(url, context); +} diff --git a/test/fixtures/es-module-loaders/worker-fail-on-load.mjs b/test/fixtures/es-module-loaders/worker-fail-on-load.mjs new file mode 100644 index 00000000000000..46e88664a03c5c --- /dev/null +++ b/test/fixtures/es-module-loaders/worker-fail-on-load.mjs @@ -0,0 +1 @@ +import 'process-exit-module-load'; diff --git a/test/fixtures/es-module-loaders/worker-fail-on-resolve.mjs b/test/fixtures/es-module-loaders/worker-fail-on-resolve.mjs new file mode 100644 index 00000000000000..e8e7adde42585f --- /dev/null +++ b/test/fixtures/es-module-loaders/worker-fail-on-resolve.mjs @@ -0,0 +1 @@ +import 'process-exit-module-resolve'; diff --git a/test/fixtures/es-module-loaders/worker-log-fail-worker-load.mjs b/test/fixtures/es-module-loaders/worker-log-fail-worker-load.mjs new file mode 100644 index 00000000000000..81797da392cb7a --- /dev/null +++ b/test/fixtures/es-module-loaders/worker-log-fail-worker-load.mjs @@ -0,0 +1,12 @@ +import { Worker } from 'worker_threads'; +import { foo } from './module-named-exports.mjs'; + +const workerURLFailOnLoad = new URL('./worker-fail-on-load.mjs', import.meta.url); +console.log(foo); + +// Spawn a worker that will fail to import a dependant module +new Worker(workerURLFailOnLoad); + +process.on('exit', (code) => { + console.log(`process exit code: ${code}`) +}); diff --git a/test/fixtures/es-module-loaders/worker-log-fail-worker-resolve.mjs b/test/fixtures/es-module-loaders/worker-log-fail-worker-resolve.mjs new file mode 100644 index 00000000000000..b5ff238967f4ef --- /dev/null +++ b/test/fixtures/es-module-loaders/worker-log-fail-worker-resolve.mjs @@ -0,0 +1,12 @@ +import { Worker } from 'worker_threads'; +import { foo } from './module-named-exports.mjs'; + +const workerURLFailOnResolve = new URL('./worker-fail-on-resolve.mjs', import.meta.url); +console.log(foo); + +// Spawn a worker that will fail to import a dependant module +new Worker(workerURLFailOnResolve); + +process.on('exit', (code) => { + console.log(`process exit code: ${code}`) +}); From 3ee4295e436ade8be7143930f87b782950b26d26 Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Tue, 30 Apr 2024 12:35:47 +0200 Subject: [PATCH 08/13] implement unregistration of workers from the hooks thread upon exit --- lib/internal/modules/esm/worker.js | 22 ++++++++++++++++++---- lib/internal/worker.js | 16 +++++++++++++--- 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/lib/internal/modules/esm/worker.js b/lib/internal/modules/esm/worker.js index e0d1317c0f1c10..79164e1cd651f5 100644 --- a/lib/internal/modules/esm/worker.js +++ b/lib/internal/modules/esm/worker.js @@ -1,6 +1,7 @@ 'use strict'; const { + ArrayPrototypeFilter, ArrayPrototypePush, AtomicsAdd, AtomicsNotify, @@ -101,7 +102,7 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { const exitMsg = wrapMessage('exit', code ?? process.exitCode); if (hooks) { for (let i = 0; i < allThreadRegisteredHandlerPorts.length; i++) { - const registeredPort = allThreadRegisteredHandlerPorts[i]; + const { port: registeredPort } = allThreadRegisteredHandlerPorts[i]; registeredPort.postMessage(exitMsg); } @@ -182,22 +183,35 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { setImmediate(() => {}); }); - const allThreadRegisteredHandlerPorts = []; + let allThreadRegisteredHandlerPorts = []; /** * @callback registerHandler * @param {MessagePort} toWorkerThread - Upon Worker creation a message channel between the new Worker * and the Hooks thread is being initialized. This is the MessagePort that the Hooks thread will use post * messages to the worker. The other MessagePort is passed to the new Worker itself via LOAD_SCRIPT message */ - function registerHandler(toWorkerThread) { + function registerHandler(toWorkerThread, registeredThreadId) { toWorkerThread.on('message', handleMessage); - ArrayPrototypePush(allThreadRegisteredHandlerPorts, toWorkerThread); + ArrayPrototypePush(allThreadRegisteredHandlerPorts, { port: toWorkerThread, registeredThreadId }); + } + + /** + * @callback registerHandler + * @param {number} unregisteredThreadId - the thread id of the worker thread that is being unregistered + * from the Hooks Thread + */ + function unregisterHandler(unregisteredThreadId) { + allThreadRegisteredHandlerPorts = ArrayPrototypeFilter( + allThreadRegisteredHandlerPorts, (el) => el.registeredThreadId !== unregisteredThreadId); } function getMessageHandler(method) { if (method === '#registerWorkerClient') { return registerHandler; } + if (method === '#unregisterWorkerClient') { + return unregisterHandler; + } return hooks[method]; } diff --git a/lib/internal/worker.js b/lib/internal/worker.js index 90c527aaf73947..766659d4464379 100644 --- a/lib/internal/worker.js +++ b/lib/internal/worker.js @@ -132,7 +132,7 @@ class Worker extends EventEmitter { constructor(filename, options = kEmptyObject) { throwIfBuildingSnapshot('Creating workers'); super(); - const isInternal = arguments[2] === kIsInternal; + const isInternal = this.#isInternal = arguments[2] === kIsInternal; debug( `[${threadId}] create new worker`, filename, @@ -289,9 +289,11 @@ class Worker extends EventEmitter { const hasCustomizations = loaderModule.hasCustomizations(); if (!isInternal && hasCustomizations) { - // - send the second side of the channel to the hooks thread + // - send the second side of the channel to the hooks thread, + // also announce the threadId of the Worker that will use that port. + // This is needed for the cleanup stage loaderModule.getHooksProxy().makeSyncRequest( - '#registerWorkerClient', [toWorkerThread], toWorkerThread); + '#registerWorkerClient', [toWorkerThread], toWorkerThread, this.threadId); } // Use this to cache the Worker's loopStart value once available. this[kLoopStartTime] = -1; @@ -312,6 +314,12 @@ class Worker extends EventEmitter { [kOnExit](code, customErr, customErrReason) { debug(`[${threadId}] hears end event for Worker ${this.threadId}`); + const loaderModule = require('internal/modules/esm/loader'); + const hasCustomizations = loaderModule.hasCustomizations(); + + if (!this.#isInternal && hasCustomizations) { + loaderModule.getHooksProxy()?.makeAsyncRequest('#unregisterWorkerClient', undefined, this.threadId); + } drainMessagePort(this[kPublicPort]); drainMessagePort(this[kPort]); this.removeAllListeners('message'); @@ -454,6 +462,8 @@ class Worker extends EventEmitter { return makeResourceLimits(this[kHandle].getResourceLimits()); } + #isInternal = false; + getHeapSnapshot(options) { const { HeapSnapshotStream, From f233e04ea708c4634ef6229d1d50594fdc49990c Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Tue, 30 Apr 2024 13:01:30 +0200 Subject: [PATCH 09/13] address review finding --- lib/internal/modules/esm/worker.js | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/lib/internal/modules/esm/worker.js b/lib/internal/modules/esm/worker.js index 79164e1cd651f5..79a516bb73f430 100644 --- a/lib/internal/modules/esm/worker.js +++ b/lib/internal/modules/esm/worker.js @@ -228,7 +228,8 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { // the main thread. let hasError = false; let shouldRemoveGlobalErrorHandler = false; - assert(typeof getMessageHandler(method) === 'function'); + const messageHandler = getMessageHandler(method); + assert(typeof messageHandler === 'function'); if (port == null && !hasUncaughtExceptionCaptureCallback()) { // When receiving sync messages, we want to unlock the main thread when there's an exception. process.on('uncaughtException', errorHandler); @@ -251,7 +252,7 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { let response; try { - response = await ReflectApply(getMessageHandler(method), hooks, args); + response = await ReflectApply(messageHandler, hooks, args); } catch (exception) { hasError = true; response = exception; From c014520ddfd0c5f4cddf11fbe6878b7c45322777 Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Tue, 30 Apr 2024 15:41:23 +0200 Subject: [PATCH 10/13] fix comment --- lib/internal/modules/esm/worker.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/internal/modules/esm/worker.js b/lib/internal/modules/esm/worker.js index 79a516bb73f430..088667f3c0d5d7 100644 --- a/lib/internal/modules/esm/worker.js +++ b/lib/internal/modules/esm/worker.js @@ -108,7 +108,7 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { for (const { port, lock: operationLock } of unsettledResponsePorts) { port.postMessage(exitMsg); - // Wake all threads that have pending operations. Is that needed??? + // Wake all threads that have pending operations. AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION); } @@ -163,7 +163,7 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { process.on('beforeExit', () => { for (const { port, lock: operationLock } of unsettledResponsePorts) { port.postMessage(wrapMessage('never-settle')); - // Wake all threads that have pending operations. Is that needed??? + // Wake all threads that have pending operations. AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION); } From e8708bd7f15549e34e840354b6aa8011f6f41e8b Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Mon, 6 May 2024 10:46:19 +0200 Subject: [PATCH 11/13] fix or skip tests that are not fit for running on worker threads --- test/common/index.mjs | 2 ++ test/es-module/test-esm-loader-mock.mjs | 7 ++++++- test/es-module/test-esm-named-exports.js | 4 +++- test/es-module/test-esm-named-exports.mjs | 9 +++++---- test/es-module/test-esm-virtual-json.mjs | 3 ++- .../es-module-loaders/builtin-named-exports.mjs | 13 ++++++++----- .../not-found-assert-loader.mjs | 17 +++++++---------- 7 files changed, 33 insertions(+), 22 deletions(-) diff --git a/test/common/index.mjs b/test/common/index.mjs index ca2994f6e1360f..ec9d6ce8b2432c 100644 --- a/test/common/index.mjs +++ b/test/common/index.mjs @@ -50,6 +50,7 @@ const { skipIfDumbTerminal, skipIfEslintMissing, skipIfInspectorDisabled, + skipIfWorker, spawnPromisified, } = common; @@ -104,5 +105,6 @@ export { skipIfDumbTerminal, skipIfEslintMissing, skipIfInspectorDisabled, + skipIfWorker, spawnPromisified, }; diff --git a/test/es-module/test-esm-loader-mock.mjs b/test/es-module/test-esm-loader-mock.mjs index 164d0ac3775039..0d39f549581a54 100644 --- a/test/es-module/test-esm-loader-mock.mjs +++ b/test/es-module/test-esm-loader-mock.mjs @@ -1,6 +1,11 @@ -import '../common/index.mjs'; +import { skipIfWorker } from '../common/index.mjs'; import assert from 'node:assert/strict'; import { mock } from '../fixtures/es-module-loaders/mock.mjs'; +// Importing mock.mjs above will call `register` to modify the loaders chain. +// Modifying the loader chain is not supported currently when running from a worker thread. +// Relevant PR: https://github.com/nodejs/node/pull/52706 +// See comment: https://github.com/nodejs/node/pull/52706/files#r1585144580 +skipIfWorker(); mock('node:events', { EventEmitter: 'This is mocked!' diff --git a/test/es-module/test-esm-named-exports.js b/test/es-module/test-esm-named-exports.js index 2c6f67288aa57c..00b7aebbfd1f46 100644 --- a/test/es-module/test-esm-named-exports.js +++ b/test/es-module/test-esm-named-exports.js @@ -1,7 +1,9 @@ // Flags: --import ./test/fixtures/es-module-loaders/builtin-named-exports.mjs 'use strict'; -require('../common'); +const common = require('../common'); +common.skipIfWorker(); + const { readFile, __fromLoader } = require('fs'); const assert = require('assert'); diff --git a/test/es-module/test-esm-named-exports.mjs b/test/es-module/test-esm-named-exports.mjs index bbe9c96b92d9b8..6e584b05aa204f 100644 --- a/test/es-module/test-esm-named-exports.mjs +++ b/test/es-module/test-esm-named-exports.mjs @@ -1,9 +1,10 @@ // Flags: --import ./test/fixtures/es-module-loaders/builtin-named-exports.mjs -import '../common/index.mjs'; -import { readFile, __fromLoader } from 'fs'; +import { skipIfWorker } from '../common/index.mjs'; +import * as fs from 'fs'; import assert from 'assert'; import ok from '../fixtures/es-modules/test-esm-ok.mjs'; +skipIfWorker(); assert(ok); -assert(readFile); -assert(__fromLoader); +assert(fs.readFile); +assert(fs.__fromLoader); diff --git a/test/es-module/test-esm-virtual-json.mjs b/test/es-module/test-esm-virtual-json.mjs index a42b037fc1f200..1064a6af5026cf 100644 --- a/test/es-module/test-esm-virtual-json.mjs +++ b/test/es-module/test-esm-virtual-json.mjs @@ -1,7 +1,8 @@ -import '../common/index.mjs'; +import { skipIfWorker } from '../common/index.mjs'; import * as fixtures from '../common/fixtures.mjs'; import { register } from 'node:module'; import assert from 'node:assert'; +skipIfWorker(); async function resolve(referrer, context, next) { const result = await next(referrer, context); diff --git a/test/fixtures/es-module-loaders/builtin-named-exports.mjs b/test/fixtures/es-module-loaders/builtin-named-exports.mjs index 123b12c26bf0c9..4e22f631eba416 100644 --- a/test/fixtures/es-module-loaders/builtin-named-exports.mjs +++ b/test/fixtures/es-module-loaders/builtin-named-exports.mjs @@ -1,3 +1,4 @@ +import { isMainThread } from '../../common/index.mjs'; import * as fixtures from '../../common/fixtures.mjs'; import { createRequire, register } from 'node:module'; @@ -10,8 +11,10 @@ Object.defineProperty(globalThis, GET_BUILTIN, { configurable: false, }); -register(fixtures.fileURL('es-module-loaders/builtin-named-exports-loader.mjs'), { - data: { - GET_BUILTIN, - }, -}); +if (isMainThread) { + register(fixtures.fileURL('es-module-loaders/builtin-named-exports-loader.mjs'), { + data: { + GET_BUILTIN, + }, + }); +} diff --git a/test/fixtures/es-module-loaders/not-found-assert-loader.mjs b/test/fixtures/es-module-loaders/not-found-assert-loader.mjs index bf66efbd0810e5..7d53e31df918a7 100644 --- a/test/fixtures/es-module-loaders/not-found-assert-loader.mjs +++ b/test/fixtures/es-module-loaders/not-found-assert-loader.mjs @@ -1,16 +1,13 @@ import assert from 'node:assert'; // A loader that asserts that the defaultResolve will throw "not found" -// (skipping the top-level main of course, and the built-in ones needed for run-worker). -let mainLoad = true; export async function resolve(specifier, { importAttributes }, next) { - if (mainLoad || specifier === 'path' || specifier === 'worker_threads') { - mainLoad = false; - return next(specifier); + if (specifier.startsWith('./not-found')) { + await assert.rejects(next(specifier), { code: 'ERR_MODULE_NOT_FOUND' }); + return { + url: 'node:fs', + importAttributes, + }; } - await assert.rejects(next(specifier), { code: 'ERR_MODULE_NOT_FOUND' }); - return { - url: 'node:fs', - importAttributes, - }; + return next(specifier); } From a25aab1d1d3010a9f14a1efb4ec13214f02eb923 Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Tue, 7 May 2024 21:15:38 +0200 Subject: [PATCH 12/13] fix flaky test with debug build --- src/handle_wrap.cc | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/handle_wrap.cc b/src/handle_wrap.cc index 69e2a389f9e148..17728380082220 100644 --- a/src/handle_wrap.cc +++ b/src/handle_wrap.cc @@ -148,7 +148,8 @@ void HandleWrap::OnClose(uv_handle_t* handle) { wrap->OnClose(); wrap->handle_wrap_queue_.Remove(); - if (!wrap->persistent().IsEmpty() && + if (!env->isolate()->IsExecutionTerminating() && + !wrap->persistent().IsEmpty() && wrap->object()->Has(env->context(), env->handle_onclose_symbol()) .FromMaybe(false)) { wrap->MakeCallback(env->handle_onclose_symbol(), 0, nullptr); From 76599010882d6f66162db882d442d5b15e7d4cab Mon Sep 17 00:00:00 2001 From: Gabriel Bota Date: Tue, 7 May 2024 21:32:51 +0200 Subject: [PATCH 13/13] fixup! fix flaky test with debug build --- src/handle_wrap.cc | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/handle_wrap.cc b/src/handle_wrap.cc index 17728380082220..1fbe0178c7a10d 100644 --- a/src/handle_wrap.cc +++ b/src/handle_wrap.cc @@ -150,8 +150,9 @@ void HandleWrap::OnClose(uv_handle_t* handle) { if (!env->isolate()->IsExecutionTerminating() && !wrap->persistent().IsEmpty() && - wrap->object()->Has(env->context(), env->handle_onclose_symbol()) - .FromMaybe(false)) { + wrap->object() + ->Has(env->context(), env->handle_onclose_symbol()) + .FromMaybe(false)) { wrap->MakeCallback(env->handle_onclose_symbol(), 0, nullptr); } }