1 files changed, 94 insertions, 8 deletions

diff --git a/src/common/Threading/MPSCQueue.h b/src/common/Threading/MPSCQueue.h
index fa5caf7a34b..ac245e1b332 100644
--- a/src/common/Threading/MPSCQueue.h
+++ b/src/common/Threading/MPSCQueue.h
@@ -21,23 +21,27 @@
 #include <atomic>
 #include <utility>
 
+namespace Trinity
+{
+namespace Impl
+{
 // C++ implementation of Dmitry Vyukov's lock free MPSC queue
 // http://www.1024cores.net/home/lock-free-algorithms/queues/non-intrusive-mpsc-node-based-queue
 template<typename T>
-class MPSCQueue
+class MPSCQueueNonIntrusive
 {
 public:
-    MPSCQueue() : _head(new Node()), _tail(_head.load(std::memory_order_relaxed))
+    MPSCQueueNonIntrusive() : _head(new Node()), _tail(_head.load(std::memory_order_relaxed))
     {
         Node* front = _head.load(std::memory_order_relaxed);
         front->Next.store(nullptr, std::memory_order_relaxed);
     }
 
-    ~MPSCQueue()
+    ~MPSCQueueNonIntrusive()
     {
         T* output;
-        while (this->Dequeue(output))
-            ;
+        while (Dequeue(output))
+            delete output;
 
         Node* front = _head.load(std::memory_order_relaxed);
         delete front;
@@ -67,7 +71,10 @@ private:
     struct Node
     {
         Node() = default;
-        explicit Node(T* data) : Data(data) { Next.store(nullptr, std::memory_order_relaxed); }
+        explicit Node(T* data) : Data(data)
+        {
+            Next.store(nullptr, std::memory_order_relaxed);
+        }
 
         T* Data;
         std::atomic<Node*> Next;
@@ -76,8 +83,87 @@ private:
     std::atomic<Node*> _head;
     std::atomic<Node*> _tail;
 
-    MPSCQueue(MPSCQueue const&) = delete;
-    MPSCQueue& operator=(MPSCQueue const&) = delete;
+    MPSCQueueNonIntrusive(MPSCQueueNonIntrusive const&) = delete;
+    MPSCQueueNonIntrusive& operator=(MPSCQueueNonIntrusive const&) = delete;
+};
+
+// C++ implementation of Dmitry Vyukov's lock free MPSC queue
+// http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue
+template<typename T, std::atomic<T*> T::* IntrusiveLink>
+class MPSCQueueIntrusive
+{
+public:
+    MPSCQueueIntrusive() : _dummyPtr(reinterpret_cast<T*>(std::addressof(_dummy))), _head(_dummyPtr), _tail(_dummyPtr)
+    {
+        T* front = _head.load(std::memory_order_relaxed);
+        (front->*IntrusiveLink).store(nullptr, std::memory_order_relaxed);
+        // _dummy is constructed from aligned_storage and is left uninitialized so we init only its Next here
+        (_dummyPtr->*IntrusiveLink).store(nullptr, std::memory_order_relaxed);
+    }
+
+    ~MPSCQueueIntrusive()
+    {
+        T* output;
+        while (Dequeue(output))
+            delete output;
+    }
+
+    void Enqueue(T* input)
+    {
+        (input->*IntrusiveLink).store(nullptr, std::memory_order_release);
+        T* prevHead = _head.exchange(input, std::memory_order_acq_rel);
+        (prevHead->*IntrusiveLink).store(input, std::memory_order_release);
+    }
+
+    bool Dequeue(T*& result)
+    {
+        T* tail = _tail.load(std::memory_order_relaxed);
+        T* next = (tail->*IntrusiveLink).load(std::memory_order_acquire);
+        if (tail == _dummyPtr)
+        {
+            if (!next)
+                return false;
+
+            _tail.store(next, std::memory_order_release);
+            tail = next;
+            next = (next->*IntrusiveLink).load(std::memory_order_acquire);
+        }
+
+        if (next)
+        {
+            _tail.store(next, std::memory_order_release);
+            result = tail;
+            return true;
+        }
+
+        T* head = _head.load(std::memory_order_acquire);
+        if (tail != head)
+            return false;
+
+        Enqueue(_dummyPtr);
+        next = (tail->*IntrusiveLink).load(std::memory_order_acquire);
+        if (next)
+        {
+            _tail.store(next, std::memory_order_release);
+            result = tail;
+            return true;
+        }
+        return false;
+    }
+
+private:
+    std::aligned_storage_t<sizeof(T), alignof(T)> _dummy;
+    T* _dummyPtr;
+    std::atomic<T*> _head;
+    std::atomic<T*> _tail;
+
+    MPSCQueueIntrusive(MPSCQueueIntrusive const&) = delete;
+    MPSCQueueIntrusive& operator=(MPSCQueueIntrusive const&) = delete;
 };
+}
+}
+
+template<typename T, std::atomic<T*> T::* IntrusiveLink = nullptr>
+using MPSCQueue = std::conditional_t<IntrusiveLink != nullptr, Trinity::Impl::MPSCQueueIntrusive<T, IntrusiveLink>, Trinity::Impl::MPSCQueueNonIntrusive<T>>;
 
 #endif // MPSCQueue_h__