SPU: Implement "quintuple" Inbound MBOX storage

rpcs3/Emu/Cell/RawSPUThread.cpp

@@ -244,7 +244,7 @@ bool spu_thread::write_reg(const u32 addr, const u32 value)
 
 	case SPU_In_MBox_offs:
 	{
-		ch_in_mbox.push(*this, value);
+		ch_in_mbox.push(value);
 		return true;
 	}
 

rpcs3/Emu/Cell/SPUThread.cpp

@@ -4124,9 +4124,9 @@ s64 spu_thread::get_ch_value(u32 ch)
 				busy_wait();
 			}
 
-			u32 out = 0;
+			const auto [old_count, out] = ch_in_mbox.pop_wait(*this);
 
-			if (const uint old_count = ch_in_mbox.try_pop(out))
+			if (old_count)
 			{
 				if (old_count == 4 /* SPU_IN_MBOX_THRESHOLD */) // TODO: check this
 				{

rpcs3/Emu/Cell/SPUThread.h

@@ -231,10 +231,9 @@ public:
 
 				// Turn off waiting bit manually (must succeed because waiting bit can only be resetted by the thread pushed to jostling_value)
 				ensure(this->data.bit_test_reset(off_wait));
+				data.notify_one();
 			}
 
-			data.notify_one();
-
 			// Return true if count has changed from 0 to 1, this condition is considered satisfied even if we pushed a value directly to the special storage for waiting SPUs 
 			return !pushed_to_data || (old & bit_count) == 0;
 		}
@@ -424,45 +423,61 @@ struct spu_channel_4_t
 	};
 
 	atomic_t<sync_var_t> values;
+	atomic_t<u64> jostling_value;
 	atomic_t<u32> value3;
 
-public:
+	static constexpr u32 off_wait  = 32;
+	static constexpr u64 bit_wait  = 1ull << off_wait;
+
 	void clear()
 	{
 		values.release({});
 	}
 
 	// push unconditionally (overwriting latest value), returns true if needs signaling
-	void push(cpu_thread& spu, u32 value)
+	void push(u32 value)
 	{
-		value3.release(value);
-
-		if (values.atomic_op([value](sync_var_t& data) -> bool
+		while (true)
 		{
-			switch (data.count++)
+			value3.release(value);
+			const auto [old, pushed_to_data] = values.fetch_op([&](sync_var_t& data)
 			{
-			case 0: data.value0 = value; break;
-			case 1: data.value1 = value; break;
-			case 2: data.value2 = value; break;
-			default:
-			{
-				data.count = 4;
-				data.value3_inval++; // Ensure the SPU reads the most recent value3 write in try_pop by re-loading
-				break;
-			}
-			}
+				if (data.waiting)
+				{
+					return false;
+				}
 
-			if (data.waiting)
-			{
-				data.waiting = 0;
+				switch (data.count++)
+				{
+				case 0: data.value0 = value; break;
+				case 1: data.value1 = value; break;
+				case 2: data.value2 = value; break;
+				default:
+				{
+					data.count = 4;
+					data.value3_inval++; // Ensure the SPU reads the most recent value3 write in try_pop by re-loading
+					break;
+				}
+				}
 
 				return true;
+			});
+
+			if (!pushed_to_data)
+			{
+				// Insert the pending value in special storage for waiting SPUs, leave no time in which the channel has data
+				if (!jostling_value.compare_and_swap_test(bit_wait, value))
+				{
+					// Other thread has inserted a value through jostling_value, retry
+					continue;
+				}
+
+				// Turn off waiting bit manually (must succeed because waiting bit can only be resetted by the thread pushing to jostling_value)
+				ensure(atomic_storage<u8>::exchange(values.raw().waiting, 0));
+				values.notify_one();
 			}
 
-			return false;
-		}))
-		{
-			spu.notify();
+			return;
 		}
 	}
 
@@ -483,15 +498,66 @@ public:
 				data.value1 = data.value2;
 				data.value2 = this->value3;
 			}
-			else
-			{
-				data.waiting = 1;
-			}
 
 			return result;
 		});
 	}
 
+	// Returns [previous count, value] (if aborted 0 count is returned)
+	std::pair<u32, u32> pop_wait(cpu_thread& spu)
+	{
+		u32 out_value = 0;
+		auto old = values.fetch_op([&](sync_var_t& data)
+		{
+			if (data.count != 0)
+			{
+				data.waiting = 0;
+				data.count--;
+				out_value = data.value0;
+
+				data.value0 = data.value1;
+				data.value1 = data.value2;
+				data.value2 = this->value3;
+			}
+			else
+			{
+				data.waiting = 1;
+				jostling_value.release(bit_wait);
+			}
+		});
+
+		if (old.count)
+		{
+			return {old.count, out_value};
+		}
+
+		old.waiting = 1;
+
+		while (true)
+		{
+			thread_ctrl::wait_on(values, old);
+			old = values;
+
+			if (!old.waiting)
+			{
+				// Count of 1 because a value has been inserted and popped in the same step.
+				return {1, static_cast<u32>(jostling_value)};
+			}
+
+			if (spu.is_stopped())
+			{
+				// Abort waiting and test if a value has been received
+				if (u64 v = jostling_value.exchange(0); !(v & bit_wait))
+				{
+					return {1, static_cast<u32>(v)};
+				}
+
+				ensure(atomic_storage<u8>::exchange(values.raw().waiting, 0));
+				return {};
+			}
+		}
+	}
+
 	// returns current queue size without modification
 	uint try_read(u32 (&out)[4]) const
 	{

rpcs3/Emu/Cell/lv2/sys_spu.cpp

@@ -1696,7 +1696,7 @@ error_code sys_spu_thread_write_spu_mb(ppu_thread& ppu, u32 id, u32 value)
 
 	std::lock_guard lock(group->mutex);
 
-	thread->ch_in_mbox.push(*thread, value);
+	thread->ch_in_mbox.push(value);
 
 	return CELL_OK;
 }