mirror/rpcs3
1
0
Fork 0
mirror of https://github.com/RPCS3/rpcs3.git synced 2025-01-27 12:35:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

sleep_queue_t rewritten, used in sys_cond/sys_mutex

Browse Source 
Some synchronization fixes
This commit is contained in:
Nekotekina 2015-07-08 01:33:24 +03:00
parent eafddd9e33
commit c0f13f7084
29 changed files with 350 additions and 427 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Utilities/Thread.cpp
View File

@ -1290,10 +1290,12 @@ void thread_t::start(std::function<std::string()> name, std::function<void()> fu
}
catch (const char* e) // obsolete
{
LOG_ERROR(GENERAL, "Deprecated exception type (const char*)");
error(e);
}
catch (const std::string& e) // obsolete
{
LOG_ERROR(GENERAL, "Deprecated exception type (std::string)");
error(e.c_str());
}
catch (const fmt::exception& e)

7
Utilities/Thread.h
View File

@ -137,8 +137,11 @@ struct waiter_map_t
check_emu_status(addr);
// lock the mutex and initialize waiter (only once)
if (!lock) lock.lock();
if (!lock)
{
lock.lock();
continue;
}
// wait on an appropriate cond var for 1 ms or until a signal arrived
cvs[hash].wait_for(lock, std::chrono::milliseconds(1));

1
rpcs3/Emu/ARMv7/ARMv7Thread.cpp
View File

@ -4,7 +4,6 @@
#include "Emu/Memory/Memory.h"
#include "Emu/System.h"
#include "Emu/IdManager.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/ARMv7/PSVFuncList.h"
#include "ARMv7Thread.h"

1
rpcs3/Emu/ARMv7/Modules/sceLibKernel.cpp
View File

@ -4,7 +4,6 @@
#include "Emu/ARMv7/PSVFuncList.h"
#include "Emu/ARMv7/PSVObjectList.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/SysCalls/Callback.h"
#include "Emu/ARMv7/ARMv7Thread.h"

65
rpcs3/Emu/CPU/CPUThread.cpp
View File

@ -6,7 +6,6 @@
#include "Emu/IdManager.h"
#include "Emu/DbgCommand.h"
#include "CPUThreadManager.h"
#include "CPUDecoder.h"
#include "CPUThread.h"
@ -57,8 +56,6 @@ CPUThread::CPUThread(CPUThreadType type, const std::string& name, std::function<
cv.wait(lock);
}
cv.notify_all();
});
}
@ -167,11 +164,23 @@ void CPUThread::Exit()
void CPUThread::Step()
{
m_state.atomic_op([](u64& state)
if (m_state.atomic_op([](u64& state) -> bool
{
const bool was_paused = (state & CPU_STATE_PAUSED) != 0;
state |= CPU_STATE_STEP;
state &= ~CPU_STATE_PAUSED;
});
return was_paused;
}))
{
if (is_current()) return;
// lock for reliable notification (only if PAUSE was removed)
std::lock_guard<std::mutex> lock(mutex);
cv.notify_one();
}
}
void CPUThread::Sleep()
@ -184,7 +193,7 @@ void CPUThread::Awake()
{
// must be called after the balanced Sleep() call
m_state.atomic_op([](u64& state)
if (m_state.atomic_op([](u64& state) -> bool
{
if (state < CPU_STATE_MAX)
{
@ -194,13 +203,43 @@ void CPUThread::Awake()
if ((state -= CPU_STATE_MAX) < CPU_STATE_MAX)
{
state &= ~CPU_STATE_SLEEP;
// notify the condition variable as well
return true;
}
});
// lock for reliable notification because the condition being checked is probably externally set
std::lock_guard<std::mutex> lock(mutex);
return false;
}))
{
if (is_current()) return;
cv.notify_one();
// lock for reliable notification; the condition being checked is probably externally set
std::lock_guard<std::mutex> lock(mutex);
cv.notify_one();
}
}
bool CPUThread::Signal()
{
// try to set SIGNAL
if (m_state._or(CPU_STATE_SIGNAL) & CPU_STATE_SIGNAL)
{
return false;
}
else
{
// not truly responsible for signal delivery, requires additional measures like LV2_LOCK
cv.notify_one();
return true;
}
}
bool CPUThread::Signaled()
{
// remove SIGNAL and return its old value
return (m_state._and_not(CPU_STATE_SIGNAL) & CPU_STATE_SIGNAL) != 0;
}
bool CPUThread::CheckStatus()
@ -213,7 +252,11 @@ bool CPUThread::CheckStatus()
if (!IsPaused()) break;
if (!lock) lock.lock();
if (!lock)
{
lock.lock();
continue;
}
cv.wait(lock);
}

11
rpcs3/Emu/CPU/CPUThread.h
View File

