nel/src/net/buf_server.cpp

   1 // NeL - MMORPG Framework <http://dev.ryzom.com/projects/nel/>
   2 // Copyright (C) 2010  Winch Gate Property Limited
   3 //
   4 // This source file has been modified by the following contributors:
   5 // Copyright (C) 2014-2016  Jan BOON (Kaetemi) <jan.boon@kaetemi.be>
   6 //
   7 // This program is free software: you can redistribute it and/or modify
   8 // it under the terms of the GNU Affero General Public License as
   9 // published by the Free Software Foundation, either version 3 of the
  10 // License, or (at your option) any later version.
  11 //
  12 // This program is distributed in the hope that it will be useful,
  13 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 // GNU Affero General Public License for more details.
  16 //
  17 // You should have received a copy of the GNU Affero General Public License
  18 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19
  20 #include "stdnet.h"
  21
  22 #include "nel/misc/hierarchical_timer.h"
  23
  24 #include "nel/net/buf_server.h"
  25 #include "nel/net/net_log.h"
  26
  27 #ifdef NL_OS_WINDOWS
  28 #       ifndef NL_COMP_MINGW
  29 #               define NOMINMAX
  30 #       endif
  31 #       include <windows.h>
  32 #elif defined NL_OS_UNIX
  33 #       include <unistd.h>
  34 #       include <cerrno>
  35 #       include <sys/types.h>
  36 #       include <sys/time.h>
  37 #endif
  38
  39 /*
  40  * On Linux, the default limit of descriptors is usually 1024, you can increase it with ulimit
  41  */
  42
  43 using namespace NLMISC;
  44 using namespace std;
  45
  46 namespace NLNET {
  47
  48 uint32  NbServerListenTask = 0;
  49 uint32  NbServerReceiveTask = 0;
  50
  51 /***************************************************************************************************
  52  * User main thread (initialization)
  53  **************************************************************************************************/
  54
  55 /*
  56  * Constructor
  57  */
  58 CBufServer::CBufServer( TThreadStategy strategy,
  59         uint16 max_threads, uint16 max_sockets_per_thread, bool nodelay, bool replaymode, bool initPipeForDataAvailable ) :
  60 #ifdef NL_OS_UNIX
  61         CBufNetBase( initPipeForDataAvailable ),
  62 #else
  63         CBufNetBase(),
  64 #endif
  65         _ThreadStrategy( strategy ),
  66         _MaxThreads( max_threads ),
  67         _MaxSocketsPerThread( max_sockets_per_thread ),
  68         _ListenTask( NULL ),
  69         _ListenThread( NULL ),
  70         _ThreadPool("CBufServer::_ThreadPool"),
  71         _ConnectionCallback( NULL ),
  72         _ConnectionCbArg( NULL ),
  73         _BytesPushedOut( 0 ),
  74         _BytesPoppedIn( 0 ),
  75         _PrevBytesPoppedIn( 0 ),
  76         _PrevBytesPushedOut( 0 ),
  77         _NbConnections (0),
  78         _NoDelay( nodelay ),
  79         _ReplayMode( replaymode )
  80 {
  81         nlnettrace( "CBufServer::CBufServer" );
  82         if ( ! _ReplayMode )
  83         {
  84                 _ListenTask = new CListenTask( this );
  85                 _ListenThread = IThread::create( _ListenTask, 1024*4*4 );
  86         }
  87         /*{
  88                 CSynchronized<uint32>::CAccessor syncbpi ( &_BytesPushedIn );
  89                 syncbpi.value() = 0;
  90         }*/
  91 }
  92
  93
  94 /*
  95  * Listens on the specified port
  96  */
  97 void CBufServer::init( uint16 port )
  98 {
  99         nlnettrace( "CBufServer::init" );
 100         if ( ! _ReplayMode )
 101         {
 102                 _ListenTask->init( port, maxExpectedBlockSize() );
 103                 _ListenThread->start();
 104         }
 105         else
 106         {
 107                 LNETL1_DEBUG( "LNETL1: Binding listen socket to any address, port %hu", port );
 108         }
 109 }
 110
 111
 112 /*
 113  * Begins to listen on the specified port (call before running thread)
 114  */
 115 void CListenTask::init( uint16 port, sint32 maxExpectedBlockSize )
 116 {
 117         nlnettrace( "CListenTask::init" );
 118         _ListenSock.init( port );
 119         _MaxExpectedBlockSize = maxExpectedBlockSize;
 120 }
 121
 122
 123 /***************************************************************************************************
 124  * User main thread (running)
 125  **************************************************************************************************/
 126
 127
 128 /*
 129  * Constructor
 130  */
 131 CServerTask::CServerTask() : NbLoop (0), _ExitRequired(false)
 132 {
 133 #ifdef NL_OS_UNIX
 134         if (pipe( _WakeUpPipeHandle ) == -1)
 135         {
 136                 nlwarning("LNETL1: pipe() failed: code=%d '%s'", errno, strerror(errno));
 137         }
 138 #endif
 139 }
 140
 141
 142
 143 #ifdef NL_OS_UNIX
 144 /*
 145  * Wake the thread up, when blocked in select (Unix only)
 146  */
 147 void CServerTask::wakeUp()
 148 {
 149         uint8 b;
 150         if ( write( _WakeUpPipeHandle[PipeWrite], &b, 1 ) == -1 )
 151         {
 152                 LNETL1_DEBUG( "LNETL1: In CServerTask::wakeUp(): write() failed" );
 153         }
 154 }
 155 #endif
 156
 157
 158 /*
 159  * Destructor
 160  */
 161 CServerTask::~CServerTask()
 162 {
 163 #ifdef NL_OS_UNIX
 164         close( _WakeUpPipeHandle[PipeRead] );
 165         close( _WakeUpPipeHandle[PipeWrite] );
 166 #endif
 167 }
 168
 169
 170 /*
 171  * Destructor
 172  */
 173 CBufServer::~CBufServer()
 174 {
 175         nlnettrace( "CBufServer::~CBufServer" );
 176
 177         // Clean listen thread exit
 178         if ( ! _ReplayMode )
 179         {
 180                 ((CListenTask*)(_ListenThread->getRunnable()))->requireExit();
 181                 ((CListenTask*)(_ListenThread->getRunnable()))->close();
 182 #ifdef NL_OS_UNIX
 183                 _ListenTask->wakeUp();
 184 #endif
 185                 _ListenThread->wait();
 186                 delete _ListenThread;
 187                 delete _ListenTask;
 188
 189                 // Clean receive thread exits
 190                 CThreadPool::iterator ipt;
 191                 {
 192                         LNETL1_DEBUG( "LNETL1: Waiting for end of threads..." );
 193                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
 194                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 195                         {
 196                                 // Tell the threads to exit and wake them up
 197                                 CServerReceiveTask *task = receiveTask(ipt);
 198                                 nlnettrace( "Requiring exit" );
 199                                 task->requireExit();
 200
 201                                 // Wake the threads up
 202         #ifdef NL_OS_UNIX
 203                                 task->wakeUp();
 204         #else
 205                                 CConnections::iterator ipb;
 206                                 nlnettrace( "Disconnecting sockets (Win32)" );
 207                                 {
 208                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
 209                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
 210                                         {
 211                                                 (*ipb)->Sock->disconnect();
 212                                         }
 213                                 }
 214         #endif
 215
 216                         }
 217
 218                         nlnettrace( "Waiting" );
 219                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 220                         {
 221                                 // Wait until the threads have exited
 222                                 (*ipt)->wait();
 223                         }
 224
 225                         LNETL1_DEBUG( "LNETL1: Deleting sockets, tasks and threads..." );
 226                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 227                         {
 228                                 // Delete the socket objects
 229                                 CServerReceiveTask *task = receiveTask(ipt);
 230                                 CConnections::iterator ipb;
 231                                 {
 232                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
 233                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
 234                                         {
 235                                                 delete (*ipb); // closes and deletes the socket
 236                                         }
 237                                 }
 238
 239                                 // Delete the task objects
 240                                 delete task;
 241
 242                                 // Delete the thread objects
 243                                 delete (*ipt);
 244                         }
 245                 }
 246         }
 247
 248         nlnettrace( "Exiting CBufServer::~CBufServer" );
 249 }
 250
 251
 252 /*
 253  * Disconnect the specified host
 254  * Set hostid to NULL to disconnect all connections.
 255  * If hostid is not null and the socket is not connected, the method does nothing.
 256  * If quick is true, any pending data will not be sent before disconnecting.
 257  */
 258 void CBufServer::disconnect( TSockId hostid, bool quick )
 259 {
 260         nlnettrace( "CBufServer::disconnect" );
 261         if ( hostid != InvalidSockId )
 262         {
 263                 if (_ConnectedClients.find(hostid) == _ConnectedClients.end())
 264                 {
 265                         // this host is not connected
 266                         return;
 267                 }
 268
 269                 // Disconnect only if physically connected
 270                 if ( hostid->Sock->connected() )
 271                 {
 272                         if ( ! quick )
 273                         {
 274                                 hostid->flush();
 275                         }
 276                         hostid->Sock->disconnect(); // the connection will be removed by the next call of update()
 277                 }
 278         }
 279         else
 280         {
 281                 // Disconnect all
 282                 CThreadPool::iterator ipt;
 283                 {
 284                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
 285                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 286                         {
 287                                 CServerReceiveTask *task = receiveTask(ipt);
 288                                 CConnections::iterator ipb;
 289                                 {
 290                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
 291                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
 292                                         {
 293                                                 if ( (*ipb)->Sock->connected() )
 294                                                 {
 295                                                         if ( ! quick )
 296                                                         {
 297                                                                 (*ipb)->flush();
 298                                                         }
 299                                                         (*ipb)->Sock->disconnect();
 300                                                 }
 301                                         }
 302                                 }
 303                         }
 304                 }
 305         }
 306 }
 307
 308
 309 /*
 310  * Send a message to the specified host
 311  */
 312 void CBufServer::send( const CMemStream& buffer, TSockId hostid )
 313 {
 314         nlnettrace( "CBufServer::send" );
 315         nlassert( buffer.length() > 0 );
 316         nlassertex( buffer.length() <= maxSentBlockSize(), ("length=%u max=%u", buffer.length(), maxSentBlockSize()) );
 317
 318         // slow down the layer H_AUTO (CBufServer_send);
 319
 320         if ( hostid != InvalidSockId )
 321         {
 322                 if (_ConnectedClients.find(hostid) == _ConnectedClients.end())
 323                 {
 324                         // this host is not connected
 325                         return;
 326                 }
 327
 328                 pushBufferToHost( buffer, hostid );
 329         }
 330         else
 331         {
 332                 // Push into all send queues
 333                 CThreadPool::iterator ipt;
 334                 {
 335                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
 336                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 337                         {
 338                                 CServerReceiveTask *task = receiveTask(ipt);
 339                                 CConnections::iterator ipb;
 340                                 {
 341                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
 342                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
 343                                         {
 344                                                 // Send only if the socket is logically connected
 345                                                 if ( (*ipb)->connectedState() )
 346                                                 {
 347                                                         pushBufferToHost( buffer, *ipb );
 348                                                 }
 349                                         }
 350                                 }
 351                         }
 352                 }
 353         }
 354 }
 355
 356
 357 /*
 358  * Checks if there are some data to receive
 359  */
 360 bool CBufServer::dataAvailable()
 361 {
 362         // slow down the layer H_AUTO (CBufServer_dataAvailable);
 363         {
 364                 /* If no data available, enter the 'while' loop and return false (1 volatile test)
 365                  * If there are user data available, enter the 'while' and return true immediately (1 volatile test + 1 short locking)
 366                  * If there is a connection/disconnection event (rare), call the callback and loop
 367                  */
 368                 while ( dataAvailableFlag() )
 369                 {
 370                         // Because _DataAvailable is true, the receive queue is not empty at this point
 371                         vector<uint8> buffer;
 372                         uint8 val;
 373                         {
 374                                 CFifoAccessor recvfifo( &receiveQueue() );
 375                                 val = recvfifo.value().frontLast();
 376                                 if ( val != CBufNetBase::User )
 377                                 {
 378                                         recvfifo.value().front( buffer );
 379                                 }
 380                         }
 381
 382                         /*sint32 mbsize = recvfifo.value().size() / 1048576;
 383                         if ( mbsize > 0 )
 384                         {
 385                           nlwarning( "The receive queue size exceeds %d MB", mbsize );
 386                         }*/
 387
 388                         /*vector<uint8> buffer;
 389                         recvfifo.value().front( buffer );*/
 390
 391 #ifdef NL_OS_UNIX
 392                         uint8 b;
 393                         if ( read( _DataAvailablePipeHandle[PipeRead], &b, 1 ) == -1 )
 394                                 nlwarning( "LNETL1: Read pipe failed in dataAvailable" );
 395                         //nldebug( "Pipe: 1 byte read (server %p)", this );
 396 #endif
 397
 398                         // Test if it the next block is a system event
 399                         //switch ( buffer[buffer.size()-1] )
 400                         switch ( val )
 401                         {
 402
 403                         // Normal message available
 404                         case CBufNetBase::User:
 405                                 {
 406                                         return true; // return immediately, do not extract the message
 407                                 }
 408
 409                         // Process disconnection event
 410                         case CBufNetBase::Disconnection:
 411                                 {
 412                                         TSockId sockid = *((TSockId*)(&*buffer.begin()));
 413                                         LNETL1_DEBUG( "LNETL1: Disconnection event for %p %s", sockid, sockid->asString().c_str());
 414
 415                                         sockid->setConnectedState( false );
 416
 417                                         // Call callback if needed
 418                                         if ( disconnectionCallback() != NULL )
 419                                         {
 420                                                 disconnectionCallback()( sockid, argOfDisconnectionCallback() );
 421                                         }
 422
 423                                         // remove from the list of valid client
 424                                         nlverify(_ConnectedClients.erase(sockid) == 1);
 425
 426                                         // Add socket object into the synchronized remove list
 427                                         LNETL1_DEBUG( "LNETL1: Adding the connection to the remove list" );
 428                                         nlassert( ((CServerBufSock*)sockid)->ownerTask() != NULL );
 429                                         ((CServerBufSock*)sockid)->ownerTask()->addToRemoveSet( sockid );
 430                                         break;
 431                                 }
 432                         // Process connection event
 433                         case CBufNetBase::Connection:
 434                                 {
 435                                         TSockId sockid = *((TSockId*)(&*buffer.begin()));
 436                                         LNETL1_DEBUG( "LNETL1: Connection event for %p %s", sockid, sockid->asString().c_str());
 437
 438                                         // add this socket in the list of client
 439                                         nlverify(_ConnectedClients.insert(sockid).second);
 440
 441                                         sockid->setConnectedState( true );
 442
 443                                         // Call callback if needed
 444                                         if ( connectionCallback() != NULL )
 445                                         {
 446                                                 connectionCallback()( sockid, argOfConnectionCallback() );
 447                                         }
 448                                         break;
 449                                 }
 450                         default: // should not occur
 451                                 LNETL1_INFO( "LNETL1: Invalid block type: %hu (should be = to %hu", (uint16)(buffer[buffer.size()-1]), (uint16)(val) );
 452                                 LNETL1_INFO( "LNETL1: Buffer (%d B): [%s]", buffer.size(), stringFromVector(buffer).c_str() );
 453                                 LNETL1_INFO( "LNETL1: Receive queue:" );
 454                                 {
 455                                         CFifoAccessor recvfifo( &receiveQueue() );
 456                                         recvfifo.value().display();
 457                                 }
 458                                 nlerror( "LNETL1: Invalid system event type in server receive queue" );
 459
 460                         }
 461
 462                         // Extract system event
 463                         {
 464                                 CFifoAccessor recvfifo( &receiveQueue() );
 465                                 recvfifo.value().pop();
 466                                 setDataAvailableFlag( ! recvfifo.value().empty() );
 467                         }
 468                 }
 469                 // _DataAvailable is false here
 470                 return false;
 471         }
 472 }
 473
 474
 475 #ifdef NL_OS_UNIX
 476 /* Wait until the receive queue contains something to read (implemented with a select()).
 477  * This is where the connection/disconnection callbacks can be called.
 478  * \param usecMax Max time to wait in microsecond (up to 1 sec)
 479  */
 480 void    CBufServer::sleepUntilDataAvailable( uint usecMax )
 481 {
 482         // Prevent looping infinitely if the system time was changed
 483         if ( usecMax > 999999 ) // limit not told in Linux man but here: http://docs.hp.com/en/B9106-90009/select.2.html
 484                 usecMax = 999999;
 485
 486         fd_set readers;
 487         timeval tv;
 488         do
 489         {
 490                 FD_ZERO( &readers );
 491                 FD_SET( _DataAvailablePipeHandle[PipeRead], &readers );
 492                 tv.tv_sec = 0;
 493                 tv.tv_usec = usecMax;
 494                 int res = ::select( _DataAvailablePipeHandle[PipeRead]+1, &readers, NULL, NULL, &tv );
 495                 if ( res == -1 )
 496                         nlerror( "LNETL1: Select failed in sleepUntilDataAvailable (code %u)", CSock::getLastError() );
 497         }
 498         while ( ! dataAvailable() ); // will loop if only a connection/disconnection event was read
 499 }
 500 #endif
 501
 502
 503 /*
 504  * Receives next block of data in the specified. The length and hostid are output arguments.
 505  * Precond: dataAvailable() has returned true, phostid not null
 506  */
 507 void CBufServer::receive( CMemStream& buffer, TSockId* phostid )
 508 {
 509         nlnettrace( "CBufServer::receive" );
 510         //nlassert( dataAvailable() );
 511         nlassert( phostid != NULL );
 512
 513         {
 514                 CFifoAccessor recvfifo( &receiveQueue() );
 515                 nlassert( ! recvfifo.value().empty() );
 516                 recvfifo.value().front( buffer );
 517                 recvfifo.value().pop();
 518                 setDataAvailableFlag( ! recvfifo.value().empty() );
 519         }
 520
 521         // Extract hostid (and event type)
 522         *phostid = *((TSockId*)&(buffer.buffer()[buffer.size()-sizeof(TSockId)-1]));
 523         nlassert( buffer.buffer()[buffer.size()-1] == CBufNetBase::User );
 524
 525         buffer.resize( buffer.size()-sizeof(TSockId)-1 );
 526
 527         // TODO OPTIM remove the nldebug for speed
 528         //commented for optimisation LNETL1_DEBUG( "LNETL1: Read buffer (%d+%d B) from %s", buffer.size(), sizeof(TSockId)+1, /*stringFromVector(buffer).c_str(), */(*phostid)->asString().c_str() );
 529
 530         // Statistics
 531         _BytesPoppedIn += buffer.size() + sizeof(TBlockSize);
 532 }
 533
 534
 535 /*
 536  * Update the network (call this method evenly)
 537  */
 538 void CBufServer::update()
 539 {
 540         //nlnettrace( "CBufServer::update-BEGIN" );
 541
 542         _NbConnections = 0;
 543
 544         // For each thread
 545         CThreadPool::iterator ipt;
 546         {
 547           //nldebug( "UPD: Acquiring the Thread Pool" );
 548                 CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
 549                 //nldebug( "UPD: Acquired." );
 550                 for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 551                 {
 552                         // For each thread of the pool
 553                         CServerReceiveTask *task = receiveTask(ipt);
 554                         CConnections::iterator ipb;
 555                         {
 556                                 CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
 557                                 for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
 558                                 {
 559                                     // For each socket of the thread, update sending
 560                                     if ( ! ((*ipb)->Sock->connected() && (*ipb)->update()) )
 561                                     {
 562                                                 // Update did not work or the socket is not connected anymore
 563                                         LNETL1_DEBUG( "LNETL1: Socket %s is disconnected", (*ipb)->asString().c_str() );
 564                                                 // Disconnection event if disconnected (known either from flush (in update) or when receiving data)
 565                                                 (*ipb)->advertiseDisconnection( this, *ipb );
 566
 567                                                 /*if ( (*ipb)->advertiseDisconnection( this, *ipb ) )
 568                                                 {
 569                                                         // Now the connection removal is in dataAvailable()
 570                                                         // POLL6
 571                                                 }*/
 572                                     }
 573                                     else
 574                                     {
 575                                                 _NbConnections++;
 576                                     }
 577                                 }
 578                         }
 579                 }
 580         }
 581
 582         //nlnettrace( "CBufServer::update-END" );
 583 }
 584
 585 uint32 CBufServer::getSendQueueSize( TSockId destid )
 586 {
 587         if ( destid != InvalidSockId )
 588         {
 589                 if (_ConnectedClients.find(destid) == _ConnectedClients.end())
 590                 {
 591                         // this host is not connected
 592                         return 0;
 593                 }
 594
 595                 return destid->SendFifo.size();
 596         }
 597         else
 598         {
 599                 // add all client buffers
 600
 601                 uint32 total = 0;
 602
 603                 // For each thread
 604                 CThreadPool::iterator ipt;
 605                 {
 606                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
 607                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 608                         {
 609                                 // For each thread of the pool
 610                                 CServerReceiveTask *task = receiveTask(ipt);
 611                                 CConnections::iterator ipb;
 612                                 {
 613                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
 614                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
 615                                         {
 616                                                 // For each socket of the thread, update sending
 617                                                 total = (*ipb)->SendFifo.size ();
 618                                         }
 619                                 }
 620                         }
 621                 }
 622                 return total;
 623         }
 624 }
 625
 626 void CBufServer::displayThreadStat (NLMISC::CLog *log)
 627 {
 628         // For each thread
 629         CThreadPool::iterator ipt;
 630         {
 631                 CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
 632                 for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 633                 {
 634                         // For each thread of the pool
 635                         CServerReceiveTask *task = receiveTask(ipt);
 636                         // For each socket of the thread, update sending
 637                         log->displayNL ("server receive thread %p nbloop %d", task, task->NbLoop);
 638                 }
 639         }
 640
 641         log->displayNL ("server listen thread %p nbloop %d", _ListenTask, _ListenTask->NbLoop);
 642 }
 643
 644 void CBufServer::setTimeFlushTrigger( TSockId destid, sint32 ms )
 645 {
 646         nlassert( destid != InvalidSockId );
 647         if (_ConnectedClients.find(destid) != _ConnectedClients.end())
 648                 destid->setTimeFlushTrigger( ms );
 649 }
 650
 651 void CBufServer::setSizeFlushTrigger( TSockId destid, sint32 size )
 652 {
 653         nlassert( destid != InvalidSockId );
 654         if (_ConnectedClients.find(destid) != _ConnectedClients.end())
 655                 destid->setSizeFlushTrigger( size );
 656 }
 657
 658 bool CBufServer::flush( TSockId destid, uint *nbBytesRemaining)
 659 {
 660         nlassert( destid != InvalidSockId );
 661         if (_ConnectedClients.find(destid) != _ConnectedClients.end())
 662                 return destid->flush( nbBytesRemaining );
 663         else
 664                 return true;
 665 }
 666 const CInetAddress& CBufServer::hostAddress( TSockId hostid )
 667 {
 668         nlassert( hostid != InvalidSockId );
 669         if (_ConnectedClients.find(hostid) != _ConnectedClients.end())
 670                 return hostid->Sock->remoteAddr();
 671
 672         static CInetAddress nullAddr;
 673         return nullAddr;
 674 }
 675
 676 void CBufServer::displaySendQueueStat (NLMISC::CLog *log, TSockId destid)
 677 {
 678         if ( destid != InvalidSockId )
 679         {
 680                 if (_ConnectedClients.find(destid) == _ConnectedClients.end())
 681                 {
 682                         // this host is not connected
 683                         return;
 684                 }
 685
 686                 destid->SendFifo.displayStats(log);
 687         }
 688         else
 689         {
 690                 // add all client buffers
 691
 692                 // For each thread
 693                 CThreadPool::iterator ipt;
 694                 {
 695                         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
 696                         for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 697                         {
 698                                 // For each thread of the pool
 699                                 CServerReceiveTask *task = receiveTask(ipt);
 700                                 CConnections::iterator ipb;
 701                                 {
 702                                         CSynchronized<CConnections>::CAccessor connectionssync( &task->_Connections );
 703                                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
 704                                         {
 705                                                 // For each socket of the thread, update sending
 706                                                 (*ipb)->SendFifo.displayStats(log);
 707                                         }
 708                                 }
 709                         }
 710                 }
 711         }
 712 }
 713
 714
 715 /*
 716  * Returns the number of bytes received since the previous call to this method
 717  */
 718 uint64 CBufServer::newBytesReceived()
 719 {
 720         uint64 b = bytesReceived();
 721         uint64 nbrecvd = b - _PrevBytesPoppedIn;
 722         //nlinfo( "b: %" NL_I64 "u   new: %" NL_I64 "u", b, nbrecvd );
 723         _PrevBytesPoppedIn = b;
 724         return nbrecvd;
 725 }
 726
 727 /*
 728  * Returns the number of bytes sent since the previous call to this method
 729  */
 730 uint64 CBufServer::newBytesSent()
 731 {
 732         uint64 b = bytesSent();
 733         uint64 nbsent = b - _PrevBytesPushedOut;
 734         //nlinfo( "b: %" NL_I64 "u   new: %" NL_I64 "u", b, nbsent );
 735         _PrevBytesPushedOut = b;
 736         return nbsent;
 737 }
 738
 739
 740 /***************************************************************************************************
 741  * Listen thread
 742  **************************************************************************************************/
 743
 744
 745 /*
 746  * Code of listening thread
 747  */
 748 void CListenTask::run()
 749 {
 750         NbNetworkTask++;
 751         NbServerListenTask++;
 752
 753         nlnettrace( "CListenTask::run" );
 754
 755         fd_set readers;
 756 #ifdef NL_OS_UNIX
 757         SOCKET descmax;
 758         descmax = _ListenSock.descriptor()>_WakeUpPipeHandle[PipeRead]?_ListenSock.descriptor():_WakeUpPipeHandle[PipeRead];
 759 #endif
 760
 761         // Accept connections
 762         while ( ! exitRequired() )
 763         {
 764                 try
 765                 {
 766                         LNETL1_DEBUG( "LNETL1: Waiting incoming connection..." );
 767                         // Get and setup the new socket
 768 #ifdef NL_OS_UNIX
 769                         FD_ZERO( &readers );
 770                         FD_SET( _ListenSock.descriptor(), &readers );
 771                         FD_SET( _WakeUpPipeHandle[PipeRead], &readers );
 772                         int res = ::select( descmax+1, &readers, NULL, NULL, NULL ); /// Wait indefinitely
 773
 774                         switch ( res )
 775                         {
 776                         //case  0 : continue; // time-out expired, no results
 777                         case -1 :
 778                                 // we'll ignore message (Interrupted system call) caused by a CTRL-C
 779                                 if (CSock::getLastError() == 4)
 780                                 {
 781                                         LNETL1_DEBUG ("LNETL1: Select failed (in listen thread): %s (code %u) but IGNORED", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError());
 782                                         continue;
 783                                 }
 784                                 nlerror( "LNETL1: Select failed (in listen thread): %s (code %u)", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError() );
 785                         }
 786
 787                         if ( FD_ISSET( _WakeUpPipeHandle[PipeRead], &readers ) )
 788                         {
 789                                 uint8 b;
 790                                 if ( read( _WakeUpPipeHandle[PipeRead], &b, 1 ) == -1 ) // we were woken-up by the wake-up pipe
 791                                 {
 792                                         LNETL1_DEBUG( "LNETL1: In CListenTask::run(): read() failed" );
 793                                 }
 794                                 LNETL1_DEBUG( "LNETL1: listen thread select woken-up" );
 795                                 continue;
 796                         }
 797 #elif defined (NL_OS_WINDOWS)
 798                         FD_ZERO( &readers );
 799                         FD_SET( _ListenSock.descriptor(), &readers );
 800                         int res = ::select( 1, &readers, NULL, NULL, NULL ); /// Wait indefinitely
 801
 802                         if ( res == -1)
 803                         {
 804                                 nlerror( "LNETL1: Select failed (in listen thread): %s (code %u)", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError() );
 805                                 continue;
 806                         }
 807 #endif
 808                         LNETL1_DEBUG( "LNETL1: Accepting an incoming connection..." );
 809                         CTcpSock *newSock = _ListenSock.accept();
 810                         if (newSock != NULL)
 811                         {
 812                                 CServerBufSock *bufsock = new CServerBufSock( newSock );
 813                                 LNETL1_DEBUG( "LNETL1: New connection : %s", bufsock->asString().c_str() );
 814                                 bufsock->setNonBlocking();
 815                                 bufsock->setMaxExpectedBlockSize( _MaxExpectedBlockSize );
 816                                 if ( _Server->noDelay() )
 817                                 {
 818                                         bufsock->Sock->setNoDelay( true );
 819                                 }
 820
 821                                 // Notify the new connection
 822                                 bufsock->advertiseConnection( _Server );
 823
 824                                 // Dispatch the socket into the thread pool
 825                                 _Server->dispatchNewSocket( bufsock );
 826                         }
 827
 828                         NbLoop++;
 829                 }
 830                 catch (const ESocket &e)
 831                 {
 832                         LNETL1_INFO( "LNETL1: Exception in listen thread: %s", e.what() );
 833                         // It can occur when too many sockets are open (e.g. 885 connections)
 834                 }
 835         }
 836
 837         nlnettrace( "Exiting CListenTask::run" );
 838         NbServerListenTask--;
 839         NbNetworkTask--;
 840 }
 841
 842 /// Close listening socket
 843 void CListenTask::close()
 844 {
 845         _ListenSock.close();
 846 //      _ListenSock.disconnect();
 847 }
 848
 849
 850 /*
 851  * Binds a new socket and send buffer to an existing or a new thread
 852  * Note: this method is called in the listening thread.
 853  */
 854 void CBufServer::dispatchNewSocket( CServerBufSock *bufsock )
 855 {
 856         nlnettrace( "CBufServer::dispatchNewSocket" );
 857
 858         CSynchronized<CThreadPool>::CAccessor poolsync( &_ThreadPool );
 859         if ( _ThreadStrategy == SpreadSockets )
 860         {
 861                 // Find the thread with the smallest number of connections and check if all
 862                 // threads do not have the same number of connections
 863                 uint min = 0xFFFFFFFF;
 864                 uint max = 0;
 865                 CThreadPool::iterator ipt, iptmin, iptmax;
 866                 for ( iptmin=iptmax=ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 867                 {
 868                         uint noc = receiveTask(ipt)->numberOfConnections();
 869                         if ( noc < min )
 870                         {
 871                                 min = noc;
 872                                 iptmin = ipt;
 873                         }
 874                         if ( noc > max )
 875                         {
 876                                 max = noc;
 877                                 iptmax = ipt;
 878                         }
 879                 }
 880
 881                 // Check if we make the pool of threads grow (if we have not found vacant room
 882                 // and if it is allowed to)
 883                 if ( (poolsync.value().empty()) ||
 884                          ((min == max) && (poolsync.value().size() < _MaxThreads)) )
 885                 {
 886                         addNewThread( poolsync.value(), bufsock );
 887                 }
 888                 else
 889                 {
 890                         // Dispatch socket to an existing thread of the pool
 891                         CServerReceiveTask *task = receiveTask(iptmin);
 892                         bufsock->setOwnerTask( task );
 893                         task->addNewSocket( bufsock );
 894 #ifdef NL_OS_UNIX
 895                         task->wakeUp();
 896 #endif
 897
 898                         if ( min >= (uint)_MaxSocketsPerThread )
 899                         {
 900                                 nlwarning( "LNETL1: Exceeding the maximum number of sockets per thread" );
 901                         }
 902                         LNETL1_DEBUG( "LNETL1: New socket dispatched to thread %d", iptmin-poolsync.value().begin() );
 903                 }
 904
 905         }
 906         else // _ThreadStrategy == FillThreads
 907         {
 908                 CThreadPool::iterator ipt;
 909                 for ( ipt=poolsync.value().begin(); ipt!=poolsync.value().end(); ++ipt )
 910                 {
 911                         uint noc = receiveTask(ipt)->numberOfConnections();
 912                         if ( noc < _MaxSocketsPerThread )
 913                         {
 914                                 break;
 915                         }
 916                 }
 917
 918                 // Check if we have to make the thread pool grow (if we have not found vacant room)
 919                 if ( ipt == poolsync.value().end() )
 920                 {
 921                         if ( poolsync.value().size() == _MaxThreads )
 922                         {
 923                                 nlwarning( "LNETL1: Exceeding the maximum number of threads" );
 924                         }
 925                         addNewThread( poolsync.value(), bufsock );
 926                 }
 927                 else
 928                 {
 929                         // Dispatch socket to an existing thread of the pool
 930                         CServerReceiveTask *task = receiveTask(ipt);
 931                         bufsock->setOwnerTask( task );
 932                         task->addNewSocket( bufsock );
 933 #ifdef NL_OS_UNIX
 934                         task->wakeUp();
 935 #endif
 936                         LNETL1_DEBUG( "LNETL1: New socket dispatched to thread %d", ipt-poolsync.value().begin() );
 937                 }
 938         }
 939 }
 940
 941
 942 /*
 943  * Creates a new task and run a new thread for it
 944  * Precond: bufsock not null
 945  */
 946 void CBufServer::addNewThread( CThreadPool& threadpool, CServerBufSock *bufsock )
 947 {
 948         nlnettrace( "CBufServer::addNewThread" );
 949         nlassert( bufsock != NULL );
 950
 951         // Create new task and dispatch the socket to it
 952         CServerReceiveTask *task = new CServerReceiveTask( this );
 953         bufsock->setOwnerTask( task );
 954         task->addNewSocket( bufsock );
 955
 956         // Add a new thread to the pool, with this task
 957         IThread *thr = IThread::create( task, 1024*4*4 );
 958         {
 959                 threadpool.push_back( thr );
 960                 thr->start();
 961                 LNETL1_DEBUG( "LNETL1: Added a new thread; pool size is %d", threadpool.size() );
 962                 LNETL1_DEBUG( "LNETL1: New socket dispatched to thread %d", threadpool.size()-1 );
 963         }
 964 }
 965
 966
 967 /***************************************************************************************************
 968  * Receive threads
 969  **************************************************************************************************/
 970
 971
 972 /*
 973  * Code of receiving threads for servers
 974  */
 975 void CServerReceiveTask::run()
 976 {
 977         NbNetworkTask++;
 978         NbServerReceiveTask++;
 979         nlnettrace( "CServerReceiveTask::run" );
 980
 981         SOCKET descmax;
 982         fd_set readers;
 983
 984 #if defined NL_OS_UNIX
 985         // POLL7
 986         if (nice( 2 ) == -1) // is this really useful as long as select() sleeps?
 987         {
 988                 nlwarning("LNETL1: nice() failed: code=%d '%s'", errno, strerror(errno));
 989         }
 990 #endif // NL_OS_UNIX
 991
 992         // Copy of _Connections
 993         vector<TSockId> connections_copy;
 994
 995         while ( ! exitRequired() )
 996         {
 997                 // 1. Remove closed connections
 998                 clearClosedConnections();
 999
1000                 // POLL8
1001
1002                 // 2-SELECT-VERSION : select() on the sockets handled in the present thread
1003
1004                 descmax = 0;
1005                 FD_ZERO( &readers );
1006                 bool skip;
1007                 bool alldisconnected = true;
1008                 CConnections::iterator ipb;
1009                 {
1010                         // Lock _Connections
1011                         CSynchronized<CConnections>::CAccessor connectionssync( &_Connections );
1012
1013                         // Prepare to avoid select if there is no connection
1014                         skip = connectionssync.value().empty();
1015
1016                         // Fill the select array and copy _Connections
1017                         connections_copy.clear();
1018                         for ( ipb=connectionssync.value().begin(); ipb!=connectionssync.value().end(); ++ipb )
1019                         {
1020                                 if ( (*ipb)->Sock->connected() ) // exclude disconnected sockets that are not deleted
1021                                 {
1022                                         alldisconnected = false;
1023                                         // Copy _Connections element
1024                                         connections_copy.push_back( *ipb );
1025
1026                                         // Add socket descriptor to the select array
1027                                         FD_SET( (*ipb)->Sock->descriptor(), &readers );
1028
1029                                         // Calculate descmax for select
1030                                         if ( (*ipb)->Sock->descriptor() > descmax )
1031                                         {
1032                                                 descmax = (*ipb)->Sock->descriptor();
1033                                         }
1034                                 }
1035                         }
1036
1037 #ifdef NL_OS_UNIX
1038                         // Add the wake-up pipe into the select array
1039                         FD_SET( _WakeUpPipeHandle[PipeRead], &readers );
1040                         if ( _WakeUpPipeHandle[PipeRead]>descmax )
1041                         {
1042                                 descmax = _WakeUpPipeHandle[PipeRead];
1043                         }
1044 #endif
1045
1046                         // Unlock _Connections, use connections_copy instead
1047                 }
1048
1049 #ifndef NL_OS_UNIX
1050                 // Avoid select if there is no connection (Windows only)
1051                 if ( skip || alldisconnected )
1052                 {
1053                         nlSleep( 1 ); // nice
1054                         continue;
1055                 }
1056 #endif
1057
1058 #ifdef NL_OS_WINDOWS
1059                 TIMEVAL tv;
1060                 tv.tv_sec = 0; // short time because the newly added connections can't be added to the select fd_set
1061                 tv.tv_usec = 10000;
1062
1063                 // Call select
1064                 int res = ::select( descmax+1, &readers, NULL, NULL, &tv );
1065
1066 #elif defined NL_OS_UNIX
1067
1068                 // Call select
1069                 int res = ::select( descmax+1, &readers, NULL, NULL, NULL );
1070
1071 #endif // NL_OS_WINDOWS
1072
1073
1074                 // POLL9
1075
1076                 // 3. Test the result
1077                 switch ( res )
1078                 {
1079 #ifdef NL_OS_WINDOWS
1080                         case  0 : continue; // time-out expired, no results
1081 #endif
1082                         case -1 :
1083                                 // we'll ignore message (Interrupted system call) caused by a CTRL-C
1084                                 /*if (CSock::getLastError() == 4)
1085                                 {
1086                                         LNETL1_DEBUG ("LNETL1: Select failed (in receive thread): %s (code %u) but IGNORED", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError());
1087                                         continue;
1088                                 }*/
1089                                 //nlerror( "LNETL1: Select failed (in receive thread): %s (code %u)", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError() );
1090                                 LNETL1_DEBUG( "LNETL1: Select failed (in receive thread): %s (code %u)", CSock::errorString( CSock::getLastError() ).c_str(), CSock::getLastError() );
1091                                 goto end;
1092                 }
1093
1094                 // 4. Get results
1095
1096                 vector<TSockId>::iterator ic;
1097                 for ( ic=connections_copy.begin(); ic!=connections_copy.end(); ++ic )
1098                 {
1099                         if ( FD_ISSET( (*ic)->Sock->descriptor(), &readers ) != 0 )
1100                         {
1101                                 CServerBufSock *serverbufsock = static_cast<CServerBufSock*>(static_cast<CBufSock*>(*ic));
1102                                 try
1103                                 {
1104                                         // 4. Receive data
1105                                         if ( serverbufsock->receivePart( sizeof(TSockId) + 1 ) ) // +1 for the event type
1106                                         {
1107                                                 serverbufsock->fillSockIdAndEventType( *ic );
1108
1109                                                 // Push message into receive queue
1110                                                 //uint32 bufsize;
1111                                                 //sint32 mbsize;
1112
1113                                                 _Server->pushMessageIntoReceiveQueue( serverbufsock->receivedBuffer() );
1114
1115                                                 //recvfifo.value().display();
1116                                                 //bufsize = serverbufsock->receivedBuffer().size();
1117                                                 //mbsize = recvfifo.value().size() / 1048576;
1118                                                 //nldebug( "RCV: Released." );
1119                                                 /*if ( mbsize > 1 )
1120                                                 {
1121                                                         nlwarning( "The receive queue size exceeds %d MB", mbsize );
1122                                                 }*/
1123                                                 /*
1124                                                 // Statistics
1125                                                 {
1126                                                         CSynchronized<uint32>::CAccessor syncbpi ( &_Server->syncBytesPushedIn() );
1127                                                         syncbpi.value() += bufsize;
1128                                                 }
1129                                                 */
1130                                         }
1131                                 }
1132 //                              catch (const ESocketConnectionClosed&)
1133 //                              {
1134 //                                      LNETL1_DEBUG( "LNETL1: Connection %s closed", serverbufsock->asString().c_str() );
1135 //                                      // The socket went to _Connected=false when throwing the exception
1136 //                              }
1137                                 catch (const ESocket&)
1138                                 {
1139                                         LNETL1_DEBUG( "LNETL1: Connection %s broken", serverbufsock->asString().c_str() );
1140                                         (*ic)->Sock->disconnect();
1141                                 }
1142 /*
1143 #ifdef NL_OS_UNIX
1144                                 skip = true; // don't check _WakeUpPipeHandle (yes, check it to read any written byte)
1145 #endif
1146
1147 */
1148                         }
1149                 }
1150
1151 #ifdef NL_OS_UNIX
1152                 // Test wake-up pipe
1153                 if ( (!skip) && (FD_ISSET( _WakeUpPipeHandle[PipeRead], &readers )) )
1154                 {
1155                         uint8 b;
1156                         if ( read( _WakeUpPipeHandle[PipeRead], &b, 1 ) == -1 ) // we were woken-up by the wake-up pipe
1157                         {
1158                                 LNETL1_DEBUG( "LNETL1: In CServerReceiveTask::run(): read() failed" );
1159                         }
1160                         LNETL1_DEBUG( "LNETL1: Receive thread select woken-up" );
1161                 }
1162 #endif
1163
1164                 NbLoop++;
1165         }
1166 end:
1167         nlnettrace( "Exiting CServerReceiveTask::run" );
1168         NbServerReceiveTask--;
1169         NbNetworkTask--;
1170 }
1171
1172
1173 /*
1174  * Delete all connections referenced in the remove list (double-mutexed)
1175  */
1176
1177 void CServerReceiveTask::clearClosedConnections()
1178 {
1179         CConnections::iterator ic;
1180         {
1181                 NLMISC::CSynchronized<CConnections>::CAccessor removesetsync( &_RemoveSet );
1182                 {
1183                         if ( ! removesetsync.value().empty() )
1184                         {
1185                                 // Delete closed connections
1186                                 NLMISC::CSynchronized<CConnections>::CAccessor connectionssync( &_Connections );
1187                                 for ( ic=removesetsync.value().begin(); ic!=removesetsync.value().end(); ++ic )
1188                                 {
1189                                         LNETL1_DEBUG( "LNETL1: Removing a connection" );
1190
1191                                         TSockId sid = (*ic);
1192
1193                                         // Remove from the connection list
1194                                         connectionssync.value().erase( *ic );
1195
1196                                         // Delete the socket object
1197                                         delete sid;
1198                                 }
1199                                 // Clear remove list
1200                                 removesetsync.value().clear();
1201                         }
1202                 }
1203         }
1204 }
1205
1206 NLMISC_CATEGORISED_VARIABLE(nel, uint32, NbServerListenTask, "Number of server listen thread");
1207 NLMISC_CATEGORISED_VARIABLE(nel, uint32, NbServerReceiveTask, "Number of server receive thread");
1208
1209 } // NLNET