Skip to content

Commit

Permalink
semaphore: Use condvars with separate signaled flag to prevent races (s…
Browse files Browse the repository at this point in the history
…hadps4-emu#1615)

* Revert "semaphore: Use binary_semaphore instead of condvar"

This reverts commit 85dc57b.

* semaphore: Use separate signaled flag to prevent races

* mutex: Few misc fixes

* condvar: Few misc fixes

* signals: Add thread name to unhandled signal message.
  • Loading branch information
squidbus authored Nov 30, 2024
1 parent 3604404 commit 07f4a03
Show file tree
Hide file tree
Showing 5 changed files with 40 additions and 25 deletions.
7 changes: 5 additions & 2 deletions src/core/libraries/kernel/threads/condvar.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ int PthreadCond::Wait(PthreadMutexT* mutex, const OrbisKernelTimespec* abstime,
SleepqUnlock(this);

//_thr_cancel_enter2(curthread, 0);
int error = curthread->Sleep(abstime, usec) ? 0 : POSIX_ETIMEDOUT;
error = curthread->Sleep(abstime, usec) ? 0 : POSIX_ETIMEDOUT;
//_thr_cancel_leave(curthread, 0);

SleepqLock(this);
Expand All @@ -145,7 +145,10 @@ int PthreadCond::Wait(PthreadMutexT* mutex, const OrbisKernelTimespec* abstime,
}
SleepqUnlock(this);
curthread->mutex_obj = nullptr;
mp->CvLock(recurse);
int error2 = mp->CvLock(recurse);
if (error == 0) {
error = error2;
}
return error;
}

Expand Down
5 changes: 3 additions & 2 deletions src/core/libraries/kernel/threads/mutex.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,7 @@ int PthreadMutex::SelfTryLock() {
switch (Type()) {
case PthreadMutexType::ErrorCheck:
case PthreadMutexType::Normal:
case PthreadMutexType::AdaptiveNp:
return POSIX_EBUSY;
case PthreadMutexType::Recursive: {
/* Increment the lock count: */
Expand Down Expand Up @@ -224,7 +225,7 @@ int PthreadMutex::Lock(const OrbisKernelTimespec* abstime, u64 usec) {
[[unlikely]] {
ret = POSIX_EINVAL;
} else {
if (THR_RELTIME) {
if (abstime == THR_RELTIME) {
ret = m_lock.try_lock_for(std::chrono::microseconds(usec)) ? 0 : POSIX_ETIMEDOUT;
} else {
ret = m_lock.try_lock_until(abstime->TimePoint()) ? 0 : POSIX_ETIMEDOUT;
Expand Down Expand Up @@ -336,7 +337,7 @@ int PS4_SYSV_ABI posix_pthread_mutex_isowned_np(PthreadMutexT* mutex) {
return m->m_owner == g_curthread;
}

bool PthreadMutex::IsOwned(Pthread* curthread) const {
int PthreadMutex::IsOwned(Pthread* curthread) const {
if (this <= THR_MUTEX_DESTROYED) [[unlikely]] {
if (this == THR_MUTEX_DESTROYED) {
return POSIX_EINVAL;
Expand Down
2 changes: 1 addition & 1 deletion src/core/libraries/kernel/threads/pthread.h
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ struct PthreadMutex {
return Unlock();
}

bool IsOwned(Pthread* curthread) const;
int IsOwned(Pthread* curthread) const;
};
using PthreadMutexT = PthreadMutex*;

Expand Down
31 changes: 16 additions & 15 deletions src/core/libraries/kernel/threads/semaphore.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -49,9 +49,7 @@ class OrbisSem {
const auto it = AddWaiter(&waiter);

// Perform the wait.
lk.unlock();
const s32 result = waiter.Wait(timeout);
lk.lock();
const s32 result = waiter.Wait(lk, timeout);
if (result == SCE_KERNEL_ERROR_ETIMEDOUT) {
wait_list.erase(it);
}
Expand All @@ -74,7 +72,8 @@ class OrbisSem {
}
it = wait_list.erase(it);
token_count -= waiter->need_count;
waiter->sema.release();
waiter->was_signaled = true;
waiter->cv.notify_one();
}

return true;
Expand All @@ -87,7 +86,7 @@ class OrbisSem {
}
for (auto* waiter : wait_list) {
waiter->was_cancled = true;
waiter->sema.release();
waiter->cv.notify_one();
}
wait_list.clear();
token_count = set_count < 0 ? init_count : set_count;
Expand All @@ -98,20 +97,21 @@ class OrbisSem {
std::scoped_lock lk{mutex};
for (auto* waiter : wait_list) {
waiter->was_deleted = true;
waiter->sema.release();
waiter->cv.notify_one();
}
wait_list.clear();
}

public:
struct WaitingThread {
std::binary_semaphore sema;
std::condition_variable cv;
u32 priority;
s32 need_count;
bool was_signaled{};
bool was_deleted{};
bool was_cancled{};

explicit WaitingThread(s32 need_count, bool is_fifo) : sema{0}, need_count{need_count} {
explicit WaitingThread(s32 need_count, bool is_fifo) : need_count{need_count} {
// Retrieve calling thread priority for sorting into waiting threads list.
if (!is_fifo) {
priority = g_curthread->attr.prio;
Expand All @@ -131,24 +131,25 @@ class OrbisSem {
return SCE_OK;
}

int Wait(u32* timeout) {
int Wait(std::unique_lock<std::mutex>& lk, u32* timeout) {
if (!timeout) {
// Wait indefinitely until we are woken up.
sema.acquire();
cv.wait(lk);
return GetResult(false);
}
// Wait until timeout runs out, recording how much remaining time there was.
const auto start = std::chrono::high_resolution_clock::now();
const auto sema_timeout = !sema.try_acquire_for(std::chrono::microseconds(*timeout));
const auto signaled = cv.wait_for(lk, std::chrono::microseconds(*timeout),
[this] { return was_signaled; });
const auto end = std::chrono::high_resolution_clock::now();
const auto time =
std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
if (sema_timeout) {
*timeout = 0;
} else {
if (signaled) {
*timeout -= time;
} else {
*timeout = 0;
}
return GetResult(sema_timeout);
return GetResult(!signaled);
}
};

Expand Down
20 changes: 15 additions & 5 deletions src/core/signals.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,14 @@ static LONG WINAPI SignalHandler(EXCEPTION_POINTERS* pExp) noexcept {

#else

static std::string GetThreadName() {
char name[256];
if (pthread_getname_np(pthread_self(), name, sizeof(name)) != 0) {
return "<unknown name>";
}
return std::string{name};
}

static std::string DisassembleInstruction(void* code_address) {
char buffer[256] = "<unable to decode>";

Expand Down Expand Up @@ -71,16 +79,18 @@ static void SignalHandler(int sig, siginfo_t* info, void* raw_context) {
case SIGBUS: {
const bool is_write = Common::IsWriteError(raw_context);
if (!signals->DispatchAccessViolation(raw_context, info->si_addr)) {
UNREACHABLE_MSG("Unhandled access violation at code address {}: {} address {}",
fmt::ptr(code_address), is_write ? "Write to" : "Read from",
fmt::ptr(info->si_addr));
UNREACHABLE_MSG(
"Unhandled access violation in thread '{}' at code address {}: {} address {}",
GetThreadName(), fmt::ptr(code_address), is_write ? "Write to" : "Read from",
fmt::ptr(info->si_addr));
}
break;
}
case SIGILL:
if (!signals->DispatchIllegalInstruction(raw_context)) {
UNREACHABLE_MSG("Unhandled illegal instruction at code address {}: {}",
fmt::ptr(code_address), DisassembleInstruction(code_address));
UNREACHABLE_MSG("Unhandled illegal instruction in thread '{}' at code address {}: {}",
GetThreadName(), fmt::ptr(code_address),
DisassembleInstruction(code_address));
}
break;
case SIGUSR1: { // Sleep thread until signal is received
Expand Down

0 comments on commit 07f4a03

Please sign in to comment.