Implement cond_one sync primitive

Change futex() args to use unsigned int

Utilities/cond.cpp

@@ -40,19 +40,13 @@ bool cond_variable::imp_wait(u32 _old, u64 _timeout) noexcept
 
 	return true;
 #else
-	if (!_timeout)
-	{
-		verify(HERE), m_value--;
-		return false;
-	}
-
 	timespec timeout;
 	timeout.tv_sec  = _timeout / 1000000;
 	timeout.tv_nsec = (_timeout % 1000000) * 1000;
 
 	for (u32 value = _old + 1;; value = m_value)
 	{
-		const int err = futex((int*)&m_value.raw(), FUTEX_WAIT_PRIVATE, value, is_inf ? nullptr : &timeout, nullptr, 0) == 0
+		const int err = futex(&m_value, FUTEX_WAIT_PRIVATE, value, is_inf ? nullptr : &timeout) == 0
 			? 0
 			: errno;
 
@@ -106,7 +100,7 @@ void cond_variable::imp_wake(u32 _count) noexcept
 			return;
 		}
 
-		if (const int res = futex((int*)&m_value.raw(), FUTEX_WAKE_PRIVATE, i > INT_MAX ? INT_MAX : i, nullptr, nullptr, 0))
+		if (const int res = futex(&m_value, FUTEX_WAKE_PRIVATE, i > INT_MAX ? INT_MAX : i))
 		{
 			verify(HERE), res >= 0 && (u32)res <= i;
 			i -= res;
@@ -214,3 +208,88 @@ bool notifier::wait(u64 usec_timeout)
 
 	return res;
 }
+
+bool cond_one::imp_wait(u32 _old, u64 _timeout) noexcept
+{
+	verify(HERE), _old == c_lock;
+
+	const bool is_inf = _timeout > cond_variable::max_timeout;
+
+#ifdef _WIN32
+	LARGE_INTEGER timeout;
+	timeout.QuadPart = _timeout * -10;
+
+	if (HRESULT rc = _timeout ? NtWaitForKeyedEvent(nullptr, &m_value, false, is_inf ? nullptr : &timeout) : WAIT_TIMEOUT)
+	{
+		verify(HERE), rc == WAIT_TIMEOUT;
+
+		// Retire
+		const bool signaled = m_value.exchange(c_lock) == c_sig;
+		while (signaled)
+		{
+			timeout.QuadPart = 0;
+
+			if (HRESULT rc2 = NtWaitForKeyedEvent(nullptr, &m_value, false, &timeout))
+			{
+				verify(HERE), rc2 == WAIT_TIMEOUT;
+				SwitchToThread();
+				continue;
+			}
+
+			return true;
+		}
+
+		return false;
+	}
+#else
+	timespec timeout;
+	timeout.tv_sec  = _timeout / 1000000;
+	timeout.tv_nsec = (_timeout % 1000000) * 1000;
+
+	for (u32 value = _old - 1; value != c_sig; value = m_value)
+	{
+		const int err = futex(&m_value, FUTEX_WAIT_PRIVATE, value, is_inf ? nullptr : &timeout) == 0
+			? 0
+			: errno;
+
+		// Normal or timeout wakeup
+		if (!err || (!is_inf && err == ETIMEDOUT))
+		{
+			return m_value.exchange(c_lock) == c_sig;
+		}
+
+		// Not a wakeup
+		verify(HERE), err == EAGAIN;
+	}
+#endif
+
+	verify(HERE), m_value.exchange(c_lock) == c_sig;
+	return true;
+}
+
+void cond_one::imp_notify() noexcept
+{
+	auto [old, ok] = m_value.fetch_op([](u32& v)
+	{
+		if (UNLIKELY(v > 0 && v < c_sig))
+		{
+			v = c_sig;
+			return true;
+		}
+
+		return false;
+	});
+
+	verify(HERE), old <= c_sig;
+
+	if (LIKELY(!ok || old == c_lock))
+	{
+		return;
+	}
+
+#ifdef _WIN32
+	NtReleaseKeyedEvent(nullptr, &m_value, false, nullptr);
+#else
+	futex(&m_value, FUTEX_WAKE_PRIVATE, 1);
+#endif
+}

Utilities/cond.h

@@ -127,3 +127,55 @@ public:
 
 	static constexpr u32 max_readers = 0x7f;
 };
+
+class cond_one
+{
+	enum : u32
+	{
+		c_wait = 1,
+		c_lock = 2,
+		c_sig  = 3,
+	};
+
+	atomic_t<u32> m_value{0};
+
+	bool imp_wait(u32 _old, u64 _timeout) noexcept;
+	void imp_notify() noexcept;
+
+public:
+	constexpr cond_one() = default;
+
+	void lock() noexcept
+	{
+		// Shouldn't be locked by more than one thread concurrently
+		while (UNLIKELY(!m_value.compare_and_swap_test(0, c_lock)))
+			;
+	}
+
+	void unlock() noexcept
+	{
+		m_value = 0;
+	}
+
+	bool wait(std::unique_lock<cond_one>& lock, u64 usec_timeout = -1) noexcept
+	{
+		AUDIT(lock.owns_lock());
+		AUDIT(lock.mutex() == this);
+
+		// State transition: c_sig -> c_lock, c_lock -> c_wait
+		const u32 _old = m_value.fetch_sub(1);
+		if (LIKELY(_old == c_sig))
+			return true;
+
+		return imp_wait(_old, usec_timeout);
+	}
+
+	void notify() noexcept
+	{
+		// Early exit if notification is not required
+		if (LIKELY(!m_value))
+			return;
+
+		imp_notify();
+	}
+};

