Core/Networking: Switch to intrusive variant of MPSC queue to halve the number of memory allocations for packet sending (#24659)

3 files changed, 110 insertions, 21 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__
diff --git a/src/server/game/Server/WorldSocket.cpp b/src/server/game/Server/WorldSocket.cpp
index 17363de5eb8..0e65704076d 100644
--- a/src/server/game/Server/WorldSocket.cpp
+++ b/src/server/game/Server/WorldSocket.cpp
@@ -31,17 +31,6 @@
 #include "WorldSession.h"
 #include <memory>
 
-class EncryptablePacket : public WorldPacket
-{
-public:
-    EncryptablePacket(WorldPacket const& packet, bool encrypt) : WorldPacket(packet), _encrypt(encrypt) { }
-
-    bool NeedsEncryption() const { return _encrypt; }
-
-private:
-    bool _encrypt;
-};
-
 using boost::asio::ip::tcp;
 
 WorldSocket::WorldSocket(tcp::socket&& socket)
diff --git a/src/server/game/Server/WorldSocket.h b/src/server/game/Server/WorldSocket.h
index 8bf230e8005..6f6b0848423 100644
--- a/src/server/game/Server/WorldSocket.h
+++ b/src/server/game/Server/WorldSocket.h
@@ -30,7 +30,21 @@
 #include <boost/asio/ip/tcp.hpp>
 
 using boost::asio::ip::tcp;
-class EncryptablePacket;
+class EncryptablePacket : public WorldPacket
+{
+public:
+    EncryptablePacket(WorldPacket const& packet, bool encrypt) : WorldPacket(packet), _encrypt(encrypt)
+    {
+        SocketQueueLink.store(nullptr, std::memory_order_relaxed);
+    }
+
+    bool NeedsEncryption() const { return _encrypt; }
+
+    std::atomic<EncryptablePacket*> SocketQueueLink;
+
+private:
+    bool _encrypt;
+};
 
 namespace WorldPackets
 {
@@ -110,7 +124,7 @@ private:
 
     MessageBuffer _headerBuffer;
     MessageBuffer _packetBuffer;
-    MPSCQueue<EncryptablePacket> _bufferQueue;
+    MPSCQueue<EncryptablePacket, &EncryptablePacket::SocketQueueLink> _bufferQueue;
     std::size_t _sendBufferSize;
 
     QueryCallbackProcessor _queryProcessor;