SPU: Use waitable atomics for SPU channels interface

Utilities/Thread.cpp

@@ -1892,6 +1892,7 @@ void thread_base::initialize(void (*error_cb)(), bool(*wait_cb)(const void*))
 
 void thread_base::notify_abort() noexcept
 {
+	m_signal.try_inc();
 	atomic_storage_futex::raw_notify(+m_state_notifier);
 }
 

Utilities/Thread.h

@@ -210,6 +210,12 @@ public:
 		static_cast<thread_base&>(thread).notify();
 	}
 
+	template <typename T>
+	static void raw_notify(named_thread<T>& thread)
+	{
+		static_cast<thread_base&>(thread).notify_abort();
+	}
+
 	// Read current state
 	static inline thread_state state()
 	{

rpcs3/Emu/Cell/RawSPUThread.cpp

@@ -22,7 +22,7 @@ inline void try_start(spu_thread& spu)
 	}).second)
 	{
 		spu.state -= cpu_flag::stop;
-		thread_ctrl::notify(static_cast<named_thread<spu_thread>&>(spu));
+		thread_ctrl::raw_notify(static_cast<named_thread<spu_thread>&>(spu));
 	}
 };
 
@@ -123,7 +123,7 @@ bool spu_thread::read_reg(const u32 addr, u32& value)
 
 	case SPU_Out_MBox_offs:
 	{
-		value = ch_out_mbox.pop(*this);
+		value = ch_out_mbox.pop();
 		return true;
 	}
 

rpcs3/Emu/Cell/SPUThread.cpp

@@ -1247,11 +1247,11 @@ void spu_thread::push_snr(u32 number, u32 value)
 	// Check corresponding SNR register settings
 	if ((snr_config >> number) & 1)
 	{
-		channel->push_or(*this, value);
+		channel->push_or(value);
 	}
 	else
 	{
-		channel->push(*this, value);
+		channel->push(value);
 	}
 }
 
@@ -2448,19 +2448,8 @@ s64 spu_thread::get_ch_value(u32 ch)
 			busy_wait();
 		}
 
-		u32 out = 0;
-
-		while (!channel.try_pop(out))
-		{
-			if (is_stopped())
-			{
-				return -1;
-			}
-
-			thread_ctrl::wait();
-		}
-
-		check_state();
+		const s64 out = channel.pop_wait(*this);
+		static_cast<void>(test_stopped());
 		return out;
 	};
 