@ -19,8 +19,9 @@ enum : u64
CPU_STATE_STEP = (1ull << 3), // forces the thread to pause after executing just one instruction or something appropriate, set by the debugger
CPU_STATE_DEAD = (1ull << 4), // indicates irreversible exit of the thread
CPU_STATE_RETURN = (1ull << 5), // used for callback return
CPU_STATE_SIGNAL = (1ull << 6),
CPU_STATE_MAX = (1ull << 6), // added to (subtracted from) m_state by Sleep()/Awake() calls to trigger status check
CPU_STATE_MAX = (1ull << 7), // added to (subtracted from) m_state by Sleep()/Awake() calls to trigger status check
};
// "HLE return" exception event
@ -34,7 +35,7 @@ class CPUThreadExit{};
class CPUDecoder;
class CPUThread : protected thread_t
class CPUThread : protected thread_t, public std::enable_shared_from_this<CPUThread>
{
protected:
atomic<u64> m_state; // thread state flags
@ -100,6 +101,12 @@ public:
// untrigger thread status check
void Awake();
// set SIGNAL and notify (returns true if set)
bool Signal();
// test SIGNAL and reset
bool Signaled();
// process m_state flags, returns true if the checker must return
bool CheckStatus();

1
rpcs3/Emu/Cell/PPUInterpreter.cpp
View File

@ -9,7 +9,6 @@
#include "PPUInstrTable.h"
#include "PPUInterpreter.h"
#include "PPUInterpreter2.h"
#include "Emu/CPU/CPUThreadManager.h"
class ppu_scale_table_t
{

1
rpcs3/Emu/Cell/PPUThread.cpp
View File

@ -12,7 +12,6 @@
#include "Emu/Cell/PPUInterpreter2.h"
#include "Emu/Cell/PPULLVMRecompiler.h"
//#include "Emu/Cell/PPURecompiler.h"
#include "Emu/CPU/CPUThreadManager.h"
#ifdef _WIN32
#include <Windows.h>

31
rpcs3/Emu/Cell/SPUThread.cpp
View File

@ -5,7 +5,6 @@
#include "Emu/System.h"
#include "Emu/IdManager.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/PPUThread.h"
#include "Emu/SysCalls/ErrorCodes.h"
#include "Emu/SysCalls/lv2/sys_spu.h"
@ -530,7 +529,11 @@ u32 SPUThread::get_ch_value(u32 ch)
if (IsStopped()) throw CPUThreadStop{};
if (!lock) lock.lock();
if (!lock)
{
lock.lock();
continue;
}
cv.wait_for(lock, std::chrono::milliseconds(1));
}
@ -555,7 +558,11 @@ u32 SPUThread::get_ch_value(u32 ch)
if (IsStopped()) throw CPUThreadStop{};
if (!lock) lock.lock();
if (!lock)
{
lock.lock();
continue;
}
cv.wait_for(lock, std::chrono::milliseconds(1));
}
@ -620,7 +627,11 @@ u32 SPUThread::get_ch_value(u32 ch)
if (IsStopped()) throw CPUThreadStop{};
if (!lock) lock.lock();
if (!lock)
{
lock.lock();
continue;
}
cv.wait_for(lock, std::chrono::milliseconds(1));
}
@ -661,7 +672,11 @@ void SPUThread::set_ch_value(u32 ch, u32 value)
if (IsStopped()) throw CPUThreadStop{};
if (!lock) lock.lock();
if (!lock)
{
lock.lock();
continue;
}
cv.wait_for(lock, std::chrono::milliseconds(1));
}
@ -857,7 +872,11 @@ void SPUThread::set_ch_value(u32 ch, u32 value)
if (IsStopped()) throw CPUThreadStop{};
if (!lock) lock.lock();
if (!lock)
{
lock.lock();
continue;
}
cv.wait_for(lock, std::chrono::milliseconds(1));
}

6
rpcs3/Emu/Memory/vm.cpp
View File

@ -102,7 +102,11 @@ namespace vm
throw EXCEPTION("Deadlock");
}
if (!lock) lock.lock();
if (!lock)
{
lock.lock();
continue;
}
m_cv.wait_for(lock, std::chrono::milliseconds(1));
}

1
rpcs3/Emu/SysCalls/Callback.cpp
View File

@ -6,7 +6,6 @@
#include "Emu/Cell/PPUThread.h"
#include "Emu/ARMv7/ARMv7Thread.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Callback.h"
void CallbackManager::Register(check_cb_t func)

1
rpcs3/Emu/SysCalls/Modules/cellAdec.cpp
View File

@ -13,7 +13,6 @@ extern "C"
#include "libswresample/swresample.h"
}
#include "Emu/CPU/CPUThreadManager.h"
#include "cellPamf.h"
#include "cellAdec.h"

1
rpcs3/Emu/SysCalls/Modules/cellDmux.cpp
View File

@ -4,7 +4,6 @@
#include "Emu/IdManager.h"
#include "Emu/SysCalls/Modules.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "cellPamf.h"
#include "cellDmux.h"

1
rpcs3/Emu/SysCalls/Modules/cellSpurs.cpp
View File

@ -5,7 +5,6 @@
#include "Emu/IdManager.h"
#include "Emu/Event.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/SPUThread.h"
#include "Emu/SysCalls/lv2/sleep_queue.h"
#include "Emu/SysCalls/lv2/sys_lwmutex.h"

1
rpcs3/Emu/SysCalls/Modules/cellVdec.cpp
View File

@ -14,7 +14,6 @@ extern "C"
#include "libswscale/swscale.h"
}
#include "Emu/CPU/CPUThreadManager.h"
#include "cellPamf.h"
#include "cellVdec.h"

1
rpcs3/Emu/SysCalls/Modules/libmixer.cpp
View File

@ -6,7 +6,6 @@
#include "Emu/SysCalls/Modules.h"
#include "Emu/Cell/PPUInstrTable.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "cellAudio.h"
#include "libmixer.h"

263
rpcs3/Emu/SysCalls/lv2/sleep_queue.cpp
View File

