net_ws_queued_packet_sender.cpp

//====== Copyright � 1996-2005, Valve Corporation, All rights reserved. =======
//
// Purpose: 
//
//=============================================================================

#include "net_structures.h"
#include "tier0/threadtools.h"
#include "net_ws_queued_packet_sender.h"
#include "wrappers.h"

#include "tier1/utlvector.h"
#include "tier1/utlpriorityqueue.h"

class CQueuedPacketSender : public CThread, public IQueuedPacketSender
{
public:
	CQueuedPacketSender();
	~CQueuedPacketSender();

	// IQueuedPacketSender

	virtual bool Setup();
	virtual void Shutdown();
	virtual bool IsRunning() { return CThread::IsAlive(); }

	virtual void ClearQueuedPacketsForChannel( INetChannel *pChan );
	virtual void QueuePacket( INetChannel *pChan, SOCKET s, const char *buf, int len, const struct sockaddr * to, int tolen, uint32 msecDelay );

private:

	// CThread Overrides
	virtual bool Start( unsigned int nBytesStack = 0 );
	virtual int Run();

private:

	class CQueuedPacket
	{
	public:
		uint32				m_unSendTime;
		const void 			*m_pChannel;  // We don't actually use the channel
		SOCKET				m_Socket;
		CUtlVector<char>	to;	// sockaddr
		CUtlVector<char>	buf;

		// We want the list sorted in ascending order, so note that we return > rather than <
		static bool LessFunc( CQueuedPacket * const &lhs, CQueuedPacket * const &rhs )
		{
			return lhs->m_unSendTime > rhs->m_unSendTime;
		}
	};

	CUtlPriorityQueue< CQueuedPacket * > m_QueuedPackets;
	CThreadMutex m_QueuedPacketsCS;
	CThreadEvent m_hThreadEvent;
	volatile bool m_bThreadShouldExit;
};

static CQueuedPacketSender g_LagPacketSender;
IQueuedPacketSender *g_pLagPackedSender = &g_LagPacketSender;


CQueuedPacketSender::CQueuedPacketSender() :
	m_QueuedPackets( 0, 0, CQueuedPacket::LessFunc )
{
	m_bThreadShouldExit = false;
}

CQueuedPacketSender::~CQueuedPacketSender()
{
	Shutdown();
}

bool CQueuedPacketSender::Setup()
{
	return Start();
}

bool CQueuedPacketSender::Start( unsigned nBytesStack )
{
	Shutdown();

	if ( CThread::Start( nBytesStack ) )
	{
		// Ahhh the perfect cross-platformness of the threads library.
#ifdef _WIN32
		SetPriority( THREAD_PRIORITY_HIGHEST );
#elif _LINUX
		//SetPriority( PRIORITY_MAX );
#endif
		m_bThreadShouldExit = false;
		return true;
	}
	else
	{
		return false;
	}
}

void CQueuedPacketSender::Shutdown()
{
	if ( !IsAlive() )
		return;
		
	m_bThreadShouldExit = true;
	m_hThreadEvent.Set();
	
	Join(); // Wait for the thread to exit.

	while ( m_QueuedPackets.Count() > 0 )
	{
		delete m_QueuedPackets.ElementAtHead();
		m_QueuedPackets.RemoveAtHead();
	}
	m_QueuedPackets.Purge();
}

void CQueuedPacketSender::ClearQueuedPacketsForChannel( INetChannel *pChan )
{
	m_QueuedPacketsCS.Lock();

	for ( int i = m_QueuedPackets.Count()-1; i >= 0; i-- )
	{
		CQueuedPacket *p = m_QueuedPackets.Element( i );
		if ( p->m_pChannel == pChan )
		{
			m_QueuedPackets.RemoveAt( i );
			delete p;
		}
	}

	m_QueuedPacketsCS.Unlock();
}

