Merge pull request #5429 from Dripl1x/perf/windows-mutex-wake

perf(libkernel): wake guest mutex waiters on release on Windows
This commit is contained in:
Dean 2026-10-14 12:31:24 +13:00 committed by GitHub
commit d1670009fb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 395 additions and 54 deletions

View File

@ -1264,6 +1264,13 @@ if(BUILD_TESTING)
add_test(NAME guest_mutex_raise_exception COMMAND guest_mutex_raise_exception_tests)
set_tests_properties(guest_mutex_raise_exception PROPERTIES TIMEOUT 20 ENVIRONMENT_MODIFICATION
"PATH=path_list_prepend:$<TARGET_FILE_DIR:libkernel>")
add_test_executable(guest_mutex_contention_tests ${LIBS_INCLUDE_DIR}/tests/GuestMutexContention.cpp)
target_include_directories(guest_mutex_contention_tests PRIVATE ${LIBS_INCLUDE_DIR})
target_link_libraries(guest_mutex_contention_tests PRIVATE libkernel libc)
configure_windows_unwind(guest_mutex_contention_tests)
add_test(NAME guest_mutex_contention COMMAND guest_mutex_contention_tests)
set_tests_properties(guest_mutex_contention PROPERTIES TIMEOUT 20 RUN_SERIAL TRUE ENVIRONMENT_MODIFICATION
"PATH=path_list_prepend:$<TARGET_FILE_DIR:libkernel>")
endif()
add_test_executable(guest_pthread_rename_tests ${LIBS_INCLUDE_DIR}/tests/GuestPthreadRename.cpp)
target_include_directories(guest_pthread_rename_tests PRIVATE ${LIBS_INCLUDE_DIR})

View File

@ -60,6 +60,19 @@ int mutexOwnership(PthreadMutex* mutex, PthreadMutex* owned) {
return 0;
}
template <class TNative>
struct WakingLock {
PthreadMutex mutex;
TNative& native;
void lock() { native.lock(); }
void unlock() {
native.unlock();
#ifdef _WIN32
TimedWait::Unpark(mutex);
#endif
}
};
int waitUntil(PthreadCond* cond, PthreadMutex* mutex, std::optional<std::uint64_t> deadlineNanos, const void* caller) {
auto* c = resolveCond(cond);
PthreadMutex m = nullptr;
@ -72,7 +85,8 @@ int waitUntil(PthreadCond* cond, PthreadMutex* mutex, std::optional<std::uint64_
~Trace() { KernelTraceWait_nid_postfix("cond", caller, static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now() - start).count()), timedOut); }
} trace{caller, timedOut, waitStart};
if (m->_type == MutexType::Recursive) {
std::unique_lock<std::recursive_timed_mutex> lock(m->_rmtx, std::adopt_lock);
WakingLock<std::recursive_timed_mutex> waking{m, m->_rmtx};
std::unique_lock lock(waking, std::adopt_lock);
const auto previousCount = m->_count;
m->_count = 0;
m->_owner.store(std::thread::id{}, std::memory_order_release);
@ -87,7 +101,8 @@ int waitUntil(PthreadCond* cond, PthreadMutex* mutex, std::optional<std::uint64_
ThreadCancel::Check();
return timedOut ? sceTimedOut : 0;
}
std::unique_lock<std::timed_mutex> lock(m->_mtx, std::adopt_lock);
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);
m->_owner.store(std::this_thread::get_id(), std::memory_order_release);

View File

