Planet
navi homePPSaboutscreenshotsdownloaddevelopmentforum

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

Last change on this file since 7613 was 7613, checked in by rennerc, 18 years ago

subprojects/network can test handshake now

File size: 18.4 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: claudio
13   co-programmer:
14*/
15
16
17/* this is for debug output. It just says, that all calls to PRINT() belong to the DEBUG_MODULE_NETWORK module
18   For more information refere to https://www.orxonox.net/cgi-bin/trac.cgi/wiki/DebugOutput
19*/
20#define DEBUG_MODULE_NETWORK
21
22
23#include "base_object.h"
24#include "network_protocol.h"
25#include "udp_socket.h"
26#include "udp_server_socket.h"
27#include "connection_monitor.h"
28#include "synchronizeable.h"
29#include "network_game_manager.h"
30#include "shared_network_data.h"
31
32#include "lib/util/loading/factory.h"
33
34#include "debug.h"
35#include "class_list.h"
36#include <algorithm>
37
38/* include your own header */
39#include "network_stream.h"
40
41/* probably unnecessary */
42using namespace std;
43
44
45#define PACKAGE_SIZE  256
46
47
48NetworkStream::NetworkStream()
49    : DataStream()
50{
51  this->init();
52  /* initialize the references */
53  this->type = NET_CLIENT;
54  this->networkProtocol = new NetworkProtocol();
55  this->connectionMonitor = new ConnectionMonitor();
56}
57
58
59NetworkStream::NetworkStream( std::string host, int port )
60{
61  this->type = NET_CLIENT;
62  this->init();
63  this->peers[0].socket = new UdpSocket( host, port );
64  this->networkProtocol = new NetworkProtocol();
65  this->connectionMonitor = new ConnectionMonitor();
66}
67
68
69NetworkStream::NetworkStream( int port )
70{
71  this->type = NET_SERVER;
72  this->init();
73  this->serverSocket = new UdpServerSocket(port);
74  this->networkProtocol = new NetworkProtocol();
75  this->connectionMonitor = new ConnectionMonitor();
76  this->bActive = true;
77}
78
79
80void NetworkStream::init()
81{
82  /* set the class id for the base object */
83  this->setClassID(CL_NETWORK_STREAM, "NetworkStream");
84  this->bActive = false;
85  this->serverSocket = NULL;
86  this->networkGameManager = NULL;
87  myHostId = 0;
88  currentState = 0;
89}
90
91
92NetworkStream::~NetworkStream()
93{
94  if ( this->serverSocket )
95  {
96    serverSocket->close();
97    delete serverSocket;
98  }
99
100  for ( PeerList::iterator i = peers.begin(); i!=peers.end(); i++)
101  {
102    if ( i->second.socket )
103    {
104      i->second.socket->disconnectServer();
105      delete i->second.socket;
106      i->second.socket = NULL;
107    }
108   
109    if ( i->second.handshake )
110    {
111      delete i->second.handshake;
112      i->second.handshake = NULL;
113    }
114  }
115 
116  if ( serverSocket )
117  {
118    delete serverSocket;
119    serverSocket = NULL;
120  }
121
122  delete connectionMonitor;
123  delete networkProtocol;
124}
125
126
127void NetworkStream::createNetworkGameManager()
128{
129  this->networkGameManager = NetworkGameManager::getInstance();
130  // setUniqueID( maxCon+2 ) because we need one id for every handshake
131  // and one for handshake to reject client maxCon+1
132  this->networkGameManager->setUniqueID( MAX_CONNECTIONS + 2 );
133
134}
135
136
137void NetworkStream::startHandshake()
138{
139  Handshake* hs = new Handshake(false);
140  hs->setUniqueID( 0 );
141  assert( peers[0].handshake == NULL );
142  peers[0].handshake = hs;
143//   peers[0].handshake->setSynchronized( true );
144  //this->connectSynchronizeable(*hs);
145  //this->connectSynchronizeable(*hs);
146  PRINTF(0)("NetworkStream: Handshake created: %s\n", hs->getName());
147}
148
149
150void NetworkStream::connectSynchronizeable(Synchronizeable& sync)
151{
152  this->synchronizeables.push_back(&sync);
153  sync.setNetworkStream( this );
154
155  this->bActive = true;
156}
157
158
159void NetworkStream::disconnectSynchronizeable(Synchronizeable& sync)
160{
161  // removing the Synchronizeable from the List.
162  std::list<Synchronizeable*>::iterator disconnectSynchro = std::find(this->synchronizeables.begin(), this->synchronizeables.end(), &sync);
163  if (disconnectSynchro != this->synchronizeables.end())
164    this->synchronizeables.erase(disconnectSynchro);
165 
166  //TODO set timestamp
167  oldSynchronizeables[sync.getUniqueID()] = 0;
168}
169
170
171void NetworkStream::processData()
172{
173  currentState++;
174 
175  if ( this->type == NET_SERVER )
176  {
177    if ( serverSocket )
178      serverSocket->update();
179   
180    this->updateConnectionList();
181  }
182  else
183  {
184    if ( peers[0].socket && !peers[0].socket->isOk() )
185    {
186      PRINTF(1)("lost connection to server\n");
187
188      peers[0].socket->disconnectServer();
189      delete peers[0].socket;
190      peers[0].socket = NULL;
191
192      if ( peers[0].handshake )
193        delete peers[0].handshake;
194      peers[0].handshake = NULL;
195    }
196  }
197
198  handleHandshakes();
199  handleUpstream();
200  handleDownstream();
201
202
203
204  /* DOWNSTREAM */
205#if 0
206
207
208  int dataLength;
209  int reciever;
210  Header header;
211  int counter;
212
213  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
214  {
215    counter = 0;
216
217    if ( (*it)!=NULL && (*it)->beSynchronized() /*&& (*it)->getOwner() == myHostId*/ )
218    {
219      do {
220        counter++;
221
222        reciever = 0;
223#warning fix this
224dataLength = 0;
225//TODO fix
226        //dataLength = (*it)->readBytes(downBuffer, DATA_STREAM_BUFFER_SIZE, &reciever);
227
228        if ( dataLength<=0 ){
229          reciever = 0;
230          continue;
231        }
232
233        dataLength = networkProtocol->createHeader((byte*)downBuffer, dataLength, DATA_STREAM_BUFFER_SIZE, static_cast<const Synchronizeable&>(*(*it)));
234
235        Header* header = (Header*)downBuffer;
236        if ( header->synchronizeableID < this->maxConnections+2 )
237        {
238          //if ( !isServer() ) PRINTF(0)("RESET UNIQUEID FROM %d TO 0 maxCon=%d\n", header->synchronizeableID, this->maxConnections);
239          header->synchronizeableID = 0;
240        }
241        else
242        {
243          //if ( !isServer() ) PRINTF(0)("UNIQUEID=%d\n", header->synchronizeableID);
244        }
245
246        if ( dataLength<=0 )
247          continue;
248
249        if ( reciever!=0 )
250        {
251          if ( reciever < 0)
252          {
253            for ( int i = 0; i<networkSockets.size(); i++)
254            {
255              if ( i!=abs(reciever) && networkSockets[i] != NULL )
256              {
257                PRINTF(0)("write %d bytes to socket %d uniqueid %d reciever %d\n", dataLength, i, (*it)->getUniqueID(), reciever);
258                networkSockets[i]->writePacket(downBuffer, dataLength);
259              }
260            }
261          }
262          else
263          {
264            if ( networkSockets[reciever] != NULL )
265            {
266              PRINTF(5)("write %d bytes to socket %d\n", dataLength, reciever);
267              networkSockets[reciever]->writePacket(downBuffer, dataLength);
268            }
269            else
270            {
271              PRINTF(1)("networkSockets[reciever] == NULL\n");
272            }
273          }
274        }
275        else
276        {
277          for ( int i = 0; i<networkSockets.size(); i++)
278          {
279            if ( networkSockets[i] != NULL )
280            {
281              PRINTF(5)("write %d bytes to socket %d\n", dataLength, i);
282              networkSockets[i]->writePacket(downBuffer, dataLength);
283            }
284          }
285        }
286
287      } while( reciever!=0 );
288    }
289  }
290
291  /* UPSTREAM */
292
293  for ( int i = 0; i<networkSockets.size(); i++)
294  {
295    if ( networkSockets[i] )
296    {
297      do {
298        dataLength = networkSockets[i]->readPacket(upBuffer, DATA_STREAM_BUFFER_SIZE);
299
300        if ( dataLength<=0 )
301          continue;
302
303        header = networkProtocol->extractHeader(upBuffer, dataLength);
304        dataLength -= sizeof(header);
305
306        PRINTF(5)("read %d bytes from socket uniqueID = %d\n", dataLength, header.synchronizeableID);
307
308        if ( dataLength != header.length )
309        {
310          PRINTF(1)("packetsize in header and real packetsize do not match! %d:%d\n", dataLength, header.length);
311          continue;
312        }
313
314        if ( header.synchronizeableID == 0 )
315        {
316          header.synchronizeableID = i;
317        }
318
319        for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
320        {
321#warning fix this
322
323          if ( *it && (*it)->getUniqueID()==header.synchronizeableID )
324          {
325            if ( (*it)->writeBytes(upBuffer+sizeof(header), dataLength, i) != header.length )
326            {
327              PRINTF(1)("%s did not read all the data id = %d!\n", (*it)->getClassName(), (*it)->getUniqueID());
328              break;
329            }
330            continue;
331          }
332
333        }
334
335      } while ( dataLength>0 );
336    }
337
338  }
339#endif
340}
341
342void NetworkStream::updateConnectionList( )
343{
344  //check for new connections
345
346  NetworkSocket* tempNetworkSocket = serverSocket->getNewSocket();
347
348  if ( tempNetworkSocket )
349  {
350    int clientId;
351    if ( freeSocketSlots.size() >0 )
352    {
353      clientId = freeSocketSlots.back();
354      freeSocketSlots.pop_back();
355      peers[clientId].socket = tempNetworkSocket;
356      peers[clientId].handshake = new Handshake(true, clientId, this->networkGameManager->getUniqueID());
357      peers[clientId].handshake->setUniqueID(clientId);
358      peers[clientId].userId = clientId;
359    } else
360    {
361      clientId = 1;
362     
363      for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
364        if ( it->first >= clientId )
365          clientId = it->first + 1;
366     
367      peers[clientId].socket = tempNetworkSocket;
368      peers[clientId].handshake = new Handshake(true, clientId, this->networkGameManager->getUniqueID());
369      peers[clientId].handshake->setUniqueID(clientId);
370      peers[clientId].userId = clientId;
371     
372      PRINTF(0)("num sync: %d\n", synchronizeables.size());
373    }
374
375    if ( clientId > MAX_CONNECTIONS )
376    {
377      peers[clientId].handshake->doReject( "too many connections" );
378      PRINTF(0)("Will reject client %d because there are to many connections!\n", clientId);
379    }
380    else
381
382    PRINTF(0)("New Client: %d\n", clientId);
383
384    //this->connectSynchronizeable(*handshakes[clientId]);
385  }
386
387
388  //check if connections are ok else remove them
389  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
390  {
391    if ( it->second.socket && !it->second.socket->isOk() )
392    {
393      PRINTF(0)("Client is gone: %d\n", it->second.userId);
394
395      it->second.socket->disconnectServer();
396      delete it->second.socket;
397      it->second.socket = NULL;
398
399      if ( it->second.handshake )
400        delete it->second.handshake;
401      it->second.handshake = NULL;
402
403
404      NetworkGameManager::getInstance()->signalLeftPlayer(it->second.userId);
405
406      freeSocketSlots.push_back( it->second.userId );
407
408    }
409  }
410
411
412}
413
414void NetworkStream::debug()
415{
416  if( this->isServer())
417    PRINT(0)(" Host ist Server with ID: %i\n", this->myHostId);
418  else
419    PRINT(0)(" Host ist Client with ID: %i\n", this->myHostId);
420
421  PRINT(0)(" Got %i connected Synchronizeables, showing active Syncs:\n", this->synchronizeables.size());
422  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
423  {
424    if( (*it)->beSynchronized() == true)
425      PRINT(0)("  Synchronizeable of class: %s::%s, with unique ID: %i, Synchronize: %i\n", (*it)->getClassName(), (*it)->getName(),
426               (*it)->getUniqueID(), (*it)->beSynchronized());
427  }
428  PRINT(0)(" Maximal Connections: %i\n", MAX_CONNECTIONS );
429
430}
431
432
433int NetworkStream::getSyncCount()
434{
435  int n = 0;
436  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
437    if( (*it)->beSynchronized() == true)
438      ++n;
439
440  //return synchronizeables.size();
441  return n;
442}
443
444/**
445 * check if handshakes completed
446 */
447void NetworkStream::handleHandshakes( )
448{
449  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
450  {
451    if ( it->second.handshake )
452    {
453      if ( it->second.handshake->completed() )
454      {
455        if ( it->second.handshake->ok() )
456        {
457          if ( type != NET_SERVER )
458          {
459            SharedNetworkData::getInstance()->setHostID( it->second.handshake->getHostId() );
460            myHostId = SharedNetworkData::getInstance()->getHostID();
461
462            this->networkGameManager = NetworkGameManager::getInstance();
463            this->networkGameManager->setUniqueID( it->second.handshake->getNetworkGameManagerId() );
464          }
465
466          PRINT(0)("handshake finished id=%d\n", it->second.handshake->getNetworkGameManagerId());
467          assert(false);
468
469          delete it->second.handshake;
470          it->second.handshake = NULL;
471        }
472        else
473        {
474          PRINT(1)("handshake failed!\n");
475          it->second.socket->disconnectServer();
476          assert(false);
477        }
478      }
479    }
480  }
481}
482
483/**
484 * handle upstream network traffic
485 */
486void NetworkStream::handleUpstream( )
487{
488  byte buf[UDP_PACKET_SIZE];
489  int offset;
490  int n;
491 
492  for ( PeerList::iterator peer = peers.begin(); peer != peers.end(); peer++ )
493  {
494    offset = INTSIZE; //make already space for length
495   
496    if ( !peer->second.socket )
497      continue;
498   
499    n = Converter::intToByteArray( currentState, buf + offset, UDP_PACKET_SIZE - offset );
500    assert( n == INTSIZE );
501    offset += n;
502   
503    n = Converter::intToByteArray( peer->second.lastAckedState, buf + offset, UDP_PACKET_SIZE - offset );
504    assert( n == INTSIZE );
505    offset += n;
506   
507    n = Converter::intToByteArray( peer->second.lastRecvedState, buf + offset, UDP_PACKET_SIZE - offset );
508    assert( n == INTSIZE );
509    offset += n;
510   
511    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
512    {
513      Synchronizeable & sync = **it;
514     
515      if ( !sync.beSynchronized() || sync.getUniqueID() < 0 )
516        continue;
517
518      //if handshake not finished only sync handshake
519      if ( peer->second.handshake && sync.getLeafClassID() != CL_HANDSHAKE )
520        continue;
521     
522      if ( isServer() && sync.getLeafClassID() == CL_HANDSHAKE && sync.getUniqueID() != peer->second.userId )
523        continue;
524
525      assert( offset + INTSIZE <= UDP_PACKET_SIZE );
526     
527      //server fakes uniqueid=0 for handshake
528      if ( this->isServer() && sync.getUniqueID() < MAX_CONNECTIONS - 1 )
529        n = Converter::intToByteArray( 0, buf + offset, UDP_PACKET_SIZE - offset );
530      else
531        n = Converter::intToByteArray( sync.getUniqueID(), buf + offset, UDP_PACKET_SIZE - offset );
532      assert( n == INTSIZE );
533      offset += n;
534     
535      //make space for size
536      offset += INTSIZE;
537     
538      n = sync.getStateDiff( peer->second.userId, buf + offset, UDP_PACKET_SIZE-offset, currentState, peer->second.lastAckedState, 0 );
539      offset += n;
540     
541      assert( Converter::intToByteArray( n, buf + offset - n, INTSIZE ) == INTSIZE );
542    }
543   
544    assert( Converter::intToByteArray( offset, buf, INTSIZE ) == INTSIZE );
545   
546    assert( peer->second.socket->writePacket( buf, offset ) );
547    PRINTF(0)("send packet: %d userId = %d %x %d\n", offset, peer->second.userId, peer->second.handshake->getLeafClassID(), peer->second.handshake->getUniqueID());
548    assert( peer->second.handshake->getLeafClassID() == CL_HANDSHAKE );
549  }
550}
551
552/**
553 * handle downstream network traffic
554 */
555void NetworkStream::handleDownstream( )
556{
557  byte buf[UDP_PACKET_SIZE];
558  int offset = 0;
559 
560  int length = 0;
561  int packetLength = 0;
562  int uniqueId = 0;
563  int state = 0;
564  int ackedState = 0;
565  int fromState = 0;
566  int syncDataLength = 0;
567 
568  for ( PeerList::iterator peer = peers.begin(); peer != peers.end(); peer++ )
569  {
570    if ( !peer->second.socket )
571      continue;
572
573    packetLength = peer->second.socket->readPacket( buf, UDP_PACKET_SIZE );
574
575    if ( packetLength < 4*INTSIZE )
576    {
577      if ( packetLength != 0 )
578        PRINTF(1)("got too small packet: %d\n", packetLength);
579      continue;
580    }
581   
582    assert( Converter::byteArrayToInt( buf, &length ) == INTSIZE );
583    assert( Converter::byteArrayToInt( buf + INTSIZE, &state ) == INTSIZE );
584    assert( Converter::byteArrayToInt( buf + 2*INTSIZE, &fromState ) == INTSIZE );
585    assert( Converter::byteArrayToInt( buf + 3*INTSIZE, &ackedState ) == INTSIZE );
586    offset = 4*INTSIZE;
587
588    PRINTF(0)("got packet: %d, %d\n", length, packetLength);
589   
590    //if this is an old state drop it
591    if ( state <= peer->second.lastRecvedState )
592      continue;
593   
594    if ( packetLength != length )
595    {
596      PRINTF(1)("real packet length (%d) and transmitted packet length (%d) do not match!\n", packetLength, length);
597      peer->second.socket->disconnectServer();
598      continue;
599    }
600   
601    while ( offset < length )
602    {
603      assert( Converter::byteArrayToInt( buf + offset, &uniqueId ) == INTSIZE );
604      offset += INTSIZE;
605     
606      assert( Converter::byteArrayToInt( buf + offset, &syncDataLength ) == INTSIZE );
607      offset += INTSIZE;
608     
609      Synchronizeable * sync = NULL;
610     
611      for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
612      { 
613        //                                        client thinks his handshake has id 0!!!!!
614        if ( (*it)->getUniqueID() == uniqueId || ( uniqueId == 0 && (*it)->getUniqueID() == peer->second.userId ) )
615        {
616          sync = *it;
617          break;
618        }
619      }
620     
621      if ( sync == NULL )
622      {
623        if ( oldSynchronizeables.find( uniqueId ) != oldSynchronizeables.end() )
624        {
625          offset += syncDataLength;
626          continue;
627        }
628       
629        //TODO dont accept new object from all peers (probably only servers)
630        int leafClassId;
631        if ( INTSIZE > length - offset )
632          break;
633       
634        Converter::byteArrayToInt( buf + offset, &leafClassId );
635       
636        BaseObject * b;
637        /* These are some small exeptions in creation: Not all objects can/should be created via Factory */
638        /* Exception 1: NullParent */
639        if( leafClassId == CL_NULL_PARENT)
640        {
641          PRINTF(1)("Can not create Class with ID %x!", (int)leafClassId);
642          break;
643        }
644        else
645          b = Factory::fabricate( (ClassID)leafClassId );
646
647        if ( !b )
648        {
649          PRINTF(1)("Could not fabricate Object with classID %x\n", leafClassId);
650          break;
651        }
652       
653        if ( b->isA(CL_SYNCHRONIZEABLE) )
654        {
655          sync = dynamic_cast<Synchronizeable*>(b);
656          sync->setUniqueID( uniqueId );
657          sync->setSynchronized(true);
658 
659          PRINTF(0)("Fabricated %s with id %d\n", sync->getClassName(), sync->getUniqueID());
660        }
661        else
662        {
663          PRINTF(1)("Class with ID %x is not a synchronizeable!", (int)leafClassId);
664          delete b;
665          break;
666        }
667      }
668      offset += sync->setStateDiff( peer->second.userId, buf+offset, length-offset, state, fromState );
669    }
670   
671    if ( offset != length )
672    {
673      peer->second.socket->disconnectServer();
674    }
675   
676    peer->second.lastAckedState = ackedState;
677  }
678}
679
680
681
682
683
684
Note: See TracBrowser for help on using the repository browser.