void CQueuedPacketSender::QueuePacket( INetChannel *pChan, SOCKET s, const char *buf, int len, const struct sockaddr * to, int tolen, uint32 msecDelay )
{
	m_QueuedPacketsCS.Lock();

	// We'll pull all packets we should have sent by now and send them out right away
	uint32 msNow = Plat_MSTime();

	int nMaxQueuedPackets = 2048; // @Forgetest: was 1024 
	if ( m_QueuedPackets.Count() < nMaxQueuedPackets )
	{
		// Add this packet to the queue.
		CQueuedPacket *pPacket = new CQueuedPacket;
		pPacket->m_unSendTime = msNow + msecDelay;
		pPacket->m_Socket = s;
		pPacket->m_pChannel = pChan;
		pPacket->buf.CopyArray( (char*)buf, len );
		pPacket->to.CopyArray( (char*)to, tolen );
		m_QueuedPackets.Insert( pPacket );
	}
	else
	{
		// static int nWarnings = 5;
		// if ( --nWarnings > 0 )
		{
			Warning( "CQueuedPacketSender: num queued packets >= nMaxQueuedPackets. Not queueing anymore.\n" );
		}
	}

	// Tell the thread that we have a queued packet.
	m_hThreadEvent.Set();

	m_QueuedPacketsCS.Unlock();
}

extern int NET_SendToImpl( SOCKET s, const char * buf, int len, const struct sockaddr * to, int tolen, int iGameDataLength );

int CQueuedPacketSender::Run()
{
	 // Normally TT_INFINITE but we wakeup every 50ms just in case.
	uint32 waitIntervalNoPackets = 50;
	uint32 waitInterval = waitIntervalNoPackets;
	while ( 1 )
	{
		if ( m_hThreadEvent.Wait( waitInterval ) )
		{
			// Someone signaled the thread. Either we're being told to exit or 
			// we're being told that a packet was just queued.
			if ( m_bThreadShouldExit )
				return 0;
		}

		// Assume nothing to do and that we'll sleep again
		waitInterval = waitIntervalNoPackets;

		// OK, now send a packet.
		m_QueuedPacketsCS.Lock();
		
		// We'll pull all packets we should have sent by now and send them out right away
		uint32 msNow = Plat_MSTime();

		// bool bTrace = net_queue_trace.GetInt() == NET_QUEUED_PACKET_THREAD_DEBUG_VALUE;

		while ( m_QueuedPackets.Count() > 0 )
		{
			CQueuedPacket *pPacket = m_QueuedPackets.ElementAtHead();
			if ( pPacket->m_unSendTime > msNow )
			{
				// Sleep until next we need this packet
				waitInterval = pPacket->m_unSendTime - msNow;
				// if ( bTrace )
				// {
				// 	char sz[ 256 ];
				// 	Q_snprintf( sz, sizeof( sz ), "SQ:  sleeping for %u msecs at %f\n", waitInterval, Plat_FloatTime() );
				// 	Warning( sz );
				// }
				break;
			}

			// If it's a bot, don't do anything. Note: we DO want this code deep here because bots only
			// try to send packets when sv_stressbots is set, in which case we want it to act as closely
			// as a real player as possible.
			sockaddr_in *pInternetAddr = (sockaddr_in*)pPacket->to.Base();
		#ifdef _WIN32
			if ( pInternetAddr->sin_addr.S_un.S_addr != 0
		#else
			if ( pInternetAddr->sin_addr.s_addr != 0 
		#endif
				&& pInternetAddr->sin_port != 0 )
			{		
				// if ( bTrace )
				// {
				// 	char sz[ 256 ];
				// 	Q_snprintf( sz, sizeof( sz ), "SQ:  sending %d bytes at %f\n", pPacket->buf.Count(), Plat_FloatTime() );
				// 	Warning( sz );
				// }

				NET_SendToImpl
				( 
					pPacket->m_Socket, 
					pPacket->buf.Base(), 
					pPacket->buf.Count(), 
					(sockaddr*)pPacket->to.Base(),
					pPacket->to.Count(), 
					NET_QUEUED_PACKET_THREAD_SEND_PACKET 
				);
			}	
			
			delete pPacket;
			m_QueuedPackets.RemoveAtHead();
		}
		
		m_QueuedPacketsCS.Unlock();
	}
}