net: support transferring net.BoundSocket to other threads

Signed-off-by: Guy Bedford <guybedford@gmail.com>

Author: guybedford

doc/api/errors.md

@@ -3004,8 +3004,9 @@ A call was made and the UDP subsystem was not running.
 ### `ERR_SOCKET_HANDLE_ADOPTED`
 
 An operation was attempted on a [`BoundSocket`][] that had already been adopted
-by a [`net.Server`][] or [`net.Socket`][]. Once a bound socket is adopted, its
-`address()` and `close()` methods can no longer be used.
+by a [`net.Server`][] or [`net.Socket`][], or transferred to another thread.
+Once a bound socket is adopted or transferred, its `address()` and `close()`
+methods can no longer be used.
 
 <a id="ERR_SOURCE_MAP_CORRUPT"></a>
 
@@ -3573,9 +3574,10 @@ The `Response` that has been passed to `WebAssembly.compileStreaming` or to
 
 ### `ERR_WORKER_HANDLE_NOT_TRANSFERABLE`
 
-An attempt was made to transfer a `net.Socket` or `net.Server` to another thread
-via a `worker_threads` `postMessage()` call while it was not in a transferable
-state, for example because it had already started reading or had buffered data.
+An attempt was made to transfer a `net.Socket`, `net.Server` or
+`net.BoundSocket` to another thread via a `worker_threads` `postMessage()` call
+while it was not in a transferable state, for example because it had already
+started reading, had buffered data, or had already been adopted.
 
 <a id="ERR_WORKER_INIT_FAILED"></a>
 

doc/api/net.md

@@ -789,6 +789,13 @@ server.listen(8000);
 A listening [`net.Server`][] can be transferred the same way, which moves the
 listening socket itself (and its pending accept queue) to the receiving thread.
 
+An un-adopted TCP [`BoundSocket`][] can also be transferred, which moves the
+bound (but not yet listening or connected) socket. This allows a port to be
+reserved synchronously on one thread and adopted by a server or outgoing
+connection on another. Pipe binds are not transferable. After the transfer, the
+source `BoundSocket` behaves as if it had been adopted: `address()`, `fd()` and
+`close()` throw [`ERR_SOCKET_HANDLE_ADOPTED`][].
+
 ### `new net.Socket([options])`
 
 <!-- YAML
@@ -1718,6 +1725,10 @@ file system entry; abstract and TCP binds have none to remove.
 When a pipe `BoundSocket` bound to a source `path` is adopted as a client, that
 path is reported as the socket's `localAddress` once it connects.
 
+An un-adopted TCP `BoundSocket` can be moved to another thread by listing it in
+the `transferList` of a [`worker_threads`][] `postMessage()` call, see
+[Transferring TCP handles to other threads][]. Pipe binds are not transferable.
+
 When an adopted `BoundSocket` connects to a numeric IP literal, `connect(2)` is
 issued synchronously, so [`socket.localAddress`][] is resolved once
 [`socket.connect()`][] returns. Connection failures are still reported via a

doc/api/worker_threads.md

