feat(libkernel): implement read and write events for guest sockets

core/libs/prx/libkernel/CMakeLists.txt

@@ -726,6 +726,16 @@ if(BUILD_TESTING)
         set_tests_properties(guest_dup PROPERTIES ENVIRONMENT_MODIFICATION
             "PATH=path_list_prepend:$<TARGET_FILE_DIR:libkernel>")
     endif()
+    add_test_executable(guest_socket_event_tests ${LIBS_INCLUDE_DIR}/tests/GuestSocketEvents.cpp)
+    target_include_directories(guest_socket_event_tests PRIVATE ${LIBS_INCLUDE_DIR})
+    target_link_libraries(guest_socket_event_tests PRIVATE libkernel libc)
+    configure_windows_unwind(guest_socket_event_tests)
+    add_test(NAME guest_socket_events COMMAND guest_socket_event_tests)
+    set_tests_properties(guest_socket_events PROPERTIES TIMEOUT 20)
+    if(WIN32)
+        set_tests_properties(guest_socket_events PROPERTIES ENVIRONMENT_MODIFICATION
+            "PATH=path_list_prepend:$<TARGET_FILE_DIR:libkernel>")
+    endif()
     add_test_executable(guest_elf_phdr_match_tests ${LIBS_INCLUDE_DIR}/tests/GuestElfPhdrMatch.cpp)
     target_include_directories(guest_elf_phdr_match_tests PRIVATE ${LIBS_INCLUDE_DIR})
     target_link_libraries(guest_elf_phdr_match_tests PRIVATE libkernel libc)

core/libs/prx/libkernel/Equeue/Equeue.cpp

@@ -1,5 +1,7 @@
 #include "Equeue.hpp"
 #include "prx/libkernel/Time/include/Time.hpp"
+#include "prx/libc/include/General.hpp"
+#include "prx/libkernel/Socket/include/SocketRuntime.hpp"
 
 #include <algorithm>
 #include <chrono>
@@ -42,6 +44,7 @@ void KernelEqueuePrivate::Close() {
     }
     m_events.clear();
     m_cond.NotifyAll();
+    WakePollers();
 }
 
 void KernelEqueuePrivate::TriggerExpiredTimers(uint64_t nowNs) {
@@ -75,12 +78,48 @@ bool KernelEqueuePrivate::NextTimerWaitMicros(uint64_t nowNs, uint32_t* out) con
     return true;
 }
 
