net: support sending net.BoundSocket to worker threads and child processes by guybedford · Pull Request #64725 · nodejs/node · GitHub
Skip to content
Closed
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
16 changes: 13 additions & 3 deletions doc/api/child_process.md
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
15 changes: 15 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,13 @@ 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][]. It can likewise be sent to a
child process as the `sendHandle` argument of [`subprocess.send()`][]. In both
cases the source is left in the adopted state. Pipe binds cannot be moved
either way.

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 Expand Up @@ -2345,6 +2359,7 @@ net.isIPv6('fhqwhgads'); // returns false
[`socket.setTimeout()`]: #socketsettimeouttimeout-callback
[`socket.setTimeout(timeout)`]: #socketsettimeouttimeout-callback
[`stream.getDefaultHighWaterMark()`]: stream.md#streamgetdefaulthighwatermarkobjectmode
[`subprocess.send()`]: child_process.md#subprocesssendmessage-sendhandle-options-callback
[`worker_threads`]: worker_threads.md
[`writable.destroy()`]: stream.md#writabledestroyerror
[`writable.destroyed`]: stream.md#writabledestroyed
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: 48 additions & 17 deletions lib/internal/child_process.js
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,10 @@ const { TCP } = internalBinding('tcp_wrap');
const { TTY } = internalBinding('tty_wrap');
const { UDP } = internalBinding('udp_wrap');
const SocketList = require('internal/socket_list');
const {
kDeserialize,
kTransfer,
} = require('internal/worker/js_transferable');
const { owner_symbol } = require('internal/async_hooks').symbols;
const { convertToValidSignal } = require('internal/util');
const { isArrayBufferView } = require('internal/util/types');
Expand Down Expand Up @@ -86,6 +90,24 @@ const kChannelHandle = Symbol('kChannelHandle');
const kIsUsedAsStdio = Symbol('kIsUsedAsStdio');
const kPendingMessages = Symbol('kPendingMessages');

// Store the handle after successfully sending it, so it can be closed when
// the NODE_HANDLE_ACK is received. If the handle could not be sent, just
// close it.
function closeHandleAfterAck(message, handle, options, callback, target) {
if (handle && !options.keepOpen) {
if (target) {
// There can only be one _pendingMessage as passing handles are
// processed one at a time: handles are stored in _handleQueue while
// waiting for the NODE_HANDLE_ACK of the current passing handle.
assert(!target._pendingMessage);
target._pendingMessage =
{ callback, message, handle, options, retransmissions: 0 };
} else {
handle.close();
}
}
}

// This object contain function to convert TCP objects to native handle objects
// and back again.
const handleConversion = {
Expand Down Expand Up @@ -116,6 +138,29 @@ const handleConversion = {
},
},

'net.BoundSocket': {
simultaneousAccepts: true,

send(message, boundSocket, options) {
// Pipe (path) binds cannot be sent over the IPC channel.
if (boundSocket.isPipe)
throw new ERR_INVALID_HANDLE_TYPE();

// Reuse the worker_threads transfer protocol: detaches the handle,
// leaving the source in the adopted state, and throws if the bound
// socket is not in a transferable state.
return boundSocket[kTransfer]().data.handle;
},

postSend: closeHandleAfterAck,

got(message, handle, emit) {
const boundSocket = new net._TransferredBoundSocket();
boundSocket[kDeserialize]({ handle });
emit(boundSocket);
},
},

'net.Socket': {
send(message, socket, options) {
if (!socket._handle)
Expand Down Expand Up @@ -165,23 +210,7 @@ const handleConversion = {
return handle;
},

postSend(message, handle, options, callback, target) {
// Store the handle after successfully sending it, so it can be closed
// when the NODE_HANDLE_ACK is received. If the handle could not be sent,
// just close it.
if (handle && !options.keepOpen) {
if (target) {
// There can only be one _pendingMessage as passing handles are
// processed one at a time: handles are stored in _handleQueue while
// waiting for the NODE_HANDLE_ACK of the current passing handle.
assert(!target._pendingMessage);
target._pendingMessage =
{ callback, message, handle, options, retransmissions: 0 };
} else {
handle.close();
}
}
},
postSend: closeHandleAfterAck,

got(message, handle, emit) {
const socket = new net.Socket({
Expand Down Expand Up @@ -822,6 +851,8 @@ function setupChannel(target, channel, serializationMode) {
message.type = 'net.Socket';
} else if (handle instanceof net.Server) {
message.type = 'net.Server';
} else if (handle instanceof net.BoundSocket) {
message.type = 'net.BoundSocket';
} else if (handle instanceof TCP || handle instanceof Pipe) {
message.type = 'net.Native';
} else if (handle instanceof dgram.Socket) {
Expand Down
65 changes: 64 additions & 1 deletion lib/net.js
Loading
Loading