fix(libkernel): preserve condition signals across cancellation

core/libs/prx/libkernel/Pthread/include/Pthread.hpp

@@ -10,6 +10,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <ctime>
+#include <memory>
 #include <mutex>
 #include <string>
 #include <thread>
@@ -61,7 +62,7 @@ struct PthreadCondattrPrivate {
 };
 
 struct PthreadCondPrivate {
-    TimedWait::Condition _cv;
+    std::shared_ptr<TimedWait::Condition> _cv = std::make_shared<TimedWait::Condition>();
     int _clockid = 0;
 };
 

core/libs/prx/libkernel/Pthread/src/Cond.cpp

@@ -2,6 +2,7 @@
 #include "prx/libkernel/Pthread/include/Mutex.hpp"
 #include "prx/libkernel/Pthread/include/Cond.hpp"
 #include "prx/libkernel/Pthread/include/Cancel.hpp"
+#include "prx/libkernel/Pthread/include/ThreadLifecycle.hpp"
 #include "prx/libkernel/Pthread/Posix/Common.hpp"
 #include "prx/libkernel/KernelErrors.hpp"
 #include "prx/libkernel/Time/include/Time.hpp"
@@ -13,6 +14,7 @@
 #include <optional>
 #include <stdexcept>
 #include <thread>
