ryzom/server/src/frontend_service/fe_send_sub.cpp

   1 // Ryzom - MMORPG Framework <http://dev.ryzom.com/projects/ryzom/>
   2 // Copyright (C) 2010  Winch Gate Property Limited
   3 //
   4 // This program is free software: you can redistribute it and/or modify
   5 // it under the terms of the GNU Affero General Public License as
   6 // published by the Free Software Foundation, either version 3 of the
   7 // License, or (at your option) any later version.
   8 //
   9 // This program is distributed in the hope that it will be useful,
  10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 // GNU Affero General Public License for more details.
  13 //
  14 // You should have received a copy of the GNU Affero General Public License
  15 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17
  18
  19 #include "stdpch.h"
  20
  21 #include "game_share/generic_xml_msg_mngr.h"
  22
  23 #include <nel/misc/stop_watch.h>
  24 #include <nel/misc/path.h>
  25 #include "fe_send_sub.h"
  26 #include "game_share/simlag.h"
  27 #include "game_share/action_factory.h"
  28 #include "game_share/action_position.h"
  29 #include "game_share/action_sync.h"
  30 #include "game_share/action_disconnection.h"
  31 #include "game_share/action_association.h"
  32 #include "game_share/action_sint64.h"
  33 #include "game_share/tick_event_handler.h"
  34 #include "history.h"
  35 #include "frontend_service.h"
  36 #include "prio_sub.h"
  37 #include "fe_stat.h"
  38
  39 #include "game_share/system_message.h"
  40
  41 using namespace std;
  42 using namespace NLNET;
  43 using namespace NLMISC;
  44 using namespace CLFECOMMON;
  45
  46 uint SendCounterRatio = 0;
  47 uint SendCounterDelay = 1000;
  48 extern CGenericXmlMsgHeaderManager GenericXmlMsgHeaderMngr;
  49
  50
  51
  52 /*
  53  * Init
  54  */
  55 void CFeSendSub::init( NLNET::CUdpSock *datasock, THostMap *clientmap, CHistory *history, CPrioSub *priosub )
  56 {
  57         nlassert( datasock && history );
  58
  59         _DataSock = datasock;
  60
  61         _ClientMap = clientmap;
  62         _History = history;
  63         _PrioSub = priosub;
  64
  65         _SendBuffers1.resize( MaxNbClients+1 );
  66         _SendBuffers2.resize( MaxNbClients+1 );
  67         _CurrentFillingBuffers = &_SendBuffers1;
  68         _CurrentFlushingBuffers = &_SendBuffers2;
  69
  70         _MsgXmlMD5 = NLMISC::getMD5("msg.xml");
  71         _DatabaseXmlMD5 = NLMISC::getMD5("database.xml");
  72
  73         /*_TotalBandwidth = totalbandwith; // initialized by setTotalBandwidth() and setClientBandwitdth()
  74         _ClientBandwidth = clientbandwith; */
  75 }
  76
  77
  78 /*
  79  * Set client bandwidth per cycle in bytes
  80  */
  81 void    CFeSendSub::setClientBandwidth( uint32 bytes )
  82 {
  83         _ClientBitBandwidth = bytes * 8;
  84
  85         // Change the bandwidth of all logged clients
  86         THostMap::iterator ihm;
  87         for ( ihm=_ClientMap->begin(); ihm!=_ClientMap->end(); ++ihm )
  88         {
  89                 GETCLIENTA(ihm)->setClientBandwidth( _ClientBitBandwidth );
  90         }
  91 }
  92
  93
  94
  95 /*
  96  * Setup headers for outgoing messages of current cycle
  97  */
  98 void    CFeSendSub::prepareHeadersAndFillImpulses()
  99 {
 100         //CFrontEndService::instance()->HeadWatch.start();
 101
 102         TTime   ctime = CTime::getLocalTime();
 103
 104         THostMap::iterator iclient;
 105         for ( iclient=_ClientMap->begin(); iclient!=_ClientMap->end(); ++iclient )
 106         {
 107                 CClientHost     *client = GETCLIENTA(iclient);
 108
 109                 if ( SendCounterRatio > 0 && ctime - client->LastCounterTime > SendCounterDelay)
 110                 {
 111                         // Send several packets when checking for packet lost
 112                         client->LastCounterTime = ctime;
 113
 114                         uint    i;
 115                         for (i=0; i<SendCounterRatio; ++i)
 116                         {
 117                                 CBitMemStream   bms;
 118                                 GenericXmlMsgHeaderMngr.pushNameToStream("DEBUG:COUNTER", bms);
 119
 120                                 bms.serial(client->LastSentCounter);
 121                                 client->LastSentCounter++;
 122
 123                                 //nldebug( "DebugCounter %s", CInstanceCounterManager().getInstance().displayCounter( "CAction" ).c_str() );
 124                                 CActionGeneric *ag = (CActionGeneric *)CActionFactory::getInstance ()->create (INVALID_SLOT, ACTION_GENERIC_CODE);
 125                                 ag->set (bms);
 126
 127                                 CFrontEndService::instance()->addImpulseToClient (client->clientId(), ag, 1);
 128                         }
 129                 }
 130
 131                 if ( client->whenToSend() )
 132                 {
 133                         TOutBox& outbox = (*_CurrentFillingBuffers)[client->clientId()].OutBox;
 134
 135                         // Clear outbox
 136                         outbox.resetBufPos();
 137
 138                         // Manage connection state (probe or not)
 139                         switch (client->ConnectionState)
 140                         {
 141                         case CClientHost::Connected:
 142                                 {
 143                                         client->NbActionsSentAtCycle = 0;
 144
 145                                         // 1. Fill headers
 146                                         client->setupOutBox( outbox );
 147
 148 #ifdef INCLUDE_FE_STATS_IN_PACKETS
 149                                         // send debug info in the message header
 150                                         uint32 value;
 151                                         value = CFrontEndService::instance()->UserLWatch.getPartialAverage();
 152                                         outbox.serialAndLog1 (value);
 153                                         value = CFrontEndService::instance()->CycleWatch.getPartialAverage();
 154                                         outbox.serialAndLog1 (value);
 155                                         value = CFrontEndService::instance()->ReceiveWatch.getPartialAverage();
 156                                         outbox.serialAndLog1 (value);
 157                                         value = CFrontEndService::instance()->SendWatch.getPartialAverage();
 158                                         outbox.serialAndLog1 (value);
 159                                         outbox.serialAndLog1 (client->PrioAmount);
 160                                         uint16 seenentities = 255 - client->NbFreeEntityItems;
 161                                         outbox.serialAndLog1 ( seenentities );
 162                                         /*float hpt = CFrontEndService::instance()->PrioSub.Prioritizer.hpThreshold();
 163                                         outbox.serialAndLog1( hpt );*/
 164
 165                                         // Important: Set STAT_HEADER_SIZE if you change the debug info sent
 166 #endif
 167                                         // 2. Fill impulse actions
 168                                         sint32 impulseFilledBits = (sint32)client->ImpulseEncoder.send( client->sendNumber(), outbox, _NbImpulseActions );
 169
 170                                         // Impulsion flow control, takes into account:
 171                                         // - the number of bits filled (possibly exceeding the max when sending forced actions (database))
 172                                         // - the number of remaining actions not forced
 173                                         client->setImpulsionThrottle( impulseFilledBits, (sint32)client->ImpulseEncoder.maxBitSize(2), client->ImpulseEncoder.queueSize() );
 174
 175                                 }
 176                                 break;
 177                         case CClientHost::Synchronize:
 178                         case CClientHost::ForceSynchronize:
 179                                 {
 180                                         // Sending of Sync is now guaranteed at this cycle (see below) => fall back to Synchronize
 181                                         if (client->ConnectionState == CClientHost::ForceSynchronize)
 182                                                 client->setSynchronizeState();
 183
 184                                         // send SYNC at defined frequency
 185                                         NLMISC::TGameCycle tick = CTickEventHandler::getGameCycle();
 186                                         client->setFirstSentPacket( client->sendNumber()+1, tick );
 187                                         client->setupSystemHeader( outbox, SYSTEM_SYNC_CODE);
 188
 189                                         //
 190                                         TGameCycle      sync = client->getSync();
 191                                         outbox.serialAndLog1(sync);
 192                                         TTime           stime = CTime::getLocalTime();
 193                                         outbox.serialAndLog1(stime);
 194                                         outbox.serialAndLog1(client->LastSentSync);
 195
 196                                         outbox.serialBuffer(_MsgXmlMD5.Data, sizeof(_MsgXmlMD5.Data));
 197                                         outbox.serialBuffer(_DatabaseXmlMD5.Data, sizeof(_DatabaseXmlMD5.Data));
 198
 199                                         nlinfo("FESEND: sent SYNC message to client %d - SYNC=%d, GameCycle=%d", client->clientId(), sync, CTickEventHandler::getGameCycle());
 200                                         break;
 201                                 }
 202                         case CClientHost::Probe:
 203                                 // send PROBE at defined frequency
 204                                 if (ctime - client->LastProbeTime > PROBESendLatency)
 205                                 {
 206                                         client->setupSystemHeader( outbox, SYSTEM_PROBE_CODE);
 207                                         client->LastSentProbe++;
 208                                         outbox.serialAndLog1(client->LastSentProbe);
 209                                         client->LastProbeTime = ctime;
 210                                         nldebug("FESEND: sent PROBE message to client %d", client->clientId());
 211                                         client->setIdleImpulsionThrottle();
 212                                 }
 213                                 break;
 214                         }
 215
 216                         //_OutputBits += outbox.getPosInBit();
 217                 }
 218         }
 219
 220         nlassert( ! _ClientMap->empty() );
 221
 222         //CFrontEndService::instance()->HeadWatch.stop();
 223 }
 224
 225
 226 /*
 227  * Fill prioritized actions into outgoing messages.
 228  * Query priority subsystem for properties,
 229  *    retrieve them using mirror,
 230  *    fill client outboxes,
 231  *    fill history.
 232  */
 233 void    CFeSendSub::fillPrioritizedActions()
 234 {
 235 #ifdef MEASURE_FRONTEND_TABLES
 236         PosSentCntFrame.setGameTick();
 237         PosSentCntFrame.reset( 0 );
 238 #endif
 239
 240         // We don't take _TotalBitBandwith into account currently
 241
 242         /* Iterate on the clients
 243          */
 244         THostMap::iterator icm;
 245         for ( icm=_ClientMap->begin(); icm!=_ClientMap->end(); ++icm )
 246         {
 247                 CClientHost *destclient = GETCLIENTA(icm);
 248
 249                 // Do not fill if client is blocked
 250                 if ( destclient->whenToSend() && (destclient->ConnectionState == CClientHost::Connected) )
 251                 {
 252                         TOutBox& outbox = (*_CurrentFillingBuffers)[destclient->clientId()].OutBox;
 253                         _PrioSub->Prioritizer.fillOutBox( *destclient, outbox );
 254                         _NbActions += destclient->NbActionsSentAtCycle;
 255                         destclient->updateThrottle( outbox );
 256                 }
 257
 258                 // Open/close the send buffers, depending on the state clienthost->whenToSend().
 259                 // Note: the send buffers must not have a reference on the clienthost objects, because
 260                 // these can be destroyed when those are still processed in the background, therefore
 261                 // we browse the clients to enable/disable the send buffers.
 262                 // We set the current filling buffer only, just before swapping them, so that
 263                 // the changes will be taken into account at next flushing.
 264
 265                 CSendBuffer& sendbuffer = (*_CurrentFillingBuffers)[destclient->clientId()];
 266                 sendbuffer.enableSendBuffer( destclient->whenToSend() );
 267                 //nldebug( "%u: client %hu %s", CTickEventHandler::getGameCycle(), destclient->clientId(), destclient->whenToSend()?"OPEN":"CLOSED" );
 268
 269                 // Switch the send state of the client
 270                 destclient->incSendCycle();
 271
 272         }
 273
 274         //nlassert( ! _ClientMap->empty() );
 275         //float fullbuffers = (float)_OutputBits / (float)_ClientBitBandwidth;
 276
 277         CFrontEndService::instance()->SentActionsLastCycle = _NbActions + _NbImpulseActions;
 278         //CFrontEndService::instance()->ScannedPropsLastCycle = _NbActions + nbnotsent - _NbImpulseActions;
 279
 280 //      nlinfo( "FESEND: Sent %u actions (%u left ; %.2f full buffers, %.2f%%) including %u impulse (%0.2f priorities, %u per client)", _NbActions, nbnotsent, fullbuffers, fullbuffers/(float)_ClientMap->size()*100.0f, _NbImpulseActions, _PrioAmount, _NbActions / _ClientMap->size() );
 281
 282         /*for ( iclient=_ClientMap->begin(); iclient!=_ClientMap->end(); ++iclient )
 283         {
 284                 nlinfo( "Client %u: size %u / %u", GETCLIENTA(iclient)->clientId(), GETCLIENTA(iclient)->OutBox.length(), _ClientBandwith );
 285         }*/
 286
 287         //CFrontEndService::instance()->FillWatch.stop();
 288
 289 #ifdef MEASURE_FRONTEND_TABLES
 290         PosSentCntFrame.commit( PosSentCntClt1 );
 291 #endif
 292
 293 }
 294
 295
 296 /*
 297  * Swap read buffers
 298  */
 299 void CFeSendSub::swapSendBuffers()
 300 {
 301         // Synchronization is done at module-level (in frontend_service.cpp)
 302
 303         // OBSOLETE: now the enabling of the send buffers depends on clienthost->whenToSend(), see below
 304         /*// Before swapping, but after the previous flushing, enable the buffers of the new clients
 305         // in the flushing buffers
 306         typename TBuffersToEnable::iterator ib;
 307         for ( ib=_BuffersToEnable.begin(); ib!=_BuffersToEnable.end(); ++ib )
 308         {
 309                 enableSendBuffer( *ib );
 310         }
 311         _BuffersToEnable.clear();*/
 312
 313         // Swap the buffers
 314         if ( _CurrentFillingBuffers == &_SendBuffers1 )
 315         {
 316                 _CurrentFillingBuffers = &_SendBuffers2;
 317                 _CurrentFlushingBuffers = &_SendBuffers1;
 318         }
 319         else
 320         {
 321                 _CurrentFillingBuffers = &_SendBuffers1;
 322                 _CurrentFlushingBuffers = &_SendBuffers2;
 323         }
 324 }
 325
 326
 327 /*
 328  * Send the current outbox
 329  */
 330 inline void     CFeSendSub::CSendBuffer::sendOutBox( NLNET::CUdpSock *datasock )
 331 {
 332         if ( OutBox.length() != 0 )
 333         {
 334                 sendUDP( datasock, OutBox.buffer(), OutBox.length(), &DestAddress );
 335                 //nlinfo( "Sent %u bytes to %s", OutBox.length(), DestAddress.asString().c_str() );
 336         }
 337
 338 #ifdef MEASURE_SENDING
 339                 static sint32 loopcount = 0;
 340                 ++loopcount;
 341                 static TTime lastdisplay = CTime::getLocalTime();
 342                 TTime tn = CTime::getLocalTime();
 343                 uint32 diff = (uint32)(tn - lastdisplay);
 344                 if ( diff > 2000 )
 345                 {
 346                         nlinfo("Sends by second: %.1f => LoopTime = %.2f ms LoopCount = %u Diff = %u ms Size=%u",(float)loopcount * 1000.0f / (float)diff, (float)diff / loopcount, loopcount, diff, OutBox.length());
 347                         loopcount = 0;
 348                         lastdisplay = tn;
 349                 }
 350 #endif
 351 }
 352
 353
 354 /*
 355  * Send outgoing messages
 356  * This can be executed by a background thread
 357  */
 358 void    CFeSendSub::flushMessages()
 359 {
 360
 361         //CFrontEndService::instance()->SndtWatch.start();
 362
 363 #ifdef MEASURE_SENDING
 364         TTicks before = CTime::getPerformanceTime();
 365 #endif
 366
 367         TSendBuffers::iterator isb;
 368         for ( isb=_CurrentFlushingBuffers->begin(); isb!=_CurrentFlushingBuffers->end(); ++isb )
 369         {
 370                 if ( (*isb).SBState )
 371                 {
 372                         try
 373                         {
 374                                 (*isb).sendOutBox( _DataSock );
 375                                 ++_SendCounter;
 376                                 //nldebug( "%u: SENDING NOW %u bytes to %s", CTickEventHandler::getGameCycle(), (*isb).OutBox.length(), (*isb).DestAddress.asString().c_str() );
 377                         }
 378                         catch (const ESocket&)
 379                         {
 380                                 nlwarning( "Could not send data to client %u", isb-_CurrentFlushingBuffers->begin() );
 381                         }
 382                 }
 383         }
 384
 385 #ifdef MEASURE_SENDING
 386         TTicks after = CTime::getPerformanceTime();
 387         static TTicks sum = 0;
 388         sum += (after-before);
 389         static uint counter = 0;
 390         ++counter;
 391         if ( counter == 20 )
 392         {
 393                 nlinfo( "Sending time: %.3f ms", CTime::ticksToSecond( sum/20 ) * 1000.0 );
 394                 sum = 0;
 395                 counter = 0;
 396         }
 397 #endif
 398
 399 }
 400
 401
 402
 403 /*
 404  * Update
 405  * Deprecated: replaced by modules (see initModules() in frontend_service.cpp)
 406  */
 407 /*void CFeSendSub::update()
 408 {
 409         if ( _ClientMap->empty() )
 410         {
 411                 return;
 412         }
 413
 414         prepareHeaders();
 415
 416         fillImpulses();
 417
 418         fillUIActions();
 419
 420         fillPrioritizedActions();
 421
 422         flushMessages();
 423
 424         _PrioSub->endCycle(); // disables the priority marked to
 425
 426         nlinfo( "FETIME: Time of sending: Headers=%u Fill=%u Send=%u EndCycle=%u",
 427                 CFrontEndService::instance()->HeadWatch.getDuration(),
 428                 CFrontEndService::instance()->FillWatch.getDuration(),
 429                 CFrontEndService::instance()->SndtWatch.getDuration(),
 430                 CFrontEndService::instance()->EndCWatch.getDuration() );
 431 }*/
 432
 433
 434 NLMISC_VARIABLE( uint, SendCounterRatio, "Set the check counter send ratio, e.g. the number of counter sent at a time (0 means don't send)");
 435 NLMISC_VARIABLE( uint, SendCounterDelay, "Set the check counter send delay, e.g. the time between 2 counter send rafale (in ms)");