Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions doc/api/errors.md
Original file line number Diff line number Diff line change
Expand Up @@ -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>

Expand Down Expand Up @@ -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>

Expand Down
11 changes: 11 additions & 0 deletions doc/api/net.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion doc/api/worker_threads.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`).

Expand All @@ -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
Expand Down
65 changes: 64 additions & 1 deletion lib/net.js
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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) {
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -2933,6 +2995,7 @@ Server.prototype.unref = function() {
module.exports = {
_createServerHandle: createServerHandle,
_normalizeArgs: normalizeArgs,
_TransferredBoundSocket,
get BlockList() {
BlockList ??= require('internal/blocklist').BlockList;
return BlockList;
Expand Down
64 changes: 64 additions & 0 deletions test/parallel/test-net-boundsocket-transfer-worker.js
Original file line number Diff line number Diff line change
@@ -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();
}));
}));
39 changes: 36 additions & 3 deletions test/parallel/test-net-transfer-guards.js
Original file line number Diff line number Diff line change
@@ -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();
Expand All @@ -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.
{
Expand Down
Loading