Changeset 8327 for code/trunk/src/libraries/network/Connection.cc

Timestamp:

Apr 25, 2011, 8:22:36 PM (13 years ago)

Author:

scheusso

Message:

merging network6 into trunk

Location:

code/trunk

Files:

• . (modified) (1 prop)
• src/libraries/network/Connection.cc (modified) (11 diffs)

code/trunk/src/libraries/network/Connection.cc

@@ -38 +38 @@
 
 #include "packet/Packet.h"
+#include <util/Sleep.h>
 
 namespace orxonox
 {
-  const boost::posix_time::millisec NETWORK_COMMUNICATION_THREAD_WAIT_TIME(20);
-
-  Connection::Connection():
-    host_(0), bCommunicationThreadRunning_(false)
+  const boost::posix_time::millisec NETWORK_COMMUNICATION_THREAD_WAIT_TIME(200);
+  const unsigned int                NETWORK_DISCONNECT_TIMEOUT = 500;
+
+  Connection::Connection(uint32_t firstPeerID):
+    host_(0), bCommunicationThreadRunning_(false), nextPeerID_(firstPeerID)
   {
     enet_initialize();
@@ -50 +52 @@
     this->incomingEventsMutex_ = new boost::mutex;
     this->outgoingEventsMutex_ = new boost::mutex;
+//     this->overallMutex_ = new boost::mutex;
   }
 
@@ -75 +78 @@
   }
 
