Cleanup after #5310 (SPU thread groups)

Move lambda into a cpu_stop()
Use running thread counter to synchronize with sys_spu_thread_group_join()
Use SPU_STATUS_STOPPED_BY_STOP exclusively for sys_spu_thread_exit() as before
Remove unnecessary waiting in sys_spu_thread_group_exit()
Rollback some minor unnecessary changes
Use shared_mutex in SPU TG

rpcs3/Emu/Cell/SPUThread.cpp

@@ -476,25 +476,28 @@ void spu_thread::cpu_init()
 	gpr[1]._u32[3] = 0x3FFF0; // initial stack frame pointer
 }
 
+void spu_thread::cpu_stop()
+{
+	if (!group && offset >= RAW_SPU_BASE_ADDR)
+	{
+		// Save next PC and current SPU Interrupt Status
+		npc = pc | (interrupts_enabled);
+	}
+	else if (group && is_stopped())
+	{
+		if (verify(HERE, group->running--) == 1)
+		{
+			// Notify on last thread stopped
+			group->mutex.lock_unlock();
+			group->cond.notify_all();
+		}
+	}
+}
+
 extern thread_local std::string(*g_tls_log_prefix)();
 
 void spu_thread::cpu_task()
 {
-	auto end_cpu_task = [&]()
-	{
-		// save next PC and current SPU Interrupt status
-		if (!group && offset >= RAW_SPU_BASE_ADDR)
-		{
-			npc = pc | (interrupts_enabled);
-		}
-
-		if (state & cpu_flag::stop)
-		{
-			status |= SPU_STATUS_STOPPED_BY_STOP;
-			if (group) group->cv.notify_all();
-		}
-	};
-
 	// Get next PC and SPU Interrupt status
 	pc = npc.exchange(0);
 
@@ -525,8 +528,7 @@ void spu_thread::cpu_task()
 
 		// Print some stats
 		LOG_NOTICE(SPU, "Stats: Block Weight: %u (Retreats: %u);", block_counter, block_failure);
-
+		cpu_stop();
-		end_cpu_task();
 		return;
 	}
 
@@ -608,7 +610,7 @@ void spu_thread::cpu_task()
 		}
 	}
 
-	end_cpu_task();
+	cpu_stop();
 }
 
 void spu_thread::cpu_mem()
@@ -2067,6 +2069,7 @@ bool spu_thread::stop_and_signal(u32 code)
 		status.atomic_op([code](u32& status)
 		{
 			status = (status & 0xffff) | (code << 16);
+			status |= SPU_STATUS_STOPPED_BY_STOP;
 			status &= ~SPU_STATUS_RUNNING;
 		});
 
@@ -2358,26 +2361,6 @@ bool spu_thread::stop_and_signal(u32 code)
 			}
 		}
 
-		while (!Emu.IsStopped())
-		{
-			bool stopped = true;
-			for (auto& thread : group->threads)
-			{
-				if (thread && thread.get() != this)
-				{
-					if ((thread->status & SPU_STATUS_STOPPED_BY_STOP) == 0)
-					{
-						stopped = false;
-						break;
-					}
-				}
-			}
-
-			if (stopped) break;
-
-			group->cv.wait(group->mutex, 1000);
-		}
-
 		group->run_state = SPU_THREAD_GROUP_STATUS_INITIALIZED;
 		group->exit_status = value;
 		group->join_state |= SPU_TGJSF_GROUP_EXIT;
@@ -2396,9 +2379,7 @@ bool spu_thread::stop_and_signal(u32 code)
 		}
 
 		LOG_TRACE(SPU, "sys_spu_thread_exit(status=0x%x)", ch_out_mbox.get_value());
-
+		status |= SPU_STATUS_STOPPED_BY_STOP;
-		std::lock_guard lock(group->mutex);
-
 		state += cpu_flag::stop;
 		return true;
 	}

rpcs3/Emu/Cell/SPUThread.h

@@ -508,6 +508,7 @@ public:
 	virtual void cpu_unmem() override;
 	virtual ~spu_thread() override;
 	void cpu_init();
+	void cpu_stop();
 
 	static const u32 id_base = 0x02000000; // TODO (used to determine thread type)
 	static const u32 id_step = 1;

rpcs3/Emu/Cell/lv2/sys_spu.cpp

