net: support sending net.BoundSocket to threads and child processes · nodejs/node@080e76b · GitHub
Skip to content

Commit 080e76b

Browse files
guybedfordaduh95
guybedford
authored and
aduh95
committed
net: support sending net.BoundSocket to threads and child processes
This adds support for transferring net.BoundSocket instances to other threads via the worker_threads postMessage() transfer list, and for sending them to child processes as the sendHandle argument of subprocess.send(), following on from the BoundSocket introduction. A BoundSocket reserves a port synchronously at construction time. Making it transferable means a port can be reserved on one thread or process and the bound (but not yet listening or connected) TCP handle handed off to another to listen or connect on, without racing on the bind. For threads, BoundSocket implements kTransfer/kTransferList/ kDeserialize, moving the underlying TCP handle with the same mechanism used for net.Socket and net.Server transfer. For child processes, the handleConversion entry reuses the same transfer protocol on the sending side and the same _TransferredBoundSocket deserialization path on the receiving side; the underlying transport is that of cluster's shared-handle scheduling: SCM_RIGHTS on Unix and WSADuplicateSocket on Windows, both of which carry bind state. In both cases the source instance is left in the adopted state: address(), fd() and close() throw ERR_SOCKET_HANDLE_ADOPTED. Transfer requires an un-adopted, open TCP handle, otherwise ERR_WORKER_HANDLE_NOT_TRANSFERABLE is thrown; pipe (path) binds are not transferable and throw ERR_INVALID_HANDLE_TYPE when sent over IPC. On the receiving side the local address is re-derived from the handle rather than trusted from serialized state. Signed-off-by: Guy Bedford <guybedford@gmail.com> PR-URL: #64725 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent e44669b commit 080e76b

9 files changed

Lines changed: 329 additions & 30 deletions

‎doc/api/child_process.md‎

Lines changed: 13 additions & 3 deletions

‎doc/api/errors.md‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3036,8 +3036,9 @@ A call was made and the UDP subsystem was not running.
30363036
### `ERR_SOCKET_HANDLE_ADOPTED`
30373037

30383038
An operation was attempted on a [`BoundSocket`][] that had already been adopted
3039-
by a [`net.Server`][] or [`net.Socket`][]. Once a bound socket is adopted, its
3040-
`address()` and `close()` methods can no longer be used.
3039+
by a [`net.Server`][] or [`net.Socket`][], or transferred to another thread.
3040+
Once a bound socket is adopted or transferred, its `address()` and `close()`
3041+
methods can no longer be used.
30413042

30423043
<a id="ERR_SOURCE_MAP_CORRUPT"></a>
30433044

@@ -3605,9 +3606,10 @@ The `Response` that has been passed to `WebAssembly.compileStreaming` or to
36053606

36063607
### `ERR_WORKER_HANDLE_NOT_TRANSFERABLE`
36073608

3608-
An attempt was made to transfer a `net.Socket` or `net.Server` to another thread
3609-
via a `worker_threads` `postMessage()` call while it was not in a transferable
3610-
state, for example because it had already started reading or had buffered data.
3609+
An attempt was made to transfer a `net.Socket`, `net.Server` or
3610+
`net.BoundSocket` to another thread via a `worker_threads` `postMessage()` call
3611+
while it was not in a transferable state, for example because it had already
3612+
started reading, had buffered data, or had already been adopted.
36113613

36123614
<a id="ERR_WORKER_INIT_FAILED"></a>
36133615

‎doc/api/net.md‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -944,6 +944,13 @@ server.listen(8000);
944944
A listening [`net.Server`][] can be transferred the same way, which moves the
945945
listening socket itself (and its pending accept queue) to the receiving thread.
946946

947+
An un-adopted TCP [`BoundSocket`][] can also be transferred, which moves the
948+
bound (but not yet listening or connected) socket. This allows a port to be
949+
reserved synchronously on one thread and adopted by a server or outgoing
950+
connection on another. Pipe binds are not transferable. After the transfer, the
951+
source `BoundSocket` behaves as if it had been adopted: `address()`, `fd()` and
952+
`close()` throw [`ERR_SOCKET_HANDLE_ADOPTED`][].
953+
947954
### `new net.Socket([options])`
948955

949956
<!-- YAML
@@ -1867,6 +1874,13 @@ file system entry; abstract and TCP binds have none to remove.
18671874
When a pipe `BoundSocket` bound to a source `path` is adopted as a client, that
18681875
path is reported as the socket's `localAddress` once it connects.
18691876

1877+
An un-adopted TCP `BoundSocket` can be moved to another thread by listing it in
1878+
the `transferList` of a [`worker_threads`][] `postMessage()` call, see
1879+
[Transferring TCP handles to other threads][]. It can likewise be sent to a
1880+
child process as the `sendHandle` argument of [`subprocess.send()`][]. In both
1881+
cases the source is left in the adopted state. Pipe binds cannot be moved
1882+
either way.
1883+
18701884
When an adopted `BoundSocket` connects to a numeric IP literal, `connect(2)` is
18711885
issued synchronously, so [`socket.localAddress`][] is resolved once
18721886
[`socket.connect()`][] returns. Connection failures are still reported via a
@@ -2497,6 +2511,7 @@ net.isIPv6('fhqwhgads'); // returns false
24972511
[`socket.setTimeout()`]: #socketsettimeouttimeout-callback
24982512
[`socket.setTimeout(timeout)`]: #socketsettimeouttimeout-callback
24992513
[`stream.getDefaultHighWaterMark()`]: stream.md#streamgetdefaulthighwatermarkobjectmode
2514+
[`subprocess.send()`]: child_process.md#subprocesssendmessage-sendhandle-options-callback
25002515
[`worker_threads`]: worker_threads.md
25012516
[`writable.destroy()`]: stream.md#writabledestroyerror
25022517
[`writable.destroyed`]: stream.md#writabledestroyed

‎doc/api/worker_threads.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1238,7 +1238,7 @@ port2.postMessage(circularData);
12381238
```
12391239
12401240
`transferList` may be a list of {ArrayBuffer}, [`MessagePort`][],
1241-
[`FileHandle`][], {net.Server}, and {net.Socket} objects.
1241+
[`FileHandle`][], {net.Server}, {net.Socket}, and {net.BoundSocket} objects.
12421242
After transferring, they are not usable on the sending side of the channel
12431243
anymore (even if they are not contained in `value`).
12441244
@@ -1249,6 +1249,8 @@ freshly accepted or created TCP connection that has not yet started reading and
12491249
has no buffered data, otherwise `postMessage()` throws
12501250
`ERR_WORKER_HANDLE_NOT_TRANSFERABLE`. This makes it possible to accept
12511251
connections on one thread and distribute them across a pool of worker threads.
1252+
Transferring a {net.BoundSocket} moves an un-adopted pre-bound socket, so a
1253+
port can be reserved synchronously on one thread and adopted on another.
12521254
Only TCP handles are supported.
12531255
12541256
If `value` contains {SharedArrayBuffer} instances, those are accessible

‎lib/internal/child_process.js‎

Lines changed: 48 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,10 @@ const { TCP } = internalBinding('tcp_wrap');
5454
const { TTY } = internalBinding('tty_wrap');
5555
const { UDP } = internalBinding('udp_wrap');
5656
const SocketList = require('internal/socket_list');
57+
const {
58+
kDeserialize,
59+
kTransfer,
60+
} = require('internal/worker/js_transferable');
5761
const { owner_symbol } = require('internal/async_hooks').symbols;
5862
const { convertToValidSignal } = require('internal/util');
5963
const { isArrayBufferView } = require('internal/util/types');
@@ -86,6 +90,24 @@ const kChannelHandle = Symbol('kChannelHandle');
8690
const kIsUsedAsStdio = Symbol('kIsUsedAsStdio');
8791
const kPendingMessages = Symbol('kPendingMessages');
8892

93+
// Store the handle after successfully sending it, so it can be closed when
94+
// the NODE_HANDLE_ACK is received. If the handle could not be sent, just
95+
// close it.
96+
function closeHandleAfterAck(message, handle, options, callback, target) {
97+
if (handle && !options.keepOpen) {
98+
if (target) {
99+
// There can only be one _pendingMessage as passing handles are
100+
// processed one at a time: handles are stored in _handleQueue while
101+
// waiting for the NODE_HANDLE_ACK of the current passing handle.
102+
assert(!target._pendingMessage);
103+
target._pendingMessage =
104+
{ callback, message, handle, options, retransmissions: 0 };
105+
} else {
106+
handle.close();
107+
}
108+
}
109+
}
110+
89111
// This object contain function to convert TCP objects to native handle objects
90112
// and back again.
91113
const handleConversion = {
@@ -116,6 +138,29 @@ const handleConversion = {
116138
},
117139
},
118140

141+
'net.BoundSocket': {
142+
simultaneousAccepts: true,
143+
144+
send(message, boundSocket, options) {
145+
// Pipe (path) binds cannot be sent over the IPC channel.
146+
if (boundSocket.isPipe)
147+
throw new ERR_INVALID_HANDLE_TYPE();
148+
149+
// Reuse the worker_threads transfer protocol: detaches the handle,
150+
// leaving the source in the adopted state, and throws if the bound
151+
// socket is not in a transferable state.
152+
return boundSocket[kTransfer]().data.handle;
153+
},
154+
155+
postSend: closeHandleAfterAck,
156+
157+
got(message, handle, emit) {
158+
const boundSocket = new net._TransferredBoundSocket();
159+
boundSocket[kDeserialize]({ handle });
160+
emit(boundSocket);
161+
},
162+
},
163+
119164
'net.Socket': {
120165
send(message, socket, options) {
121166
if (!socket._handle)
@@ -165,23 +210,7 @@ const handleConversion = {
165210
return handle;
166211
},
167212

168-
postSend(message, handle, options, callback, target) {
169-
// Store the handle after successfully sending it, so it can be closed
170-
// when the NODE_HANDLE_ACK is received. If the handle could not be sent,
171-
// just close it.
172-
if (handle && !options.keepOpen) {
173-
if (target) {
174-
// There can only be one _pendingMessage as passing handles are
175-
// processed one at a time: handles are stored in _handleQueue while
176-
// waiting for the NODE_HANDLE_ACK of the current passing handle.
177-
assert(!target._pendingMessage);
178-
target._pendingMessage =
179-
{ callback, message, handle, options, retransmissions: 0 };
180-
} else {
181-
handle.close();
182-
}
183-
}
184-
},
213+
postSend: closeHandleAfterAck,
185214

186215
got(message, handle, emit) {
187216
const socket = new net.Socket({
@@ -822,6 +851,8 @@ function setupChannel(target, channel, serializationMode) {
822851
message.type = 'net.Socket';
823852
} else if (handle instanceof net.Server) {
824853
message.type = 'net.Server';
854+
} else if (handle instanceof net.BoundSocket) {
855+
message.type = 'net.BoundSocket';
825856
} else if (handle instanceof TCP || handle instanceof Pipe) {
826857
message.type = 'net.Native';
827858
} else if (handle instanceof dgram.Socket) {

‎lib/net.js‎

Lines changed: 64 additions & 1 deletion

0 commit comments

Comments
 (0)