diff --git a/src/core/avatarmanager.h b/src/core/avatarmanager.h index 5aa44616..ea2e0afb 100644 --- a/src/core/avatarmanager.h +++ b/src/core/avatarmanager.h @@ -47,10 +47,10 @@ public: void AddSingleAvatar(const std::string &fileName); - boost::shared_ptr OpenAvatarFileForChunkRead(const std::string &fileName, unsigned &outFileSize, AvatarFileType &outFileType); - unsigned ChunkReadAvatarFile(boost::shared_ptr fileState, unsigned char *data, unsigned chunkSize); + static boost::shared_ptr OpenAvatarFileForChunkRead(const std::string &fileName, unsigned &outFileSize, AvatarFileType &outFileType); + static unsigned ChunkReadAvatarFile(boost::shared_ptr fileState, unsigned char *data, unsigned chunkSize); - int AvatarFileToNetPackets(const std::string &fileName, unsigned requestId, NetPacketList &packets); + static int AvatarFileToNetPackets(const std::string &fileName, unsigned requestId, NetPacketList &packets); bool GetHashForAvatar(const std::string &fileName, MD5Buf &md5buf) const; bool GetAvatarFileName(const MD5Buf &md5buf, std::string &fileName) const; @@ -77,8 +77,6 @@ private: mutable boost::mutex m_cacheDirMutex; std::string m_cacheDir; - - mutable boost::mutex m_loadMutex; }; #endif diff --git a/src/core/common/avatarmanager.cpp b/src/core/common/avatarmanager.cpp index 1eed443a..f1773a84 100644 --- a/src/core/common/avatarmanager.cpp +++ b/src/core/common/avatarmanager.cpp @@ -192,9 +192,6 @@ AvatarManager::ChunkReadAvatarFile(boost::shared_ptr fileState, int AvatarManager::AvatarFileToNetPackets(const string &fileName, unsigned requestId, NetPacketList &packets) { - // Serialize avatar access for now. - boost::mutex::scoped_lock lock(m_loadMutex); - int retVal = ERR_NET_INVALID_AVATAR_FILE; unsigned fileSize; AvatarFileType fileType; diff --git a/src/net/common/clientstate.cpp b/src/net/common/clientstate.cpp index bd80dc1d..ee8f2215 100644 --- a/src/net/common/clientstate.cpp +++ b/src/net/common/clientstate.cpp @@ -592,7 +592,7 @@ ClientStateWaitSession::InternalProcess(ClientThread &client, boost::shared_ptr< tmpList); if (!avatarError) - client.GetSender().SendLowPrio(client.GetContext().GetSessionData(), tmpList); + client.GetSender().Send(client.GetContext().GetSessionData(), tmpList); else throw ClientException(__FILE__, __LINE__, avatarError, 0); } diff --git a/src/net/common/senderthread.cpp b/src/net/common/senderthread.cpp index 5344a039..f65a15c3 100644 --- a/src/net/common/senderthread.cpp +++ b/src/net/common/senderthread.cpp @@ -32,16 +32,10 @@ using namespace std; #define SEND_ERROR_TIMEOUT_MSEC 20000 #define SEND_TIMEOUT_MSEC 10 -#ifdef POKERTH_DEDICATED_SERVER - #define SEND_QUEUE_SIZE 10000 - #define SEND_LOW_PRIO_QUEUE_SIZE 10000000 -#else - #define SEND_QUEUE_SIZE 1000 - #define SEND_LOW_PRIO_QUEUE_SIZE 10000 -#endif +#define SEND_QUEUE_SIZE 10000000 SenderThread::SenderThread(SenderCallback &cb) -: m_tmpOutBufSize(0), m_lastInvalidSessionId(INVALID_SESSION), m_callback(cb) +: m_callback(cb) { } @@ -54,8 +48,8 @@ SenderThread::Send(boost::shared_ptr session, boost::shared_ptr session, const NetPacketList & { if (!packetList.empty() && session.get()) { - boost::mutex::scoped_lock lock(m_outBufMutex); - InternalStore(m_outBuf, SEND_QUEUE_SIZE, session, packetList); - } -} - -void -SenderThread::SendLowPrio(boost::shared_ptr session, boost::shared_ptr packet) -{ - if (packet.get() && session.get()) - { - boost::mutex::scoped_lock lock(m_lowPrioOutBufMutex); - InternalStore(m_lowPrioOutBuf, SEND_LOW_PRIO_QUEUE_SIZE, session, packet); - } -} - -void -SenderThread::SendLowPrio(boost::shared_ptr session, const NetPacketList &packetList) -{ - if (!packetList.empty() && session.get()) - { - boost::mutex::scoped_lock lock(m_lowPrioOutBufMutex); - InternalStore(m_lowPrioOutBuf, SEND_LOW_PRIO_QUEUE_SIZE, session, packetList); + boost::mutex::scoped_lock lock(m_sendQueueMutex); + InternalStore(m_sendQueue, SEND_QUEUE_SIZE, session, packetList); } } @@ -94,12 +68,12 @@ SenderThread::GetNumPacketsInQueue() const { unsigned numPackets; { - boost::mutex::scoped_lock lock(m_lowPrioOutBufMutex); - numPackets = m_lowPrioOutBuf.size(); + boost::mutex::scoped_lock lock(m_sendQueueMutex); + numPackets = m_sendQueue.size(); } { - boost::mutex::scoped_lock lock(m_outBufMutex); - numPackets += m_outBuf.size(); + boost::mutex::scoped_lock lock(m_stalledQueueMutex); + numPackets += m_stalledQueue.size(); } return numPackets; } @@ -111,169 +85,178 @@ SenderThread::operator<(const SenderThread &other) const } void -SenderThread::InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boost::shared_ptr session, boost::shared_ptr packet) +SenderThread::InternalStore(SendDataList &sendQueue, unsigned maxQueueSize, boost::shared_ptr session, boost::shared_ptr packet) { if (sendQueue.size() < maxQueueSize) // Queue is limited in size. - sendQueue.push_back(std::make_pair(packet, session)); + sendQueue.push_back(SendData(packet, session)); // TODO: Throw exception if failed. } void -SenderThread::InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boost::shared_ptr session, const NetPacketList &packetList) +SenderThread::InternalStore(SendDataList &sendQueue, unsigned maxQueueSize, boost::shared_ptr session, const NetPacketList &packetList) { - if (sendQueue.size() + packetList.size() < maxQueueSize) + if (sendQueue.size() + packetList.size() <= maxQueueSize) { NetPacketList::const_iterator i = packetList.begin(); NetPacketList::const_iterator end = packetList.end(); while (i != end) { - sendQueue.push_back(std::make_pair(*i, session)); + sendQueue.push_back(SendData(*i, session)); ++i; } } // TODO: Throw exception if failed. } -void -SenderThread::RemoveCurSendData() -{ - m_tmpOutBufSize = 0; - m_curSession.reset(); - // TODO use callback to remove session. -} - void SenderThread::Main() { - boost::timers::portable::microsec_timer sendTimer(boost::posix_time::time_duration(0, 0, 0), boost::timers::portable::microsec_timer::manual_start); - while (!ShouldTerminate()) { - // Send remaining bytes of output buffer OR - // copy ONE packet to output buffer. + /* + * To prevent stalling of the sender, keeping the order + * of the packets in the sender queue is not guaranteed. + * Instead, only the order of the packets for a single + * target session is maintained. + * + * This could also be done by using one sender thread + * for each session, but that would require too many + * resources. + * + * The send queue is a list of packets. Whenever a + * select timeout occurs, the session of the current + * packet is placed in the stalled list, and all other + * packets from the queue for that session are also + * attached to the stalled list. This process is + * continued with the next packet in the send queue. + * + * When the send queue is empty, the list of stalled + * packets is copied back to the send queue, and + * everything starts from the beginning. + * + * No packets are lost in this algorithm, and at the same + * time, the send process is never fully stalled, + * except when only packets for stalled sessions are + * present. + * + * Note: If someone keeps putting in packets to send, the + * stalled packets will never be sent, but this is + * considered more a theoretical problem. + */ + // For reasons of simplicity, only one packet is sent at a time. - if (!m_tmpOutBufSize) + SendData tmpData; + // Check main queue. { - bool isLowPrio = false; - SendData tmpData; - // Check main queue first. + boost::mutex::scoped_lock lock(m_sendQueueMutex); + if (m_sendQueue.empty()) { - boost::mutex::scoped_lock lock(m_outBufMutex); - if (!m_outBuf.empty()) - { - tmpData = m_outBuf.front(); - m_outBuf.pop_front(); - } + // Check stalled queue. + // Attention: double lock (on purpose). + boost::mutex::scoped_lock lock2(m_stalledQueueMutex); + if (!m_stalledQueue.empty()) + m_sendQueue.swap(m_stalledQueue); } - - // Check low prio queue only if there is nothing in the main queue. - if (!tmpData.first.get()) + if (!m_sendQueue.empty()) { - boost::mutex::scoped_lock lock(m_lowPrioOutBufMutex); - if (!m_lowPrioOutBuf.empty()) - { - tmpData = m_lowPrioOutBuf.front(); - m_lowPrioOutBuf.pop_front(); - isLowPrio = true; - } - } - - if (tmpData.first.get() && tmpData.second.get()) - { - if (!isLowPrio || tmpData.second->GetId() != m_lastInvalidSessionId) - { - u_int16_t tmpLen = tmpData.first->GetLen(); - if (tmpLen <= MAX_PACKET_SIZE) - { - m_curSession = tmpData.second; - m_tmpOutBufSize = tmpLen; - memcpy(m_tmpOutBuf, tmpData.first->GetRawData(), tmpLen); - sendTimer.restart(); - } - } + tmpData = m_sendQueue.front(); + m_sendQueue.pop_front(); } } - if (m_tmpOutBufSize) + + if (tmpData.packet && tmpData.session) { - SOCKET tmpSocket = m_curSession->GetSocket(); - - // send next chunk of data - int bytesSent = send(tmpSocket, m_tmpOutBuf, m_tmpOutBufSize, SOCKET_SEND_FLAGS); - - if (!IS_VALID_SEND(bytesSent)) + const unsigned tmpLen = tmpData.packet->GetLen(); + if (tmpLen <= MAX_PACKET_SIZE) { - // Never assume that this is a fatal error. - int errCode = SOCKET_ERRNO(); - if (IS_SOCKET_ERR_WOULDBLOCK(errCode)) + SOCKET tmpSocket = tmpData.session->GetSocket(); + + // send next chunk of data + int bytesSent = send(tmpSocket, ((const char *)tmpData.packet->GetRawData()) + tmpData.bytesSent, tmpLen - tmpData.bytesSent, SOCKET_SEND_FLAGS); + + if (!IS_VALID_SEND(bytesSent)) { - 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)) + // Never assume that this is a fatal error. + int errCode = SOCKET_ERRNO(); + if (IS_SOCKET_ERR_WOULDBLOCK(errCode)) { - // Never assume that this is a fatal error. - int errCode = SOCKET_ERRNO(); - if (!IS_SOCKET_ERR_WOULDBLOCK(errCode)) + 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)) { - // 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) - m_callback.SignalNetError(m_curSession->GetId(), ERR_SOCK_SELECT_FAILED, errCode); - RemoveCurSendData(); + // 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) + m_callback.SignalNetError(tmpData.session->GetId(), ERR_SOCK_SELECT_FAILED, errCode); + } + Msleep(SEND_TIMEOUT_MSEC); } + else if (selectResult == 0) + { + // A timeout occured - don't block the thread. + { + // Attention: double lock (on purpose). + // Stall all packets for that sender. + boost::mutex::scoped_lock lock(m_sendQueueMutex); + boost::mutex::scoped_lock lock2(m_stalledQueueMutex); + m_stalledQueue.push_back(tmpData); + SendDataList::iterator i = m_sendQueue.begin(); + SendDataList::iterator end = m_sendQueue.end(); + while (i != end) + { + SendDataList::iterator next = i; + ++next; + if ((*i).session && (*i).session->GetId() == tmpData.session->GetId()) + { + m_stalledQueue.push_back(*i); + m_sendQueue.erase(i); + } + i = next; + } + } + } + else + { + // Select was successful - store the packet back in the main queue. + boost::mutex::scoped_lock lock(m_sendQueueMutex); + m_sendQueue.push_back(tmpData); + } + } + 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) + m_callback.SignalNetError(tmpData.session->GetId(), ERR_SOCK_SEND_FAILED, errCode); Msleep(SEND_TIMEOUT_MSEC); } } - else // other errors than would block + else if ((unsigned)bytesSent + tmpData.bytesSent < tmpLen) { - // 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) - m_callback.SignalNetError(m_curSession->GetId(), ERR_SOCK_SEND_FAILED, errCode); - RemoveCurSendData(); - Msleep(SEND_TIMEOUT_MSEC); + if (bytesSent) + { + tmpData.bytesSent += bytesSent; + // Send was partly successful - store the packet back in the main queue. + boost::mutex::scoped_lock lock(m_sendQueueMutex); + m_sendQueue.push_back(tmpData); + } + else + Msleep(SEND_TIMEOUT_MSEC); } } - else if ((unsigned)bytesSent < m_tmpOutBufSize) - { - if (bytesSent) - { - m_tmpOutBufSize -= (unsigned)bytesSent; - memmove(m_tmpOutBuf, m_tmpOutBuf + bytesSent, m_tmpOutBufSize); - } - else - Msleep(SEND_TIMEOUT_MSEC); - } else - { - assert(bytesSent == m_tmpOutBufSize); - m_tmpOutBufSize = 0; - m_curSession.reset(); - sendTimer.reset(); - } - } - else - Msleep(SEND_TIMEOUT_MSEC); - - // Check whether the send timed out. - if (sendTimer.is_running()) - { - if (sendTimer.elapsed().total_milliseconds() > SEND_ERROR_TIMEOUT_MSEC) - { - if (m_curSession.get()) - { - m_lastInvalidSessionId = m_curSession->GetId(); - LOG_MSG("Send operation for session " << m_lastInvalidSessionId << " timed out."); - } - RemoveCurSendData(); - sendTimer.reset(); - } + Msleep(SEND_TIMEOUT_MSEC); } } } diff --git a/src/net/common/servergamestate.cpp b/src/net/common/servergamestate.cpp index c7e5a82e..eca82ebd 100644 --- a/src/net/common/servergamestate.cpp +++ b/src/net/common/servergamestate.cpp @@ -36,7 +36,7 @@ using namespace std; -//#define SERVER_TEST +#define SERVER_TEST #define SERVER_START_GAME_TIMEOUT_SEC 10 diff --git a/src/net/common/serverlobbythread.cpp b/src/net/common/serverlobbythread.cpp index 62f2d045..f034adf2 100644 --- a/src/net/common/serverlobbythread.cpp +++ b/src/net/common/serverlobbythread.cpp @@ -165,8 +165,8 @@ ServerLobbyThread::NotifyPlayerJoinedGame(unsigned gameId, unsigned playerId) packetData.gameId = gameId; packetData.playerId = playerId; static_cast(packet.get())->SetData(packetData); - m_sessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Established); - m_gameSessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Game); + m_sessionManager.SendToAllSessions(GetSender(), packet, SessionData::Established); + m_gameSessionManager.SendToAllSessions(GetSender(), packet, SessionData::Game); } void @@ -178,8 +178,8 @@ ServerLobbyThread::NotifyPlayerLeftGame(unsigned gameId, unsigned playerId) packetData.gameId = gameId; packetData.playerId = playerId; static_cast(packet.get())->SetData(packetData); - m_sessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Established); - m_gameSessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Game); + m_sessionManager.SendToAllSessions(GetSender(), packet, SessionData::Established); + m_gameSessionManager.SendToAllSessions(GetSender(), packet, SessionData::Game); } void @@ -191,24 +191,24 @@ ServerLobbyThread::NotifyGameAdminChanged(unsigned gameId, unsigned newAdminPlay packetData.gameId = gameId; packetData.newAdminplayerId = newAdminPlayerId; static_cast(packet.get())->SetData(packetData); - m_sessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Established); - m_gameSessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Game); + m_sessionManager.SendToAllSessions(GetSender(), packet, SessionData::Established); + m_gameSessionManager.SendToAllSessions(GetSender(), packet, SessionData::Game); } void ServerLobbyThread::NotifyStartingGame(unsigned gameId) { boost::shared_ptr packet = CreateNetPacketGameListUpdate(gameId, GAME_MODE_STARTED); - m_sessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Established); - m_gameSessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Game); + m_sessionManager.SendToAllSessions(GetSender(), packet, SessionData::Established); + m_gameSessionManager.SendToAllSessions(GetSender(), packet, SessionData::Game); } void ServerLobbyThread::NotifyReopeningGame(unsigned gameId) { boost::shared_ptr packet = CreateNetPacketGameListUpdate(gameId, GAME_MODE_CREATED); - m_sessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Established); - m_gameSessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Game); + m_sessionManager.SendToAllSessions(GetSender(), packet, SessionData::Established); + m_gameSessionManager.SendToAllSessions(GetSender(), packet, SessionData::Game); } void @@ -592,7 +592,7 @@ ServerLobbyThread::HandleNetPacketRetrieveAvatar(SessionWrapper session, const N avatarFound = true; SenderThreadList::iterator pos = min_element(m_avatarSenderThreadPool.begin(), m_avatarSenderThreadPool.end(), *boost::lambda::_1 < *boost::lambda::_2); if (pos != m_avatarSenderThreadPool.end()) - (*pos)->SendLowPrio(session.sessionData, tmpPackets); + (*pos)->Send(session.sessionData, tmpPackets); else LOG_ERROR("Load balancing for avatar sender threads failed."); } @@ -827,8 +827,8 @@ ServerLobbyThread::InternalAddGame(boost::shared_ptr game) // Add game to list. m_gameMap.insert(GameMap::value_type(game->GetId(), game)); // Notify all players. - m_sessionManager.SendToAllSessionsLowPrio(GetSender(), CreateNetPacketGameListNew(*game), SessionData::Established); - m_gameSessionManager.SendToAllSessionsLowPrio(GetSender(), CreateNetPacketGameListNew(*game), SessionData::Game); + m_sessionManager.SendToAllSessions(GetSender(), CreateNetPacketGameListNew(*game), SessionData::Established); + m_gameSessionManager.SendToAllSessions(GetSender(), CreateNetPacketGameListNew(*game), SessionData::Game); { boost::mutex::scoped_lock lock(m_statMutex); @@ -849,8 +849,8 @@ ServerLobbyThread::InternalRemoveGame(boost::shared_ptr game) game->RemoveAllSessions(); // Notify all players. boost::shared_ptr packet = CreateNetPacketGameListUpdate(game->GetId(), GAME_MODE_CLOSED); - m_sessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Established); - m_gameSessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Game); + m_sessionManager.SendToAllSessions(GetSender(), packet, SessionData::Established); + m_gameSessionManager.SendToAllSessions(GetSender(), packet, SessionData::Game); } void @@ -1024,8 +1024,8 @@ ServerLobbyThread::BroadcastStatisticsUpdate(const ServerStats &stats) try { static_cast(packet.get())->SetData(statData); - m_sessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Established); - m_gameSessionManager.SendToAllSessionsLowPrio(GetSender(), packet, SessionData::Game); + m_sessionManager.SendToAllSessions(GetSender(), packet, SessionData::Established); + m_gameSessionManager.SendToAllSessions(GetSender(), packet, SessionData::Game); } catch (const NetException &) { // Ignore errors for now. diff --git a/src/net/common/sessionmanager.cpp b/src/net/common/sessionmanager.cpp index 10fea918..bb507470 100644 --- a/src/net/common/sessionmanager.cpp +++ b/src/net/common/sessionmanager.cpp @@ -363,26 +363,6 @@ SessionManager::SendToAllSessions(SenderThread &sender, boost::shared_ptr packet, SessionData::State state) -{ - boost::mutex::scoped_lock lock(m_sessionMapMutex); - - SessionMap::iterator i = m_sessionMap.begin(); - SessionMap::iterator end = m_sessionMap.end(); - - while (i != end) - { - if (!i->second.sessionData.get()) - throw ServerException(__FILE__, __LINE__, ERR_NET_INVALID_SESSION, 0); - - // Send each client (with a certain state) a copy of the packet. - if (i->second.sessionData->GetState() == state) - sender.SendLowPrio(i->second.sessionData, boost::shared_ptr(packet->Clone())); - ++i; - } -} - void SessionManager::SendToAllButOneSessions(SenderThread &sender, boost::shared_ptr packet, SessionId except, SessionData::State state) { diff --git a/src/net/senderthread.h b/src/net/senderthread.h index c2ff4863..6a6c5c8f 100644 --- a/src/net/senderthread.h +++ b/src/net/senderthread.h @@ -26,7 +26,7 @@ #include #include -#include +#include #include #define SENDER_THREAD_TERMINATE_TIMEOUT THREAD_WAIT_INFINITE @@ -40,38 +40,35 @@ public: void Send(boost::shared_ptr session, boost::shared_ptr packet); void Send(boost::shared_ptr session, const NetPacketList &packetList); - void SendLowPrio(boost::shared_ptr session, boost::shared_ptr packet); - void SendLowPrio(boost::shared_ptr session, const NetPacketList &packetList); - unsigned GetNumPacketsInQueue() const; bool operator<(const SenderThread &other) const; protected: - typedef std::pair, boost::shared_ptr > SendData; - typedef std::deque SendDataDeque; + struct SendData + { + SendData() + : bytesSent(0) {} + SendData(boost::shared_ptr p, boost::shared_ptr s) + : packet(p), session(s), bytesSent(0) {} + boost::shared_ptr packet; + boost::shared_ptr session; + unsigned bytesSent; + }; + typedef std::list SendDataList; // Main function of the thread. virtual void Main(); - void InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boost::shared_ptr session, boost::shared_ptr packet); - void InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boost::shared_ptr session, const NetPacketList &packetList); - - void RemoveCurSendData(); + void InternalStore(SendDataList &sendQueue, unsigned maxQueueSize, boost::shared_ptr session, boost::shared_ptr packet); + void InternalStore(SendDataList &sendQueue, unsigned maxQueueSize, boost::shared_ptr session, const NetPacketList &packetList); private: - boost::shared_ptr m_curSession; + SendDataList m_sendQueue; + mutable boost::mutex m_sendQueueMutex; - std::deque m_outBuf; - mutable boost::mutex m_outBufMutex; - - std::deque m_lowPrioOutBuf; - mutable boost::mutex m_lowPrioOutBufMutex; - - char m_tmpOutBuf[MAX_PACKET_SIZE]; - unsigned m_tmpOutBufSize; - bool m_tmpIsLowPrio; - unsigned m_lastInvalidSessionId; + SendDataList m_stalledQueue; + mutable boost::mutex m_stalledQueueMutex; SenderCallback &m_callback; }; diff --git a/src/net/sessionmanager.h b/src/net/sessionmanager.h index 6e8c28f5..4b5983c5 100644 --- a/src/net/sessionmanager.h +++ b/src/net/sessionmanager.h @@ -74,7 +74,6 @@ public: unsigned GetRawSessionCount(); void SendToAllSessions(SenderThread &sender, boost::shared_ptr packet, SessionData::State state); - void SendToAllSessionsLowPrio(SenderThread &sender, boost::shared_ptr packet, SessionData::State state); void SendToAllButOneSessions(SenderThread &sender, boost::shared_ptr packet, SessionId except, SessionData::State state); protected: