Implement lf_queue<>, lf_value<>

lf_queue<>: unbound FIFO queue with dynamic linked-list
lf_value<>: concurrently-assignable value readable without locking at the cost of memory (using dynamic linked list)

Add atomic_t<>::compare_exchange

Utilities/Atomic.h

@@ -610,6 +610,12 @@ public:
 		return atomic_storage<type>::compare_exchange(m_data, old, exch);
 	}
 
+	// As in std::atomic
+	bool compare_exchange(type& cmp_and_old, const type& exch)
+	{
+		return atomic_storage<type>::compare_exchange(m_data, cmp_and_old, exch);
+	}
+
 	// Atomic operation; returns old value, or pair of old value and return value (cancel op if evaluates to false)
 	template <typename F, typename RT = std::invoke_result_t<F, T&>>
 	std::conditional_t<std::is_void_v<RT>, type, std::pair<type, RT>> fetch_op(F&& func)

Utilities/lockless.h

@@ -8,7 +8,7 @@
 //!
 //! T is the type of elements. Currently, default constructor of T shall be constexpr.
 //! N is initial element count, available without any memory allocation and only stored contiguously.
-template<typename T, std::size_t N>
+template <typename T, std::size_t N>
 class lf_array
 {
 	// Data (default-initialized)
@@ -328,3 +328,214 @@ public:
 	using lf_spsc<T, N>::size;
 	using lf_spsc<T, N>::operator [];
 };
+
+// Helper type, linked list element
+template <typename T>
+class lf_item final
+{
+	lf_item* m_link = nullptr;
+
+	T m_data;
+
+	template <typename U>
+	friend class lf_queue;
+
+	constexpr lf_item() = default;
+
+	template <typename... Args>
+	constexpr lf_item(lf_item* link, Args&&... args)
+	    : m_link(link)
+	    , m_data(std::forward<Args>(args)...)
+	{
+	}
+
+public:
+	lf_item(const lf_item&) = delete;
+
+	lf_item& operator=(const lf_item&) = delete;
+
+	~lf_item()
+	{
+		for (lf_item* ptr = m_link; ptr;)
+		{
+			delete std::exchange(ptr, std::exchange(ptr->m_link, nullptr));
+		}
+	}
+
+	// Withdraw all other elements
+	std::unique_ptr<lf_item<T>> pop_all()
+	{
+		return std::unique_ptr<lf_item<T>>(std::exchange(m_link, nullptr));
+	}
+
+	[[nodiscard]] T& get()
+	{
+		return m_data;
+	}
+
+	[[nodiscard]] const T& get() const
+	{
+		return m_data;
+	}
+};
+
+// Full-dynamic multi-producer queue (consumer consumes everything or nothing, thread-safe)
+template <typename T>
+class lf_queue
+{
+	// Elements are added by replacing m_head
+	atomic_t<lf_item<T>*> m_head = nullptr;
+
+	// Extract all elements and reverse element order (FILO to FIFO)
+	lf_item<T>* reverse() noexcept
+	{
+		if (lf_item<T>* head = m_head.load() ? m_head.exchange(nullptr) : nullptr)
+		{
+			if (lf_item<T>* prev = head->m_link)
+			{
+				head->m_link = nullptr;
+
+				do
+				{
+					lf_item<T>* pprev = prev->m_link;
+					prev->m_link      = head;
+					head              = std::exchange(prev, pprev);
+				} while (prev);
+			}
+
+			return head;
+		}
+
+		return nullptr;
+	}
+
+public:
+	constexpr lf_queue() = default;
+
+	~lf_queue()
+	{
+		delete m_head.load();
+	}
+
+	template <typename... Args>
+	void push(Args&&... args)
+	{
+		lf_item<T>* old  = m_head.load();
+		lf_item<T>* item = new lf_item<T>(old, std::forward<Args>(args)...);
+
+		while (!m_head.compare_exchange(old, item))
+		{
+			item->m_link = old;
+		}
+	}
+
+	// Withdraw the list
+	std::unique_ptr<lf_item<T>> pop_all()
+	{
+		return std::unique_ptr<lf_item<T>>(reverse());
+	}
+
+	// Withdraw the list and apply func(data) to each element, return the total length
+	template <typename F>
+	std::size_t apply(F&& func)
+	{
+		std::size_t count = 0;
+
+		for (std::unique_ptr<lf_item<T>> ptr(reverse()); ptr; ptr = ptr->pop_all(), count++)
+		{
+			std::invoke(std::forward<F>(func), ptr->m_data);
+		}
+
+		return count;
+	}
+
+	// apply_all() overload for callable template argument
+	template <auto F>
+	std::size_t apply()
+	{
+		std::size_t count = 0;
+
+		for (std::unique_ptr<lf_item<T>> ptr(reverse()); ptr; ptr = ptr->pop_all(), count++)
+		{
+			std::invoke(F, ptr->m_data);
+		}
+
+		return count;
+	}
+};
+
+// Assignable lock-free thread-safe value of any type (memory-inefficient)
+template <typename T>
+class lf_value final
+{
+	atomic_t<lf_value*> m_head;
+
+	T m_data;
+
+public:
+	template <typename... Args>
+	explicit constexpr lf_value(Args&&... args)
+	    : m_head(this)
+	    , m_data(std::forward<Args>(args)...)
+	{
+	}
+
+	~lf_value()
+	{
+		// All values are kept in the queue until the end
+		for (lf_value* ptr = m_head.load(); ptr != this;)
+		{
+			delete std::exchange(ptr, std::exchange(ptr->m_head.raw(), ptr));
+		}
+	}
+
+	// Get current head, allows to inspect old values
+	[[nodiscard]] const lf_value* head() const
+	{
+		return m_head.load();
+	}
+
+	// Inspect the initial (oldest) value
+	[[nodiscard]] const T& first() const
+	{
+		return m_data;
+	}
+
+	[[nodiscard]] const T& get() const
+	{
+		return m_head.load()->m_data;
+	}
+
+	[[nodiscard]] operator const T&() const
+	{
+		return m_head.load()->m_data;
+	}
+
+	// Construct new value in-place
+	template <typename... Args>
+	const T& assign(Args&&... args)
+	{
+		lf_value* val = new lf_value(std::forward<Args>(args)...);
+		lf_value* old = m_head.load();
+
+		do
+		{
+			val->m_head = old;
+		}
+		while (!m_head.compare_exchange(old, val));
+
+		return val->m_data;
+	}
+
+	// Copy-assign new value
+	const T& operator =(const T& value)
+	{
+		return assign(value);
+	}
+
+	// Move-assign new value
+	const T& operator =(T&& value)
+	{
+		return assign(std::move(value));
+	}
+};

rpcs3/Emu/Cell/SPUThread.cpp

@@ -1,6 +1,5 @@
 #include "stdafx.h"
 #include "Utilities/JIT.h"
-#include "Utilities/lockless.h"
 #include "Utilities/sysinfo.h"
 #include "Emu/Memory/vm.h"
 #include "Emu/System.h"