@@ -1238,7 +1238,7 @@ port2.postMessage(circularData);
 ```
 
 `transferList` may be a list of {ArrayBuffer}, [`MessagePort`][],
-[`FileHandle`][], {net.Server}, and {net.Socket} objects.
+[`FileHandle`][], {net.Server}, {net.Socket}, and {net.BoundSocket} objects.
 After transferring, they are not usable on the sending side of the channel
 anymore (even if they are not contained in `value`).
 
@@ -1249,6 +1249,8 @@ freshly accepted or created TCP connection that has not yet started reading and
 has no buffered data, otherwise `postMessage()` throws
 `ERR_WORKER_HANDLE_NOT_TRANSFERABLE`. This makes it possible to accept
 connections on one thread and distribute them across a pool of worker threads.
+Transferring a {net.BoundSocket} moves an un-adopted pre-bound socket, so a
+port can be reserved synchronously on one thread and adopted on another.
 Only TCP handles are supported.
 
 If `value` contains {SharedArrayBuffer} instances, those are accessible

lib/net.js

@@ -388,6 +388,10 @@ const kBoundPath = Symbol('kBoundPath');
 
 const isLinux = process.platform === 'linux';
 
+// Internal: construct an empty BoundSocket shell during postMessage()
+// deserialization; the transferred handle is installed by [kDeserialize].
+const kBoundSocketDeserialize = Symbol('kBoundSocketDeserialize');
+
 // A role-neutral wrapper over a synchronously bound libuv handle: bound to a
 // local address (a numeric IP literal for TCP, or a filesystem/abstract path
 // for a unix-domain socket via { path }) but neither listening nor connecting
@@ -396,11 +400,20 @@ const isLinux = process.platform === 'linux';
 // handle must be closed by the caller. bind(2) is non-blocking, so binding
 // happens inline and errors throw synchronously. No DNS is performed.
 class BoundSocket {
-  #handle;
+  #handle = null;
   #address = {};
   #path;
 
   constructor(options = kEmptyObject) {
+    // An un-adopted BoundSocket can be moved to another thread by listing it
+    // in the transferList of a worker_threads postMessage() call. See
+    // [kTransfer]().
+    markTransferMode(this, false, true);
+
+    if (options === kBoundSocketDeserialize) {
+      return;
+    }
+
     validateObject(options, 'options');
 
     if (options.path !== undefined) {
@@ -544,7 +557,56 @@ class BoundSocket {
   get isPipe() {
     return this.#path !== undefined;
   }
+
+  // A BoundSocket can be transferred only while it still owns its handle,
+  // i.e. before it has been adopted, closed or already transferred. Only TCP
+  // binds are transferable; pipe handles cannot move between event loops.
+  #assertTransferable() {
+    if (this.#handle === null || !(this.#handle instanceof TCP)) {
+      throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.BoundSocket');
+    }
+  }
+
+  [kTransferList]() {
+    this.#assertTransferable();
+    return [this.#handle];
+  }
+
+  [kTransfer]() {
+    this.#assertTransferable();
+    const handle = this.#handle;
+    // Detach the handle; the messaging layer takes ownership of it via
+    // TCPWrap::TransferForMessaging(). Further use on the sending side throws
+    // ERR_SOCKET_HANDLE_ADOPTED, as after adoption.
+    this.#handle = null;
+    return {
+      data: { handle },
+      deserializeInfo: 'net:_TransferredBoundSocket',
+    };
+  }
+
+  [kDeserialize](data) {
+    const handle = data?.handle;
+    if (handle == null || !(handle instanceof TCP)) {
+      throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.BoundSocket');
+    }
+    // Re-derive the bound address from the transferred handle rather than
+    // trusting serialized state.
+    const err = handle.getsockname(this.#address);
+    if (err) {
+      handle.close();
+      throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.BoundSocket');
+    }
+    this.#handle = handle;
+  }
+}
+
+// Deserialization target for a transferred BoundSocket: constructs an empty
+// shell without binding a new socket. Internal, not part of the public API.
+function _TransferredBoundSocket() {
+  return new BoundSocket(kBoundSocketDeserialize);
 }
+_TransferredBoundSocket.prototype = BoundSocket.prototype;
 
 function Socket(options) {
   if (!(this instanceof Socket)) return new Socket(options);
@@ -2933,6 +2995,7 @@ Server.prototype.unref = function() {
 module.exports = {
   _createServerHandle: createServerHandle,
   _normalizeArgs: normalizeArgs,
+  _TransferredBoundSocket,
   get BlockList() {
     BlockList ??= require('internal/blocklist').BlockList;
     return BlockList;

test/parallel/test-net-boundsocket-transfer-worker.js

@@ -0,0 +1,64 @@
+'use strict';
+
+// This test verifies that an un-adopted net.BoundSocket can be transferred to
+// a worker thread via worker_threads postMessage()'s transferList. The parent
+// thread binds synchronously to reserve the port, then hands the bound socket
+// off; the worker adopts it via server.listen() and accepts connections.
+
+const common = require('../common');
+
+const assert = require('assert');
+const net = require('net');
+const {
+  Worker,
+  parentPort,
+  threadId,
+  workerData,
+} = require('worker_threads');
+
+if (workerData?.role === 'server') {
+  parentPort.on('message', common.mustCall(({ bound, port }) => {
+    assert.ok(bound instanceof net.BoundSocket);
+    // The bound address survives the transfer.
+    assert.strictEqual(bound.address().port, port);
+
+    const server = net.createServer((socket) => {
+      socket.end(`served-by:${threadId}`);
+    });
+    server.listen(bound, common.mustCall(() => {
+      assert.strictEqual(server.address().port, port);
+      // Adoption consumed the bound socket.
+      assert.throws(() => bound.address(), {
+        code: 'ERR_SOCKET_HANDLE_ADOPTED',
+      });
+      parentPort.postMessage('listening');
+    }));
+  }));
+  return;
+}
+
+const worker = new Worker(__filename, { workerData: { role: 'server' } });
+
+const bound = new net.BoundSocket({ host: '127.0.0.1', port: 0 });
+const { port } = bound.address();
+
+// Move the bound socket to the worker thread.
+worker.postMessage({ bound, port }, [bound]);
+
+// The source is left in the adopted state; further use throws.
+assert.throws(() => bound.address(), { code: 'ERR_SOCKET_HANDLE_ADOPTED' });
+assert.throws(() => bound.fd(), { code: 'ERR_SOCKET_HANDLE_ADOPTED' });
+assert.throws(() => bound.close(), { code: 'ERR_SOCKET_HANDLE_ADOPTED' });
+
+worker.on('message', common.mustCall((msg) => {
+  assert.strictEqual(msg, 'listening');
+  const client = net.connect(port, '127.0.0.1');
+  client.setEncoding('utf8');
+  let response = '';
+  client.on('data', (chunk) => { response += chunk; });
+  client.on('end', common.mustCall(() => {
+    assert.match(response, /^served-by:\d+$/);
+    assert.notStrictEqual(response, `served-by:${threadId}`);
+    worker.terminate();
+  }));
+}));

test/parallel/test-net-transfer-guards.js

@@ -1,17 +1,31 @@
 'use strict';
 
-// This test verifies the guards that reject transferring a net.Socket or
-// net.Server that is not in a clean, movable state. Serialization is triggered
-// with a MessageChannel, so no worker thread is required.
+// This test verifies the guards that reject transferring a net.Socket,
+// net.Server or net.BoundSocket that is not in a clean, movable state.
+// Serialization is triggered with a MessageChannel, so no worker thread is
+// required.
 
 const common = require('../common');
 
 const assert = require('assert');
 const net = require('net');
 const { MessageChannel } = require('worker_threads');
+const tmpdir = require('../common/tmpdir');
+
+tmpdir.refresh();
 
 const { port1 } = new MessageChannel();
 
+// A pipe BoundSocket is not transferable; only TCP binds can move between
+// event loops.
+{
+  const bound = new net.BoundSocket({ path: common.PIPE });
+  assert.throws(() => port1.postMessage({ bound }, [bound]), {
+    code: 'ERR_WORKER_HANDLE_NOT_TRANSFERABLE',
+  });
+  bound.close();
+}
+
 // A Server that is not listening (no handle) cannot be transferred.
 {
   const server = net.createServer();
@@ -20,6 +34,25 @@ const { port1 } = new MessageChannel();
   });
 }
 
+// A closed BoundSocket no longer owns a handle and cannot be transferred.
+{
+  const bound = new net.BoundSocket({ host: '127.0.0.1', port: 0 });
+  bound.close();
+  assert.throws(() => port1.postMessage({ bound }, [bound]), {
+    code: 'ERR_WORKER_HANDLE_NOT_TRANSFERABLE',
+  });
+}
+
+// An adopted BoundSocket no longer owns a handle and cannot be transferred.
+{
+  const bound = new net.BoundSocket({ host: '127.0.0.1', port: 0 });
+  const socket = new net.Socket({ handle: bound });
+  assert.throws(() => port1.postMessage({ bound }, [bound]), {
+    code: 'ERR_WORKER_HANDLE_NOT_TRANSFERABLE',
+  });
+  socket.destroy();
+}
+
 // A Socket that has already consumed data cannot be transferred, because that
 // buffered data would be lost on the sending side.
 {