diff --git a/src/common/Threading/MPSCQueue.h b/src/common/Threading/MPSCQueue.h index fa5caf7a34..ac245e1b33 100644 --- a/src/common/Threading/MPSCQueue.h +++ b/src/common/Threading/MPSCQueue.h @@ -21,23 +21,27 @@ #include #include +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 -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 Next; @@ -76,8 +83,87 @@ private: std::atomic _head; std::atomic _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 T::* IntrusiveLink> +class MPSCQueueIntrusive +{ +public: + MPSCQueueIntrusive() : _dummyPtr(reinterpret_cast(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 _dummy; + T* _dummyPtr; + std::atomic _head; + std::atomic _tail; + + MPSCQueueIntrusive(MPSCQueueIntrusive const&) = delete; + MPSCQueueIntrusive& operator=(MPSCQueueIntrusive const&) = delete; +}; +} +} + +template T::* IntrusiveLink = nullptr> +using MPSCQueue = std::conditional_t, Trinity::Impl::MPSCQueueNonIntrusive>; + #endif // MPSCQueue_h__ diff --git a/src/server/game/Server/WorldSocket.cpp b/src/server/game/Server/WorldSocket.cpp index 7fde9e3a6b..4214a1940e 100644 --- a/src/server/game/Server/WorldSocket.cpp +++ b/src/server/game/Server/WorldSocket.cpp @@ -35,17 +35,6 @@ #include #include -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; std::string const WorldSocket::ServerConnectionInitialize("WORLD OF WARCRAFT CONNECTION - SERVER TO CLIENT"); diff --git a/src/server/game/Server/WorldSocket.h b/src/server/game/Server/WorldSocket.h index b4f0f9d7a0..1b97c0aad4 100644 --- a/src/server/game/Server/WorldSocket.h +++ b/src/server/game/Server/WorldSocket.h @@ -32,7 +32,21 @@ #include 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 SocketQueueLink; + +private: + bool _encrypt; +}; struct z_stream_s; @@ -138,7 +152,7 @@ private: z_stream_s* _compressionStream; - MPSCQueue _bufferQueue; + MPSCQueue _bufferQueue; std::size_t _sendBufferSize; bool _initialized;