@@ -329,6 +329,12 @@ error_code sys_spu_thread_group_destroy(u32 id)
 
 	const auto group = idm::withdraw<lv2_spu_group>(id, [](lv2_spu_group& group) -> CellError
 	{
+		if (group.running)
+		{
+			// Cannot destroy while threads are running
+			return CELL_EBUSY;
+		}
+
 		const auto _old = group.run_state.compare_and_swap(SPU_THREAD_GROUP_STATUS_INITIALIZED, SPU_THREAD_GROUP_STATUS_NOT_INITIALIZED);
 
 		if (_old > SPU_THREAD_GROUP_STATUS_INITIALIZED)
@@ -365,10 +371,16 @@ error_code sys_spu_thread_group_start(ppu_thread& ppu, u32 id)
 {
 	vm::temporary_unlock(ppu);
 
-	sys_spu.warning("sys_spu_thread_group_start(id=0x%x)", id);
+	sys_spu.trace("sys_spu_thread_group_start(id=0x%x)", id);
 
 	const auto group = idm::get<lv2_spu_group>(id, [](lv2_spu_group& group)
 	{
+		if (group.running)
+		{
+			// Can't start while threads are (still) running
+			return false;
+		}
+
 		// SPU_THREAD_GROUP_STATUS_READY state is not used
 		return group.run_state.compare_and_swap_test(SPU_THREAD_GROUP_STATUS_INITIALIZED, SPU_THREAD_GROUP_STATUS_RUNNING);
 	});
@@ -386,6 +398,7 @@ error_code sys_spu_thread_group_start(ppu_thread& ppu, u32 id)
 	std::lock_guard lock(group->mutex);
 
 	group->join_state = 0;
+	group->running = +group->init;
 
 	for (auto& thread : group->threads)
 	{
@@ -547,9 +560,9 @@ error_code sys_spu_thread_group_yield(u32 id)
 	return CELL_OK;
 }
 
-error_code sys_spu_thread_group_terminate(ppu_thread& ppu, u32 id, s32 value)
+error_code sys_spu_thread_group_terminate(u32 id, s32 value)
 {
-	sys_spu.warning("sys_spu_thread_group_terminate(id=0x%x, value=0x%x)", id, value);
+	sys_spu.trace("sys_spu_thread_group_terminate(id=0x%x, value=0x%x)", id, value);
 
 	// The id can be either SPU Thread Group or SPU Thread
 	const auto thread = idm::get<named_thread<spu_thread>>(id);
@@ -580,7 +593,7 @@ error_code sys_spu_thread_group_terminate(ppu_thread& ppu, u32 id, s32 value)
 		}
 	}
 
-	std::lock_guard lock(group->mutex);
+	std::unique_lock lock(group->mutex);
 
 	if (group->run_state <= SPU_THREAD_GROUP_STATUS_INITIALIZED ||
 		group->run_state == SPU_THREAD_GROUP_STATUS_WAITING ||
@@ -594,38 +607,27 @@ error_code sys_spu_thread_group_terminate(ppu_thread& ppu, u32 id, s32 value)
 		if (thread)
 		{
 			thread->state += cpu_flag::stop;
+		}
+	}
+
+	for (auto& thread : group->threads)
+	{
+		if (thread && group->running)
+		{
 			thread_ctrl::notify(*thread);
 		}
 	}
 
-	while (!ppu.is_stopped())
-	{
-		bool stopped = true;
-
-		for (auto& t : group->threads)
-		{
-			if (t)
-			{
-				if ((t->status & SPU_STATUS_STOPPED_BY_STOP) == 0)
-				{
-					stopped = false;
-					break;
-				}
-			}
-		}
-
-		if (stopped) break;
-
-		// TODO
-		group->cv.wait(group->mutex, 1000);
-	}
-
-	if (ppu.is_stopped()) return 0;
-
 	group->run_state = SPU_THREAD_GROUP_STATUS_INITIALIZED;
 	group->exit_status = value;
 	group->join_state |= SPU_TGJSF_TERMINATED;
 
+	// Wait until the threads are actually stopped
+	while (group->running)
+	{
+		group->cond.wait(lock);
+	}
+
 	return CELL_OK;
 }
 
@@ -633,7 +635,7 @@ error_code sys_spu_thread_group_join(ppu_thread& ppu, u32 id, vm::ptr<u32> cause
 {
 	vm::temporary_unlock(ppu);
 
-	sys_spu.warning("sys_spu_thread_group_join(id=0x%x, cause=*0x%x, status=*0x%x)", id, cause, status);
+	sys_spu.trace("sys_spu_thread_group_join(id=0x%x, cause=*0x%x, status=*0x%x)", id, cause, status);
 
 	const auto group = idm::get<lv2_spu_group>(id);
 
@@ -646,7 +648,7 @@ error_code sys_spu_thread_group_join(ppu_thread& ppu, u32 id, vm::ptr<u32> cause
 	s32 exit_value = 0;
 
 	{
-		std::lock_guard lock(group->mutex);
+		std::unique_lock lock(group->mutex);
 
 		if (group->run_state < SPU_THREAD_GROUP_STATUS_INITIALIZED)
 		{
@@ -661,26 +663,14 @@ error_code sys_spu_thread_group_join(ppu_thread& ppu, u32 id, vm::ptr<u32> cause
 
 		lv2_obj::sleep(ppu);
 
-		while (!ppu.is_stopped())
+		while (group->running)
 		{
-			bool stopped = true;
+			if (ppu.is_stopped())
-
-			for (auto& t : group->threads)
 			{
-				if (t)
+				return 0;
-				{
-					if ((t->status & SPU_STATUS_STOPPED_BY_STOP) == 0)
-					{
-						stopped = false;
-						break;
-					}
-				}
 			}
 
-			if (stopped) break;
+			group->cond.wait(lock);
-
-			// TODO
-			group->cv.wait(group->mutex, 1000);
 		}
 
 		join_state = group->join_state;
@@ -689,7 +679,10 @@ error_code sys_spu_thread_group_join(ppu_thread& ppu, u32 id, vm::ptr<u32> cause
 		group->run_state = SPU_THREAD_GROUP_STATUS_INITIALIZED; // hack
 	}
 
-	if (ppu.test_stopped()) return 0;
+	if (ppu.test_stopped())
+	{
+		return 0;
+	}
 
 	switch (join_state & ~SPU_TGJSF_IS_JOINING)
 	{

rpcs3/Emu/Cell/lv2/sys_spu.h

@@ -227,13 +227,15 @@ struct lv2_spu_group
 	const s32 type; // SPU Thread Group Type
 	const u32 ct; // Memory Container Id
 
-	semaphore<> mutex;
+	shared_mutex mutex;
+
 	atomic_t<u32> init; // Initialization Counter
 	atomic_t<s32> prio; // SPU Thread Group Priority
 	atomic_t<u32> run_state; // SPU Thread Group State
 	atomic_t<s32> exit_status; // SPU Thread Group Exit Status
 	atomic_t<u32> join_state; // flags used to detect exit cause
-	cond_variable cv; // used to signal waiting PPU thread
+	atomic_t<u32> running; // Number of running threads
+	cond_variable cond; // used to signal waiting PPU thread
 
 	std::array<std::shared_ptr<named_thread<spu_thread>>, 256> threads; // SPU Threads
 	std::array<std::pair<sys_spu_image, std::vector<sys_spu_segment>>, 256> imgs; // SPU Images
@@ -254,6 +256,7 @@ struct lv2_spu_group
 		, run_state(SPU_THREAD_GROUP_STATUS_NOT_INITIALIZED)
 		, exit_status(0)
 		, join_state(0)
+		, running(0)
 	{
 	}
 
@@ -300,7 +303,7 @@ error_code sys_spu_thread_group_start(ppu_thread&, u32 id);
 error_code sys_spu_thread_group_suspend(u32 id);
 error_code sys_spu_thread_group_resume(u32 id);
 error_code sys_spu_thread_group_yield(u32 id);
-error_code sys_spu_thread_group_terminate(ppu_thread&, u32 id, s32 value);
+error_code sys_spu_thread_group_terminate(u32 id, s32 value);
 error_code sys_spu_thread_group_join(ppu_thread&, u32 id, vm::ptr<u32> cause, vm::ptr<u32> status);
 error_code sys_spu_thread_group_set_priority(u32 id, s32 priority);
 error_code sys_spu_thread_group_get_priority(u32 id, vm::ptr<s32> priority);