Skip to content

Commit ee16c17

Browse files
dygabodygabo
committed
esm: have a single hooks thread for all workers
1 parent 4221631 commit ee16c17

10 files changed

Lines changed: 214 additions & 53 deletions

File tree

‎lib/internal/main/worker_thread.js‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,7 @@ port.on('message', (message) => {
9797
manifestSrc,
9898
manifestURL,
9999
publicPort,
100+
hooksPort,
100101
workerData,
101102
} = message;
102103

@@ -111,6 +112,7 @@ port.on('message', (message) => {
111112
}
112113

113114
require('internal/worker').assignEnvironmentData(environmentData);
115+
require('internal/worker').hooksPort = hooksPort;
114116

115117
if (SharedArrayBuffer !== undefined) {
116118
// The counter is only passed to the workers created by the main thread,

‎lib/internal/modules/esm/hooks.js‎

Lines changed: 66 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ const {
3535
const { exitCodes: { kUnsettledTopLevelAwait } } = internalBinding('errors');
3636
const { URL } = require('internal/url');
3737
const { canParse: URLCanParse } = internalBinding('url');
38-
const { receiveMessageOnPort } = require('worker_threads');
38+
const { receiveMessageOnPort, isMainThread } = require('worker_threads');
3939
const {
4040
isAnyArrayBuffer,
4141
isArrayBufferView,
@@ -481,6 +481,7 @@ class HooksProxy {
481481
* The InternalWorker instance, which lets us communicate with the loader thread.
482482
*/
483483
#worker;
484+
#portToHooksThread;
484485

485486
/**
486487
* The last notification ID received from the worker. This is used to detect
@@ -499,26 +500,40 @@ class HooksProxy {
499500
#isReady = false;
500501

501502
constructor() {
502-
const { InternalWorker } = require('internal/worker');
503+
const { InternalWorker, hooksPort } = require('internal/worker');
503504
MessageChannel ??= require('internal/worker/io').MessageChannel;
504505

505506
const lock = new SharedArrayBuffer(SHARED_MEMORY_BYTE_LENGTH);
506507
this.#lock = new Int32Array(lock);
507508

508-
this.#worker = new InternalWorker(loaderWorkerId, {
509-
stderr: false,
510-
stdin: false,
511-
stdout: false,
512-
trackUnmanagedFds: false,
513-
workerData: {
514-
lock,
515-
},
516-
});
517-
this.#worker.unref(); // ! Allows the process to eventually exit.
518-
this.#worker.on('exit', process.exit);
509+
if (isMainThread) {
510+
// Main thread is the only one that creates the internal single hooks worker
511+
this.#worker = new InternalWorker(loaderWorkerId, {
512+
stderr: false,
513+
stdin: false,
514+
stdout: false,
515+
trackUnmanagedFds: false,
516+
workerData: {
517+
lock,
518+
},
519+
});
520+
this.#worker.unref(); // ! Allows the process to eventually exit.
521+
this.#worker.on('exit', process.exit);
522+
this.#portToHooksThread = this.#worker;
523+
} else {
524+
this.#portToHooksThread = hooksPort;
525+
}
519526
}
520527

521528
waitForWorker() {
529+
// There is one Hooks instance for each worker thread. But only one of these Hooks instances
530+
// has an InternalWorker. That was the Hooks instance created for the main thread.
531+
// It means for all Hooks instances that are not on the main thread => they are ready because they
532+
// delegate to the single InternalWorker anyway.
533+
if (!isMainThread) {
534+
return;
535+
}
536+
522537
if (!this.#isReady) {
523538
const { kIsOnline } = require('internal/worker');
524539
if (!this.#worker[kIsOnline]) {
@@ -535,6 +550,23 @@ class HooksProxy {
535550
}
536551
}
537552

553+
#postMessageToWorker(method, type, transferList, ...args) {
554+
this.waitForWorker();
555+
MessageChannel ??= require('internal/worker/io').MessageChannel;
556+
const { port1: fromHooksThread, port2: toHooksThread } = new MessageChannel();
557+
558+
// Pass work to the worker.
559+
debug(`post ${type} message to worker`, { method, args, transferList });
560+
const usedTransferList = [toHooksThread];
561+
if (transferList) {
562+
ArrayPrototypePushApply(usedTransferList, transferList);
563+
}
564+
this.#portToHooksThread.postMessage(
565+
{ __proto__: null, method, args, lock: this.#lock, port: toHooksThread }, usedTransferList);
566+
567+
return fromHooksThread;
568+
}
569+
538570
/**
539571
* Invoke a remote method asynchronously.
540572
* @param {string} method Method to invoke
@@ -543,22 +575,7 @@ class HooksProxy {
543575
* @returns {Promise<any>}
544576
*/
545577
async makeAsyncRequest(method, transferList, ...args) {
546-
this.waitForWorker();
547-
548-
MessageChannel ??= require('internal/worker/io').MessageChannel;
549-
const asyncCommChannel = new MessageChannel();
550-
551-
// Pass work to the worker.
552-
debug('post async message to worker', { method, args, transferList });
553-
const finalTransferList = [asyncCommChannel.port2];
554-
if (transferList) {
555-
ArrayPrototypePushApply(finalTransferList, transferList);
556-
}
557-
this.#worker.postMessage({
558-
__proto__: null,
559-
method, args,
560-
port: asyncCommChannel.port2,
561-
}, finalTransferList);
578+
const fromHooksThread = this.#postMessageToWorker(method, 'Async', transferList, ...args);
562579

563580
if (this.#numberOfPendingAsyncResponses++ === 0) {
564581
// On the next lines, the main thread will await a response from the worker thread that might
@@ -567,7 +584,11 @@ class HooksProxy {
567584
// However we want to keep the process alive until the worker thread responds (or until the
568585
// event loop of the worker thread is also empty), so we ref the worker until we get all the
569586
// responses back.
570-
this.#worker.ref();
587+
if (this.#worker) {
588+
this.#worker.ref();
589+
} else {
590+
this.#portToHooksThread.ref();
591+
}
571592
}
572593

573594
let response;
@@ -576,18 +597,25 @@ class HooksProxy {
576597
await AtomicsWaitAsync(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, this.#workerNotificationLastId).value;
577598
this.#workerNotificationLastId = AtomicsLoad(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION);
578599

579-
response = receiveMessageOnPort(asyncCommChannel.port1);
600+
response = receiveMessageOnPort(fromHooksThread);
580601
} while (response == null);
581602
debug('got async response from worker', { method, args }, this.#lock);
582603

583604
if (--this.#numberOfPendingAsyncResponses === 0) {
584605
// We got all the responses from the worker, its job is done (until next time).
585-
this.#worker.unref();
606+
if (this.#worker) {
607+
this.#worker.unref();
608+
} else {
609+
this.#portToHooksThread.unref();
610+
}
586611
}
587612

588-
const body = this.#unwrapMessage(response);
589-
asyncCommChannel.port1.close();
590-
return body;
613+
if (response.message.status === 'exit') {
614+
process.exit(response.message.body);
615+
}
616+
617+
fromHooksThread.close();
618+
return this.#unwrapMessage(response);
591619
}
592620

593621
/**
@@ -598,11 +626,7 @@ class HooksProxy {
598626
* @returns {any}
599627
*/
600628
makeSyncRequest(method, transferList, ...args) {
601-
this.waitForWorker();
602-
603-
// Pass work to the worker.
604-
debug('post sync message to worker', { method, args, transferList });
605-
this.#worker.postMessage({ __proto__: null, method, args }, transferList);
629+
const fromHooksThread = this.#postMessageToWorker(method, 'Sync', transferList, ...args);
606630

607631
let response;
608632
do {
@@ -611,14 +635,15 @@ class HooksProxy {
611635
AtomicsWait(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, this.#workerNotificationLastId);
612636
this.#workerNotificationLastId = AtomicsLoad(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION);
613637

614-
response = this.#worker.receiveMessageSync();
638+
response = receiveMessageOnPort(fromHooksThread);
615639
} while (response == null);
616640
debug('got sync response from worker', { method, args });
617641
if (response.message.status === 'never-settle') {
618642
process.exit(kUnsettledTopLevelAwait);
619643
} else if (response.message.status === 'exit') {
620644
process.exit(response.message.body);
621645
}
646+
fromHooksThread.close();
622647
return this.#unwrapMessage(response);
623648
}
624649

‎lib/internal/modules/esm/loader.js‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ const { ModuleWrap, kEvaluating, kEvaluated } = internalBinding('module_wrap');
4040
const {
4141
urlToFilename,
4242
} = require('internal/modules/helpers');
43+
const { isMainThread } = require('worker_threads');
4344
let defaultResolve, defaultLoad, defaultLoadSync, importMetaInitializer;
4445

4546
/**
@@ -594,21 +595,26 @@ class CustomizedModuleLoader {
594595
*/
595596
constructor() {
596597
getHooksProxy();
598+
_hasCustomizations = true;
597599
}
598600

599601
/**
600-
* Register some loader specifier.
602+
* Register a loader specifier.
601603
* @param {string} originalSpecifier The specified URL path of the loader to
602604
* be registered.
603605
* @param {string} parentURL The parent URL from where the loader will be
604606
* registered if using it package name as specifier
605607
* @param {any} [data] Arbitrary data to be passed from the custom loader
606608
* (user-land) to the worker.
607609
* @param {any[]} [transferList] Objects in `data` that are changing ownership
608-
* @returns {{ format: string, url: URL['href'] }}
610+
* @returns {{ format: string, url: URL['href'] } | undefined}
609611
*/
610612
register(originalSpecifier, parentURL, data, transferList) {
611-
return hooksProxy.makeSyncRequest('register', transferList, originalSpecifier, parentURL, data);
613+
if (isMainThread) {
614+
// Only the main thread has a Hooks instance with worker thread. All other Worker threads
615+
// delegate thier hooks to the HooksThread of the main thread.
616+
return hooksProxy.makeSyncRequest('register', transferList, originalSpecifier, parentURL, data);
617+
}
612618
}
613619

614620
/**
@@ -617,7 +623,7 @@ class CustomizedModuleLoader {
617623
* be resolved.
618624
* @param {string} [parentURL] The URL path of the module's parent.
619625
* @param {ImportAttributes} importAttributes Attributes from the import
620-
* statement or expression.
626+
* statement or exp-ression.
621627
* @returns {{ format: string, url: URL['href'] }}
622628
*/
623629
resolve(originalSpecifier, parentURL, importAttributes) {
@@ -706,6 +712,12 @@ function getHooksProxy() {
706712
return hooksProxy;
707713
}
708714

715+
let _hasCustomizations = false;
716+
function hasCustomizations() {
717+
return _hasCustomizations;
718+
}
719+
720+
709721
let cascadedLoader;
710722

711723
/**
@@ -767,6 +779,7 @@ function register(specifier, parentURL = undefined, options) {
767779

768780
module.exports = {
769781
createModuleLoader,
782+
hasCustomizations,
770783
getHooksProxy,
771784
getOrInitializeCascadedLoader,
772785
register,

‎lib/internal/modules/esm/worker.js‎

Lines changed: 38 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,17 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) {
9797
// so it can detect the exit event.
9898
const { exit } = process;
9999
process.exit = function(code) {
100+
if (hooks) {
101+
for (const registeredPort of allThreadRegisteredHandlerPorts) {
102+
registeredPort.postMessage(wrapMessage('exit', code ?? process.exitCode));
103+
}
104+
for (const { port, lock: operationLock } of unsettledResponsePorts) {
105+
port.postMessage(wrapMessage('exit', code ?? process.exitCode));
106+
// Wake all threads that have pending operations. Is that needed???
107+
AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1);
108+
AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION);
109+
}
110+
}
100111
syncCommPort.postMessage(wrapMessage('exit', code ?? process.exitCode));
101112
AtomicsAdd(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1);
102113
AtomicsNotify(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION);
@@ -145,8 +156,11 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) {
145156
const unsettledResponsePorts = new SafeSet();
146157

147158
process.on('beforeExit', () => {
148-
for (const port of unsettledResponsePorts) {
159+
for (const { port, lock: operationLock } of unsettledResponsePorts) {
149160
port.postMessage(wrapMessage('never-settle'));
161+
// Wake all threads that have pending operations. Is that needed???
162+
AtomicsAdd(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1);
163+
AtomicsNotify(operationLock, WORKER_TO_MAIN_THREAD_NOTIFICATION);
150164
}
151165
unsettledResponsePorts.clear();
152166

@@ -164,24 +178,39 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) {
164178
setImmediate(() => {});
165179
});
166180

181+
const allThreadRegisteredHandlerPorts = [];
182+
function registerHandler(toWorkerThread) {
183+
toWorkerThread.on('message', handleMessage);
184+
allThreadRegisteredHandlerPorts.push(toWorkerThread);
185+
}
186+
187+
function getMessageHandler(method) {
188+
if (method === '#registerWorkerClient') {
189+
return registerHandler;
190+
}
191+
return hooks[method];
192+
}
193+
167194
/**
168195
* Handles incoming messages from the main thread or other workers.
169196
* @param {object} options - The options object.
170197
* @param {string} options.method - The name of the hook.
171198
* @param {Array} options.args - The arguments to pass to the method.
172199
* @param {MessagePort} options.port - The message port to use for communication.
200+
* @param {Int32Array} options.lock - The shared memory where the caller expects to get awaken.
173201
*/
174-
async function handleMessage({ method, args, port }) {
202+
async function handleMessage({ method, args, port, lock: msgLock }) {
175203
// Each potential exception needs to be caught individually so that the correct error is sent to
176204
// the main thread.
177205
let hasError = false;
178206
let shouldRemoveGlobalErrorHandler = false;
179-
assert(typeof hooks[method] === 'function');
207+
assert(typeof getMessageHandler(method) === 'function');
180208
if (port == null && !hasUncaughtExceptionCaptureCallback()) {
181209
// When receiving sync messages, we want to unlock the main thread when there's an exception.
182210
process.on('uncaughtException', errorHandler);
183211
shouldRemoveGlobalErrorHandler = true;
184212
}
213+
const usedLock = msgLock ?? lock;
185214

186215
// We are about to yield the execution with `await ReflectApply` below. In case the code
187216
// following the `await` never runs, we remove the message handler so the `beforeExit` event
@@ -192,17 +221,18 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) {
192221
clearImmediate(immediate);
193222
immediate = setImmediate(checkForMessages).unref();
194223

195-
unsettledResponsePorts.add(port ?? syncCommPort);
224+
const unsettledActionData = { port: port ?? syncCommPort, lock: usedLock };
225+
unsettledResponsePorts.add(unsettledActionData);
196226

197227
let response;
198228
try {
199-
response = await ReflectApply(hooks[method], hooks, args);
229+
response = await ReflectApply(getMessageHandler(method), hooks, args);
200230
} catch (exception) {
201231
hasError = true;
202232
response = exception;
203233
}
204234

205-
unsettledResponsePorts.delete(port ?? syncCommPort);
235+
unsettledResponsePorts.delete(unsettledActionData);
206236

207237
// Send the method response (or exception) to the main thread.
208238
try {
@@ -215,8 +245,8 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) {
215245
(port ?? syncCommPort).postMessage(wrapMessage('error', exception));
216246
}
217247

218-
AtomicsAdd(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1);
219-
AtomicsNotify(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION);
248+
AtomicsAdd(usedLock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 1);
249+
AtomicsNotify(usedLock, WORKER_TO_MAIN_THREAD_NOTIFICATION);
220250
if (shouldRemoveGlobalErrorHandler) {
221251
process.off('uncaughtException', errorHandler);
222252
}

0 commit comments

Comments
 (0)