From 9b7535c6a4305800812dbf12ec92b1dd31414f94 Mon Sep 17 00:00:00 2001 From: kjhq Date: Wed, 7 Oct 2026 10:38:09 +0000 Subject: [PATCH 1/2] feat(libkernel): implement read and write events for guest sockets sceKernelAddReadEvent/DeleteReadEvent and sceKernelAddWriteEvent/ DeleteWriteEvent register EVFILT_READ and EVFILT_WRITE events on guest sockets. The equeue polls the host socket for readiness on each wait and reports the readable byte count or the send buffer size in data, with EV_EOF on hang-up or error. --- core/libs/prx/libkernel/CMakeLists.txt | 10 ++ core/libs/prx/libkernel/Equeue/Equeue.cpp | 79 ++++++++++++- core/libs/prx/libkernel/Equeue/Equeue.hpp | 6 + .../Socket/include/SocketRuntime.hpp | 1 + core/libs/prx/libkernel/Socket/src/Udp.cpp | 33 ++++++ core/libs/tests/GuestSocketEvents.cpp | 105 ++++++++++++++++++ docs/dev/TechnicalDebt.md | 1 + 7 files changed, 232 insertions(+), 3 deletions(-) create mode 100644 core/libs/tests/GuestSocketEvents.cpp diff --git a/core/libs/prx/libkernel/CMakeLists.txt b/core/libs/prx/libkernel/CMakeLists.txt index 4961439df..6f5ccdd85 100644 --- a/core/libs/prx/libkernel/CMakeLists.txt +++ b/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:$") 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:$") + 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) diff --git a/core/libs/prx/libkernel/Equeue/Equeue.cpp b/core/libs/prx/libkernel/Equeue/Equeue.cpp index b8d41b762..1cd163ec0 100644 --- a/core/libs/prx/libkernel/Equeue/Equeue.cpp +++ b/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 #include @@ -11,6 +13,7 @@ static std::unordered_map g_equeues; static std::mutex g_equeueMutex; static uint64_t g_nextEqueue = 1; +static constexpr std::uint64_t PollIntervalNs = 1000000; extern "C" { @@ -75,12 +78,24 @@ 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; +} + 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 +138,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 +175,14 @@ 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(timerWait) * 1000ULL); + const bool hasTimer = NextTimerWaitMicros(MonotonicNs(), &timerWait); + if (hasTimer || polling) { + const std::uint64_t now = TimedWait::NowNanos(); + std::uint64_t wake = hasTimer ? now + static_cast(timerWait) * 1000ULL : now + PollIntervalNs; + if (polling) { + wake = std::min(wake, now + PollIntervalNs); + } + m_cond.WaitUntil(lock, wake); } else { m_cond.Wait(lock); } @@ -172,7 +194,11 @@ int KernelEqueuePrivate::WaitForEvents(KernelEvent* ev, int num, uint32_t micros return 0; } const std::uint64_t timerDeadline = now + static_cast(timerWait) * 1000ULL; - m_cond.WaitUntil(lock, hasTimer ? std::min(deadline, timerDeadline) : deadline); + std::uint64_t wake = hasTimer ? std::min(deadline, timerDeadline) : deadline; + if (polling) { + wake = std::min(wake, now + PollIntervalNs); + } + m_cond.WaitUntil(lock, wake); } } } @@ -481,4 +507,51 @@ 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(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(e->event.ident), e->event.filter == EVFILT_WRITE, &data, &eof)) { + return false; + } + e->event.data = static_cast(data); + e->event.flags = static_cast(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(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(fd), EVFILT_WRITE); +} + } diff --git a/core/libs/prx/libkernel/Equeue/Equeue.hpp b/core/libs/prx/libkernel/Equeue/Equeue.hpp index dd3a47706..dd7dea008 100644 --- a/core/libs/prx/libkernel/Equeue/Equeue.hpp +++ b/core/libs/prx/libkernel/Equeue/Equeue.hpp @@ -14,6 +14,8 @@ #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 +26,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,6 +35,7 @@ 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; @@ -39,6 +43,7 @@ struct KernelFilter { EqueueTriggerFunc triggerFunc = nullptr; EqueueResetFunc resetFunc = nullptr; EqueueDeleteFunc deleteEventFunc = nullptr; + EqueuePollFunc pollFunc = nullptr; }; struct KernelEqueueEvent { @@ -72,6 +77,7 @@ public: private: void TriggerExpiredTimers(uint64_t nowNs); bool NextTimerWaitMicros(uint64_t nowNs, uint32_t* out) const; + bool PollEvents(); std::list m_events; std::mutex m_mutex; diff --git a/core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp b/core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp index db775297c..9a8640add 100644 --- a/core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp +++ b/core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp @@ -10,6 +10,7 @@ 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); } extern "C" bool GuestSocketIsOpen_nid_no_patch(int descriptor); diff --git a/core/libs/prx/libkernel/Socket/src/Udp.cpp b/core/libs/prx/libkernel/Socket/src/Udp.cpp index e87ecd8f5..8cb6d12cc 100644 --- a/core/libs/prx/libkernel/Socket/src/Udp.cpp +++ b/core/libs/prx/libkernel/Socket/src/Udp.cpp @@ -349,6 +349,39 @@ int UnnamedAddress(void* address, std::uint32_t* length) { } } +bool GuestSockets::Ready(int descriptor, bool write, std::int64_t* data, bool* eof) { + std::shared_ptr socket; + { + std::lock_guard lock(socketsMutex); + const auto found = sockets.find(descriptor); + if (found == sockets.end()) return false; + socket = found->second; + } +#ifdef _WIN32 + WSAPOLLFD entry{socket->value, static_cast(write ? POLLWRNORM : POLLRDNORM), 0}; + if (WSAPoll(&entry, 1, 0) <= 0) return false; +#else + pollfd entry{socket->value, static_cast(write ? POLLOUT : POLLIN), 0}; + if (::poll(&entry, 1, 0) <= 0) return false; +#endif + *eof = (entry.revents & (POLLHUP | POLLERR)) != 0; + *data = 0; + if (write) { + int size = 0; + socklen_t length = sizeof(size); + if (getsockopt(socket->value, SOL_SOCKET, SO_SNDBUF, reinterpret_cast(&size), &length) == 0) *data = size; + } else { +#ifdef _WIN32 + unsigned long available = 0; + if (ioctlsocket(socket->value, FIONREAD, &available) == 0) *data = static_cast(available); +#else + int available = 0; + if (::ioctl(socket->value, FIONREAD, &available) == 0) *data = available; +#endif + } + return true; +} + extern "C" { int APS5_VABI fcntl_nid_postfix(int descriptor, int command, ...) { const auto socket = Lookup(descriptor); diff --git a/core/libs/tests/GuestSocketEvents.cpp b/core/libs/tests/GuestSocketEvents.cpp new file mode 100644 index 000000000..011577219 --- /dev/null +++ b/core/libs/tests/GuestSocketEvents.cpp @@ -0,0 +1,105 @@ +#include "SceTypes.hpp" +#include +#include +#include +#include + +extern "C" { +int APS5_VABI socket_nid_postfix(int, int, int); +int APS5_VABI bind_nid_postfix(int, const void*, std::uint32_t); +int APS5_VABI listen_nid_postfix(int, int); +int APS5_VABI getsockname_nid_postfix(int, void*, std::uint32_t*); +int APS5_VABI connect_nid_postfix(int, const void*, std::uint32_t); +int APS5_VABI accept_nid_postfix(int, void*, std::uint32_t*); +std::int64_t APS5_VABI send_nid_postfix(int, const void*, std::uint64_t, int); +std::int64_t APS5_VABI recv_nid_postfix(int, void*, std::uint64_t, int); +int APS5_VABI close_nid_postfix(int); +int APS5_VABI sceKernelCreateEqueue(KernelEqueue* eq, const char* name); +int APS5_VABI sceKernelDeleteEqueue(KernelEqueue eq); +int APS5_VABI sceKernelWaitEqueue(KernelEqueue eq, KernelEvent* ev, int num, int* out, const KernelUseconds* timo); +int APS5_VABI sceKernelAddReadEvent(KernelEqueue eq, int fd, std::size_t size, void* udata); +int APS5_VABI sceKernelDeleteReadEvent(KernelEqueue eq, int fd); +int APS5_VABI sceKernelAddWriteEvent(KernelEqueue eq, int fd, std::size_t size, void* udata); +int APS5_VABI sceKernelDeleteWriteEvent(KernelEqueue eq, int fd); +} + +static constexpr int SCE_OK = 0; +static constexpr int SCE_KERNEL_ERROR_ENOENT = static_cast(0x80020002); +static constexpr int SCE_KERNEL_ERROR_EBADF = static_cast(0x80020009); +static constexpr int SCE_KERNEL_ERROR_ETIMEDOUT = static_cast(0x8002003c); +static constexpr int EVFILT_READ = -1; +static constexpr int EVFILT_WRITE = -2; +static constexpr std::uint16_t EV_EOF = 0x8000; + +static void Require(bool value) { if (!value) std::abort(); } + +static bool Rejects(int (APS5_VABI *add)(KernelEqueue, int, std::size_t, void*), KernelEqueue eq, int fd, std::size_t size) { + try { + add(eq, fd, size, nullptr); + } catch (const std::exception&) { + return true; + } + return false; +} + +int main() { + const int listener = socket_nid_postfix(2, 1, 6); + Require(listener >= 0); + std::array address{16, 2, 0, 0, 127, 0, 0, 1}; + Require(bind_nid_postfix(listener, address.data(), address.size()) == 0); + Require(listen_nid_postfix(listener, 4) == 0); + std::uint32_t addressSize = address.size(); + Require(getsockname_nid_postfix(listener, address.data(), &addressSize) == 0); + + KernelEqueue eq = 0; + Require(sceKernelCreateEqueue(&eq, "sockets") == SCE_OK); + const KernelUseconds shortTimeout = 20000; + const KernelUseconds timeout = 1000000; + int marker = 0; + int count = 0; + std::array events{}; + + Require(sceKernelAddReadEvent(eq, listener, 1, &marker) == SCE_OK); + Require(sceKernelWaitEqueue(eq, events.data(), events.size(), &count, &shortTimeout) == SCE_KERNEL_ERROR_ETIMEDOUT); + Require(count == 0); + + const int client = socket_nid_postfix(2, 1, 0); + Require(client >= 0); + Require(connect_nid_postfix(client, address.data(), address.size()) == 0); + Require(sceKernelWaitEqueue(eq, events.data(), events.size(), &count, &timeout) == SCE_OK); + Require(count == 1); + Require(events[0].ident == static_cast(listener)); + Require(events[0].filter == EVFILT_READ); + Require(events[0].udata == &marker); + const int accepted = accept_nid_postfix(listener, nullptr, nullptr); + Require(accepted >= 0); + Require(sceKernelDeleteReadEvent(eq, listener) == SCE_OK); + Require(sceKernelDeleteReadEvent(eq, listener) == SCE_KERNEL_ERROR_ENOENT); + + Require(sceKernelAddWriteEvent(eq, client, 0, nullptr) == SCE_OK); + Require(sceKernelWaitEqueue(eq, events.data(), events.size(), &count, &timeout) == SCE_OK); + Require(count == 1 && events[0].filter == EVFILT_WRITE && events[0].data > 0); + Require(sceKernelDeleteWriteEvent(eq, client) == SCE_OK); + + Require(sceKernelAddReadEvent(eq, accepted, 1, nullptr) == SCE_OK); + const char message[] = "ready"; + Require(send_nid_postfix(client, message, sizeof(message), 0) == sizeof(message)); + Require(sceKernelWaitEqueue(eq, events.data(), events.size(), &count, &timeout) == SCE_OK); + Require(count == 1 && events[0].ident == static_cast(accepted)); + Require(events[0].data == sizeof(message) && (events[0].flags & EV_EOF) == 0); + char received[sizeof(message)]{}; + Require(recv_nid_postfix(accepted, received, sizeof(received), 0) == sizeof(received)); + Require(sceKernelWaitEqueue(eq, events.data(), events.size(), &count, &shortTimeout) == SCE_KERNEL_ERROR_ETIMEDOUT); + + Require(close_nid_postfix(client) == 0); + Require(sceKernelWaitEqueue(eq, events.data(), events.size(), &count, &timeout) == SCE_OK); + Require(count == 1 && events[0].data == 0); + + Require(sceKernelAddReadEvent(eq, client, 1, nullptr) == SCE_KERNEL_ERROR_EBADF); + Require(Rejects(sceKernelAddReadEvent, eq, 0, 1)); + Require(Rejects(sceKernelAddWriteEvent, eq, accepted, 64)); + + Require(close_nid_postfix(accepted) == 0); + Require(close_nid_postfix(listener) == 0); + Require(sceKernelDeleteEqueue(eq) == SCE_OK); +} diff --git a/docs/dev/TechnicalDebt.md b/docs/dev/TechnicalDebt.md index 8c103b8c8..2fe26defd 100644 --- a/docs/dev/TechnicalDebt.md +++ b/docs/dev/TechnicalDebt.md @@ -332,6 +332,7 @@ Throughout the project, every function at every stage either **does exactly what - [Pending storage-image results](../../core/libs/prx/libSceAgcDriver/Graphics/src/Texture.cpp) (libSceAgcDriver) are stored when something reads their memory, not when the GPU work that wrote them completes. A title that waits for that work, frees the image and reuses its memory can have the new data overwritten by the late store (#1970). `APS5_STORE_AT_FLIP=1` stores every pending result at each flip; storing them at the title's GPU waits (labels, fences) is unimplemented. - Windows host-pointer imports of the guest mapping itself are compared instead of write-watched: a driver can mark imported pages dirty after the synchronous import window. Shared memory imported through a separate writable alias retains protection-based tracking on the guest views; driver access to the import alias does not disarm those views. [GuestMemory](../../core/libs/prx/libSceAgcDriver/Execution/src/GuestMemory.cpp) excludes every overlapping 64 KiB tracker block while its backing is retained. A full block replaced by fresh private backing regains tracking and gets a new generation; protection changes, partial replacements and shared remappings do not restore it. This preserves CPU edits through byte comparisons but disables snapshot reuse and unit shadows for excluded blocks; restoring tracking on import retirement without replacing the backing and measuring the comparison cost remain unimplemented. - [shm_open](../../core/libs/prx/libkernel/File/src/SharedMemory.cpp) (libkernel) creates only anonymous objects (`SHM_ANON`), each backed by a host anonymous file: a `memfd` on Linux, a delete-on-close temporary file on Windows. The flag checks follow FreeBSD 11's `kern_shm_open`; the mode is ignored. Named objects throw. The descriptor cannot be mapped yet: `mmap` accepts only private anonymous mappings and fails a descriptor mapping with `EOPNOTSUPP`, so the known caller (the bundled Mesa in PPSA59322, which truncates the object and maps it `MAP_SHARED`) gets that failure +- [sceKernelAddReadEvent and sceKernelAddWriteEvent](../../core/libs/prx/libkernel/Equeue/Equeue.cpp) (libkernel) poll the host socket instead of being woken by it: while such an event is registered, every wait on its equeue checks the socket at least once per millisecond, so readiness is reported up to 1 ms late and an idle wait wakes the thread 1000 times a second. Only guest sockets are supported; other descriptors and a low watermark above 1 throw - [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 - [Additional content](../../core/libs/prx/libSceAppContent/Export.cpp) (libSceAppContent) is not installed or mounted: `sceAppContentAddcontMount` answers `NOT_FOUND` for every entitlement label and `sceAppContentAddcontUnmount` for every mount point, as no add-content mount is ever handed out; `sceAppContentGetAddcontInfo` answers `DRM_NO_ENTITLEMENT` for every label and `sceAppContentGetAddcontInfoList` reports 0 entries - [sceKernelPread, sceKernelPwrite, sceKernelPreadv, sceKernelPwritev](../../core/libs/prx/libkernel/File/src/Stdio.cpp) (libkernel) on Windows transfer at the given offset with an overlapped `ReadFile`/`WriteFile` and restore the descriptor's file pointer afterwards, so a concurrent `read`/`write` on the same descriptor can see the moved pointer in between, unlike on the console. Handles that are not disk files return `ESPIPE`. `readv` on a pipe or console handle stops after the first iovec read, a short read the console is assumed to allow; each iovec is transferred separately, so a failure after the first transfer returns the bytes already done From 0abe0b832babdf1a642431798004b7e21c2132f4 Mon Sep 17 00:00:00 2001 From: kjhq Date: Wed, 7 Oct 2026 12:13:39 +0000 Subject: [PATCH 2/2] feat(libkernel): wake socket read and write events instead of polling A wait on an equeue that holds socket events now blocks in poll/WSAPoll on those sockets plus a per-thread loopback wake socket, instead of rechecking every millisecond. Adding, triggering or deleting an event and closing the equeue wake the blocked waiters. Closing a socket removes its read and write events, as on FreeBSD. A read event also reports EV_EOF once the peer has shut down its sending side. --- core/libs/prx/libkernel/Equeue/Equeue.cpp | 69 +++++++++-- core/libs/prx/libkernel/Equeue/Equeue.hpp | 6 + .../Socket/include/SocketRuntime.hpp | 8 ++ core/libs/prx/libkernel/Socket/src/Udp.cpp | 108 +++++++++++++++++- core/libs/tests/GuestSocketEvents.cpp | 38 +++++- docs/dev/TechnicalDebt.md | 2 +- 6 files changed, 216 insertions(+), 15 deletions(-) diff --git a/core/libs/prx/libkernel/Equeue/Equeue.cpp b/core/libs/prx/libkernel/Equeue/Equeue.cpp index 1cd163ec0..b3b130ba0 100644 --- a/core/libs/prx/libkernel/Equeue/Equeue.cpp +++ b/core/libs/prx/libkernel/Equeue/Equeue.cpp @@ -13,7 +13,6 @@ static std::unordered_map g_equeues; static std::mutex g_equeueMutex; static uint64_t g_nextEqueue = 1; -static constexpr std::uint64_t PollIntervalNs = 1000000; extern "C" { @@ -45,6 +44,7 @@ void KernelEqueuePrivate::Close() { } m_events.clear(); m_cond.NotifyAll(); + WakePollers(); } void KernelEqueuePrivate::TriggerExpiredTimers(uint64_t nowNs) { @@ -89,6 +89,30 @@ bool KernelEqueuePrivate::PollEvents() { return polled; } +void KernelEqueuePrivate::WakePollers() { + for (const auto waker : m_pollers) { + GuestSockets::Wake(waker); + } +} + +void KernelEqueuePrivate::WaitForDescriptors(std::unique_lock& lock, std::uint64_t deadlineNanos) { + std::vector interests; + for (const auto& e : m_events) { + if (e.filter.pollFunc != nullptr) { + interests.push_back({static_cast(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) { @@ -176,12 +200,10 @@ int KernelEqueuePrivate::WaitForEvents(KernelEvent* ev, int num, uint32_t micros if (micros == 0) { uint32_t timerWait = 0; const bool hasTimer = NextTimerWaitMicros(MonotonicNs(), &timerWait); - if (hasTimer || polling) { - const std::uint64_t now = TimedWait::NowNanos(); - std::uint64_t wake = hasTimer ? now + static_cast(timerWait) * 1000ULL : now + PollIntervalNs; - if (polling) { - wake = std::min(wake, now + PollIntervalNs); - } + const std::uint64_t wake = hasTimer ? TimedWait::NowNanos() + static_cast(timerWait) * 1000ULL : 0; + if (polling) { + WaitForDescriptors(lock, wake); + } else if (hasTimer) { m_cond.WaitUntil(lock, wake); } else { m_cond.Wait(lock); @@ -194,11 +216,12 @@ int KernelEqueuePrivate::WaitForEvents(KernelEvent* ev, int num, uint32_t micros return 0; } const std::uint64_t timerDeadline = now + static_cast(timerWait) * 1000ULL; - std::uint64_t wake = hasTimer ? std::min(deadline, timerDeadline) : deadline; + const std::uint64_t wake = hasTimer ? std::min(deadline, timerDeadline) : deadline; if (polling) { - wake = std::min(wake, now + PollIntervalNs); + WaitForDescriptors(lock, wake); + } else { + m_cond.WaitUntil(lock, wake); } - m_cond.WaitUntil(lock, wake); } } } @@ -227,6 +250,7 @@ int KernelEqueuePrivate::AddEvent(const KernelEqueueEvent& event) { m_events.push_back(event); } m_cond.NotifyOne(); + WakePollers(); return EQUEUE_OK; } @@ -249,6 +273,7 @@ int KernelEqueuePrivate::TriggerEvent(uintptr_t ident, int16_t filter, void* tri it->triggered = true; } m_cond.NotifyOne(); + WakePollers(); return EQUEUE_OK; } @@ -270,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 {}; @@ -555,3 +591,16 @@ int APS5_VABI sceKernelDeleteWriteEvent(KernelEqueue eq, int fd) { } } + +void EqueueDescriptorClosed(int descriptor) { + std::vector 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(descriptor)); + } +} diff --git a/core/libs/prx/libkernel/Equeue/Equeue.hpp b/core/libs/prx/libkernel/Equeue/Equeue.hpp index dd7dea008..c63f505fb 100644 --- a/core/libs/prx/libkernel/Equeue/Equeue.hpp +++ b/core/libs/prx/libkernel/Equeue/Equeue.hpp @@ -8,6 +8,7 @@ #include #include #include +#include #include "SceTypes.hpp" #include "prx/libc/include/general/VabiMacros.hpp" @@ -71,6 +72,7 @@ public: 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(); @@ -78,8 +80,11 @@ private: void TriggerExpiredTimers(uint64_t nowNs); bool NextTimerWaitMicros(uint64_t nowNs, uint32_t* out) const; bool PollEvents(); + void WakePollers(); + void WaitForDescriptors(std::unique_lock& lock, std::uint64_t deadlineNanos); std::list m_events; + std::vector m_pollers; std::mutex m_mutex; TimedWait::Condition m_cond; std::string m_name; @@ -88,6 +93,7 @@ private: }; using KernelEqueueRef = std::shared_ptr; +void EqueueDescriptorClosed(int descriptor); extern "C" { KernelEqueueRef EqueuePin_nid_postfix(KernelEqueue eq); diff --git a/core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp b/core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp index 9a8640add..0703293dc 100644 --- a/core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp +++ b/core/libs/prx/libkernel/Socket/include/SocketRuntime.hpp @@ -1,6 +1,7 @@ #pragma once #include #include +#include namespace GuestSockets { constexpr int FirstDescriptor = 0x10000000; int Duplicate(int descriptor); @@ -11,6 +12,13 @@ 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& interests, std::uint64_t deadlineNanos); } extern "C" bool GuestSocketIsOpen_nid_no_patch(int descriptor); diff --git a/core/libs/prx/libkernel/Socket/src/Udp.cpp b/core/libs/prx/libkernel/Socket/src/Udp.cpp index 8cb6d12cc..9ff5d7b33 100644 --- a/core/libs/prx/libkernel/Socket/src/Udp.cpp +++ b/core/libs/prx/libkernel/Socket/src/Udp.cpp @@ -5,6 +5,8 @@ #else #include #include +#include +#include #include #include #include @@ -13,6 +15,8 @@ #include "prx/libc/include/GuestArena.hpp" #include "prx/libc/include/general/VabiMacros.hpp" #include "prx/libkernel/Socket/include/SocketRuntime.hpp" +#include "prx/libkernel/Equeue/Equeue.hpp" +#include "prx/libkernel/Time/include/TimedWait.hpp" #include #include #include @@ -109,6 +113,43 @@ struct Socket { #endif } }; +struct Waker { + NativeSocket value = Invalid; + Waker() { + value = ::socket(AF_INET, SOCK_DGRAM, 0); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + socklen_t size = sizeof(address); +#ifdef _WIN32 + u_long nonblocking = 1; + const bool configured = value != Invalid && ioctlsocket(value, FIONBIO, &nonblocking) == 0; +#else + const bool configured = value != Invalid && ::fcntl(value, F_SETFL, ::fcntl(value, F_GETFL) | O_NONBLOCK) == 0; +#endif + if (!configured || bind(value, reinterpret_cast(&address), size) != 0 || + getsockname(value, reinterpret_cast(&address), &size) != 0 || + connect(value, reinterpret_cast(&address), size) != 0) { + const int error = NativeError(); + Close(); + throw std::runtime_error("equeue waker: loopback socket setup failed, guest errno " + std::to_string(error)); + } + } + ~Waker() { Close(); } + void Close() { + if (value == Invalid) return; +#ifdef _WIN32 + closesocket(value); +#else + ::close(value); +#endif + value = Invalid; + } +}; +Waker& ThreadWaker() { + thread_local Waker waker; + return waker; +} std::mutex socketsMutex; std::map> sockets; int nextDescriptor = GuestSockets::FirstDescriptor; @@ -173,8 +214,68 @@ void GuestAddress(const sockaddr_storage& native, void* output, std::uint32_t* l } int GuestSockets::Close(int descriptor) { - std::lock_guard lock(socketsMutex); - return sockets.erase(descriptor) ? 0 : Fail(9); + { + std::lock_guard lock(socketsMutex); + if (!sockets.erase(descriptor)) return Fail(9); + } + EqueueDescriptorClosed(descriptor); + return 0; +} + +std::uintptr_t GuestSockets::CurrentWaker() { + return static_cast(ThreadWaker().value); +} + +void GuestSockets::Wake(std::uintptr_t waker) { + const char signal = 0; + if (send(static_cast(waker), &signal, 1, 0) < 0 && NativeError() != 35) + throw std::runtime_error("equeue waker: send failed, guest errno " + std::to_string(NativeError())); +} + +bool GuestSockets::WaitAny(const std::vector& interests, std::uint64_t deadlineNanos) { + const NativeSocket waker = ThreadWaker().value; + std::vector> held; +#ifdef _WIN32 + std::vector entries; + constexpr SHORT ReadMask = POLLRDNORM; + constexpr SHORT WriteMask = POLLWRNORM; +#else + std::vector entries; + constexpr short ReadMask = POLLIN; + constexpr short WriteMask = POLLOUT; +#endif + { + std::lock_guard lock(socketsMutex); + for (const auto& interest : interests) { + const auto found = sockets.find(interest.descriptor); + if (found == sockets.end()) continue; + held.push_back(found->second); + entries.push_back({found->second->value, interest.write ? WriteMask : ReadMask, 0}); + } + } + entries.push_back({waker, ReadMask, 0}); + std::uint64_t remaining = 0; + if (deadlineNanos != 0) { + const std::uint64_t now = TimedWait::NowNanos(); + if (now >= deadlineNanos) return true; + remaining = deadlineNanos - now; + } +#ifdef _WIN32 + INT timeout = -1; + if (deadlineNanos != 0) { + if (remaining < 1000000) return false; + timeout = static_cast(std::min(remaining / 1000000, INT_MAX)); + } + if (WSAPoll(entries.data(), static_cast(entries.size()), timeout) == SOCKET_ERROR) + throw std::runtime_error("equeue wait: WSAPoll failed, guest errno " + std::to_string(NativeError())); +#else + timespec timeout{static_cast(remaining / 1000000000), static_cast(remaining % 1000000000)}; + if (::ppoll(entries.data(), entries.size(), deadlineNanos != 0 ? &timeout : nullptr, nullptr) < 0 && errno != EINTR) + throw std::runtime_error("equeue wait: ppoll failed, guest errno " + std::to_string(NativeError())); +#endif + char drained[16]; + while (recv(waker, drained, sizeof(drained), 0) > 0) {} + return true; } bool GuestSockets::IsOpen(int descriptor) { @@ -361,8 +462,9 @@ bool GuestSockets::Ready(int descriptor, bool write, std::int64_t* data, bool* e WSAPOLLFD entry{socket->value, static_cast(write ? POLLWRNORM : POLLRDNORM), 0}; if (WSAPoll(&entry, 1, 0) <= 0) return false; #else - pollfd entry{socket->value, static_cast(write ? POLLOUT : POLLIN), 0}; + pollfd entry{socket->value, static_cast(write ? POLLOUT : (POLLIN | POLLRDHUP)), 0}; if (::poll(&entry, 1, 0) <= 0) return false; + if (!write && (entry.revents & POLLRDHUP)) entry.revents |= POLLHUP; #endif *eof = (entry.revents & (POLLHUP | POLLERR)) != 0; *data = 0; diff --git a/core/libs/tests/GuestSocketEvents.cpp b/core/libs/tests/GuestSocketEvents.cpp index 011577219..94bf0656d 100644 --- a/core/libs/tests/GuestSocketEvents.cpp +++ b/core/libs/tests/GuestSocketEvents.cpp @@ -1,8 +1,10 @@ #include "SceTypes.hpp" #include +#include #include #include #include +#include extern "C" { int APS5_VABI socket_nid_postfix(int, int, int); @@ -93,7 +95,41 @@ int main() { Require(close_nid_postfix(client) == 0); Require(sceKernelWaitEqueue(eq, events.data(), events.size(), &count, &timeout) == SCE_OK); - Require(count == 1 && events[0].data == 0); + Require(count == 1 && events[0].data == 0 && (events[0].flags & EV_EOF) != 0); + + KernelEqueue blocking = 0; + Require(sceKernelCreateEqueue(&blocking, "blocking") == SCE_OK); + const int writer = socket_nid_postfix(2, 1, 0); + Require(writer >= 0 && connect_nid_postfix(writer, address.data(), address.size()) == 0); + const int reader = accept_nid_postfix(listener, nullptr, nullptr); + Require(reader >= 0); + Require(sceKernelAddReadEvent(blocking, reader, 1, nullptr) == SCE_OK); + std::thread sender([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + Require(send_nid_postfix(writer, message, sizeof(message), 0) == sizeof(message)); + }); + const auto start = std::chrono::steady_clock::now(); + Require(sceKernelWaitEqueue(blocking, events.data(), events.size(), &count, nullptr) == SCE_OK); + sender.join(); + Require(std::chrono::steady_clock::now() - start >= std::chrono::milliseconds(40)); + Require(count == 1 && events[0].ident == static_cast(reader) && events[0].data == sizeof(message)); + Require(recv_nid_postfix(reader, received, sizeof(received), 0) == sizeof(received)); + + std::thread waiter([&] { + std::array woken{}; + int wokenCount = 0; + Require(sceKernelWaitEqueue(blocking, woken.data(), woken.size(), &wokenCount, nullptr) == SCE_OK); + Require(wokenCount == 1 && woken[0].filter == EVFILT_WRITE && woken[0].ident == static_cast(writer)); + }); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + Require(sceKernelAddWriteEvent(blocking, writer, 0, nullptr) == SCE_OK); + waiter.join(); + Require(sceKernelDeleteWriteEvent(blocking, writer) == SCE_OK); + + Require(close_nid_postfix(reader) == 0); + Require(sceKernelDeleteReadEvent(blocking, reader) == SCE_KERNEL_ERROR_ENOENT); + Require(close_nid_postfix(writer) == 0); + Require(sceKernelDeleteEqueue(blocking) == SCE_OK); Require(sceKernelAddReadEvent(eq, client, 1, nullptr) == SCE_KERNEL_ERROR_EBADF); Require(Rejects(sceKernelAddReadEvent, eq, 0, 1)); diff --git a/docs/dev/TechnicalDebt.md b/docs/dev/TechnicalDebt.md index 2fe26defd..f06d91026 100644 --- a/docs/dev/TechnicalDebt.md +++ b/docs/dev/TechnicalDebt.md @@ -332,7 +332,7 @@ Throughout the project, every function at every stage either **does exactly what - [Pending storage-image results](../../core/libs/prx/libSceAgcDriver/Graphics/src/Texture.cpp) (libSceAgcDriver) are stored when something reads their memory, not when the GPU work that wrote them completes. A title that waits for that work, frees the image and reuses its memory can have the new data overwritten by the late store (#1970). `APS5_STORE_AT_FLIP=1` stores every pending result at each flip; storing them at the title's GPU waits (labels, fences) is unimplemented. - Windows host-pointer imports of the guest mapping itself are compared instead of write-watched: a driver can mark imported pages dirty after the synchronous import window. Shared memory imported through a separate writable alias retains protection-based tracking on the guest views; driver access to the import alias does not disarm those views. [GuestMemory](../../core/libs/prx/libSceAgcDriver/Execution/src/GuestMemory.cpp) excludes every overlapping 64 KiB tracker block while its backing is retained. A full block replaced by fresh private backing regains tracking and gets a new generation; protection changes, partial replacements and shared remappings do not restore it. This preserves CPU edits through byte comparisons but disables snapshot reuse and unit shadows for excluded blocks; restoring tracking on import retirement without replacing the backing and measuring the comparison cost remain unimplemented. - [shm_open](../../core/libs/prx/libkernel/File/src/SharedMemory.cpp) (libkernel) creates only anonymous objects (`SHM_ANON`), each backed by a host anonymous file: a `memfd` on Linux, a delete-on-close temporary file on Windows. The flag checks follow FreeBSD 11's `kern_shm_open`; the mode is ignored. Named objects throw. The descriptor cannot be mapped yet: `mmap` accepts only private anonymous mappings and fails a descriptor mapping with `EOPNOTSUPP`, so the known caller (the bundled Mesa in PPSA59322, which truncates the object and maps it `MAP_SHARED`) gets that failure -- [sceKernelAddReadEvent and sceKernelAddWriteEvent](../../core/libs/prx/libkernel/Equeue/Equeue.cpp) (libkernel) poll the host socket instead of being woken by it: while such an event is registered, every wait on its equeue checks the socket at least once per millisecond, so readiness is reported up to 1 ms late and an idle wait wakes the thread 1000 times a second. Only guest sockets are supported; other descriptors and a low watermark above 1 throw +- [sceKernelAddReadEvent and sceKernelAddWriteEvent](../../core/libs/prx/libkernel/Equeue/Equeue.cpp) (libkernel): a write event reports the socket's send buffer size (`SO_SNDBUF`) in `data`, not the free space left in it, as Windows has no query for that space. Only guest sockets are supported; other descriptors and a low watermark above 1 throw - [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 - [Additional content](../../core/libs/prx/libSceAppContent/Export.cpp) (libSceAppContent) is not installed or mounted: `sceAppContentAddcontMount` answers `NOT_FOUND` for every entitlement label and `sceAppContentAddcontUnmount` for every mount point, as no add-content mount is ever handed out; `sceAppContentGetAddcontInfo` answers `DRM_NO_ENTITLEMENT` for every label and `sceAppContentGetAddcontInfoList` reports 0 entries - [sceKernelPread, sceKernelPwrite, sceKernelPreadv, sceKernelPwritev](../../core/libs/prx/libkernel/File/src/Stdio.cpp) (libkernel) on Windows transfer at the given offset with an overlapped `ReadFile`/`WriteFile` and restore the descriptor's file pointer afterwards, so a concurrent `read`/`write` on the same descriptor can see the moved pointer in between, unlike on the console. Handles that are not disk files return `ESPIPE`. `readv` on a pipe or console handle stops after the first iovec read, a short read the console is assumed to allow; each iovec is transferred separately, so a failure after the first transfer returns the bytes already done