fix(libSceNet): pace hot zero-timeout epoll polls and record epoll events in the observation log

core/libs/prx/libSceNet/CMakeLists.txt

@@ -230,4 +230,19 @@ if(BUILD_TESTING)
                 "PATH=path_list_prepend:$<TARGET_FILE_DIR:libkernel>")
         endif()
     endforeach()
+    add_test_executable(net_epoll_per_call_tests ${LIBS_INCLUDE_DIR}/tests/NetEpollPerCall.cpp)
+    target_include_directories(net_epoll_per_call_tests PRIVATE ${LIBS_INCLUDE_DIR})
+    target_link_libraries(net_epoll_per_call_tests PRIVATE libkernel libc libSceNetCtl)
+    configure_windows_unwind(net_epoll_per_call_tests)
+    if(WIN32)
+        target_link_libraries(net_epoll_per_call_tests PRIVATE ws2_32)
+    endif()
+    foreach(scenario IN ITEMS udp tcp refused closed spin observe)
+        add_test(NAME net_epoll_per_call_${scenario} COMMAND net_epoll_per_call_tests ${scenario})
+        set_tests_properties(net_epoll_per_call_${scenario} PROPERTIES TIMEOUT 40)
+        if(WIN32)
+            set_tests_properties(net_epoll_per_call_${scenario} PROPERTIES ENVIRONMENT_MODIFICATION
+                "PATH=path_list_prepend:$<TARGET_FILE_DIR:libkernel>")
+        endif()
+    endforeach()
 endif()

core/libs/prx/libSceNet/Export.cpp

@@ -38,6 +38,7 @@
 #include <set>
 #include <stdexcept>
 #include <string>
+#include <thread>
 #include <utility>
 #include "SceTypes.hpp"
 #include "prx/libSceNetCtl/include/NetBridge.hpp"
@@ -2390,6 +2391,20 @@ int APS5_VABI sceNetGetSockInfo6(std::int32_t s, NetSockInfo6* info, std::uint32
     return netCall.Ret(written);
 }
 
