khanat-opennel-code/code/nel/src/net/buf_client.cpp

471 lines
12 KiB
C++

// NeL - MMORPG Framework <http://dev.ryzom.com/projects/nel/>
// Copyright (C) 2010 Winch Gate Property Limited
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#include "stdnet.h"
#include "nel/misc/hierarchical_timer.h"
#include "nel/net/buf_client.h"
#include "nel/misc/thread.h"
#include "nel/net/dummy_tcp_sock.h"
#include "nel/net/net_log.h"
#ifdef NL_OS_WINDOWS
# define NOMINMAX
# include <windows.h>
#elif defined NL_OS_UNIX
# include <netinet/in.h>
#endif
using namespace NLMISC;
using namespace std;
namespace NLNET {
uint32 NbClientReceiveTask = 0;
/***************************************************************************************************
* User main thread (initialization)
**************************************************************************************************/
/*
* Constructor
*/
#ifdef NL_OS_UNIX
CBufClient::CBufClient( bool nodelay, bool replaymode, bool initPipeForDataAvailable ) :
CBufNetBase( initPipeForDataAvailable ),
#else
CBufClient::CBufClient( bool nodelay, bool replaymode, bool ) :
CBufNetBase(),
#endif
_NoDelay( nodelay ),
_PrevBytesDownloaded( 0 ),
_PrevBytesUploaded( 0 ),
_RecvTask( NULL ),
_RecvThread( NULL )
/*_PrevBytesReceived( 0 ),
_PrevBytesSent( 0 )*/
{
nlnettrace( "CBufClient::CBufClient" ); // don't define a global object
if ( replaymode )
{
_BufSock = new CNonBlockingBufSock( new CDummyTcpSock(), CBufNetBase::DefaultMaxExpectedBlockSize );
}
else
{
_BufSock = new CNonBlockingBufSock( NULL, CBufNetBase::DefaultMaxExpectedBlockSize );
_RecvTask = new CClientReceiveTask( this, _BufSock );
}
}
/*
* Connects to the specified host
* Precond: not connected
*/
void CBufClient::connect( const CInetAddress& addr )
{
nlnettrace( "CBufClient::connect" );
nlassert( ! _BufSock->Sock->connected() );
_BufSock->setMaxExpectedBlockSize( maxExpectedBlockSize() );
_BufSock->connect( addr, _NoDelay, true );
_BufSock->setNonBlocking(); // ADDED: non-blocking client connection
_PrevBytesDownloaded = 0;
_PrevBytesUploaded = 0;
/*_PrevBytesReceived = 0;
_PrevBytesSent = 0;*/
// Allow reconnection
if ( _RecvThread != NULL )
{
delete _RecvThread;
}
_RecvThread = IThread::create( _RecvTask, 1024*4*4 );
_RecvThread->start();
}
/***************************************************************************************************
* User main thread (running)
**************************************************************************************************/
void CBufClient::displayThreadStat (NLMISC::CLog *log)
{
log->displayNL ("client thread %p nbloop %d", _RecvTask, _RecvTask->NbLoop);
}
/*
* Sends a message to the remote host
*/
void CBufClient::send( const NLMISC::CMemStream& buffer )
{
nlnettrace( "CBufClient::send" );
nlassert( buffer.length() > 0 );
nlassert( buffer.length() <= maxSentBlockSize() );
// slow down the layer H_AUTO (CBufServer_send);
if ( ! _BufSock->pushBuffer( buffer ) )
{
// Disconnection event if disconnected
_BufSock->advertiseDisconnection( this, NULL );
}
}
/*
* Checks if there are some data to receive
*/
bool CBufClient::dataAvailable()
{
// slow down the layer H_AUTO (CBufClient_dataAvailable);
{
/* If no data available, enter the 'while' loop and return false (1 volatile test)
* If there are user data available, enter the 'while' and return true immediately (1 volatile test + 1 short locking)
* If there is a disconnection event (rare), call the callback and loop
*/
while ( dataAvailableFlag() )
{
// Because _DataAvailable is true, the receive queue is not empty at this point
uint8 val;
{
CFifoAccessor recvfifo( &receiveQueue() );
val = recvfifo.value().frontLast ();
}
#ifdef NL_OS_UNIX
uint8 b;
if ( read( _DataAvailablePipeHandle[PipeRead], &b, 1 ) == -1 )
nlwarning( "LNETL1: Read pipe failed in dataAvailable" );
//nldebug( "Pipe: 1 byte read (client %p)", this );
#endif
// Test if it the next block is a system event
switch ( val )
{
// Normal message available
case CBufNetBase::User:
return true; // return immediatly, do not extract the message
// Process disconnection event
case CBufNetBase::Disconnection:
LNETL1_DEBUG( "LNETL1: Disconnection event" );
_BufSock->setConnectedState( false );
// Call callback if needed
if ( disconnectionCallback() != NULL )
{
disconnectionCallback()( id(), argOfDisconnectionCallback() );
}
// Unlike the server version, we do not delete the CBufSock object here,
// it will be done in the destructor of CBufClient
break;
default: // should not occur
{
CFifoAccessor recvfifo( &receiveQueue() );
vector<uint8> buffer;
recvfifo.value().front (buffer);
LNETL1_INFO( "LNETL1: Invalid block type: %hu (should be = %hu)", (uint16)(buffer[buffer.size()-1]), (uint16)val );
LNETL1_INFO( "LNETL1: Buffer (%d B): [%s]", buffer.size(), stringFromVector(buffer).c_str() );
LNETL1_INFO( "LNETL1: Receive queue:" );
recvfifo.value().display();
nlerror( "LNETL1: Invalid system event type in client receive queue" );
}
}
// Extract system event
{
CFifoAccessor recvfifo( &receiveQueue() );
recvfifo.value().pop();
setDataAvailableFlag( ! recvfifo.value().empty() );
}
}
// _DataAvailable is false here
return false;
}
}
#ifdef NL_OS_UNIX
/* Wait until the receive queue contains something to read (implemented with a select()).
* This is where the connection/disconnection callbacks can be called.
* \param usecMax Max time to wait in microsecond (up to 1 sec)
*/
void CBufClient::sleepUntilDataAvailable( uint usecMax )
{
// Prevent looping infinitely if the system time was changed
if ( usecMax > 999999 ) // limit not told in Linux man but here: http://docs.hp.com/en/B9106-90009/select.2.html
usecMax = 999999;
fd_set readers;
timeval tv;
do
{
FD_ZERO( &readers );
FD_SET( _DataAvailablePipeHandle[PipeRead], &readers );
tv.tv_sec = 0;
tv.tv_usec = usecMax;
int res = ::select( _DataAvailablePipeHandle[PipeRead]+1, &readers, NULL, NULL, &tv );
if ( res == -1 )
nlerror( "LNETL1: Select failed in sleepUntilDataAvailable (code %u)", CSock::getLastError() );
}
while ( ! dataAvailable() ); // will loop if only a connection/disconnection event was read
}
#endif
/*
* Receives next block of data in the specified buffer (resizes the vector)
* Precond: dataAvailable() has returned true
*/
void CBufClient::receive( NLMISC::CMemStream& buffer )
{
nlnettrace( "CBufClient::receive" );
//nlassert( dataAvailable() );
// Extract buffer from the receive queue
{
CFifoAccessor recvfifo( &receiveQueue() );
nlassert( ! recvfifo.value().empty() );
recvfifo.value().front( buffer );
recvfifo.value().pop();
setDataAvailableFlag( ! recvfifo.value().empty() );
}
// Extract event type
nlassert( buffer.buffer()[buffer.size()-1] == CBufNetBase::User );
//commented for optimisation LNETL1_DEBUG( "LNETL1: Client read buffer (%d+%d B)", buffer.size(), sizeof(TSockId)+1 );
buffer.resize( buffer.size()-1 );
}
/*
* Update the network (call this method evenly)
*/
void CBufClient::update()
{
//nlnettrace( "CBufClient::update" );
// Update sending
bool sendingok = _BufSock->update();
// Disconnection event if disconnected
if ( ! ( _BufSock->Sock->connected() && sendingok ) )
{
if ( _BufSock->Sock->connected() )
{
_BufSock->Sock->disconnect();
}
_BufSock->advertiseDisconnection( this, NULL );
}
}
/*
* Disconnect the remote host
*/
void CBufClient::disconnect( bool quick )
{
nlnettrace( "CBufClient::disconnect" );
// Do not allow to disconnect a socket that is not connected
nlassert( _BufSock->connectedState() );
// When the NS tells us to remove this connection AND the connection has physically
// disconnected but not yet logically (i.e. disconnection event not processed yet),
// skip flushing and physical active disconnection
if ( _BufSock->Sock->connected() )
{
// Flush sending is asked for
if ( ! quick )
{
_BufSock->flush();
}
// Disconnect and prevent from advertising the disconnection
_BufSock->disconnect( false );
}
// Empty the receive queue
{
CFifoAccessor recvfifo( &receiveQueue() );
recvfifo.value().clear();
setDataAvailableFlag( false );
}
}
// Utility function for newBytes...()
inline uint64 updateStatCounter( uint64& counter, uint64 newvalue )
{
uint64 result = newvalue - counter;
counter = newvalue;
return result;
}
/*
* Returns the number of bytes downloaded since the previous call to this method
*/
uint64 CBufClient::newBytesDownloaded()
{
return updateStatCounter( _PrevBytesDownloaded, bytesDownloaded() );
}
/*
* Returns the number of bytes uploaded since the previous call to this method
*/
uint64 CBufClient::newBytesUploaded()
{
return updateStatCounter( _PrevBytesUploaded, bytesUploaded() );
}
/*
* Returns the number of bytes popped by receive() since the previous call to this method
*/
/*uint64 CBufClient::newBytesReceived()
{
return updateStatCounter( _PrevBytesReceived, bytesReceived() );
}*/
/*
* Returns the number of bytes pushed by send() since the previous call to this method
*/
/*uint64 CBufClient::newBytesSent()
{
return updateStatCounter( _PrevBytesSent, bytesSent() );
}*/
/*
* Destructor
*/
CBufClient::~CBufClient()
{
nlnettrace( "CBufClient::~CBufClient" );
// Disconnect if not done
if ( _BufSock->Sock->connected() )
{
nlassert( _BufSock->connectedState() );
disconnect( true );
}
// Clean thread termination
if ( _RecvThread != NULL )
{
LNETL1_DEBUG( "LNETL1: Waiting for the end of the receive thread..." );
_RecvThread->wait();
}
if ( _RecvTask != NULL )
delete _RecvTask;
if ( _RecvThread != NULL )
delete _RecvThread;
if ( _BufSock != NULL )
delete _BufSock;
nlnettrace( "Exiting CBufClient::~CBufClient" );
}
/***************************************************************************************************
* Receive thread
**************************************************************************************************/
/*
* Code of receiving thread for clients
*/
void CClientReceiveTask::run()
{
NbClientReceiveTask++;
NbNetworkTask++;
nlnettrace( "CClientReceiveTask::run" );
// 18/08/2005 : sonix : Changed time out from 60s to 1s, in some case, it
// can generate a 60 s wait on destruction of the CBufSock
// By the way, checking every 1s is not a time consuming
_NBBufSock->Sock->setTimeOutValue( 1, 0 );
bool connected = true;
while ( connected && _NBBufSock->Sock->connected())
{
try
{
// ADDED: non-blocking client connection
// Wait until some data are received (sleepin' select inside)
while ( ! _NBBufSock->Sock->dataAvailable() )
{
if ( ! _NBBufSock->Sock->connected() )
{
LNETL1_DEBUG( "LNETL1: Client connection %s closed", sockId()->asString().c_str() );
// The socket went to _Connected=false when throwing the exception
connected = false;
break;
}
}
// Process the data received
if ( _NBBufSock->receivePart( 1 ) ) // 1 for the event type
{
//commented out for optimisation: LNETL1_DEBUG( "LNETL1: Client %s received buffer (%u bytes)", _SockId->asString().c_str(), buffer.size()/*, stringFromVector(buffer).c_str()*/ );
// Add event type
_NBBufSock->fillEventTypeOnly();
// Push message into receive queue
_Client->pushMessageIntoReceiveQueue( _NBBufSock->receivedBuffer() );
}
NbLoop++;
}
catch (const ESocket&)
{
LNETL1_DEBUG( "LNETL1: Client connection %s broken", sockId()->asString().c_str() );
sockId()->Sock->disconnect();
connected = false;
}
}
nlnettrace( "Exiting CClientReceiveTask::run()" );
NbClientReceiveTask--;
NbNetworkTask--;
}
NLMISC_CATEGORISED_VARIABLE(nel, uint32, NbClientReceiveTask, "Number of client receive thread");
} // NLNET