@ -2,257 +2,46 @@
#include "Utilities/Log.h"
#include "Emu/Memory/Memory.h"
#include "Emu/System.h"
#include "Emu/IdManager.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/PPUThread.h"
#include "Emu/CPU/CPUThread.h"
#include "sleep_queue.h"
sleep_queue_t::~sleep_queue_t()
sleep_queue_entry_t::sleep_queue_entry_t(CPUThread& cpu, sleep_queue_t& queue)
: m_queue(queue)
, m_thread(cpu)
{
for (auto& tid : m_waiting)
{
LOG_NOTICE(HLE, "~sleep_queue_t['%s']: m_waiting[%lld]=%d", m_name.c_str(), &tid - m_waiting.data(), tid);
}
for (auto& tid : m_signaled)
{
LOG_NOTICE(HLE, "~sleep_queue_t['%s']: m_signaled[%lld]=%d", m_name.c_str(), &tid - m_signaled.data(), tid);
}
m_queue.emplace_back(std::move(cpu.shared_from_this()));
m_thread.Sleep();
}
void sleep_queue_t::push(u32 tid, u32 protocol)
sleep_queue_entry_t::~sleep_queue_entry_t() noexcept(false)
{
assert(tid);
m_thread.Awake();
switch (protocol & SYS_SYNC_ATTR_PROTOCOL_MASK)
if (m_queue.front().get() == &m_thread)
{
case SYS_SYNC_FIFO:
case SYS_SYNC_PRIORITY:
{
std::lock_guard<std::mutex> lock(m_mutex);
for (auto& v : m_waiting)
{
if (v == tid)
{
LOG_ERROR(HLE, "sleep_queue_t['%s']::push() failed: thread already waiting (%d)", m_name.c_str(), tid);
Emu.Pause();
return;
}
}
for (auto& v : m_signaled)
{
if (v == tid)
{
LOG_ERROR(HLE, "sleep_queue_t['%s']::push() failed: thread already signaled (%d)", m_name.c_str(), tid);
Emu.Pause();
return;
}
}
m_waiting.push_back(tid);
m_queue.pop_front();
return;
}
case SYS_SYNC_RETRY:
if (m_queue.back().get() == &m_thread)
{
m_queue.pop_back();
return;
}
for (auto it = m_queue.begin(); it != m_queue.end(); it++)
{
if (it->get() == &m_thread)
{
m_queue.erase(it);
return;
}
}
LOG_ERROR(HLE, "sleep_queue_t['%s']::push() failed: unsupported protocol (0x%x)", m_name.c_str(), protocol);
Emu.Pause();
}
bool sleep_queue_t::pop(u32 tid, u32 protocol)
{
assert(tid);
switch (protocol & SYS_SYNC_ATTR_PROTOCOL_MASK)
{
case SYS_SYNC_FIFO:
case SYS_SYNC_PRIORITY:
{
std::lock_guard<std::mutex> lock(m_mutex);
if (m_signaled.size() && m_signaled[0] == tid)
{
m_signaled.erase(m_signaled.begin());
return true;
}
for (auto& v : m_signaled)
{
if (v == tid)
{
return false;
}
}
for (auto& v : m_waiting)
{
if (v == tid)
{
return false;
}
}
LOG_ERROR(HLE, "sleep_queue_t['%s']::pop() failed: thread not found (%d)", m_name.c_str(), tid);
Emu.Pause();
return true; // ???
}
//case SYS_SYNC_RETRY: // ???
//{
// return true; // ???
//}
}
LOG_ERROR(HLE, "sleep_queue_t['%s']::pop() failed: unsupported protocol (0x%x)", m_name.c_str(), protocol);
Emu.Pause();
return false; // ???
}
u32 sleep_queue_t::signal(u32 protocol)
{
u32 res = ~0;
switch (protocol & SYS_SYNC_ATTR_PROTOCOL_MASK)
{
case SYS_SYNC_FIFO:
{
std::lock_guard<std::mutex> lock(m_mutex);
if (m_waiting.size())
{
res = m_waiting[0];
if (!Emu.GetIdManager().check_id<CPUThread>(res))
{
LOG_ERROR(HLE, "sleep_queue_t['%s']::signal(SYS_SYNC_FIFO) failed: invalid thread (%d)", m_name.c_str(), res);
Emu.Pause();
}
m_waiting.erase(m_waiting.begin());
m_signaled.push_back(res);
}
else
{
res = 0;
}
return res;
}
case SYS_SYNC_PRIORITY:
{
std::lock_guard<std::mutex> lock(m_mutex);
s32 highest_prio = INT32_MAX;
u64 sel = ~0ull;
for (auto& v : m_waiting)
{
if (const auto t = Emu.GetIdManager().get<PPUThread>(v))
{
const s32 prio = t->prio;
if (prio < highest_prio)
{
highest_prio = prio;
sel = &v - m_waiting.data();
}
}
else
{
LOG_ERROR(HLE, "sleep_queue_t['%s']::signal(SYS_SYNC_PRIORITY) failed: invalid thread (%d)", m_name.c_str(), v);
Emu.Pause();
}
}
if (~sel)
{
res = m_waiting[sel];
m_waiting.erase(m_waiting.begin() + sel);
m_signaled.push_back(res);
return res;
}
// fallthrough
}
case SYS_SYNC_RETRY:
{
return 0;
}
}
LOG_ERROR(HLE, "sleep_queue_t['%s']::signal(): unsupported protocol (0x%x)", m_name.c_str(), protocol);
Emu.Pause();
return 0;
}
bool sleep_queue_t::invalidate(u32 tid, u32 protocol)
{
assert(tid);
switch (protocol & SYS_SYNC_ATTR_PROTOCOL_MASK)
{
case SYS_SYNC_FIFO:
case SYS_SYNC_PRIORITY:
{
std::lock_guard<std::mutex> lock(m_mutex);
for (auto& v : m_waiting)
{
if (v == tid)
{
m_waiting.erase(m_waiting.begin() + (&v - m_waiting.data()));
return true;
}
}
for (auto& v : m_signaled)
{
if (v == tid)
{
if (&v == m_signaled.data())
{
return false; // if the thread is signaled, pop() should be used
}
m_signaled.erase(m_signaled.begin() + (&v - m_signaled.data()));
return true;
}
}
return false;
}
case SYS_SYNC_RETRY:
{
return true;
}
}
LOG_ERROR(HLE, "sleep_queue_t['%s']::invalidate(): unsupported protocol (0x%x)", m_name.c_str(), protocol);
Emu.Pause();
return 0;
}
bool sleep_queue_t::signal_selected(u32 tid)
{
assert(tid);
std::lock_guard<std::mutex> lock(m_mutex);
for (auto& v : m_waiting)
{
if (v == tid)
{
m_waiting.erase(m_waiting.begin() + (&v - m_waiting.data()));
m_signaled.push_back(tid);
return true;
}
}
return false;
}
u32 sleep_queue_t::count()
{
std::lock_guard<std::mutex> lock(m_mutex);
return (u32)m_waiting.size() + (u32)m_signaled.size();
if (!std::uncaught_exception())
{
throw EXCEPTION("Thread not found");
}
}

32
rpcs3/Emu/SysCalls/lv2/sleep_queue.h
View File

@ -41,30 +41,18 @@ enum
SYS_SYNC_NOT_ADAPTIVE = 0x2000,
};
class sleep_queue_t
using sleep_queue_t = std::deque<std::shared_ptr<CPUThread>>;
// automatic object handling adding threads to the sleep queue
class sleep_queue_entry_t final
{
std::vector<u32> m_waiting;
std::vector<u32> m_signaled;
std::mutex m_mutex;
std::string m_name;
CPUThread& m_thread;
sleep_queue_t& m_queue;
public:
const u64 name;
// adds specified thread to the sleep queue
sleep_queue_entry_t(CPUThread& cpu, sleep_queue_t& queue);
sleep_queue_t(u64 name = 0)
: name(name)
{
}
~sleep_queue_t();
void set_full_name(const std::string& name) { m_name = name; }
const std::string& get_full_name() { return m_name; }
void push(u32 tid, u32 protocol);
bool pop(u32 tid, u32 protocol);
u32 signal(u32 protocol);
bool signal_selected(u32 tid);
bool invalidate(u32 tid, u32 protocol);
u32 count();
// removes specified thread from the sleep queue
~sleep_queue_entry_t() noexcept(false);
};

