From 97a79af4701621ec04b88c8b548dbc35d120e99e Mon Sep 17 00:00:00 2001 From: Shauren Date: Fri, 19 Feb 2016 19:23:04 +0100 Subject: Core/Networking: Rewrite networking threading model Each network thread has its own io_service - this means that all operations on a given socket except queueing packets run from a single thread, removing the need for locking Sending packets now writes to a lockfree intermediate queue directly, encryption is applied in network thread if it was required at the time of sending the packet --- src/common/Threading/MPSCQueue.h | 83 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 src/common/Threading/MPSCQueue.h (limited to 'src/common/Threading/MPSCQueue.h') diff --git a/src/common/Threading/MPSCQueue.h b/src/common/Threading/MPSCQueue.h new file mode 100644 index 00000000000..09648b844be --- /dev/null +++ b/src/common/Threading/MPSCQueue.h @@ -0,0 +1,83 @@ +/* + * Copyright (C) 2008-2016 TrinityCore + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for + * more details. + * + * You should have received a copy of the GNU General Public License along + * with this program. If not, see . + */ + +#ifndef MPSCQueue_h__ +#define MPSCQueue_h__ + +#include +#include + +// 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 +{ +public: + MPSCQueue() : _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() + { + T* output; + while (this->Dequeue(output)) + ; + + Node* front = _head.load(std::memory_order_relaxed); + delete front; + } + + void Enqueue(T* input) + { + Node* node = new Node(input); + Node* prevHead = _head.exchange(node, std::memory_order_acq_rel); + prevHead->Next.store(node, std::memory_order_release); + } + + bool Dequeue(T*& result) + { + Node* tail = _tail.load(std::memory_order_relaxed); + Node* next = tail->Next.load(std::memory_order_acquire); + if (!next) + return false; + + result = next->Data; + _tail.store(next, std::memory_order_release); + delete tail; + return true; + } + +private: + struct Node + { + Node() = default; + explicit Node(T* data) : Data(data) { Next.store(nullptr, std::memory_order_relaxed); } + + T* Data; + std::atomic Next; + }; + + std::atomic _head; + std::atomic _tail; + + MPSCQueue(MPSCQueue const&) = delete; + MPSCQueue& operator=(MPSCQueue const&) = delete; +}; + +#endif // MPSCQueue_h__ -- cgit v1.2.3