module: allow module.register from workers

Author: dygabo

lib/internal/modules/esm/hooks.js

@@ -1,8 +1,11 @@
 'use strict';
 
 const {
+  ArrayPrototypeFilter,
+  ArrayPrototypeMap,
   ArrayPrototypePush,
   ArrayPrototypePushApply,
+  ArrayPrototypeReduce,
   AtomicsLoad,
   AtomicsWait,
   AtomicsWaitAsync,
@@ -35,7 +38,7 @@ const {
 const { exitCodes: { kUnsettledTopLevelAwait } } = internalBinding('errors');
 const { URL } = require('internal/url');
 const { canParse: URLCanParse } = internalBinding('url');
-const { receiveMessageOnPort, isMainThread } = require('worker_threads');
+const { receiveMessageOnPort } = require('worker_threads');
 const {
   isAnyArrayBuffer,
   isArrayBufferView,
@@ -164,6 +167,18 @@ class Hooks {
    * @returns {any | Promise<any>} User data, ignored unless it's a promise, in which case it will be awaited.
    */
   addCustomLoader(url, exports, data) {
+    const alreadyKnown = ArrayPrototypeReduce(
+      ArrayPrototypeMap(['initialize', 'resolve', 'load'], (hookName) => {
+        if (this.#chains[hookName]) {
+          const res2 = ArrayPrototypeFilter(this.#chains[hookName], (el) => el.url === url);
+          return res2.length;
+        }
+      }), (acc, val) => acc || (val === 1), false);
+
+    if (alreadyKnown) {
+      return undefined;
+    }
+
     const {
       initialize,
       resolve,
@@ -499,13 +514,14 @@ class HooksProxy {
   #numberOfPendingAsyncResponses = 0;
 
   #isReady = false;
+  #isWorkerOwner = false;
 
   constructor() {
-    const { InternalWorker, hooksPort } = require('internal/worker');
+    const { InternalWorker, hooksPort, hasHooksThread } = require('internal/worker');
     const lock = new SharedArrayBuffer(SHARED_MEMORY_BYTE_LENGTH);
     this.#lock = new Int32Array(lock);
 
-    if (isMainThread) {
+    if (!hasHooksThread()) {
       // Main thread is the only one that creates the internal single hooks worker
       this.#worker = new InternalWorker(loaderWorkerId, {
         stderr: false,
@@ -518,6 +534,7 @@ class HooksProxy {
       });
       this.#worker.unref(); // ! Allows the process to eventually exit.
       this.#worker.on('exit', process.exit);
+      this.#isWorkerOwner = true;
       this.#portToHooksThread = this.#worker;
     } else {
       this.#portToHooksThread = hooksPort;
@@ -529,7 +546,7 @@ class HooksProxy {
     // 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) {
+    if (!this.#isWorkerOwner) {
       return;
     }
 

lib/internal/modules/esm/loader.js

@@ -41,7 +41,6 @@ const { ModuleWrap, kEvaluating, kEvaluated } = internalBinding('module_wrap');
 const {
   urlToFilename,
 } = require('internal/modules/helpers');
-const { isMainThread } = require('worker_threads');
 let defaultResolve, defaultLoad, defaultLoadSync, importMetaInitializer;
 
 /**
@@ -623,11 +622,9 @@ class CustomizedModuleLoader {
    * @returns {{ format: string, url: URL['href'] } | undefined}
    */
   register(originalSpecifier, parentURL, data, transferList) {
-    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);
-    }
+    // 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);
   }
 
   /**
@@ -640,6 +637,9 @@ class CustomizedModuleLoader {
    * @returns {{ format: string, url: URL['href'] }}
    */
   resolve(originalSpecifier, parentURL, importAttributes) {
+    // const FS = require('fs');
+    // const UTIL = require('util');
+    // FS.writeFileSync(1, `resolve(${originalSpecifier}). ${Error().stack}\n`);
     return hooksProxy.makeAsyncRequest('resolve', undefined, originalSpecifier, parentURL, importAttributes);
   }
 

lib/internal/worker.js

@@ -69,6 +69,7 @@ const {
   resourceLimits: resourceLimitsRaw,
   threadId,
   Worker: WorkerImpl,
+  hasHooksThread,
   kMaxYoungGenerationSizeMb,
   kMaxOldGenerationSizeMb,
   kCodeRangeSizeMb,
@@ -562,6 +563,7 @@ module.exports = {
   isMainThread,
   SHARE_ENV,
   hooksPort: undefined,
+  hasHooksThread,
   resourceLimits:
     !isMainThread ? makeResourceLimits(resourceLimitsRaw) : {},
   setEnvironmentData,

src/node_worker.cc

@@ -46,6 +46,7 @@ namespace node {
 namespace worker {
 
 constexpr double kMB = 1024 * 1024;
+std::atomic_bool Worker::internalExists{false};
 
 Worker::Worker(Environment* env,
                Local<Object> wrap,
@@ -489,6 +490,8 @@ void Worker::New(const FunctionCallbackInfo<Value>& args) {
   if (is_internal->IsFalse()) {
     THROW_IF_INSUFFICIENT_PERMISSIONS(
         env, permission::PermissionScope::kWorkerThreads, "");
+  } else {
+    internalExists = true;
   }
   Isolate* isolate = args.GetIsolate();
 
@@ -903,6 +906,10 @@ void Worker::LoopStartTime(const FunctionCallbackInfo<Value>& args) {
   args.GetReturnValue().Set(loop_start_time / 1e6);
 }
 
+void Worker::HasHooksThreadAlready(const FunctionCallbackInfo<Value>& args) {
+  args.GetReturnValue().Set(Worker::internalExists);
+}
+
 namespace {
 
 // Return the MessagePort that is global for this Environment and communicates
@@ -940,6 +947,7 @@ void CreateWorkerPerIsolateProperties(IsolateData* isolate_data,
     SetProtoMethod(isolate, w, "loopStartTime", Worker::LoopStartTime);
 
     SetConstructorFunction(isolate, target, "Worker", w);
+    SetMethodNoSideEffect(isolate, target, "hasHooksThread", Worker::HasHooksThread);
   }
 
   {
@@ -1011,6 +1019,7 @@ void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(Worker::TakeHeapSnapshot);
   registry->Register(Worker::LoopIdleTime);
   registry->Register(Worker::LoopStartTime);
+  registry->Register(Worker::HasHooksThreadAlready);
 }
 
 }  // anonymous namespace

src/node_worker.h

@@ -5,6 +5,7 @@
 
 #include <optional>
 #include <unordered_map>
+#include <atomic>
 #include "node_exit_code.h"
 #include "node_messaging.h"
 #include "uv.h"
@@ -76,6 +77,7 @@ class Worker : public AsyncWrap {
   static void TakeHeapSnapshot(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void LoopIdleTime(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void LoopStartTime(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void HasHooksThreadAlready(const v8::FunctionCallbackInfo<v8::Value>& args);
 
  private:
   bool CreateEnvMessagePort(Environment* env);
@@ -102,6 +104,7 @@ class Worker : public AsyncWrap {
   uintptr_t stack_base_ = 0;
   // Optional name used for debugging in inspector and trace events.
   std::string name_;
+  static std::atomic_bool internalExists;
 
   // Custom resource constraints:
   double resource_limits_[kTotalResourceLimitCount];

test/es-module/test-esm-loader-hooks.mjs

@@ -612,7 +612,7 @@ describe('Loader hooks', { concurrency: !process.env.TEST_PARALLEL }, () => {
       ]);
 
       assert.strictEqual(stderr, '');
-      assert.deepStrictEqual(stdout.split('\n'), ['resolve passthru', 'resolve passthru', '']);
+      assert.deepStrictEqual(stdout.split('\n'), ['resolve passthru', '']);
       assert.strictEqual(code, 0);
       assert.strictEqual(signal, null);
     });

test/es-module/test-esm-loader-programmatically.mjs

@@ -163,12 +163,10 @@ describe('ESM: programmatically register loaders', { concurrency: !process.env.T
     const lines = stdout.split('\n');
 
     assert.match(lines[0], /resolve passthru/);
-    assert.match(lines[1], /resolve passthru/);
-    assert.match(lines[2], /load passthru/);
-    assert.match(lines[3], /load passthru/);
-    assert.match(lines[4], /Hello from dynamic import/);
+    assert.match(lines[1], /load passthru/);
+    assert.match(lines[2], /Hello from dynamic import/);
 
-    assert.strictEqual(lines[5], '');
+    assert.strictEqual(lines[3], '');
   });
 
   it('works registering loaders as package name', async () => {

test/es-module/test-esm-loader-threads.mjs

@@ -16,6 +16,9 @@ describe('off-thread hooks', { concurrency: true }, () => {
       fixtures.path('es-module-loaders/workers-spawned.mjs'),
     ]);
 
+    console.log(stderr);
+    console.log(stdout);
+
     strictEqual(stderr, '');
     strictEqual(stdout.split('\n').filter((line) => line.startsWith('initialize')).length, 1);
     strictEqual(stdout.split('\n').filter((line) => line === 'foo').length, 2);
@@ -30,7 +33,9 @@ describe('off-thread hooks', { concurrency: true }, () => {
     // 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);
+    // 6 additional calls to resolve because of the modeul.register being allowed from worker threads (happens
+    // implicitly because of the --import on the main thread)
+    strictEqual(stdout.split('\n').filter((line) => line.startsWith('hooked resolve')).length, 22);
     strictEqual(stdout.split('\n').filter((line) => line.startsWith('hooked load')).length, 16);
     strictEqual(code, 0);
     strictEqual(signal, null);

test/fixtures/es-module-loaders/hooks-log.mjs

@@ -9,7 +9,7 @@ export function initialize() {
 }
 
 export function resolve(specifier, context, next) {
-  writeFileSync(1, `hooked resolve ${++resolveCount} ${specifier}\n`);
+  writeFileSync(1, `hooked resolve ${++resolveCount} ${specifier} \n`);
   return next(specifier, context);
 }