From 17377f5a6b3d970680802acc70c092c1605a6ed9 Mon Sep 17 00:00:00 2001 From: lotodore Date: Sun, 25 Jan 2009 20:19:09 +0000 Subject: [PATCH] New boost::asio based sender thread. Does not work on windows yet, because the completion routine is never called. No idea why not. --- src/net/common/senderthread.cpp | 153 +++++++++----------------- src/net/common/serveracceptthread.cpp | 5 + src/net/common/serverlobbythread.cpp | 47 ++------ src/net/senderthread.h | 59 ++++++++-- src/net/serverlobbythread.h | 9 +- 5 files changed, 116 insertions(+), 157 deletions(-) diff --git a/src/net/common/senderthread.cpp b/src/net/common/senderthread.cpp index da14ebc2..7a297787 100644 --- a/src/net/common/senderthread.cpp +++ b/src/net/common/senderthread.cpp @@ -26,16 +26,29 @@ #include #include +#include using namespace std; +using boost::asio::ip::tcp; + #define SEND_ERROR_TIMEOUT_MSEC 20000 #define SEND_TIMEOUT_MSEC 10 #define SEND_QUEUE_SIZE 10000000 #define SEND_LOG_INTERVAL_SEC 60 + + +void +SenderThread::SendDataManager::HandleWrite(const boost::system::error_code& error) +{ + SetWriteInProgress(false); + SetCompleted(true); +} + + SenderThread::SenderThread(SenderCallback &cb) -: m_bytesSent(0), m_callback(cb) +: m_callback(cb) { } @@ -66,18 +79,12 @@ SenderThread::Send(boost::shared_ptr session, boost::shared_ptrGetSocket(); - m_sessionId = session->GetId(); - } + 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, m_ioService)))).first; + if (pos->second->list.size() < SEND_QUEUE_SIZE) + pos->second->list.push_back(packet); } } @@ -86,22 +93,20 @@ SenderThread::Send(boost::shared_ptr session, const NetPacketList & { if (!packetList.empty() && session.get()) { - boost::mutex::scoped_lock lock(m_sendQueueMutex); - if (m_sendQueue.size() + packetList.size() <= SEND_QUEUE_SIZE) + 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, m_ioService)))).first; + if (pos->second->list.size() + packetList.size() <= SEND_QUEUE_SIZE) { NetPacketList::const_iterator i = packetList.begin(); NetPacketList::const_iterator end = packetList.end(); while (i != end) { - m_sendQueue.push_back(*i); + pos->second->list.push_back(*i); ++i; } } - { - boost::mutex::scoped_lock lock(m_sessionDataMutex); - m_sessionSocket = session->GetSocket(); - m_sessionId = session->GetId(); - } } } @@ -110,98 +115,40 @@ SenderThread::Main() { while (!ShouldTerminate()) { - if (!m_curPacket) { - boost::mutex::scoped_lock lock(m_sendQueueMutex); - if (!m_sendQueue.empty()) + boost::mutex::scoped_lock lock(m_sendQueueMapMutex); + SendQueueMap::iterator i = m_sendQueueMap.begin(); + SendQueueMap::iterator end = m_sendQueueMap.end(); + while (i != end) { - m_curPacket = m_sendQueue.front(); - m_sendQueue.pop_front(); - } - } - - if (m_curPacket) - { - const unsigned tmpLen = m_curPacket->GetLen(); - if (tmpLen > MAX_PACKET_SIZE) - m_curPacket.reset(); // TODO log - else - { - SOCKET tmpSocket; + SendQueueMap::iterator next = i; + ++next; + boost::shared_ptr tmpManager = i->second; + if (tmpManager->list.empty()) + m_sendQueueMap.erase(i); + else { - boost::mutex::scoped_lock lock(m_sessionDataMutex); - tmpSocket = m_sessionSocket; - } - - // send next chunk of data - int bytesSent = send(tmpSocket, ((const char *)m_curPacket->GetRawData()) + m_bytesSent, tmpLen - m_bytesSent, SOCKET_SEND_FLAGS); - - if (!IS_VALID_SEND(bytesSent)) - { - // Never assume that this is a fatal error. - int errCode = SOCKET_ERRNO(); - if (IS_SOCKET_ERR_WOULDBLOCK(errCode)) + if (!tmpManager->IsWriteInProgress()) { - fd_set writeSet; - struct timeval timeout; - - FD_ZERO(&writeSet); - FD_SET(tmpSocket, &writeSet); - - timeout.tv_sec = 0; - timeout.tv_usec = SEND_TIMEOUT_MSEC * 1000; - int selectResult = select(tmpSocket + 1, NULL, &writeSet, NULL, &timeout); - if (!IS_VALID_SELECT(selectResult)) + if (tmpManager->IsCompleted()) + tmpManager->list.pop_front(); + else { - // Never assume that this is a fatal error. - int errCode = SOCKET_ERRNO(); - if (!IS_SOCKET_ERR_WOULDBLOCK(errCode)) - { - // Skip this packet - this is bad, and is therefore reported. - // Ignore invalid or not connected sockets. - if (errCode != SOCKET_ERR_NOTCONN && errCode != SOCKET_ERR_NOTSOCK) - { - boost::mutex::scoped_lock lock(m_sessionDataMutex); - m_callback.SignalNetError(m_sessionId, ERR_SOCK_SELECT_FAILED, errCode); - } - } - Msleep(SEND_TIMEOUT_MSEC); + boost::shared_ptr tmpPacket = tmpManager->list.front(); + boost::asio::async_write( + *tmpManager->socket, + boost::asio::buffer(tmpPacket->GetRawData(), + tmpPacket->GetLen()), + boost::bind(&SendDataManager::HandleWrite, tmpManager, + boost::asio::placeholders::error)); + tmpManager->SetWriteInProgress(true); } } - else // other errors than would block - { - // Skip this packet - this is bad, and is therefore reported. - // Ignore invalid or not connected sockets. - if (errCode != SOCKET_ERR_NOTCONN && errCode != SOCKET_ERR_NOTSOCK) - { - boost::mutex::scoped_lock lock(m_sessionDataMutex); - m_callback.SignalNetError(m_sessionId, ERR_SOCK_SEND_FAILED, errCode); - } - Msleep(SEND_TIMEOUT_MSEC); - } - } - else if ((unsigned)bytesSent + m_bytesSent < tmpLen) - { - if (bytesSent) - { - // Send was partly successful. - m_bytesSent += bytesSent; - } - else - Msleep(SEND_TIMEOUT_MSEC); - } - else //if ((unsigned)bytesSent + m_bytesSent == tmpLen) - { - m_curPacket.reset(); - m_bytesSent = 0; } + i = next; } - } - else Msleep(SEND_TIMEOUT_MSEC); + } } - boost::mutex::scoped_lock lock(m_sessionDataMutex); - if (m_sessionSocket != INVALID_SOCKET) - CLOSESOCKET(m_sessionSocket); } diff --git a/src/net/common/serveracceptthread.cpp b/src/net/common/serveracceptthread.cpp index a19999a6..9daa4f5a 100644 --- a/src/net/common/serveracceptthread.cpp +++ b/src/net/common/serveracceptthread.cpp @@ -99,7 +99,12 @@ ServerAcceptThread::Listen() if (context.GetServerPort() < 1024) throw ServerException(__FILE__, __LINE__, ERR_SOCK_INVALID_PORT, 0); +#ifdef _WIN32 + context.SetSocket(WSASocket(context.GetAddrFamily(), SOCK_STREAM, context.GetProtocol(), 0, 0, WSA_FLAG_OVERLAPPED)); +#else context.SetSocket(socket(context.GetAddrFamily(), SOCK_STREAM, context.GetProtocol())); +#endif + if (!IS_VALID_SOCKET(context.GetSocket())) throw ServerException(__FILE__, __LINE__, ERR_SOCK_CREATION_FAILED, SOCKET_ERRNO()); diff --git a/src/net/common/serverlobbythread.cpp b/src/net/common/serverlobbythread.cpp index 1e6de971..534c6475 100644 --- a/src/net/common/serverlobbythread.cpp +++ b/src/net/common/serverlobbythread.cpp @@ -69,9 +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.RemoveSender(session); + // Nothing to do, since we only have one sender thread. } private: @@ -85,6 +85,7 @@ ServerLobbyThread::ServerLobbyThread(GuiInterface &gui, ConfigFile *playerConfig m_statDataChanged(false), m_startTime(boost::posix_time::second_clock::local_time()) { m_senderCallback.reset(new ServerSenderCallback(*this)); + m_sender.reset(new SenderThread(*m_senderCallback)); m_receiver.reset(new ReceiverHelper); } @@ -304,13 +305,6 @@ ServerLobbyThread::RemoveGame(unsigned id) m_removeGameList.push_back(id); } -void -ServerLobbyThread::RemoveSender(unsigned session) -{ - boost::mutex::scoped_lock lock(m_removeSenderListMutex); - m_removeSenderList.push_back(session); -} - AvatarManager & ServerLobbyThread::GetAvatarManager() { @@ -356,6 +350,8 @@ ServerLobbyThread::Main() { try { + m_sender->Start(); + while (!ShouldTerminate()) { // Process new connections. @@ -368,8 +364,6 @@ ServerLobbyThread::Main() RemoveGameLoop(); // Kick players. RemovePlayerLoop(); - // Remove sender threads. - RemoveSenderLoop(); // Resubscribe Lobby Messages if needed. ResubscribeLobbyMsgLoop(); // Check session timeouts. @@ -388,6 +382,8 @@ ServerLobbyThread::Main() } TerminateGames(); + m_sender->SignalStop(); + m_sender->WaitStop(); CleanupConnectQueue(); } @@ -875,30 +871,6 @@ ServerLobbyThread::RemovePlayerLoop() } } -void -ServerLobbyThread::RemoveSenderLoop() -{ - boost::mutex::scoped_lock lock(m_removeSenderListMutex); - - RemoveSenderList::iterator i = m_removeSenderList.begin(); - RemoveSenderList::iterator end = m_removeSenderList.end(); - - // Synchronously remove Sender Threads whose Sessions have been closed. - while (i != end) - { - SenderMap::iterator pos = m_senderMap.find(*i); - if (pos != m_senderMap.end()) - { - boost::shared_ptr tmpSender = pos->second; - tmpSender->SignalStop(); - tmpSender->WaitStop(); - m_senderMap.erase(pos); - } - ++i; - } - m_removeSenderList.clear(); -} - void ServerLobbyThread::ResubscribeLobbyMsgLoop() { @@ -1084,11 +1056,8 @@ ServerLobbyThread::HandleNewConnection(boost::shared_ptr connData) //} // Create a new session. - boost::shared_ptr senderThread(new SenderThread(*m_senderCallback)); - senderThread->Start(); - boost::shared_ptr sessionData(new SessionData(connData->ReleaseSocket(), m_curSessionId++, senderThread, *m_senderCallback)); + boost::shared_ptr sessionData(new SessionData(connData->ReleaseSocket(), m_curSessionId++, m_sender, *m_senderCallback)); m_sessionManager.AddSession(sessionData); - m_senderMap[sessionData->GetId()] = senderThread; LOG_VERBOSE("Accepted connection - session #" << sessionData->GetId() << "."); diff --git a/src/net/senderthread.h b/src/net/senderthread.h index 0405b084..f6e91dfd 100644 --- a/src/net/senderthread.h +++ b/src/net/senderthread.h @@ -28,6 +28,7 @@ #include #include +#include class SessionData; #define SENDER_THREAD_TERMINATE_TIMEOUT THREAD_WAIT_INFINITE @@ -48,19 +49,63 @@ public: protected: typedef std::list > SendDataList; + class SendDataManager + { + public: + SendDataManager(boost::shared_ptr s, boost::asio::io_service &ioService) + : session(s), m_writeInProgress(false), m_completed(false) + { + socket.reset(new boost::asio::ip::tcp::socket( + ioService, boost::asio::ip::tcp::v6(), s->GetSocket())); + } + + void HandleWrite(const boost::system::error_code& error); + + bool IsWriteInProgress() const + { + boost::mutex::scoped_lock lock(m_mutex); + return m_writeInProgress; + } + + void SetWriteInProgress(bool v) + { + boost::mutex::scoped_lock lock(m_mutex); + m_writeInProgress = v; + } + + bool IsCompleted() const + { + boost::mutex::scoped_lock lock(m_mutex); + return m_completed; + } + + void SetCompleted(bool v) + { + boost::mutex::scoped_lock lock(m_mutex); + m_completed = v; + } + + boost::shared_ptr session; + boost::shared_ptr socket; + SendDataList list; + + private: + mutable boost::mutex m_mutex; + bool m_writeInProgress; + bool m_completed; + }; + typedef std::map > SendQueueMap; + // Main function of the thread. virtual void Main(); private: - SendDataList m_sendQueue; - mutable boost::mutex m_sendQueueMutex; + boost::asio::io_service m_ioService; + + SendQueueMap m_sendQueueMap; + mutable boost::mutex m_sendQueueMapMutex; - mutable boost::mutex m_sessionDataMutex; - SOCKET m_sessionSocket; - unsigned m_sessionId; - boost::shared_ptr m_curPacket; - unsigned m_bytesSent; SenderCallback &m_callback; }; diff --git a/src/net/serverlobbythread.h b/src/net/serverlobbythread.h index e9ca775f..7612d925 100644 --- a/src/net/serverlobbythread.h +++ b/src/net/serverlobbythread.h @@ -77,7 +77,6 @@ public: void RemoveComputerPlayer(boost::shared_ptr player); void RemoveGame(unsigned id); - void RemoveSender(unsigned session); u_int32_t GetNextUniquePlayerId(); u_int32_t GetNextGameId(); @@ -98,8 +97,6 @@ protected: typedef std::map > GameMap; typedef std::map TimerClientAddressMap; typedef std::list RemoveGameList; - typedef std::list RemoveSenderList; - typedef std::map >SenderMap; // Main function of the thread. virtual void Main(); @@ -120,7 +117,6 @@ protected: void NewSessionLoop(); void RemoveGameLoop(); void RemovePlayerLoop(); - void RemoveSenderLoop(); void ResubscribeLobbyMsgLoop(); void CheckSessionTimeoutsLoop(); void UpdateAvatarClientTimerLoop(); @@ -183,9 +179,6 @@ private: RemovePlayerList m_removePlayerList; mutable boost::mutex m_removePlayerListMutex; - RemoveSenderList m_removeSenderList; - mutable boost::mutex m_removeSenderListMutex; - PlayerDataMap m_computerPlayers; mutable boost::mutex m_computerPlayersMutex; @@ -193,8 +186,8 @@ private: mutable boost::mutex m_resubscribeListMutex; GameMap m_gameMap; - SenderMap m_senderMap; + boost::shared_ptr m_sender; boost::shared_ptr m_receiver; boost::shared_ptr m_senderCallback; GuiInterface &m_gui;