@@ -119,9 +119,17 @@ const { |
119 | 119 | ERR_SOCKET_CLOSED_BEFORE_CONNECTION, |
120 | 120 | ERR_SOCKET_CONNECTION_TIMEOUT, |
121 | 121 | ERR_SOCKET_HANDLE_ADOPTED, |
| 122 | + ERR_WORKER_HANDLE_NOT_TRANSFERABLE, |
| 123 | + ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED, |
122 | 124 | }, |
123 | 125 | genericNodeError, |
124 | 126 | } = require('internal/errors'); |
| 127 | +const { |
| 128 | + markTransferMode, |
| 129 | + kDeserialize, |
| 130 | + kTransfer, |
| 131 | + kTransferList, |
| 132 | +} = require('internal/worker/js_transferable'); |
125 | 133 | const { isUint8Array } = require('internal/util/types'); |
126 | 134 | const { queueMicrotask } = require('internal/process/task_queues'); |
127 | 135 | const { |
@@ -484,6 +492,9 @@ class BoundSocket { |
484 | 492 |
|
485 | 493 | function Socket(options) { |
486 | 494 | if (!(this instanceof Socket)) return new Socket(options); |
| 495 | + // A connected TCP Socket can be moved to another thread by listing it in the |
| 496 | + // transferList of a worker_threads postMessage() call. See [kTransfer](). |
| 497 | + markTransferMode(this, false, true); |
487 | 498 | if (options?.objectMode) { |
488 | 499 | throw new ERR_INVALID_ARG_VALUE( |
489 | 500 | 'options.objectMode', |
@@ -1501,6 +1512,58 @@ Socket.prototype[kReinitializeHandle] = function reinitializeHandle(handle) { |
1501 | 1512 | initSocketHandle(this); |
1502 | 1513 | }; |
1503 | 1514 |
|
| 1515 | +// A Socket can be transferred to another thread only while it is a freshly |
| 1516 | +// accepted/created TCP connection: still attached to a live handle, not |
| 1517 | +// connecting or destroyed, and with no data already buffered in either |
| 1518 | +// direction (which would otherwise be lost on the sending side). |
| 1519 | +function assertTransferableSocket(socket) { |
| 1520 | + if (isWindows) { |
| 1521 | + throw new ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED('net.Socket'); |
| 1522 | + } |
| 1523 | + const handle = socket._handle; |
| 1524 | + if (handle == null || !(handle instanceof TCP) || |
| 1525 | + socket.destroyed || socket.connecting || |
| 1526 | + socket.bytesRead > 0 || socket.bytesWritten > 0 || |
| 1527 | + socket.readableLength > 0 || socket.writableLength > 0 || |
| 1528 | + socket.readableEncoding != null) { |
| 1529 | + throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Socket'); |
| 1530 | + } |
| 1531 | +} |
| 1532 | + |
| 1533 | +Socket.prototype[kTransferList] = function() { |
| 1534 | + assertTransferableSocket(this); |
| 1535 | + return [this._handle]; |
| 1536 | +}; |
| 1537 | + |
| 1538 | +Socket.prototype[kTransfer] = function() { |
| 1539 | + assertTransferableSocket(this); |
| 1540 | + const handle = this._handle; |
| 1541 | + const data = { |
| 1542 | + handle, |
| 1543 | + allowHalfOpen: this.allowHalfOpen, |
| 1544 | + }; |
| 1545 | + // Detach the handle from this source socket; the messaging layer takes |
| 1546 | + // ownership of it via TCPWrap::TransferForMessaging(). Destroy the source so |
| 1547 | + // any further use on the sending side fails cleanly (the socket is now owned |
| 1548 | + // by the receiving thread) instead of silently dropping data. |
| 1549 | + this._handle = null; |
| 1550 | + this.destroy(); |
| 1551 | + return { |
| 1552 | + data, |
| 1553 | + deserializeInfo: 'net:Socket', |
| 1554 | + }; |
| 1555 | +}; |
| 1556 | + |
| 1557 | +Socket.prototype[kDeserialize] = function(data) { |
| 1558 | + const handle = data?.handle; |
| 1559 | + if (handle == null || !(handle instanceof TCP)) { |
| 1560 | + throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Socket'); |
| 1561 | + } |
| 1562 | + this.allowHalfOpen = Boolean(data.allowHalfOpen); |
| 1563 | + this[kReinitializeHandle](handle); |
| 1564 | + this.readable = this.writable = true; |
| 1565 | +}; |
| 1566 | + |
1504 | 1567 | function socketToDnsFamily(family) { |
1505 | 1568 | switch (family) { |
1506 | 1569 | case 'IPv4': |
@@ -1993,6 +2056,10 @@ function Server(options, connectionListener) { |
1993 | 2056 |
|
1994 | 2057 | EventEmitter.call(this); |
1995 | 2058 |
|
| 2059 | + // A listening TCP Server can be moved to another thread by listing it in the |
| 2060 | + // transferList of a worker_threads postMessage() call. See [kTransfer](). |
| 2061 | + markTransferMode(this, false, true); |
| 2062 | + |
1996 | 2063 | if (typeof options === 'function') { |
1997 | 2064 | connectionListener = options; |
1998 | 2065 | options = kEmptyObject; |
@@ -2199,6 +2266,61 @@ function setupListenHandle(address, port, addressType, backlog, fd, flags) { |
2199 | 2266 |
|
2200 | 2267 | Server.prototype._listen2 = setupListenHandle; // legacy alias |
2201 | 2268 |
|
| 2269 | +// A listening TCP Server can be transferred to another thread, which moves the |
| 2270 | +// underlying listening socket (and its pending accept queue) to that thread's |
| 2271 | +// event loop. Only a server bound to a live TCP handle can be transferred. |
| 2272 | +function assertTransferableServer(server) { |
| 2273 | + if (isWindows) { |
| 2274 | + throw new ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED('net.Server'); |
| 2275 | + } |
| 2276 | + if (server._handle == null || !(server._handle instanceof TCP)) { |
| 2277 | + throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Server'); |
| 2278 | + } |
| 2279 | +} |
| 2280 | + |
| 2281 | +Server.prototype[kTransferList] = function() { |
| 2282 | + assertTransferableServer(this); |
| 2283 | + return [this._handle]; |
| 2284 | +}; |
| 2285 | + |
| 2286 | +Server.prototype[kTransfer] = function() { |
| 2287 | + assertTransferableServer(this); |
| 2288 | + const handle = this._handle; |
| 2289 | + const data = { |
| 2290 | + handle, |
| 2291 | + // Construction-time options that govern accepted sockets, so the receiving |
| 2292 | + // server reproduces the same behaviour. |
| 2293 | + allowHalfOpen: this.allowHalfOpen, |
| 2294 | + pauseOnConnect: this.pauseOnConnect, |
| 2295 | + noDelay: this.noDelay, |
| 2296 | + keepAlive: this.keepAlive, |
| 2297 | + keepAliveInitialDelay: this.keepAliveInitialDelay * 1000, |
| 2298 | + highWaterMark: this.highWaterMark, |
| 2299 | + }; |
| 2300 | + // Detach so the source server no longer references the handle being moved. |
| 2301 | + this._handle = null; |
| 2302 | + return { |
| 2303 | + data, |
| 2304 | + deserializeInfo: 'net:Server', |
| 2305 | + }; |
| 2306 | +}; |
| 2307 | + |
| 2308 | +Server.prototype[kDeserialize] = function(data) { |
| 2309 | + const { handle, ...options } = data ?? {}; |
| 2310 | + if (handle == null || !(handle instanceof TCP)) { |
| 2311 | + throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Server'); |
| 2312 | + } |
| 2313 | + this.allowHalfOpen = Boolean(options.allowHalfOpen); |
| 2314 | + this.pauseOnConnect = Boolean(options.pauseOnConnect); |
| 2315 | + this.noDelay = Boolean(options.noDelay); |
| 2316 | + this.keepAlive = Boolean(options.keepAlive); |
| 2317 | + this.keepAliveInitialDelay = ~~(options.keepAliveInitialDelay / 1000); |
| 2318 | + this.highWaterMark = options.highWaterMark ?? getDefaultHighWaterMark(); |
| 2319 | + // Adopt the transferred listening handle (mirrors the child_process server |
| 2320 | + // hand-off path), which re-arms accept() on this thread's event loop. |
| 2321 | + this.listen(handle); |
| 2322 | +}; |
| 2323 | + |
2202 | 2324 | function emitErrorNT(self, err) { |
2203 | 2325 | self.emit('error', err); |
2204 | 2326 | } |
|