diff --git a/src/net/common/clientcontext.cpp b/src/net/common/clientcontext.cpp index 427c18ee..26e7d585 100644 --- a/src/net/common/clientcontext.cpp +++ b/src/net/common/clientcontext.cpp @@ -43,10 +43,10 @@ ClientContext::ClientContext() m_hasSubscribedLobbyMsg(true) { bzero(&m_clientSockaddr, sizeof(m_clientSockaddr)); + m_ioService.reset(new boost::asio::io_service()); m_senderCallback.reset(new ClientSenderCallback()); - m_senderThread.reset(new SenderThread(*m_senderCallback)); + m_senderThread.reset(new SenderThread(*m_senderCallback, m_ioService)); m_senderThread->Start(); - m_ioService = dynamic_cast(m_senderThread.get())->GetIOService(); } ClientContext::~ClientContext() diff --git a/src/net/common/senderthread.cpp b/src/net/common/senderthread.cpp index 6046d8e4..402ed4e9 100644 --- a/src/net/common/senderthread.cpp +++ b/src/net/common/senderthread.cpp @@ -43,8 +43,8 @@ typedef std::list > SendDataList; class SendDataManager : public boost::enable_shared_from_this { public: - SendDataManager(boost::shared_ptr s) - : session(s), writeInProgress(false) + SendDataManager(boost::shared_ptr s) + : socket(s), writeInProgress(false) { } @@ -52,7 +52,7 @@ class SendDataManager : public boost::enable_shared_from_this void AsyncSendNextPacket(bool handlerMode = false); - boost::shared_ptr session; + boost::shared_ptr socket; mutable boost::mutex dataMutex; SendDataList list; @@ -79,7 +79,7 @@ SendDataManager::AsyncSendNextPacket(bool handlerMode) { boost::shared_ptr nextPacket = list.front(); boost::asio::async_write( - *session->GetAsioSocket(), + *socket, boost::asio::buffer(nextPacket->GetRawData(), nextPacket->GetLen()), boost::bind(&SendDataManager::HandleWrite, shared_from_this(), @@ -91,8 +91,8 @@ SendDataManager::AsyncSendNextPacket(bool handlerMode) } } -SenderThread::SenderThread(SenderCallback &cb) -: m_callback(cb) +SenderThread::SenderThread(SenderCallback &cb, boost::shared_ptr ioService) +: m_callback(cb), m_ioService(ioService) { } @@ -103,9 +103,7 @@ SenderThread::~SenderThread() void SenderThread::Start() { - m_ioServiceBarrier.reset(new boost::barrier(2)); Run(); - m_ioServiceBarrier->wait(); } void @@ -131,7 +129,7 @@ SenderThread::Send(boost::shared_ptr session, boost::shared_ptrGetId()); if (pos == m_sendQueueMap.end()) - pos = m_sendQueueMap.insert(SendQueueMap::value_type(session->GetId(), boost::shared_ptr(new SendDataManager(session)))).first; + pos = m_sendQueueMap.insert(SendQueueMap::value_type(session->GetId(), boost::shared_ptr(new SendDataManager(session->GetAsioSocket())))).first; tmpManager = pos->second; } { @@ -143,7 +141,7 @@ SenderThread::Send(boost::shared_ptr session, boost::shared_ptrsession->GetId()); + m_changedSessions.push_back(session->GetId()); } } } @@ -159,7 +157,7 @@ SenderThread::Send(boost::shared_ptr session, const NetPacketList & boost::mutex::scoped_lock lock(m_sendQueueMapMutex); SendQueueMap::iterator pos = m_sendQueueMap.find(session->GetId()); if (pos == m_sendQueueMap.end()) - pos = m_sendQueueMap.insert(SendQueueMap::value_type(session->GetId(), boost::shared_ptr(new SendDataManager(session)))).first; + pos = m_sendQueueMap.insert(SendQueueMap::value_type(session->GetId(), boost::shared_ptr(new SendDataManager(session->GetAsioSocket())))).first; tmpManager = pos->second; } { @@ -179,25 +177,52 @@ SenderThread::Send(boost::shared_ptr session, const NetPacketList & { // Third: Update notification list. boost::mutex::scoped_lock lock(m_changedSessionsMutex); - m_changedSessions.push_back(tmpManager->session->GetId()); + m_changedSessions.push_back(session->GetId()); } } } -boost::shared_ptr -SenderThread::GetIOService() +void +SenderThread::SignalSessionTerminated(unsigned sessionId) { - return m_ioService; + boost::mutex::scoped_lock lock(m_removedSessionsMutex); + m_removedSessions.push_back(sessionId); } void SenderThread::Main() { - m_ioService.reset(new boost::asio::io_service()); - m_ioServiceBarrier->wait(); boost::asio::io_service::work ioWork(*m_ioService); while (!ShouldTerminate()) { + // Close sessions if they were destructed. + { + boost::mutex::scoped_lock lock(m_removedSessionsMutex); + if (!m_removedSessions.empty()) + { + SessionIdList newRemovedSessions; + SessionIdList::iterator i = m_removedSessions.begin(); + SessionIdList::iterator end = m_removedSessions.end(); + + boost::mutex::scoped_lock lock(m_sendQueueMapMutex); + + while (i != end) + { + SendQueueMap::iterator pos = m_sendQueueMap.find(*i); + if (pos != m_sendQueueMap.end()) + { + // Remove session if no write is in progress, else wait. + if (!pos->second->writeInProgress) + m_sendQueueMap.erase(pos); + else + newRemovedSessions.push_back(*i); + } + ++i; + } + m_removedSessions = newRemovedSessions; + } + } + // Iterate through all changed sessions, and send data if needed. bool sessionValid; do { diff --git a/src/net/common/serverlobbythread.cpp b/src/net/common/serverlobbythread.cpp index 9907343c..ac886a72 100644 --- a/src/net/common/serverlobbythread.cpp +++ b/src/net/common/serverlobbythread.cpp @@ -69,8 +69,9 @@ public: // A serious send error should trigger a read error or a read // returning 0 afterwards, and we will handle this error. } - virtual void SignalSessionTerminated(unsigned /*session*/) + virtual void SignalSessionTerminated(unsigned session) { + m_server.GetSender(); } private: @@ -83,8 +84,9 @@ ServerLobbyThread::ServerLobbyThread(GuiInterface &gui, ConfigFile *playerConfig m_curGameId(0), m_curUniquePlayerId(0), m_curSessionId(INVALID_SESSION + 1), m_statDataChanged(false), m_startTime(boost::posix_time::second_clock::local_time()) { + m_ioService.reset(new boost::asio::io_service()); m_senderCallback.reset(new ServerSenderCallback(*this)); - m_sender.reset(new SenderThread(*m_senderCallback)); + m_sender.reset(new SenderThread(*m_senderCallback, m_ioService)); m_receiver.reset(new ReceiverHelper); } @@ -323,6 +325,13 @@ ServerLobbyThread::GetStartTime() const return m_startTime; } +SenderInterface & +ServerLobbyThread::GetSender() +{ + assert(m_sender); + return *m_sender; +} + u_int32_t ServerLobbyThread::GetNextUniquePlayerId() { @@ -350,7 +359,6 @@ ServerLobbyThread::Main() try { m_sender->Start(); - m_ioService = dynamic_cast(m_sender.get())->GetIOService(); while (!ShouldTerminate()) { diff --git a/src/net/senderinterface.h b/src/net/senderinterface.h index 38609c55..01d94e10 100644 --- a/src/net/senderinterface.h +++ b/src/net/senderinterface.h @@ -36,6 +36,8 @@ public: virtual void Send(boost::shared_ptr session, boost::shared_ptr packet) = 0; virtual void Send(boost::shared_ptr session, const NetPacketList &packetList) = 0; + + virtual void SignalSessionTerminated(unsigned sessionId) = 0; }; #endif diff --git a/src/net/senderthread.h b/src/net/senderthread.h index 490abaa8..8261d348 100644 --- a/src/net/senderthread.h +++ b/src/net/senderthread.h @@ -36,7 +36,7 @@ class SendDataManager; class SenderThread : public Thread, public SenderInterface { public: - SenderThread(SenderCallback &cb); + SenderThread(SenderCallback &cb, boost::shared_ptr ioService); virtual ~SenderThread(); virtual void Start(); @@ -46,10 +46,10 @@ public: virtual void Send(boost::shared_ptr session, boost::shared_ptr packet); virtual void Send(boost::shared_ptr session, const NetPacketList &packetList); - boost::shared_ptr GetIOService(); + virtual void SignalSessionTerminated(unsigned sessionId); protected: - typedef std::list ChangedSessionList; + typedef std::list SessionIdList; typedef std::map > SendQueueMap; @@ -61,13 +61,14 @@ private: SendQueueMap m_sendQueueMap; mutable boost::mutex m_sendQueueMapMutex; - ChangedSessionList m_changedSessions; + SessionIdList m_changedSessions; mutable boost::mutex m_changedSessionsMutex; + SessionIdList m_removedSessions; + mutable boost::mutex m_removedSessionsMutex; + SenderCallback &m_callback; boost::shared_ptr m_ioService; - - mutable boost::shared_ptr m_ioServiceBarrier; }; #endif diff --git a/src/net/serverlobbythread.h b/src/net/serverlobbythread.h index 2b4c5140..3a081dbc 100644 --- a/src/net/serverlobbythread.h +++ b/src/net/serverlobbythread.h @@ -87,6 +87,8 @@ public: ServerStats GetStats() const; boost::posix_time::ptime GetStartTime() const; + SenderInterface &GetSender(); + protected: typedef std::deque > ConnectQueue;