reuse the public port instead of creating an additional communication…

… channel
nodejs · aduh95 · Apr 13, 2023 · Dec 19, 2022 · Jan 10, 2023 · Dec 20, 2022
commit f6679bc0d7f4e76c3d274b3b8d7118a6640755a3
diff --git a/lib/internal/modules/esm/hooks.js b/lib/internal/modules/esm/hooks.js
@@ -471,9 +471,9 @@ class HooksProxy {
    */
   #lock;
   /**
-   * A MessagePort used to synchronously communicate with the worker.
+   * The InternalWorker instance, which let us communicate with the loader thread.
    */
-  #syncCommPort;
+  #worker;
 
   #isReady = false;
 
@@ -483,30 +483,26 @@ class HooksProxy {
 
     const lock = new SharedArrayBuffer(SHARED_MEMORY_BYTE_LENGTH);
     this.#lock = new Int32Array(lock);
-    const syncCommChannel = new MessageChannel();
-    this.#syncCommPort = syncCommChannel.port1;
 
-    const worker = this.worker = new InternalWorker('internal/modules/esm/worker', {
+    this.#worker = new InternalWorker('internal/modules/esm/worker', {
       stderr: false,
       stdin: false,
       stdout: false,
       trackUnmanagedFds: false,
       workerData: {
         lock,
-        syncCommPort: syncCommChannel.port2,
       },
-      transferList: [syncCommChannel.port2],
     });
-    worker.unref(); // ! Allows the process to eventually exit.
+    this.#worker.unref(); // ! Allows the process to eventually exit.
   }
 
   #waitForWorker() {
     if (!this.#isReady) {
       const { kIsOnline } = require('internal/worker');
-      if (!this.worker[kIsOnline]) {
+      if (!this.#worker[kIsOnline]) {
         debug('wait for signal from worker');
         AtomicsWait(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 0);
-        const response = receiveMessageOnPort(this.#syncCommPort);
+        const response = this.#worker.receiveMessageSync();
         const { preloadScripts } = this.#unwrapMessage(response);
         this.#executePreloadScripts(preloadScripts);
       }
@@ -521,11 +517,11 @@ class HooksProxy {
     if (value & WORKER_THREAD_IS_IDLE && AtomicsLoad(this.#lock, NUMBER_OF_MESSAGES_IN_TRANSIT) === 0) {
       debug('send terminate order to worker');
       // The worker has already reported waiting for us to die together.
-      this.worker.terminate();
+      this.#worker.terminate();
     } else {
       // The worker is still in good health, and we just told it it's the
       // only thing keeping us alive.
-      this.worker.ref();
+      this.#worker.ref();
     }
   };
 
@@ -538,7 +534,7 @@ class HooksProxy {
 
     // Pass work to the worker.
     debug('post async message to worker', { method, args });
-    this.#syncCommPort.postMessage({ method, args, port: asyncCommChannel.port2 }, [asyncCommChannel.port2]);
+    this.#worker.postMessage({ method, args, port: asyncCommChannel.port2 }, [asyncCommChannel.port2]);
     // In case the worker was bored, it might lock itself waiting for a signal.
     AtomicsNotify(this.#lock, NUMBER_OF_MESSAGES_IN_TRANSIT);
 
@@ -561,13 +557,15 @@ class HooksProxy {
     debug('start waiting for async response from worker', { method, args });
     while (response == null) {
       await AtomicsWaitAsync(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 0).value;
-      AtomicsAnd(this.#lock, IDLENESS_OF_THREADS, ~MAIN_THREAD_IS_IDLE); // We are no longer bored.
+
+      // In case the beforeExit handler was called during the await, we revert its actions.
+      AtomicsAnd(this.#lock, IDLENESS_OF_THREADS, ~MAIN_THREAD_IS_IDLE);
+      this.#worker.unref();
 
       response = receiveMessageOnPort(asyncCommChannel.port1);
     }
     debug('got async response from worker', { method, args }, this.#lock);
 
-    this.worker.unref(); // We unref the worker here so the beforeExit handler can be called.
     if (AtomicsSub(this.#lock, NUMBER_OF_INCOMING_ASYNC_RESPONSES, 1) <= 1) {
       // We got all the responses from the worker, its job is done (until next time).
       process.off('beforeExit', this.#beforeExitHandler);
@@ -584,7 +582,7 @@ class HooksProxy {
 
     // Pass work to the worker.
     debug('post sync message to worker', { method, args });
-    this.#syncCommPort.postMessage({ method, args });
+    this.#worker.postMessage({ method, args });
 
     // In case the worker was bored, it might lock itself waiting for a signal.
     AtomicsNotify(this.#lock, NUMBER_OF_MESSAGES_IN_TRANSIT);
@@ -594,7 +592,7 @@ class HooksProxy {
     while (response == null) {
       AtomicsWait(this.#lock, WORKER_TO_MAIN_THREAD_NOTIFICATION, 0); // Sleep until worker responds.
 
-      response = receiveMessageOnPort(this.#syncCommPort);
+      response = this.#worker.receiveMessageSync();
 
       if (response == null && AtomicsLoad(this.#lock, IDLENESS_OF_THREADS) & WORKER_THREAD_IS_IDLE) {
         process.exit(kUnfinishedTopLevelAwait);

diff --git a/lib/internal/modules/esm/worker.js b/lib/internal/modules/esm/worker.js
@@ -193,9 +193,8 @@ async function customizedModuleWorker(lock, syncCommPort, errorHandler) {
 /**
  * ! Run everything possible within this function so errors get reported.
  */
-module.exports = function(workerData) {
+module.exports = function(workerData, syncCommPort) {
   const lock = new Int32Array(workerData.lock);
-  const { syncCommPort } = workerData;
 
   function errorHandler(err, origin = 'unhandledRejection') {
     AtomicsNotify(lock, WORKER_TO_MAIN_THREAD_NOTIFICATION);

diff --git a/lib/internal/worker.js b/lib/internal/worker.js
@@ -45,6 +45,7 @@ const { getOptionValue } = require('internal/options');
 const workerIo = require('internal/worker/io');
 const {
   drainMessagePort,
+  receiveMessageOnPort,
   MessageChannel,
   messageTypes,
   kPort,
@@ -455,10 +456,13 @@ class Worker extends EventEmitter {
 }
 
 class InternalWorker extends Worker {
-class InternalWorker extends Worker {
+/**
+ * A worker to be used internally by node (ex module/esm/worker). This functions as a siganal to
+ * handle its setup and script in a particular way.
+ */
+class InternalWorker extends Worker {
-class InternalWorker extends Worker {
+/**
+ * A worker to be used internally by node (ex module/esm/worker). This functions as a siganal to
+ * handle its setup and script in a particular way.
+ */
+class InternalWorker extends Worker {
-
   constructor(filename, options) {
     super(filename, options, kIsInternal);
   }
+
+  receiveMessageSync() {
+    return receiveMessageOnPort(this[kPublicPort]);
+  }
 }
 
 function pipeWithoutWarning(source, dest) {