nevrax.org : docs

00001 
00007 /* Copyright, 2001 Nevrax Ltd.
00008  *
00009  * This file is part of NEVRAX NEL.
00010  * NEVRAX NEL is free software; you can redistribute it and/or modify
00011  * it under the terms of the GNU General Public License as published by
00012  * the Free Software Foundation; either version 2, or (at your option)
00013  * any later version.
00014 
00015  * NEVRAX NEL is distributed in the hope that it will be useful, but
00016  * WITHOUT ANY WARRANTY; without even the implied warranty of
00017  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
00018  * General Public License for more details.
00019 
00020  * You should have received a copy of the GNU General Public License
00021  * along with NEVRAX NEL; see the file COPYING. If not, write to the
00022  * Free Software Foundation, Inc., 59 Temple Place - Suite 330, Boston,
00023  * MA 02111-1307, USA.
00024  */
00025 
00026 #include "stdnet.h"
00027 
00028 #include "nel/misc/hierarchical_timer.h"
00029 
00030 #include "nel/net/buf_server.h"
00031 
00032 #ifdef NL_OS_WINDOWS
00033 #include <winsock2.h>
00034 //typedef sint socklen_t;
00035 
00036 #elif defined NL_OS_UNIX
00037 #include <unistd.h>
00038 #include <sys/types.h>
00039 #include <sys/time.h>
00040 #endif
00041 
00042 
00043 using namespace NLMISC;
00044 using namespace std;
00045 
00046 namespace NLNET {
00047 
00048 
00049 /***************************************************************************************************
00050  * User main thread (initialization)
00051  **************************************************************************************************/
00052 
00053 
00054 /*
00055  * Constructor
00056  */
00057 CBufServer::CBufServer( TThreadStategy strategy,
00058         uint16 max_threads, uint16 max_sockets_per_thread, bool nodelay, bool replaymode ) :
00059         CBufNetBase(),
00060         _NoDelay( nodelay ),
00061         _ThreadStrategy( strategy ),
00062         _MaxThreads( max_threads ),
00063         _MaxSocketsPerThread( max_sockets_per_thread ),
00064         _ListenTask( NULL ),
00065         _ListenThread( NULL ),
00066         _ThreadPool("CBufServer::_ThreadPool"),
00067         _ConnectionCallback( NULL ),
00068         _ConnectionCbArg( NULL ),
00069         _BytesPushedOut( 0 ),
00070         _BytesPoppedIn( 0 ),
00071         _PrevBytesPoppedIn( 0 ),
00072         _PrevBytesPushedOut( 0 ),
00073         _NbConnections (0),
00074         _ReplayMode( replaymode )
00075 {
00076         nlnettrace( "CBufServer::CBufServer" );
00077         if ( ! _ReplayMode )
00078         {
00079                 _ListenTask = new CListenTask( this );
00080                 _ListenThread = IThread::create( _ListenTask );
00081         }
00082         /*{
00083                 CSynchronized<uint32>::CAccessor syncbpi ( &_BytesPushedIn );
00084                 syncbpi.value() = 0;
00085         }*/
00086 }
00087 
00088 
00089 /*
00090  * Listens on the specified port
00091  */
00092 void CBufServer::init( uint16 port )
00093 {
00094         nlnettrace( "CBufServer::init" );
00095         if ( ! _ReplayMode )
00096         {
00097                 _ListenTask->init( port );
00098                 _ListenThread->start();
00099         }
00100         else
00101         {
00102                 nldebug( "LNETL0: Binding listen socket to any address, port %hu", port );
00103         }
00104 }
00105 
00106 
00107 /*
00108  * Begins to listen on the specified port (call before running thread)
00109  */
00110 void CListenTask::init( uint16 port )
00111 {
00112         nlnettrace( "CListenTask::init" );
00113         _ListenSock.init( port );
00114 }
00115 
00116 
00117 /***************************************************************************************************
00118  * User main thread (running)
00119  **************************************************************************************************/
00120 
00121 
00122 /*
00123  * Constructor
00124  */
00125 CServerTask::CServerTask() : _ExitRequired(false)
00126 {
00127 #ifdef NL_OS_UNIX
00128         pipe( _WakeUpPipeHandle );
00129 #endif
00130 }
00131 
00132 
00133 
00134 #ifdef NL_OS_UNIX
00135 /*
00136  * Wake the thread up, when blocked in select (Unix only)
00137  */
00138 void CServerTask::wakeUp()
00139 {
00140         uint8 b;
00141         if ( write( _WakeUpPipeHandle[PipeWrite], &b, 1 ) == -1 )
00142         {
00143                 nldebug( "LNETL1: In CServerTask::wakeUp(): write() failed" );
00144         }
00145 }
00146 #endif
00147 
00148 
00149 /*
00150  * Destructor
00151  */
00152 CServerTask::~CServerTask()
00153 {
00154 #ifdef NL_OS_UNIX
00155         close( _WakeUpPipeHandle[PipeRead] );
00156         close( _WakeUpPipeHandle[PipeWrite] );
00157 #endif
00158 }
00159 
00160 
00161 /*
00162  * Destructor
00163  */
00164 CBufServer::~CBufServer()
00165 {
00166         nlnettrace( "CBufServer::~CBufServer" );
00167 
00168         // Clean listen thread exit
00169         if ( ! _ReplayMode )
00170         {
00171                 ((CListenTask*)(_ListenThread->getRunnable()))->requireExit();
00172                 ((CListenTask*)(_ListenThread->getRunnable()))->close();
00173 #ifdef NL_OS_UNIX
00174                 _ListenTask->wakeUp();
00175 #endif
00176                 _ListenThread->wait();
00177                 delete _ListenThread;
00178                 delete _ListenTask;
00179 
00180                 // Clean receive thread exits
00181                 CThreadPool::iterator ipt;
00182                 {
00183                         nldebug( "LNETL1: Waiting for end of threads..." );
00184                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
00185                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00186                         {
00187                                 // Tell the threads to exit and wake them up
00188                                 CServerReceiveTask *task = receiveTask(ipt);
00189                                 nlnettrace( "Requiring exit" );
00190                                 task->requireExit();
00191 
00192                                 // Wake the threads up
00193         #ifdef NL_OS_UNIX
00194                                 task->wakeUp();
00195         #else
00196                                 CConnections::iterator ipb;
00197                                 nlnettrace( "Closing sockets (Win32)" );
00198                                 {
00199                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
00200                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
00201                                         {
00202                                                 (*ipb)->Sock->close();
00203                                         }
00204                                 }
00205         #endif
00206 
00207                         }
00208                         
00209                         nlnettrace( "Waiting" );
00210                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00211                         {
00212                                 // Wait until the threads have exited
00213                                 (*ipt)->wait();
00214                         }
00215 
00216                         nldebug( "LNETL1: Deleting sockets, tasks and threads..." );
00217                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00218                         {
00219                                 // Delete the socket objects
00220                                 CServerReceiveTask *task = receiveTask(ipt);
00221                                 CConnections::iterator ipb;
00222                                 {
00223                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
00224                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
00225                                         {
00226                                                 delete (*ipb);
00227                                         }
00228                                 }
00229 
00230         #ifdef NL_OS_UNIX
00231                                 // Under Unix, close the sockets now
00232                                 nlnettrace( "Closing sockets (Unix)" );
00233                                 {
00234                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
00235                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
00236                                         {
00237                                                 (*ipb)->Sock->close();
00238                                         }
00239                                 }
00240         #endif
00241                                 
00242                                 // Delete the task objects
00243                                 delete task;
00244 
00245                                 // Delete the thread objects
00246                                 delete (*ipt);
00247                         }
00248                 }
00249         }
00250 
00251         nlnettrace( "Exiting CBufServer::~CBufServer" );
00252 }
00253 
00254 
00255 /*
00256  * Disconnect the specified host
00257  * Set hostid to NULL to disconnect all connections.
00258  * If hostid is not null and the socket is not connected, the method does nothing.
00259  * If quick is true, any pending data will not be sent before disconnecting.
00260  */
00261 void CBufServer::disconnect( TSockId hostid, bool quick )
00262 {
00263         nlnettrace( "CBufServer::disconnect" );
00264         if ( hostid != InvalidSockId )
00265         {
00266                 // Disconnect only if physically connected
00267                 if ( hostid->Sock->connected() )
00268                 {
00269                         if ( ! quick )
00270                         {
00271                                 hostid->flush();
00272                         }
00273                         hostid->Sock->disconnect(); // the connection will be removed by the next call of update()
00274                 }
00275         }
00276         else
00277         {
00278                 // Disconnect all
00279                 CThreadPool::iterator ipt;
00280                 {
00281                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
00282                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00283                         {
00284                                 CServerReceiveTask *task = receiveTask(ipt);
00285                                 CConnections::iterator ipb;
00286                                 {
00287                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
00288                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
00289                                         {
00290                                                 if ( (*ipb)->Sock->connected() )
00291                                                 {
00292                                                         if ( ! quick )
00293                                                         {
00294                                                                 (*ipb)->flush();
00295                                                         }
00296                                                         (*ipb)->Sock->disconnect();
00297                                                 }
00298                                         }
00299                                 }
00300                         }
00301                 }
00302         }
00303 }
00304 
00305 
00306 /*
00307  * Send a message to the specified host
00308  */
00309 void CBufServer::send( const CMemStream& buffer, TSockId hostid )
00310 {
00311         nlnettrace( "CBufServer::send" );
00312         nlassert( buffer.length() > 0);
00313         nlassert( buffer.length() <= maxSentBlockSize() );
00314 
00315         // slow down the layer H_AUTO (CBufServer_send);
00316 
00317         if ( hostid != InvalidSockId )
00318         {
00319                 // debug features, we number all packet to be sure that they are all sent and received
00320                 // \todo remove this debug feature when ok
00321 //              nldebug ("send message number %u", hostid->SendNextValue);
00322 #ifdef NL_BIG_ENDIAN
00323                 uint32 val = NLMISC_BSWAP32(hostid->SendNextValue);
00324 #else
00325                 uint32 val = hostid->SendNextValue;
00326 #endif
00327 
00328                 *(uint32*)buffer.buffer() = val;
00329                 hostid->SendNextValue++;
00330 
00331                 pushBufferToHost( buffer, hostid );
00332         }
00333         else
00334         {
00335                 // Push into all send queues
00336                 CThreadPool::iterator ipt;
00337                 {
00338                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
00339                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00340                         {
00341                                 CServerReceiveTask *task = receiveTask(ipt);
00342                                 CConnections::iterator ipb;
00343                                 {
00344                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
00345                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
00346                                         {
00347                                                 // Send only if the socket is logically connected
00348                                                 if ( (*ipb)->connectedState() ) 
00349                                                 {
00350                                                         // debug features, we number all packet to be sure that they are all sent and received
00351                                                         // \todo remove this debug feature when ok
00352 //                                                      nldebug ("send message number %u", (*ipb)->SendNextValue);
00353 #ifdef NL_BIG_ENDIAN
00354                                                         uint32 val = NLMISC_BSWAP32((*ipb)->SendNextValue);
00355 #else
00356                                                         uint32 val = (*ipb)->SendNextValue;
00357 #endif
00358                                                         *(uint32*)buffer.buffer() = val;
00359                                                         (*ipb)->SendNextValue++;
00360 
00361                                                         pushBufferToHost( buffer, *ipb );
00362                                                 }
00363                                         }
00364                                 }
00365                         }
00366                 }
00367         }
00368 }
00369 
00370 
00371 /*
00372  * Checks if there are some data to receive
00373  */
00374 bool CBufServer::dataAvailable()
00375 {
00376         // slow down the layer H_AUTO (CBufServer_dataAvailable);
00377         {
00378                 /* If no data available, enter the 'while' loop and return false (1 volatile test)
00379                  * If there are user data available, enter the 'while' and return true immediately (1 volatile test + 1 short locking)
00380                  * If there is a connection/disconnection event (rare), call the callback and loop
00381                  */
00382                 while ( dataAvailableFlag() )
00383                 {
00384                         // Because _DataAvailable is true, the receive queue is not empty at this point
00385                         vector<uint8> buffer;
00386                         uint8 val;
00387                         {
00388                                 CFifoAccessor recvfifo( &receiveQueue() );
00389                                 val = recvfifo.value().frontLast();
00390                                 if ( val != CBufNetBase::User )
00391                                 {
00392                                         recvfifo.value().front( buffer );
00393                                 }
00394                         }
00395 
00396                         /*sint32 mbsize = recvfifo.value().size() / 1048576;
00397                         if ( mbsize > 0 )
00398                         {
00399                           nlwarning( "The receive queue size exceeds %d MB", mbsize );
00400                         }*/
00401 
00402                         /*vector<uint8> buffer;
00403                         recvfifo.value().front( buffer );*/
00404 
00405                         // Test if it the next block is a system event
00406                         //switch ( buffer[buffer.size()-1] )
00407                         switch ( val )
00408                         {
00409                                 
00410                         // Normal message available
00411                         case CBufNetBase::User:
00412                                 return true; // return immediatly, do not extract the message
00413 
00414                         // Process disconnection event
00415                         case CBufNetBase::Disconnection:
00416                         {
00417                                 TSockId sockid = *((TSockId*)(&*buffer.begin()));
00418                                 nldebug( "LNETL1: Disconnection event for %p %s", sockid, sockid->asString().c_str());
00419 
00420                                 sockid->setConnectedState( false );
00421 
00422                                 // Call callback if needed
00423                                 if ( disconnectionCallback() != NULL )
00424                                 {
00425                                         disconnectionCallback()( sockid, argOfDisconnectionCallback() );
00426                                 }
00427 
00428                                 // Add socket object into the synchronized remove list
00429                                 nldebug( "LNETL1: Adding the connection to the remove list" );
00430                                 nlassert( ((CServerBufSock*)sockid)->ownerTask() != NULL );
00431                                 ((CServerBufSock*)sockid)->ownerTask()->addToRemoveSet( sockid );
00432                                 break;
00433                         }
00434                         // Process connection event
00435                         case CBufNetBase::Connection:
00436                         {
00437                                 TSockId sockid = *((TSockId*)(&*buffer.begin()));
00438                                 nldebug( "LNETL1: Connection event for %p %s", sockid, sockid->asString().c_str());
00439 
00440                                 sockid->setConnectedState( true );
00441                                 
00442                                 // Call callback if needed
00443                                 if ( connectionCallback() != NULL )
00444                                 {
00445                                         connectionCallback()( sockid, argOfConnectionCallback() );
00446                                 }
00447                                 break;
00448                         }
00449                         default: // should not occur
00450                                 nlinfo( "LNETL1: Invalid block type: %hu (should be = to %hu", (uint16)(buffer[buffer.size()-1]), (uint16)(val) );
00451                                 nlinfo( "LNETL1: Buffer (%d B): [%s]", buffer.size(), stringFromVector(buffer).c_str() );
00452                                 nlinfo( "LNETL1: Receive queue:" );
00453                                 {
00454                                         CFifoAccessor recvfifo( &receiveQueue() );
00455                                         recvfifo.value().display();
00456                                 }
00457                                 nlerror( "LNETL1: Invalid system event type in server receive queue" );
00458 
00459                         }
00460 
00461                         // Extract system event
00462                         {
00463                                 CFifoAccessor recvfifo( &receiveQueue() );
00464                                 recvfifo.value().pop();
00465                                 setDataAvailableFlag( ! recvfifo.value().empty() );
00466                         }
00467                 }
00468                 // _DataAvailable is false here
00469                 return false;
00470         }
00471 }
00472 
00473 
00474 /* // OLD VERSION
00475 bool CBufServer::dataAvailable()
00476 {
00477         // slow down the layer H_AUTO (CBufServer_dataAvailable);
00478         {
00479                 CFifoAccessor recvfifo( &receiveQueue() );
00480                 do
00481                 {
00482                         // Check if the receive queue is empty
00483                         if ( recvfifo.value().empty() )
00484                         {
00485                                 return false;
00486                         }
00487                         else
00488                         {
00489                           //sint32 mbsize = recvfifo.value().size() / 1048576;
00490                           //if ( mbsize > 0 )
00491                             //{
00492                             //  nlwarning( "The receive queue size exceeds %d MB", mbsize );
00493                             //}
00494 
00495                                 uint8 val = recvfifo.value().frontLast();
00496                                 
00497                                 //vector<uint8> buffer;
00498                                 //recvfifo.value().front( buffer );
00499 
00500                                 // Test if it the next block is a system event
00501                                 //switch ( buffer[buffer.size()-1] )
00502                                 switch ( val )
00503                                 {
00504                                         
00505                                 // Normal message available
00506                                 case CBufNetBase::User:
00507                                         return true; // return immediatly, do not extract the message
00508 
00509                                 // Process disconnection event
00510                                 case CBufNetBase::Disconnection:
00511                                 {
00512                                         vector<uint8> buffer;
00513                                         recvfifo.value().front( buffer );
00514 
00515                                         TSockId sockid = *((TSockId*)(&*buffer.begin()));
00516                                         nldebug( "LNETL1: Disconnection event for %p %s", sockid, sockid->asString().c_str());
00517 
00518                                         sockid->setConnectedState( false );
00519 
00520                                         // Call callback if needed
00521                                         if ( disconnectionCallback() != NULL )
00522                                         {
00523                                                 disconnectionCallback()( sockid, argOfDisconnectionCallback() );
00524                                         }
00525 
00526                                         // Add socket object into the synchronized remove list
00527                                         nldebug( "LNETL1: Adding the connection to the remove list" );
00528                                         nlassert( ((CServerBufSock*)sockid)->ownerTask() != NULL );
00529                                         ((CServerBufSock*)sockid)->ownerTask()->addToRemoveSet( sockid );
00530                                         break;
00531                                 }
00532                                 // Process connection event
00533                                 case CBufNetBase::Connection:
00534                                 {
00535                                         vector<uint8> buffer;
00536                                         recvfifo.value().front( buffer );
00537 
00538                                         TSockId sockid = *((TSockId*)(&*buffer.begin()));
00539                                         nldebug( "LNETL1: Connection event for %p %s", sockid, sockid->asString().c_str());
00540 
00541                                         sockid->setConnectedState( true );
00542                                         
00543                                         // Call callback if needed
00544                                         if ( connectionCallback() != NULL )
00545                                         {
00546                                                 connectionCallback()( sockid, argOfConnectionCallback() );
00547                                         }
00548                                         break;
00549                                 }
00550                                 default:
00551                                         vector<uint8> buffer;
00552                                         recvfifo.value().front( buffer );
00553 
00554                                         nlinfo( "LNETL1: Invalid block type: %hu (should be = to %hu", (uint16)(buffer[buffer.size()-1]), (uint16)(val) );
00555                                         nlinfo( "LNETL1: Buffer (%d B): [%s]", buffer.size(), stringFromVector(buffer).c_str() );
00556                                         nlinfo( "LNETL1: Receive queue:" );
00557                                         recvfifo.value().display();
00558                                         nlerror( "LNETL1: Invalid system event type in server receive queue" );
00559 
00560                                 }
00561 
00562                                 // Extract system event
00563                                 recvfifo.value().pop();
00564                         }
00565                 }
00566                 while ( true );
00567         }
00568 }
00569 */
00570  
00571 /*
00572  * Receives next block of data in the specified. The length and hostid are output arguments.
00573  * Precond: dataAvailable() has returned true, phostid not null
00574  */
00575 void CBufServer::receive( CMemStream& buffer, TSockId* phostid )
00576 {
00577         nlnettrace( "CBufServer::receive" );
00578         //nlassert( dataAvailable() );
00579         nlassert( phostid != NULL );
00580         {
00581                 CFifoAccessor recvfifo( &receiveQueue() );
00582                 nlassert( ! recvfifo.value().empty() );
00583                 recvfifo.value().front( buffer );
00584                 recvfifo.value().pop();
00585                 setDataAvailableFlag( ! recvfifo.value().empty() );
00586         }
00587 
00588         // Extract hostid (and event type)
00589         *phostid = *((TSockId*)&(buffer.buffer()[buffer.length()-sizeof(TSockId)-1]));
00590         nlassert( buffer.buffer()[buffer.length()-1] == CBufNetBase::User );
00591 
00592         // debug features, we number all packet to be sure that they are all sent and received
00593         // \todo remove this debug feature when ok
00594 #ifdef NL_BIG_ENDIAN
00595         uint32 val = NLMISC_BSWAP32(*(uint32*)buffer.buffer());
00596 #else
00597         uint32 val = *(uint32*)buffer.buffer();
00598 #endif
00599 
00600         //      nldebug ("receive message number %u", val);
00601         if ((*phostid)->ReceiveNextValue != val)
00602         {
00603                 nlwarning (("LNETL1: !!!LOST A MESSAGE!!! I received the message number %u but I'm waiting the message number %u (cnx %s), warn lecroart@nevrax.com with the log now please", val, (*phostid)->ReceiveNextValue, (*phostid)->asString().c_str()));
00604                 // resync the message number
00605                 (*phostid)->ReceiveNextValue = val;
00606         }
00607 
00608         (*phostid)->ReceiveNextValue++;
00609 
00610         buffer.resize( buffer.length()-sizeof(TSockId)-1 );
00611 
00612         // TODO OPTIM remove the nldebug for speed
00613         //commented for optimisation nldebug( "LNETL1: Read buffer (%d+%d B) from %s", buffer.length(), sizeof(TSockId)+1, /*stringFromVector(buffer).c_str(), */(*phostid)->asString().c_str() );
00614 
00615         // Statistics
00616         _BytesPoppedIn += buffer.length() + sizeof(TBlockSize);
00617 }
00618 
00619 
00620 /*
00621  * Update the network (call this method evenly)
00622  */
00623 void CBufServer::update()
00624 {
00625         //nlnettrace( "CBufServer::update-BEGIN" );
00626 
00627         _NbConnections = 0;
00628 
00629         // For each thread
00630         CThreadPool::iterator ipt;
00631         {
00632           //nldebug( "UPD: Acquiring the Thread Pool" );
00633                 CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
00634                 //nldebug( "UPD: Acquired." );
00635                 for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00636                 {
00637                         // For each thread of the pool
00638                         CServerReceiveTask *task = receiveTask(ipt);
00639                         CConnections::iterator ipb;
00640                         {
00641                                 CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
00642                                 for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
00643                                 {
00644                                     // For each socket of the thread, update sending
00645                                     if ( ! ((*ipb)->Sock->connected() && (*ipb)->update()) )
00646                                     {
00647                                                 // Update did not work or the socket is not connected anymore
00648                                         nldebug( "LNETL1: Socket %s is disconnected", (*ipb)->asString().c_str() );
00649                                                 // Disconnection event if disconnected (known either from flush (in update) or when receiving data)
00650                                                 (*ipb)->advertiseDisconnection( this, *ipb );
00651                                         
00652                                                 /*if ( (*ipb)->advertiseDisconnection( this, *ipb ) )
00653                                                 {
00654                                                         // Now the connection removal is in dataAvailable()
00655                                                         // POLL6
00656                                                 }*/
00657                                     }
00658                                     else
00659                                     {
00660                                                 _NbConnections++;
00661                                     }
00662                                 }
00663                         }
00664                 }
00665         }
00666 
00667         //nlnettrace( "CBufServer::update-END" );
00668 }
00669 
00670 uint32 CBufServer::getSendQueueSize( TSockId destid )
00671 {
00672         if ( destid != InvalidSockId )
00673         {
00674                 return destid->SendFifo.size();
00675         }
00676         else
00677         {
00678                 // add all client buffers
00679 
00680                 uint32 total = 0;
00681 
00682                 // For each thread
00683                 CThreadPool::iterator ipt;
00684                 {
00685                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
00686                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00687                         {
00688                                 // For each thread of the pool
00689                                 CServerReceiveTask *task = receiveTask(ipt);
00690                                 CConnections::iterator ipb;
00691                                 {
00692                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
00693                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
00694                                         {
00695                                                 // For each socket of the thread, update sending
00696                                                 total = (*ipb)->SendFifo.size ();
00697                                         }
00698                                 }
00699                         }
00700                 }
00701                 return total;
00702         }
00703 }
00704 
00705 
00706 /*
00707  * Returns the number of bytes received since the previous call to this method
00708  */
00709 uint64 CBufServer::newBytesReceived()
00710 {
00711         uint64 b = bytesReceived();
00712         uint64 nbrecvd = b - _PrevBytesPoppedIn;
00713         //nlinfo( "b: %"NL_I64"u   new: %"NL_I64"u", b, nbrecvd );
00714         _PrevBytesPoppedIn = b;
00715         return nbrecvd;
00716 }
00717 
00718 /*
00719  * Returns the number of bytes sent since the previous call to this method
00720  */
00721 uint64 CBufServer::newBytesSent()
00722 {
00723         uint64 b = bytesSent();
00724         uint64 nbsent = b - _PrevBytesPushedOut;
00725         //nlinfo( "b: %"NL_I64"u   new: %"NL_I64"u", b, nbsent );
00726         _PrevBytesPushedOut = b;
00727         return nbsent;
00728 }
00729 
00730 
00731 /***************************************************************************************************
00732  * Listen thread
00733  **************************************************************************************************/
00734 
00735 
00736 /*
00737  * Code of listening thread
00738  */
00739 void CListenTask::run()
00740 {
00741         nlnettrace( "CListenTask::run" );
00742 
00743 #ifdef NL_OS_UNIX
00744         SOCKET descmax;
00745         fd_set readers;
00746         timeval tv;
00747         descmax = _ListenSock.descriptor()>_WakeUpPipeHandle[PipeRead]?_ListenSock.descriptor():_WakeUpPipeHandle[PipeRead];
00748 #endif
00749 
00750         // Accept connections
00751         while ( ! exitRequired() )
00752         {
00753                 try
00754                 {
00755                         // Get and setup the new socket
00756 #ifdef NL_OS_UNIX
00757                         FD_ZERO( &readers );
00758                         FD_SET( _ListenSock.descriptor(), &readers );
00759                         FD_SET( _WakeUpPipeHandle[PipeRead], &readers );
00760                         tv.tv_sec = 60; 
00761                         tv.tv_usec = 0;
00762                         int res = ::select( descmax+1, &readers, NULL, NULL, &tv );
00763 
00764                         switch ( res )
00765                         {
00766                         case  0 : continue; // time-out expired, no results
00767                         case -1 :
00768                                 // we'll ignore message (Interrupted system call) caused by a CTRL-C
00769                                 if (CSock::getLastError() == 4)
00770                                 {
00771                                         nldebug ("LNETL1: Select failed (in listen thread): %s (code %u) but IGNORED", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError());
00772                                         continue;
00773                                 }
00774                                 nlerror( "LNETL1: Select failed (in listen thread): %s (code %u)", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError() );
00775                         }
00776 
00777                         if ( FD_ISSET( _WakeUpPipeHandle[PipeRead], &readers ) )
00778                         {
00779                                 uint8 b;
00780                                 if ( read( _WakeUpPipeHandle[PipeRead], &b, 1 ) == -1 ) // we were woken-up by the wake-up pipe
00781                                 {
00782                                         nldebug( "LNETL1: In CListenTask::run(): read() failed" );
00783                                 }
00784                                 nldebug( "LNETL1: listen thread select woken-up" );
00785                                 continue;
00786                         }
00787 #endif
00788                         nldebug( "LNETL1: Waiting incoming connection..." );
00789                         CServerBufSock *bufsock = new CServerBufSock( _ListenSock.accept() );
00790                         nldebug( "New connection : %s", bufsock->asString().c_str() );
00791                         bufsock->Sock->setNonBlockingMode( true );
00792                         if ( _Server->noDelay() )
00793                         {
00794                                 bufsock->Sock->setNoDelay( true );
00795                         }
00796 
00797                         // Notify the new connection
00798                         bufsock->advertiseConnection( _Server );
00799 
00800                         // Dispatch the socket into the thread pool
00801                         _Server->dispatchNewSocket( bufsock );
00802                 }
00803                 catch ( ESocket& e )
00804                 {
00805                         nlinfo( "Exception in listen thread: %s", e.what() ); // It can occur in normal behavior (e.g. when exiting)
00806                         // It can also occur when too many sockets are open (e.g. 885 connections)
00807                 }
00808         }
00809 
00810         nlnettrace( "Exiting CListenTask::run" );
00811 }
00812 
00813 
00814 /*
00815  * Binds a new socket and send buffer to an existing or a new thread
00816  * Note: this method is called in the listening thread.
00817  */
00818 void CBufServer::dispatchNewSocket( CServerBufSock *bufsock )
00819 {
00820         nlnettrace( "CBufServer::dispatchNewSocket" );
00821 
00822         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
00823         if ( _ThreadStrategy == SpreadSockets ) 
00824         {
00825                 // Find the thread with the smallest number of connections and check if all
00826                 // threads do not have the same number of connections
00827                 uint min = 0xFFFFFFFF;
00828                 uint max = 0;
00829                 CThreadPool::iterator ipt, iptmin, iptmax;
00830                 for ( iptmin=iptmax=ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00831                 {
00832                         uint noc = receiveTask(ipt)->numberOfConnections();
00833                         if ( noc < min )
00834                         {
00835                                 min = noc;
00836                                 iptmin = ipt;
00837                         }
00838                         if ( noc > max )
00839                         {
00840                                 max = noc;
00841                                 iptmax = ipt;
00842                         }
00843                 }
00844 
00845                 // Check if we make the pool of threads grow (if we have not found vacant room
00846                 // and if it is allowed to)
00847                 if ( (poolsync.value().empty()) ||
00848                          ((min == max) && (poolsync.value().size() < _MaxThreads)) )
00849                 {
00850                         addNewThread( poolsync.value(), bufsock );
00851                 }
00852                 else
00853                 {
00854                         // Dispatch socket to an existing thread of the pool
00855                         CServerReceiveTask *task = receiveTask(iptmin);
00856                         bufsock->setOwnerTask( task );
00857                         task->addNewSocket( bufsock );
00858 #ifdef NL_OS_UNIX
00859                         task->wakeUp();
00860 #endif                  
00861                         
00862                         if ( min >= (uint)_MaxSocketsPerThread )
00863                         {
00864                                 nlwarning( "LNETL1: Exceeding the maximum number of sockets per thread" );
00865                         }
00866                         nldebug( "LNETL1: New socket dispatched to thread %d", iptmin-poolsync.value().begin() );
00867                 }
00868 
00869         }
00870         else // _ThreadStrategy == FillThreads
00871         {
00872                 CThreadPool::iterator ipt;
00873                 for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
00874                 {
00875                         uint noc = receiveTask(ipt)->numberOfConnections();
00876                         if ( noc < _MaxSocketsPerThread )
00877                         {
00878                                 break;
00879                         }
00880                 }
00881 
00882                 // Check if we have to make the thread pool grow (if we have not found vacant room)
00883                 if ( ipt == poolsync.value().end() )
00884                 {
00885                         if ( poolsync.value().size() == _MaxThreads )
00886                         {
00887                                 nlwarning( "LNETL1: Exceeding the maximum number of threads" );
00888                         }
00889                         addNewThread( poolsync.value(), bufsock );
00890                 }
00891                 else
00892                 {
00893                         // Dispatch socket to an existing thread of the pool
00894                         CServerReceiveTask *task = receiveTask(ipt);
00895                         bufsock->setOwnerTask( task );
00896                         task->addNewSocket( bufsock );
00897 #ifdef NL_OS_UNIX
00898                         task->wakeUp();
00899 #endif                  
00900                         nldebug( "LNETL1: New socket dispatched to thread %d", ipt-poolsync.value().begin() );
00901                 }
00902         }
00903 }
00904 
00905 
00906 /*
00907  * Creates a new task and run a new thread for it
00908  * Precond: bufsock not null
00909  */
00910 void CBufServer::addNewThread( CThreadPool& threadpool, CServerBufSock *bufsock )
00911 {
00912         nlnettrace( "CBufServer::addNewThread" );
00913         nlassert( bufsock != NULL );
00914 
00915         // Create new task and dispatch the socket to it
00916         CServerReceiveTask *task = new CServerReceiveTask( this );
00917         bufsock->setOwnerTask( task );
00918         task->addNewSocket( bufsock );
00919 
00920         // Add a new thread to the pool, with this task
00921         IThread *thr = IThread::create( task );
00922         {
00923                 threadpool.push_back( thr );
00924                 thr->start();
00925                 nldebug( "LNETL1: Added a new thread; pool size is %d", threadpool.size() );
00926                 nldebug( "LNETL1: New socket dispatched to thread %d", threadpool.size()-1 );
00927         }
00928 }
00929 
00930 
00931 /***************************************************************************************************
00932  * Receive threads
00933  **************************************************************************************************/
00934 
00935 
00936 /*
00937  * Code of receiving threads for servers
00938  */
00939 void CServerReceiveTask::run()
00940 {
00941         nlnettrace( "CServerReceiveTask::run" );
00942 
00943         SOCKET descmax;
00944         fd_set readers;
00945 
00946         // Time-out value for select (it can be long because we do not do any thing else in this thread)
00947         timeval tv;
00948 #if defined NL_OS_UNIX
00949         // POLL7
00950         nice( 2 );
00951 #endif // NL_OS_UNIX
00952         
00953         // Copy of _Connections
00954         vector<TSockId> connections_copy;       
00955 
00956         while ( ! exitRequired() )
00957         {
00958                 // 1. Remove closed connections
00959                 clearClosedConnections();
00960 
00961                 // POLL8
00962 
00963                 // 2-SELECT-VERSION : select() on the sockets handled in the present thread
00964 
00965                 descmax = 0;
00966                 FD_ZERO( &readers );
00967                 bool skip;
00968                 bool alldisconnected = true;
00969                 CConnections::iterator ipb;
00970                 {
00971                         // Lock _Connections
00972                         CSynchronized<CConnections>::CAccessor connectionssync( &_Connections );
00973 
00974                         // Prepare to avoid select if there is no connection
00975                         skip = connectionssync.value().empty();
00976 
00977                         // Fill the select array and copy _Connections
00978                         connections_copy.clear();
00979                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
00980                         {
00981                                 if ( (*ipb)->Sock->connected() ) // exclude disconnected sockets that are not deleted
00982                                 {
00983                                         alldisconnected = false;
00984                                         // Copy _Connections element
00985                                         connections_copy.push_back( *ipb );
00986 
00987                                         // Add socket descriptor to the select array
00988                                         FD_SET( (*ipb)->Sock->descriptor(), &readers );
00989 
00990                                         // Calculate descmax for select
00991                                         if ( (*ipb)->Sock->descriptor() > descmax )
00992                                         {
00993                                                 descmax = (*ipb)->Sock->descriptor();
00994                                         }
00995                                 }
00996                         }
00997 
00998 #ifdef NL_OS_UNIX
00999                         // Add the wake-up pipe into the select array
01000                         FD_SET( _WakeUpPipeHandle[PipeRead], &readers );
01001                         if ( _WakeUpPipeHandle[PipeRead]>descmax )
01002                         {
01003                                 descmax = _WakeUpPipeHandle[PipeRead];
01004                         }
01005 #endif
01006                         
01007                         // Unlock _Connections, use connections_copy instead
01008                 }
01009 
01010 #ifndef NL_OS_UNIX
01011                 // Avoid select if there is no connection (Windows only)
01012                 if ( skip || alldisconnected )
01013                 {
01014                         nlSleep( 1 ); // nice
01015                         continue;
01016                 }
01017 #endif
01018 
01019 #ifdef NL_OS_WINDOWS
01020                 tv.tv_sec = 0; // short time because the newly added connections can't be added to the select fd_set
01021                 tv.tv_usec = 10000; // NEW: set to 500ms because otherwise new connections handling are too slow
01022 #elif defined NL_OS_UNIX
01023                 // POLL7
01024                 tv.tv_sec = 3600;               // 1 hour (=> 1 select every 3.6 second for 1000 connections)
01025                 tv.tv_usec = 0;
01026 #endif // NL_OS_WINDOWS
01027 
01028                 // Call select
01029                 int res = ::select( descmax+1, &readers, NULL, NULL, &tv );
01030 
01031                 // POLL9
01032 
01033                 // 3. Test the result
01034                 switch ( res )
01035                 {
01036                         case  0 : continue; // time-out expired, no results
01037 
01039                         case -1 :
01040                                 // we'll ignore message (Interrupted system call) caused by a CTRL-C
01041                                 /*if (CSock::getLastError() == 4)
01042                                 {
01043                                         nldebug ("LNETL1: Select failed (in receive thread): %s (code %u) but IGNORED", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError());
01044                                         continue;
01045                                 }*/
01046                                 //nlerror( "LNETL1: Select failed (in receive thread): %s (code %u)", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError() );
01047                                 nldebug( "LNETL1: Select failed (in receive thread): %s (code %u)", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError() );
01048                                 return;
01049                 }
01050 
01051                 // 4. Get results
01052 
01053                 vector<TSockId>::iterator ic;
01054                 for ( ic=connections_copy.begin(); ic!=connections_copy.end(); ++ic )
01055                 {
01056                         if ( FD_ISSET( (*ic)->Sock->descriptor(), &readers ) != 0 )
01057                         {
01058                                 CServerBufSock *serverbufsock = static_cast<CServerBufSock*>(static_cast<CBufSock*>(*ic));
01059                                 try
01060                                 {
01061                                         // 4. Receive data
01062                                         if ( serverbufsock->receivePart() )
01063                                         {
01064                                                 // Copy sockid
01065                                                 vector<uint8> hidvec;
01066                                                 hidvec.resize( sizeof(TSockId)+1 );
01067                                                 memcpy( &*hidvec.begin(), &(*ic), sizeof(TSockId) );
01068 
01069                                                 // Add event type to hidvec
01070                                                 hidvec[sizeof(TSockId)] = (uint8)CBufNetBase::User;
01071 
01072                                                 // Push message into receive queue
01073                                                 //uint32 bufsize;
01074                                                 //sint32 mbsize;
01075                                                 {
01076                                                         //nldebug( "RCV: Acquiring the receive queue... ");
01077                                                         CFifoAccessor recvfifo( &_Server->receiveQueue() );
01078                                                         //nldebug( "RCV: Acquired, pushing the received buffer... ");
01079                                                         recvfifo.value().push( serverbufsock->receivedBuffer(), hidvec );
01080                                                         _Server->setDataAvailableFlag( true );
01081                                                         //nldebug( "RCV: Pushed, releasing the receive queue..." );
01082                                                         //recvfifo.value().display();
01083                                                         //bufsize = serverbufsock->receivedBuffer().size();
01084                                                         //mbsize = recvfifo.value().size() / 1048576;
01085                                                 }
01086                                                 //nldebug( "RCV: Released." );
01087                                                 /*if ( mbsize > 1 )
01088                                                 {
01089                                                         nlwarning( "The receive queue size exceeds %d MB", mbsize );
01090                                                 }*/
01091                                                 /*
01092                                                 // Statistics
01093                                                 {
01094                                                         CSynchronized<uint32>::CAccessor syncbpi ( &_Server->syncBytesPushedIn() );
01095                                                         syncbpi.value() += bufsize;
01096                                                 }
01097                                                 */
01098                                         }
01099                                 }
01100                                 catch ( ESocketConnectionClosed& )
01101                                 {
01102                                         nldebug( "LNETL1: Connection %s closed", serverbufsock->asString().c_str() );
01103                                         // The socket went to _Connected=false when throwing the exception
01104                                 }
01105                                 catch ( ESocket& )
01106                                 {
01107                                         nldebug( "LNETL1: Connection %s broken", serverbufsock->asString().c_str() );
01108                                         (*ic)->Sock->disconnect();
01109                                 }
01110 /*
01111 #ifdef NL_OS_UNIX
01112                                 skip = true; // don't check _WakeUpPipeHandle (yes, check it to read any written byte)
01113 #endif
01114 
01115 */
01116                         }
01117 
01118                 }
01119 
01120 #ifdef NL_OS_UNIX
01121                 // Test wake-up pipe
01122                 if ( (!skip) && (FD_ISSET( _WakeUpPipeHandle[PipeRead], &readers )) )
01123                 {
01124                         uint8 b;
01125                         if ( read( _WakeUpPipeHandle[PipeRead], &b, 1 ) == -1 ) // we were woken-up by the wake-up pipe
01126                         {
01127                                 nldebug( "LNETL1: In CServerReceiveTask::run(): read() failed" );
01128                         }
01129                         nldebug( "LNETL1: Receive thread select woken-up" );
01130                 }
01131 #endif
01132                 
01133         }
01134         nlnettrace( "Exiting CServerReceiveTask::run" );
01135 }
01136 
01137 
01138 /*
01139  * Delete all connections referenced in the remove list (double-mutexed)
01140  */
01141 
01142 void CServerReceiveTask::clearClosedConnections()
01143 {
01144         CConnections::iterator ic;
01145         {
01146                 NLMISC::CSynchronized<CConnections>::CAccessor removesetsync( &_RemoveSet );
01147                 {
01148                         if ( ! removesetsync.value().empty() )
01149                         {
01150                                 // Delete closed connections
01151                                 NLMISC::CSynchronized<CConnections>::CAccessor connectionssync( &_Connections );
01152                                 for ( ic=removesetsync.value().begin(); ic!=removesetsync.value().end(); ++ic )
01153                                 {
01154                                         nldebug( "LNETL1: Removing a connection" );
01155 
01156                                         TSockId sid = (*ic);
01157 
01158                                         // Remove from the connection list
01159                                         connectionssync.value().erase( *ic );
01160 
01161                                         // Delete the socket object
01162                                         delete sid;
01163                                 }
01164                                 // Clear remove list
01165                                 removesetsync.value().clear();
01166                         }
01167                 }
01168         }
01169 }
01170 
01171 
01172 } // NLNET
buf_server.cpp
