From 9b56bf38cb92c42abd776d95bdd61f96bf9d1e9a Mon Sep 17 00:00:00 2001 From: cirakg Date: Tue, 3 Nov 2026 01:15:30 +0100 Subject: network: keep fcntl modes across MSG_DONTWAIT and decompress room packets once A receive with MSG_DONTWAIT restored the blocking mode from the descriptor copy taken before the call. When fcntl(SETFL, O_NONBLOCK) ran during the receive, the fd kept reporting O_NONBLOCK while its socket blocked again. Receives now switch the mode only for MSG_DONTWAIT and read the current flags under file_descriptors_mutex at both ends, and SETFL changes the mode and the stored flags in the same critical section. OnProxyPacketReceived decompressed each room packet once per matching socket while holding file_descriptors_mutex, so every bsd call waited for it. It now copies the distinct sockets under the lock, decompresses once without it and delivers the result. Also reject fd == MAX_FD in IsFileDescriptorValid and PollImpl, which read past the table, and guard the ProxySocket blocking and receive timeout setters with packets_mutex, matching their reader. Assisted-by: Claude Opus 5.5 (investigation and implementation) --- src/core/hle/service/sockets/bsd.cpp | 92 ++++++++++++++++++------------ src/core/hle/service/sockets/bsd.h | 2 + src/core/internal_network/socket_proxy.cpp | 23 ++------ src/core/internal_network/sockets.h | 1 + 4 files changed, 65 insertions(+), 53 deletions(-) (limited to 'src/core') diff --git a/src/core/hle/service/sockets/bsd.cpp b/src/core/hle/service/sockets/bsd.cpp index 477f32626d..e6d23b7bf6 100644 --- a/src/core/hle/service/sockets/bsd.cpp +++ b/src/core/hle/service/sockets/bsd.cpp @@ -14,6 +14,7 @@ #include "common/logging.h" #include "common/socket_types.h" +#include "common/zstd_compression.h" #include "core/core.h" #include "core/hle/kernel/k_thread.h" #include "core/hle/service/ipc_helpers.h" @@ -608,7 +609,7 @@ std::pair BSD_USA::PollImpl(std::vector& write_buffer, std::span PollFD& pollfd = fds[i]; ASSERT(False(pollfd.revents)); - if (pollfd.fd > static_cast(MAX_FD) || pollfd.fd < 0) { + if (pollfd.fd >= static_cast(MAX_FD) || pollfd.fd < 0) { LOG_ERROR(Service, "File descriptor handle={} is invalid", pollfd.fd); pollfd.revents = PollEvents{}; return {0, Errno::SUCCESS}; @@ -764,11 +765,12 @@ std::pair BSD_USA::FcntlImpl(s32 fd, FcntlCmd cmd, s32 arg) { return {descriptor->flags, Errno::SUCCESS}; case FcntlCmd::SETFL: { const bool enable = (arg & Network::FLAG_O_NONBLOCK) != 0; + // Set the mode and the flags together, so SetDontWait never sees one without the other + std::scoped_lock lock{file_descriptors_mutex}; const Errno bsd_errno = Translate(descriptor->socket->SetNonBlock(enable)); if (bsd_errno != Errno::SUCCESS) { return {-1, bsd_errno}; } - std::scoped_lock lock{file_descriptors_mutex}; if (auto& entry = file_descriptors[fd]; entry && entry->socket == descriptor->socket) { entry->flags = arg; } @@ -880,19 +882,14 @@ std::pair BSD_USA::RecvImpl(s32 fd, u32 flags, std::vector& mess // Apply flags using Network::FLAG_MSG_DONTWAIT; - using Network::FLAG_O_NONBLOCK; - if ((flags & FLAG_MSG_DONTWAIT) != 0) { - flags &= ~FLAG_MSG_DONTWAIT; - if ((descriptor->flags & FLAG_O_NONBLOCK) == 0) { - descriptor->socket->SetNonBlock(true); - } - } + const bool dont_wait = (flags & FLAG_MSG_DONTWAIT) != 0 && SetDontWait(fd, *descriptor, true); + flags &= ~FLAG_MSG_DONTWAIT; const auto [ret, bsd_errno] = Translate(descriptor->socket->Recv(flags, message)); // Restore original state - if ((descriptor->flags & FLAG_O_NONBLOCK) == 0) { - descriptor->socket->SetNonBlock(false); + if (dont_wait) { + SetDontWait(fd, *descriptor, false); } return {ret, bsd_errno}; @@ -916,20 +913,15 @@ std::pair BSD_USA::RecvFromImpl(s32 fd, u32 flags, std::vector& // Apply flags using Network::FLAG_MSG_DONTWAIT; - using Network::FLAG_O_NONBLOCK; - if ((flags & FLAG_MSG_DONTWAIT) != 0) { - flags &= ~FLAG_MSG_DONTWAIT; - if ((descriptor->flags & FLAG_O_NONBLOCK) == 0) { - descriptor->socket->SetNonBlock(true); - } - } + const bool dont_wait = (flags & FLAG_MSG_DONTWAIT) != 0 && SetDontWait(fd, *descriptor, true); + flags &= ~FLAG_MSG_DONTWAIT; const auto [ret, bsd_errno] = Translate(descriptor->socket->RecvFrom(flags, message, p_addr_in)); // Restore original state - if ((descriptor->flags & FLAG_O_NONBLOCK) == 0) { - descriptor->socket->SetNonBlock(false); + if (dont_wait) { + SetDontWait(fd, *descriptor, false); } if (p_addr_in) { @@ -1043,8 +1035,21 @@ s32 BSD_USA::FindFreeFileDescriptorHandle() noexcept { return -1; } +bool BSD_USA::SetDontWait(s32 fd, const FileDescriptor& descriptor, bool enable) { + // Read the current flags, not the copy: fcntl may have changed them during the receive + std::scoped_lock lock{file_descriptors_mutex}; + const auto& entry = file_descriptors[fd]; + const s32 current_flags = + entry && entry->socket == descriptor.socket ? entry->flags : descriptor.flags; + if ((current_flags & Network::FLAG_O_NONBLOCK) != 0) { + return false; + } + descriptor.socket->SetNonBlock(enable); + return true; +} + bool BSD_USA::IsFileDescriptorValid(s32 fd) const noexcept { - if (fd > static_cast(MAX_FD) || fd < 0) { + if (fd >= static_cast(MAX_FD) || fd < 0) { LOG_ERROR(Service, "Invalid file descriptor handle={}", fd); return false; } @@ -1064,22 +1069,37 @@ void BSD_USA::BuildErrnoResponse(HLERequestContext& ctx, Errno bsd_errno) const } void BSD_USA::OnProxyPacketReceived(const Network::ProxyPacket& packet) { - // Runs on the room thread. The lock keeps a socket alive while a bsd thread closes it. - std::scoped_lock lock{file_descriptors_mutex}; - // A duplicated fd shares its socket, which must still get each packet only once - std::array delivered{}; - size_t num_delivered = 0; - for (auto& optional_descriptor : file_descriptors) { - if (!optional_descriptor.has_value() || !optional_descriptor->socket) { - continue; - } - Network::SocketBase* const socket = optional_descriptor->socket.get(); - const auto delivered_end = delivered.begin() + num_delivered; - if (std::find(delivered.begin(), delivered_end, socket) != delivered_end) { - continue; + // Runs on the room thread. The copies keep each socket alive if a bsd thread closes it, so + // the bsd threads do not wait for the decompression below. + std::array, MAX_FD> sockets{}; + size_t num_sockets = 0; + { + std::scoped_lock lock{file_descriptors_mutex}; + for (const auto& descriptor : file_descriptors) { + if (!descriptor.has_value() || !descriptor->socket) { + continue; + } + // A duplicated fd shares its socket, which must still get each packet only once + const auto sockets_end = sockets.begin() + num_sockets; + if (std::find(sockets.begin(), sockets_end, descriptor->socket) == sockets_end) { + sockets[num_sockets++] = descriptor->socket; + } } - delivered[num_delivered++] = socket; - socket->HandleProxyPacket(packet); + } + if (num_sockets == 0) { + return; + } + + // Decompress once for all sockets that share the port + const Network::ProxyPacket decompressed{ + .local_endpoint = packet.local_endpoint, + .remote_endpoint = packet.remote_endpoint, + .protocol = packet.protocol, + .broadcast = packet.broadcast, + .data = Common::Compression::DecompressDataZSTD(packet.data), + }; + for (size_t i = 0; i < num_sockets; ++i) { + sockets[i]->HandleProxyPacket(decompressed); } } diff --git a/src/core/hle/service/sockets/bsd.h b/src/core/hle/service/sockets/bsd.h index f908edc7d2..559830d7a1 100644 --- a/src/core/hle/service/sockets/bsd.h +++ b/src/core/hle/service/sockets/bsd.h @@ -180,6 +180,8 @@ private: bool IsFileDescriptorValid(s32 fd) const noexcept; /// Copies a descriptor under file_descriptors_mutex; nullopt if fd has no socket std::optional CopyDescriptor(s32 fd); + /// Switches a blocking socket for one MSG_DONTWAIT call; false if fd is already nonblocking + bool SetDontWait(s32 fd, const FileDescriptor& descriptor, bool enable); void BuildErrnoResponse(HLERequestContext& ctx, Errno bsd_errno) const noexcept; diff --git a/src/core/internal_network/socket_proxy.cpp b/src/core/internal_network/socket_proxy.cpp index 1b93ce93e5..95f64b9bd0 100644 --- a/src/core/internal_network/socket_proxy.cpp +++ b/src/core/internal_network/socket_proxy.cpp @@ -43,22 +43,7 @@ void ProxySocket::HandleProxyPacket(const ProxyPacket& packet) { LOG_INFO(Network, "Received broadcast packet, but not configured for broadcast mode"); return; } - } - - // Decompress without the lock, so the game thread does not wait for it - ProxyPacket decompressed{ - .local_endpoint = packet.local_endpoint, - .remote_endpoint = packet.remote_endpoint, - .protocol = packet.protocol, - .broadcast = packet.broadcast, - .data = Common::Compression::DecompressDataZSTD(packet.data), - }; - { - std::lock_guard guard(packets_mutex); - if (closed) { - return; - } - received_packets.push(std::move(decompressed)); + received_packets.push(packet); } // A MSG_PEEK reader leaves the packet queued, so every waiter must see it packets_cv.notify_all(); @@ -304,11 +289,15 @@ Errno ProxySocket::SetSndTimeo(u32 value) { } Errno ProxySocket::SetRcvTimeo(u32 value) { - receive_timeout = value; + { + std::lock_guard guard(packets_mutex); + receive_timeout = value; + } return SetSockOpt(fd, SO_RCVTIMEO, static_cast(value)); } Errno ProxySocket::SetNonBlock(bool enable) { + std::lock_guard guard(packets_mutex); blocking = !enable; return Errno::SUCCESS; } diff --git a/src/core/internal_network/sockets.h b/src/core/internal_network/sockets.h index 0b751b6622..37cffd4651 100644 --- a/src/core/internal_network/sockets.h +++ b/src/core/internal_network/sockets.h @@ -89,6 +89,7 @@ public: virtual bool IsOpened() const = 0; + /// Queues a room packet whose data is already decompressed virtual void HandleProxyPacket(const ProxyPacket& packet) = 0; [[nodiscard]] SOCKET GetFD() const { -- cgit v1.2.3