net: support TCP handle transfer on Windows

Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #64460
Fixes: #64456
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Stefan Stojanovic <stefan.stojanovic@janeasystems.com>
Reviewed-By: Marco Ippolito <marcoippolito54@gmail.com>

Author: mcollina

doc/api/errors.md

@@ -3577,14 +3577,6 @@ 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.
 
-<a id="ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED"></a>
-
-### `ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED`
-
-An attempt was made to transfer a `net.Socket` or `net.Server` to another thread
-on a platform where moving the underlying handle between event loops is not
-supported (currently Windows).
-
 <a id="ERR_WORKER_INIT_FAILED"></a>
 
 ### `ERR_WORKER_INIT_FAILED`

doc/api/net.md

@@ -770,9 +770,7 @@ threads.
 The socket must be a freshly accepted or created TCP connection: it must still
 be attached to a live handle, must not be connecting or destroyed, and must not
 have started reading or have buffered data. Otherwise `postMessage()` throws
-`ERR_WORKER_HANDLE_NOT_TRANSFERABLE`. Only TCP sockets are supported, and only
-on Unix-like platforms; on Windows `postMessage()` throws
-`ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED`.
+`ERR_WORKER_HANDLE_NOT_TRANSFERABLE`. Only TCP sockets are supported.
 
 ```cjs
 const net = require('node:net');

doc/api/worker_threads.md

@@ -1249,8 +1249,7 @@ 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.
-Only TCP handles are supported, and only on Unix-like platforms; on Windows
-`postMessage()` throws `ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED`.
+Only TCP handles are supported.
 
 If `value` contains {SharedArrayBuffer} instances, those are accessible
 from either thread. They cannot be listed in `transferList`.

lib/internal/errors.js

@@ -1978,8 +1978,6 @@ E('ERR_WORKER_HANDLE_NOT_TRANSFERABLE',
   '%s cannot be transferred in its current state; it must be a freshly ' +
   'created or accepted handle that has not started reading and has no ' +
   'pending writes', Error);
-E('ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED',
-  'Transferring a %s to another thread is not supported on this platform', Error);
 E('ERR_WORKER_INIT_FAILED', 'Worker initialization failure: %s', Error);
 E('ERR_WORKER_INVALID_EXEC_ARGV', (errors, msg = 'invalid execArgv flags') =>
   `Initiated Worker with ${msg}: ${ArrayPrototypeJoin(errors, ', ')}`,

lib/net.js

@@ -120,7 +120,6 @@ const {
     ERR_SOCKET_CONNECTION_TIMEOUT,
     ERR_SOCKET_HANDLE_ADOPTED,
     ERR_WORKER_HANDLE_NOT_TRANSFERABLE,
-    ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED,
   },
   genericNodeError,
 } = require('internal/errors');
@@ -1600,9 +1599,6 @@ Socket.prototype[kReinitializeHandle] = function reinitializeHandle(handle) {
 // connecting or destroyed, and with no data already buffered in either
 // direction (which would otherwise be lost on the sending side).
 function assertTransferableSocket(socket) {
-  if (isWindows) {
-    throw new ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED('net.Socket');
-  }
   const handle = socket._handle;
   if (handle == null || !(handle instanceof TCP) ||
       socket.destroyed || socket.connecting ||
@@ -2364,9 +2360,6 @@ Server.prototype._listen2 = setupListenHandle;  // legacy alias
 // underlying listening socket (and its pending accept queue) to that thread's
 // event loop. Only a server bound to a live TCP handle can be transferred.
 function assertTransferableServer(server) {
-  if (isWindows) {
-    throw new ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED('net.Server');
-  }
   if (server._handle == null || !(server._handle instanceof TCP)) {
     throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Server');
   }

src/tcp_wrap.cc

@@ -381,57 +381,66 @@ void TCPWrap::Open(const FunctionCallbackInfo<Value>& args) {
 }
 
 BaseObject::TransferMode TCPWrap::GetTransferMode() const {
-#ifdef _WIN32
-  // Re-adopting a socket into another thread's event loop requires
-  // re-associating it with that loop's IOCP, which needs same-process
-  // WSADuplicateSocket support that is not wired up yet. The JS net layer
-  // throws a clearer error before reaching here; this is the backstop for the
-  // low-level `socket._handle` transfer path.
-  return TransferMode::kDisallowCloneAndTransfer;
-#else
   // Only a live handle that is not already being torn down can be transferred.
   // Higher-level guards (no buffered reads, no pending writes) are enforced by
   // the JS net.Socket/net.Server layer before a handle reaches here.
   if (!HandleWrap::IsAlive(this) || IsHandleClosing())
     return TransferMode::kDisallowCloneAndTransfer;
   return TransferMode::kTransferable;
-#endif
 }
 
 std::unique_ptr<worker::TransferData> TCPWrap::TransferForMessaging() {
-#ifdef _WIN32
-  return {};
-#else
   CHECK_NE(GetTransferMode(), TransferMode::kDisallowCloneAndTransfer);
 
   uv_os_fd_t fd;
   if (uv_fileno(reinterpret_cast<uv_handle_t*>(&handle_), &fd) != 0) return {};
 
-  // dup() the descriptor so the receiving event loop owns an independent
-  // reference to the same socket. We then close the source handle, which
-  // renders it unusable on this side (true transfer semantics) while the dup
-  // keeps the underlying socket alive for the destination thread.
-  int dup_fd = dup(fd);
-  if (dup_fd < 0) return {};
+#ifdef _WIN32
+  // A socket that is already associated with an IOCP cannot be associated with
+  // another one. Create a same-process duplicate that is not associated with
+  // any IOCP yet; uv_tcp_open() will associate it with the receiving loop.
+  WSAPROTOCOL_INFOW protocol_info;
+  uv_os_sock_t source_socket = reinterpret_cast<uv_os_sock_t>(fd);
+  if (WSADuplicateSocketW(
+          source_socket, GetCurrentProcessId(), &protocol_info) != 0) {
+    return {};
+  }
+  uv_os_sock_t duplicate = WSASocketW(FROM_PROTOCOL_INFO,
+                                      FROM_PROTOCOL_INFO,
+                                      FROM_PROTOCOL_INFO,
+                                      &protocol_info,
+                                      0,
+                                      WSA_FLAG_OVERLAPPED);
+  if (duplicate == static_cast<uv_os_sock_t>(-1)) return {};
+#else
+  // Unix threads share the descriptor table, so dup() creates an independent
+  // reference to the same socket for the receiving event loop.
+  uv_os_sock_t duplicate = dup(fd);
+  if (duplicate < 0) return {};
+#endif
 
   SocketType type =
       provider_type() == ProviderType::PROVIDER_TCPSERVERWRAP ? SERVER : SOCKET;
 
-  // Stop watching the fd and tear down the source handle.
+  // Stop watching the original socket and tear down the source handle. The
+  // duplicate keeps the underlying socket alive until the destination adopts
+  // it, or until TransferData is destroyed if the message is not delivered.
   Close();
 
-  return std::make_unique<TransferData>(dup_fd, type);
-#endif
+  return std::make_unique<TransferData>(duplicate, type);
 }
 
 TCPWrap::TransferData::~TransferData() {
-  // Only reached if the message was never delivered (e.g. the destination port
-  // closed in flight); close the dup'd fd so it is not leaked.
-  if (fd_ >= 0) {
+#ifdef _WIN32
+  if (socket_ != static_cast<uv_os_sock_t>(-1))
+    CHECK_EQ(0, closesocket(socket_));
+#else
+  if (socket_ >= 0) {
     uv_fs_t req;
-    CHECK_EQ(0, uv_fs_close(nullptr, &req, fd_, nullptr));
+    CHECK_EQ(0, uv_fs_close(nullptr, &req, socket_, nullptr));
     uv_fs_req_cleanup(&req);
   }
+#endif
 }
 
 BaseObjectPtr<BaseObject> TCPWrap::TransferData::Deserialize(
@@ -454,10 +463,14 @@ BaseObjectPtr<BaseObject> TCPWrap::TransferData::Deserialize(
   TCPWrap* wrap = BaseObject::Unwrap<TCPWrap>(obj);
   if (wrap == nullptr) return {};
 
-  if (uv_tcp_open(&wrap->handle_, fd_) != 0) return {};
+  if (uv_tcp_open(&wrap->handle_, socket_) != 0) return {};
 
-  wrap->set_fd(fd_);
-  fd_ = -1;  // Ownership has been handed to the new handle.
+#ifdef _WIN32
+  socket_ = static_cast<uv_os_sock_t>(-1);
+#else
+  wrap->set_fd(socket_);
+  socket_ = -1;
+#endif
   return BaseObjectPtr<BaseObject>(wrap);
 }
 

src/tcp_wrap.h

@@ -62,11 +62,10 @@ class TCPWrap : public ConnectionWrap<TCPWrap, uv_tcp_t> {
     }
   }
 
-  // Transfer the underlying socket to another thread via .postMessage(). Within
-  // a single process all threads share the same file descriptor table, so the
-  // transfer dup()s the fd and re-adopts it (uv_tcp_open) in the receiving
-  // event loop. This is the building block for distributing listening sockets
-  // and accepted connections across worker_threads.
+  // Transfer the underlying socket to another thread via .postMessage(). The
+  // transfer duplicates the socket and re-adopts it (uv_tcp_open) in the
+  // receiving event loop. This is the building block for distributing
+  // listening sockets and accepted connections across worker_threads.
   BaseObject::TransferMode GetTransferMode() const override;
   std::unique_ptr<worker::TransferData> TransferForMessaging() override;
 
@@ -75,7 +74,8 @@ class TCPWrap : public ConnectionWrap<TCPWrap, uv_tcp_t> {
 
   class TransferData : public worker::TransferData {
    public:
-    explicit TransferData(int fd, SocketType type) : fd_(fd), type_(type) {}
+    explicit TransferData(uv_os_sock_t socket, SocketType type)
+        : socket_(socket), type_(type) {}
     ~TransferData() override;
 
     BaseObjectPtr<BaseObject> Deserialize(
@@ -88,7 +88,7 @@ class TCPWrap : public ConnectionWrap<TCPWrap, uv_tcp_t> {
     SET_SELF_SIZE(TransferData)
 
    private:
-    int fd_;
+    uv_os_sock_t socket_;
     SocketType type_;
   };
 

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

@@ -7,11 +7,6 @@
 
 const common = require('../common');
 
-if (common.isWindows) {
-  common.skip('transferring TCP handles between threads is not supported on ' +
-              'Windows yet');
-}
-
 const assert = require('assert');
 const net = require('net');
 const {

test/parallel/test-net-socket-transfer-worker-http.js

@@ -7,11 +7,6 @@
 
 const common = require('../common');
 
-if (common.isWindows) {
-  common.skip('transferring TCP handles between threads is not supported on ' +
-              'Windows yet');
-}
-
 const assert = require('assert');
 const net = require('net');
 const http = require('http');

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

@@ -7,11 +7,6 @@
 
 const common = require('../common');
 
-if (common.isWindows) {
-  common.skip('transferring TCP handles between threads is not supported on ' +
-              'Windows yet');
-}
-
 const assert = require('assert');
 const net = require('net');
 const {