The sender thread has been rewritten with a new algorithm in mind. The sender should not stall any more, and no timeout is required. There are no more send priorities.
The avatar manager no longer serializes access to avatar files.
This commit is contained in:
lotodore
+139
-156
@@ -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<SessionData> session, boost::shared_ptr<Net
|
||||
{
|
||||
if (packet.get() && session.get())
|
||||
{
|
||||
boost::mutex::scoped_lock lock(m_outBufMutex);
|
||||
InternalStore(m_outBuf, SEND_QUEUE_SIZE, session, packet);
|
||||
boost::mutex::scoped_lock lock(m_sendQueueMutex);
|
||||
InternalStore(m_sendQueue, SEND_QUEUE_SIZE, session, packet);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -64,28 +58,8 @@ SenderThread::Send(boost::shared_ptr<SessionData> 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<SessionData> session, boost::shared_ptr<NetPacket> 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<SessionData> 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<SessionData> session, boost::shared_ptr<NetPacket> packet)
|
||||
SenderThread::InternalStore(SendDataList &sendQueue, unsigned maxQueueSize, boost::shared_ptr<SessionData> session, boost::shared_ptr<NetPacket> 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<SessionData> session, const NetPacketList &packetList)
|
||||
SenderThread::InternalStore(SendDataList &sendQueue, unsigned maxQueueSize, boost::shared_ptr<SessionData> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user