Added mpscqueue (for future use)

2 files changed, 113 insertions, 0 deletions

diff --git a/zencore/include/zencore/mpscqueue.h b/zencore/include/zencore/mpscqueue.h
new file mode 100644
index 000000000..bb558bb5a
--- /dev/null
+++ b/zencore/include/zencore/mpscqueue.h
@@ -0,0 +1,109 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <atomic>
+#include <new>
+#include <optional>
+
+#ifdef __cpp_lib_hardware_interference_size
+using std::hardware_constructive_interference_size;
+using std::hardware_destructive_interference_size;
+#else
+// 64 bytes on x86-64 │ L1_CACHE_BYTES │ L1_CACHE_SHIFT │ __cacheline_aligned │ ...
+constexpr std::size_t hardware_constructive_interference_size = 64;
+constexpr std::size_t hardware_destructive_interference_size  = 64;
+#endif
+
+namespace zen {
+
+/** An untyped array of data with compile-time alignment and size derived from another type. */
+template<typename ElementType>
+struct TypeCompatibleStorage
+{
+	ElementType*	   Data() { return (ElementType*)this; }
+	const ElementType* Data() const { return (const ElementType*)this; }
+
+	char alignas(ElementType) DataMember;
+};
+
+/** Fast multi-producer/single-consumer unbounded concurrent queue.
+
+	Based on http://www.1024cores.net/home/lock-free-algorithms/queues/non-intrusive-mpsc-node-based-queue
+  */
+
+template<typename T>
+class MpscQueue final
+{
+public:
+	using ElementType = T;
+
+	MpscQueue()
+	{
+		Node* Sentinel = new Node;
+		Head.store(Sentinel, std::memory_order_relaxed);
+		Tail = Sentinel;
+	}
+
+	~MpscQueue()
+	{
+		Node* Next = Tail->Next.load(std::memory_order_relaxed);
+
+		// sentinel's value is already destroyed
+		delete Tail;
+
+		while (Next != nullptr)
+		{
+			Tail = Next;
+			Next = Tail->Next.load(std::memory_order_relaxed);
+
+			std::destroy_at((ElementType*)&Tail->Value);
+			delete Tail;
+		}
+	}
+
+	template<typename... ArgTypes>
+	void Enqueue(ArgTypes&&... Args)
+	{
+		Node* New = new Node;
+		new (&New->Value) ElementType(std::forward<ArgTypes>(Args)...);
+
+		Node* Prev = Head.exchange(New, std::memory_order_acq_rel);
+		Prev->Next.store(New, std::memory_order_release);
+	}
+
+	std::optional<ElementType> Dequeue()
+	{
+		Node* Next = Tail->Next.load(std::memory_order_acquire);
+
+		if (Next == nullptr)
+		{
+			return {};
+		}
+
+		ElementType*			   ValuePtr = (ElementType*)&Next->Value;
+		std::optional<ElementType> Res{std::move(*ValuePtr)};
+		std::destroy_at(ValuePtr);
+
+		delete Tail;  // current sentinel
+
+		Tail = Next;  // new sentinel
+		return Res;
+	}
+
+private:
+	struct Node
+	{
+		std::atomic<Node*>				   Next{nullptr};
+		TypeCompatibleStorage<ElementType> Value;
+	};
+
+private:
+	std::atomic<Node*> Head;  // accessed only by producers
+	alignas(hardware_constructive_interference_size)
+		Node* Tail;	 // accessed only by consumer, hence should be on a different cache line than `Head`
+};
+
+void mpscqueue_forcelink();
+
+}  // namespace zen
diff --git a/zencore/include/zencore/zencore.h b/zencore/include/zencore/zencore.h
index 4fad0b7a4..2b411af0e 100644
--- a/zencore/include/zencore/zencore.h
+++ b/zencore/include/zencore/zencore.h
@@ -60,6 +60,10 @@
 #	endif
 #endif
 
+#if ZEN_COMPILER_MSC
+#	pragma warning(disable : 4324)	 // warning C4324: '<type>': structure was padded due to alignment specifier
+#endif
+
 //////////////////////////////////////////////////////////////////////////
 // Architecture
 //