+bool KernelEqueuePrivate::PollEvents() {
+    bool polled = false;
+    for (auto& ev : m_events) {
+        if (ev.filter.pollFunc != nullptr) {
+            ev.triggered = ev.filter.pollFunc(&ev);
+            polled = true;
+        }
+    }
+    return polled;
+}
+
+void KernelEqueuePrivate::WakePollers() {
+    for (const auto waker : m_pollers) {
+        GuestSockets::Wake(waker);
+    }
+}
+
+void KernelEqueuePrivate::WaitForDescriptors(std::unique_lock<std::mutex>& lock, std::uint64_t deadlineNanos) {
+    std::vector<GuestSockets::Interest> interests;
+    for (const auto& e : m_events) {
+        if (e.filter.pollFunc != nullptr) {
+            interests.push_back({static_cast<int>(e.event.ident), e.event.filter == EVFILT_WRITE});
+        }
+    }
+    const auto waker = GuestSockets::CurrentWaker();
+    m_pollers.push_back(waker);
+    lock.unlock();
+    const bool waited = GuestSockets::WaitAny(interests, deadlineNanos);
+    lock.lock();
+    m_pollers.erase(std::find(m_pollers.begin(), m_pollers.end(), waker));
+    if (!waited) {
+        m_cond.WaitUntil(lock, deadlineNanos);
+    }
+}
+
 int KernelEqueuePrivate::GetTriggeredEvents(KernelEvent* ev, int num) {
     std::unique_lock lock(m_mutex);
     if (m_closed) {
         return SCE_KERNEL_ERROR_EBADF;
     }
     TriggerExpiredTimers(MonotonicNs());
+    PollEvents();
     int ret = 0;
     for (auto it = m_events.begin(); it != m_events.end();) {
         auto& e = *it;
@@ -123,6 +162,7 @@ int KernelEqueuePrivate::WaitForEvents(KernelEvent* ev, int num, uint32_t micros
     const std::uint64_t deadline = TimedWait::DeadlineNanos(micros);
     for (;;) {
         TriggerExpiredTimers(MonotonicNs());
+        const bool polling = PollEvents();
         int ret = 0;
         for (auto it = m_events.begin(); it != m_events.end() && ret < num;) {
             auto& e = *it;
@@ -159,8 +199,12 @@ int KernelEqueuePrivate::WaitForEvents(KernelEvent* ev, int num, uint32_t micros
         }
         if (micros == 0) {
             uint32_t timerWait = 0;
-            if (NextTimerWaitMicros(MonotonicNs(), &timerWait)) {
-                m_cond.WaitUntil(lock, TimedWait::NowNanos() + static_cast<std::uint64_t>(timerWait) * 1000ULL);
+            const bool hasTimer = NextTimerWaitMicros(MonotonicNs(), &timerWait);
+            const std::uint64_t wake = hasTimer ? TimedWait::NowNanos() + static_cast<std::uint64_t>(timerWait) * 1000ULL : 0;
+            if (polling) {
+                WaitForDescriptors(lock, wake);
+            } else if (hasTimer) {
+                m_cond.WaitUntil(lock, wake);
             } else {
                 m_cond.Wait(lock);
             }
@@ -172,7 +216,12 @@ int KernelEqueuePrivate::WaitForEvents(KernelEvent* ev, int num, uint32_t micros
                 return 0;
             }
             const std::uint64_t timerDeadline = now + static_cast<std::uint64_t>(timerWait) * 1000ULL;
-            m_cond.WaitUntil(lock, hasTimer ? std::min(deadline, timerDeadline) : deadline);
+            const std::uint64_t wake = hasTimer ? std::min(deadline, timerDeadline) : deadline;
+            if (polling) {
+                WaitForDescriptors(lock, wake);
+            } else {
+                m_cond.WaitUntil(lock, wake);
+            }
         }
     }
 }
@@ -201,6 +250,7 @@ int KernelEqueuePrivate::AddEvent(const KernelEqueueEvent& event) {
         m_events.push_back(event);
     }
     m_cond.NotifyOne();
+    WakePollers();
     return EQUEUE_OK;
 }
 
@@ -223,6 +273,7 @@ int KernelEqueuePrivate::TriggerEvent(uintptr_t ident, int16_t filter, void* tri
         it->triggered = true;
     }
     m_cond.NotifyOne();
+    WakePollers();
     return EQUEUE_OK;
 }
 
@@ -244,9 +295,20 @@ int KernelEqueuePrivate::DeleteEvent(uintptr_t ident, int16_t filter) {
         it->filter.deleteEventFunc(m_handle, &*it);
     }
     m_events.erase(it);
+    WakePollers();
     return EQUEUE_OK;
 }
 
+void KernelEqueuePrivate::RemoveDescriptorEvents(uintptr_t ident) {
+    std::unique_lock lock(m_mutex);
+    const auto removed = std::erase_if(m_events, [ident](const auto& e) {
+        return e.event.ident == ident && (e.event.filter == EVFILT_READ || e.event.filter == EVFILT_WRITE);
+    });
+    if (removed != 0) {
+        WakePollers();
+    }
+}
+
 KernelEqueueRef EqueuePin_nid_postfix(KernelEqueue eq) {
     if (eq == 0) {
         return {};
@@ -481,4 +543,64 @@ int APS5_VABI sceKernelDeleteAmprSystemEvent(KernelEqueue eq, int id) {
     return sceKernelDeleteAmprEvent(eq, id);
 }
 
+static int AddDescriptorEvent(KernelEqueue eq, int fd, std::size_t size, void* udata, int16_t filter) {
+    if (fd < GuestSockets::FirstDescriptor) {
+        NotImplemented_nid_no_patch(filter == EVFILT_READ ? "sceKernelAddReadEvent: non-socket descriptor" : "sceKernelAddWriteEvent: non-socket descriptor");
+    }
+    if (size > 1) {
+        NotImplemented_nid_no_patch(filter == EVFILT_READ ? "sceKernelAddReadEvent: low watermark" : "sceKernelAddWriteEvent: low watermark");
+    }
+    if (!GuestSockets::IsOpen(fd)) {
+        return SCE_KERNEL_ERROR_EBADF;
+    }
+    KernelEqueueEvent event{};
+    event.event.ident = static_cast<uintptr_t>(fd);
+    event.event.filter = filter;
+    event.event.flags = EV_ADD;
+    event.event.udata = udata;
+    event.filter.pollFunc = [](KernelEqueueEvent* e) {
+        std::int64_t data = 0;
+        bool eof = false;
+        if (!GuestSockets::Ready(static_cast<int>(e->event.ident), e->event.filter == EVFILT_WRITE, &data, &eof)) {
+            return false;
+        }
+        e->event.data = static_cast<intptr_t>(data);
+        e->event.flags = static_cast<uint16_t>(eof ? (e->event.flags | EV_EOF) : (e->event.flags & ~EV_EOF));
+        return true;
+    };
+    event.filter.resetFunc = [](KernelEqueueEvent* e) {
+        e->triggered = false;
+    };
+    return EqueueAddEvent_nid_postfix(eq, event);
+}
+
+int APS5_VABI sceKernelAddReadEvent(KernelEqueue eq, int fd, std::size_t size, void* udata) {
+    return AddDescriptorEvent(eq, fd, size, udata, EVFILT_READ);
+}
+
+int APS5_VABI sceKernelDeleteReadEvent(KernelEqueue eq, int fd) {
+    return EqueueDeleteEvent_nid_postfix(eq, static_cast<uintptr_t>(fd), EVFILT_READ);
+}
+
+int APS5_VABI sceKernelAddWriteEvent(KernelEqueue eq, int fd, std::size_t size, void* udata) {
+    return AddDescriptorEvent(eq, fd, size, udata, EVFILT_WRITE);
+}
+
+int APS5_VABI sceKernelDeleteWriteEvent(KernelEqueue eq, int fd) {
+    return EqueueDeleteEvent_nid_postfix(eq, static_cast<uintptr_t>(fd), EVFILT_WRITE);
+}
+
+}
+
+void EqueueDescriptorClosed(int descriptor) {
+    std::vector<KernelEqueueRef> owners;
+    {
+        std::unique_lock lock(g_equeueMutex);
+        for (const auto& [handle, owner] : g_equeues) {
+            owners.push_back(owner);
+        }
+    }
+    for (const auto& owner : owners) {
+        owner->RemoveDescriptorEvents(static_cast<uintptr_t>(descriptor));
+    }
 }

core/libs/prx/libkernel/Equeue/Equeue.hpp

@@ -8,12 +8,15 @@
 #include <mutex>
 #include <string>
 #include <unordered_map>
+#include <vector>
 
 #include "SceTypes.hpp"
 #include "prx/libc/include/general/VabiMacros.hpp"
 #include "prx/libkernel/KernelErrors.hpp"
 #include "prx/libkernel/Time/include/TimedWait.hpp"
 
+static constexpr int16_t EVFILT_READ = -1;
+static constexpr int16_t EVFILT_WRITE = -2;
 static constexpr int16_t EVFILT_TIMER = -7;
 static constexpr int16_t EVFILT_USER = -11;
 static constexpr int16_t EVFILT_VIDEO_OUT = -13;
@@ -24,6 +27,7 @@ static constexpr uint16_t EV_ADD = 0x0001;
 static constexpr uint16_t EV_ONESHOT = 0x0010;
 static constexpr uint16_t EV_CLEAR = 0x0020;
 static constexpr uint16_t EV_ERROR = 0x4000;
+static constexpr uint16_t EV_EOF = 0x8000;
 
 static constexpr int EQUEUE_OK = 0;
 
@@ -32,13 +36,15 @@ struct KernelEqueueEvent;
 using EqueueTriggerFunc = void (*)(KernelEqueueEvent* event, void* triggerData);
 using EqueueResetFunc = void (*)(KernelEqueueEvent* event);
 using EqueueDeleteFunc = void (*)(KernelEqueue eq, KernelEqueueEvent* event);
+using EqueuePollFunc = bool (*)(KernelEqueueEvent* event);
 
 struct KernelFilter {
     void* data = nullptr;
     std::shared_ptr<void> owner;
     EqueueTriggerFunc triggerFunc = nullptr;
     EqueueResetFunc resetFunc = nullptr;
     EqueueDeleteFunc deleteEventFunc = nullptr;
+    EqueuePollFunc pollFunc = nullptr;
 };
 
 struct KernelEqueueEvent {
@@ -66,14 +72,19 @@ class KernelEqueuePrivate {
     int DeleteEvent(uintptr_t ident, int16_t filter);
     int GetTriggeredEvents(KernelEvent* ev, int num);
     int WaitForEvents(KernelEvent* ev, int num, uint32_t micros);
+    void RemoveDescriptorEvents(uintptr_t ident);
     void Close();
     static uint64_t MonotonicNs();
 
 private:
     void TriggerExpiredTimers(uint64_t nowNs);
     bool NextTimerWaitMicros(uint64_t nowNs, uint32_t* out) const;
+    bool PollEvents();
+    void WakePollers();
+    void WaitForDescriptors(std::unique_lock<std::mutex>& lock, std::uint64_t deadlineNanos);
 
     std::list<KernelEqueueEvent> m_events;
+    std::vector<std::uintptr_t> m_pollers;
     std::mutex m_mutex;
     TimedWait::Condition m_cond;
     std::string m_name;
@@ -82,6 +93,7 @@ class KernelEqueuePrivate {
 };
 
 using KernelEqueueRef = std::shared_ptr<KernelEqueuePrivate>;
+void EqueueDescriptorClosed(int descriptor);
 extern "C" {
 
 KernelEqueueRef EqueuePin_nid_postfix(KernelEqueue eq);

core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp

@@ -1,6 +1,7 @@
 #pragma once
 #include <cstddef>
 #include <cstdint>
+#include <vector>
 namespace GuestSockets {
 constexpr int FirstDescriptor = 0x10000000;
 int Duplicate(int descriptor);
@@ -10,6 +11,14 @@ bool IsOpen(int descriptor);
 int Family(int descriptor);
 std::int64_t Read(int descriptor, void* buffer, std::size_t length);
 std::int64_t Write(int descriptor, const void* buffer, std::size_t length);
+bool Ready(int descriptor, bool write, std::int64_t* data, bool* eof);
+struct Interest {
+    int descriptor;
+    bool write;
+};
+std::uintptr_t CurrentWaker();
+void Wake(std::uintptr_t waker);
+bool WaitAny(const std::vector<Interest>& interests, std::uint64_t deadlineNanos);
 }
 
 extern "C" bool GuestSocketIsOpen_nid_no_patch(int descriptor);