Core/Socket: Fixed race conditions with instance socket

Helgrind logs:
==41785== Possible data race during write of size 8 at 0x6D80D590 by thread #4
==41785== Locks held: none
==41785==    at 0x19603A2: void std::swap<std::__future_base::_State_baseV2*>(std::__future_base::_State_baseV2*&, std::__future_base::_State_baseV2*&) (move.h:177)
==41785==    by 0x195FD7C: std::__shared_ptr<std::__future_base::_State_baseV2, (__gnu_cxx::_Lock_policy)2>::swap(std::__shared_ptr<std::__future_base::_State_baseV2, (__gnu_cxx::_Lock_policy)2>&) (shared_ptr_base.h:1069)
==41785==    by 0x1FC5B84: std::__basic_future<SQLQueryHolder*>::_M_swap(std::__basic_future<SQLQueryHolder*>&) (future:629)
==41785==    by 0x1FC5155: std::future<SQLQueryHolder*>::operator=(std::future<SQLQueryHolder*>&&) (future:691)
==41785==    by 0x1FB94C6: WorldSession::HandleContinuePlayerLogin() (CharacterHandler.cpp:832)
==41785==    by 0x1ABD31A: WorldSocket::HandleAuthContinuedSession(WorldPackets::Auth::AuthContinuedSession&) (WorldSocket.cpp:742)
==41785==    by 0x1AB8322: WorldSocket::ReadDataHandler() (WorldSocket.cpp:261)
==41785==    by 0x1AB77D1: WorldSocket::ReadHandler() (WorldSocket.cpp:157)
==41785==    by 0x1AC068A: Socket<WorldSocket>::ReadHandlerInternal(boost::system::error_code, unsigned long) (Socket.h:201)
==41785==    by 0x1AC39ED: void std::_Mem_fn<void (Socket<WorldSocket>::*)(boost::system::error_code, unsigned long)>::_M_call<std::shared_ptr<WorldSocket>&, boost::system::error_code const&, unsigned long const&>(std::shared_ptr<WorldSocket>&, void const volatile*, boost::system::error_code const&, unsigned long const&) const (in /usr/local/bin/worldserver)
==41785==    by 0x1AC393F: void std::_Mem_fn<void (Socket<WorldSocket>::*)(boost::system::error_code, unsigned long)>::operator()<std::shared_ptr<WorldSocket>&, boost::system::error_code const&, unsigned long const&, void>(std::shared_ptr<WorldSocket>&, boost::system::error_code const&, unsigned long const&) const (functional:578)
==41785==    by 0x1AC370F: void std::_Bind<std::_Mem_fn<void (Socket<WorldSocket>::*)(boost::system::error_code, unsigned long)> (std::shared_ptr<WorldSocket>, std::_Placeholder<1>, std::_Placeholder<2>)>::__call<void, boost::system::error_code const&, unsigned long const&, 0ul, 1ul, 2ul>(std::tuple<boost::system::error_code const&, unsigned long const&>&&, std::_Index_tuple<0ul, 1ul, 2ul>) (functional:1264)
==41785==
==41785== This conflicts with a previous read of size 8 by thread #1
==41785== Locks held: none
==41785==    at 0x144BF6A: std::__shared_ptr<std::__future_base::_State_baseV2, (__gnu_cxx::_Lock_policy)2>::operator bool() const (shared_ptr_base.h:1056)
==41785==    by 0x1AAF7BD: std::__basic_future<SQLQueryHolder*>::valid() const (future:590)
==41785==    by 0x1AAA04C: WorldSession::ProcessQueryCallbacks() (WorldSession.cpp:1006)
==41785==    by 0x1AA5167: WorldSession::Update(unsigned int, PacketFilter&) (WorldSession.cpp:452)
==41785==    by 0x1D2D667: World::UpdateSessions(unsigned int) (World.cpp:2793)
==41785==    by 0x1D2A8C6: World::Update(unsigned int) (World.cpp:2159)
==41785==    by 0x145B823: WorldUpdateLoop() (Main.cpp:397)
==41785==    by 0x145957F: main (Main.cpp:255)

