diff --git a/doc/api/errors.md b/doc/api/errors.md index 29cd9e8a4937d1..5b31cf729d3f40 100644 --- a/doc/api/errors.md +++ b/doc/api/errors.md @@ -3590,14 +3590,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. - - -### `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). - ### `ERR_WORKER_INIT_FAILED` diff --git a/doc/api/net.md b/doc/api/net.md index 81863d9b0a13c5..ec14ae5165182b 100644 --- a/doc/api/net.md +++ b/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'); diff --git a/doc/api/worker_threads.md b/doc/api/worker_threads.md index d0b4f0a830b03e..e137722674b39b 100644 --- a/doc/api/worker_threads.md +++ b/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`. diff --git a/lib/internal/errors.js b/lib/internal/errors.js index 5f4cba63a64499..c39c3a413bd722 100644 --- a/lib/internal/errors.js +++ b/lib/internal/errors.js @@ -1980,8 +1980,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, ', ')}`, diff --git a/lib/net.js b/lib/net.js index d0620449c31e7b..cfaa4242bdba5e 100644 --- a/lib/net.js +++ b/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'); @@ -1517,9 +1516,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 || @@ -2270,9 +2266,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'); } diff --git a/src/tcp_wrap.cc b/src/tcp_wrap.cc index dc3cd6553d4f9b..9bb742fb3edcbc 100644 --- a/src/tcp_wrap.cc +++ b/src/tcp_wrap.cc @@ -381,57 +381,66 @@ void TCPWrap::Open(const FunctionCallbackInfo& 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 TCPWrap::TransferForMessaging() { -#ifdef _WIN32 - return {}; -#else CHECK_NE(GetTransferMode(), TransferMode::kDisallowCloneAndTransfer); uv_os_fd_t fd; if (uv_fileno(reinterpret_cast(&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(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(-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(dup_fd, type); -#endif + return std::make_unique(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(-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 TCPWrap::TransferData::Deserialize( @@ -454,10 +463,14 @@ BaseObjectPtr TCPWrap::TransferData::Deserialize( TCPWrap* wrap = BaseObject::Unwrap(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(-1); +#else + wrap->set_fd(socket_); + socket_ = -1; +#endif return BaseObjectPtr(wrap); } diff --git a/src/tcp_wrap.h b/src/tcp_wrap.h index d1bcb528157479..4802b62727a264 100644 --- a/src/tcp_wrap.h +++ b/src/tcp_wrap.h @@ -62,11 +62,10 @@ class TCPWrap : public ConnectionWrap { } } - // 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 TransferForMessaging() override; @@ -75,7 +74,8 @@ class TCPWrap : public ConnectionWrap { 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 Deserialize( @@ -88,7 +88,7 @@ class TCPWrap : public ConnectionWrap { SET_SELF_SIZE(TransferData) private: - int fd_; + uv_os_sock_t socket_; SocketType type_; }; diff --git a/test/parallel/test-net-server-transfer-worker.js b/test/parallel/test-net-server-transfer-worker.js index 6a8fc6a770b5c6..18a52bc1a0c561 100644 --- a/test/parallel/test-net-server-transfer-worker.js +++ b/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 { diff --git a/test/parallel/test-net-socket-transfer-worker-http.js b/test/parallel/test-net-socket-transfer-worker-http.js index 23a3ec1790c808..86245cfeeea19a 100644 --- a/test/parallel/test-net-socket-transfer-worker-http.js +++ b/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'); diff --git a/test/parallel/test-net-socket-transfer-worker.js b/test/parallel/test-net-socket-transfer-worker.js index fe512ee3723453..32a07471fc6162 100644 --- a/test/parallel/test-net-socket-transfer-worker.js +++ b/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 { diff --git a/test/parallel/test-net-transfer-guards.js b/test/parallel/test-net-transfer-guards.js index 4724913b9f3352..0879ec9da81a8e 100644 --- a/test/parallel/test-net-transfer-guards.js +++ b/test/parallel/test-net-transfer-guards.js @@ -6,11 +6,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 { MessageChannel } = require('worker_threads');