fix(libkernel): defer raises while a thread holds the address-wait lock on Windows

core/libs/prx/libkernel/SyncOnAddress/src/SyncOnAddress.cpp

@@ -59,12 +59,26 @@ bool IsAlignedAddress(std::uintptr_t address) {
     return address != 0 && address % alignof(TValue) == 0;
 }
 
+template <class TFunction>
+void DeferringDelivery(TFunction function) {
+#ifdef _WIN32
+    {
+        TimedWait::DeliveryDeferral deferral;
+        function();
+    }
+    TimedWait::DeliverDeferred();
+#else
+    function();
+#endif
+}
+
 template <class TValue>
-bool WaitOnAddressUntil(TValue* address, TValue expected, std::optional<std::uint64_t> deadlineNanos, const void* caller) {
+std::optional<std::chrono::nanoseconds> WaitOnAddressLocked(TValue* address, TValue expected, std::optional<std::uint64_t> deadlineNanos, bool& woken) {
     const auto key = reinterpret_cast<std::uintptr_t>(address);
     std::unique_lock<std::mutex> lock(g_waitersLock);
     if (std::atomic_ref<TValue>(*address).load() != expected) {
-        return true;
+        woken = true;
+        return std::nullopt;
     }
 
     AddressWaiter waiter;
@@ -79,15 +93,24 @@ bool WaitOnAddressUntil(TValue* address, TValue expected, std::optional<std::uin
     }
     const auto waited = std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now() - waitStart);
 
-    const bool woken = waiter.woken;
+    woken = waiter.woken;
     if (!woken) {
         queue.erase(position);
         if (queue.empty()) {
             g_waiters.erase(key);
         }
     }
-    lock.unlock();
-    KernelTraceWait_nid_postfix("addr", caller, static_cast<std::uint64_t>(waited.count()), !woken);
+    return waited;
+}
+
+template <class TValue>
+bool WaitOnAddressUntil(TValue* address, TValue expected, std::optional<std::uint64_t> deadlineNanos, const void* caller) {
+    bool woken = false;
+    std::optional<std::chrono::nanoseconds> waited;
+    DeferringDelivery([&] { waited = WaitOnAddressLocked(address, expected, deadlineNanos, woken); });
+    if (waited) {
+        KernelTraceWait_nid_postfix("addr", caller, static_cast<std::uint64_t>(waited->count()), !woken);
+    }
     return woken;
 }
 
@@ -97,7 +120,7 @@ int WaitOnAddress(TValue* address, TValue expected, const KernelUseconds* timeou
     return WaitOnAddressUntil(address, expected, deadline, caller) ? SYNC_ON_ADDRESS_OK : SCE_KERNEL_ERROR_ETIMEDOUT;
 }
 
