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

@@ -0,0 +1,83 @@
+/*
+ * Copyright (C) 2008-2016 TrinityCore <http://www.trinitycore.org/>
+ *
+ * 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 <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef MPSCQueue_h__
+#define MPSCQueue_h__
+
+#include <atomic>
+#include <utility>
+
+// 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
+{
+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<Node*> Next;
+    };
+
+    std::atomic<Node*> _head;
+    std::atomic<Node*> _tail;
+
+    MPSCQueue(MPSCQueue const&) = delete;
+    MPSCQueue& operator=(MPSCQueue const&) = delete;
+};
+
+#endif // MPSCQueue_h__

src/server/bnetserver/Server/Session.cpp

@@ -654,12 +654,37 @@ void Battlenet::Session::CheckIpCallback(PreparedQueryResult result)
 
 bool Battlenet::Session::Update()
 {
+    EncryptableBuffer* queued;
+    MessageBuffer buffer((std::size_t(BufferSizes::Read)));
+    while (_bufferQueue.Dequeue(queued))
+    {
+        std::size_t packetSize = queued->Buffer.GetActiveSize();
+        if (queued->Encrypt)
+            _crypt.EncryptSend(queued->Buffer.GetReadPointer(), packetSize);
+
+        if (buffer.GetRemainingSpace() < packetSize)
+        {
+            QueuePacket(std::move(buffer));
+            buffer.Resize(std::size_t(BufferSizes::Read));
+        }
+
+        if (buffer.GetRemainingSpace() >= packetSize)
+            buffer.Write(queued->Buffer.GetReadPointer(), packetSize);
+        else    // single packet larger than 16384 bytes - client will reject.
+            QueuePacket(std::move(queued->Buffer));
+
+        delete queued;
+    }
+
+    if (buffer.GetActiveSize() > 0)
+        QueuePacket(std::move(buffer));
+
     if (!BattlenetSocket::Update())
         return false;
 
     if (_queryFuture.valid() && _queryFuture.wait_for(std::chrono::seconds(0)) == std::future_status::ready)
     {
-        auto callback = std::move(_queryCallback);
+        auto callback = _queryCallback;
         _queryCallback = nullptr;
         callback(_queryFuture.get());
     }
@@ -679,15 +704,12 @@ void Battlenet::Session::AsyncWrite(ServerPacket* packet)
 
     packet->Write();
 
-    MessageBuffer buffer;
-    buffer.Write(packet->GetData(), packet->GetSize());
+    EncryptableBuffer* buffer = new EncryptableBuffer();
+    buffer->Buffer.Write(packet->GetData(), packet->GetSize());
+    buffer->Encrypt = _crypt.IsInitialized();
     delete packet;
 
-    std::unique_lock<std::mutex> guard(_writeLock);
-
-    _crypt.EncryptSend(buffer.GetReadPointer(), buffer.GetActiveSize());
-
-    QueuePacket(std::move(buffer), guard);
+    _bufferQueue.Enqueue(buffer);
 }
 
 inline void ReplaceResponse(Battlenet::ServerPacket** oldResponse, Battlenet::ServerPacket* newResponse)

src/server/bnetserver/Server/Session.h

@@ -23,6 +23,7 @@
 #include "Socket.h"
 #include "BigNumber.h"
 #include "Callback.h"
+#include "MPSCQueue.h"
 #include <memory>
 #include <boost/asio/ip/tcp.hpp>
 
@@ -174,6 +175,13 @@ namespace Battlenet
 
         std::queue<ModuleType> _modulesWaitingForData;
 
+        struct EncryptableBuffer
+        {
+            MessageBuffer Buffer;
+            bool Encrypt;
+        };
+
+        MPSCQueue<EncryptableBuffer> _bufferQueue;
         PacketCrypt _crypt;
         bool _authed;
         bool _subscribedToRealmListUpdates;

src/server/bnetserver/Server/SessionManager.cpp

@@ -22,7 +22,8 @@ bool Battlenet::SessionManager::StartNetwork(boost::asio::io_service& service, s
     if (!BaseSocketMgr::StartNetwork(service, bindIp, port))
         return false;
 
-    _acceptor->AsyncAcceptManaged(&OnSocketAccept);
+    _acceptor->SetSocketFactory(std::bind(&BaseSocketMgr::GetSocketForAccept, this));
+    _acceptor->AsyncAcceptWithCallback<&OnSocketAccept>();
     return true;
 }
 
@@ -31,9 +32,9 @@ NetworkThread<Battlenet::Session>* Battlenet::SessionManager::CreateThreads() co
     return new NetworkThread<Session>[GetNetworkThreadCount()];
 }
 
-void Battlenet::SessionManager::OnSocketAccept(tcp::socket&& sock)
+void Battlenet::SessionManager::OnSocketAccept(tcp::socket&& sock, uint32 threadIndex)
 {
-    sSessionMgr.OnSocketOpen(std::forward<tcp::socket>(sock));
+    sSessionMgr.OnSocketOpen(std::forward<tcp::socket>(sock), threadIndex);
 }
 
 void Battlenet::SessionManager::AddSession(Session* session)

src/server/bnetserver/Server/SessionManager.h

@@ -75,7 +75,7 @@ namespace Battlenet
         NetworkThread<Session>* CreateThreads() const override;
 
     private:
-        static void OnSocketAccept(tcp::socket&& sock);
+        static void OnSocketAccept(tcp::socket&& sock, uint32 threadIndex);
 
         SessionMap _sessions;
         SessionByAccountMap _sessionsByAccountId;

src/server/game/Server/WorldSocket.cpp

@@ -38,6 +38,17 @@ struct CompressedWorldPacket
     uint32 CompressedAdler;
 };
 
+class EncryptablePacket : public WorldPacket
+{
+public:
+    EncryptablePacket(WorldPacket const& packet, bool encrypt) : WorldPacket(packet), _encrypt(encrypt) { }
+
+    bool NeedsEncryption() const { return _encrypt; }
+
+private:
+    bool _encrypt;
+};
+
 #pragma pack(pop)
 
 using boost::asio::ip::tcp;
@@ -76,11 +87,8 @@ void WorldSocket::Start()
     stmt->setString(0, ip_address);
     stmt->setUInt32(1, inet_addr(ip_address.c_str()));
 
-    {
-        std::lock_guard<std::mutex> guard(_queryLock);
-        _queryCallback = io_service().wrap(std::bind(&WorldSocket::CheckIpCallback, this, std::placeholders::_1));
-        _queryFuture = LoginDatabase.AsyncQuery(stmt);
-    }
+    _queryCallback = std::bind(&WorldSocket::CheckIpCallback, this, std::placeholders::_1);
+    _queryFuture = LoginDatabase.AsyncQuery(stmt);
 }
 
 void WorldSocket::CheckIpCallback(PreparedQueryResult result)
@@ -116,23 +124,50 @@ void WorldSocket::CheckIpCallback(PreparedQueryResult result)
     initializer.Write(&header, sizeof(header.Setup.Size));
     initializer.Write(ServerConnectionInitialize.c_str(), ServerConnectionInitialize.length());
 
-    std::unique_lock<std::mutex> guard(_writeLock);
-    QueuePacket(std::move(initializer), guard);
+    // - io_service.run thread, safe.
+    QueuePacket(std::move(initializer));
 }
 
 bool WorldSocket::Update()
 {
+    EncryptablePacket* queued;
+    MessageBuffer buffer;
+    while (_bufferQueue.Dequeue(queued))
+    {
+        uint32 sizeOfHeader = SizeOfServerHeader[queued->NeedsEncryption()];
+        uint32 packetSize = queued->size();
+        if (packetSize > MinSizeForCompression && queued->NeedsEncryption())
+            packetSize = compressBound(packetSize) + sizeof(CompressedWorldPacket);
+
+        if (buffer.GetRemainingSpace() < packetSize + sizeOfHeader)
+        {
+            QueuePacket(std::move(buffer));
+            buffer.Resize(4096);
+        }
+
+        if (buffer.GetRemainingSpace() >= packetSize + sizeOfHeader)
+            WritePacketToBuffer(*queued, buffer);
+        else    // single packet larger than 4096 bytes
+        {
+            MessageBuffer packetBuffer(packetSize + sizeOfHeader);
+            WritePacketToBuffer(*queued, packetBuffer);
+            QueuePacket(std::move(packetBuffer));
+        }
+
+        delete queued;
+    }
+
+    if (buffer.GetActiveSize() > 0)
+        QueuePacket(std::move(buffer));
+
     if (!BaseSocket::Update())
         return false;
 
+    if (_queryFuture.valid() && _queryFuture.wait_for(std::chrono::seconds(0)) == std::future_status::ready)
     {
-        std::lock_guard<std::mutex> guard(_queryLock);
-        if (_queryFuture.valid() && _queryFuture.wait_for(std::chrono::seconds(0)) == std::future_status::ready)
-        {
-            auto callback = std::move(_queryCallback);
-            _queryCallback = nullptr;
-            callback(_queryFuture.get());
-        }
+        auto callback = _queryCallback;
+        _queryCallback = nullptr;
+        callback(_queryFuture.get());
     }
 
     return true;
@@ -428,37 +463,21 @@ void WorldSocket::SendPacket(WorldPacket const& packet)
     if (sPacketLog->CanLogPacket())
         sPacketLog->LogPacket(packet, SERVER_TO_CLIENT, GetRemoteIpAddress(), GetRemotePort(), GetConnectionType());
 
-    uint32 packetSize = packet.size();
-    uint32 sizeOfHeader = SizeOfServerHeader[_authCrypt.IsInitialized()];
-    if (packetSize > MinSizeForCompression && _authCrypt.IsInitialized())
-        packetSize = compressBound(packetSize) + sizeof(CompressedWorldPacket);
-
-    std::unique_lock<std::mutex> guard(_writeLock);
-
-#ifndef TC_SOCKET_USE_IOCP
-    if (_writeQueue.empty() && _writeBuffer.GetRemainingSpace() >= sizeOfHeader + packetSize)
-        WritePacketToBuffer(packet, _writeBuffer);
-    else
-#endif
-    {
-        MessageBuffer buffer(sizeOfHeader + packetSize);
-        WritePacketToBuffer(packet, buffer);
-        QueuePacket(std::move(buffer), guard);
-    }
+    _bufferQueue.Enqueue(new EncryptablePacket(packet, _authCrypt.IsInitialized()));
 }
 
-void WorldSocket::WritePacketToBuffer(WorldPacket const& packet, MessageBuffer& buffer)
+void WorldSocket::WritePacketToBuffer(EncryptablePacket const& packet, MessageBuffer& buffer)
 {
     ServerPktHeader header;
-    uint32 sizeOfHeader = SizeOfServerHeader[_authCrypt.IsInitialized()];
+    uint32 sizeOfHeader = SizeOfServerHeader[packet.NeedsEncryption()];
     uint32 opcode = packet.GetOpcode();
     uint32 packetSize = packet.size();
 
     // Reserve space for buffer
     uint8* headerPos = buffer.GetWritePointer();
     buffer.WriteCompleted(sizeOfHeader);
 
-    if (packetSize > MinSizeForCompression && _authCrypt.IsInitialized())
+    if (packetSize > MinSizeForCompression && packet.NeedsEncryption())
     {
         CompressedWorldPacket cmp;
         cmp.UncompressedSize = packetSize + 4;
@@ -481,7 +500,7 @@ void WorldSocket::WritePacketToBuffer(WorldPacket const& packet, MessageBuffer&
     else if (!packet.empty())
         buffer.Write(packet.contents(), packet.size());
 
-    if (_authCrypt.IsInitialized())
+    if (packet.NeedsEncryption())
     {
         header.Normal.Size = packetSize;
         header.Normal.Command = opcode;
@@ -598,11 +617,8 @@ void WorldSocket::HandleAuthSession(std::shared_ptr<WorldPackets::Auth::AuthSess
     stmt->setInt32(0, int32(realm.Id.Realm));
     stmt->setString(1, authSession->Account);
 
-    {
-        std::lock_guard<std::mutex> guard(_queryLock);
-        _queryCallback = io_service().wrap(std::bind(&WorldSocket::HandleAuthSessionCallback, this, authSession, std::placeholders::_1));
-        _queryFuture = LoginDatabase.AsyncQuery(stmt);
-    }
+    _queryCallback = std::bind(&WorldSocket::HandleAuthSessionCallback, this, authSession, std::placeholders::_1);
+    _queryFuture = LoginDatabase.AsyncQuery(stmt);
 }
 
 void WorldSocket::HandleAuthSessionCallback(std::shared_ptr<WorldPackets::Auth::AuthSession> authSession, PreparedQueryResult result)
@@ -768,7 +784,7 @@ void WorldSocket::HandleAuthSessionCallback(std::shared_ptr<WorldPackets::Auth::
     if (wardenActive)
         _worldSession->InitWarden(&account.Game.SessionKey, account.BattleNet.OS);
 
-    _queryCallback = io_service().wrap(std::bind(&WorldSocket::LoadSessionPermissionsCallback, this, std::placeholders::_1));
+    _queryCallback = std::bind(&WorldSocket::LoadSessionPermissionsCallback, this, std::placeholders::_1);
     _queryFuture = _worldSession->LoadPermissionsAsync();
     AsyncRead();
 }
@@ -801,11 +817,8 @@ void WorldSocket::HandleAuthContinuedSession(std::shared_ptr<WorldPackets::Auth:
     PreparedStatement* stmt = LoginDatabase.GetPreparedStatement(LOGIN_SEL_ACCOUNT_INFO_CONTINUED_SESSION);
     stmt->setUInt32(0, accountId);
 
-    {
-        std::lock_guard<std::mutex> guard(_queryLock);
-        _queryCallback = io_service().wrap(std::bind(&WorldSocket::HandleAuthContinuedSessionCallback, this, authSession, std::placeholders::_1));
-        _queryFuture = LoginDatabase.AsyncQuery(stmt);
-    }
+    _queryCallback = std::bind(&WorldSocket::HandleAuthContinuedSessionCallback, this, authSession, std::placeholders::_1);
+    _queryFuture = LoginDatabase.AsyncQuery(stmt);
 }
 
 void WorldSocket::HandleAuthContinuedSessionCallback(std::shared_ptr<WorldPackets::Auth::AuthContinuedSession> authSession, PreparedQueryResult result)