diff --git a/src/net/common/senderthread.cpp b/src/net/common/senderthread.cpp index 42b9993e..5783890f 100644 --- a/src/net/common/senderthread.cpp +++ b/src/net/common/senderthread.cpp @@ -49,8 +49,18 @@ SenderThread::Send(boost::shared_ptr session, boost::shared_ptrGetId()) == m_sessionsStalled.end()) + { + boost::mutex::scoped_lock lock1(m_sendQueueMutex); + InternalStore(m_sendQueue, SEND_QUEUE_SIZE, session, packet); + } + else + { + // This session is stalled. + boost::mutex::scoped_lock lock2(m_stalledQueueMutex); + InternalStore(m_stalledQueue, SEND_QUEUE_SIZE, session, packet); + } } } @@ -59,8 +69,18 @@ SenderThread::Send(boost::shared_ptr session, const NetPacketList & { if (!packetList.empty() && session.get()) { - boost::mutex::scoped_lock lock(m_sendQueueMutex); - InternalStore(m_sendQueue, SEND_QUEUE_SIZE, session, packetList); + boost::mutex::scoped_lock lock0(m_sessionsStalledMutex); + if (find(m_sessionsStalled.begin(), m_sessionsStalled.end(), session->GetId()) == m_sessionsStalled.end()) + { + boost::mutex::scoped_lock lock1(m_sendQueueMutex); + InternalStore(m_sendQueue, SEND_QUEUE_SIZE, session, packetList); + } + else + { + // This session is stalled. + boost::mutex::scoped_lock lock2(m_stalledQueueMutex); + InternalStore(m_stalledQueue, SEND_QUEUE_SIZE, session, packetList); + } } } @@ -69,11 +89,11 @@ SenderThread::GetNumPacketsInQueue() const { unsigned numPackets; { - boost::mutex::scoped_lock lock(m_sendQueueMutex); + boost::mutex::scoped_lock lock1(m_sendQueueMutex); numPackets = m_sendQueue.size(); } { - boost::mutex::scoped_lock lock(m_stalledQueueMutex); + boost::mutex::scoped_lock lock2(m_stalledQueueMutex); numPackets += m_stalledQueue.size(); } return numPackets; @@ -149,14 +169,18 @@ SenderThread::Main() SendData tmpData; // Check main queue. { - boost::mutex::scoped_lock lock(m_sendQueueMutex); + boost::mutex::scoped_lock lock0(m_sessionsStalledMutex); + boost::mutex::scoped_lock lock1(m_sendQueueMutex); if (m_sendQueue.empty()) { // Check stalled queue. - // Attention: double lock (on purpose). + // Attention: TRIPLE lock (on purpose). boost::mutex::scoped_lock lock2(m_stalledQueueMutex); if (!m_stalledQueue.empty()) + { m_sendQueue.swap(m_stalledQueue); + m_sessionsStalled.clear(); // No more sessions stalled, all in send list. + } } if (!m_sendQueue.empty()) { @@ -207,11 +231,13 @@ SenderThread::Main() { // A timeout occured - don't block the thread. { - // Attention: double lock (on purpose). + // Attention: TRIPLE lock (on purpose). // Stall all packets for that sender. - boost::mutex::scoped_lock lock(m_sendQueueMutex); + boost::mutex::scoped_lock lock0(m_sessionsStalledMutex); + boost::mutex::scoped_lock lock1(m_sendQueueMutex); boost::mutex::scoped_lock lock2(m_stalledQueueMutex); m_stalledQueue.push_back(tmpData); + m_sessionsStalled.push_back(tmpData.session->GetId()); SendDataList::iterator i = m_sendQueue.begin(); SendDataList::iterator end = m_sendQueue.end(); while (i != end) @@ -230,7 +256,7 @@ SenderThread::Main() else { // Select was successful - store the packet back in the main queue. - boost::mutex::scoped_lock lock(m_sendQueueMutex); + boost::mutex::scoped_lock lock1(m_sendQueueMutex); m_sendQueue.push_front(tmpData); } } @@ -249,7 +275,7 @@ SenderThread::Main() { tmpData.bytesSent += bytesSent; // Send was partly successful - store the packet back in the main queue. - boost::mutex::scoped_lock lock(m_sendQueueMutex); + boost::mutex::scoped_lock lock1(m_sendQueueMutex); m_sendQueue.push_front(tmpData); } else @@ -267,11 +293,11 @@ SenderThread::Main() unsigned sendQueueSize = 0; unsigned stalledQueueSize = 0; { - boost::mutex::scoped_lock lock(m_sendQueueMutex); + boost::mutex::scoped_lock lock1(m_sendQueueMutex); sendQueueSize = m_sendQueue.size(); } { - boost::mutex::scoped_lock lock(m_stalledQueueMutex); + boost::mutex::scoped_lock lock2(m_stalledQueueMutex); stalledQueueSize = m_stalledQueue.size(); } LOG_VERBOSE("Sender TICK - send queue " << sendQueueSize << ", stalled " << stalledQueueSize << "."); diff --git a/src/net/senderthread.h b/src/net/senderthread.h index 536928d0..70e08ea3 100644 --- a/src/net/senderthread.h +++ b/src/net/senderthread.h @@ -56,6 +56,7 @@ protected: unsigned bytesSent; }; typedef std::list SendDataList; + typedef std::list SessionIdList; // Main function of the thread. virtual void Main(); @@ -71,6 +72,9 @@ private: SendDataList m_stalledQueue; mutable boost::mutex m_stalledQueueMutex; + SessionIdList m_sessionsStalled; // Cache + mutable boost::mutex m_sessionsStalledMutex; + SenderCallback &m_callback; boost::timers::portable::microsec_timer m_logTimer; diff --git a/src/net/sessiondata.h b/src/net/sessiondata.h index b6f5e03a..7a93a519 100644 --- a/src/net/sessiondata.h +++ b/src/net/sessiondata.h @@ -32,11 +32,6 @@ #define SESSION_ID_GENERIC 0xFFFFFFFF typedef unsigned SessionId; -/*struct SessionId -{ - unsigned id; -};*/ - class SessionData {