+#include <utility>
 
 namespace {
 
@@ -31,21 +33,20 @@ PthreadMutex adaptiveInitializer() {
     return reinterpret_cast<PthreadMutex>(std::uintptr_t{1});
 }
 
-PthreadCond resolveCond(PthreadCond* cond) {
+std::shared_ptr<TimedWait::Condition> resolveCond(PthreadCond* cond, int* clockId = nullptr) {
     if (!cond)
         throw std::invalid_argument("Condition variable pointer is null");
-    std::atomic_ref<PthreadCond> slot(*cond);
-    if (const auto current = slot.load(std::memory_order_acquire); current && current != destroyedCond())
-        return current;
     std::lock_guard lock(condInitializationMutex);
-    const auto current = slot.load(std::memory_order_acquire);
+    std::atomic_ref<PthreadCond> slot(*cond);
+    auto current = slot.load(std::memory_order_acquire);
     if (current == destroyedCond())
         throw std::runtime_error("Condition variable has been destroyed");
-    if (current)
-        return current;
-    auto* created = new PthreadCondPrivate();
-    slot.store(created, std::memory_order_release);
-    return created;
+    if (!current) {
+        current = new PthreadCondPrivate();
+        slot.store(current, std::memory_order_release);
+    }
+    if (clockId) *clockId = current->_clockid;
+    return current->_cv;
 }
 
 int mutexOwnership(PthreadMutex* mutex, PthreadMutex* owned) {
@@ -73,8 +74,16 @@ struct WakingLock {
     }
 };
 
-int waitUntil(PthreadCond* cond, PthreadMutex* mutex, std::optional<std::uint64_t> deadlineNanos, const void* caller) {
-    auto* c = resolveCond(cond);
+int finishWait(std::shared_ptr<TimedWait::Condition>& condition, bool timedOut) {
+    if (ThreadCancel::Requested()) {
+        condition->NotifyOne();
+        condition.reset();
+        scePthreadExit(GUEST_CANCELED);
+    }
+    return timedOut ? sceTimedOut : 0;
+}
+
+int waitUntil(std::shared_ptr<TimedWait::Condition> condition, PthreadMutex* mutex, std::optional<std::uint64_t> deadlineNanos, const void* caller) {
     PthreadMutex m = nullptr;
     if (const int error = mutexOwnership(mutex, &m))
         return error;
@@ -92,32 +101,32 @@ int waitUntil(PthreadCond* cond, PthreadMutex* mutex, std::optional<std::uint64_
         m->_owner.store(std::thread::id{}, std::memory_order_release);
         for (int level = 1; level < previousCount; ++level)
             m->_rmtx.unlock();
-        timedOut = !ThreadCancel::Wait(c->_cv, lock, deadlineNanos);
+        timedOut = !ThreadCancel::Wait(*condition, lock, deadlineNanos);
         for (int level = 1; level < previousCount; ++level)
             m->_rmtx.lock();
         m->_owner.store(std::this_thread::get_id(), std::memory_order_release);
         m->_count = previousCount;
         lock.release();
-        ThreadCancel::Check();
-        return timedOut ? sceTimedOut : 0;
+        return finishWait(condition, timedOut);
     }
     WakingLock<std::timed_mutex> waking{m, m->_mtx};
     std::unique_lock lock(waking, std::adopt_lock);
     m->_owner.store(std::thread::id{}, std::memory_order_release);
-    timedOut = !ThreadCancel::Wait(c->_cv, lock, deadlineNanos);
+    timedOut = !ThreadCancel::Wait(*condition, lock, deadlineNanos);
     m->_owner.store(std::this_thread::get_id(), std::memory_order_release);
     lock.release();
-    ThreadCancel::Check();
-    return timedOut ? sceTimedOut : 0;
+    return finishWait(condition, timedOut);
 }
 
 }
 
 int CondOperations::AbsoluteTimedwait(PthreadCond* cond, PthreadMutex* mutex, const KernelTimespec* abstime) {
     KernelUseconds usec = 0;
-    if (!PosixThread::RelativeMicroseconds(resolveCond(cond)->_clockid, abstime, &usec))
+    int clockId = 0;
+    auto condition = resolveCond(cond, &clockId);
+    if (!PosixThread::RelativeMicroseconds(clockId, abstime, &usec))
         throw std::invalid_argument("Invalid absolute condition variable timeout");
-    return waitUntil(cond, mutex, TimedWait::DeadlineNanos(usec), __builtin_return_address(0));
+    return waitUntil(std::move(condition), mutex, TimedWait::DeadlineNanos(usec), __builtin_return_address(0));
 }
 
 extern "C" {
@@ -169,27 +178,27 @@ int APS5_VABI scePthreadCondDestroy(PthreadCond* cond) {
 }
 
 int APS5_VABI scePthreadCondSignal(PthreadCond* cond) {
-    resolveCond(cond)->_cv.NotifyOne();
+    resolveCond(cond)->NotifyOne();
     return 0;
 }
 
 int APS5_VABI scePthreadCondBroadcast(PthreadCond* cond) {
-    resolveCond(cond)->_cv.NotifyAll();
+    resolveCond(cond)->NotifyAll();
     return 0;
 }
 
 int APS5_VABI scePthreadCondSignalto(PthreadCond* cond, Pthread thread) {
     (void)thread;
-    resolveCond(cond)->_cv.NotifyAll();
+    resolveCond(cond)->NotifyAll();
     return 0;
 }
 
 int APS5_VABI scePthreadCondWait(PthreadCond* cond, PthreadMutex* mutex) {
-    return waitUntil(cond, mutex, std::nullopt, __builtin_return_address(0));
+    return waitUntil(resolveCond(cond), mutex, std::nullopt, __builtin_return_address(0));
 }
 
 int APS5_VABI scePthreadCondTimedwait(PthreadCond* cond, PthreadMutex* mutex, KernelUseconds usec) {
-    return waitUntil(cond, mutex, TimedWait::DeadlineNanos(usec), __builtin_return_address(0));
+    return waitUntil(resolveCond(cond), mutex, TimedWait::DeadlineNanos(usec), __builtin_return_address(0));
 }
 
 }

core/libs/tests/GuestPthreadCancelRecovery.cpp

@@ -1,10 +1,21 @@
 #include "SceTypes.hpp"
+#include "prx/libkernel/Pthread/include/Pthread.hpp"
 #include <array>
 #include <atomic>
+#include <chrono>
 #include <cstdint>
 #include <cstdio>
 #include <cstdlib>
 #include <source_location>
+#include <thread>
+#include <memory>
+
+#ifdef _WIN32
+#include <windows.h>
+#else
+#include <sys/mman.h>
+#include <unistd.h>
+#endif
 
 extern "C" {
 int APS5_VABI scePthreadCreate(Pthread* thread, const PthreadAttr* attr, PthreadEntry entry, void* arg, const char* name);
@@ -17,13 +28,20 @@ int APS5_VABI scePthreadKeyCreate(PthreadKey* key, pthread_key_destructor_func_t
 int APS5_VABI scePthreadKeyDelete(PthreadKey key);
 int APS5_VABI scePthreadSetspecific(PthreadKey key, void* value);
 int APS5_VABI scePthreadMutexInit(PthreadMutex* mutex, const PthreadMutexattr* attr, const char* name);
+int APS5_VABI scePthreadMutexattrInit(PthreadMutexattr* attr);
+int APS5_VABI scePthreadMutexattrDestroy(PthreadMutexattr* attr);
+int APS5_VABI scePthreadMutexattrSettype(PthreadMutexattr* attr, int type);
 int APS5_VABI scePthreadMutexDestroy(PthreadMutex* mutex);
 int APS5_VABI scePthreadMutexLock(PthreadMutex* mutex);
 int APS5_VABI scePthreadMutexUnlock(PthreadMutex* mutex);
 int APS5_VABI scePthreadCondInit(PthreadCond* cond, const PthreadCondattr* attr, const char* name);
 int APS5_VABI scePthreadCondDestroy(PthreadCond* cond);
 int APS5_VABI scePthreadCondSignal(PthreadCond* cond);
 int APS5_VABI scePthreadCondTimedwait(PthreadCond* cond, PthreadMutex* mutex, KernelUseconds usec);
+int APS5_VABI scePthreadCondWait(PthreadCond* cond, PthreadMutex* mutex);
+int APS5_VABI pthread_cond_wait_nid_postfix(PthreadCond* cond, PthreadMutex* mutex);
+int APS5_VABI pthread_cond_timedwait_nid_postfix(PthreadCond* cond, PthreadMutex* mutex, const KernelTimespec* abstime);
+int APS5_VABI clock_gettime_nid_postfix(int clockId, KernelTimespec* time);
 int APS5_VABI scePthreadSemInit(PthreadSem* sem, int flag, unsigned int value, const char* name);
 int APS5_VABI scePthreadSemDestroy(PthreadSem* sem);
 int APS5_VABI scePthreadSemWait(PthreadSem* sem);
@@ -226,6 +244,166 @@ static void ConditionRecovery() {
     Require(scePthreadMutexDestroy(&mutex) == 0);
 }
 
+enum class ConditionWaitMode { SceTimed, Sce, PosixTimed, Posix };
+
+struct SignaledCondition {
+    PthreadMutex mutex = nullptr;
+    PthreadCond condition = nullptr;
+    PthreadCond* slot = &condition;
+    ConditionWaitMode mode = ConditionWaitMode::SceTimed;
+    int lockCount = 1;
+    bool cancellationDisabled = false;
+    bool released = false;
+    std::atomic<int> ready{0};
+    std::array<int, 2> results{-1, -1};
+    std::array<bool, 2> cleaned{};
+};
+
+struct SignaledConditionWorker {
+    SignaledCondition* context;
+    int index;
+};
+
+static void APS5_VABI UnlockSignaledCondition(void* value) {
+    auto& worker = *static_cast<SignaledConditionWorker*>(value);
+    auto& context = *worker.context;
+    for (int level = 0; level < context.lockCount; ++level) Require(scePthreadMutexUnlock(&context.mutex) == 0);
+    Require(scePthreadMutexUnlock(&context.mutex) == static_cast<int>(0x80020001));
+    context.cleaned[worker.index] = true;
+}
+
+static void* APS5_VABI SignaledConditionWaiter(void* value) {
+    auto& worker = *static_cast<SignaledConditionWorker*>(value);
+    auto& context = *worker.context;
+    if (context.cancellationDisabled) Require(scePthreadSetcancelstate(CANCEL_DISABLE, nullptr) == 0);
+    for (int level = 0; level < context.lockCount; ++level) Require(scePthreadMutexLock(&context.mutex) == 0);
+    std::uintptr_t cleanup[8];
+    __pthread_cleanup_push_imp_nid_postfix(UnlockSignaledCondition, &worker, cleanup);
+    context.ready.fetch_add(1);
+    int result = 0;
+    do {
+        switch (context.mode) {
+        case ConditionWaitMode::SceTimed:
+            result = scePthreadCondTimedwait(context.slot, &context.mutex, 2000000);
+            break;
+        case ConditionWaitMode::Sce:
+            result = scePthreadCondWait(context.slot, &context.mutex);
+            break;
+        case ConditionWaitMode::PosixTimed: {
+            KernelTimespec deadline{};
+            Require(clock_gettime_nid_postfix(0, &deadline) == 0);
+            deadline.tv_sec += 2;
+            result = pthread_cond_timedwait_nid_postfix(context.slot, &context.mutex, &deadline);
+            break;
+        }
+        case ConditionWaitMode::Posix:
+            result = pthread_cond_wait_nid_postfix(context.slot, &context.mutex);
+            break;
+        }
+    } while (result == 0 && !context.released);
+    context.results[worker.index] = result;
+    __pthread_cleanup_pop_imp_nid_postfix(1);
+    return nullptr;
+}
+
+static bool ConditionRegistered(Pthread thread) {
+    const std::lock_guard lock(thread->cancelLock);
+    return thread->cancelWait != nullptr;
+}
+
+static void InitializeSignaledCondition(SignaledCondition& context, bool recursive) {
+    PthreadMutexattr attr = nullptr;
+    Require(scePthreadMutexattrInit(&attr) == 0);
+    Require(scePthreadMutexattrSettype(&attr, recursive ? 2 : 1) == 0);
+    Require(scePthreadMutexInit(&context.mutex, &attr, nullptr) == 0);
+    Require(scePthreadMutexattrDestroy(&attr) == 0);
+    context.lockCount = recursive ? 3 : 1;
+    Require(scePthreadCondInit(context.slot, nullptr, nullptr) == 0);
+}
+
+static void CanceledConditionSignal(ConditionWaitMode mode, bool recursive) {
+    SignaledCondition context;
+    context.mode = mode;
+    InitializeSignaledCondition(context, recursive);
+    std::array<SignaledConditionWorker, 2> workers{{{&context, 0}, {&context, 1}}};
+    std::array<Pthread, 2> threads{};
+    for (int index = 0; index < 2; ++index) {
+        Require(scePthreadCreate(&threads[index], nullptr, SignaledConditionWaiter, &workers[index], nullptr) == 0);
+        while (context.ready.load() <= index) std::this_thread::yield();
+        Require(scePthreadMutexLock(&context.mutex) == 0);
+        const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(1);
+        while (!ConditionRegistered(threads[index]) && std::chrono::steady_clock::now() < deadline) {
+            Require(scePthreadMutexUnlock(&context.mutex) == 0);
+            std::this_thread::yield();
+            Require(scePthreadMutexLock(&context.mutex) == 0);
+        }
+        Require(ConditionRegistered(threads[index]));
+        Require(scePthreadMutexUnlock(&context.mutex) == 0);
+    }
+    Require(scePthreadMutexLock(&context.mutex) == 0);
+    context.released = true;
+    Require(scePthreadCondSignal(&context.condition) == 0);
+    int selected = -1;
+    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(1);
+    while (selected < 0 && std::chrono::steady_clock::now() < deadline) {
+        for (int index = 0; index < 2; ++index) if (!ConditionRegistered(threads[index])) selected = index;
+        std::this_thread::yield();
+    }
+    Require(selected >= 0);
+    Require(scePthreadCancel(threads[selected]) == 0);
+    Require(scePthreadMutexUnlock(&context.mutex) == 0);
+    Join(threads[selected], CANCELED);
+    Join(threads[1 - selected], nullptr);
+    Require(context.results[selected] == -1 && context.results[1 - selected] == 0);
+    Require(context.cleaned[0] && context.cleaned[1]);
+    Require(scePthreadCondDestroy(&context.condition) == 0);
+    Require(scePthreadMutexDestroy(&context.mutex) == 0);
+}
+
+static void DestroyedSignaledCondition(ConditionWaitMode mode, bool recursive, bool cancel, bool cancellationDisabled) {
+    SignaledCondition context;
+    context.mode = mode;
+    context.cancellationDisabled = cancellationDisabled;
+#ifdef _WIN32
+    SYSTEM_INFO system{};
+    GetSystemInfo(&system);
+    const auto bytes = static_cast<std::size_t>(system.dwPageSize);
+    context.slot = static_cast<PthreadCond*>(VirtualAlloc(nullptr, bytes, MEM_RESERVE | MEM_COMMIT, PAGE_READWRITE));
+    Require(context.slot != nullptr);
+#else
+    const auto bytes = static_cast<std::size_t>(sysconf(_SC_PAGESIZE));
+    context.slot = static_cast<PthreadCond*>(mmap(nullptr, bytes, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0));
+    Require(context.slot != MAP_FAILED);
+#endif
+    *context.slot = nullptr;
+    InitializeSignaledCondition(context, recursive);
+    const std::weak_ptr<TimedWait::Condition> retained = (*context.slot)->_cv;
+    SignaledConditionWorker worker{&context, 0};
+    Pthread thread = nullptr;
+    Require(scePthreadCreate(&thread, nullptr, SignaledConditionWaiter, &worker, nullptr) == 0);
+    while (context.ready.load() == 0) std::this_thread::yield();
+    Require(scePthreadMutexLock(&context.mutex) == 0);
+    context.released = true;
+    Require(scePthreadCondSignal(context.slot) == 0);
+    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(1);
+    while (ConditionRegistered(thread) && std::chrono::steady_clock::now() < deadline) std::this_thread::yield();
+    Require(!ConditionRegistered(thread));
+    Require(scePthreadCondDestroy(context.slot) == 0);
+    Require(!retained.expired());
+#ifdef _WIN32
+    Require(VirtualFree(context.slot, 0, MEM_RELEASE) != 0);
+#else
+    Require(munmap(context.slot, bytes) == 0);
+#endif
+    if (cancel) Require(scePthreadCancel(thread) == 0);
+    Require(scePthreadMutexUnlock(&context.mutex) == 0);
+    Join(thread, cancel && !cancellationDisabled ? CANCELED : nullptr);
+    Require(context.cleaned[0]);
+    Require(context.results[0] == (cancel && !cancellationDisabled ? -1 : 0));
+    Require(retained.expired());
+    Require(scePthreadMutexDestroy(&context.mutex) == 0);
+}
+
 static void* APS5_VABI HeldTarget(void*) {
     CheckDefaults();
     targetRelease.wait(false);
@@ -274,6 +452,14 @@ int main() {
     }
     Require(scePthreadKeyDelete(key) == 0);
     ConditionRecovery();
+    for (const auto mode : {ConditionWaitMode::SceTimed, ConditionWaitMode::Sce, ConditionWaitMode::PosixTimed, ConditionWaitMode::Posix}) {
+        for (bool recursive : {false, true}) {
+            CanceledConditionSignal(mode, recursive);
+            DestroyedSignaledCondition(mode, recursive, true, false);
+            DestroyedSignaledCondition(mode, recursive, false, false);
+            DestroyedSignaledCondition(mode, recursive, true, true);
+        }
+    }
     disabled = true;
     ConditionRecovery();
     JoinRecovery();