fix(libSceNet): report EAGAIN when a stream send or receive times out on Windows

core/libs/prx/libSceNet/Export.cpp

@@ -376,6 +376,25 @@ int set_kernel_timeout(int descriptor, int optname, const void* value, std::uint
     return 0;
 }
 
+int wait_for_stream(const Sock& socket, std::size_t length, bool sending) {
+#ifdef _WIN32
+    const int timeout_us = sending ? socket.snd_timeout_us : socket.rcv_timeout_us;
+    if (socket.type != NET_SOCK_STREAM || socket.nonblock || timeout_us <= 0 || length == 0) return 0;
+    sockaddr_storage peer{};
+    NativeLength peer_length = sizeof(peer);
+    if (::getpeername(socket.native->value, reinterpret_cast<sockaddr*>(&peer), &peer_length) != 0) return 0;
+    WSAPOLLFD descriptor{socket.native->value, static_cast<SHORT>(sending ? POLLWRNORM : POLLRDNORM), 0};
+    const int ready = ::WSAPoll(&descriptor, 1, static_cast<INT>((static_cast<std::int64_t>(timeout_us) + 999) / 1000));
+    if (ready < 0) return native_error();
+    return ready == 0 ? NET_EAGAIN : 0;
+#else
+    (void)socket;
+    (void)length;
+    (void)sending;
+    return 0;
+#endif
+}
+
 int begin_lookup(int rid, bool Resolver::*preservedAbort, std::uint64_t& generation) {
     std::lock_guard lk(g_mutex);
     const auto resolver = g_resolvers.find(rid);
@@ -453,7 +472,7 @@ std::int64_t message_length(const NetMsghdr* message) {
 
 int receive_error(const Sock& socket) {
 #ifdef _WIN32
-    if (WSAGetLastError() == WSAETIMEDOUT && socket.type != NET_SOCK_STREAM && socket.rcv_timeout_us > 0) return NET_EAGAIN;
+    if (WSAGetLastError() == WSAETIMEDOUT && socket.rcv_timeout_us > 0) return NET_EAGAIN;
 #else
     (void)socket;
 #endif
@@ -778,6 +797,7 @@ int64_t APS5_VABI sceNetRecv(int s, void* buf, size_t len, int flags) {
         if (it == g_socks.end()) return fail(NET_EBADF);
         socket = it->second;
     }
+    if (const int error = wait_for_stream(socket, len, false)) return fail(error);
     const GuestArena::HostWrite destination(len == 0 ? nullptr : buf, len);
     if (!destination.Open()) return fail(NET_EFAULT);
     const int native_flags = flags == 2 ? MSG_PEEK : 0;
@@ -800,6 +820,7 @@ int64_t APS5_VABI sceNetRecvfrom(int s, void* buf, size_t len, int flags, void*
         if (it == g_socks.end()) return fail(NET_EBADF);
         socket = it->second;
     }
+    if (const int error = wait_for_stream(socket, len, false)) return fail(error);
     const GuestArena::HostWrite destination(len == 0 ? nullptr : buf, len);
     if (!destination.Open()) return fail(NET_EFAULT);
     sockaddr_storage peer{};
@@ -826,6 +847,7 @@ int64_t APS5_VABI sceNetSend(int s, const void* buf, size_t len, int flags) {
         if (it == g_socks.end()) return fail(NET_EBADF);
         socket = it->second;
     }
+    if (const int error = wait_for_stream(socket, len, true)) return fail(error);
     const int result = ::send(socket.native->value, static_cast<const char*>(buf), static_cast<int>(len), NATIVE_SEND_FLAGS);
     return result >= 0 ? result : fail(native_error());
 }
@@ -841,6 +863,7 @@ int64_t APS5_VABI sceNetSendto(int s, const void* buf, size_t len, int flags, co
         if (it == g_socks.end()) return fail(NET_EBADF);
         socket = it->second;
     }
+    if (const int error = wait_for_stream(socket, len, true)) return fail(error);
     int result;
     if (!to) {
         if (tolen != 0) return fail(NET_EINVAL);
@@ -885,6 +908,7 @@ int64_t APS5_VABI sceNetRecvmsg(int s, NetMsghdr* msg, int flags) {
         if (it == g_socks.end()) return fail(NET_EBADF);
         socket = it->second;
     }
+    if (const int error = wait_for_stream(socket, static_cast<std::size_t>(total), false)) return fail(error);
     const bool datagram = socket.type != NET_SOCK_STREAM;
     std::vector<char> buffer;
     try {

core/libs/tests/GuestSceNet.cpp

@@ -296,6 +296,73 @@ static void RequireTimeout(int socket, int option, int expected) {
     }
 }
 
+static void CheckStreamTimeouts() {
+    const int listener = sceNetSocket(nullptr, 2, 1, 6);
+    Require(listener >= 0);
+    std::array<std::uint8_t, 16> address{16, 2, 0, 0, 127, 0, 0, 1};
+    Require(sceNetBind_nid_postfix(listener, address.data(), address.size()) == 0 && sceNetListen(listener, 1) == 0);
+    std::uint32_t size = address.size();
+    Require(sceNetGetsockname(listener, address.data(), &size) == 0 && size == address.size());
+    const int client = sceNetSocket(nullptr, 2, 1, 6);
+    Require(client >= 0 && sceNetConnect(client, address.data(), address.size()) == 0);
+    const int accepted = sceNetAccept(listener, nullptr, nullptr);
+    Require(accepted >= 0);
+    const auto timesOut = [](auto call) {
+        const auto start = std::chrono::steady_clock::now();
+        return Failed(call(), 35) && std::chrono::steady_clock::now() - start >= std::chrono::milliseconds(40);
+    };
+    const int timeout = 50000;
+    Require(sceNetSetsockopt(accepted, 0xffff, 0x1006, &timeout, sizeof(timeout)) == 0);
+    static std::array<char, 1 << 16> buffer;
+    std::array<std::uint8_t, 16> source{};
+    NetIovec in[] = {{buffer.data(), 8}};
+    NetMsghdr receive{source.data(), static_cast<std::uint32_t>(source.size()), in, 1, nullptr, 0, 0};
+    Require(timesOut([&] { return sceNetRecv(accepted, buffer.data(), 8, 0); }));
+    Require(timesOut([&] { return sceNetRecv(accepted, buffer.data(), 8, 2); }));
+    size = source.size();
+    Require(timesOut([&] { return sceNetRecvfrom(accepted, buffer.data(), 8, 0, source.data(), &size); }));
+    Require(timesOut([&] { return sceNetRecvmsg(accepted, &receive, 0); }));
+    const int unconnected = sceNetSocket(nullptr, 2, 1, 6);
+    Require(unconnected >= 0 && sceNetSetsockopt(unconnected, 0xffff, 0x1006, &timeout, sizeof(timeout)) == 0);
+    Require(sceNetSetsockopt(listener, 0xffff, 0x1006, &timeout, sizeof(timeout)) == 0);
+    for (const int descriptor : {unconnected, listener}) Require(Failed(sceNetRecv(descriptor, buffer.data(), 8, 0), 57));
+    Require(sceNetSocketClose(unconnected) == 0);
+    const char late[] = "late";
+    Require(sceNetSend(client, late, sizeof(late), 0) == sizeof(late));
+    Require(sceNetRecv(accepted, buffer.data(), buffer.size(), 0) == sizeof(late) && std::strcmp(buffer.data(), late) == 0);
+
+    const int small = 4096;
+    Require(sceNetSetsockopt(client, 0xffff, 0x1001, &small, sizeof(small)) == 0);
+    Require(sceNetSetsockopt(client, 0xffff, 0x1005, &timeout, sizeof(timeout)) == 0);
+    static std::array<char, 1 << 16> chunk;
+    chunk.fill('s');
+    std::int64_t sent = 0;
+    for (int round = 0;; ++round) {
+        Require(round < 4096);
+        const auto start = std::chrono::steady_clock::now();
+        const auto result = sceNetSend(client, chunk.data(), chunk.size(), 0);
+        if (result > 0) {
+            Require(result <= static_cast<std::int64_t>(chunk.size()));
+            sent += result;
+            continue;
+        }
+        Require(Failed(result, 35) && std::chrono::steady_clock::now() - start >= std::chrono::milliseconds(40));
+        break;
+    }
+    std::int64_t received = 0;
+    while (received < sent) {
+        buffer.fill(0);
+        const auto result = sceNetRecv(accepted, buffer.data(), buffer.size(), 0);
+        Require(result > 0 && std::all_of(buffer.begin(), buffer.begin() + result, [](char byte) { return byte == 's'; }));
+        received += result;
+    }
+    Require(received == sent);
+    Require(timesOut([&] { return sceNetRecv(accepted, buffer.data(), buffer.size(), 0); }));
+    Require(sceNetSend(client, late, sizeof(late), 0) == sizeof(late));
+    Require(sceNetRecv(accepted, buffer.data(), buffer.size(), 0) == sizeof(late) && std::strcmp(buffer.data(), late) == 0);
+    for (const int descriptor : {client, accepted, listener}) Require(sceNetSocketClose(descriptor) == 0);
+}
+
 static void CheckKernelTimeouts() {
     const int descriptor = sceNetSocket("kernel-timeouts", 2, 2, 17);
     Require(descriptor >= 0);
@@ -492,6 +559,7 @@ int main() {
     Require(sceNetSetsockopt(closed, 0xffff, 0x4, &enabled, sizeof(enabled)) == 0);
     Require(close_nid_postfix(closed) == 0);
     Require(Failed(sceNetSocketClose(closed), 9));
+    CheckStreamTimeouts();
     CheckKernelTimeouts();
 
     const int listener = sceNetSocket(nullptr, 2, 1, 6);

docs/dev/debt/libSceNet.md

@@ -8,6 +8,6 @@
 
 ### Functional
 
-- [sceNetSetsockopt](../../../core/libs/prx/libSceNet/Export.cpp) (libSceNet) sets `SO_RCVTIMEO` and `SO_SNDTIMEO` as the host socket's timeouts. On Windows a receive or send on a stream socket that times out fails with `ETIMEDOUT`, where FreeBSD returns `EAGAIN` or the partial count: Winsock leaves the connection in an indeterminate state after a timeout and reports no partial count ([SOL_SOCKET options](https://learn.microsoft.com/en-us/windows/win32/winsock/sol-socket-socket-options)), so FreeBSD's behaviour needs the timeout waited for without the host option. A receive on a datagram socket that times out fails with `EAGAIN`
+- [sceNetSetsockopt](../../../core/libs/prx/libSceNet/Export.cpp) (libSceNet) sets `SO_RCVTIMEO` and `SO_SNDTIMEO` as the host socket's timeouts. Winsock leaves a connection in an indeterminate state after such a timeout and reports no partial count ([SOL_SOCKET options](https://learn.microsoft.com/en-us/windows/win32/winsock/sol-socket-socket-options)), so on Windows a blocking receive or send on a stream socket first waits for the socket with `WSAPoll` for the timeout and fails with `EAGAIN` when the wait ends, as FreeBSD does. A send that starts while the socket is writable and still blocks past the timeout fails with `ETIMEDOUT`, where FreeBSD returns the partial count; a receive whose data another thread takes after the wait gets `EAGAIN` from the host timeout. A receive on a datagram socket that times out fails with `EAGAIN`
 - [sceNetSendmsg and sceNetRecvmsg](../../../core/libs/prx/libSceNet/Export.cpp) (libSceNet) copy the scatter/gather list through one host buffer and accept only the flags `sceNetSend`/`sceNetRecv` accept. Sending control data throws; a receive never returns control data (`msg_controllen` 0), since no socket option that queues it is supported
 - [sceNetResolverAbort and sceNetResolverStartNtoaMultipleRecords](../../../core/libs/prx/libSceNet/Export.cpp) (libSceNet) follow shadPS4 ([net.cpp](https://github.com/shadps4-emu/shadPS4/blob/811bfcd7d17ed8ae975c477d9ff28abc744e1ca8/src/core/libraries/net/net.cpp), [net_resolver.cpp](https://github.com/shadps4-emu/shadPS4/blob/811bfcd7d17ed8ae975c477d9ff28abc744e1ca8/src/core/libraries/net/net_resolver.cpp)); the `(int rid, int flags)` signature of the abort also matches [SharpProspero](https://github.com/SvenGDK/SharpProspero/blob/5dfdb4b0b166e83dc9ae275001afb695f12c7174/src/SharpProspero/Interop/Net/Socket.cs). Unverified on hardware. Lookups run synchronously over the host resolver, which cannot be interrupted: an abort while a lookup runs on another thread returns at once, and that lookup fails with `EINTR` without writing its output only when the host call returns; the abort flags are then ignored, as shadPS4 does for a running lookup. With nothing running, abort flag 1 (NTOA preservation) or 2 (ATON preservation) makes the next `sceNetResolverStartNtoa*` or `sceNetResolverStartAton` call on that resolver fail with `EINTR`, reported by `sceNetResolverGetError`, after its arguments are checked. Unlike shadPS4, which ignores them, other abort flag bits throw, and `sceNetResolverStartNtoaMultipleRecords` throws for any flag, including async (1) and 0x10000 (IP address literals disabled), as `sceNetResolverStartNtoaMultipleRecordsEx` does