168
rpcs3/Emu/SysCalls/lv2/sys_cond.cpp
View File

@ -4,9 +4,7 @@
#include "Emu/IdManager.h"
#include "Emu/SysCalls/SysCalls.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/PPUThread.h"
#include "sleep_queue.h"
#include "sys_mutex.h"
#include "sys_cond.h"
@ -56,7 +54,7 @@ s32 sys_cond_destroy(u32 cond_id)
return CELL_ESRCH;
}
if (!cond->waiters.empty() || cond->signaled)
if (!cond->sq.empty())
{
return CELL_EBUSY;
}
@ -84,11 +82,21 @@ s32 sys_cond_signal(u32 cond_id)
return CELL_ESRCH;
}
if (!cond->waiters.empty())
for (auto& thread : cond->sq)
{
cond->signaled++;
cond->waiters.erase(cond->waiters.begin());
cond->cv.notify_one();
// signal one waiting thread; protocol is ignored in current implementation
if (thread->Signal())
{
cond->sent++;
if (!cond->mutex->owner)
{
// set the appropriate mutex owner if free; protocol is ignored in current implementation
cond->mutex->owner = thread;
}
return CELL_OK;
}
}
return CELL_OK;
@ -107,11 +115,19 @@ s32 sys_cond_signal_all(u32 cond_id)
return CELL_ESRCH;
}
if (const u64 count = cond->waiters.size())
for (auto& thread : cond->sq)
{
cond->signaled += count;
cond->waiters.clear();
cond->cv.notify_all();
// signal all waiting threads; protocol is ignored in current implementation
if (thread->Signal())
{
cond->sent++;
if (!cond->mutex->owner)
{
// set the appropriate mutex owner if free; protocol is ignored in current implementation
cond->mutex->owner = thread;
}
}
}
return CELL_OK;
@ -130,26 +146,29 @@ s32 sys_cond_signal_to(u32 cond_id, u32 thread_id)
return CELL_ESRCH;
}
if (!Emu.GetIdManager().check_id<CPUThread>(thread_id))
// TODO: check if CELL_ESRCH is returned if thread_id is invalid
for (auto& thread : cond->sq)
{
return CELL_ESRCH;
// signal specified thread
if (thread->GetId() == thread_id && thread->Signal())
{
cond->sent++;
if (!cond->mutex->owner)
{
// set the appropriate mutex owner if free; protocol is ignored in current implementation
cond->mutex->owner = thread;
}
return CELL_OK;
}
}
const auto found = cond->waiters.find(thread_id);
if (found == cond->waiters.end())
{
return CELL_EPERM;
}
cond->signaled++;
cond->waiters.erase(found);
cond->cv.notify_one();
return CELL_OK;
return CELL_EPERM;
}
s32 sys_cond_wait(PPUThread& CPU, u32 cond_id, u64 timeout)
s32 sys_cond_wait(PPUThread& ppu, u32 cond_id, u64 timeout)
{
sys_cond.Log("sys_cond_wait(cond_id=0x%x, timeout=%lld)", cond_id, timeout);
@ -164,56 +183,87 @@ s32 sys_cond_wait(PPUThread& CPU, u32 cond_id, u64 timeout)
return CELL_ESRCH;
}
const auto thread = Emu.GetIdManager().get<PPUThread>(CPU.GetId());
if (cond->mutex->owner.owner_before(thread) || thread.owner_before(cond->mutex->owner)) // check equality
// check current ownership
if (cond->mutex->owner.get() != &ppu)
{
return CELL_EPERM;
}
// add waiter; protocol is ignored in current implementation
cond->waiters.emplace(CPU.GetId());
// unlock mutex
cond->mutex->owner.reset();
if (cond->mutex->waiters)
{
cond->mutex->cv.notify_one();
}
// save recursive value
const u32 recursive_value = cond->mutex->recursive_count.exchange(0);
while (!cond->mutex->owner.expired() || !cond->signaled || cond->waiters.count(CPU.GetId()))
if (cond->mutex->sq.size())
{
CHECK_EMU_STATUS;
const bool is_timedout = timeout && get_system_time() - start_time > timeout;
// pick another owner; protocol is ignored in current implementation
cond->mutex->owner = cond->mutex->sq.front();
// check timeout
if (is_timedout && cond->mutex->owner.expired())
if (!cond->mutex->owner->Signal())
{
// cancel waiting if the mutex is free, restore its owner and recursive value
cond->mutex->owner = thread;
cond->mutex->recursive_count = recursive_value;
if (!cond->waiters.erase(CPU.GetId()))
{
throw EXCEPTION("Unexpected");
}
return CELL_ETIMEDOUT;
throw EXCEPTION("Mutex owner not signaled");
}
// wait on appropriate condition variable
(cond->signaled || is_timedout ? cond->mutex->cv : cond->cv).wait_for(lv2_lock, std::chrono::milliseconds(1));
}
// reown the mutex and restore its recursive value
cond->mutex->owner = thread;
{
// add waiter; protocol is ignored in current implementation
sleep_queue_entry_t waiter(ppu, cond->sq);
while (!ppu.Signaled())
{
if (timeout)
{
const u64 passed = get_system_time() - start_time;
if (passed >= timeout || ppu.cv.wait_for(lv2_lock, std::chrono::microseconds(timeout - passed)) == std::cv_status::timeout)
{
break;
}
}
else
{
ppu.cv.wait(lv2_lock);
}
CHECK_EMU_STATUS;
}
cond->recv++;
}
// reown the mutex (could be set when notified)
if (!cond->mutex->owner)
{
cond->mutex->owner = std::move(ppu.shared_from_this());
}
if (cond->mutex->owner.get() != &ppu)
{
// add waiter; protocol is ignored in current implementation
sleep_queue_entry_t waiter(ppu, cond->mutex->sq);
while (!ppu.Signaled())
{
ppu.cv.wait(lv2_lock);
CHECK_EMU_STATUS;
}
if (cond->mutex->owner.get() != &ppu)
{
throw EXCEPTION("Unexpected mutex owner");
}
}
// restore the recursive value
cond->mutex->recursive_count = recursive_value;
cond->signaled--;
// check timeout
if (timeout && get_system_time() - start_time > timeout)
{
return CELL_ETIMEDOUT;
}
return CELL_OK;
}