==41785== Possible data race during read of size 8 at 0x6D80D638 by thread #1
==41785== Locks held: none
==41785==    at 0x1A5E914: std::__shared_ptr<WorldSocket, (__gnu_cxx::_Lock_policy)2>::operator bool() const (shared_ptr_base.h:1056)
==41785==    by 0x1AA52A9: WorldSession::Update(unsigned int, PacketFilter&) (WorldSession.cpp:468)
==41785==    by 0x1D2D667: World::UpdateSessions(unsigned int) (World.cpp:2793)
==41785==    by 0x1D2A8C6: World::Update(unsigned int) (World.cpp:2159)
==41785==    by 0x145B823: WorldUpdateLoop() (Main.cpp:397)
==41785==    by 0x145957F: main (Main.cpp:255)
==41785==
==41785== This conflicts with a previous write of size 8 by thread #4
==41785== Locks held: none
==41785==    at 0x1AAD12D: std::__shared_ptr<WorldSocket, (__gnu_cxx::_Lock_policy)2>::operator=(std::__shared_ptr<WorldSocket, (__gnu_cxx::_Lock_policy)2> const&) (shared_ptr_base.h:860)
==41785==    by 0x1AAD174: std::shared_ptr<WorldSocket>::operator=(std::shared_ptr<WorldSocket> const&) (shared_ptr.h:93)
==41785==    by 0x1ABF0B5: WorldSession::AddInstanceConnection(std::shared_ptr<WorldSocket>) (WorldSession.h:684)
==41785==    by 0x1ABD2F8: WorldSocket::HandleAuthContinuedSession(WorldPackets::Auth::AuthContinuedSession&) (WorldSocket.cpp:741)
==41785==    by 0x1AB8322: WorldSocket::ReadDataHandler() (WorldSocket.cpp:261)
==41785==    by 0x1AB77D1: WorldSocket::ReadHandler() (WorldSocket.cpp:157)
==41785==    by 0x1AC068A: Socket<WorldSocket>::ReadHandlerInternal(boost::system::error_code, unsigned long) (Socket.h:201)
==41785==    by 0x1AC39ED: void std::_Mem_fn<void (Socket<WorldSocket>::*)(boost::system::error_code, unsigned long)>::_M_call<std::shared_ptr<WorldSocket>&, boost::system::error_code const&, unsigned long const&>(std::shared_ptr<WorldSocket>&, void const volatile*, boost::system::error_code const&, unsigned long const&) const (in /usr/local/bin/worldserver)

7 files changed, 44 insertions, 20 deletions

diff --git a/src/server/game/Handlers/CharacterHandler.cpp b/src/server/game/Handlers/CharacterHandler.cpp
index bcad74d50fd..e79cc992725 100644
--- a/src/server/game/Handlers/CharacterHandler.cpp
+++ b/src/server/game/Handlers/CharacterHandler.cpp
@@ -829,6 +829,8 @@ void WorldSession::HandleContinuePlayerLogin()
         return;
     }
 
+    SendPacket(WorldPackets::Auth::ResumeComms(CONNECTION_TYPE_INSTANCE).Write());
+
     _charLoginCallback = CharacterDatabase.DelayQueryHolder(holder);
 }
 