-
-//   int Connection::service(ENetEvent* event) {
-//     return enet_host_service( this->host_, event, NETWORK_WAIT_TIMEOUT );
-//   }
-
-  void Connection::disconnectPeer(ENetPeer *peer)
-  {
-    assert(peer);
-    outgoingEvent outEvent = { peer, outgoingEventType::disconnectPeer, (ENetPacket*)10, 15 };
+  void Connection::disconnectPeer(uint32_t peerID)
+  {
+//     this->overallMutex_->lock();
+    outgoingEvent outEvent = { peerID, outgoingEventType::disconnectPeer, 0, 0 };
     
     this->outgoingEventsMutex_->lock();
     this->outgoingEvents_.push_back(outEvent);
     this->outgoingEventsMutex_->unlock();
-  }
-
-  void Connection::addPacket(ENetPacket *packet, ENetPeer *peer, uint8_t channelID)
-  {
-    assert(peer);
-    outgoingEvent outEvent = { peer, outgoingEventType::sendPacket, packet, channelID };
+//     this->overallMutex_->unlock();
+  }
+  
+  void Connection::disconnectPeers()
+  {
+    outgoingEvent outEvent = { 0, outgoingEventType::disconnectPeers, 0, 0 };
     
     this->outgoingEventsMutex_->lock();
@@ -99 +97 @@
     this->outgoingEventsMutex_->unlock();
   }
+
+  void Connection::addPacket(ENetPacket* packet, uint32_t peerID, uint8_t channelID)
+  {
+//     this->overallMutex_->lock();
+    outgoingEvent outEvent = { peerID, outgoingEventType::sendPacket, packet, channelID };
+    
+    this->outgoingEventsMutex_->lock();
+    this->outgoingEvents_.push_back(outEvent);
+    this->outgoingEventsMutex_->unlock();
+//     this->overallMutex_->unlock();
+  }
   
   void Connection::broadcastPacket(ENetPacket* packet, uint8_t channelID)
   {
-    outgoingEvent outEvent = { (ENetPeer*)15, outgoingEventType::broadcastPacket, packet, channelID };
+//     this->overallMutex_->lock();
+    outgoingEvent outEvent = { 0, outgoingEventType::broadcastPacket, packet, channelID };
     
     this->outgoingEventsMutex_->lock();
     this->outgoingEvents_.push_back(outEvent);
     this->outgoingEventsMutex_->unlock();
+//     this->overallMutex_->unlock();
   }
 
@@ -114 +125 @@
     ENetEvent event;
     
+//     this->overallMutex_->lock();
     while( bCommunicationThreadRunning_ )
     {
@@ -119 +131 @@
       while( enet_host_check_events( this->host_, &event ) > 0 )
       {
-//         COUT(0) << "incoming event" << endl;
-        // received an event
-        this->incomingEventsMutex_->lock();
-        this->incomingEvents_.push_back(event);
-        this->incomingEventsMutex_->unlock();
-      }
+        processIncomingEvent(event);
+      }
+      
+//       this->overallMutex_->unlock();
+      msleep(1);
+//       this->overallMutex_->lock();
       
       // Send all waiting outgoing packets
@@ -138 +150 @@
         this->outgoingEventsMutex_->unlock();
         
-        switch( outEvent.type )
-        {
-          case outgoingEventType::sendPacket:
-            enet_peer_send( outEvent.peer, outEvent.channelID, outEvent.packet );
-            break;
-          case outgoingEventType::disconnectPeer:
-            enet_peer_disconnect(outEvent.peer, 0);
-            break;
-          case outgoingEventType::broadcastPacket:
-            enet_host_broadcast( this->host_, outEvent.channelID, outEvent.packet );
-            break;
-          default:
-            assert(0);
-        }
+        processOutgoingEvent(outEvent);
+        
         this->outgoingEventsMutex_->lock();
         outgoingEventsCount = this->outgoingEvents_.size();
@@ -160 +160 @@
       if( enet_host_service( this->host_, &event, NETWORK_WAIT_TIMEOUT ) > 0 )
       {
-//         COUT(0) << "incoming event after wait" << endl;
-        //received an event
-        this->incomingEventsMutex_->lock();
-        this->incomingEvents_.push_back(event);
-        this->incomingEventsMutex_->unlock();
+        processIncomingEvent(event);
+      }
+    }
+//     this->overallMutex_->unlock();
+  }
+  
+  void Connection::processIncomingEvent(ENetEvent& event)
+  {
+    incomingEvent inEvent;
+    // preprocess event
+    switch( event.type )
+    {
+      case ENET_EVENT_TYPE_CONNECT:
+        inEvent = preprocessConnectEvent(event);
+        break;
+      case ENET_EVENT_TYPE_RECEIVE:
+        inEvent = preprocessReceiveEvent(event);
+        break;
+      case ENET_EVENT_TYPE_DISCONNECT:
+        inEvent = preprocessDisconnectEvent(event);
+        break;
+      case ENET_EVENT_TYPE_NONE:
+      default:
+        return;
+    }
+    
+    // pushing event to queue
+    this->incomingEventsMutex_->lock();
+    this->incomingEvents_.push_back(inEvent);
+    this->incomingEventsMutex_->unlock();
+  }
+  
+  void Connection::processOutgoingEvent(outgoingEvent& event)
+  {
+    ENetPeer* peer;
+    switch( event.type )
+    {
+      case outgoingEventType::sendPacket:
+        // check whether the peer is still/already in the peer list
+        if( this->peerMap_.find(event.peerID) != this->peerMap_.end() )
+        {
+          peer = this->peerMap_[event.peerID];
+          enet_peer_send( peer, event.channelID, event.packet );
+        }
+        else
+        {
+          // peer probably already disconnected so just discard packet
+          assert(event.peerID<this->nextPeerID_);
+          enet_packet_destroy(event.packet);
+        }
+        break;
+      case outgoingEventType::disconnectPeer:
+        if( this->peerMap_.find(event.peerID) != this->peerMap_.end() )
+        {
+          peer = this->peerMap_[event.peerID];
+          enet_peer_disconnect(peer, 0);
+        }
+        else
+        {
+          // peer probably already disconnected so just discard disconnect event
+          assert(event.peerID<this->nextPeerID_);
+        }
+        break;
+      case outgoingEventType::disconnectPeers:
+        disconnectPeersInternal();
+        break;
+      case outgoingEventType::broadcastPacket:
+        enet_host_broadcast( this->host_, event.channelID, event.packet );
+        break;
+      default:
+        assert(0);
+    }
+  }
+
+
+  void Connection::disconnectPeersInternal()
+  {
+    std::map<uint32_t, ENetPeer*>::iterator it;
+    for( it=this->peerMap_.begin(); it!=this->peerMap_.end(); ++it )
+    {
+      enet_peer_disconnect(it->second, 0);
+    }
+    uint32_t iterations = NETWORK_DISCONNECT_TIMEOUT/NETWORK_WAIT_TIMEOUT;
+    uint32_t i = 0;
+    while( this->peerMap_.size() && i++ < iterations )
+    {
+      ENetEvent event;
+      if( enet_host_service( this->host_, &event, NETWORK_WAIT_TIMEOUT ) > 0 )
+      {
+        processIncomingEvent(event);
       }
     }
@@ -171 +256 @@
   void Connection::processQueue()
   {
-    ENetEvent event;
+    incomingEvent inEvent;
 
     this->incomingEventsMutex_->lock();
@@ -178 +263 @@
     while( incomingEventsCount > 0 )
     {
-      packet::Packet* p;
+      // pop event from queue
       this->incomingEventsMutex_->lock();
-      event = this->incomingEvents_.front();
+      inEvent = this->incomingEvents_.front();
       this->incomingEvents_.pop_front();
       this->incomingEventsMutex_->unlock();
       
-      switch(event.type)
-      {
-        // log handling ================
-        case ENET_EVENT_TYPE_CONNECT:
-          addPeer( &event );
+      // process event
+      switch( inEvent.type )
+      {
+        case incomingEventType::peerConnect:
+          addPeer(inEvent.peerID);
           break;
-        case ENET_EVENT_TYPE_DISCONNECT:
-          removePeer( &event );
+        case incomingEventType::peerDisconnect:
+          removePeer(inEvent.peerID);
           break;
-        case ENET_EVENT_TYPE_RECEIVE:
-//           COUT(0) << "ENET_EVENT_TYPE_RECEIVE" << endl;
-          p = createPacket( &event );
-          processPacket(p);
+        case incomingEventType::receivePacket:
+          processPacket(inEvent.packet);
           break;
-        case ENET_EVENT_TYPE_NONE:
+        default:
           break;
       }
       
+      // check whether there are still events in the queue
       this->incomingEventsMutex_->lock();
       incomingEventsCount = this->incomingEvents_.size();
@@ -207 +291 @@
     }
   }
-
-  packet::Packet* Connection::createPacket(ENetEvent* event)
-  {
-    packet::Packet *p = packet::Packet::createPacket(event->packet, event->peer);
-    return p;
-//     return p->process();
-  }
+  
+  void Connection::waitOutgoingQueue()
+  {
+    uint32_t outgoingEventsCount;
+    this->outgoingEventsMutex_->lock();
+    outgoingEventsCount = this->outgoingEvents_.size();
+    this->outgoingEventsMutex_->unlock();
+    while( outgoingEventsCount )
+    {
+      msleep(1);
+      this->outgoingEventsMutex_->lock();
+      outgoingEventsCount = this->outgoingEvents_.size();
+      this->outgoingEventsMutex_->unlock();
+    }
+  }
+
+
+  incomingEvent Connection::preprocessConnectEvent(ENetEvent& event)
+  {
+    // make sure this peer doesn't exist
+    assert( this->peerMap_.find(this->nextPeerID_) == this->peerMap_.end() );
+    assert( this->peerIDMap_.find(event.peer) == this->peerIDMap_.end() );
+    
+    // give peer a new id and increase peerID for next peer
+    uint32_t peerID = this->nextPeerID_;
+    ++this->nextPeerID_;
+    
+    // add peer/peerID into peerMap_ and peerIDMap_
+    this->peerMap_[peerID] = event.peer;
+    this->peerIDMap_[event.peer] = peerID;
+    
+    // create new peerEvent and return it
+    incomingEvent inEvent = { peerID, incomingEventType::peerConnect, 0 };
+    return inEvent;
+  }
+  
+  incomingEvent Connection::preprocessDisconnectEvent(ENetEvent& event)
+  {
+    // assert that the peer exists and get peerID
+    assert( this->peerIDMap_.find(event.peer) != this->peerIDMap_.end() );
+    uint32_t peerID = this->peerIDMap_[event.peer];
+    
+    // remove peer/peerID from maps
+    this->peerIDMap_.erase(this->peerIDMap_.find(event.peer));
+    this->peerMap_.erase(this->peerMap_.find(peerID));
+    
+    // create new peerEvent and return it
+    incomingEvent inEvent = { peerID, incomingEventType::peerDisconnect, 0 };
+    return inEvent;
+  }
+  
+  incomingEvent Connection::preprocessReceiveEvent(ENetEvent& event)
+  {
+    // assert that the peer exists and get peerID
+    assert( this->peerIDMap_.find(event.peer) != this->peerIDMap_.end() );
+    uint32_t peerID = this->peerIDMap_[event.peer];
+    
+    // create new Packet from ENetPacket
+    packet::Packet* p = packet::Packet::createPacket(event.packet, peerID);
+    
+    // create new peerEvent and return it
+    incomingEvent inEvent = { peerID, incomingEventType::receivePacket, p };
+    return inEvent;
+  }
+
   
   void Connection::enableCompression()