13
rpcs3/Emu/SysCalls/lv2/sys_cond.h
View File

@ -1,5 +1,7 @@
#pragma once
#include "sleep_queue.h"
namespace vm { using namespace ps3; }
struct lv2_mutex_t;
@ -22,17 +24,14 @@ struct lv2_cond_t
const u64 name;
const std::shared_ptr<lv2_mutex_t> mutex; // associated mutex
std::atomic<u64> signaled;
sleep_queue_t sq;
// TODO: use sleep queue, possibly remove condition variable
std::condition_variable cv;
std::unordered_set<u32> waiters;
std::atomic<u32> sent{ 0 };
std::atomic<u32> recv{ 0 };
lv2_cond_t(const std::shared_ptr<lv2_mutex_t>& mutex, u64 name)
: mutex(mutex)
, name(name)
, signaled(0)
//, waiters(0)
{
}
};
@ -44,7 +43,7 @@ class PPUThread;
// SysCalls
s32 sys_cond_create(vm::ptr<u32> cond_id, u32 mutex_id, vm::ptr<sys_cond_attribute_t> attr);
s32 sys_cond_destroy(u32 cond_id);
s32 sys_cond_wait(PPUThread& CPU, u32 cond_id, u64 timeout);
s32 sys_cond_wait(PPUThread& ppu, u32 cond_id, u64 timeout);
s32 sys_cond_signal(u32 cond_id);
s32 sys_cond_signal_all(u32 cond_id);
s32 sys_cond_signal_to(u32 cond_id, u32 thread_id);