diff --git a/src/server/game/Server/Packet.h b/src/server/game/Server/Packet.h
index bf538803ad6..85d65e967be 100644
--- a/src/server/game/Server/Packet.h
+++ b/src/server/game/Server/Packet.h
@@ -45,7 +45,7 @@ namespace WorldPackets
     class ServerPacket : public Packet
     {
     public:
-        ServerPacket(OpcodeServer opcode, size_t initialSize = 200) : Packet(WorldPacket(opcode, initialSize)) { }
+        ServerPacket(OpcodeServer opcode, size_t initialSize = 200, ConnectionType connection = CONNECTION_TYPE_DEFAULT) : Packet(WorldPacket(opcode, initialSize, connection)) { }
 
         void Read() override final { ASSERT(!"Read not implemented for server packets."); }
 
diff --git a/src/server/game/Server/Packets/AuthenticationPackets.h b/src/server/game/Server/Packets/AuthenticationPackets.h
index eae0cae34f1..f914e18f6bc 100644
--- a/src/server/game/Server/Packets/AuthenticationPackets.h
+++ b/src/server/game/Server/Packets/AuthenticationPackets.h
@@ -198,7 +198,7 @@ namespace WorldPackets
         class ResumeComms final : public ServerPacket
         {
         public:
-            ResumeComms() : ServerPacket(SMSG_RESUME_COMMS, 0) { }
+            ResumeComms(ConnectionType connection) : ServerPacket(SMSG_RESUME_COMMS, 0, connection) { }
 
             WorldPacket const* Write() override { return &_worldPacket; }
         };
diff --git a/src/server/game/Server/WorldSocket.cpp b/src/server/game/Server/WorldSocket.cpp
index 2b2b2458993..949e783a3ba 100644
--- a/src/server/game/Server/WorldSocket.cpp
+++ b/src/server/game/Server/WorldSocket.cpp
@@ -182,6 +182,13 @@ void WorldSocket::ExtractOpcodeAndSize(ClientPktHeader const* header, uint32& op
     }
 }
 
+void WorldSocket::SetWorldSession(WorldSession* session)
+{
+    std::lock_guard<std::mutex> sessionGuard(_worldSessionLock);
+    _worldSession = session;
+    _authed = true;
+}
+
 bool WorldSocket::ReadHeaderHandler()
 {
     ASSERT(_headerBuffer.GetActiveSize() == SizeOfClientHeader[_initialized][_authCrypt.IsInitialized()], "Header size " SZFMTD " different than expected %u", _headerBuffer.GetActiveSize(), SizeOfClientHeader[_initialized][_authCrypt.IsInitialized()]);
@@ -724,22 +731,7 @@ void WorldSocket::HandleAuthContinuedSession(WorldPackets::Auth::AuthContinuedSe
         return;
     }
 
-    _worldSession = sWorld->FindSession(accountId);
-    if (!_worldSession)
-    {
-        SendAuthResponseError(AUTH_SESSION_EXPIRED);
-        TC_LOG_ERROR("network", "WorldSocket::HandleAuthContinuedSession: No active session found for account: %u ('%s') address: %s", accountId, login.c_str(), GetRemoteIpAddress().to_string().c_str());
-        DelayedCloseSocket();
-        return;
-    }
-
-    _authed = true;
-
-    WorldPackets::Auth::ResumeComms resumeComms;
-    SendPacketAndLogOpcode(*resumeComms.Write());
-
-    _worldSession->AddInstanceConnection(shared_from_this());
-    _worldSession->HandleContinuePlayerLogin();
+    sWorld->AddInstanceSocket(shared_from_this(), accountId);
 }
 
 void WorldSocket::HandleConnectToFailed(WorldPackets::Auth::ConnectToFailed& connectToFailed)
diff --git a/src/server/game/Server/WorldSocket.h b/src/server/game/Server/WorldSocket.h
index 482f794344b..e42a310f470 100644
--- a/src/server/game/Server/WorldSocket.h
+++ b/src/server/game/Server/WorldSocket.h
@@ -85,12 +85,14 @@ public:
 
     ConnectionType GetConnectionType() const { return _type; }
 
+    void SendAuthResponseError(uint8 code);
+    void SetWorldSession(WorldSession* session);
+
 protected:
     void OnClose() override;
     void ReadHandler() override;
     bool ReadHeaderHandler();
     bool ReadDataHandler();
