worker: add connect and setConnectionsListener

Author: ShogunPanda

doc/api/errors.md

@@ -3072,6 +3072,16 @@ added: v18.1.0
 The `Response` that has been passed to `WebAssembly.compileStreaming` or to
 `WebAssembly.instantiateStreaming` is not a valid WebAssembly response.
 
+<a id="ERR_WORKER_CONNECTION_REFUSED"></a>
+
+### `ERR_WORKER_CONNECTION_REFUSED`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+The thread requested in [`connect()`][] refused the connection or has no connections listener provided.
+
 <a id="ERR_WORKER_INIT_FAILED"></a>
 
 ### `ERR_WORKER_INIT_FAILED`
@@ -3085,6 +3095,16 @@ The `Worker` initialization failed.
 The `execArgv` option passed to the `Worker` constructor contains
 invalid flags.
 
+<a id="ERR_WORKER_INVALID_ID"></a>
+
+### `ERR_WORKER_INVALID_ID`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+The thread id requested in [`connect()`][] is invalid.
+
 <a id="ERR_WORKER_NOT_RUNNING"></a>
 
 ### `ERR_WORKER_NOT_RUNNING`
@@ -3104,6 +3124,16 @@ The `Worker` instance terminated because it reached its memory limit.
 The path for the main script of a worker is neither an absolute path
 nor a relative path starting with `./` or `../`.
 
+<a id="ERR_WORKER_SAME_THREAD"></a>
+
+### `ERR_WORKER_SAME_THREAD`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+The thread id requested in [`connect()`][] is the current thread id.
+
 <a id="ERR_WORKER_UNSERIALIZABLE_ERROR"></a>
 
 ### `ERR_WORKER_UNSERIALIZABLE_ERROR`
@@ -3999,6 +4029,7 @@ An error occurred trying to allocate memory. This should never happen.
 [`Writable`]: stream.md#class-streamwritable
 [`child_process`]: child_process.md
 [`cipher.getAuthTag()`]: crypto.md#ciphergetauthtag
+[`connect()`]: worker_threads.md#workerconnecttarget-data
 [`crypto.getDiffieHellman()`]: crypto.md#cryptogetdiffiehellmangroupname
 [`crypto.scrypt()`]: crypto.md#cryptoscryptpassword-salt-keylen-options-callback
 [`crypto.scryptSync()`]: crypto.md#cryptoscryptsyncpassword-salt-keylen-options

doc/api/worker_threads.md

@@ -61,6 +61,24 @@ Worker threads inherit non-process-specific options by default. Refer to
 [`Worker constructor options`][] to know how to customize worker thread options,
 specifically `argv` and `execArgv` options.
 
+## `worker.connect(target, data)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1.1 - Active development
+
+* `target` {number} The target thread id.
+* `data` {any} Any arbitrary, cloneable JavaScript value.
+* Returns: {Promise} A promise for a `MessagePort`.
+
+Establishes a connection to another worker thread in the same process, returning a
+`MessagePort` that can be used for the communication.
+
+The target thread must have a connection listener setup via [`worker.setConnectionsListener()`][]
+otherwise the connection request will fail.
+
 ## `worker.getEnvironmentData(key)`
 
 <!-- YAML
@@ -325,6 +343,30 @@ new Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV })
   });
 ```
 
+## `worker.setConnectionsListener(fn)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1.1 - Active development
+
+* `fn` {Function} A callback to be executed when [`worker.connect()`][] is called from another thread.
+  The function will receive the following arguments:
+
+  * `sender` {number} The other thread id.
+  * `port` {MessagePort} The port than can be used to communicate with the other thread.
+  * `data` {any} The data passed to [`worker.connect()`][].
+
+  The function must return `true` to accept the connection or any other value to
+  refuse the connection. If the function returns a `Promise`, it will be awaited.
+
+Sets the callback that handles connection from other worker threads in the same process.
+If the callback is `null` or `undefined` then the current listener is removed.
+
+When no listeners are present (the default) all connection requests are immediately
+refused.
+
 ## `worker.setEnvironmentData(key[, value])`
 
 <!-- YAML
@@ -1437,8 +1479,10 @@ thread spawned will spawn another until the application crashes.
 [`v8.getHeapSnapshot()`]: v8.md#v8getheapsnapshotoptions
 [`vm`]: vm.md
 [`worker.SHARE_ENV`]: #workershare_env
+[`worker.connect()`]: #workerconnecttarget-data
 [`worker.on('message')`]: #event-message_1
 [`worker.postMessage()`]: #workerpostmessagevalue-transferlist
+[`worker.setConnectionsListener()`]: #workersetconnectionslistenerfn
 [`worker.terminate()`]: #workerterminate
 [`worker.threadId`]: #workerthreadid_1
 [async-resource-worker-pool]: async_context.md#using-asyncresource-for-a-worker-thread-pool

lib/internal/errors.js

@@ -1858,10 +1858,12 @@ E('ERR_VM_MODULE_NOT_MODULE',
 E('ERR_VM_MODULE_STATUS', 'Module status %s', Error);
 E('ERR_WASI_ALREADY_STARTED', 'WASI instance has already started', Error);
 E('ERR_WEBASSEMBLY_RESPONSE', 'WebAssembly response %s', TypeError);
+E('ERR_WORKER_CONNECTION_REFUSED', 'Connection refused from worker', Error);
 E('ERR_WORKER_INIT_FAILED', 'Worker initialization failure: %s', Error);
 E('ERR_WORKER_INVALID_EXEC_ARGV', (errors, msg = 'invalid execArgv flags') =>
   `Initiated Worker with ${msg}: ${ArrayPrototypeJoin(errors, ', ')}`,
   Error);
+E('ERR_WORKER_INVALID_ID', 'Invalid worker id %d', Error);
 E('ERR_WORKER_NOT_RUNNING', 'Worker instance not running', Error);
 E('ERR_WORKER_OUT_OF_MEMORY',
   'Worker terminated due to reaching memory limit: %s', Error);
@@ -1876,6 +1878,7 @@ E('ERR_WORKER_PATH', (filename) =>
   ) +
   ` Received "${filename}"`,
   TypeError);
