Planet
navi homePPSaboutscreenshotsdownloaddevelopmentforum

source: orxonox.OLD/branches/proxy/src/lib/network/network_stream.cc @ 9541

Last change on this file since 9541 was 9541, checked in by patrick, 18 years ago

discovered a bug in the client id assignement algorithm. corrected. proxy control center should work now. testing

File size: 34.7 KB
Line 
1/*
2   orxonox - the future of 3D-vertical-scrollers
3
4   Copyright (C) 2004 orx
5
6   This program is free software; you can redistribute it and/or modify
7   it under the terms of the GNU General Public License as published by
8   the Free Software Foundation; either version 2, or (at your option)
9   any later version.
10
11### File Specific:
12   main-programmer: Christoph Renner rennerc@ee.ethz.ch
13   co-programmer:   Patrick Boenzli  patrick@orxonox.ethz.ch
14
15     June 2006: finishing work on the network stream for pps presentation (rennerc@ee.ethz.ch)
16     July 2006: some code rearangement and integration of the proxy server mechanism (patrick@orxonox.ethz.ch)
17*/
18
19
20#define DEBUG_MODULE_NETWORK
21
22#include "proxy/proxy_control.h"
23
24#include "base_object.h"
25#include "network_protocol.h"
26#include "udp_socket.h"
27#include "udp_server_socket.h"
28#include "monitor/connection_monitor.h"
29#include "monitor/network_monitor.h"
30#include "synchronizeable.h"
31#include "ip.h"
32#include "network_game_manager.h"
33#include "shared_network_data.h"
34#include "message_manager.h"
35#include "preferences.h"
36#include "zip.h"
37
38#include "src/lib/util/loading/resource_manager.h"
39
40#include "network_log.h"
41
42#include "player_stats.h"
43
44#include "lib/util/loading/factory.h"
45
46#include "debug.h"
47#include "class_list.h"
48#include <algorithm>
49
50
51#include "network_stream.h"
52
53
54#include "converter.h"
55
56
57#define PACKAGE_SIZE  256
58
59
60/**
61 * empty constructor
62 */
63NetworkStream::NetworkStream()
64    : DataStream()
65{
66  this->init();
67  /* initialize the references */
68  this->pInfo->nodeType = NET_UNASSIGNED;
69}
70
71
72NetworkStream::NetworkStream( int nodeType)
73{
74  this->init();
75
76  this->pInfo->nodeType = nodeType;
77
78  switch( nodeType)
79  {
80    case NET_MASTER_SERVER:
81      // init the shared network data
82      SharedNetworkData::getInstance()->setHostID(NET_ID_MASTER_SERVER);
83      break;
84    case NET_PROXY_SERVER_ACTIVE:
85      // init the shared network data
86      SharedNetworkData::getInstance()->setHostID(NET_ID_PROXY_SERVER_01);
87      break;
88    case NET_PROXY_SERVER_PASSIVE:
89      // init the shared network data
90      SharedNetworkData::getInstance()->setHostID(NET_ID_PROXY_SERVER_01);
91      break;
92    case NET_CLIENT:
93      SharedNetworkData::getInstance()->setHostID(NET_ID_UNASSIGNED);
94      break;
95  }
96
97  SharedNetworkData::getInstance()->setDefaultSyncStream(this);
98
99  // get the local ip address
100  IPaddress ip;
101  SDLNet_ResolveHost( &ip, NULL, 0);
102  this->pInfo->ip = ip;
103}
104
105
106
107/**
108 * generic init functions
109 */
110void NetworkStream::init()
111{
112  /* set the class id for the base object */
113  this->setClassID(CL_NETWORK_STREAM, "NetworkStream");
114  this->clientSocket = NULL;
115  this->proxySocket = NULL;
116  this->networkGameManager = NULL;
117  this->networkMonitor = NULL;
118
119  this->pInfo = new PeerInfo();
120  this->pInfo->userId = 0;
121  this->pInfo->lastAckedState = 0;
122  this->pInfo->lastRecvedState = 0;
123
124  this->bRedirect = false;
125
126  this->currentState = 0;
127
128  remainingBytesToWriteToDict = Preferences::getInstance()->getInt( "compression", "writedict", 0 );
129
130  assert( Zip::getInstance()->loadDictionary( "testdict" ) >= 0 );
131  this->dictClient = Zip::getInstance()->loadDictionary( "dict2pl_client" );
132  assert( this->dictClient >= 0 );
133  this->dictServer = Zip::getInstance()->loadDictionary( "dict2p_server" );
134  assert( this->dictServer >= 0 );
135}
136
137
138/**
139 * deconstructor
140 */
141NetworkStream::~NetworkStream()
142{
143  if ( this->clientSocket )
144  {
145    clientSocket->close();
146    delete clientSocket;
147    clientSocket = NULL;
148  }
149  if ( this->proxySocket)
150  {
151    proxySocket->close();
152    delete proxySocket;
153    proxySocket = NULL;
154  }
155  for ( PeerList::iterator i = peers.begin(); i!=peers.end(); i++)
156  {
157    if ( i->second.socket )
158    {
159      i->second.socket->disconnectServer();
160      delete i->second.socket;
161      i->second.socket = NULL;
162    }
163
164    if ( i->second.handshake )
165    {
166      delete i->second.handshake;
167      i->second.handshake = NULL;
168    }
169
170    if ( i->second.connectionMonitor )
171    {
172      delete i->second.connectionMonitor;
173      i->second.connectionMonitor = NULL;
174    }
175  }
176  for ( SynchronizeableList::const_iterator it = getSyncBegin(); it != getSyncEnd(); it ++ )
177    (*it)->setNetworkStream( NULL );
178
179  if( this->pInfo)
180    delete this->pInfo;
181
182  if( this->networkMonitor)
183    delete this->networkMonitor;
184}
185
186
187/**
188 * establish a connection to a remote master server
189 * @param host: host name
190 * @param port: the port number
191 */
192void NetworkStream::connectToMasterServer(std::string host, int port)
193{
194  int node = NET_ID_MASTER_SERVER;
195  // this create the new node in the peers map
196  this->peers[node].socket = new UdpSocket( host, port );
197  this->peers[node].userId = NET_ID_MASTER_SERVER;
198
199  this->peers[node].nodeType = NET_MASTER_SERVER;
200  this->peers[node].connectionMonitor = new ConnectionMonitor( NET_ID_MASTER_SERVER );
201  this->peers[node].ip = this->peers[node].socket->getRemoteAddress();
202}
203
204
205/**
206 * establish a connection to a remote proxy server
207 * @param host: host name
208 * @param port: the port number
209 */
210void NetworkStream::connectToProxyServer(int proxyId, std::string host, int port)
211{
212  PRINTF(0)("connect to proxy %s, this is proxyId %i\n", host.c_str(), proxyId);
213
214  // this creates the new proxyId in the peers map
215  this->peers[proxyId].socket = new UdpSocket( host, port );
216  this->peers[proxyId].userId = proxyId;
217
218  this->peers[proxyId].nodeType = NET_PROXY_SERVER_ACTIVE;
219  this->peers[proxyId].connectionMonitor = new ConnectionMonitor( proxyId );
220  this->peers[proxyId].ip = this->peers[proxyId].socket->getRemoteAddress();
221}
222
223
224/**
225 * create a server
226 * @param port: interface port for all clients
227 */
228void NetworkStream::createServer(int clientPort, int proxyPort)
229{
230  PRINTF(0)(" Creating new Server: listening for clients on port %i and for proxies on port %i", clientPort, proxyPort);
231  this->clientSocket= new UdpServerSocket(clientPort);
232  this->proxySocket = new UdpServerSocket(proxyPort);
233}
234
235
236/**
237 * creates a new instance of the network game manager
238 */
239void NetworkStream::createNetworkGameManager()
240{
241  this->networkGameManager = NetworkGameManager::getInstance();
242
243  this->networkGameManager->setUniqueID( SharedNetworkData::getInstance()->getNewUniqueID() );
244  MessageManager::getInstance()->setUniqueID( SharedNetworkData::getInstance()->getNewUniqueID() );
245}
246
247
248/**
249 * starts the network handshake
250 * handsakes are always initialized from the client side first. this starts the handshake and therefore is only
251 * executed as client
252 * @param userId: start handshake for this user id (optional, default == 0)
253 */
254void NetworkStream::startHandshake(int userId)
255{
256  Handshake* hs = new Handshake(this->pInfo->nodeType);
257  // fake the unique id
258  hs->setUniqueID( NET_UID_HANDSHAKE );
259  assert( peers[userId].handshake == NULL );
260  peers[userId].handshake = hs;
261
262  // set the preferred nick name
263  hs->setPreferedNickName( Preferences::getInstance()->getString( "multiplayer", "nickname", "Player" ) );
264
265  PRINTF(0)("NetworkStream: Handshake created: %s\n", hs->getCName());
266}
267
268
269/**
270 * this functions connects a synchronizeable to the networkstream, therefore synchronizeing
271 * it all over the network and creating it on the other platforms (if and only if it is a
272 * server
273 * @param sync: the synchronizeable to add
274 */
275void NetworkStream::connectSynchronizeable(Synchronizeable& sync)
276{
277  this->synchronizeables.push_back(&sync);
278  sync.setNetworkStream( this );
279}
280
281
282/**
283 * removes the synchronizeable from the list of synchronized entities
284 * @param sync: the syncronizeable to remove
285 */
286void NetworkStream::disconnectSynchronizeable(Synchronizeable& sync)
287{
288  // removing the Synchronizeable from the List.
289  std::list<Synchronizeable*>::iterator disconnectSynchro = std::find(this->synchronizeables.begin(), this->synchronizeables.end(), &sync);
290  if (disconnectSynchro != this->synchronizeables.end())
291    this->synchronizeables.erase(disconnectSynchro);
292
293  oldSynchronizeables[sync.getUniqueID()] = SDL_GetTicks();
294}
295
296
297/**
298 * this is called to process data from the network socket to the synchronizeable and vice versa
299 */
300void NetworkStream::processData()
301{
302  // create the network monitor after all the init work and before there is any connection handlings
303  if( this->networkMonitor == NULL)
304    this->networkMonitor = new NetworkMonitor(this);
305
306
307  int tick = SDL_GetTicks();
308
309  this->currentState++;
310  // there was a wrap around
311  if( this->currentState < 0)
312  {
313    PRINTF(1)("A wrap around in the state variable as occured. The server was running so long? Pls restart server or write a mail to the supporters!\n");
314  }
315
316  if ( SharedNetworkData::getInstance()->isMasterServer())
317  {
318    // execute everytthing the master server shoudl do
319    if ( this->clientSocket )
320      this->clientSocket->update();
321    if( this->proxySocket)
322      this->proxySocket->update();
323
324    this->updateConnectionList();
325  }
326  else if( SharedNetworkData::getInstance()->isProxyServerActive())
327  {
328    //execute everything the proxy server should do
329    if ( this->clientSocket )
330      this->clientSocket->update();
331    if( this->proxySocket)
332      this->proxySocket->update();
333
334    this->updateConnectionList();
335  }
336
337#warning make this more modular: every proxy/master server connection should be watched for termination
338  if( !SharedNetworkData::getInstance()->isMasterServer())
339  {
340    // check if the connection is ok else terminate and remove
341    if ( !peers.empty() && peers[NET_ID_MASTER_SERVER].socket &&
342          ( !peers[NET_ID_MASTER_SERVER].socket->isOk() ||
343          peers[NET_ID_MASTER_SERVER].connectionMonitor->hasTimedOut() ) )
344    {
345      this->handleDisconnect( NET_ID_MASTER_SERVER);
346      PRINTF(1)("lost connection to server\n");
347    }
348    // check if there is a redirection command
349    if( this->bRedirect)
350    {
351      this->handleReconnect( NET_ID_MASTER_SERVER);
352    }
353  }
354
355  this->cleanUpOldSyncList();
356  this->handleHandshakes();
357
358  // update the network monitor
359  this->networkMonitor->process();
360
361  // order of up/downstream is important!!!!
362  // don't change it
363  this->handleDownstream( tick );
364  this->handleUpstream( tick );
365}
366
367
368/**
369 * @brief handles incoming connections
370 *
371 * if we are a NET_MASTER_SERVER or NET_PROXY_SERVER_ACTIVE update the connection list to accept new connections (clients)
372 * start and initialize the handsake for the new clients
373 */
374void NetworkStream::updateConnectionList( )
375{
376  //check for new connections
377
378  NetworkSocket* tempNetworkSocket = NULL;
379  int userId;
380
381  if( this->clientSocket != NULL)
382  {
383    tempNetworkSocket = this->clientSocket->getNewSocket();
384
385    // we got new NET_CLIENT connecting
386    if ( tempNetworkSocket )
387    {
388      // determine the network node id
389      if ( freeSocketSlots.size() > 0 )
390      {
391        // this should never be called
392        assert(false);
393        userId = freeSocketSlots.back();
394        freeSocketSlots.pop_back();
395      }
396      else
397      {
398        //userId = 1;
399        // each server (proxy and master) have an address space for new network nodes of 1000 nodes
400        userId = SharedNetworkData::getInstance()->getHostID() * 1000 + 1;
401
402        for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
403          if ( it->first >= userId )
404            userId = it->first + 1;
405
406        assert( userId < (SharedNetworkData::getInstance()->getHostID() + 1) * 1000);
407      }
408      // this creates a new entry in the peers list
409      peers[userId].socket = tempNetworkSocket;
410      peers[userId].nodeType = NET_CLIENT;
411
412      // handle the newly connected client
413      this->handleConnect(userId);
414
415      PRINTF(0)("New Client: %d\n", userId);
416    }
417  }
418
419
420  if( this->proxySocket != NULL)
421  {
422    tempNetworkSocket = this->proxySocket->getNewSocket();
423
424    // we got new NET_PROXY_SERVER_ACTIVE connecting
425    if ( tempNetworkSocket )
426    {
427      // determine the network node id
428      if ( freeSocketSlots.size() > 0 )
429      {
430        userId = freeSocketSlots.back();
431        freeSocketSlots.pop_back();
432      }
433      else
434      {
435        userId = 1;
436
437        for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
438          if ( it->first >= userId )
439            userId = it->first + 1;
440      }
441
442      // this creates a new entry in the peers list
443      peers[userId].socket = tempNetworkSocket;
444      peers[userId].nodeType = NET_PROXY_SERVER_ACTIVE;
445
446      // handle the newly connected proxy server
447      this->handleConnect(userId);
448
449      PRINTF(0)("New proxy connected: %d\n", userId);
450    }
451  }
452
453
454  //check if connections are ok else remove them
455  for ( PeerList::iterator it = peers.begin(); it != peers.end(); )
456  {
457    if (
458          it->second.socket &&
459          (
460            !it->second.socket->isOk()  ||
461            it->second.connectionMonitor->hasTimedOut()
462          )
463       )
464    {
465      std::string reason = "disconnected";
466      if ( it->second.connectionMonitor->hasTimedOut() )
467        reason = "timeout";
468      PRINTF(0)("Client is gone: %d (%s)\n", it->second.userId, reason.c_str());
469
470      this->handleDisconnect( it->second.userId);
471
472      it++;
473      continue;
474    }
475
476    it++;
477  }
478
479
480}
481
482
483/**
484 * this handles new connections
485 * @param userId: the id of the new user node
486 */
487void NetworkStream::handleConnect( int userId)
488{
489  // create new handshake and init its variables
490  peers[userId].handshake = new Handshake(this->pInfo->nodeType, userId, this->networkGameManager->getUniqueID(), MessageManager::getInstance()->getUniqueID());
491  peers[userId].handshake->setUniqueID(userId);
492
493  peers[userId].connectionMonitor = new ConnectionMonitor( userId );
494  peers[userId].userId = userId;
495
496  PRINTF(0)("num sync: %d\n", synchronizeables.size());
497
498  // get the proxy server informations and write them to the handshake, if any (proxy)
499  assert( this->networkMonitor != NULL);
500  PeerInfo* pi = this->networkMonitor->getFirstChoiceProxy();
501  if( pi != NULL)
502  {
503    peers[userId].handshake->setProxy1Address( pi->ip);
504  }
505  pi = this->networkMonitor->getSecondChoiceProxy();
506  if( pi != NULL)
507    peers[userId].handshake->setProxy2Address( pi->ip);
508
509  // check if the connecting client should reconnect to a proxy server
510  if( SharedNetworkData::getInstance()->isMasterServer())
511    peers[userId].handshake->setRedirect(/*this->networkMonitor->isReconnectNextClient()*/false);
512
513  // the connecting node of course is a client
514  peers[userId].ip = peers[userId].socket->getRemoteAddress();
515}
516
517
518
519/**
520 * some debug output
521 */
522void NetworkStream::debug()
523{
524  if( SharedNetworkData::getInstance()->isMasterServer()) {
525    PRINT(0)(" Host ist Master Server with ID: %i\n", this->pInfo->userId);
526  }
527  else if( SharedNetworkData::getInstance()->isProxyServerActive()) {
528    PRINT(0)(" Host ist Proxy Server with ID: %i\n", this->pInfo->userId);
529  }
530  else {
531    PRINT(0)(" Host ist Client with ID: %i\n", this->pInfo->userId);
532  }
533
534  PRINT(0)(" Got %i connected Synchronizeables, showing active Syncs:\n", this->synchronizeables.size());
535  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
536  {
537    if( (*it)->beSynchronized() == true)
538      PRINT(0)("  Synchronizeable of class: %s::%s, with unique ID: %i, Synchronize: %i\n", (*it)->getClassCName(), (*it)->getCName(),
539               (*it)->getUniqueID(), (*it)->beSynchronized());
540  }
541  PRINT(0)(" Maximal Connections: %i\n", SharedNetworkData::getInstance()->getMaxPlayer() );
542
543}
544
545
546/**
547 * @returns the number of synchronizeables registered to this stream
548 */
549int NetworkStream::getSyncCount()
550{
551  int n = 0;
552  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
553    if( (*it)->beSynchronized() == true)
554      ++n;
555
556  //return synchronizeables.size();
557  return n;
558}
559
560
561/**
562 * check if handshakes completed. if so create the network game manager else remove it again
563 */
564void NetworkStream::handleHandshakes( )
565{
566  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
567  {
568    if ( it->second.handshake )
569    {
570      // handshake finished
571      if ( it->second.handshake->completed() )
572      {
573        //handshake is correct
574        if ( it->second.handshake->ok() )
575        {
576          // write the first informations into the node so they can be read from there for case differentiation
577          it->second.nodeType = it->second.handshake->getRemoteNodeType();
578
579          // the counter part didn't mark it free for deletion yet
580          if ( !it->second.handshake->allowDel() )
581          {
582            // make sure this is a connection:
583            // - client       <==> master server
584            // - proxy server <==> master server
585            if(  SharedNetworkData::getInstance()->isClient() || SharedNetworkData::getInstance()->isProxyServerActive() && it->second.isMasterServer())
586            {
587              PRINTF(0)("Handshake: i am in client role\n");
588
589              SharedNetworkData::getInstance()->setHostID( it->second.handshake->getHostId() );
590              this->pInfo->userId = SharedNetworkData::getInstance()->getHostID();
591
592#warning the ip address is not set here because it results in a segfault when connecting to a proxy server => trace this later
593//               it->second.ip = it->second.socket->getRemoteAddress();
594
595              // it->second.nodeType = it->second.handshake->getRemoteNodeType();
596              // it->second.ip = it->second.socket->getRemoteAddress();
597              // add the new server to the nodes list (it can be a NET_MASTER_SERVER or NET_PROXY_SERVER)
598              this->networkMonitor->addNode(&it->second);
599              // get proxy 1 address and add it
600              this->networkMonitor->addNode(it->second.handshake->getProxy1Address(), NET_PROXY_SERVER_ACTIVE);
601              // get proxy 2 address and add it
602              this->networkMonitor->addNode(it->second.handshake->getProxy2Address(), NET_PROXY_SERVER_ACTIVE);
603
604              // now check if the server accepted the connection
605              if( SharedNetworkData::getInstance()->isClient() && it->second.handshake->redirect() )
606              {
607                this->bRedirect = true;
608              }
609
610              // create the new network game manager and init it
611              this->networkGameManager = NetworkGameManager::getInstance();
612              this->networkGameManager->setUniqueID( it->second.handshake->getNetworkGameManagerId() );
613              // init the new message manager
614              MessageManager::getInstance()->setUniqueID( it->second.handshake->getMessageManagerId() );
615            }
616
617            PRINT(0)("handshake finished id=%d\n", it->second.handshake->getNetworkGameManagerId());
618            it->second.handshake->del();
619
620          }
621          else
622          {
623            // handshake finished registring new player
624            if ( it->second.handshake->canDel() )
625            {
626
627              if (  SharedNetworkData::getInstance()->isMasterServer() )
628              {
629                it->second.ip = it->second.socket->getRemoteAddress();
630
631                this->networkMonitor->addNode(&it->second);
632
633                this->handleNewClient( it->second.userId );
634
635                if ( PlayerStats::getStats( it->second.userId ) && it->second.handshake->getPreferedNickName() != "" )
636                {
637                  PlayerStats::getStats( it->second.userId )->setNickName( it->second.handshake->getPreferedNickName() );
638                }
639              }
640              else if ( SharedNetworkData::getInstance()->isProxyServerActive() && it->second.isClient() )
641              {
642                PRINTF(0)("Handshake: i am in server role\n");
643
644                it->second.ip = it->second.socket->getRemoteAddress();
645
646                this->networkMonitor->addNode(&it->second);
647
648                // work with the ProxyControl to init the new client
649                ProxyControl::getInstance()->signalNewClient(it->second.userId);
650
651
652//                 this->handleNewClient( it->second.userId );
653
654//                 if ( PlayerStats::getStats( it->second.userId ) && it->second.handshake->getPreferedNickName() != "" )
655//                 {
656//                   PlayerStats::getStats( it->second.userId )->setNickName( it->second.handshake->getPreferedNickName() );
657//                 }
658              }
659
660              PRINT(0)("handshake finished delete it\n");
661              delete it->second.handshake;
662              it->second.handshake = NULL;
663            }
664          }
665
666        }
667        else
668        {
669          PRINT(1)("handshake failed!\n");
670          it->second.socket->disconnectServer();
671        }
672      }
673    }
674  }
675}
676
677
678/**
679 * this functions handles a reconnect event received from the a NET_MASTER_SERVER or NET_PROXY_SERVER
680 */
681void NetworkStream::handleReconnect(int userId)
682{
683  this->bRedirect = false;
684  PeerInfo* pInfo = &this->peers[userId];
685
686  PRINTF(0)("===============================================\n");
687  PRINTF(0)("Client is redirected to the other proxy servers\n");
688  PRINTF(0)("  user id: %i\n", userId);
689  PRINTF(0)("  connecting to: %s\n", this->networkMonitor->getFirstChoiceProxy()->ip.ipString().c_str());
690  PRINTF(0)("===============================================\n");
691
692  // flush the old synchronization states, since the numbering could be completely different
693  pInfo->lastAckedState = 0;
694  pInfo->lastRecvedState = 0;
695
696  // temp save the ip address here
697  IP proxyIP = pInfo->handshake->getProxy1Address();
698
699  // disconnect from the current server and reconnect to proxy server
700  this->handleDisconnect( userId);
701  this->connectToProxyServer(NET_ID_PROXY_SERVER_01, proxyIP.ipString(), 9999);
702  #warning the ports are not yet integrated correctly in the ip class
703
704  // and restart the handshake
705  this->startHandshake( userId);
706}
707
708
709/**
710 * handles the disconnect event
711 * @param userId id of the user to remove
712 */
713void NetworkStream::handleDisconnect( int userId )
714{
715  peers[userId].socket->disconnectServer();
716  delete peers[userId].socket;
717  peers[userId].socket = NULL;
718
719  if ( peers[userId].handshake )
720    delete peers[userId].handshake;
721  peers[userId].handshake = NULL;
722
723  if ( peers[userId].connectionMonitor )
724    delete peers[userId].connectionMonitor;
725  peers[userId].connectionMonitor = NULL;
726
727
728  for ( SynchronizeableList::iterator it2 = synchronizeables.begin(); it2 != synchronizeables.end(); it2++ )  {
729    (*it2)->cleanUpUser( userId );
730  }
731
732  if( SharedNetworkData::getInstance()->isMasterServer())
733    NetworkGameManager::getInstance()->signalLeftPlayer(userId);
734
735  this->freeSocketSlots.push_back( userId );
736
737  peers.erase( userId);
738}
739
740
741
742/**
743 * handle upstream network traffic
744 * @param tick: seconds elapsed since last update
745 */
746void NetworkStream::handleUpstream( int tick )
747{
748  int offset;
749  int n;
750
751  for ( PeerList::reverse_iterator peer = peers.rbegin(); peer != peers.rend(); peer++ )
752  {
753    offset = INTSIZE; // reserve enough space for the packet length
754
755    // continue with the next peer if this peer has no socket assigned (therefore no network)
756    if ( !peer->second.socket )
757      continue;
758
759    // header informations: current state
760    n = Converter::intToByteArray( currentState, buf + offset, UDP_PACKET_SIZE - offset );
761    assert( n == INTSIZE );
762    offset += n;
763
764    // header informations: last acked state
765    n = Converter::intToByteArray( peer->second.lastAckedState, buf + offset, UDP_PACKET_SIZE - offset );
766    assert( n == INTSIZE );
767    offset += n;
768
769    // header informations: last recved state
770    n = Converter::intToByteArray( peer->second.lastRecvedState, buf + offset, UDP_PACKET_SIZE - offset );
771    assert( n == INTSIZE );
772    offset += n;
773
774    // now write all synchronizeables in the packet
775    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
776    {
777
778      int oldOffset = offset;
779      Synchronizeable & sync = **it;
780
781
782      // do not include synchronizeables with uninit id and syncs that don't want to be synchronized
783      if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED )
784        continue;
785
786      // if handshake not finished only sync handshake
787      if ( peer->second.handshake && sync.getLeafClassID() != CL_HANDSHAKE )
788        continue;
789
790      // if we are a server (both master and proxy servers) and this is not our handshake
791      if ( ( SharedNetworkData::getInstance()->isMasterServer() ||
792             SharedNetworkData::getInstance()->isProxyServerActive() &&  peer->second.isClient())
793             && sync.getLeafClassID() == CL_HANDSHAKE && sync.getUniqueID() != peer->second.userId )
794        continue;
795
796      /* list of synchronizeables that will never be synchronized over the network: */
797      // do not sync null parent
798      if ( sync.getLeafClassID() == CL_NULL_PARENT )
799        continue;
800
801
802      assert( sync.getLeafClassID() != 0);
803
804      assert( offset + INTSIZE <= UDP_PACKET_SIZE );
805
806      // server fakes uniqueid == 0 for handshake synchronizeable
807      if ( ( SharedNetworkData::getInstance()->isMasterServer() ||
808             SharedNetworkData::getInstance()->isProxyServerActive() &&  peer->second.isClient() ) &&
809             ( sync.getUniqueID() >= 1000 || sync.getUniqueID() <= SharedNetworkData::getInstance()->getMaxPlayer() + 1)
810             /*<= SharedNetworkData::getInstance()->getMaxPlayer() + 1*/) // plus one to handle one client more than the max to redirect it
811        n = Converter::intToByteArray( 0, buf + offset, UDP_PACKET_SIZE - offset );
812      else
813        n = Converter::intToByteArray( sync.getUniqueID(), buf + offset, UDP_PACKET_SIZE - offset );
814
815
816      assert( n == INTSIZE );
817      offset += n;
818
819      // make space for packet size
820      offset += INTSIZE;
821
822      n = sync.getStateDiff( peer->second.userId, buf + offset, UDP_PACKET_SIZE-offset, currentState, peer->second.lastAckedState, -1000 );
823      offset += n;
824
825      assert( Converter::intToByteArray( n, buf + offset - n - INTSIZE, INTSIZE ) == INTSIZE );
826
827      // check if all data bytes == 0 -> remove data and the synchronizeable from the sync process since there is no update
828      // TODO not all synchronizeables like this maybe add Synchronizeable::canRemoveZeroDiff()
829      bool allZero = true;
830      for ( int i = 0; i < n; i++ )
831      {
832         if ( buf[i+oldOffset+2*INTSIZE] != 0 )
833           allZero = false;
834      }
835      // if there is no new data in this synchronizeable reset the data offset to the last state -> dont synchronizes
836      // data that hast not changed
837      if ( allZero )
838      {
839        offset = oldOffset;
840      }
841    } // all synchronizeables written
842
843
844
845    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
846    {
847      Synchronizeable & sync = **it;
848
849      // again exclude all unwanted syncs
850      if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED)
851        continue;
852
853      sync.handleSentState( peer->second.userId, currentState, peer->second.lastAckedState );
854    }
855
856
857    assert( Converter::intToByteArray( offset, buf, INTSIZE ) == INTSIZE );
858
859    // now compress the data with the zip library
860    int compLength = 0;
861    if ( SharedNetworkData::getInstance()->isMasterServer() ||
862         SharedNetworkData::getInstance()->isProxyServerActive())
863      compLength = Zip::getInstance()->zip( buf, offset, compBuf, UDP_PACKET_SIZE, dictServer );
864    else
865      compLength = Zip::getInstance()->zip( buf, offset, compBuf, UDP_PACKET_SIZE, dictClient );
866
867    if ( compLength <= 0 )
868    {
869      PRINTF(1)("compression failed!\n");
870      continue;
871    }
872
873    assert( peer->second.socket->writePacket( compBuf, compLength ) );
874
875    if ( this->remainingBytesToWriteToDict > 0 )
876      writeToNewDict( buf, offset, true );
877
878    peer->second.connectionMonitor->processUnzippedOutgoingPacket( tick, buf, offset, currentState );
879    peer->second.connectionMonitor->processZippedOutgoingPacket( tick, compBuf, compLength, currentState );
880
881  }
882}
883
884/**
885 * handle downstream network traffic
886 */
887void NetworkStream::handleDownstream( int tick )
888{
889  int offset = 0;
890
891  int length = 0;
892  int packetLength = 0;
893  int compLength = 0;
894  int uniqueId = 0;
895  int state = 0;
896  int ackedState = 0;
897  int fromState = 0;
898  int syncDataLength = 0;
899
900  for ( PeerList::iterator peer = peers.begin(); peer != peers.end(); peer++ )
901  {
902
903    if ( !peer->second.socket )
904      continue;
905
906    while ( 0 < (compLength = peer->second.socket->readPacket( compBuf, UDP_PACKET_SIZE )) )
907    {
908      peer->second.connectionMonitor->processZippedIncomingPacket( tick, compBuf, compLength );
909
910      packetLength = Zip::getInstance()->unZip( compBuf, compLength, buf, UDP_PACKET_SIZE );
911
912      if ( packetLength < 4*INTSIZE )
913      {
914        if ( packetLength != 0 )
915          PRINTF(1)("got too small packet: %d\n", packetLength);
916        continue;
917      }
918
919      if ( this->remainingBytesToWriteToDict > 0 )
920        writeToNewDict( buf, packetLength, false );
921
922      assert( Converter::byteArrayToInt( buf, &length ) == INTSIZE );
923      assert( Converter::byteArrayToInt( buf + INTSIZE, &state ) == INTSIZE );
924      assert( Converter::byteArrayToInt( buf + 2*INTSIZE, &fromState ) == INTSIZE );
925      assert( Converter::byteArrayToInt( buf + 3*INTSIZE, &ackedState ) == INTSIZE );
926      offset = 4*INTSIZE;
927
928      peer->second.connectionMonitor->processUnzippedIncomingPacket( tick, buf, packetLength, state, ackedState );
929
930
931      //if this is an old state drop it
932      if ( state <= peer->second.lastRecvedState )
933        continue;
934
935      if ( packetLength != length )
936      {
937        PRINTF(1)("real packet length (%d) and transmitted packet length (%d) do not match!\n", packetLength, length);
938        peer->second.socket->disconnectServer();
939        continue;
940      }
941
942      while ( offset + 2 * INTSIZE < length )
943      {
944        // read the unique id of the sync
945        assert( offset > 0 );
946        assert( Converter::byteArrayToInt( buf + offset, &uniqueId ) == INTSIZE );
947        offset += INTSIZE;
948
949        // read the data length
950        assert( Converter::byteArrayToInt( buf + offset, &syncDataLength ) == INTSIZE );
951        offset += INTSIZE;
952
953        assert( syncDataLength > 0 );
954        assert( syncDataLength < 10000 );
955
956        Synchronizeable * sync = NULL;
957
958        // look for the synchronizeable in question
959        for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
960        {
961          // client thinks his handshake has a special id: hostId * 1000 (host id of this server)
962          if ( (*it)->getUniqueID() == uniqueId ||
963                 ( uniqueId == 0  && (*it)->getUniqueID() == peer->second.userId ) )
964          {
965            sync = *it;
966            break;
967          }
968        }
969
970        // this synchronizeable does not yet exist! create it
971        if ( sync == NULL )
972        {
973          PRINTF(0)("could not find sync with id %d. try to create it\n", uniqueId);
974
975          // if it is an old synchronizeable already removed, ignore it
976          if ( oldSynchronizeables.find( uniqueId ) != oldSynchronizeables.end() )
977          {
978            offset += syncDataLength;
979            continue;
980          }
981
982          // if the node we got this unknown sync we ignore it if:
983          //  - the remote host is a client
984          //  - the remote host is a proxy server and we are master server
985          // (since it has no rights to create a new sync)
986          if ( peers[peer->second.userId].isClient() ||
987               (peers[peer->second.userId].isProxyServerActive() && SharedNetworkData::getInstance()->isMasterServer()))
988          {
989            offset += syncDataLength;
990            continue;
991          }
992
993          int leafClassId;
994          if ( INTSIZE > length - offset )
995          {
996            offset += syncDataLength;
997            continue;
998          }
999
1000          Converter::byteArrayToInt( buf + offset, &leafClassId );
1001
1002          assert( leafClassId != 0 );
1003
1004
1005          BaseObject * b = NULL;
1006          /* These are some small exeptions in creation: Not all objects can/should be created via Factory */
1007          /* Exception 1: NullParent */
1008          if( leafClassId == CL_NULL_PARENT || leafClassId == CL_SYNCHRONIZEABLE || leafClassId == CL_NETWORK_GAME_MANAGER )
1009          {
1010            PRINTF(1)("Don't create Object with ID %x, ignored!\n", (int)leafClassId);
1011            offset += syncDataLength;
1012            continue;
1013          }
1014          else
1015            b = Factory::fabricate( (ClassID)leafClassId );
1016
1017          if ( !b )
1018          {
1019            PRINTF(1)("Could not fabricate Object with classID %x\n", leafClassId);
1020            offset += syncDataLength;
1021            continue;
1022          }
1023
1024          if ( b->isA(CL_SYNCHRONIZEABLE) )
1025          {
1026            sync = dynamic_cast<Synchronizeable*>(b);
1027            sync->setUniqueID( uniqueId );
1028            sync->setSynchronized(true);
1029
1030            PRINTF(0)("Fabricated %s with id %d\n", sync->getClassCName(), sync->getUniqueID());
1031          }
1032          else
1033          {
1034            PRINTF(1)("Class with ID %x is not a synchronizeable!\n", (int)leafClassId);
1035            delete b;
1036            offset += syncDataLength;
1037            continue;
1038          }
1039        }
1040
1041
1042        int n = sync->setStateDiff( peer->second.userId, buf+offset, syncDataLength, state, fromState );
1043        offset += n;
1044
1045      }
1046
1047      if ( offset != length )
1048      {
1049        PRINTF(0)("offset (%d) != length (%d)\n", offset, length);
1050        peer->second.socket->disconnectServer();
1051      }
1052
1053
1054      for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
1055      {
1056        Synchronizeable & sync = **it;
1057
1058        if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED )
1059          continue;
1060
1061        sync.handleRecvState( peer->second.userId, state, fromState );
1062      }
1063
1064      assert( peer->second.lastAckedState <= ackedState );
1065      peer->second.lastAckedState = ackedState;
1066
1067      assert( peer->second.lastRecvedState < state );
1068      peer->second.lastRecvedState = state;
1069
1070    }
1071
1072  }
1073
1074}
1075
1076/**
1077 * is executed when a handshake has finished
1078 */
1079void NetworkStream::handleNewClient( int userId )
1080{
1081  // init and assign the message manager
1082  MessageManager::getInstance()->initUser( userId );
1083  // do all game relevant stuff here
1084  networkGameManager->signalNewPlayer( userId );
1085}
1086
1087
1088/**
1089 * removes old items from oldSynchronizeables
1090 */
1091void NetworkStream::cleanUpOldSyncList( )
1092{
1093  int now = SDL_GetTicks();
1094
1095  for ( std::map<int,int>::iterator it = oldSynchronizeables.begin(); it != oldSynchronizeables.end();  )
1096  {
1097    if ( it->second < now - 10*1000 )
1098    {
1099      std::map<int,int>::iterator delIt = it;
1100      it++;
1101      oldSynchronizeables.erase( delIt );
1102      continue;
1103    }
1104    it++;
1105  }
1106}
1107
1108/**
1109 * writes data to DATA/dicts/newdict
1110 * @param data pointer to data
1111 * @param length length
1112 */
1113void NetworkStream::writeToNewDict( byte * data, int length, bool upstream )
1114{
1115  if ( remainingBytesToWriteToDict <= 0 )
1116    return;
1117
1118  if ( length > remainingBytesToWriteToDict )
1119    length = remainingBytesToWriteToDict;
1120
1121  std::string fileName = ResourceManager::getInstance()->getDataDir();
1122  fileName += "/dicts/newdict";
1123
1124  if ( upstream )
1125    fileName += "_upstream";
1126  else
1127    fileName += "_downstream";
1128
1129  FILE * f = fopen( fileName.c_str(), "a" );
1130
1131  if ( !f )
1132  {
1133    PRINTF(2)("could not open %s\n", fileName.c_str());
1134    remainingBytesToWriteToDict = 0;
1135    return;
1136  }
1137
1138  if ( fwrite( data, 1, length, f ) != length )
1139  {
1140    PRINTF(2)("could not write to file\n");
1141    fclose( f );
1142    return;
1143  }
1144
1145  fclose( f );
1146
1147  remainingBytesToWriteToDict -= length;
1148}
1149
1150
1151
1152
1153
1154
Note: See TracBrowser for help on using the repository browser.