+static void pace_empty_poll(bool ready) {
+    constexpr auto HotGap = std::chrono::microseconds(500);
+    constexpr unsigned YieldAfter = 32;
+    constexpr unsigned SleepAfter = 256;
+    thread_local std::chrono::steady_clock::time_point last{};
+    thread_local unsigned streak = 0;
+    const auto now = std::chrono::steady_clock::now();
+    if (ready || now - last >= HotGap) streak = 0;
+    else if (streak < SleepAfter) ++streak;
+    if (streak >= SleepAfter) std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    else if (streak >= YieldAfter) std::this_thread::yield();
+    last = streak == 0 ? now : std::chrono::steady_clock::now();
+}
+
 int APS5_VABI sceNetEpollCreate(const char* name, int flags) {
     NetBridgeCall netCall("libSceNet", __func__, [&](NetBridgeJson& netIn) { netIn.Int("flags", flags); });
     if (!name || flags != 0) return netCall.Ret(fail(NET_EINVAL));
@@ -2459,7 +2474,10 @@ int APS5_VABI sceNetEpollAbort(int eid, int flags) {
 }
 
 int APS5_VABI sceNetEpollControl(int eid, int op, int id, const NetEpollEvent* event) {
-    NetBridgeCall netCall("libSceNet", __func__, [&](NetBridgeJson& netIn) { netIn.Int("eid", eid); netIn.Int("op", op); netIn.Int("id", id); });
+    NetBridgeCall netCall("libSceNet", __func__, [&](NetBridgeJson& netIn) {
+        netIn.Int("eid", eid); netIn.Int("op", op); netIn.Int("id", id);
+        if (event != nullptr) netIn.Uint("events", event->events);
+    });
     bool kernel = false;
     {
         std::lock_guard lk(g_mutex);
@@ -2491,6 +2509,15 @@ int APS5_VABI sceNetEpollControl(int eid, int op, int id, const NetEpollEvent* e
 int APS5_VABI sceNetEpollWait(int eid, NetEpollEvent* events, int maxevents, int timeout) {
     NetBridgeCall netCall("libSceNet", __func__, [&](NetBridgeJson& netIn) { netIn.Int("eid", eid); netIn.Int("maxevents", maxevents); netIn.Int("timeout", timeout); });
     if (!events || maxevents <= 0) return netCall.Ret(fail(NET_EINVAL));
+    const auto readyResult = [&](int count) {
+        if (netCall.On() && count > 0) {
+            netCall.Out().Begin("ready");
+            for (int index = 0; index < count; ++index) netCall.Out().Uint(std::to_string(events[index].ident), events[index].events);
+            netCall.Out().End();
+        }
+        if (timeout == 0) pace_empty_poll(count > 0);
+        return netCall.Ret(count);
+    };
     struct Watch {
         int descriptor;
         std::shared_ptr<const NetEpollEvent> registration;
@@ -2539,7 +2566,7 @@ int APS5_VABI sceNetEpollWait(int eid, NetEpollEvent* events, int maxevents, int
             }
         }
         if (watches.empty()) {
-            if (timeout == 0) return netCall.Ret(0);
+            if (timeout == 0) return readyResult(0);
             std::unique_lock lk(g_mutex);
             const auto now = std::chrono::steady_clock::now();
             if (timeout > 0 && now >= deadline) return netCall.Ret(0);
@@ -2609,10 +2636,10 @@ int APS5_VABI sceNetEpollWait(int eid, NetEpollEvent* events, int maxevents, int
                     events[count].events = guest_events;
                     ++count;
                 }
-                if (count > 0) return netCall.Ret(count);
+                if (count > 0) return readyResult(count);
             }
         }
-        if (timeout == 0) return netCall.Ret(0);
+        if (timeout == 0) return readyResult(0);
     }
 }
 

core/libs/tests/NetEpollPerCall.cpp

@@ -0,0 +1,196 @@
+#include "../prx/libSceNet/Export.cpp"
+#include <filesystem>
+#include <fstream>
+#include <future>
+#include <source_location>
+
+static void Require(bool value, const std::source_location location = std::source_location::current()) {
+    if (value) return;
+    std::fprintf(stderr, "requirement failed at %s:%u\n", location.file_name(), location.line());
+    std::abort();
+}
+
+static constexpr std::uint32_t In = 1, Out = 2, Err = 8, Hup = 16;
+static constexpr int NonBlockingOption = 0x1200;
+
+static std::int64_t PollOnce(int descriptor, std::uint32_t mask) {
+    const int epoll = sceNetEpollCreate("per-call", 0);
+    Require(epoll >= 0);
+    NetEpollEvent request{};
+    request.events = mask;
+    request.data.u64 = 0x1234;
+    Require(sceNetEpollControl(epoll, 1, descriptor, &request) == 0);
+    NetEpollEvent ready{};
+    const int count = sceNetEpollWait(epoll, &ready, 16, 0);
+    Require(count == 0 || count == 1);
+    if (count == 1) Require(ready.ident == static_cast<std::uint64_t>(descriptor) && ready.data.u64 == 0x1234);
+    Require(sceNetEpollControl(epoll, 3, descriptor, nullptr) == 0);
+    Require(sceNetEpollDestroy(epoll) == 0);
+    return count == 0 ? -1 : static_cast<std::int64_t>(ready.events);
+}
+
+static std::int64_t PollUntil(int descriptor, std::uint32_t mask, std::chrono::milliseconds limit) {
+    const auto deadline = std::chrono::steady_clock::now() + limit;
+    for (;;) {
+        const auto seen = PollOnce(descriptor, mask);
+        if (seen != -1 || std::chrono::steady_clock::now() >= deadline) return seen;
+        std::this_thread::sleep_for(std::chrono::milliseconds(2));
+    }
+}
+
+static std::array<std::uint8_t, 16> LoopbackAddress(int descriptor) {
+    std::array<std::uint8_t, 16> address{16, 2, 0, 0, 127, 0, 0, 1};
+    Require(sceNetBind_nid_postfix(descriptor, address.data(), address.size()) == 0);
+    std::uint32_t size = address.size();
+    Require(sceNetGetsockname(descriptor, address.data(), &size) == 0 && size == address.size());
+    return address;
+}
+
+static void SetNonBlocking(int descriptor) {
+    const int on = 1;
+    Require(sceNetSetsockopt(descriptor, 0xffff, NonBlockingOption, &on, sizeof(on)) == 0);
+}
+
+static void CheckUdp() {
+    const int sender = sceNetSocket("per-call-udp", 2, 2, 17);
+    const int receiver = sceNetSocket("per-call-udp", 2, 2, 17);
+    Require(sender >= 0 && receiver >= 0);
+    SetNonBlocking(sender);
+    const auto address = LoopbackAddress(receiver);
+    for (int round = 0; round != 2000; ++round) {
+        Require(PollOnce(sender, Out) == Out);
+        Require(PollOnce(sender, In | Out) == Out);
+    }
+    for (int round = 0; round != 300; ++round) Require(PollOnce(receiver, In) == -1);
+    const char payload[] = "per-call";
+    Require(sceNetSendto(sender, payload, sizeof(payload), 0, address.data(), address.size()) == sizeof(payload));
+    Require(PollUntil(receiver, In, std::chrono::seconds(5)) == In);
+    Require(PollOnce(receiver, In) == In);
+    Require(PollOnce(receiver, Out) == Out);
+    Require(sceNetSocketClose(sender) == 0 && sceNetSocketClose(receiver) == 0);
+}
+
+static void CheckTcpConnect() {
+    const int listener = sceNetSocket("per-call-tcp", 2, 1, 6);
+    Require(listener >= 0);
+    const auto address = LoopbackAddress(listener);
+    Require(sceNetListen(listener, 4) == 0);
+    const int client = sceNetSocket("per-call-tcp", 2, 1, 6);
+    Require(client >= 0);
+    SetNonBlocking(client);
+    const int result = sceNetConnect(client, address.data(), address.size());
+    Require(result == 0 || (result == static_cast<int>(0x80410124) && *sceNetErrnoLoc() == 36));
+    const auto writable = PollUntil(client, Out, std::chrono::seconds(5));
+    Require(writable > 0 && (writable & Out) && !(writable & Err));
+    int error = -1;
+    std::uint32_t size = sizeof(error);
+    Require(sceNetGetsockopt(client, 0xffff, 0x1007, &error, &size) == 0 && error == 0);
+    const int accepted = sceNetAccept(listener, nullptr, nullptr);
+    Require(accepted >= 0);
+    Require(PollOnce(client, In) == -1);
+    const char byte = 'x';
+    Require(sceNetSend(accepted, &byte, 1, 0) == 1);
+    Require(PollUntil(client, In, std::chrono::seconds(5)) == In);
+    char received = 0;
+    Require(sceNetRecv(client, &received, 1, 0) == 1);
+    Require(sceNetSocketClose(accepted) == 0);
+    const auto closed = PollUntil(client, In, std::chrono::seconds(5));
+    Require(closed > 0 && (closed & (In | Hup)));
+    for (const int descriptor : {client, listener}) Require(sceNetSocketClose(descriptor) == 0);
+}
+
+static void CheckTcpRefused() {
+    const int probe = sceNetSocket("per-call-closed", 2, 1, 6);
+    Require(probe >= 0);
+    const auto address = LoopbackAddress(probe);
+    Require(sceNetSocketClose(probe) == 0);
+    const int client = sceNetSocket("per-call-closed", 2, 1, 6);
+    Require(client >= 0);
+    SetNonBlocking(client);
+    const int result = sceNetConnect(client, address.data(), address.size());
+    if (result != 0) {
+        const auto seen = PollUntil(client, Out | Err, std::chrono::seconds(10));
+        Require(seen > 0 && (seen & (Err | Hup | Out)));
+    }
+    Require(sceNetSocketClose(client) == 0);
+}
+
+static void CheckClosedDescriptor() {
+    const int socket = sceNetSocket("per-call-close", 2, 2, 17);
+    Require(socket >= 0);
+    const int epoll = sceNetEpollCreate("per-call-close", 0);
+    NetEpollEvent request{};
+    request.events = Out;
+    Require(sceNetEpollControl(epoll, 1, socket, &request) == 0);
+    Require(sceNetSocketClose(socket) == 0);
+    NetEpollEvent ready{};
+    Require(sceNetEpollWait(epoll, &ready, 1, 0) == 0);
+    for (const int operation : {1, 2, 3}) {
+        Require(sceNetEpollControl(epoll, operation, socket, operation == 3 ? nullptr : &request) == static_cast<int>(0x80410109));
+        Require(*sceNetErrnoLoc() == 9);
+    }
+    Require(sceNetEpollDestroy(epoll) == 0);
+    Require(sceNetEpollDestroy(epoll) == static_cast<int>(0x80410109));
+}
+
+static void CheckSpinningThreads() {
+    std::array<int, 7> sockets{};
+    for (int& socket : sockets) {
+        socket = sceNetSocket("per-call-spin", 2, 2, 17);
+        Require(socket >= 0);
+    }
+    const auto start = std::chrono::steady_clock::now();
+    std::vector<std::future<int>> workers;
+    for (const int socket : sockets)
+        workers.push_back(std::async(std::launch::async, [socket] {
+            int polls = 0;
+            while (polls < 1500) {
+                Require(PollOnce(socket, In) == -1);
+                ++polls;
+            }
+            return polls;
+        }));
+    for (auto& worker : workers) Require(worker.get() == 1500);
+    Require(std::chrono::steady_clock::now() - start > std::chrono::milliseconds(200));
+    for (const int socket : sockets) Require(sceNetSocketClose(socket) == 0);
+}
+
+static void CheckObservation(const std::filesystem::path& path) {
+    const int socket = sceNetSocket("per-call-observe", 2, 2, 17);
+    Require(socket >= 0);
+    Require(PollOnce(socket, In | Out) == Out);
+    Require(sceNetSocketClose(socket) == 0);
+    std::string text;
+    for (int attempt = 0; attempt != 40 && text.find("sceNetEpollWait") == std::string::npos; ++attempt) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        std::ifstream input(path, std::ios::binary);
+        text.assign(std::istreambuf_iterator<char>(input), std::istreambuf_iterator<char>());
+    }
+    Require(text.find("\"fn\":\"sceNetEpollControl\",\"in\":{\"eid\":") != std::string::npos);
+    Require(text.find("\"events\":3}") != std::string::npos);
+    Require(text.find("\"ready\":{\"" + std::to_string(socket) + "\":2}") != std::string::npos);
+}
+
+int main(int argc, char** argv) {
+    Require(argc == 2);
+    const std::string mode = argv[1];
+    const auto observePath = std::filesystem::temp_directory_path() / "anyps5-net-epoll-per-call.jsonl";
+    if (mode == "observe") {
+        std::error_code ignored;
+        std::filesystem::remove(observePath, ignored);
+#ifdef _WIN32
+        Require(_putenv_s("ANYPS5_NET_OBSERVE", observePath.string().c_str()) == 0);
+#else
+        Require(setenv("ANYPS5_NET_OBSERVE", observePath.c_str(), 1) == 0);
+#endif
+    }
+    Require(sceNetInit_nid_postfix() == 0);
+    if (mode == "udp") CheckUdp();
+    else if (mode == "tcp") CheckTcpConnect();
+    else if (mode == "refused") CheckTcpRefused();
+    else if (mode == "closed") CheckClosedDescriptor();
+    else if (mode == "spin") CheckSpinningThreads();
+    else if (mode == "observe") CheckObservation(observePath);
+    else Require(false);
+    std::printf("epoll per call %s passed\n", mode.c_str());
+}