+E('ERR_WORKER_SAME_THREAD', 'Cannot connect to the same thread', Error);
 E('ERR_WORKER_UNSERIALIZABLE_ERROR',
   'Serializing an uncaught exception failed', Error);
 E('ERR_WORKER_UNSUPPORTED_OPERATION',

lib/internal/main/worker_thread.js

@@ -28,6 +28,7 @@ const {
   getEnvMessagePort,
 } = internalBinding('worker');
 
+const { processConnectionRequest } = require('internal/worker');
 const workerIo = require('internal/worker/io');
 const {
   messageTypes: {
@@ -40,6 +41,7 @@ const {
     // Messages that may be either received or posted
     STDIO_PAYLOAD,
     STDIO_WANTS_MORE_DATA,
+    CONNECT,
   },
   kStdioWantsMoreDataCallback,
 } = workerIo;
@@ -182,6 +184,8 @@ port.on('message', (message) => {
         break;
       }
     }
+  } else if (message.type === CONNECT) {
+    processConnectionRequest(message);
   } else if (message.type === STDIO_PAYLOAD) {
     const { stream, chunks } = message;
     ArrayPrototypeForEach(chunks, ({ chunk, encoding }) => {

lib/internal/worker.js

@@ -5,8 +5,12 @@ const {
   ArrayPrototypeMap,
   ArrayPrototypePush,
   AtomicsAdd,
+  AtomicsNotify,
+  AtomicsStore,
+  AtomicsWaitAsync,
   Float64Array,
   FunctionPrototypeBind,
+  Int32Array,
   JSONStringify,
   MathMax,
   ObjectEntries,
@@ -34,12 +38,14 @@ const {
 
 const errorCodes = require('internal/errors').codes;
 const {
+  ERR_INVALID_ARG_TYPE,
+  ERR_INVALID_ARG_VALUE,
+  ERR_WORKER_CONNECTION_REFUSED,
+  ERR_WORKER_INVALID_EXEC_ARGV,
   ERR_WORKER_NOT_RUNNING,
   ERR_WORKER_PATH,
+  ERR_WORKER_SAME_THREAD,
   ERR_WORKER_UNSERIALIZABLE_ERROR,
-  ERR_WORKER_INVALID_EXEC_ARGV,
-  ERR_INVALID_ARG_TYPE,
-  ERR_INVALID_ARG_VALUE,
 } = errorCodes;
 
 const workerIo = require('internal/worker/io');
@@ -59,7 +65,7 @@ const {
 const { deserializeError } = require('internal/error_serdes');
 const { fileURLToPath, isURL, pathToFileURL } = require('internal/url');
 const { kEmptyObject } = require('internal/util');
-const { validateArray, validateString } = require('internal/validators');
+const { validateArray, validateFunction, validateString } = require('internal/validators');
 const {
   throwIfBuildingSnapshot,
 } = require('internal/v8/startup_snapshot');
@@ -74,6 +80,8 @@ const {
   kCodeRangeSizeMb,
   kStackSizeMb,
   kTotalResourceLimitCount,
+  sendToWorker,
+  setMainPort,
 } = internalBinding('worker');
 
 const kHandle = Symbol('kHandle');
@@ -100,6 +108,14 @@ let cwdCounter;
 
 const environmentData = new SafeMap();
 
+// SharedArrayBuffer must always be Int32, so it's * 4.
+// We need one for the operation status (performing / performed) and one for the result (success / failure).
+const WORKER_MESSAGING_SHARED_DATA = 2 * 4;
+const WORKER_MESSAGING_STATUS_INDEX = 0;
+const WORKER_MESSAGING_RESULT_INDEX = 1;
+let connectionsListener = null;
+let mainPortWasSetup = false;
+
 // SharedArrayBuffers can be disabled with --enable-sharedarraybuffer-per-context.
 if (isMainThread && SharedArrayBuffer !== undefined) {
   cwdCounter = new Uint32Array(new SharedArrayBuffer(4));
@@ -527,6 +543,79 @@ function eventLoopUtilization(util1, util2) {
   );
 }
 
+function setConnectionsListener(fn) {
+  if (isMainThread && !mainPortWasSetup) {
+    setupMainPort();
+    mainPortWasSetup = true;
+  }
+
+  if (typeof fn === 'undefined' || fn === null) {
+    connectionsListener = null;
+    return;
+  }
+
+  validateFunction(fn, 'fn');
+  connectionsListener = fn;
+}
+
+async function processConnectionRequest(request) {
+  const status = new Int32Array(request.memory);
+
+  try {
+    const result = await connectionsListener?.(request.sender, request.port, request.data);
+    AtomicsStore(status, WORKER_MESSAGING_RESULT_INDEX, result === true ? 0 : 1);
+  } catch (e) {
+    debug('connections listener rejected', e);
+    AtomicsStore(status, WORKER_MESSAGING_RESULT_INDEX, 2);
+  } finally {
+    AtomicsNotify(status, WORKER_MESSAGING_STATUS_INDEX, 1);
+  }
+}
+
+async function connect(target, data) {
+  if (target === threadId) {
+    throw new ERR_WORKER_SAME_THREAD();
+  }
+
+  // Create a shared array to exchange the status and the result
+  const memory = new SharedArrayBuffer(WORKER_MESSAGING_SHARED_DATA);
+  const status = new Int32Array(memory);
+  const promise = AtomicsWaitAsync(status, WORKER_MESSAGING_STATUS_INDEX, 0).value;
+
+  // Create the channel and send it to the other thread
+  const { port1, port2 } = new MessageChannel();
+  sendToWorker(target, { type: messageTypes.CONNECT, sender: threadId, port: port2, memory, data }, [port2]);
+
+  // Wait for the response
+  await promise;
+
+  if (status[WORKER_MESSAGING_RESULT_INDEX] === 1) {
+    port1.close();
+    port2.close();
+    throw new ERR_WORKER_CONNECTION_REFUSED();
+  }
+
+  return port1;
+}
+
+function setupMainPort() {
+  const { port1, port2 } = new MessageChannel();
+  setMainPort(port2);
+
+  // Set message management
+  port1.on('message', (message) => {
+    if (message.type === messageTypes.CONNECT) {
+      processConnectionRequest(message);
+    } else {
+      assert(message.type === messageTypes.CONNECT, `Unknown worker message type ${message.type}`);
+    }
+  });
+
+  // Never block the process on this channel
+  port1.unref();
+  port2.unref();
+}
+
 module.exports = {
   ownsProcessState,
   kIsOnline,
@@ -537,6 +626,10 @@ module.exports = {
   setEnvironmentData,
   getEnvironmentData,
   assignEnvironmentData,
+  setConnectionsListener,
+  setupMainPort,
+  processConnectionRequest,
+  connect,
   threadId,
   InternalWorker,
   Worker,

lib/internal/worker/io.js

@@ -83,6 +83,7 @@ const messageTypes = {
   STDIO_PAYLOAD: 'stdioPayload',
   STDIO_WANTS_MORE_DATA: 'stdioWantsMoreData',
   LOAD_SCRIPT: 'loadScript',
+  CONNECT: 'connect',
 };
 
 // createFastMessageEvent skips webidl argument validation when the arguments

lib/worker_threads.js

@@ -6,8 +6,10 @@ const {
   resourceLimits,
   setEnvironmentData,
   getEnvironmentData,
+  connect,
   threadId,
   Worker,
+  setConnectionsListener,
 } = require('internal/worker');
 
 const {
@@ -40,4 +42,6 @@ module.exports = {
   BroadcastChannel,
   setEnvironmentData,
   getEnvironmentData,
+  setConnectionsListener,
+  connect,
 };

src/node_errors.h

@@ -102,6 +102,7 @@ void OOMErrorHandler(const char* location, const v8::OOMDetails& details);
   V(ERR_VM_MODULE_LINK_FAILURE, Error)                                         \
   V(ERR_WASI_NOT_STARTED, Error)                                               \
   V(ERR_WORKER_INIT_FAILED, Error)                                             \
+  V(ERR_WORKER_INVALID_ID, Error)                                              \
   V(ERR_PROTO_ACCESS, Error)
 
 #define V(code, type)                                                          \

src/node_messaging.h

@@ -19,6 +19,12 @@ class MessagePort;
 
 typedef MaybeStackBuffer<v8::Local<v8::Value>, 8> TransferList;
 
+// Get a transfer list out of an array of Javascript objects.
+bool GetTransferList(Environment* env,
+                     v8::Local<v8::Context> context,
+                     v8::Local<v8::Value> transfer_list_v,
+                     TransferList* transfer_list_out);
+
 // Used to represent the in-flight structure of an object that is being
 // transferred or cloned using postMessage().
 class TransferData : public MemoryRetainer {
@@ -288,7 +294,7 @@ class MessagePort : public HandleWrap {
   // alone is often not enough, since the backing C++ MessagePort object may
   // have been deleted already. For all intents and purposes, an object with a
   // NULL pointer to the C++ MessagePort object is also detached.
-  inline bool IsDetached() const;
+  bool IsDetached() const;
 
   BaseObject::TransferMode GetTransferMode() const override;
   std::unique_ptr<TransferData> TransferForMessaging() override;