ryzom/server/src/frontend_service/fe_receive_task.cpp

   1 // Ryzom - MMORPG Framework <http://dev.ryzom.com/projects/ryzom/>
   2 // Copyright (C) 2010  Winch Gate Property Limited
   3 //
   4 // This source file has been modified by the following contributors:
   5 // Copyright (C) 2014  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
  21
  22 #include "stdpch.h"
  23
  24 #include "fe_receive_task.h"
  25 #include "fe_types.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
  34 #include <unistd.h>
  35 #include <sys/socket.h>
  36 #include <arpa/inet.h>
  37 #include <netinet/in.h>
  38 #include <netdb.h>
  39
  40 #define WSAGetLastError() 0
  41
  42 #endif
  43
  44
  45 using namespace NLMISC;
  46 using namespace NLNET;
  47
  48
  49 volatile uint32 CFEReceiveTask::LastUDPPacketReceived = 0;
  50
  51 /*
  52  * TReceivedMessage
  53  */
  54
  55 /// Constructor
  56 TReceivedMessage::TReceivedMessage()
  57 {
  58         VAddrFrom.resize( sizeof(sockaddr_in) );
  59 }
  60
  61 /// Return a vector containing the address info
  62 void    TReceivedMessage::addressToVector()
  63 {
  64         memcpy( &*VAddrFrom.begin(), AddrFrom.sockAddr(), sizeof(sockaddr_in) );
  65 }
  66
  67 /// Set address with address info from specified vector
  68 void    TReceivedMessage::vectorToAddress()
  69 {
  70         AddrFrom.setSockAddr( (sockaddr_in*)&*VAddrFrom.begin() );
  71 }
  72
  73
  74 /*
  75  * Constructor (note: called from the main thread)
  76  */
  77 CFEReceiveTask::CFEReceiveTask( uint16 firstAcceptablePort, uint16 lastAcceptablePort, uint32 msgsize ) :
  78         _ReceivedMessage(),
  79         _WriteQueue( "WriteQueue" ), // value unspecified
  80         _DatagramLength( msgsize ),
  81         _ExitRequired( false ),
  82         _NbRejectedDatagrams( 0 )
  83 {
  84         // Socket
  85         DataSock = new CUdpSock( false );
  86         nlassert( DataSock );
  87
  88         // Test of multihomed host: bind the first address
  89 /*      vector<CInetAddress> addrlist;
  90         addrlist = CInetAddress::localAddresses();
  91         vector<CInetAddress>::iterator ivi;
  92         nlinfo( "Listing local interfaces:" );
  93         for ( ivi=addrlist.begin(); ivi!=addrlist.end(); ++ivi )
  94         {
  95                 nlinfo( "%s", (*ivi).asIPString().c_str() );
  96         }
  97         addrlist[0].setPort( port );
  98         DataSock->bind( addrlist[0] );
  99 */
 100
 101         // Bind on all network interfaces (TODO: find a simple way to bind only on the external interface; note: change releaase code for Linux in fe_receive_sub.cpp as well)
 102         uint16 actualPort;
 103         for ( actualPort=firstAcceptablePort; actualPort<=lastAcceptablePort; ++actualPort )
 104         {
 105                 try
 106                 {
 107                         DataSock->bind( actualPort );
 108                         break;
 109                 }
 110                 catch (const ESocket &e)
 111                 {
 112                         nlinfo( "Port %u not available: %s", actualPort, e.what() );
 113                 }
 114         }
 115         if ( actualPort > lastAcceptablePort )
 116                 nlerror( "Could not find an available port between %hu and %hu", firstAcceptablePort, lastAcceptablePort );
 117         nlinfo( "Binding all network interfaces on port %hu (%hu asked)", actualPort, firstAcceptablePort );
 118 }
 119
 120
 121
 122 /*
 123  * Destructor
 124  */
 125 CFEReceiveTask::~CFEReceiveTask()
 126 {
 127         nlassert( DataSock != NULL );
 128         delete DataSock;
 129         DataSock = NULL;
 130 }
 131
 132
 133 /*
 134  * Run
 135  */
 136 void CFEReceiveTask::run()
 137 {
 138         uint maxrecvlength = _DatagramLength;
 139         while ( ! _ExitRequired )
 140         {
 141 #ifndef SIMUL_CLIENTS
 142
 143 #ifdef MEASURE_RECEIVE_TASK
 144                 static sint32 loopcount = 0;
 145                 ++loopcount;
 146                 static TTime lastdisplay = CTime::getLocalTime();
 147                 TTime tn = CTime::getLocalTime();
 148                 uint32 diff = (uint32)(tn - lastdisplay);
 149                 if ( diff > 2000 )
 150                 {
 151                         nlinfo("Reads by second: %.1f => LoopTime = %.2f ms LoopCount = %u Diff = %u ms",(float)loopcount * 1000.0f / (float)diff, (float)diff / loopcount, loopcount, diff);
 152                         loopcount = 0;
 153                         lastdisplay = tn;
 154                 }
 155
 156 #endif
 157                 try
 158                 {
 159                         // Receive into _ReceivedMessage
 160                         _DatagramLength = maxrecvlength;
 161                         _ReceivedMessage.resizeData( _DatagramLength );
 162                         _ReceivedMessage.setTypeEvent( TReceivedMessage::User );
 163                         DataSock->receivedFrom( _ReceivedMessage.userDataW(), _DatagramLength, _ReceivedMessage.AddrFrom );
 164                 }
 165                 catch (const ESocket&)
 166                 {
 167                         // Remove the client corresponding to the address
 168                         _ReceivedMessage.setTypeEvent( TReceivedMessage::RemoveClient );
 169                         _DatagramLength = 0;
 170                 }
 171
 172                 // update the last datagram receive date
 173                 LastUDPPacketReceived = CTime::getSecondsSince1970();
 174
 175                 // Check the size. Consider a big size as a hacked message
 176 //              if ( _DatagramLength < 512 )
 177                 {
 178                         // Push into the write queue
 179                         _ReceivedMessage.addressToVector();
 180                         _ReceivedMessage.resizeData( _DatagramLength ); // _DatagramLength was modified by receivedFrom()
 181                         {
 182                                 CSynchronized<CBufFIFO*>::CAccessor wq( &_WriteQueue );
 183                                 wq.value()->push( _ReceivedMessage.data() );
 184                                 wq.value()->push( _ReceivedMessage.VAddrFrom );
 185                         }
 186                 }
 187 //              else
 188 //              {
 189 //                      // Reject message
 190 //                      ++_NbRejectedDatagrams;
 191 //              }
 192
 193 #else
 194                 nlSleep( 1000 );
 195 #endif
 196         }
 197
 198         nlinfo( "Exiting from front-end receive task" );
 199 }
 200
 201
 202 /*
 203  * Set new write queue
 204  */
 205 void CFEReceiveTask::setWriteQueue( CBufFIFO *writequeue )
 206 {
 207         CSynchronized<CBufFIFO*>::CAccessor wq( &_WriteQueue );
 208         wq.value() = writequeue;
 209 }
 210