1
rpcs3/Emu/SysCalls/lv2/sys_event_flag.cpp
View File

@ -4,7 +4,6 @@
#include "Emu/IdManager.h"
#include "Emu/SysCalls/SysCalls.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/PPUThread.h"
#include "sleep_queue.h"
#include "sys_event_flag.h"

1
rpcs3/Emu/SysCalls/lv2/sys_lwmutex.cpp
View File

@ -4,7 +4,6 @@
#include "Emu/IdManager.h"
#include "Emu/SysCalls/SysCalls.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/PPUThread.h"
#include "sleep_queue.h"
#include "sys_lwmutex.h"

86
rpcs3/Emu/SysCalls/lv2/sys_mutex.cpp
View File

@ -4,9 +4,7 @@
#include "Emu/IdManager.h"
#include "Emu/SysCalls/SysCalls.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/PPUThread.h"
#include "sleep_queue.h"
#include "sys_mutex.h"
SysCallBase sys_mutex("sys_mutex");
@ -36,8 +34,7 @@ s32 sys_mutex_create(vm::ptr<u32> mutex_id, vm::ptr<sys_mutex_attribute_t> attr)
if ((!recursive && attr->recursive != SYS_SYNC_NOT_RECURSIVE) || attr->pshared != SYS_SYNC_NOT_PROCESS_SHARED || attr->adaptive != SYS_SYNC_NOT_ADAPTIVE || attr->ipc_key.data() || attr->flags.data())
{
sys_mutex.Error("sys_mutex_create(): unknown attributes (recursive=0x%x, pshared=0x%x, adaptive=0x%x, ipc_key=0x%llx, flags=0x%x)",
attr->recursive, attr->pshared, attr->adaptive, attr->ipc_key, attr->flags);
sys_mutex.Error("sys_mutex_create(): unknown attributes (recursive=0x%x, pshared=0x%x, adaptive=0x%x, ipc_key=0x%llx, flags=0x%x)", attr->recursive, attr->pshared, attr->adaptive, attr->ipc_key, attr->flags);
return CELL_EINVAL;
}
@ -60,13 +57,8 @@ s32 sys_mutex_destroy(u32 mutex_id)
return CELL_ESRCH;
}
if (!mutex->owner.expired())
{
return CELL_EBUSY;
}
// assuming that the mutex is locked immediately by another waiting thread when unlocked
if (mutex->waiters)
if (!mutex->owner || !mutex->sq.empty())
{
return CELL_EBUSY;
}
@ -81,7 +73,7 @@ s32 sys_mutex_destroy(u32 mutex_id)
return CELL_OK;
}
s32 sys_mutex_lock(PPUThread& CPU, u32 mutex_id, u64 timeout)
s32 sys_mutex_lock(PPUThread& ppu, u32 mutex_id, u64 timeout)
{
sys_mutex.Log("sys_mutex_lock(mutex_id=0x%x, timeout=0x%llx)", mutex_id, timeout);
@ -96,9 +88,8 @@ s32 sys_mutex_lock(PPUThread& CPU, u32 mutex_id, u64 timeout)
return CELL_ESRCH;
}
const auto thread = Emu.GetIdManager().get<PPUThread>(CPU.GetId());
if (!mutex->owner.owner_before(thread) && !thread.owner_before(mutex->owner)) // check equality
// check current ownership
if (mutex->owner.get() == &ppu)
{
if (mutex->recursive)
{
@ -115,29 +106,46 @@ s32 sys_mutex_lock(PPUThread& CPU, u32 mutex_id, u64 timeout)
return CELL_EDEADLK;
}
// protocol is ignored in current implementation
mutex->waiters++;
// lock immediately if not locked
if (!mutex->owner)
{
mutex->owner = std::move(ppu.shared_from_this());
while (!mutex->owner.expired())
return CELL_OK;
}
// add waiter; protocol is ignored in current implementation
sleep_queue_entry_t waiter(ppu, mutex->sq);
while (!ppu.Signaled())
{
CHECK_EMU_STATUS;
if (timeout && get_system_time() - start_time > timeout)
if (timeout)
{
mutex->waiters--;
return CELL_ETIMEDOUT;
}
const u64 passed = get_system_time() - start_time;
mutex->cv.wait_for(lv2_lock, std::chrono::milliseconds(1));
if (passed >= timeout || ppu.cv.wait_for(lv2_lock, std::chrono::microseconds(timeout - passed)) == std::cv_status::timeout)
{
return CELL_ETIMEDOUT;
}
}
else
{
ppu.cv.wait(lv2_lock);
}
}
mutex->owner = thread;
mutex->waiters--;
// new owner must be set when unlocked
if (mutex->owner.get() != &ppu)
{
throw EXCEPTION("Unexpected mutex owner");
}
return CELL_OK;
}
s32 sys_mutex_trylock(PPUThread& CPU, u32 mutex_id)
s32 sys_mutex_trylock(PPUThread& ppu, u32 mutex_id)
{
sys_mutex.Log("sys_mutex_trylock(mutex_id=0x%x)", mutex_id);
@ -150,9 +158,8 @@ s32 sys_mutex_trylock(PPUThread& CPU, u32 mutex_id)
return CELL_ESRCH;
}
const auto thread = Emu.GetIdManager().get<PPUThread>(CPU.GetId());
if (!mutex->owner.owner_before(thread) && !thread.owner_before(mutex->owner)) // check equality
// check current ownership
if (mutex->owner.get() == &ppu)
{
if (mutex->recursive)
{
@ -169,17 +176,18 @@ s32 sys_mutex_trylock(PPUThread& CPU, u32 mutex_id)
return CELL_EDEADLK;
}
if (!mutex->owner.expired())
if (mutex->owner)
{
return CELL_EBUSY;
}
mutex->owner = thread;
// own the mutex if free
mutex->owner = std::move(ppu.shared_from_this());
return CELL_OK;
}
s32 sys_mutex_unlock(PPUThread& CPU, u32 mutex_id)
s32 sys_mutex_unlock(PPUThread& ppu, u32 mutex_id)
{
sys_mutex.Log("sys_mutex_unlock(mutex_id=0x%x)", mutex_id);
@ -192,9 +200,8 @@ s32 sys_mutex_unlock(PPUThread& CPU, u32 mutex_id)
return CELL_ESRCH;
}
const auto thread = Emu.GetIdManager().get<PPUThread>(CPU.GetId());
if (mutex->owner.owner_before(thread) || thread.owner_before(mutex->owner)) // check inequality
// check current ownership
if (mutex->owner.get() != &ppu)
{
return CELL_EPERM;
}
@ -210,11 +217,18 @@ s32 sys_mutex_unlock(PPUThread& CPU, u32 mutex_id)
}
else
{
// free the mutex
mutex->owner.reset();
if (mutex->waiters)
if (mutex->sq.size())
{
mutex->cv.notify_one();
// pick another owner; protocol is ignored in current implementation
mutex->owner = mutex->sq.front();
if (!mutex->owner->Signal())
{
throw EXCEPTION("Mutex owner not signaled");
}
}
}

