event_queue_t, event_port_t rewritten

rpcs3/Emu/ARMv7/ARMv7Interpreter.cpp

@@ -296,7 +296,7 @@ namespace ARMv7_instrs
 
 			context.fmt_debug_str("0x%08x: %s", context.thread.PC, context.debug_str);
 
-			LV2_LOCK(0);
+			LV2_LOCK;
 
 			auto found = g_armv7_dump.find(context.thread.PC);
 			if (found != g_armv7_dump.end())

rpcs3/Emu/ARMv7/Modules/sceLibc.cpp

@@ -152,7 +152,7 @@ namespace sce_libc_func
 	{
 		sceLibc.Warning("__cxa_atexit(func=0x%x, arg=0x%x, dso=0x%x)", func, arg, dso);
 		
-		LV2_LOCK(0);
+		LV2_LOCK;
 
 		g_atexit.insert(g_atexit.begin(), [func, arg, dso](ARMv7Context& context)
 		{
@@ -164,7 +164,7 @@ namespace sce_libc_func
 	{
 		sceLibc.Warning("__aeabi_atexit(arg=0x%x, func=0x%x, dso=0x%x)", arg, func, dso);
 
-		LV2_LOCK(0);
+		LV2_LOCK;
 
 		g_atexit.insert(g_atexit.begin(), [func, arg, dso](ARMv7Context& context)
 		{
@@ -176,7 +176,7 @@ namespace sce_libc_func
 	{
 		sceLibc.Warning("exit()");
 		
-		LV2_LOCK(0);
+		LV2_LOCK;
 
 		for (auto func : g_atexit)
 		{

rpcs3/Emu/Cell/SPUThread.cpp

@@ -10,6 +10,7 @@
 #include "Emu/SysCalls/ErrorCodes.h"
 #include "Emu/SysCalls/lv2/sys_spu.h"
 #include "Emu/SysCalls/lv2/sys_event_flag.h"
+#include "Emu/SysCalls/lv2/sys_event.h"
 #include "Emu/SysCalls/lv2/sys_time.h"
 
 #include "Emu/Cell/SPUDisAsm.h"
@@ -589,24 +590,26 @@ void SPUThread::set_ch_value(u32 ch, u32 value)
 					LOG_NOTICE(SPU, "sys_spu_thread_send_event(spup=%d, data0=0x%x, data1=0x%x)", spup, value & 0x00ffffff, data);
 				}
 
-				std::shared_ptr<EventPort> port;// = SPUPs[spup];
+				LV2_LOCK;
 
-				std::lock_guard<std::mutex> lock(port->m_mutex);
+				std::shared_ptr<event_queue_t> queue = this->spup[spup].lock();
 
-				if (!port->eq)
+				if (!queue)
 				{
 					LOG_WARNING(SPU, "sys_spu_thread_send_event(spup=%d, data0=0x%x, data1=0x%x): event queue not connected", spup, (value & 0x00ffffff), data);
 					ch_in_mbox.push_uncond(CELL_ENOTCONN); // TODO: check error passing
 					return;
 				}
 
-				if (!port->eq->events.push(SYS_SPU_THREAD_EVENT_USER_KEY, GetId(), ((u64)spup << 32) | (value & 0x00ffffff), data))
+				if (queue->events.size() >= queue->size)
 				{
 					ch_in_mbox.push_uncond(CELL_EBUSY);
 					return;
 				}
 
+				queue->events.emplace_back(SYS_SPU_THREAD_EVENT_USER_KEY, GetId(), ((u64)spup << 32) | (value & 0x00ffffff), data);
 				ch_in_mbox.push_uncond(CELL_OK);
+				queue->cv.notify_one();
 				return;
 			}
 			else if (code < 128)
@@ -627,23 +630,25 @@ void SPUThread::set_ch_value(u32 ch, u32 value)
 					LOG_WARNING(SPU, "sys_spu_thread_throw_event(spup=%d, data0=0x%x, data1=0x%x)", spup, value & 0x00ffffff, data);
 				}
 
-				std::shared_ptr<EventPort> port;// = SPUPs[spup];
+				LV2_LOCK;
 
-				std::lock_guard<std::mutex> lock(port->m_mutex);
+				std::shared_ptr<event_queue_t> queue = this->spup[spup].lock();
 
-				if (!port->eq)
+				if (!queue)
 				{
 					LOG_WARNING(SPU, "sys_spu_thread_throw_event(spup=%d, data0=0x%x, data1=0x%x): event queue not connected", spup, (value & 0x00ffffff), data);
 					return;
 				}
 
 				// TODO: check passing spup value
-				if (!port->eq->events.push(SYS_SPU_THREAD_EVENT_USER_KEY, GetId(), ((u64)spup << 32) | (value & 0x00ffffff), data))
+				if (queue->events.size() >= queue->size)
 				{
 					LOG_WARNING(SPU, "sys_spu_thread_throw_event(spup=%d, data0=0x%x, data1=0x%x) failed (queue is full)", spup, (value & 0x00ffffff), data);
 					return;
 				}
 
+				queue->events.emplace_back(SYS_SPU_THREAD_EVENT_USER_KEY, GetId(), ((u64)spup << 32) | (value & 0x00ffffff), data);
+				queue->cv.notify_one();
 				return;
 			}
 			else if (code == 128)
@@ -954,74 +959,46 @@ void SPUThread::stop_and_signal(u32 code)
 			LOG_NOTICE(SPU, "sys_spu_thread_receive_event(spuq=0x%x)", spuq);
 		}
 
-		std::shared_ptr<EventQueue> eq;
+		LV2_LOCK;
-		//if (!SPUQs.GetEventQueue(FIX_SPUQ(spuq), eq))
-		//{
-		//	ch_in_mbox.push_uncond(CELL_EINVAL); // TODO: check error value
-		//	return;
-		//}
 
-		u32 tid = GetId();
+		auto found = this->spuq.find(spuq);
-
+		if (found == this->spuq.end())
-		eq->sq.push(tid, eq->protocol); // add thread to sleep queue
-
-		while (true)
 		{
-			u32 old_owner = eq->owner.compare_and_swap(0, tid);
+			ch_in_mbox.push_uncond(CELL_EINVAL); // TODO: check error value
+			return;
+		}
 		
-			switch (s32 res = old_owner ? (old_owner == tid ? 1 : 2) : 0)
+		std::shared_ptr<event_queue_t> queue = found->second;
-			{
-			case 0:
-			{
-				const u32 next = eq->events.count() ? eq->sq.signal(eq->protocol) : 0;
-				if (next != tid)
-				{
-					if (!eq->owner.compare_and_swap_test(tid, next))
-					{
-						assert(!"sys_spu_thread_receive_event() failed (I)");
-					}
-					break;
-				}
-				// fallthrough
-			}
-			case 1:
-			{
-				sys_event_data event;
-				eq->events.pop(event);
-				if (!eq->owner.compare_and_swap_test(tid, 0))
-				{
-					assert(!"sys_spu_thread_receive_event() failed (II)");
-				}
-				ch_in_mbox.push_uncond(CELL_OK);
-				ch_in_mbox.push_uncond((u32)event.data1);
-				ch_in_mbox.push_uncond((u32)event.data2);
-				ch_in_mbox.push_uncond((u32)event.data3);
-				if (!eq->sq.invalidate(tid, eq->protocol) && !eq->sq.pop(tid, eq->protocol))
-				{
-					assert(!"sys_spu_thread_receive_event() failed (receiving)");
-				}
-				return;
-			}
-			}
 
-			if (!~old_owner)
+		// protocol is ignored in current implementation
+		queue->waiters++;
+
+		while (queue->events.empty())
+		{
+			if (queue->waiters < 0)
 			{
-				if (!eq->sq.invalidate(tid, eq->protocol))
-				{
-					assert(!"sys_spu_thread_receive_event() failed (cancelling)");
-				}
 				ch_in_mbox.push_uncond(CELL_ECANCELED);
+				queue->waiters--;
 				return;
 			}
 
-			std::this_thread::sleep_for(std::chrono::milliseconds(1)); // hack
 			if (Emu.IsStopped())
 			{
 				LOG_WARNING(SPU, "sys_spu_thread_receive_event(spuq=0x%x) aborted", spuq);
 				return;
 			}
+
+			queue->cv.wait_for(lv2_lock, std::chrono::milliseconds(1));
 		}
 
+		auto& event = queue->events.front();
+		ch_in_mbox.push_uncond(CELL_OK);
+		ch_in_mbox.push_uncond((u32)event.data1);
+		ch_in_mbox.push_uncond((u32)event.data2);
+		ch_in_mbox.push_uncond((u32)event.data3);
+		
+		queue->events.pop_front();
+		queue->waiters--;
 		return;
 	}
 

rpcs3/Emu/Cell/SPUThread.h

@@ -2,11 +2,11 @@
 #include "Emu/Cell/Common.h"
 #include "Emu/CPU/CPUThread.h"
 #include "Emu/Cell/SPUContext.h"
-#include "Emu/SysCalls/lv2/sleep_queue.h"
-#include "Emu/SysCalls/lv2/sys_event.h"
-#include "Emu/Event.h"
 #include "MFC.h"
 
+struct event_queue_t;
+struct event_port_t;
+
 // SPU Channels
 enum : u32
 {
@@ -507,6 +507,9 @@ public:
 
 	u32 tg_id; // SPU Thread Group Id
 
+	std::unordered_map<u32, std::shared_ptr<event_queue_t>> spuq; // Event Queue Keys for SPU Thread
+	std::weak_ptr<event_queue_t> spup[64]; // SPU Ports
+
 	void write_snr(bool number, u32 value)
 	{
 		if (!number)

rpcs3/Emu/Event.cpp

@@ -11,74 +11,77 @@ void EventManager::Init()
 
 void EventManager::Clear()
 {
-	key_map.clear();
+	eq_map.clear();
 }
 
 bool EventManager::CheckKey(u64 key)
 {
-	if (!key) return true;
+	if (!key)
-	std::lock_guard<std::mutex> lock(m_lock);
-
-	return key_map.find(key) != key_map.end();
-}
-
-bool EventManager::RegisterKey(std::shared_ptr<EventQueue>& data, u64 key)
-{
-	if (!key) return true;
-	std::lock_guard<std::mutex> lock(m_lock);
-
-	if (key_map.find(key) != key_map.end()) return false;
-
-	for (auto& v : key_map)
 	{
-		if (v.second == data) return false;
+		// never exists
+		return false;
 	}
 
-	key_map[key] = data;
-
-	return true;
-}
-
-bool EventManager::GetEventQueue(u64 key, std::shared_ptr<EventQueue>& data)
-{
-	data = nullptr;
-	if (!key) return false;
 	std::lock_guard<std::mutex> lock(m_lock);
 
-	auto f = key_map.find(key);
+	return eq_map.find(key) != eq_map.end();
-	if (f != key_map.end())
+}
+
+bool EventManager::RegisterKey(std::shared_ptr<event_queue_t>& data, u64 key)
+{
+	if (!key)
 	{
-		data = f->second;
+		// always ok
 		return true;
 	}
-	return false;
-}
 
-bool EventManager::UnregisterKey(u64 key)
-{
-	if (!key) return false;
 	std::lock_guard<std::mutex> lock(m_lock);
 
-	auto f = key_map.find(key);
+	if (eq_map.find(key) != eq_map.end())
-	if (f != key_map.end())
-	{
-		key_map.erase(f);
-		return true;
-	}
-	return false;
-}
-
-bool EventManager::SendEvent(u64 key, u64 source, u64 d1, u64 d2, u64 d3)
-{
-	if (!key) return false;
-	std::lock_guard<std::mutex> lock(m_lock);
-
-	auto f = key_map.find(key);
-	if (f == key_map.end())
 	{
 		return false;
 	}
 
-	f->second->events.push(source, d1, d2, d3);
+	eq_map[key] = data;
+
 	return true;
 }
+
+bool EventManager::UnregisterKey(u64 key)
+{
+	if (!key)
+	{
+		// always ok
+		return true;
+	}
+
+	std::lock_guard<std::mutex> lock(m_lock);
+
+	auto f = eq_map.find(key);
+	if (f != eq_map.end())
+	{
+		eq_map.erase(f);
+		return true;
+	}
+
+	return false;
+}
+
+std::shared_ptr<event_queue_t> EventManager::GetEventQueue(u64 key)
+{
+	if (!key)
+	{
+		// never exists
+		return nullptr;
+	}
+
+	std::lock_guard<std::mutex> lock(m_lock);
+
+	auto f = eq_map.find(key);
+	if (f != eq_map.end())
+	{
+		return f->second;
+	}
+
+	return nullptr;
+}

rpcs3/Emu/Event.h

@@ -1,19 +1,19 @@
 #pragma once
 #include <unordered_map>
 
-struct EventQueue;
+struct event_queue_t;
 
 class EventManager
 {
 	std::mutex m_lock;
-	std::unordered_map<u64, std::shared_ptr<EventQueue>> key_map;
+	std::unordered_map<u64, std::shared_ptr<event_queue_t>> eq_map;
 
 public:
 	void Init();
 	void Clear();
 	bool CheckKey(u64 key);
-	bool RegisterKey(std::shared_ptr<EventQueue>& data, u64 key);
+	bool RegisterKey(std::shared_ptr<event_queue_t>& data, u64 key);
-	bool GetEventQueue(u64 key, std::shared_ptr<EventQueue>& data);
 	bool UnregisterKey(u64 key);
-	bool SendEvent(u64 key, u64 source, u64 d1, u64 d2, u64 d3);
+
+	std::shared_ptr<event_queue_t> GetEventQueue(u64 key);
 };

rpcs3/Emu/Memory/Memory.cpp

@@ -8,7 +8,7 @@ MemoryBase Memory;
 
 u32 MemoryBase::InitRawSPU(MemoryBlock* raw_spu)
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	u32 index;
 	for (index = 0; index < sizeof(RawSPUMem) / sizeof(RawSPUMem[0]); index++)
@@ -26,7 +26,7 @@ u32 MemoryBase::InitRawSPU(MemoryBlock* raw_spu)
 
 void MemoryBase::CloseRawSPU(MemoryBlock* raw_spu, const u32 num)
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	for (int i = 0; i < MemoryBlocks.size(); ++i)
 	{
@@ -41,8 +41,6 @@ void MemoryBase::CloseRawSPU(MemoryBlock* raw_spu, const u32 num)
 
 void MemoryBase::Init(MemoryType type)
 {
-	LV2_LOCK(0);
-
 	if (m_inited) return;
 	m_inited = true;
 
@@ -88,8 +86,6 @@ void MemoryBase::Init(MemoryType type)
 
 void MemoryBase::Close()
 {
-	LV2_LOCK(0);
-
 	if (!m_inited) return;
 	m_inited = false;
 
@@ -107,7 +103,7 @@ void MemoryBase::Close()
 
 bool MemoryBase::WriteMMIO32(u32 addr, const u32 data)
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	if (RawSPUMem[(addr - RAW_SPU_BASE_ADDR) / RAW_SPU_OFFSET] && ((RawSPUThread*)RawSPUMem[(addr - RAW_SPU_BASE_ADDR) / RAW_SPU_OFFSET])->Write32(addr, data))
 	{
@@ -119,7 +115,7 @@ bool MemoryBase::WriteMMIO32(u32 addr, const u32 data)
 
 bool MemoryBase::ReadMMIO32(u32 addr, u32& result)
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	if (RawSPUMem[(addr - RAW_SPU_BASE_ADDR) / RAW_SPU_OFFSET] && ((RawSPUThread*)RawSPUMem[(addr - RAW_SPU_BASE_ADDR) / RAW_SPU_OFFSET])->Read32(addr, &result))
 	{
@@ -133,7 +129,7 @@ bool MemoryBase::Map(const u32 addr, const u32 size)
 {
 	assert(size && (size | addr) % 4096 == 0);
 
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	for (u32 i = addr / 4096; i < addr / 4096 + size / 4096; i++)
 	{
@@ -151,7 +147,7 @@ bool MemoryBase::Map(const u32 addr, const u32 size)
 
 bool MemoryBase::Unmap(const u32 addr)
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	for (u32 i = 0; i < MemoryBlocks.size(); i++)
 	{
@@ -238,7 +234,7 @@ DynamicMemoryBlockBase::DynamicMemoryBlockBase()
 
 const u32 DynamicMemoryBlockBase::GetUsedSize() const
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	u32 size = 0;
 
@@ -257,7 +253,7 @@ bool DynamicMemoryBlockBase::IsInMyRange(const u32 addr, const u32 size)
 
 MemoryBlock* DynamicMemoryBlockBase::SetRange(const u32 start, const u32 size)
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	m_max_size = PAGE_4K(size);
 	if (!MemoryBlock::SetRange(start, 0))
@@ -271,7 +267,7 @@ MemoryBlock* DynamicMemoryBlockBase::SetRange(const u32 start, const u32 size)
 
 void DynamicMemoryBlockBase::Delete()
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	m_allocated.clear();
 	m_max_size = 0;
@@ -293,7 +289,7 @@ bool DynamicMemoryBlockBase::AllocFixed(u32 addr, u32 size)
 		return false;
 	}
 
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	for (u32 i = 0; i<m_allocated.size(); ++i)
 	{
@@ -334,7 +330,7 @@ u32 DynamicMemoryBlockBase::AllocAlign(u32 size, u32 align)
 		exsize = size + align - 1;
 	}
 
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	for (u32 addr = MemoryBlock::GetStartAddr(); addr <= MemoryBlock::GetEndAddr() - exsize;)
 	{
@@ -375,7 +371,7 @@ bool DynamicMemoryBlockBase::Alloc()
 
 bool DynamicMemoryBlockBase::Free(u32 addr)
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	for (u32 num = 0; num < m_allocated.size(); num++)
 	{

rpcs3/Emu/SysCalls/Modules/cellAudio.cpp

@@ -395,8 +395,16 @@ s32 cellAudioInit()
 			}
 			for (u32 i = 0; i < keys.size(); i++)
 			{
-				// TODO: check event source
+				auto eq = Emu.GetEventManager().GetEventQueue(keys[i]);
-				Emu.GetEventManager().SendEvent(keys[i], 0x10103000e010e07, 0, 0, 0);
+
+				if (eq)
+				{
+					LV2_LOCK;
+
+					// TODO: check event source
+					eq->events.emplace_back(0x10103000e010e07, 0, 0, 0);
+					eq->cv.notify_one();
+				}
 			}
 
 			//const u64 stamp3 = get_system_time();
@@ -776,7 +784,7 @@ int cellAudioCreateNotifyEventQueue(vm::ptr<u32> id, vm::ptr<u64> key)
 	}
 	event_key = (event_key << 48) | 0x80004d494f323221; // left part: 0x8000, 0x8001, 0x8002 ...
 
-	std::shared_ptr<EventQueue> eq(new EventQueue(SYS_SYNC_FIFO, SYS_PPU_QUEUE, event_key, event_key, 32));
+	std::shared_ptr<event_queue_t> eq(new event_queue_t(SYS_SYNC_FIFO, SYS_PPU_QUEUE, event_key, event_key, 32));
 
 	if (!Emu.GetEventManager().RegisterKey(eq, event_key))
 	{

rpcs3/Emu/SysCalls/Modules/cellSpurs.cpp

@@ -1770,8 +1770,8 @@ s32 _cellSpursEventFlagWait(vm::ptr<CellSpursEventFlag> eventFlag, vm::ptr<u16>
 	u16 receivedEventFlag;
 	if (recv) {
 		// Block till something happens
-		vm::var<sys_event_data> data;
+		vm::var<sys_event_t> data;
-		auto rc = sys_event_queue_receive(eventFlag->m.eventQueueId, data, 0);
+		auto rc = sys_event_queue_receive(GetCurrentPPUThread(), eventFlag->m.eventQueueId, data, 0);
 		if (rc != CELL_OK)
 		{
 			assert(0);
@@ -2967,7 +2967,7 @@ bool spursIsLibProfLoaded()
 
 void spursTraceStatusUpdate(vm::ptr<CellSpurs> spurs)
 {
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	if (spurs->m.xCC != 0)
 	{

rpcs3/Emu/SysCalls/Modules/cellSpursSpu.cpp

@@ -3,6 +3,7 @@
 #include "Emu/System.h"
 #include "Emu/Cell/SPUThread.h"
 #include "Emu/SysCalls/Modules.h"
+#include "Emu/SysCalls/lv2/sleep_queue.h"
 #include "Emu/SysCalls/lv2/sys_lwmutex.h"
 #include "Emu/SysCalls/lv2/sys_lwcond.h"
 #include "Emu/SysCalls/lv2/sys_spu.h"

rpcs3/Emu/SysCalls/Modules/cellSync.cpp

@@ -1070,7 +1070,7 @@ s32 syncLFQueueGetPushPointer(vm::ptr<CellSyncLFQueue> queue, s32& pointer, u32
 			}
 		}
 
-		if (s32 res = sys_event_queue_receive(queue->m_eq_id, vm::ptr<sys_event_data>::make(0), 0))
+		if (s32 res = sys_event_queue_receive(GetCurrentPPUThread(), queue->m_eq_id, vm::ptr<sys_event_t>::make(0), 0))
 		{
 			assert(!"sys_event_queue_receive() failed");
 		}
@@ -1421,7 +1421,7 @@ s32 syncLFQueueGetPopPointer(vm::ptr<CellSyncLFQueue> queue, s32& pointer, u32 i
 			}
 		}
 
-		if (s32 res = sys_event_queue_receive(queue->m_eq_id, vm::ptr<sys_event_data>::make(0), 0))
+		if (s32 res = sys_event_queue_receive(GetCurrentPPUThread(), queue->m_eq_id, vm::ptr<sys_event_t>::make(0), 0))
 		{
 			assert(!"sys_event_queue_receive() failed");
 		}

rpcs3/Emu/SysCalls/lv2/cellFs.cpp

@@ -525,8 +525,6 @@ s32 cellFsFGetBlockSize(u32 fd, vm::ptr<u64> sector_size, vm::ptr<u64> block_siz
 {
 	sys_fs.Warning("cellFsFGetBlockSize(fd=0x%x, sector_size=0x%x, block_size=0x%x)", fd, sector_size, block_size);
 
-	LV2_LOCK(0);
-
 	std::shared_ptr<vfsStream> file;
 	if (!sys_fs.CheckId(fd, file))
 		return CELL_ESRCH;

rpcs3/Emu/SysCalls/lv2/sys_event.cpp

@@ -14,66 +14,68 @@ SysCallBase sys_event("sys_event");
 
 u32 event_queue_create(u32 protocol, s32 type, u64 name_u64, u64 event_queue_key, s32 size)
 {
-	std::shared_ptr<EventQueue> eq(new EventQueue(protocol, type, name_u64, event_queue_key, size));
+	std::shared_ptr<event_queue_t> queue(new event_queue_t(protocol, type, name_u64, event_queue_key, size));
 
-	if (event_queue_key && !Emu.GetEventManager().RegisterKey(eq, event_queue_key))
+	Emu.GetEventManager().RegisterKey(queue, event_queue_key);
-	{
-		return 0;
-	}
 
-	const u32 id = sys_event.GetNewId(eq, TYPE_EVENT_QUEUE);
+	return sys_event.GetNewId(queue, TYPE_EVENT_QUEUE);
-	eq->sq.set_full_name(fmt::Format("EventQueue(%d)", id));
-	sys_event.Warning("*** event_queue created [%s] (protocol=0x%x, type=0x%x, key=0x%llx, size=0x%x): id = %d",
-		std::string((const char*)&name_u64, 8).c_str(), protocol, type, event_queue_key, size, id);
-	return id;
 }
 
 s32 sys_event_queue_create(vm::ptr<u32> equeue_id, vm::ptr<sys_event_queue_attr> attr, u64 event_queue_key, s32 size)
 {
-	sys_event.Warning("sys_event_queue_create(equeue_id_addr=0x%x, attr_addr=0x%x, event_queue_key=0x%llx, size=%d)",
+	sys_event.Warning("sys_event_queue_create(equeue_id=*0x%x, attr=*0x%x, event_queue_key=0x%llx, size=%d)", equeue_id, attr, event_queue_key, size);
-		equeue_id.addr(), attr.addr(), event_queue_key, size);
 
-	if(size <= 0 || size > 127)
+	if (size <= 0 || size > 127)
 	{
 		return CELL_EINVAL;
 	}
 
-	switch (attr->protocol.data())
+	const u32 protocol = attr->protocol;
+
+	switch (protocol)
 	{
-	case se32(SYS_SYNC_PRIORITY): break;
+	case SYS_SYNC_PRIORITY: break;
-	case se32(SYS_SYNC_RETRY): sys_event.Error("Invalid protocol (SYS_SYNC_RETRY)"); return CELL_EINVAL;
+	case SYS_SYNC_RETRY: sys_event.Error("Invalid protocol (SYS_SYNC_RETRY)"); return CELL_EINVAL;
-	case se32(SYS_SYNC_PRIORITY_INHERIT): sys_event.Error("Invalid protocol (SYS_SYNC_PRIORITY_INHERIT)"); return CELL_EINVAL;
+	case SYS_SYNC_PRIORITY_INHERIT: sys_event.Error("Invalid protocol (SYS_SYNC_PRIORITY_INHERIT)"); return CELL_EINVAL;
-	case se32(SYS_SYNC_FIFO): break;
+	case SYS_SYNC_FIFO: break;
-	default: sys_event.Error("Unknown protocol (0x%x)", attr->protocol); return CELL_EINVAL;
+	default: sys_event.Error("Unknown protocol (0x%x)", protocol); return CELL_EINVAL;
 	}
 
-	switch (attr->type.data())
+	const u32 type = attr->type;
+
+	switch (type)
 	{
-	case se32(SYS_PPU_QUEUE): break;
+	case SYS_PPU_QUEUE: break;
-	case se32(SYS_SPU_QUEUE): break;
+	case SYS_SPU_QUEUE: break;
-	default: sys_event.Error("Unknown event queue type (0x%x)", attr->type); return CELL_EINVAL;
+	default: sys_event.Error("Unknown event queue type (0x%x)", type); return CELL_EINVAL;
 	}
 
-	if (event_queue_key && Emu.GetEventManager().CheckKey(event_queue_key))
+	LV2_LOCK;
+
+	if (Emu.GetEventManager().CheckKey(event_queue_key))
 	{
 		return CELL_EEXIST;
 	}
 
-	if (u32 id = event_queue_create(attr->protocol, attr->type, attr->name_u64, event_queue_key, size))
+	std::shared_ptr<event_queue_t> queue(new event_queue_t(protocol, type, attr->name_u64, event_queue_key, size));
+
+	if (!Emu.GetEventManager().RegisterKey(queue, event_queue_key))
 	{
-		*equeue_id = id;
+		return CELL_EAGAIN;
-		return CELL_OK;
 	}
 
-	return CELL_EAGAIN;	
+	*equeue_id = sys_event.GetNewId(queue, TYPE_EVENT_QUEUE);
+	return CELL_OK;
 }
 
-s32 sys_event_queue_destroy(u32 equeue_id, int mode)
+s32 sys_event_queue_destroy(u32 equeue_id, s32 mode)
 {
-	sys_event.Todo("sys_event_queue_destroy(equeue_id=%d, mode=0x%x)", equeue_id, mode);
+	sys_event.Warning("sys_event_queue_destroy(equeue_id=%d, mode=%d)", equeue_id, mode);
 
-	std::shared_ptr<EventQueue> eq;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(equeue_id, eq))
+
+	std::shared_ptr<event_queue_t> queue;
+	if (!Emu.GetIdManager().GetIDData(equeue_id, queue))
 	{
 		return CELL_ESRCH;
 	}
@@ -83,203 +85,145 @@ s32 sys_event_queue_destroy(u32 equeue_id, int mode)
 		return CELL_EINVAL;
 	}
 
-	//u32 tid = GetCurrentPPUThread().GetId();
+	if (!mode && queue->waiters)
-	//eq->sq.m_mutex.lock();
+	{
-	//eq->owner.lock(tid);
+		return CELL_EBUSY;
-	// check if some threads are waiting for an event
+	}
-	//if (!mode && eq->sq.list.size())
+	else
-	//{
+	{
-	//	eq->owner.unlock(tid);
+		// set special value for waiters
-	//	eq->sq.m_mutex.unlock();
+		queue->waiters.exchange(-1);
-	//	return CELL_EBUSY;
+	}
-	//}
-	//eq->owner.unlock(tid, ~0);
-	//eq->sq.m_mutex.unlock();
-	//while (eq->sq.list.size())
-	//{
-	//	std::this_thread::sleep_for(std::chrono::milliseconds(1)); // hack
-	//	if (Emu.IsStopped())
-	//	{
-	//		sys_event.Warning("sys_event_queue_destroy(equeue=%d) aborted", equeue_id);
-	//		break;
-	//	}
-	//}
 
-	Emu.GetEventManager().UnregisterKey(eq->key);
+	Emu.GetEventManager().UnregisterKey(queue->key);
-	eq->ports.clear();
 	Emu.GetIdManager().RemoveID(equeue_id);
-
 	return CELL_OK;
 }
 
-s32 sys_event_queue_tryreceive(u32 equeue_id, vm::ptr<sys_event_data> event_array, s32 size, vm::ptr<u32> number)
+s32 sys_event_queue_tryreceive(u32 equeue_id, vm::ptr<sys_event_t> event_array, s32 size, vm::ptr<u32> number)
 {
-	sys_event.Todo("sys_event_queue_tryreceive(equeue_id=%d, event_array_addr=0x%x, size=%d, number_addr=0x%x)",
+	sys_event.Warning("sys_event_queue_tryreceive(equeue_id=%d, event_array=*0x%x, size=%d, number=*0x%x)", equeue_id, event_array, size, number);
-		equeue_id, event_array.addr(), size, number.addr());
 
-	std::shared_ptr<EventQueue> eq;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(equeue_id, eq))
+
+	std::shared_ptr<event_queue_t> queue;
+	if (!Emu.GetIdManager().GetIDData(equeue_id, queue))
 	{
 		return CELL_ESRCH;
 	}
 
-	if (eq->type != SYS_PPU_QUEUE)
+	if (queue->type != SYS_PPU_QUEUE || size < 0) // ???
 	{
 		return CELL_EINVAL;
 	}
 
-	if (size == 0)
+	s32 count = 0;
+
+	while (count < size && queue->events.size())
 	{
-		*number = 0;
+		auto& event = queue->events.front();
-		return CELL_OK;
+		event_array[count++] = { be_t<u64>::make(event.source), be_t<u64>::make(event.data1), be_t<u64>::make(event.data2), be_t<u64>::make(event.data3) };
+
+		queue->events.pop_front();
 	}
 
-	//u32 tid = GetCurrentPPUThread().GetId();
+	*number = count;
-	//eq->sq.m_mutex.lock();
-	//eq->owner.lock(tid);
-	//if (eq->sq.list.size())
-	//{
-	//	*number = 0;
-	//	eq->owner.unlock(tid);
-	//	eq->sq.m_mutex.unlock();
-	//	return CELL_OK;
-	//}
-	*number = eq->events.pop_all(event_array.get_ptr(), size);
-	//eq->owner.unlock(tid);
-	//eq->sq.m_mutex.unlock();
 	return CELL_OK;
 }
 
-s32 sys_event_queue_receive(u32 equeue_id, vm::ptr<sys_event_data> dummy_event, u64 timeout)
+s32 sys_event_queue_receive(PPUThread& CPU, u32 equeue_id, vm::ptr<sys_event_t> dummy_event, u64 timeout)
 {
-	// dummy_event argument is ignored, data returned in registers
+	sys_event.Log("sys_event_queue_receive(equeue_id=%d, event=*0x%x, timeout=0x%llx)", equeue_id, dummy_event, timeout);
-	sys_event.Log("sys_event_queue_receive(equeue_id=%d, dummy_event_addr=0x%x, timeout=%lld)",
-		equeue_id, dummy_event.addr(), timeout);
 
 	const u64 start_time = get_system_time();
 
-	std::shared_ptr<EventQueue> eq;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(equeue_id, eq))
+
+	std::shared_ptr<event_queue_t> queue;
+	if (!Emu.GetIdManager().GetIDData(equeue_id, queue))
 	{
 		return CELL_ESRCH;
 	}
 
-	if (eq->type != SYS_PPU_QUEUE)
+	if (queue->type != SYS_PPU_QUEUE)
 	{
 		return CELL_EINVAL;
 	}
 
-	const u32 tid = GetCurrentPPUThread().GetId();
+	// protocol is ignored in current implementation
+	queue->waiters++;
 
-	eq->sq.push(tid, eq->protocol); // add thread to sleep queue
+	while (queue->events.empty())
-
-	while (true)
 	{
-		const u32 old_owner = eq->owner.compare_and_swap(0, tid);
+		if (queue->waiters < 0)
-		const s32 res = old_owner ? (old_owner == tid ? 1 : 2) : 0;
-
-		switch (res)
 		{
-		case 0:
+			queue->waiters--;
-		{
-			const u32 next = eq->events.count() ? eq->sq.signal(eq->protocol) : 0;
-			if (next != tid)
-			{
-				if (!eq->owner.compare_and_swap_test(tid, next))
-				{
-					assert(!"sys_event_queue_receive() failed (I)");
-				}
-				break;
-			}
-			// fallthrough
-		}
-		case 1:
-		{
-			sys_event_data event;
-			eq->events.pop(event);
-			if (!eq->owner.compare_and_swap_test(tid, 0))
-			{
-				assert(!"sys_event_queue_receive() failed (II)");
-			}
-			sys_event.Log(" *** event received: source=0x%llx, d1=0x%llx, d2=0x%llx, d3=0x%llx", 
-				(u64)event.source, (u64)event.data1, (u64)event.data2, (u64)event.data3);
-			/* passing event data in registers */
-			PPUThread& t = GetCurrentPPUThread();
-			t.GPR[4] = event.source;
-			t.GPR[5] = event.data1;
-			t.GPR[6] = event.data2;
-			t.GPR[7] = event.data3;
-			if (!eq->sq.invalidate(tid, eq->protocol) && !eq->sq.pop(tid, eq->protocol))
-			{
-				assert(!"sys_event_queue_receive() failed (receiving)");
-			}
-			return CELL_OK;
-		}
-		}
-
-		if (!~old_owner)
-		{
-			if (!eq->sq.invalidate(tid, eq->protocol))
-			{
-				assert(!"sys_event_queue_receive() failed (cancelling)");
-			}
 			return CELL_ECANCELED;
 		}
 
-		std::this_thread::sleep_for(std::chrono::milliseconds(1)); // hack
-
 		if (timeout && get_system_time() - start_time > timeout)
 		{
-			if (!eq->sq.invalidate(tid, eq->protocol))
+			queue->waiters--;
-			{
-				assert(!"sys_event_queue_receive() failed (timeout)");
-			}
 			return CELL_ETIMEDOUT;
 		}
 
 		if (Emu.IsStopped())
 		{
-			sys_event.Warning("sys_event_queue_receive(equeue=%d) aborted", equeue_id);
+			sys_event.Warning("sys_event_queue_receive(equeue_id=%d) aborted", equeue_id);
 			return CELL_OK;
 		}
+
+		queue->cv.wait_for(lv2_lock, std::chrono::milliseconds(1));
 	}
+
+	// event data is returned in registers, second arg is not used
+	auto& event = queue->events.front();
+	CPU.GPR[4] = event.source;
+	CPU.GPR[5] = event.data1;
+	CPU.GPR[6] = event.data2;
+	CPU.GPR[7] = event.data3;
+
+	queue->events.pop_front();
+	queue->waiters--;
+	return CELL_OK;
 }
 
 s32 sys_event_queue_drain(u32 equeue_id)
 {
 	sys_event.Log("sys_event_queue_drain(equeue_id=%d)", equeue_id);
 
-	std::shared_ptr<EventQueue> eq;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(equeue_id, eq))
+
+	std::shared_ptr<event_queue_t> queue;
+	if (!Emu.GetIdManager().GetIDData(equeue_id, queue))
 	{
 		return CELL_ESRCH;
 	}
 
-	eq->events.clear();
+	queue->events = {};
-
 	return CELL_OK;
 }
 
 u32 event_port_create(u64 name)
 {
-	std::shared_ptr<EventPort> eport(new EventPort());
+	std::shared_ptr<event_port_t> eport(new event_port_t());
-	u32 id = sys_event.GetNewId(eport, TYPE_EVENT_PORT);
+	const u32 id = sys_event.GetNewId(eport, TYPE_EVENT_PORT);
 	eport->name = name ? name : ((u64)process_getpid() << 32) | (u64)id;
-	sys_event.Warning("*** sys_event_port created: id = %d, name=0x%llx", id, eport->name);
 	return id;
 }
 
 s32 sys_event_port_create(vm::ptr<u32> eport_id, s32 port_type, u64 name)
 {
-	sys_event.Warning("sys_event_port_create(eport_id_addr=0x%x, port_type=0x%x, name=0x%llx)",
+	sys_event.Warning("sys_event_port_create(eport_id=*0x%x, port_type=%d, name=0x%llx)", eport_id, port_type, name);
-		eport_id.addr(), port_type, name);
 
 	if (port_type != SYS_EVENT_PORT_LOCAL)
 	{
-		sys_event.Error("sys_event_port_create: invalid port_type(0x%x)", port_type);
+		sys_event.Error("sys_event_port_create(): invalid port_type (%d)", port_type);
 		return CELL_EINVAL;
 	}
 
+	LV2_LOCK;
+
 	*eport_id = event_port_create(name);
 	return CELL_OK;
 }
@@ -288,24 +232,19 @@ s32 sys_event_port_destroy(u32 eport_id)
 {
 	sys_event.Warning("sys_event_port_destroy(eport_id=%d)", eport_id);
 
-	std::shared_ptr<EventPort> eport;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(eport_id, eport))
+
+	std::shared_ptr<event_port_t> port;
+	if (!Emu.GetIdManager().GetIDData(eport_id, port))
 	{
 		return CELL_ESRCH;
 	}
 
-	if (!eport->m_mutex.try_lock())
+	if (!port->queue.expired())
 	{
 		return CELL_EISCONN;
 	}
 
-	if (eport->eq)
-	{
-		eport->m_mutex.unlock();
-		return CELL_EISCONN;
-	}
-
-	eport->m_mutex.unlock();
 	Emu.GetIdManager().RemoveID(eport_id);
 	return CELL_OK;
 }
@@ -314,37 +253,23 @@ s32 sys_event_port_connect_local(u32 eport_id, u32 equeue_id)
 {
 	sys_event.Warning("sys_event_port_connect_local(eport_id=%d, equeue_id=%d)", eport_id, equeue_id);
 
-	std::shared_ptr<EventPort> eport;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(eport_id, eport))
+
+	std::shared_ptr<event_port_t> port;
+	std::shared_ptr<event_queue_t> queue;
+	if (!Emu.GetIdManager().GetIDData(eport_id, port) || !Emu.GetIdManager().GetIDData(equeue_id, queue))
 	{
 		return CELL_ESRCH;
 	}
 
-	if (!eport->m_mutex.try_lock())
+	// CELL_EINVAL is never returned (I have no idea if SYS_EVENT_PORT_LOCAL is the only possible type)
+
+	if (!port->queue.expired())
 	{
 		return CELL_EISCONN;
 	}
 
-	if (eport->eq)
+	port->queue = queue;
-	{
-		eport->m_mutex.unlock();
-		return CELL_EISCONN;
-	}
-
-	std::shared_ptr<EventQueue> equeue;
-	if (!Emu.GetIdManager().GetIDData(equeue_id, equeue))
-	{
-		sys_event.Error("sys_event_port_connect_local: event_queue(%d) not found!", equeue_id);
-		eport->m_mutex.unlock();
-		return CELL_ESRCH;
-	}
-	else
-	{
-		equeue->ports.add(eport);
-	}
-
-	eport->eq = equeue;
-	eport->m_mutex.unlock();
 	return CELL_OK;
 }
 
@@ -352,51 +277,58 @@ s32 sys_event_port_disconnect(u32 eport_id)
 {
 	sys_event.Warning("sys_event_port_disconnect(eport_id=%d)", eport_id);
 
-	std::shared_ptr<EventPort> eport;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(eport_id, eport))
+
+	std::shared_ptr<event_port_t> port;
+	if (!Emu.GetIdManager().GetIDData(eport_id, port))
 	{
 		return CELL_ESRCH;
 	}
 
-	if (!eport->eq)
+	std::shared_ptr<event_queue_t> queue = port->queue.lock();
+
+	if (!queue)
 	{
 		return CELL_ENOTCONN;
 	}
 
-	if (!eport->m_mutex.try_lock())
+	//for (auto& event : queue->events)
-	{
+	//{
-		return CELL_EBUSY;
+	//	if (event.source == port->name)
-	}
+	//	{
+	//		return CELL_EBUSY; // not sure about it
+	//	}
+	//}
 
-	eport->eq->ports.remove(eport);
+	port->queue.reset();
-	eport->eq = nullptr;
-	eport->m_mutex.unlock();
 	return CELL_OK;
 }
 
 s32 sys_event_port_send(u32 eport_id, u64 data1, u64 data2, u64 data3)
 {
-	sys_event.Log("sys_event_port_send(eport_id=%d, data1=0x%llx, data2=0x%llx, data3=0x%llx)",
+	sys_event.Log("sys_event_port_send(eport_id=%d, data1=0x%llx, data2=0x%llx, data3=0x%llx)", eport_id, data1, data2, data3);
-		eport_id, data1, data2, data3);
 
-	std::shared_ptr<EventPort> eport;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(eport_id, eport))
+
+	std::shared_ptr<event_port_t> port;
+	if (!Emu.GetIdManager().GetIDData(eport_id, port))
 	{
 		return CELL_ESRCH;
 	}
 
-	std::lock_guard<std::mutex> lock(eport->m_mutex);
+	std::shared_ptr<event_queue_t> queue = port->queue.lock();
 
-	std::shared_ptr<EventQueue> eq = eport->eq;
+	if (!queue)
-	if (!eq)
 	{
 		return CELL_ENOTCONN;
 	}
 
-	if (!eq->events.push(eport->name, data1, data2, data3))
+	if (queue->events.size() >= queue->size)
 	{
 		return CELL_EBUSY;
 	}
 
+	queue->events.emplace_back(port->name, data1, data2, data3);
+	queue->cv.notify_one();
 	return CELL_OK;
 }

rpcs3/Emu/SysCalls/lv2/sys_event.h

@@ -1,41 +1,43 @@
 #pragma once
 
-#define FIX_SPUQ(x) ((u64)x | 0x5350555100000000ULL)
+// Event Queue Type
-// arbitrary code to prevent "special" zero value in key argument
+enum : u32
-
-enum EventQueueType
 {
 	SYS_PPU_QUEUE = 1,
 	SYS_SPU_QUEUE = 2,
 };
 
-enum EventQueueDestroyMode
+// Event Queue Destroy Mode
+enum : s32
 {
-	// DEFAULT = 0,
 	SYS_EVENT_QUEUE_DESTROY_FORCE = 1,
 };
 
-enum EventPortType
+// Event Port Type
+enum : s32
 {
 	SYS_EVENT_PORT_LOCAL = 1,
 };
 
-enum EventSourceType
+// Event Source Type
+enum : u32
 {
 	SYS_SPU_THREAD_EVENT_USER = 1,
-	/* SYS_SPU_THREAD_EVENT_DMA = 2, */ // not supported
+	SYS_SPU_THREAD_EVENT_DMA  = 2, // not supported
 };
 
-enum EventSourceKey : u64
+// Event Source Key
+enum : u64
 {
 	SYS_SPU_THREAD_EVENT_USER_KEY = 0xFFFFFFFF53505501,
-	/* SYS_SPU_THREAD_EVENT_DMA_KEY = 0xFFFFFFFF53505502, */
+	SYS_SPU_THREAD_EVENT_DMA_KEY  = 0xFFFFFFFF53505502, // ???
 };
 
 struct sys_event_queue_attr
 {
 	be_t<u32> protocol; // SYS_SYNC_PRIORITY or SYS_SYNC_FIFO
 	be_t<s32> type; // SYS_PPU_QUEUE or SYS_SPU_QUEUE
+
 	union
 	{
 		char name[8];
@@ -43,7 +45,7 @@ struct sys_event_queue_attr
 	};
 };
 
-struct sys_event_data
+struct sys_event_t
 {
 	be_t<u64> source;
 	be_t<u64> data1;
@@ -51,169 +53,59 @@ struct sys_event_data
 	be_t<u64> data3;
 };
 
-struct EventQueue;
+struct event_t
-
-struct EventPort
 {
-	u64 name; // generated or user-specified code that is passed to sys_event_data struct
+	u64 source;
-	std::shared_ptr<EventQueue> eq; // event queue this port has been connected to
+	u64 data1;
-	std::mutex m_mutex; // may be locked until the event sending is finished
+	u64 data2;
+	u64 data3;
 
-	EventPort(u64 name = 0)
+	event_t(u64 source, u64 data1, u64 data2, u64 data3)
-		: eq(nullptr)
+		: source(source)
-		, name(name)
+		, data1(data1)
+		, data2(data2)
+		, data3(data3)
 	{
 	}
 };
 
-class EventRingBuffer
+struct event_queue_t
 {
-	std::vector<sys_event_data> data;
-	std::mutex m_mutex;
-	u32 buf_pos;
-	u32 buf_count;
-
-public:
-	const u32 size;
-
-	EventRingBuffer(u32 size)
-		: size(size)
-		, buf_pos(0)
-		, buf_count(0)
-	{
-		data.resize(size);
-	}
-
-	void clear()
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-		buf_count = 0;
-		buf_pos = 0;
-	}
-
-	bool push(u64 name, u64 d1, u64 d2, u64 d3)
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-		if (buf_count >= size) return false;
-
-		sys_event_data& ref = data[(buf_pos + buf_count++) % size];
-		ref.source = name;
-		ref.data1 = d1;
-		ref.data2 = d2;
-		ref.data3 = d3;
-
-		return true;
-	}
-
-	bool pop(sys_event_data& ref)
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-		if (!buf_count) return false;
-
-		sys_event_data& from = data[buf_pos];
-		buf_pos = (buf_pos + 1) % size;
-		buf_count--;
-		ref.source = from.source;
-		ref.data1 = from.data1;
-		ref.data2 = from.data2;
-		ref.data3 = from.data3;
-
-		return true;
-	}
-
-	u32 pop_all(sys_event_data* ptr, u32 max)
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-
-		u32 res = 0;
-		while (buf_count && max)
-		{
-			sys_event_data& from = data[buf_pos];
-			ptr->source = from.source;
-			ptr->data1 = from.data1;
-			ptr->data2 = from.data2;
-			ptr->data3 = from.data3;
-			buf_pos = (buf_pos + 1) % size;
-			buf_count--;
-			max--;
-			ptr++;
-			res++;
-		}
-		return res;
-	}
-
-	u32 count() const
-	{
-		return buf_count;
-	}
-};
-
-class EventPortList
-{
-	std::vector<std::shared_ptr<EventPort>> data;
-	std::mutex m_mutex;
-
-public:
-
-	void clear()
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-		for (u32 i = 0; i < data.size(); i++)
-		{
-			// TODO: force all ports to disconnect
-			//std::lock_guard<std::mutex> lock2(data[i]->m_mutex);
-			//data[i]->eq = nullptr;
-		}
-		data.clear();
-	}
-
-	void add(std::shared_ptr<EventPort>& port)
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-		data.push_back(port);
-	}
-
-	void remove(std::shared_ptr<EventPort>& port)
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-		for (u32 i = 0; i < data.size(); i++)
-		{
-			if (data[i].get() == port.get())
-			{
-				data.erase(data.begin() + i);
-				return;
-			}
-		}
-	}
-};
-
-struct EventQueue
-{
-	sleep_queue_t sq;
-	EventPortList ports;
-	EventRingBuffer events;
-	atomic_le_t<u32> owner;
-
-	const union
-	{
-		u64 name_u64;
-		char name[8];
-	};
 	const u32 protocol;
-	const int type;
+	const s32 type;
+	const u64 name;
 	const u64 key;
+	const s32 size;
 
-	EventQueue(u32 protocol, int type, u64 name, u64 key, int size)
+	std::deque<event_t> events;
-		: type(type)
+
-		, protocol(protocol)
+	std::condition_variable cv;
-		, name_u64(name)
+	std::atomic<s32> waiters;
+
+	event_queue_t(u32 protocol, s32 type, u64 name, u64 key, s32 size)
+		: protocol(protocol)
+		, type(type)
+		, name(name)
 		, key(key)
-		, events(size) // size: max event count this queue can hold
+		, size(size)
+		, waiters(0)
 	{
-		owner.write_relaxed(0);
 	}
 };
 
+struct event_port_t
+{
+	u64 name; // generated or user-specified code that is passed as event source
+	std::weak_ptr<event_queue_t> queue; // event queue this port is connected to
+
+	event_port_t(u64 name = 0)
+		: name(name)
+	{
+	}
+};
+
+class PPUThread;
+
 // Aux
 u32 event_port_create(u64 name);
 u32 event_queue_create(u32 protocol, s32 type, u64 name_u64, u64 event_queue_key, s32 size);
@@ -221,8 +113,8 @@ u32 event_queue_create(u32 protocol, s32 type, u64 name_u64, u64 event_queue_key
 // SysCalls
 s32 sys_event_queue_create(vm::ptr<u32> equeue_id, vm::ptr<sys_event_queue_attr> attr, u64 event_queue_key, s32 size);
 s32 sys_event_queue_destroy(u32 equeue_id, s32 mode);
-s32 sys_event_queue_receive(u32 equeue_id, vm::ptr<sys_event_data> dummy_event, u64 timeout);
+s32 sys_event_queue_receive(PPUThread& CPU, u32 equeue_id, vm::ptr<sys_event_t> dummy_event, u64 timeout);
-s32 sys_event_queue_tryreceive(u32 equeue_id, vm::ptr<sys_event_data> event_array, s32 size, vm::ptr<u32> number);
+s32 sys_event_queue_tryreceive(u32 equeue_id, vm::ptr<sys_event_t> event_array, s32 size, vm::ptr<u32> number);
 s32 sys_event_queue_drain(u32 event_queue_id);
 
 s32 sys_event_port_create(vm::ptr<u32> eport_id, s32 port_type, u64 name);

rpcs3/Emu/SysCalls/lv2/sys_interrupt.cpp

@@ -82,7 +82,7 @@ s32 sys_interrupt_thread_establish(vm::ptr<u32> ih, u32 intrtag, u64 intrthread,
 	PPUThread& ppu = static_cast<PPUThread&>(*it);
 
 	{
-		LV2_LOCK(0);
+		LV2_LOCK;
 
 		if (ppu.custom_task)
 		{

rpcs3/Emu/SysCalls/lv2/sys_lwmutex.cpp

@@ -55,8 +55,6 @@ s32 sys_lwmutex_destroy(PPUThread& CPU, vm::ptr<sys_lwmutex_t> lwmutex)
 {
 	sys_lwmutex.Warning("sys_lwmutex_destroy(lwmutex_addr=0x%x)", lwmutex.addr());
 
-	LV2_LOCK(0);
-
 	u32 sq_id = lwmutex->sleep_queue;
 	if (!Emu.GetIdManager().CheckID(sq_id)) return CELL_ESRCH;
 

rpcs3/Emu/SysCalls/lv2/sys_ppu_thread.cpp

@@ -231,7 +231,7 @@ void sys_ppu_thread_once(PPUThread& CPU, vm::ptr<atomic_t<u32>> once_ctrl, vm::p
 {
 	sys_ppu_thread.Warning("sys_ppu_thread_once(once_ctrl_addr=0x%x, init_addr=0x%x)", once_ctrl.addr(), init.addr());
 
-	LV2_LOCK(0);
+	LV2_LOCK;
 
 	if (once_ctrl->compare_and_swap_test(be_t<u32>::make(SYS_PPU_THREAD_ONCE_INIT), be_t<u32>::make(SYS_PPU_THREAD_DONE_INIT)))
 	{

rpcs3/Emu/SysCalls/lv2/sys_spu.cpp

@@ -9,6 +9,7 @@
 #include "Emu/FS/vfsFile.h"
 #include "Loader/ELF32.h"
 #include "Crypto/unself.h"
+#include "sys_event.h"
 #include "sys_spu.h"
 
 SysCallBase sys_spu("sys_spu");
@@ -708,140 +709,131 @@ s32 sys_spu_thread_throw_event(u8 spup, u24 data0, u32 data1);
 s32 sys_spu_thread_tryreceive_event(u32 spuq_num, mem32_t d1, mem32_t d2, mem32_t d3);
 */
 
-s32 sys_spu_thread_connect_event(u32 id, u32 eq_id, u32 et, u8 spup)
+s32 sys_spu_thread_connect_event(u32 id, u32 eq, u32 et, u8 spup)
 {
-	sys_spu.Warning("sys_spu_thread_connect_event(id=%d, eq_id=%d, event_type=0x%x, spup=%d)", id, eq_id, et, spup);
+	sys_spu.Warning("sys_spu_thread_connect_event(id=%d, eq=%d, et=%d, spup=%d)", id, eq, et, spup);
 
-	std::shared_ptr<CPUThread> thr = Emu.GetCPU().GetThread(id);
+	LV2_LOCK;
 
-	if(!thr || thr->GetType() != CPU_THREAD_SPU)
+	std::shared_ptr<CPUThread> t = Emu.GetCPU().GetThread(id);
+
+	std::shared_ptr<event_queue_t> queue;
+
+	if (!t || t->GetType() != CPU_THREAD_SPU || !Emu.GetIdManager().GetIDData(eq, queue))
 	{
 		return CELL_ESRCH;
 	}
 
-	std::shared_ptr<EventQueue> eq;
+	auto& spu = static_cast<SPUThread&>(*t);
-	if (!Emu.GetIdManager().GetIDData(eq_id, eq))
-	{
-		return CELL_ESRCH;
-	}
 
-	if (spup > 63)
+	if (et != SYS_SPU_THREAD_EVENT_USER || spup > 63 || queue->type != SYS_PPU_QUEUE)
 	{
-		sys_spu.Error("sys_spu_thread_connect_event: invalid spup (%d)", spup);
+		sys_spu.Error("sys_spu_thread_connect_event(): invalid arguments (et=%d, spup=%d, queue->type=%d)", et, spup, queue->type);
 		return CELL_EINVAL;
 	}
 
-	if (et != SYS_SPU_THREAD_EVENT_USER)
+	auto& port = spu.spup[spup];
-	{
-		sys_spu.Error("sys_spu_thread_connect_event: unsupported event type (0x%x)", et);
-		return CELL_EINVAL;
-	}
 
-	// TODO: check if can receive these events
+	if (!port.expired())
-
-	SPUThread& spu = *(SPUThread*)thr.get();
-
-	std::shared_ptr<EventPort> port; //= spu.SPUPs[spup];
-
-	std::lock_guard<std::mutex> lock(port->m_mutex);
-
-	if (port->eq)
 	{
 		return CELL_EISCONN;
 	}
 
-	eq->ports.add(port);
+	port = queue;
-	port->eq = eq;
-
 	return CELL_OK;
 }
 
 s32 sys_spu_thread_disconnect_event(u32 id, u32 et, u8 spup)
 {
-	sys_spu.Warning("sys_spu_thread_disconnect_event(id=%d, event_type=0x%x, spup=%d)", id, et, spup);
+	sys_spu.Warning("sys_spu_thread_disconnect_event(id=%d, event_type=%d, spup=%d)", id, et, spup);
 
-	std::shared_ptr<CPUThread> thr = Emu.GetCPU().GetThread(id);
+	LV2_LOCK;
 
-	if(!thr || thr->GetType() != CPU_THREAD_SPU)
+	std::shared_ptr<CPUThread> t = Emu.GetCPU().GetThread(id);
+
+	if (!t || t->GetType() != CPU_THREAD_SPU)
 	{
 		return CELL_ESRCH;
 	}
 
-	if (spup > 63)
+	auto& spu = static_cast<SPUThread&>(*t);
+
+	if (et != SYS_SPU_THREAD_EVENT_USER || spup > 63)
 	{
-		sys_spu.Error("sys_spu_thread_connect_event: invalid spup (%d)", spup);
+		sys_spu.Error("sys_spu_thread_disconnect_event(): invalid arguments (et=%d, spup=%d)", et, spup);
 		return CELL_EINVAL;
 	}
 
-	if (et != SYS_SPU_THREAD_EVENT_USER)
+	auto& port = spu.spup[spup];
-	{
-		sys_spu.Error("sys_spu_thread_connect_event: unsupported event type (0x%x)", et);
-		return CELL_EINVAL;
-	}
 
-	SPUThread& spu = *(SPUThread*)thr.get();
+	if (port.expired())
-
-	std::shared_ptr<EventPort> port;// = spu.SPUPs[spup];
-
-	std::lock_guard<std::mutex> lock(port->m_mutex);
-
-	if (!port->eq)
 	{
 		return CELL_ENOTCONN;
 	}
 
-	port->eq->ports.remove(port);
+	port.reset();
-	port->eq = nullptr;
-
 	return CELL_OK;
 }
 
-s32 sys_spu_thread_bind_queue(u32 id, u32 eq_id, u32 spuq_num)
+s32 sys_spu_thread_bind_queue(u32 id, u32 spuq, u32 spuq_num)
 {
-	sys_spu.Warning("sys_spu_thread_bind_queue(id=%d, equeue_id=%d, spuq_num=0x%x)", id, eq_id, spuq_num);
+	sys_spu.Warning("sys_spu_thread_bind_queue(id=%d, spuq=%d, spuq_num=0x%x)", id, spuq, spuq_num);
 
-	std::shared_ptr<EventQueue> eq;
+	LV2_LOCK;
-	if (!Emu.GetIdManager().GetIDData(eq_id, eq))
+
+	std::shared_ptr<CPUThread> t = Emu.GetCPU().GetThread(id);
+
+	std::shared_ptr<event_queue_t> queue;
+
+	if (!t || t->GetType() != CPU_THREAD_SPU || !Emu.GetIdManager().GetIDData(spuq, queue))
 	{
 		return CELL_ESRCH;
 	}
 
-	if (eq->type != SYS_SPU_QUEUE)
+	auto& spu = static_cast<SPUThread&>(*t);
+
+	if (queue->type != SYS_SPU_QUEUE)
 	{
 		return CELL_EINVAL;
 	}
 
-	std::shared_ptr<CPUThread> thr = Emu.GetCPU().GetThread(id);
+	if (spu.spuq.size() >= 32)
-
-	if(!thr || thr->GetType() != CPU_THREAD_SPU)
 	{
-		return CELL_ESRCH;
+		return CELL_EAGAIN;
 	}
 
-	//if (!(*(SPUThread*)thr.get()).SPUQs.RegisterKey(eq, FIX_SPUQ(spuq_num)))
+	auto found = spu.spuq.find(spuq_num);
-	//{
+	if (found != spu.spuq.end())
-	//	return CELL_EBUSY;
+	{
-	//}
+		return CELL_EBUSY;
+	}
 
+	spu.spuq[spuq_num] = queue;
 	return CELL_OK;
 }
 
 s32 sys_spu_thread_unbind_queue(u32 id, u32 spuq_num)
 {
-	sys_spu.Warning("sys_spu_thread_unbind_queue(id=0x%x, spuq_num=0x%x)", id, spuq_num);
+	sys_spu.Warning("sys_spu_thread_unbind_queue(id=%d, spuq_num=0x%x)", id, spuq_num);
 
-	std::shared_ptr<CPUThread> thr = Emu.GetCPU().GetThread(id);
+	LV2_LOCK;
 
-	if(!thr || thr->GetType() != CPU_THREAD_SPU)
+	std::shared_ptr<CPUThread> t = Emu.GetCPU().GetThread(id);
+
+	if (!t || t->GetType() != CPU_THREAD_SPU)
 	{
 		return CELL_ESRCH;
 	}
 
-	//if (!(*(SPUThread*)thr.get()).SPUQs.UnregisterKey(FIX_SPUQ(spuq_num)))
+	auto& spu = static_cast<SPUThread&>(*t);
-	//{
-	//	return CELL_ESRCH; // may be CELL_EINVAL
-	//}
 
+	auto found = spu.spuq.find(spuq_num);
+	if (found == spu.spuq.end())
+	{
+		return CELL_ESRCH;
+	}
+
+	spu.spuq.erase(found);
 	return CELL_OK;
 }
 
@@ -850,7 +842,7 @@ s32 sys_spu_thread_group_connect_event_all_threads(u32 id, u32 eq_id, u64 req, v
 	sys_spu.Warning("sys_spu_thread_group_connect_event_all_threads(id=%d, eq_id=%d, req=0x%llx, spup_addr=0x%x)",
 		id, eq_id, req, spup.addr());
 
-	std::shared_ptr<EventQueue> eq;
+	std::shared_ptr<event_queue_t> eq;
 	if (!Emu.GetIdManager().GetIDData(eq_id, eq))
 	{
 		return CELL_ESRCH;

rpcs3/Emu/SysCalls/lv2/sys_timer.cpp

@@ -79,7 +79,7 @@ s32 sys_timer_connect_event_queue(u32 timer_id, u32 queue_id, u64 name, u64 data
 		timer_id, queue_id, name, data1, data2);
 
 	std::shared_ptr<timer> timer_data = nullptr;
-	std::shared_ptr<EventQueue> equeue = nullptr;
+	std::shared_ptr<event_queue_t> equeue = nullptr;
 	if(!sys_timer.CheckId(timer_id, timer_data)) return CELL_ESRCH;
 	if(!sys_timer.CheckId(queue_id, equeue)) return CELL_ESRCH;
 

rpcs3/Emu/System.cpp

@@ -25,6 +25,7 @@
 #include "Emu/RSX/GSManager.h"
 #include "Emu/Audio/AudioManager.h"
 #include "Emu/FS/VFS.h"
+#include "Emu/Event.h"
 #include "Emu/SysCalls/SyncPrimitivesManager.h"
 
 #include "Loader/PSF.h"

rpcs3/Emu/System.h

@@ -79,7 +79,7 @@ class Emulator
 	std::vector<u64> m_break_points;
 	std::vector<u64> m_marked_points;
 
-	std::recursive_mutex m_core_mutex;
+	std::mutex m_core_mutex;
 
 	CPUThreadManager* m_thread_manager;
 	PadManager* m_pad_manager;
@@ -137,7 +137,7 @@ public:
 		m_emu_path = path;
 	}
 
-	std::recursive_mutex& GetCoreMutex()   { return m_core_mutex; }
+	std::mutex&       GetCoreMutex()       { return m_core_mutex; }
 
 	CPUThreadManager& GetCPU()             { return *m_thread_manager; }
 	PadManager&       GetPadManager()      { return *m_pad_manager; }
@@ -199,7 +199,7 @@ public:
 	__forceinline bool IsReady()   const { return m_status == Ready; }
 };
 
-#define LV2_LOCK(x) std::lock_guard<std::recursive_mutex> core_lock##x(Emu.GetCoreMutex())
+#define LV2_LOCK std::unique_lock<std::mutex> lv2_lock(Emu.GetCoreMutex())
 
 extern Emulator Emu;
 

rpcs3/stdafx.h

@@ -27,6 +27,7 @@
 #include <condition_variable>
 #include <memory>
 #include <vector>
+#include <queue>
 #include <set>
 #include <array>
 #include <string>