Utilities/mutex.cpp

@@ -66,7 +66,7 @@ void shared_mutex::imp_wait()
 			return;
 		}
 
-		futex(reinterpret_cast<int*>(&m_value.raw()), FUTEX_WAIT_BITSET_PRIVATE, value, nullptr, nullptr, c_sig);
+		futex(&m_value, FUTEX_WAIT_BITSET_PRIVATE, value, nullptr, c_sig);
 	}
 #endif
 }
@@ -77,8 +77,8 @@ void shared_mutex::imp_signal()
 	NtReleaseKeyedEvent(nullptr, &m_value, false, nullptr);
 #else
 	m_value += c_sig;
-	futex(reinterpret_cast<int*>(&m_value.raw()), FUTEX_WAKE_BITSET_PRIVATE, 1, nullptr, nullptr, c_sig);
-	//futex(reinterpret_cast<int*>(&m_value.raw()), FUTEX_WAKE_BITSET_PRIVATE, c_one, nullptr, nullptr, c_sig - 1);
+	futex(&m_value, FUTEX_WAKE_BITSET_PRIVATE, 1, nullptr, c_sig);
+	//futex(&m_value, FUTEX_WAKE_BITSET_PRIVATE, c_one, nullptr, c_sig - 1);
 #endif
 }
 
@@ -185,7 +185,7 @@ void shared_mutex::imp_lock_unlock()
 		_max = val / c_one;
 
 		// Monitor all bits except c_sig
-		futex(reinterpret_cast<int*>(&m_value.raw()), FUTEX_WAIT_BITSET_PRIVATE, val, nullptr, nullptr, c_sig - 1);
+		futex(&m_value, FUTEX_WAIT_BITSET_PRIVATE, val, nullptr, c_sig - 1);
 	}
 #endif
 

Utilities/sema.cpp

@@ -51,11 +51,11 @@ void semaphore_base::imp_wait()
 		if (value >= 0)
 		{
 			// Signal other waiter to wake up or to restore sign bit
-			futex(&m_value.raw(), FUTEX_WAKE_PRIVATE, 1, nullptr, nullptr, 0);
+			futex(&m_value, FUTEX_WAKE_PRIVATE, 1);
 			return;
 		}
 
-		futex(&m_value.raw(), FUTEX_WAIT_PRIVATE, value, nullptr, nullptr, 0);
+		futex(&m_value, FUTEX_WAIT_PRIVATE, value);
 	}
 #endif
 }
@@ -67,7 +67,7 @@ void semaphore_base::imp_post(s32 _old)
 #ifdef _WIN32
 	NtReleaseKeyedEvent(nullptr, &m_value, false, nullptr);
 #else
-	futex(&m_value.raw(), FUTEX_WAKE_PRIVATE, 1, nullptr, nullptr, 0);
+	futex(&m_value, FUTEX_WAKE_PRIVATE, 1);
 #endif
 }
 

Utilities/sync.h

@@ -49,24 +49,24 @@ enum
 };
 #endif
 