21
rpcs3/Emu/SysCalls/lv2/sys_mutex.h
View File

@ -1,5 +1,7 @@
#pragma once
#include "sleep_queue.h"
namespace vm { using namespace ps3; }
struct sys_mutex_attribute_t
@ -25,21 +27,16 @@ struct lv2_mutex_t
const u32 protocol;
const u64 name;
std::atomic<u32> cond_count; // count of condition variables associated
std::atomic<u32> recursive_count;
std::weak_ptr<CPUThread> owner;
std::atomic<u32> cond_count{ 0 }; // count of condition variables associated
std::atomic<u32> recursive_count{ 0 };
std::shared_ptr<CPUThread> owner;
// TODO: use sleep queue, possibly remove condition variable
std::condition_variable cv;
std::atomic<u32> waiters;
sleep_queue_t sq;
lv2_mutex_t(bool recursive, u32 protocol, u64 name)
: recursive(recursive)
, protocol(protocol)
, name(name)
, cond_count(0)
, recursive_count(0)
, waiters(0)
{
}
};
@ -51,6 +48,6 @@ class PPUThread;
// SysCalls
s32 sys_mutex_create(vm::ptr<u32> mutex_id, vm::ptr<sys_mutex_attribute_t> attr);
s32 sys_mutex_destroy(u32 mutex_id);
s32 sys_mutex_lock(PPUThread& CPU, u32 mutex_id, u64 timeout);
s32 sys_mutex_trylock(PPUThread& CPU, u32 mutex_id);
s32 sys_mutex_unlock(PPUThread& CPU, u32 mutex_id);
s32 sys_mutex_lock(PPUThread& ppu, u32 mutex_id, u64 timeout);
s32 sys_mutex_trylock(PPUThread& ppu, u32 mutex_id);
s32 sys_mutex_unlock(PPUThread& ppu, u32 mutex_id);

44
rpcs3/Emu/SysCalls/lv2/sys_ppu_thread.cpp
View File

