If a sender thread is stalled while trying to send, time out after 2 seconds. Use infinite timeout when terminating threads.
This commit is contained in:
@@ -22,10 +22,12 @@
|
|||||||
#include <net/socket_msg.h>
|
#include <net/socket_msg.h>
|
||||||
#include <cstring>
|
#include <cstring>
|
||||||
|
|
||||||
|
#include <core/boost/timers.hpp>
|
||||||
#include <boost/bind.hpp>
|
#include <boost/bind.hpp>
|
||||||
|
|
||||||
using namespace std;
|
using namespace std;
|
||||||
|
|
||||||
|
#define SEND_TIMEOUT_MSEC 2000
|
||||||
|
|
||||||
SenderThread::SenderThread(SenderCallback &cb)
|
SenderThread::SenderThread(SenderCallback &cb)
|
||||||
: m_tmpOutBufSize(0), m_callback(cb)
|
: m_tmpOutBufSize(0), m_callback(cb)
|
||||||
@@ -100,11 +102,26 @@ SenderThread::InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boo
|
|||||||
// TODO: Throw exception if failed.
|
// TODO: Throw exception if failed.
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void
|
||||||
|
SenderThread::RemoveCurSendData()
|
||||||
|
{
|
||||||
|
m_tmpOutBufSize = 0;
|
||||||
|
m_curSession.reset();
|
||||||
|
// TODO use callback to remove session.
|
||||||
|
}
|
||||||
|
|
||||||
void
|
void
|
||||||
SenderThread::Main()
|
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())
|
while (!ShouldTerminate())
|
||||||
{
|
{
|
||||||
|
if (sendTimer.is_running() && sendTimer.elapsed().total_milliseconds() > SEND_TIMEOUT_MSEC)
|
||||||
|
{
|
||||||
|
RemoveCurSendData();
|
||||||
|
sendTimer.reset();
|
||||||
|
}
|
||||||
// Send remaining bytes of output buffer OR
|
// Send remaining bytes of output buffer OR
|
||||||
// copy ONE packet to output buffer.
|
// copy ONE packet to output buffer.
|
||||||
// For reasons of simplicity, only one packet is sent at a time.
|
// For reasons of simplicity, only one packet is sent at a time.
|
||||||
@@ -135,13 +152,15 @@ SenderThread::Main()
|
|||||||
if (tmpData.first.get())
|
if (tmpData.first.get())
|
||||||
{
|
{
|
||||||
if (tmpData.second.get())
|
if (tmpData.second.get())
|
||||||
m_curSession = tmpData.second;
|
{
|
||||||
|
|
||||||
u_int16_t tmpLen = tmpData.first->GetLen();
|
u_int16_t tmpLen = tmpData.first->GetLen();
|
||||||
if (tmpLen <= MAX_PACKET_SIZE)
|
if (tmpLen <= MAX_PACKET_SIZE)
|
||||||
{
|
{
|
||||||
|
m_curSession = tmpData.second;
|
||||||
m_tmpOutBufSize = tmpLen;
|
m_tmpOutBufSize = tmpLen;
|
||||||
memcpy(m_tmpOutBuf, tmpData.first->GetRawData(), tmpLen);
|
memcpy(m_tmpOutBuf, tmpData.first->GetRawData(), tmpLen);
|
||||||
|
sendTimer.restart();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -167,8 +186,7 @@ SenderThread::Main()
|
|||||||
// Ignore invalid or not connected sockets.
|
// Ignore invalid or not connected sockets.
|
||||||
if (errCode != SOCKET_ERR_NOTCONN && errCode != SOCKET_ERR_NOTSOCK)
|
if (errCode != SOCKET_ERR_NOTCONN && errCode != SOCKET_ERR_NOTSOCK)
|
||||||
m_callback.SignalNetError(m_curSession->GetId(), ERR_SOCK_SELECT_FAILED, errCode);
|
m_callback.SignalNetError(m_curSession->GetId(), ERR_SOCK_SELECT_FAILED, errCode);
|
||||||
m_tmpOutBufSize = 0;
|
RemoveCurSendData();
|
||||||
m_curSession.reset();
|
|
||||||
}
|
}
|
||||||
Msleep(SEND_TIMEOUT_MSEC);
|
Msleep(SEND_TIMEOUT_MSEC);
|
||||||
}
|
}
|
||||||
@@ -187,8 +205,7 @@ SenderThread::Main()
|
|||||||
// Ignore invalid or not connected sockets.
|
// Ignore invalid or not connected sockets.
|
||||||
if (errCode != SOCKET_ERR_NOTCONN && errCode != SOCKET_ERR_NOTSOCK)
|
if (errCode != SOCKET_ERR_NOTCONN && errCode != SOCKET_ERR_NOTSOCK)
|
||||||
m_callback.SignalNetError(m_curSession->GetId(), ERR_SOCK_SEND_FAILED, errCode);
|
m_callback.SignalNetError(m_curSession->GetId(), ERR_SOCK_SEND_FAILED, errCode);
|
||||||
m_tmpOutBufSize = 0;
|
RemoveCurSendData();
|
||||||
m_curSession.reset();
|
|
||||||
}
|
}
|
||||||
Msleep(SEND_TIMEOUT_MSEC);
|
Msleep(SEND_TIMEOUT_MSEC);
|
||||||
}
|
}
|
||||||
@@ -201,6 +218,7 @@ SenderThread::Main()
|
|||||||
{
|
{
|
||||||
m_tmpOutBufSize = 0;
|
m_tmpOutBufSize = 0;
|
||||||
m_curSession.reset();
|
m_curSession.reset();
|
||||||
|
sendTimer.reset();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -30,7 +30,7 @@
|
|||||||
#include <deque>
|
#include <deque>
|
||||||
#include <boost/shared_ptr.hpp>
|
#include <boost/shared_ptr.hpp>
|
||||||
|
|
||||||
#define SENDER_THREAD_TERMINATE_TIMEOUT 10000
|
#define SENDER_THREAD_TERMINATE_TIMEOUT THREAD_WAIT_INFINITE
|
||||||
#define SEND_TIMEOUT_MSEC 10
|
#define SEND_TIMEOUT_MSEC 10
|
||||||
#define SEND_QUEUE_SIZE 1000
|
#define SEND_QUEUE_SIZE 1000
|
||||||
#define SEND_LOW_PRIO_QUEUE_SIZE 50000
|
#define SEND_LOW_PRIO_QUEUE_SIZE 50000
|
||||||
@@ -57,6 +57,8 @@ protected:
|
|||||||
void InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boost::shared_ptr<SessionData> session, boost::shared_ptr<NetPacket> packet);
|
void InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boost::shared_ptr<SessionData> session, boost::shared_ptr<NetPacket> packet);
|
||||||
void InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boost::shared_ptr<SessionData> session, const NetPacketList &packetList);
|
void InternalStore(SendDataDeque &sendQueue, unsigned maxQueueSize, boost::shared_ptr<SessionData> session, const NetPacketList &packetList);
|
||||||
|
|
||||||
|
void RemoveCurSendData();
|
||||||
|
|
||||||
private:
|
private:
|
||||||
|
|
||||||
boost::shared_ptr<SessionData> m_curSession;
|
boost::shared_ptr<SessionData> m_curSession;
|
||||||
|
|||||||
@@ -27,7 +27,7 @@
|
|||||||
|
|
||||||
#include <deque>
|
#include <deque>
|
||||||
|
|
||||||
#define GAME_THREAD_TERMINATE_TIMEOUT 20000
|
#define GAME_THREAD_TERMINATE_TIMEOUT THREAD_WAIT_INFINITE
|
||||||
|
|
||||||
|
|
||||||
class SenderThread;
|
class SenderThread;
|
||||||
|
|||||||
Reference in New Issue
Block a user