-inline int futex(int* uaddr, int futex_op, int val, const timespec* timeout, int* uaddr2, int val3)
+inline int futex(volatile void* uaddr, int futex_op, uint val, const timespec* timeout = nullptr, uint mask = 0)
 {
 #ifdef __linux__
-	return syscall(SYS_futex, uaddr, futex_op, val, timeout, uaddr, val3);
+	return syscall(SYS_futex, uaddr, futex_op, static_cast<int>(val), timeout, nullptr, static_cast<int>(mask));
 #else
-	static struct futex_map
+	static struct futex_manager
 	{
 		struct waiter
 		{
-			 int val;
+			 uint val;
 			 uint mask;
 			 std::condition_variable cv;
 		};
 
 		std::mutex mutex;
-		std::unordered_multimap<int*, waiter*, pointer_hash<int>> map;
+		std::unordered_multimap<volatile void*, waiter*, pointer_hash<volatile void, alignof(int)>> map;
 
-		int operator()(int* uaddr, int futex_op, int val, const timespec* timeout, int*, uint val3)
+		int operator()(volatile void* uaddr, int futex_op, uint val, const timespec* timeout, uint mask)
 		{
 			std::unique_lock lock(mutex);
 
@@ -74,12 +74,12 @@ inline int futex(int* uaddr, int futex_op, int val, const timespec* timeout, int
 			{
 			case FUTEX_WAIT:
 			{
-				val3 = -1;
-				// Fallthrough
+				mask = -1;
+				[[fallthrough]];
 			}
 			case FUTEX_WAIT_BITSET:
 			{
-				if (*(volatile int*)uaddr != val)
+				if (*reinterpret_cast<volatile uint*>(uaddr) != val)
 				{
 					errno = EAGAIN;
 					return -1;
@@ -87,7 +87,7 @@ inline int futex(int* uaddr, int futex_op, int val, const timespec* timeout, int
 
 				waiter rec;
 				rec.val = val;
-				rec.mask = val3;
+				rec.mask = mask;
 				const auto& ref = *map.emplace(uaddr, &rec);
 
 				int res = 0;
@@ -117,8 +117,8 @@ inline int futex(int* uaddr, int futex_op, int val, const timespec* timeout, int
 
 			case FUTEX_WAKE:
 			{
-				val3 = -1;
-				// Fallthrough
+				mask = -1;
+				[[fallthrough]];
 			}
 			case FUTEX_WAKE_BITSET:
 			{
@@ -128,7 +128,7 @@ inline int futex(int* uaddr, int futex_op, int val, const timespec* timeout, int
 				{
 					auto& entry = *range.first->second;
 
-					if (entry.mask & val3)
+					if (entry.mask & mask)
 					{
 						entry.cv.notify_one();
 						entry.mask = 0;
@@ -146,6 +146,6 @@ inline int futex(int* uaddr, int futex_op, int val, const timespec* timeout, int
 		}
 	} g_futex;
 
-	return g_futex(uaddr, futex_op, val, timeout, uaddr2, val3);
+	return g_futex(uaddr, futex_op, val, timeout, mask);
 #endif
 }

rpcs3/Emu/Cell/Modules/cellVdec.cpp

@@ -89,7 +89,7 @@ struct vdec_context final
 
 	atomic_t<u32> au_count{0};
 
-	notifier in_cv;
+	cond_one in_cv;
 	lf_queue<std::variant<vdec_start_seq_t, vdec_close_t, vdec_cmd, CellVdecFrameRate>> in_cmd;
 
 	vdec_context(s32 type, u32 profile, u32 addr, u32 size, vm::ptr<CellVdecCbMsg> func, u32 arg)
@@ -160,13 +160,13 @@ struct vdec_context final
 	{
 		ppu_tid = ppu.id;
 
-		std::shared_lock no_lock(in_cv, std::try_to_lock);
+		std::unique_lock cv_lock(in_cv);
 
 		for (auto cmds = in_cmd.pop_all(); !Emu.IsStopped(); cmds ? cmds = cmds->pop_all() : cmds = in_cmd.pop_all())
 		{
 			if (!cmds)
 			{
-				in_cv.wait(1000);
+				in_cv.wait(cv_lock, 1000);
 				continue;
 			}
 
@@ -378,7 +378,7 @@ struct vdec_context final
 
 				while (!Emu.IsStopped() && out_max && (std::lock_guard{mutex}, out.size() > out_max))
 				{
-					in_cv.wait(1000);
+					in_cv.wait(cv_lock, 1000);
 				}
 			}
 			else if (auto* frc = std::get_if<CellVdecFrameRate>(&cmds->get()))
@@ -486,7 +486,7 @@ s32 cellVdecClose(ppu_thread& ppu, u32 handle)
 	lv2_obj::sleep(ppu);
 	vdec->out_max = 0;
 	vdec->in_cmd.push(vdec_close);
-	vdec->in_cv.notify_all();
+	vdec->in_cv.notify();
 	ppu_execute<&sys_interrupt_thread_disestablish>(ppu, vdec->ppu_tid);
 	return CELL_OK;
 }
@@ -503,7 +503,7 @@ s32 cellVdecStartSeq(u32 handle)
 	}
 
 	vdec->in_cmd.push(vdec_start_seq);
-	vdec->in_cv.notify_all();
+	vdec->in_cv.notify();
 	return CELL_OK;
 }
 
@@ -519,7 +519,7 @@ s32 cellVdecEndSeq(u32 handle)
 	}
 
 	vdec->in_cmd.push(vdec_cmd{-1});
-	vdec->in_cv.notify_all();
+	vdec->in_cv.notify();
 	return CELL_OK;
 }
 
@@ -541,7 +541,7 @@ s32 cellVdecDecodeAu(u32 handle, CellVdecDecodeMode mode, vm::cptr<CellVdecAuInf
 
 	// TODO: check info
 	vdec->in_cmd.push(vdec_cmd{mode, *auInfo});
-	vdec->in_cv.notify_all();
+	vdec->in_cv.notify();
 	return CELL_OK;
 }
 
@@ -574,7 +574,7 @@ s32 cellVdecGetPicture(u32 handle, vm::cptr<CellVdecPicFormat> format, vm::ptr<u
 	}
 
 	if (notify)
-		vdec->in_cv.notify_all();
+		vdec->in_cv.notify();
 
 	if (outBuff)
 	{
@@ -878,7 +878,7 @@ s32 cellVdecSetFrameRate(u32 handle, CellVdecFrameRate frc)
 
 	// TODO: check frc value
 	vdec->in_cmd.push(frc);
-	vdec->in_cv.notify_all();
+	vdec->in_cv.notify();
 	return CELL_OK;
 }