@ -5,21 +5,39 @@
#include "Emu/IdManager.h"
#include "Emu/DbgCommand.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/PPUThread.h"
#include "sys_mutex.h"
#include "sys_ppu_thread.h"
SysCallBase sys_ppu_thread("sys_ppu_thread");
void _sys_ppu_thread_exit(PPUThread& CPU, u64 errorcode)
void _sys_ppu_thread_exit(PPUThread& ppu, u64 errorcode)
{
sys_ppu_thread.Log("_sys_ppu_thread_exit(errorcode=0x%llx)", errorcode);
LV2_LOCK;
if (!CPU.is_joinable)
// get all sys_mutex objects
for (auto& _id : Emu.GetIdManager().get_data<lv2_mutex_t>())
{
const u32 id = CPU.GetId();
const auto mutex = std::static_pointer_cast<lv2_mutex_t>(_id.data);
// unlock mutex if locked by this thread
if (mutex->owner.get() == &ppu)
{
mutex->owner.reset();
if (mutex->sq.size())
{
mutex->owner = mutex->sq.front();
mutex->owner->Signal();
}
}
}
if (!ppu.is_joinable)
{
const u32 id = ppu.GetId();
CallAfter([id]()
{
@ -27,7 +45,7 @@ void _sys_ppu_thread_exit(PPUThread& CPU, u64 errorcode)
});
}
CPU.Exit();
ppu.Exit();
}
void sys_ppu_thread_yield()
@ -37,7 +55,7 @@ void sys_ppu_thread_yield()
std::this_thread::yield();
}
s32 sys_ppu_thread_join(PPUThread& CPU, u32 thread_id, vm::ptr<u64> vptr)
s32 sys_ppu_thread_join(PPUThread& ppu, u32 thread_id, vm::ptr<u64> vptr)
{
sys_ppu_thread.Warning("sys_ppu_thread_join(thread_id=0x%x, vptr=*0x%x)", thread_id, vptr);
@ -55,7 +73,7 @@ s32 sys_ppu_thread_join(PPUThread& CPU, u32 thread_id, vm::ptr<u64> vptr)
return CELL_EINVAL;
}
if (&CPU == thread.get())
if (&ppu == thread.get())
{
return CELL_EDEADLK;
}
@ -68,7 +86,7 @@ s32 sys_ppu_thread_join(PPUThread& CPU, u32 thread_id, vm::ptr<u64> vptr)
{
CHECK_EMU_STATUS;
thread->cv.wait_for(lv2_lock, std::chrono::milliseconds(1));
ppu.cv.wait_for(lv2_lock, std::chrono::milliseconds(1));
}
// get exit status from the register
@ -109,13 +127,13 @@ s32 sys_ppu_thread_detach(u32 thread_id)
return CELL_OK;
}
void sys_ppu_thread_get_join_state(PPUThread& CPU, vm::ptr<s32> isjoinable)
void sys_ppu_thread_get_join_state(PPUThread& ppu, vm::ptr<s32> isjoinable)
{
sys_ppu_thread.Warning("sys_ppu_thread_get_join_state(isjoinable=*0x%x)", isjoinable);
LV2_LOCK;
*isjoinable = CPU.is_joinable;
*isjoinable = ppu.is_joinable;
}
s32 sys_ppu_thread_set_priority(u32 thread_id, s32 prio)
@ -159,12 +177,12 @@ s32 sys_ppu_thread_get_priority(u32 thread_id, vm::ptr<s32> priop)
return CELL_OK;
}
s32 sys_ppu_thread_get_stack_information(PPUThread& CPU, vm::ptr<sys_ppu_thread_stack_t> sp)
s32 sys_ppu_thread_get_stack_information(PPUThread& ppu, vm::ptr<sys_ppu_thread_stack_t> sp)
{
sys_ppu_thread.Log("sys_ppu_thread_get_stack_information(sp=*0x%x)", sp);
sp->pst_addr = CPU.stack_addr;
sp->pst_size = CPU.stack_size;
sp->pst_addr = ppu.stack_addr;
sp->pst_size = ppu.stack_size;
return CELL_OK;
}

8
rpcs3/Emu/SysCalls/lv2/sys_ppu_thread.h
View File

@ -33,14 +33,14 @@ struct ppu_thread_param_t
u32 ppu_thread_create(u32 entry, u64 arg, s32 prio, u32 stacksize, bool is_joinable, bool is_interrupt, std::string name, std::function<void(PPUThread&)> task = nullptr);
// SysCalls
void _sys_ppu_thread_exit(PPUThread& CPU, u64 errorcode);
void _sys_ppu_thread_exit(PPUThread& ppu, u64 errorcode);
void sys_ppu_thread_yield();
s32 sys_ppu_thread_join(PPUThread& CPU, u32 thread_id, vm::ptr<u64> vptr);
s32 sys_ppu_thread_join(PPUThread& ppu, u32 thread_id, vm::ptr<u64> vptr);
s32 sys_ppu_thread_detach(u32 thread_id);
void sys_ppu_thread_get_join_state(PPUThread& CPU, vm::ptr<s32> isjoinable);
void sys_ppu_thread_get_join_state(PPUThread& ppu, vm::ptr<s32> isjoinable);
s32 sys_ppu_thread_set_priority(u32 thread_id, s32 prio);
s32 sys_ppu_thread_get_priority(u32 thread_id, vm::ptr<s32> priop);
s32 sys_ppu_thread_get_stack_information(PPUThread& CPU, vm::ptr<sys_ppu_thread_stack_t> sp);
s32 sys_ppu_thread_get_stack_information(PPUThread& ppu, vm::ptr<sys_ppu_thread_stack_t> sp);
s32 sys_ppu_thread_stop(u32 thread_id);
s32 sys_ppu_thread_restart(u32 thread_id);
s32 _sys_ppu_thread_create(vm::ptr<u64> thread_id, vm::ptr<ppu_thread_param_t> param, u64 arg, u64 arg4, s32 prio, u32 stacksize, u64 flags, vm::cptr<char> threadname);

1
rpcs3/Emu/SysCalls/lv2/sys_semaphore.cpp
View File

@ -4,7 +4,6 @@
#include "Emu/IdManager.h"
#include "Emu/SysCalls/SysCalls.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "Emu/Cell/PPUThread.h"
#include "sleep_queue.h"
#include "sys_semaphore.h"

6
rpcs3/Emu/System.cpp
View File

@ -357,7 +357,11 @@ void Emulator::Stop()
// notify all threads
for (auto& t : GetCPU().GetAllThreads())
{
t->Stop(); // signal / trigger status check
std::lock_guard<std::mutex> lock(t->mutex);
t->Sleep(); // trigger status check
t->cv.notify_one(); // signal
}
}

1
rpcs3/Loader/ELF64.cpp
View File

@ -10,7 +10,6 @@
#include "Emu/SysCalls/ModuleManager.h"
#include "Emu/SysCalls/lv2/sys_prx.h"
#include "Emu/Cell/PPUInstrTable.h"
#include "Emu/CPU/CPUThreadManager.h"
#include "ELF64.h"
#include "Ini.h"