atomic_t<>: extend fetch_op to support cancellation

Use std::invoke inside atomic_op/fetch_op
Remove op_fetch because it's easily replaced
Add fetch_dec_sat algorithm (conditional decrement)

Utilities/Atomic.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "types.h"
+#include <functional>
 
 // Helper class, provides access to compiler-specific atomic intrinsics
 template<typename T, std::size_t Size = sizeof(T)>
@@ -609,36 +610,31 @@ public:
 		return atomic_storage<type>::compare_exchange(m_data, old, exch);
 	}
 
-	// Atomic operation; returns old value
+	// Atomic operation; returns old value, or pair of old value and return value (cancel op if evaluates to false)
-	template <typename F>
+	template <typename F, typename RT = std::invoke_result_t<F, T&>>
-	std::enable_if_t<std::is_void<std::invoke_result_t<F, T&>>::value, type> fetch_op(F&& func)
+	std::conditional_t<std::is_void_v<RT>, type, std::pair<type, RT>> fetch_op(F&& func)
 	{
 		type _new, old = atomic_storage<type>::load(m_data);
 
 		while (true)
 		{
-			func((_new = old));
+			if constexpr (std::is_void_v<RT>)
+			{
+				std::invoke(func, (_new = old));
 
 				if (LIKELY(atomic_storage<type>::compare_exchange(m_data, old, _new)))
 				{
 					return old;
 				}
 			}
+			else
+			{
+				RT ret = std::invoke(func, (_new = old));
+
+				if (LIKELY(!ret || atomic_storage<type>::compare_exchange(m_data, old, _new)))
+				{
+					return {old, std::move(ret)};
 				}
-
-	// Atomic operation; returns new value
-	template <typename F>
-	std::enable_if_t<std::is_void<std::invoke_result_t<F, T&>>::value, type> op_fetch(F&& func)
-	{
-		type _new, old = atomic_storage<type>::load(m_data);
-
-		while (true)
-		{
-			func((_new = old));
-
-			if (LIKELY(atomic_storage<type>::compare_exchange(m_data, old, _new)))
-			{
-				return _new;
 			}
 		}
 	}
@@ -651,9 +647,9 @@ public:
 
 		while (true)
 		{
-			if constexpr(std::is_void<RT>::value)
+			if constexpr (std::is_void_v<RT>)
 			{
-				func((_new = old), args...);
+				std::invoke(func, (_new = old), args...);
 
 				if (LIKELY(atomic_storage<type>::compare_exchange(m_data, old, _new)))
 				{
@@ -662,7 +658,7 @@ public:
 			}
 			else
 			{
-				RT result = func((_new = old), args...);
+				RT result = std::invoke(func, (_new = old), args...);
 
 				if (LIKELY(atomic_storage<type>::compare_exchange(m_data, old, _new)))
 				{
@@ -722,9 +718,10 @@ public:
 			return atomic_storage<type>::add_fetch(m_data, rhs);
 		}
 
-		return op_fetch([&](T& v)
+		return atomic_op([&](T& v)
 		{
 			v += rhs;
+			return v;
 		});
 	}
 
@@ -761,9 +758,10 @@ public:
 			return atomic_storage<type>::sub_fetch(m_data, rhs);
 		}
 
-		return op_fetch([&](T& v)
+		return atomic_op([&](T& v)
 		{
 			v -= rhs;
+			return v;
 		});
 	}
 
@@ -800,9 +798,10 @@ public:
 			return atomic_storage<type>::and_fetch(m_data, rhs);
 		}
 
-		return op_fetch([&](T& v)
+		return atomic_op([&](T& v)
 		{
 			v &= rhs;
+			return v;
 		});
 	}
 
@@ -839,9 +838,10 @@ public:
 			return atomic_storage<type>::or_fetch(m_data, rhs);
 		}
 
-		return op_fetch([&](T& v)
+		return atomic_op([&](T& v)
 		{
 			v |= rhs;
+			return v;
 		});
 	}
 
@@ -878,9 +878,10 @@ public:
 			return atomic_storage<type>::xor_fetch(m_data, rhs);
 		}
 
-		return op_fetch([&](T& v)
+		return atomic_op([&](T& v)
 		{
 			v ^= rhs;
+			return v;
 		});
 	}
 
@@ -948,4 +949,28 @@ public:
 			return v--;
 		});
 	}
+
+	// Conditionally decrement
+	simple_type fetch_dec_sat(simple_type greater_than = std::numeric_limits<simple_type>::min(), simple_type amount = 1)
+	{
+		type _new, old = atomic_storage<type>::load(m_data);
+
+		while (true)
+		{
+			_new = old;
+
+			if (_new <= greater_than)
+			{
+				// Early exit
+				return old;
+			}
+
+			_new -= amount;
+
+			if (LIKELY(atomic_storage<type>::compare_exchange(m_data, old, _new)))
+			{
+				return old;
+			}
+		}
+	}
 };

Utilities/cond.cpp