-void WakeAddress(std::uintptr_t key, std::int32_t count, bool wakesAtLeastOne) {
+void WakeAddressLocked(std::uintptr_t key, std::int32_t count, bool wakesAtLeastOne) {
     std::lock_guard<std::mutex> lock(g_waitersLock);
     const auto entry = g_waiters.find(key);
     if (entry == g_waiters.end()) {
@@ -115,6 +138,10 @@ void WakeAddress(std::uintptr_t key, std::int32_t count, bool wakesAtLeastOne) {
     }
 }
 
+void WakeAddress(std::uintptr_t key, std::int32_t count, bool wakesAtLeastOne) {
+    DeferringDelivery([&] { WakeAddressLocked(key, count, wakesAtLeastOne); });
+}
+
 std::uint64_t SaturatingNanos(std::int64_t seconds, std::int64_t nanos) {
     if (seconds < 0) {
         return 0;

core/libs/prx/libkernel/Time/TimedWait.cpp

@@ -241,6 +241,18 @@ void Condition::NotifyAll() {
     }
 }
 
+DeliveryDeferral::DeliveryDeferral() {
+    WaitState().bound->fetch_add(1, std::memory_order_seq_cst);
+}
+
+DeliveryDeferral::~DeliveryDeferral() {
+    WaitState().bound->fetch_sub(1, std::memory_order_seq_cst);
+}
+
+void DeliverDeferred() {
+    DrainApcs();
+}
+
 namespace {
 
 struct alignas(64) ParkBucket {
@@ -256,11 +268,6 @@ ParkBucket& ParkBucketFor(const void* key) {
     return parkBuckets[(reinterpret_cast<std::uintptr_t>(key) * 0x9E3779B97F4A7C15ULL) >> 56];
 }
 
-struct DeliveryDeferral {
-    DeliveryDeferral() { WaitState().bound->fetch_add(1, std::memory_order_seq_cst); }
-    ~DeliveryDeferral() { WaitState().bound->fetch_sub(1, std::memory_order_seq_cst); }
-};
-
 void UnlinkParked(ParkBucket& bucket, Waiter* waiter) {
     if (waiter->previous) waiter->previous->next = waiter->next;
     else bucket.head = waiter->next;

core/libs/prx/libkernel/Time/include/TimedWait.hpp

@@ -22,6 +22,15 @@ struct Waiter;
 #ifdef _WIN32
 constexpr unsigned PARK_SPINS = 1000;
 
+class DeliveryDeferral {
+public:
+    DeliveryDeferral();
+    ~DeliveryDeferral();
+    DeliveryDeferral(const DeliveryDeferral&) = delete;
+    DeliveryDeferral& operator=(const DeliveryDeferral&) = delete;
+};
+
+void DeliverDeferred();
 void SpinPause();
 Waiter* BeginPark(const void* key);
 bool FinishPark(const void* key, Waiter* waiter, const std::uint64_t* deadlineNanos);

core/libs/tests/GuestRaiseException.cpp

@@ -3,6 +3,7 @@
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
+#include <cstdio>
 #include <cstdlib>
 #include <cstring>
 #include <mutex>
@@ -33,6 +34,7 @@ int APS5_VABI scePthreadRwlockDestroy(PthreadRwlock* rwlock);
 int APS5_VABI scePthreadRwlockWrlock(PthreadRwlock* rwlock);
 int APS5_VABI scePthreadRwlockUnlock(PthreadRwlock* rwlock);
 int APS5_VABI sceKernelSyncOnAddressWait(std::uint32_t* address, std::uint32_t expected, const KernelUseconds* timeout, const char* name);
+int APS5_VABI sceKernelSyncOnAddressWake(void* address, std::int32_t count);
 }
 
 static constexpr int SCE_KERNEL_ERROR_EINVAL = static_cast<int>(0x80020016);
@@ -402,6 +404,87 @@ static void NestedWaitKeepsOuterWake() {
     Require(sceKernelDeleteSema(sem) == 0);
 }
 
+#ifdef _WIN32
+static constexpr int AddressRounds = 100;
+static std::uint32_t addressWord = 0;
+static std::uint32_t otherAddressWord = 0;
+static std::atomic<bool> holdingEntered{false};
+static std::atomic<bool> holdingRelease{false};
+static std::atomic<bool> holdingLeft{false};
+static std::atomic<std::thread::id> holdingThread;
+
+static void APS5_VABI HoldingHandler(int signum, void*) {
+    Require(signum == SIGUSR1);
+    holdingThread.store(std::this_thread::get_id());
+    holdingEntered.store(true);
+    while (!holdingRelease.load()) std::this_thread::yield();
+    holdingLeft.store(true);
+}
+
+static void* APS5_VABI AddressWaking(void* arg) {
+    auto& worker = *static_cast<Worker*>(arg);
+    worker.id = std::this_thread::get_id();
+    worker.started.store(true);
+    while (!worker.stop.load()) Require(sceKernelSyncOnAddressWake(&addressWord, 1) == 0);
+    return nullptr;
+}
+
+static void* APS5_VABI AddressWaiting(void* arg) {
+    auto& worker = *static_cast<Worker*>(arg);
+    worker.id = std::this_thread::get_id();
+    worker.started.store(true);
+    while (!worker.stop.load()) {
+        KernelUseconds timeout = 20;
+        const int result = sceKernelSyncOnAddressWait(&addressWord, 0, &timeout, "address waiting");
+        Require(result == 0 || result == SCE_KERNEL_ERROR_ETIMEDOUT);
+    }
+    return nullptr;
+}
+
+static bool AwaitWithin(const std::atomic<bool>& flag, std::chrono::milliseconds limit) {
+    const auto deadline = std::chrono::steady_clock::now() + limit;
+    while (!flag.load()) {
+        if (std::chrono::steady_clock::now() >= deadline) return false;
+        std::this_thread::yield();
+    }
+    return true;
+}
+
+static void RequireInRound(bool value, const char* name, int round, const char* check) {
+    if (value) return;
+    std::fprintf(stderr, "%s round %d: %s\n", name, round, check);
+    std::fflush(stderr);
+    std::abort();
+}
+
+static void HandlerLeavesAddressLock(PthreadEntry entry, const char* name) {
+    Worker worker;
+    Pthread thread = nullptr;
+    Require(scePthreadCreate(&thread, nullptr, entry, &worker, name) == 0);
+    while (!worker.started.load()) std::this_thread::yield();
+    for (int round = 0; round < AddressRounds; ++round) {
+        holdingEntered.store(false);
+        holdingLeft.store(false);
+        holdingRelease.store(false);
+        RequireInRound(sceKernelRaiseException(thread, SIGUSR1) == 0, name, round, "the raise failed");
+        RequireInRound(AwaitWithin(holdingEntered, std::chrono::seconds(5)), name, round, "the handler did not run within 5 s");
+        RequireInRound(holdingThread.load() == worker.id, name, round, "the handler ran on another thread");
+        std::atomic<bool> woke{false};
+        std::thread waker([&] {
+            Require(sceKernelSyncOnAddressWake(&otherAddressWord, 1) == 0);
+            woke.store(true);
+        });
+        const bool returned = AwaitWithin(woke, std::chrono::milliseconds(500));
+        holdingRelease.store(true);
+        waker.join();
+        RequireInRound(returned, name, round, "a wake on another address blocked for 500 ms while the handler ran");
+        RequireInRound(AwaitWithin(holdingLeft, std::chrono::seconds(5)), name, round, "the handler did not return within 5 s of its release");
+    }
+    worker.stop.store(true);
+    Require(scePthreadJoin(thread, nullptr) == 0);
+}
+#endif
+
 static void ExpectDelivery(int before, std::thread::id thread) {
     for (int attempt = 0; attempt < 5000 && calls.load() == before; ++attempt) std::this_thread::sleep_for(std::chrono::milliseconds(1));
     Require(calls.load() == before + 1);
@@ -660,4 +743,11 @@ int main() {
     NestedWaitKeepsLaterWaiters();
     NestedWaitKeepsOuterWake();
     Require(sceKernelRemoveExceptionHandler(SIGUSR1) == 0);
+
+#ifdef _WIN32
+    Require(sceKernelInstallExceptionHandler(SIGUSR1, reinterpret_cast<void*>(&HoldingHandler)) == 0);
+    HandlerLeavesAddressLock(AddressWaking, "address waking");
+    HandlerLeavesAddressLock(AddressWaiting, "address waiting");
+    Require(sceKernelRemoveExceptionHandler(SIGUSR1) == 0);
+#endif
 }

docs/dev/debt/libkernel.md

@@ -129,7 +129,7 @@
 - [sceKernelRaiseException](../../../core/libs/prx/libkernel/System/src/Exception.cpp) does not consult the target thread's signal mask
 - [pthread_sigmask](../../../core/libs/prx/libkernel/System/src/Signals.cpp) (libkernel) shares the process-wide mask of `sigprocmask` and `_sigprocmask`, which keeps only the first 32 signals; FreeBSD keeps a mask per thread
 - [sceKernelRaiseException](../../../core/libs/prx/libkernel/System/src/Exception.cpp) reaches a thread that waits in an alertable `TimedWait` wait through an APC. A thread blocked anywhere else, including all waits under `APS5_COARSE_TIMED_WAITS` and host waits inside the emulator, is redirected and runs the handler only when that wait returns
-- [sceKernelRaiseException](../../../core/libs/prx/libkernel/System/src/Exception.cpp) on Windows runs the handler wherever the target thread was interrupted. It waits only while the thread is inside winpthreads or restoring a context through `NtContinue`, and throws if that lasts 1 s, so a handler that calls into other host code the thread was interrupted in, such as the C runtime's heap or a lock of the emulator, can deadlock on it or reenter it
+- [sceKernelRaiseException](../../../core/libs/prx/libkernel/System/src/Exception.cpp) on Windows runs the handler wherever the target thread was interrupted. It waits only while the thread is inside winpthreads or restoring a context through `NtContinue`, and throws if that lasts 1 s, so a handler that calls into other host code the thread was interrupted in, such as the C runtime's heap or a lock of the emulator, can deadlock on it or reenter it. A thread that holds the address-wait lock of `sceKernelSyncOnAddressWait`, `sceKernelSyncOnAddressWake` and `_umtx_op`, or the park queue lock of the guest mutexes, gets the raise as an APC instead, which runs once it has released the lock or while it waits; the other locks of the emulator are not deferred this way
 - [sigaltstack](../../../core/libs/prx/libkernel/System/src/Signals.cpp) (libkernel) records each thread's alternate signal stack and reports it back, following FreeBSD's `kern_sigaltstack` (`EINVAL` for flags other than `SS_DISABLE`, `ENOMEM` below `MINSIGSTKSZ` 2048), but handlers never run on it, so `SS_ONSTACK` and `EPERM` are never reported
 - [sigpending](../../../core/libs/prx/libkernel/System/src/Signals.cpp) (libkernel) reports the intersection of the process-wide pending set of `sigtimedwait` and the blocked mask of `_sigprocmask`, as FreeBSD's `sys_sigpending` intersects the pending list with `td_sigmask`, and returns -1 with `EFAULT` for a null set. Because `_sigprocmask` keeps only the first 32 signals, signals 33 to 128 are never blocked and never appear in the set, and setting `SIG_IGN` (or `SIG_DFL` on a signal whose default action is to ignore) does not clear a signal that was already marked pending. PPSA99420 imports it
 - [scePthreadJoin, scePthreadCreate](../../../core/libs/prx/libkernel/Pthread/src/Thread.cpp) (libkernel) - on a PS5 (firmware 13.60) joining a thread a second time returns `ESRCH` (`0x80020003`); here a joined thread is released, so a second join reads freed memory. A null entry crashed the payload and still throws. Joining the calling thread returns `EDEADLK` and a zeroed attribute creates a default thread, as on the console