@@ -2508,9 +2497,8 @@ s64 spu_thread::get_ch_value(u32 ch)
 
 	case MFC_RdTagStat:
 	{
-		if (ch_tag_stat.get_count())
+		if (u32 out; ch_tag_stat.try_read(out))
 		{
-			u32 out = ch_tag_stat.get_value();
 			ch_tag_stat.set_value(0, false);
 			return out;
 		}
@@ -2536,9 +2524,8 @@ s64 spu_thread::get_ch_value(u32 ch)
 
 	case MFC_RdAtomicStat:
 	{
-		if (ch_atomic_stat.get_count())
+		if (u32 out; ch_atomic_stat.try_read(out))
 		{
-			u32 out = ch_atomic_stat.get_value();
 			ch_atomic_stat.set_value(0, false);
 			return out;
 		}
@@ -2549,9 +2536,8 @@ s64 spu_thread::get_ch_value(u32 ch)
 
 	case MFC_RdListStallStat:
 	{
-		if (ch_stall_stat.get_count())
+		if (u32 out; ch_stall_stat.try_read(out))
 		{
-			u32 out = ch_stall_stat.get_value();
 			ch_stall_stat.set_value(0, false);
 			return out;
 		}
@@ -2658,16 +2644,14 @@ bool spu_thread::set_ch_value(u32 ch, u32 value)
 	{
 		if (get_type() >= spu_type::raw)
 		{
-			while (!ch_out_intr_mbox.try_push(value))
+			if (ch_out_intr_mbox.get_count())
 			{
 				state += cpu_flag::wait;
-
-				if (is_stopped())
-				{
-					return false;
 			}
 
-				thread_ctrl::wait();
+			if (!ch_out_intr_mbox.push_wait(*this, value))
+			{
+				return false;
 			}
 
 			int_ctrl[2].set(SPU_INT2_STAT_MAILBOX_INT);
@@ -2798,16 +2782,14 @@ bool spu_thread::set_ch_value(u32 ch, u32 value)
 
 	case SPU_WrOutMbox:
 	{
-		while (!ch_out_mbox.try_push(value))
+		if (ch_out_mbox.get_count())
 		{
 			state += cpu_flag::wait;
-
-			if (is_stopped())
-			{
-				return false;
 		}
 
-			thread_ctrl::wait();
+		if (!ch_out_mbox.push_wait(*this, value))
+		{
+			return false;
 		}
 
 		check_state();
@@ -3196,7 +3178,7 @@ bool spu_thread::stop_and_signal(u32 code)
 
 				if (thread.get() != this)
 				{
-					thread_ctrl::notify(*thread);
+					thread_ctrl::raw_notify(*thread);
 				}
 			}
 		}
@@ -3325,7 +3307,7 @@ bool spu_thread::stop_and_signal(u32 code)
 				if (thread && thread.get() != this)
 				{
 					thread->state += cpu_flag::stop;
-					thread_ctrl::notify(*thread);
+					thread_ctrl::raw_notify(*thread);
 				}
 			}
 

rpcs3/Emu/Cell/SPUThread.h

@@ -175,23 +175,20 @@ public:
 	// Returns true on success
 	bool try_push(u32 value)
 	{
-		const u64 old = data.fetch_op([value](u64& data)
+		return data.fetch_op([value](u64& data)
 		{
-			if (data & bit_count) [[unlikely]]
-			{
-				data |= bit_wait;
-			}
-			else
+			if (!(data & bit_count)) [[likely]]
 			{
 				data = bit_count | value;
+				return true;
 			}
-		});
 
-		return !(old & bit_count);
+			return false;
+		}).second;
 	}
 
 	// Push performing bitwise OR with previous value, may require notification
-	void push_or(cpu_thread& spu, u32 value)
+	void push_or(u32 value)
 	{
 		const u64 old = data.fetch_op([value](u64& data)
 		{
@@ -201,7 +198,7 @@ public:
 
 		if (old & bit_wait)
 		{
-			spu.notify();
+			data.notify_one();
 		}
 	}
 
@@ -211,31 +208,28 @@ public:
 	}
 
 	// Push unconditionally (overwriting previous value), may require notification
-	void push(cpu_thread& spu, u32 value)
+	void push(u32 value)
 	{
 		if (data.exchange(bit_count | value) & bit_wait)
 		{
-			spu.notify();
+			data.notify_one();
 		}
 	}
 
 	// Returns true on success
 	bool try_pop(u32& out)
 	{
-		const u64 old = data.fetch_op([&](u64& data)
+		return data.fetch_op([&](u64& data)
 		{
 			if (data & bit_count) [[likely]]
 			{
 				out = static_cast<u32>(data);
 				data = 0;
+				return true;
 			}
-			else
-			{
-				data |= bit_wait;
-			}
-		});
 
-		return (old & bit_count) != 0;
+			return false;
+		}).second;
 	}
 
 	// Reading without modification
@@ -253,19 +247,95 @@ public:
 	}
 
 	// Pop unconditionally (loading last value), may require notification
-	u32 pop(cpu_thread& spu)
+	u32 pop()
 	{
 		// Value is not cleared and may be read again
 		const u64 old = data.fetch_and(~(bit_count | bit_wait));
 
 		if (old & bit_wait)
 		{
-			spu.notify();
+			data.notify_one();
 		}
 
 		return static_cast<u32>(old);
 	}
 
+	// Waiting for channel pop state availability, actually popping if specified
+	s64 pop_wait(cpu_thread& spu, bool pop = true)
+	{
+		while (true)
+		{
+			const u64 old = data.fetch_op([&](u64& data)
+			{
+				if (data & bit_count) [[likely]]
+				{
+					if (pop)
+					{
+						data = 0;
+						return true;
+					}
+
+					return false;
+				}
+
+				data = bit_wait;
+				return true;
+			}).first;
+
+			if (old & bit_count)
+			{
+				return static_cast<u32>(old);
+			}
+
+			if (spu.is_stopped())
+			{
+				return -1;
+			}
+
+			data.wait(bit_wait);
+		}
+	}
+
+	// Waiting for channel push state availability, actually pushing if specified
+	bool push_wait(cpu_thread& spu, u32 value, bool push = true)
+	{
+		while (true) 
+		{
+			u64 state;
+			data.fetch_op([&](u64& data)
+			{
+				if (data & bit_count) [[unlikely]]
+				{
+					data |= bit_wait;
+				}
+				else if (push)
+				{
+					data = bit_count | value;
+				}
+				else
+				{
+					state = data;
+					return false;
+				}
+
+				state = data;
+				return true;
+			});
+
+			if (!(state & bit_wait))
+			{
+				return true;
+			}
+
+			if (spu.is_stopped())
+			{
+				return false;
+			}
+
+			data.wait(state);
+		}
+	}
+
 	void set_value(u32 value, bool count = true)
 	{
 		data.release(u64{count} << off_count | value);

rpcs3/Emu/Cell/lv2/sys_spu.cpp

@@ -754,7 +754,7 @@ error_code sys_spu_thread_group_start(ppu_thread& ppu, u32 id)
 		if (thread && ran_threads--)
 		{
 			thread->state -= cpu_flag::stop;
-			thread_ctrl::notify(*thread);
+			thread_ctrl::raw_notify(*thread);
 		}
 	}
 
@@ -904,7 +904,7 @@ error_code sys_spu_thread_group_resume(ppu_thread& ppu, u32 id)
 		if (thread)
 		{
 			thread->state -= cpu_flag::suspend;
-			thread_ctrl::notify(*thread);
+			thread_ctrl::raw_notify(*thread);
 		}
 	}
 
@@ -1011,7 +1011,7 @@ error_code sys_spu_thread_group_terminate(ppu_thread& ppu, u32 id, s32 value)
 	{
 		if (thread && group->running)
 		{
-			thread_ctrl::notify(*thread);
+			thread_ctrl::raw_notify(*thread);
 		}
 	}
 
@@ -2203,7 +2203,7 @@ error_code raw_spu_read_puint_mb(u32 id, vm::ptr<u32> value)
 		return CELL_ESRCH;
 	}
 
-	*value = thread->ch_out_intr_mbox.pop(*thread);
+	*value = thread->ch_out_intr_mbox.pop();
 
 	return CELL_OK;
 }