@@ -21,7 +21,7 @@ bool cond_variable::imp_wait(u32 _old, u64 _timeout) noexcept
 		verify(HERE), rc == WAIT_TIMEOUT;
 
 		// Retire
-		while (!m_value.fetch_op([](u32& value) { if (value) value--; }))
+		while (!m_value.fetch_dec_sat())
 		{
 			timeout.QuadPart = 0;
 

Utilities/event.h

@@ -111,9 +111,10 @@ public:
 	// Unconditionally set next state
 	void state_next()
 	{
-		T _old, state = m_value.op_fetch([&](T& value)
+		T _old, state = m_value.atomic_op([&](T& value)
 		{
 			_old = value++;
+			return value;
 		});
 
 		(static_cast<CRT*>(this)->*transition_get(state))(_old);
@@ -122,9 +123,10 @@ public:
 	// Unconditionally set previous state
 	void state_prev()
 	{
-		T _old, state = m_value.op_fetch([&](T& value)
+		T _old, state = m_value.atomic_op([&](T& value)
 		{
 			_old = value--;
+			return value;
 		});
 
 		(static_cast<CRT*>(this)->*transition_get(state))(_old);

Utilities/mutex.cpp

@@ -237,7 +237,7 @@ void shared_mutex::imp_lock_degrade()
 bool shared_mutex::try_lock_shared()
 {
 	// Conditional decrement
-	return m_value.fetch_op([](s64& value) { if (value >= c_min) value -= c_min; }) >= c_min;
+	return m_value.fetch_dec_sat(c_min - 1, c_min) >= c_min;
 }
 
 bool shared_mutex::try_lock()

Utilities/sema.cpp

@@ -26,7 +26,7 @@ void semaphore_base::imp_wait()
 	while (true)
 	{
 		// Try hard way
-		const s32 value = m_value.op_fetch([](s32& value)
+		const s32 value = m_value.atomic_op([](s32& value)
 		{
 			// Use sign bit to acknowledge waiter presence
 			if (value && value > INT32_MIN)
@@ -44,6 +44,8 @@ void semaphore_base::imp_wait()
 				// Set sign bit
 				value = INT32_MIN;
 			}
+
+			return value;
 		});
 
 		if (value >= 0)
@@ -69,20 +71,6 @@ void semaphore_base::imp_post(s32 _old)
 #endif
 }
 
-bool semaphore_base::try_wait()
-{
-	// Conditional decrement
-	const s32 value = m_value.fetch_op([](s32& value)
-	{
-		if (value > 0)
-		{
-			value -= 1;
-		}
-	});
-
-	return value > 0;
-}
-
 bool semaphore_base::try_post(s32 _max)
 {
 	// Conditional increment

Utilities/sema.h

@@ -32,7 +32,10 @@ protected:
 		}
 	}
 
-	bool try_wait();
+	bool try_wait()
+	{
+		return m_value.fetch_dec_sat(0) > 0;
+	}
 
 	void post(s32 _max)
 	{

Utilities/types.h

@@ -13,6 +13,7 @@
 #include <type_traits>
 #include <utility>
 #include <chrono>
+#include <limits>
 #include <array>
 
 // Assume little-endian

rpcs3/Emu/Cell/lv2/lv2.cpp

@@ -1012,15 +1012,18 @@ void lv2_obj::sleep_timeout(named_thread& thread, u64 timeout)
 	{
 		LOG_TRACE(PPU, "sleep() - waiting (%zu)", g_pending.size());
 
-		auto state = ppu->state.fetch_op([&](auto& val)
+		const auto [_, ok] = ppu->state.fetch_op([&](bs_t<cpu_flag>& val)
 		{
 			if (!(val & cpu_flag::signal))
 			{
 				val += cpu_flag::suspend;
+				return true;
 			}
+
+			return false;
 		});
 
-		if (state & cpu_flag::signal)
+		if (!ok)
 		{
 			LOG_TRACE(PPU, "sleep() failed (signaled)");
 			return;

rpcs3/Emu/Cell/lv2/sys_lwcond.cpp

@@ -290,7 +290,7 @@ error_code _sys_lwcond_queue_wait(ppu_thread& ppu, u32 lwcond_id, u32 lwmutex_id
 
 				cond->waiters--;
 
-				if (mutex->signaled.fetch_op([](u32& v) { if (v) v--; }))
+				if (mutex->signaled.fetch_dec_sat())
 				{
 					ppu.gpr[3] = CELL_EDEADLK;
 					break;

rpcs3/Emu/Cell/lv2/sys_memory.h

@@ -66,20 +66,18 @@ struct lv2_memory_container
 	// Try to get specified amount of "physical" memory
 	u32 take(u32 amount)
 	{
-		const u32 old_value = used.fetch_op([&](u32& value)
+		auto [_, result] = used.fetch_op([&](u32& value) -> u32
 		{
 			if (size - value >= amount)
 			{
 				value += amount;
-			}
-		});
-
-		if (size - old_value >= amount)
-		{
 				return amount;
 			}
 
 			return 0;
+		});
+
+		return result;
 	}
 };