-
 private:
     /// writes network.opcode log
     /// accessing WorldSession is not threadsafe, only do it when holding _worldSessionLock
@@ -104,7 +106,6 @@ private:
     void HandleAuthSession(WorldPackets::Auth::AuthSession& authSession);
     void HandleAuthContinuedSession(WorldPackets::Auth::AuthContinuedSession& authSession);
     void HandleConnectToFailed(WorldPackets::Auth::ConnectToFailed& connectToFailed);
-    void SendAuthResponseError(uint8 code);
 
     bool HandlePing(WorldPacket& recvPacket);
 
diff --git a/src/server/game/World/World.cpp b/src/server/game/World/World.cpp
index cdbb15bf539..7010982dc2b 100644
--- a/src/server/game/World/World.cpp
+++ b/src/server/game/World/World.cpp
@@ -66,6 +66,7 @@
 #include "WeatherMgr.h"
 #include "WorldSession.h"
 #include "ChatPackets.h"
+#include "WorldSocket.h"
 
 #include <boost/algorithm/string.hpp>
 
@@ -223,6 +224,11 @@ void World::AddSession(WorldSession* s)
     addSessQueue.add(s);
 }
 
+void World::AddInstanceSocket(std::shared_ptr<WorldSocket> sock, uint32 sessionAccountId)
+{
+    _linkSocketQueue.add(std::make_pair(sock, sessionAccountId));
+}
+
 void World::AddSession_(WorldSession* s)
 {
     ASSERT(s);
@@ -296,6 +302,21 @@ void World::AddSession_(WorldSession* s)
     }
 }
 
+void World::ProcessLinkInstanceSocket(std::pair<std::shared_ptr<WorldSocket>, uint32> linkInfo)
+{
+    WorldSession* session = FindSession(linkInfo.second);
+    if (!session)
+    {
+        linkInfo.first->SendAuthResponseError(AUTH_SESSION_EXPIRED);
+        linkInfo.first->DelayedCloseSocket();
+        return;
+    }
+
+    linkInfo.first->SetWorldSession(session);
+    session->AddInstanceConnection(linkInfo.first);
+    session->HandleContinuePlayerLogin();
+}
+
 bool World::HasRecentlyDisconnected(WorldSession* session)
 {
     if (!session)
@@ -2775,6 +2796,10 @@ void World::SendServerMessage(ServerMessageType type, const char *text, Player*
 
 void World::UpdateSessions(uint32 diff)
 {
+    std::pair<std::shared_ptr<WorldSocket>, uint32> linkInfo;
+    while (_linkSocketQueue.next(linkInfo))
+        ProcessLinkInstanceSocket(std::move(linkInfo));
+
     ///- Add new sessions
     WorldSession* sess = NULL;
     while (addSessQueue.next(sess))
diff --git a/src/server/game/World/World.h b/src/server/game/World/World.h
index 78243b19374..b3bebca08e3 100644
--- a/src/server/game/World/World.h
+++ b/src/server/game/World/World.h
@@ -575,6 +575,7 @@ class World
 
         WorldSession* FindSession(uint32 id) const;
         void AddSession(WorldSession* s);
+        void AddInstanceSocket(std::shared_ptr<WorldSocket> sock, uint32 sessionAccountId);
         void SendAutoBroadcast();
         bool RemoveSession(uint32 id);
         /// Get the number of current active sessions
@@ -883,6 +884,9 @@ class World
         void AddSession_(WorldSession* s);
         LockedQueue<WorldSession*> addSessQueue;
 
+        void ProcessLinkInstanceSocket(std::pair<std::shared_ptr<WorldSocket>, uint32> linkInfo);
+        LockedQueue<std::pair<std::shared_ptr<WorldSocket>, uint32>> _linkSocketQueue;
+
         // used versions
         std::string m_DBVersion;