@ -2,6 +2,7 @@
#include "../include/Mutex.hpp"
#include "prx/libkernel/KernelErrors.hpp"
#include "prx/libkernel/Time/include/TimedWait.hpp"
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdint>
@ -87,14 +88,18 @@ int MutexOperations::Timedlock(PthreadMutex* mutex, const KernelTimespec* abstim
if (std::chrono::duration<long double>(duration) >= std::chrono::duration<long double>(std::chrono::system_clock::duration::max()))
throw std::overflow_error("Absolute mutex timeout exceeds the host clock range");
const auto deadline = std::chrono::system_clock::time_point(std::chrono::duration_cast<std::chrono::system_clock::duration>(duration));
return acquireMutex(resolveMutex(mutex, true), [&](auto& native) {
auto* current = resolveMutex(mutex, true);
return acquireMutex(current, [&](auto& native) {
#ifdef _WIN32
while (!native.try_lock()) {
if (std::chrono::system_clock::now() >= deadline) return false;
TimedWait::SleepNanos(1000000);
for (;;) {
const auto now = std::chrono::system_clock::now();
if (now >= deadline) return native.try_lock();
const auto remaining = std::chrono::duration_cast<std::chrono::nanoseconds>(deadline - now).count();
const std::uint64_t slice = TimedWait::NowNanos() + static_cast<std::uint64_t>(std::min<std::int64_t>(remaining, 100000000));
if (TimedWait::ParkAcquire(current, [&] { return native.try_lock(); }, &slice)) return true;
}
return true;
#else
static_cast<void>(current);
return native.try_lock_until(deadline);
#endif
}, sceTimedOut, false);
@ -169,13 +174,15 @@ int APS5_VABI scePthreadMutexDestroy(PthreadMutex* mutex) {
}
int APS5_VABI scePthreadMutexLock(PthreadMutex* mutex) {
return acquireMutex(resolveMutex(mutex, true), [](auto& native) {
auto* current = resolveMutex(mutex, true);
return acquireMutex(current, [current](auto& native) {
#ifdef _WIN32
while (!native.try_lock()) TimedWait::SleepNanos(1000000);
return TimedWait::ParkAcquire(current, [&] { return native.try_lock(); }, nullptr);
#else
static_cast<void>(current);
native.lock();
#endif
return true;
#endif
}, 0, false);
}
@ -183,21 +190,29 @@ int APS5_VABI scePthreadMutexUnlock(PthreadMutex* mutex) {
auto* current = resolveMutex(mutex, false);
if (current->_owner.load(std::memory_order_acquire) != std::this_thread::get_id())
return sceNotPermitted;
bool released = true;
if (current->_type == MutexType::Recursive) {
if (--current->_count == 0)
released = --current->_count == 0;
if (released)
current->_owner.store(std::thread::id{}, std::memory_order_release);
current->_rmtx.unlock();
} else {
current->_owner.store(std::thread::id{}, std::memory_order_release);
current->_mtx.unlock();
}
#ifdef _WIN32
if (released) TimedWait::Unpark(current);
#else
static_cast<void>(released);
#endif
return 0;
}
int APS5_VABI scePthreadMutexTimedlock(PthreadMutex* mutex, KernelUseconds usec) {
const auto deadline = TimedWait::DeadlineNanos(usec);
return acquireMutex(resolveMutex(mutex, true), [=](auto& native) {
return TimedWait::AcquireUntil(deadline, [&] { return native.try_lock(); }, [&](std::uint64_t micros) { return native.try_lock_for(std::chrono::microseconds(micros)); });
auto* current = resolveMutex(mutex, true);
return acquireMutex(current, [=](auto& native) {
return TimedWait::AcquireUntil(current, deadline, [&] { return native.try_lock(); }, [&](std::uint64_t micros) { return native.try_lock_for(std::chrono::microseconds(micros)); });
}, sceTimedOut, false);
}

View File

@ -3,6 +3,7 @@
#include <cstdlib>
#include <limits>
#include <stdexcept>
#ifdef _WIN32
#include <windows.h>
@ -108,10 +109,11 @@ void SleepNanosCoarse(std::uint64_t nanos) {
struct Waiter {
HANDLE event = CreateEventW(nullptr, FALSE, FALSE, nullptr);
HANDLE timer = CreateHighResolutionTimer();
HANDLE sleepTimers[3]{};
HANDLE sleepTimers[2]{};
Waiter* previous = nullptr;
Waiter* next = nullptr;
Waiter* nested = nullptr;
const void* key = nullptr;
bool queued = false;
bool busy = false;
@ -239,6 +241,111 @@ void Condition::NotifyAll() {
}
}
namespace {
struct alignas(64) ParkBucket {
std::mutex lock;
Waiter* head = nullptr;
Waiter* tail = nullptr;
std::atomic<int> parked{0};
};
ParkBucket parkBuckets[256];
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;
if (waiter->next) waiter->next->previous = waiter->previous;
else bucket.tail = waiter->previous;
waiter->previous = nullptr;
waiter->next = nullptr;
waiter->queued = false;
}
bool LeavePark(ParkBucket& bucket, Waiter* waiter) {
bool signaled;
{
DeliveryDeferral deferral;
std::lock_guard lock(bucket.lock);
signaled = !waiter->queued;
if (!signaled) UnlinkParked(bucket, waiter);
}
if (signaled) WaitForSingleObject(waiter->event, 0);
return signaled;
}
void ReleaseParked(ParkBucket& bucket, Waiter* waiter) {
bucket.parked.fetch_sub(1, std::memory_order_seq_cst);
waiter->busy = false;
DrainApcs();
}
} // namespace
void SpinPause() {
YieldProcessor();
}
Waiter* BeginPark(const void* key) {
auto& bucket = ParkBucketFor(key);
Waiter* waiter;
{
DeliveryDeferral deferral;
waiter = AcquireWaiter();
waiter->key = key;
bucket.parked.fetch_add(1, std::memory_order_seq_cst);
std::lock_guard lock(bucket.lock);
waiter->queued = true;
waiter->next = nullptr;
waiter->previous = bucket.tail;
if (bucket.tail) bucket.tail->next = waiter;
else bucket.head = waiter;
bucket.tail = waiter;
}
DrainApcs();
return waiter;
}
bool FinishPark(const void* key, Waiter* waiter, const std::uint64_t* deadlineNanos) {
auto& bucket = ParkBucketFor(key);
bool signaled = deadlineNanos ? WaitEventUntil(waiter, *deadlineNanos) : AlertableWait(1, &waiter->event, INFINITE) == WAIT_OBJECT_0;
if (!signaled) signaled = LeavePark(bucket, waiter);
ReleaseParked(bucket, waiter);
if (!signaled && !deadlineNanos) throw std::runtime_error("TimedWait: an untimed parked wait ended without a wake");
return signaled;
}
void CancelPark(const void* key, Waiter* waiter) {
auto& bucket = ParkBucketFor(key);
LeavePark(bucket, waiter);
ReleaseParked(bucket, waiter);
}
void Unpark(const void* key) {
auto& bucket = ParkBucketFor(key);
if (bucket.parked.load(std::memory_order_seq_cst) == 0) return;
{
DeliveryDeferral deferral;
std::lock_guard lock(bucket.lock);
for (Waiter* waiter = bucket.head; waiter; waiter = waiter->next) {
if (waiter->key != key) continue;
UnlinkParked(bucket, waiter);
SetEvent(waiter->event);
break;
}
}
DrainApcs();
}
#else
void Condition::NotifyOne() {
@ -312,17 +419,6 @@ void SleepUntil(std::uint64_t deadlineNanos) {
#endif
}
void PollSleepUntil(std::uint64_t deadlineNanos) {
#ifdef _WIN32
const std::uint64_t now = NowNanos();
if (deadlineNanos <= now) return;
const HANDLE timer = ThreadTimer(2);
if (!TimerWait(timer, deadlineNanos - now)) AlertableSleep(WholeMilliseconds(deadlineNanos - now));
#else
SleepUntil(deadlineNanos);
#endif
}
void SleepNanos(std::uint64_t nanos) {
if (nanos == 0) return;
#ifdef _WIN32

View File

@ -16,38 +16,46 @@ std::uint64_t DeadlineNanos(std::uint64_t microseconds);
std::uint64_t RemainingMicros(std::uint64_t deadlineNanos);
void SleepNanos(std::uint64_t nanos);
void SleepUntil(std::uint64_t deadlineNanos);
void PollSleepUntil(std::uint64_t deadlineNanos);
constexpr std::uint64_t LOCK_BULK_NANOS = 40000000ULL;
constexpr std::uint64_t LOCK_POLL_NANOS = 500000ULL;
constexpr std::uint64_t LOCK_TAIL_NANOS = 1000000ULL;
template <class TTryLock, class TTryLockFor>
bool AcquireUntil(std::uint64_t deadlineNanos, TTryLock tryLock, TTryLockFor tryLockFor) {
if (Coarse()) return tryLockFor(RemainingMicros(deadlineNanos));
for (;;) {
if (tryLock()) return true;
const std::uint64_t now = NowNanos();
if (now >= deadlineNanos) return false;
const std::uint64_t remaining = deadlineNanos - now;
#ifdef _WIN32
if (remaining > LOCK_TAIL_NANOS) {
PollSleepUntil(now + (remaining < LOCK_POLL_NANOS ? remaining : LOCK_POLL_NANOS));
continue;
}
#else
if (remaining > LOCK_BULK_NANOS) {
if (tryLockFor((remaining - LOCK_BULK_NANOS) / 1000ULL)) return true;
continue;
}
#endif
if (remaining <= LOCK_TAIL_NANOS) SleepUntil(deadlineNanos);
else PollSleepUntil(now + LOCK_POLL_NANOS);
}
}
struct Waiter;
#ifdef _WIN32
constexpr unsigned PARK_SPINS = 1000;
void SpinPause();
Waiter* BeginPark(const void* key);
bool FinishPark(const void* key, Waiter* waiter, const std::uint64_t* deadlineNanos);
void CancelPark(const void* key, Waiter* waiter);
void Unpark(const void* key);
template <class TTryLock>
bool ParkAcquire(const void* key, TTryLock tryLock, const std::uint64_t* deadlineNanos) {
for (unsigned spin = 0; spin < PARK_SPINS; ++spin) {
if (tryLock()) return true;
SpinPause();
}
for (;;) {
Waiter* waiter = BeginPark(key);
if (tryLock()) {
CancelPark(key, waiter);
return true;
}
if (!FinishPark(key, waiter, deadlineNanos)) return tryLock();
if (tryLock()) return true;
}
}
#endif
template <class TTryLock, class TTryLockFor>
bool AcquireUntil(const void* key, std::uint64_t deadlineNanos, TTryLock tryLock, TTryLockFor tryLockFor) {
#ifdef _WIN32
if (!Coarse()) return ParkAcquire(key, tryLock, &deadlineNanos);
#endif
static_cast<void>(key);
static_cast<void>(tryLock);
return tryLockFor(RemainingMicros(deadlineNanos));
}
class Condition {
public:
Condition() = default;

View File

@ -0,0 +1,200 @@
#include "SceTypes.hpp"
#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <thread>
#include <vector>
#include <windows.h>
extern "C" {
int APS5_VABI scePthreadCreate(Pthread*, const PthreadAttr*, PthreadEntry, void*, const char*);
int APS5_VABI scePthreadJoin(Pthread, void**);
int APS5_VABI scePthreadMutexattrInit(PthreadMutexattr*);
int APS5_VABI scePthreadMutexattrSettype(PthreadMutexattr*, int);
int APS5_VABI scePthreadMutexattrDestroy(PthreadMutexattr*);
int APS5_VABI scePthreadMutexInit(PthreadMutex*, const PthreadMutexattr*, const char*);
int APS5_VABI scePthreadMutexDestroy(PthreadMutex*);
int APS5_VABI scePthreadMutexLock(PthreadMutex*);
int APS5_VABI scePthreadMutexTimedlock(PthreadMutex*, KernelUseconds);
int APS5_VABI scePthreadMutexUnlock(PthreadMutex*);
}
namespace {
using Clock = std::chrono::steady_clock;
constexpr int TimedOut = static_cast<int>(0x8002003cu);
void Require(bool value, const char* message) {
if (!value) {
std::fprintf(stderr, "%s\n", message);
std::abort();
}
}
std::uint64_t ThreadCycles() {
ULONG64 cycles = 0;
Require(QueryThreadCycleTime(GetCurrentThread(), &cycles) != 0, "QueryThreadCycleTime failed");
return cycles;
}
std::uint64_t BusyCycles(int milliseconds) {
const auto before = ThreadCycles();
const auto until = Clock::now() + std::chrono::milliseconds(milliseconds);
while (Clock::now() < until) YieldProcessor();
return ThreadCycles() - before;
}
struct Handoff {
PthreadMutex mutex = nullptr;
bool timed = false;
bool destroy = false;
std::atomic<bool> waiting{false};
std::atomic<std::int64_t> released{0};
std::int64_t latencyNanos = 0;
std::uint64_t waitCycles = 0;
};
struct HandoffResult {
std::int64_t latencyNanos;
std::uint64_t waitCycles;
};
void* APS5_VABI HandoffWorker(void* opaque) {
auto& context = *static_cast<Handoff*>(opaque);
const auto cyclesBefore = ThreadCycles();
context.waiting.store(true);
Require((context.timed ? scePthreadMutexTimedlock(&context.mutex, 5000000) : scePthreadMutexLock(&context.mutex)) == 0, "the waiter did not acquire the mutex");
const auto acquired = Clock::now().time_since_epoch().count();
context.waitCycles = ThreadCycles() - cyclesBefore;
context.latencyNanos = acquired - context.released.load();
Require(scePthreadMutexUnlock(&context.mutex) == 0, "the waiter could not unlock");
if (context.destroy) Require(scePthreadMutexDestroy(&context.mutex) == 0, "the waiter could not destroy the mutex");
return nullptr;
}
HandoffResult RunHandoff(PthreadMutex mutex, bool timed, bool destroy, int holdMilliseconds, int extraMicroseconds = 0) {
Handoff context;
context.mutex = mutex;
context.timed = timed;
context.destroy = destroy;
PthreadMutex holder = mutex;
Require(scePthreadMutexLock(&holder) == 0, "the holder could not lock");
Pthread worker = nullptr;
Require(scePthreadCreate(&worker, nullptr, HandoffWorker, &context, "mutex waiter") == 0, "cannot create the waiter");
while (!context.waiting.load()) std::this_thread::yield();
std::this_thread::sleep_for(std::chrono::milliseconds(holdMilliseconds));
const auto extra = Clock::now() + std::chrono::microseconds(extraMicroseconds);
while (Clock::now() < extra) YieldProcessor();
context.released.store(Clock::now().time_since_epoch().count());
Require(scePthreadMutexUnlock(&holder) == 0, "the holder could not unlock");
Require(scePthreadJoin(worker, nullptr) == 0, "cannot join the waiter");
return {context.latencyNanos, context.waitCycles};
}
PthreadMutex CreateMutex(bool recursive) {
PthreadMutexattr attributes = nullptr;
Require(scePthreadMutexattrInit(&attributes) == 0 && scePthreadMutexattrSettype(&attributes, recursive ? 2 : 1) == 0, "cannot prepare mutex attributes");
PthreadMutex mutex = nullptr;
Require(scePthreadMutexInit(&mutex, &attributes, nullptr) == 0, "cannot create the mutex");
Require(scePthreadMutexattrDestroy(&attributes) == 0, "cannot destroy mutex attributes");
return mutex;
}
struct Contention {
PthreadMutex mutex = nullptr;
std::uint64_t rounds = 0;
std::atomic<bool> start{false};
std::atomic<bool> stop{false};
std::uint64_t shared = 0;
std::array<std::uint64_t, 4> acquired{};
std::atomic<int> next{0};
};
void* APS5_VABI ContentionWorker(void* opaque) {
auto& context = *static_cast<Contention*>(opaque);
const int index = context.next.fetch_add(1);
PthreadMutex mutex = context.mutex;
while (!context.start.load()) std::this_thread::yield();
std::uint64_t count = 0;
while (context.rounds ? count < context.rounds : !context.stop.load(std::memory_order_relaxed)) {
Require(scePthreadMutexLock(&mutex) == 0, "contended lock failed");
const std::uint64_t seen = context.shared;
for (int spin = 0; spin < 64; ++spin) YieldProcessor();
context.shared = seen + 1;
Require(scePthreadMutexUnlock(&mutex) == 0, "contended unlock failed");
++count;
for (int spin = 0; spin < 64; ++spin) YieldProcessor();
}
context.acquired[index] = count;
return nullptr;
}
void* APS5_VABI TimeoutWorker(void* opaque) {
PthreadMutex mutex = *static_cast<PthreadMutex*>(opaque);
const auto started = Clock::now();
Require(scePthreadMutexTimedlock(&mutex, 20000) == TimedOut, "a timed lock of a held mutex did not time out");
Require(Clock::now() - started >= std::chrono::milliseconds(20), "a timed lock of a held mutex timed out before its deadline");
return nullptr;
}
}
int main(int argc, char** argv) {
const bool report = argc > 1 && std::strcmp(argv[1], "--report") == 0;
const std::uint64_t busyCycles = report ? BusyCycles(20) * 10 : 0;
for (const bool recursive : {false, true}) {
PthreadMutex mutex = CreateMutex(recursive);
for (const bool timed : {false, true}) {
if (!report) {
for (int round = 0; round < 4; ++round) RunHandoff(mutex, timed, false, 3);
continue;
}
std::vector<std::int64_t> latencies;
for (int round = 0; round < 31; ++round) latencies.push_back(RunHandoff(mutex, timed, false, 3, round * 617 % 1000).latencyNanos);
std::sort(latencies.begin(), latencies.end());
const auto blocked = RunHandoff(mutex, timed, false, 200);
const double share = 100.0 * static_cast<double>(blocked.waitCycles) / static_cast<double>(busyCycles);
std::printf("%s %s: handoff median %.1f us, p90 %.1f us, max %.1f us; a waiter blocked for 200 ms used %.1f %% of one processor\n",
recursive ? "recursive" : "errorcheck", timed ? "timed lock" : "lock",
latencies[15] / 1000.0, latencies[27] / 1000.0, latencies[30] / 1000.0, share);
}
PthreadMutex held = mutex;
Require(scePthreadMutexLock(&held) == 0, "cannot hold the mutex for the timeout");
Pthread timeout = nullptr;
Require(scePthreadCreate(&timeout, nullptr, TimeoutWorker, &mutex, "mutex timeout") == 0, "cannot create the timeout waiter");
Require(scePthreadJoin(timeout, nullptr) == 0, "cannot join the timeout waiter");
Require(scePthreadMutexUnlock(&held) == 0, "cannot release the mutex after the timeout");
Contention context;
context.mutex = mutex;
context.rounds = report ? 0 : 20000;
std::array<Pthread, 4> workers{};
for (auto& worker : workers) Require(scePthreadCreate(&worker, nullptr, ContentionWorker, &context, "mutex contender") == 0, "cannot create a contender");
const auto started = Clock::now();
context.start.store(true);
if (report) {
std::this_thread::sleep_for(std::chrono::milliseconds(300));
context.stop.store(true);
}
for (auto& worker : workers) Require(scePthreadJoin(worker, nullptr) == 0, "cannot join a contender");
const double seconds = std::chrono::duration<double>(Clock::now() - started).count();
std::uint64_t total = 0;
for (const auto count : context.acquired) total += count;
Require(total == context.shared, "the mutex did not exclude its contenders");
const auto fewest = *std::min_element(context.acquired.begin(), context.acquired.end());
if (report) {
std::printf("%s: four contenders made %.0f acquisitions per second, the slowest %.0f per second\n",
recursive ? "recursive" : "errorcheck", total / seconds, fewest / seconds);
} else {
Require(fewest == context.rounds, "a contender did not finish its acquisitions");
}
Require(scePthreadMutexDestroy(&mutex) == 0, "cannot destroy the mutex");
for (int round = 0; round < 200; ++round) RunHandoff(CreateMutex(recursive), round % 2 != 0, true, 0);
}
std::puts("guest mutex contention: PASS");
return 0;
}