diff --git a/lib/internal/main/worker_thread.js b/lib/internal/main/worker_thread.js index aa329b9fe04..32bccca9b53 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 ba655116a0b..f5833ad61cd 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, @@ -482,6 +482,8 @@ class HooksProxy { */ #worker; + #portToHooksThread; + /** * The last notification ID received from the worker. This is used to detect * if the worker has already sent a notification before putting the main @@ -499,26 +501,38 @@ class HooksProxy { #isReady = false; constructor() { - const { InternalWorker } = require('internal/worker'); - MessageChannel ??= require('internal/worker/io').MessageChannel; - + const { InternalWorker, hooksPort } = require('internal/worker'); 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 +549,37 @@ 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, + args, + lock: this.#lock, + method, + port: toHooksThread, + }, + usedTransferList, + ); + + return fromHooksThread; + } + /** * Invoke a remote method asynchronously. * @param {string} method Method to invoke @@ -543,22 +588,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 +597,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 +610,26 @@ 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 +640,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 +649,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 +657,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 8fbcb56aa65..5eb7cd17d02 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 their hooks to the HooksThread of the main thread. + return hooksProxy.makeSyncRequest('register', transferList, originalSpecifier, parentURL, data); + } } /** @@ -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 311d77fb099..088667f3c0d 100644 --- a/lib/internal/modules/esm/worker.js +++ b/lib/internal/modules/esm/worker.js @@ -1,6 +1,8 @@ 'use strict'; const { + ArrayPrototypeFilter, + ArrayPrototypePush, AtomicsAdd, AtomicsNotify, DataViewPrototypeGetBuffer, @@ -97,7 +99,21 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { // so it can detect the exit event. const { exit } = process; process.exit = function(code) { - syncCommPort.postMessage(wrapMessage('exit', code ?? process.exitCode)); + const exitMsg = wrapMessage('exit', code ?? process.exitCode); + if (hooks) { + for (let i = 0; i < allThreadRegisteredHandlerPorts.length; i++) { + const { port: registeredPort } = allThreadRegisteredHandlerPorts[i]; + registeredPort.postMessage(exitMsg); + } + + for (const { port, lock: operationLock } of unsettledResponsePorts) { + port.postMessage(exitMsg); + // Wake all threads that have pending operations. + AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); + AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION); + } + } + syncCommPort.postMessage(exitMsg); AtomicsAdd(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); AtomicsNotify(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION); return ReflectApply(exit, this, arguments); @@ -145,8 +161,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. + AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1); + AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION); } unsettledResponsePorts.clear(); @@ -164,24 +183,59 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) { setImmediate(() => {}); }); + 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, registeredThreadId) { + toWorkerThread.on('message', handleMessage); + 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]; + } + /** * 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'); + 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); 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 +246,19 @@ 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(messageHandler, 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 +271,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 9eefaa97021..766659d4464 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, @@ -258,6 +258,15 @@ 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 + ArrayPrototypePush(transferList, toHooksThread); + } ArrayPrototypeForEach(['message', 'messageerror'], (event) => { this[kPublicPort].on(event, (message) => this.emit(event, message)); }); @@ -272,8 +281,20 @@ 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, + // 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, this.threadId); + } // Use this to cache the Worker's loopStart value once available. this[kLoopStartTime] = -1; this[kIsOnline] = false; @@ -293,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'); @@ -435,6 +462,8 @@ class Worker extends EventEmitter { return makeResourceLimits(this[kHandle].getResourceLimits()); } + #isInternal = false; + getHeapSnapshot(options) { const { HeapSnapshotStream, @@ -532,6 +561,7 @@ module.exports = { kIsOnline, isMainThread, SHARE_ENV, + hooksPort: undefined, resourceLimits: !isMainThread ? makeResourceLimits(resourceLimitsRaw) : {}, setEnvironmentData, diff --git a/src/handle_wrap.cc b/src/handle_wrap.cc index 69e2a389f9e..1fbe0178c7a 100644 --- a/src/handle_wrap.cc +++ b/src/handle_wrap.cc @@ -148,9 +148,11 @@ void HandleWrap::OnClose(uv_handle_t* handle) { wrap->OnClose(); wrap->handle_wrap_queue_.Remove(); - if (!wrap->persistent().IsEmpty() && - wrap->object()->Has(env->context(), env->handle_onclose_symbol()) - .FromMaybe(false)) { + 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); } } diff --git a/test/common/index.mjs b/test/common/index.mjs index ca2994f6e13..ec9d6ce8b24 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 164d0ac3775..0d39f549581 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-loader-threads.mjs b/test/es-module/test-esm-loader-threads.mjs new file mode 100644 index 00000000000..7310a9ac5b5 --- /dev/null +++ b/test/es-module/test-esm-loader-threads.mjs @@ -0,0 +1,74 @@ +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); + }); + + 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/es-module/test-esm-named-exports.js b/test/es-module/test-esm-named-exports.js index 2c6f67288aa..00b7aebbfd1 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 bbe9c96b92d..6e584b05aa2 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 a42b037fc1f..1064a6af502 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 123b12c26bf..4e22f631eba 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/hooks-exit-worker.mjs b/test/fixtures/es-module-loaders/hooks-exit-worker.mjs new file mode 100644 index 00000000000..d499a835e64 --- /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/hooks-log.mjs b/test/fixtures/es-module-loaders/hooks-log.mjs new file mode 100644 index 00000000000..2d2512281e8 --- /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/not-found-assert-loader.mjs b/test/fixtures/es-module-loaders/not-found-assert-loader.mjs index bf66efbd081..7d53e31df91 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); } 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 00000000000..46e88664a03 --- /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 00000000000..e8e7adde425 --- /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-again.mjs b/test/fixtures/es-module-loaders/worker-log-again.mjs new file mode 100644 index 00000000000..2969edc8dac --- /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-fail-worker-load.mjs b/test/fixtures/es-module-loaders/worker-log-fail-worker-load.mjs new file mode 100644 index 00000000000..81797da392c --- /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 00000000000..b5ff238967f --- /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}`) +}); 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 00000000000..13290c37d07 --- /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 00000000000..439847656fe --- /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);