Skip to content
Open
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
8 changes: 0 additions & 8 deletions doc/api/errors.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

<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`
Expand Down
4 changes: 1 addition & 3 deletions doc/api/net.md
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down
3 changes: 1 addition & 2 deletions doc/api/worker_threads.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
Expand Down
2 changes: 0 additions & 2 deletions lib/internal/errors.js
Original file line number Diff line number Diff line change
Expand Up @@ -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, ', ')}`,
Expand Down
7 changes: 0 additions & 7 deletions lib/net.js
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down Expand Up @@ -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 ||
Expand Down Expand Up @@ -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');
}
Expand Down
69 changes: 41 additions & 28 deletions src/tcp_wrap.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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);
}

Expand Down
14 changes: 7 additions & 7 deletions src/tcp_wrap.h
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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(
Expand All @@ -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_;
};

Expand Down
5 changes: 0 additions & 5 deletions test/parallel/test-net-server-transfer-worker.js
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
5 changes: 0 additions & 5 deletions test/parallel/test-net-socket-transfer-worker-http.js
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down
5 changes: 0 additions & 5 deletions test/parallel/test-net-socket-transfer-worker.js
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
5 changes: 0 additions & 5 deletions test/parallel/test-net-transfer-guards.js
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down
Loading