Changeset 2483

Timestamp:

Feb 24, 2009, 10:06:43 PM (16 years ago)

Author:

Christoph Mayer

Message:

-autolinks impl. (funktioniert noch nicht komplett, macht aber im moment nichts schlechter)
-einige fixes im ablauf etc.

Files:

• etc/pingpong/settings_initiator.cnf (modified) (1 diff)
• etc/pingpong/settings_node2.cnf (modified) (1 diff)
• sample/pingpong/PingPong.cpp (modified) (6 diffs)
• sample/pingpong/PingPong.h (modified) (2 diffs)
• source/ariba/communication/BaseCommunication.cpp (modified) (21 diffs)
• source/ariba/communication/BaseCommunication.h (modified) (5 diffs)
• source/ariba/overlay/BaseOverlay.cpp (modified) (12 diffs)
• source/ariba/overlay/BaseOverlay.h (modified) (8 diffs)

etc/pingpong/settings_initiator.cnf

@@ -4 +4 @@
 ariba.tcp.port = 5002
 #ariba.udp.port =
-ariba.bootstrap.hints=pingpong{ip{127.0.0.1},tcp(ip,{5002})}
+#ariba.bootstrap.hints=pingpong{ip{127.0.0.1},tcp(ip,{5002})}

etc/pingpong/settings_node2.cnf

@@ -2 +2 @@
 node.initiator = false
 #ariba.ip.addr =
-ariba.tcp.port = 5010
+ariba.tcp.port = 5004
 #ariba.udp.port =
 ariba.bootstrap.hints=pingpong{ip{127.0.0.1},tcp(ip,{5002})}

sample/pingpong/PingPong.cpp

@@ -17 +17 @@
 // construction
 PingPong::PingPong() : pingId( 0 ) {
+        Timer::setInterval( 5000 );
 }
 
@@ -70 +71 @@
 
         // bind communication and node listener
-        node->bind(this);
-        node->bind(this, PingPong::PINGPONG_ID);
+        node->bind( this );
+        node->bind( this, PingPong::PINGPONG_ID);
         
+        // start the ping timer. if we are not
+        // the initiator this will happen in onJoinCompleted
+        if( isInitiator ) Timer::start();
+
         // ping pong started up...
         logging_info( "pingpong started up ");
@@ -85 +90 @@
         Timer::stop();
 
-        // unbind listeners
+        // unbind communication and node listener
         node->unbind( this );
         node->unbind( this, PingPong::PINGPONG_ID );
@@ -108 +113 @@
 
         // start the timer to ping every second
-        Timer::setInterval( 1000 );
         Timer::start();
 }
@@ -117 +121 @@
 
 // communication listener
-bool PingPong::onLinkRequest(const NodeID& remote, Message* msg) {
+bool PingPong::onLinkRequest(const NodeID& remote, const DataMessage& msg) {
         return false;
 }
 
-void PingPong::onMessage(Message* msg, const NodeID& remote, 
-                const LinkID& lnk = LinkID::UNSPECIFIED) {
+void PingPong::onMessage(const DataMessage& msg, const NodeID& remote, const LinkID& lnk) {
 
-        PingPongMessage* pingmsg = msg->decapsulate<PingPongMessage> ();
+        PingPongMessage* pingmsg = msg.getMessage()->decapsulate<PingPongMessage> ();
 
         logging_info( "received ping message on link " << lnk.toString()
                         << " from node " << remote.toString()
-                        << ": " << pingmsg->toString() );
+                        << ": " << pingmsg->info() );
 }
 
@@ -142 +145 @@
         PingPongMessage pingmsg( pingId );
         node->sendBroadcastMessage( pingmsg, PingPong::PINGPONG_ID );
-
 }
 

sample/pingpong/PingPong.h

@@ -35 +35 @@
 protected:
         // communication listener interface
-        virtual bool onLinkRequest(const NodeID& remote, Message* msg);
-        virtual void onMessage(Message* msg, const NodeID& remote, const LinkID& lnk);
+        virtual bool onLinkRequest(const NodeID& remote, const DataMessage& msg);
+        virtual void onMessage(const DataMessage& msg, const NodeID& remote, const LinkID& lnk= LinkID::UNSPECIFIED);
 
         // node listener interface
@@ -62 +62 @@
         // the current ping id
         unsigned long pingId;
+
 };
 

source/ariba/communication/BaseCommunication.cpp

@@ -85 +85 @@
 
         for( ; i != iend; i++){
-                logging_debug( "local locator found " + (*i)->toString() );
+                logging_debug( "local locator found " << (*i)->toString() );
                 IPv4Locator* ipv4locator = dynamic_cast<IPv4Locator*>(*i);
 
@@ -101 +101 @@
                         localDescriptor.locator = ipv4locator;
                         localDescriptor.isUnspec = false;
-                        logging_info( "binding to addr = " + ipv4locator->toString() );
+                        logging_info( "binding to addr = " << ipv4locator->toString() );
                         foundLocator = true;
                         break;
@@ -180 +180 @@
         LinkID linkid = LinkID::create();
 
-        logging_debug( "creating new local descriptor entry with local link id " + linkid.toString() );
-        LinkDescriptor linkDescriptor( linkid, local, LinkID::UNSPECIFIED, remote, descriptor );
+        logging_debug( "creating new local descriptor entry with local link id " << linkid.toString() );
+        LinkDescriptor linkDescriptor( linkid, local, LinkID::UNSPECIFIED, remote, descriptor, false );
         addLink( linkDescriptor );
 
@@ -189 +189 @@
         //
 
-        logging_debug( "sending out base messages with request to open link to " + remote->toString() );
+        logging_debug( "sending out base messages with request to open link to " << remote->toString() );
         AribaBaseMsg baseMsg( remote, AribaBaseMsg::LINK_STATE_OPEN_REQUEST, linkid,
                                                                 LinkID::UNSPECIFIED );
@@ -206 +206 @@
                 logging_error( "don't know the link you want to drop" );
                 return;
+        }
+
+        // warn if this link has some queued messages attached
+        if( descriptor.waitingmsg.size() > 0 ){
+                logging_warn( "dropping link " << link.toString() << 
+                        " that has " << descriptor.waitingmsg.size() << " waiting messages" );
         }
 
@@ -231 +237 @@
 seqnum_t BaseCommunication::sendMessage( const LinkID lid, const Message* message) {
 
-        logging_debug( "sending out message to link " + lid.toString() );
+        logging_debug( "sending out message to link " << lid.toString() );
 
         // query local link info
         LinkDescriptor& linkDesc = queryLocalLink(lid);
         if( linkDesc.isUnspecified() ){
-                logging_error( "don't know the link with id " + lid.toString() );
-                return 0;
+                logging_error( "don't know the link with id " << lid.toString() );
+                return -1;
         }
 
@@ -251 +257 @@
         msg.encapsulate( const_cast<Message*>(message) );
 
-        // send message
-        transport->sendMessage( &msg );
-
-        return ++currentSeqnum;
+        if( linkDesc.linkup ){
+
+                // send message
+                transport->sendMessage( &msg );
+                return ++currentSeqnum;
+
+        } else {
+
+                // queue message
+                logging_info( "link " << lid.toString() << " is not up yet, queueing message" );
+                linkDesc.waitingmsg.push_back( new Message(msg) ); // TODO ooooo
+
+                return 0;
+
+        } // if( linkDesc.linkup )
 }
 
@@ -300 +317 @@
 
         AribaBaseMsg* spovmsg = ((Message*)message)->decapsulate<AribaBaseMsg>();
-        logging_debug( "receiving base comm message of type " + spovmsg->getTypeString() );
+        logging_debug( "receiving base comm message of type " << spovmsg->getTypeString() );
 
         //
@@ -351 +368 @@
                 const NetworkLocator* remoteLocator = dynamic_cast<const NetworkLocator*>(message->getSourceAddress());
 
-                logging_debug( "localLocator=" + localLocator->toString() + " remoteLocator="+remoteLocator->toString());
+                logging_debug( "localLocator=" << localLocator->toString() 
+                                << " remoteLocator=" << remoteLocator->toString());
 
                 // ask the registered listeners if this link
@@ -372 +390 @@
 
                 LinkDescriptor linkDescriptor(localLink, localLocator, remoteLink,
-                                        remoteLocator, EndpointDescriptor(remoteLocator));
+                                        remoteLocator, EndpointDescriptor(remoteLocator), true);
 
                 logging_debug( "saving new link descriptor with " <<
@@ -378 +396 @@
                                 "[local locator " << localLocator->toString() << "] " <<
                                 "[remote link " << remoteLink.toString() << "] " <<
-                                "[remote locator " << remoteLocator->toString() << "]" );
+                                "[remote locator " << remoteLocator->toString() << "]" <<
+                                "[link up true]" );
 
                 addLink( linkDescriptor );
@@ -400 +419 @@
                 // the link is now open
                 //
-
+                
                 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
                         i->onLinkUp( localLink, localLocator, remoteLocator );
@@ -420 +439 @@
 
                 if (linkDesc.isUnspecified()) {
-                        logging_warn("Failed to find local link "+spovmsg->getRemoteLink().toString());
+                        logging_warn("Failed to find local link " << spovmsg->getRemoteLink().toString());
                         return false;
                 }
 
                 linkDesc.remoteLink = spovmsg->getLocalLink();
-                logging_debug( "the link is now up with local link id " + spovmsg->getRemoteLink().toString() );
+                linkDesc.linkup = true; 
+        
+                logging_debug( "the link is now up with local link id " << spovmsg->getRemoteLink().toString() );
 
                 BOOST_FOREACH( CommunicationEvents* i, eventListener ){
@@ -431 +452 @@
                 }
 
+                if( linkDesc.waitingmsg.size() > 0 ){ 
+                        logging_info( "sending out queued messages on link " << linkDesc.localLink.toString() );
+
+                        BOOST_FOREACH( Message* msg, linkDesc.waitingmsg ){
+                                sendMessage( linkDesc.localLink, msg );
+                                delete msg;
+                        }
+
+                        linkDesc.waitingmsg.clear();
+                }
+
         } // LINK_STATE_OPEN_REPLY
 
@@ -440 +472 @@
 
                 const LinkID& localLink = spovmsg->getRemoteLink();
-                logging_debug( "received link close request for link " + localLink.toString() );
+                logging_debug( "received link close request for link " << localLink.toString() );
 
                 //
@@ -450 +482 @@
                 LinkDescriptor& linkDesc = queryLocalLink( localLink );
                 if (linkDesc.isUnspecified()) {
-                        logging_warn("Failed to find local link "+localLink.toString());
+                        logging_warn("Failed to find local link " << localLink.toString());
                         return false;
                 }
@@ -469 +501 @@
 
                 const LinkID& localLink = spovmsg->getRemoteLink();
-                logging_debug( "received link update for link " + localLink.toString() );
+                logging_debug( "received link update for link " << localLink.toString() );
 
                 //
@@ -477 +509 @@
                 LinkDescriptor& linkDesc = queryLocalLink( localLink );
                 if (linkDesc.isUnspecified()) {
-                        logging_warn("Failed to update local link "+localLink.toString());
+                        logging_warn("Failed to update local link " << localLink.toString());
                         return false;
                 }
@@ -517 +549 @@
 
         for( ; i != iend; i++){
-                if( (*i).localLink == localLink){
-                        linkSet.erase( i );
-                        break;
-                }
+                if( (*i).localLink != localLink) continue;
+                
+                BOOST_FOREACH( Message* msg, i->waitingmsg ){
+                        delete msg;
+                }
+
+                i->waitingmsg.clear();
+                linkSet.erase( i );
+
+                break;
         }
 }
@@ -568 +606 @@
                 info.type != NetworkChangeInterface::EventTypeAddressDelete ) return;
 
-        logging_info("base communication is handling network address changes");
+        logging_info( "base communication is handling network address changes" );
 
         //
@@ -610 +648 @@
 
         for( ; i != iend; i++){
-                logging_debug( "local locator found " + (*i)->toString() );
+                logging_debug( "local locator found " << (*i)->toString() );
                 IPv4Locator* ipv4locator = dynamic_cast<IPv4Locator*>(*i);
 

source/ariba/communication/BaseCommunication.h

@@ -46 +46 @@
 #include <vector>
 #include <iostream>
+#include <deque>
 #include <algorithm>
 #include <boost/foreach.hpp>
@@ -77 +78 @@
 using std::map;
 using std::vector;
+using std::deque;
 using std::pair;
 using std::make_pair;
@@ -222 +224 @@
                         remoteLink(),
                         remoteLocator(NULL),
-                        remoteEndpoint(EndpointDescriptor::UNSPECIFIED){
+                        remoteEndpoint(EndpointDescriptor::UNSPECIFIED),
+                        linkup(false) {
                 }
 
                 LinkDescriptor(const LinkID& _localLink, const NetworkLocator*& _localLocator,
                                 const LinkID& _remoteLink, const NetworkLocator*& _remoteLocator,
-                                const EndpointDescriptor& _remoteEndpoint ) :
+                                const EndpointDescriptor& _remoteEndpoint, bool _linkup ) :
                         localLink(_localLink),
                         localLocator(_localLocator),
                         remoteLink(_remoteLink),
                         remoteLocator(_remoteLocator),
-                        remoteEndpoint(_remoteEndpoint){
+                        remoteEndpoint(_remoteEndpoint),
+                        linkup(_linkup) {
                 }
 
@@ -240 +244 @@
                         remoteLink(desc.remoteLink),
                         remoteLocator(desc.remoteLocator),
-                        remoteEndpoint(desc.remoteEndpoint){
+                        remoteEndpoint(desc.remoteEndpoint),
+                        linkup(desc.linkup) {
+
+                        BOOST_FOREACH( Message* msg, desc.waitingmsg ){
+                                waitingmsg.push_back( msg );
+                        }
                 }
 
@@ -252 +261 @@
                 const NetworkLocator*   remoteLocator;
                 EndpointDescriptor      remoteEndpoint;
+
+                bool                    linkup;
+                deque<Message*>         waitingmsg;
         };
 

source/ariba/overlay/BaseOverlay.cpp

@@ -69 +69 @@
 //              ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CHARACTER_W);
 //      }
+
+        // timer for auto link management
+        Timer::setInterval( 5000 );
+        Timer::start();
 }
 
@@ -75 +79 @@
         logging_info("deleting base overlay");
 
+        Timer::stop();
         bc.unregisterMessageReceiver( this );
         bc.unregisterEventListener( this );
@@ -147 +152 @@
         //
         // create the overlay
-        // and bootstrap against ourselfs
         //
 
         overlayInterface->createOverlay();
+        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
+                i->onOverlayCreate( spovnetId );
+        }
+
+        //
+        // bootstrap against ourselfs
+        //
+
         overlayInterface->joinOverlay();
+        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
+                i->onJoinSuccess( spovnetId );
+        }
 
         ovl.visChangeNodeIcon ( ovlId, nodeId, OvlVis::ICON_ID_CAMERA );
         ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN);
-
-        // inform all registered services of the event
-        BOOST_FOREACH( ServiceInterface* i, listenerMux.getOneList() ){
-                i->onOverlayCreate( spovnetId );
-        }
 }
 
@@ -245 +255 @@
         overmsg.encapsulate( const_cast<Message*>(message) );
 
+        i->second.markused();
         return bc.sendMessage( link, &overmsg );
 }
@@ -263 +274 @@
 
         if( link == LinkID::UNSPECIFIED ){
-                logging_error( "no link could be found to send message to node " <<
-                                node.toString() << " for service " << service.toString() );
-                return -1;
-        }
-
+
+                logging_info( "no link could be found to send message to node " <<
+                                node.toString() << " for service " << service.toString() <<
+                                ". creating auto link ...");
+                
+                const LinkID link = establishLink( node, service );
+                LinkMapping::iterator i = linkMapping.find( link );
+
+                if( i == linkMapping.end() ){
+                        logging_error( "failed to establish auto link to node " << node.toString() <<
+                                        " for service " << service.toString() );
+                        return -1;
+                }
+
+                i->second.autolink = true;
+
+        } // if( link != LinkID::UNSPECIFIED )
+
+        i->second.markused();
         return sendMessage( message, link );
 }
@@ -352 +377 @@
                                 " on link " << id.toString() );
 
-                OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeUpdate, i->second.service, nodeId );
+                OverlayMsg overMsg( 
+                        OverlayMsg::OverlayMessageTypeUpdate, 
+                        i->second.service, 
+                        nodeId 
+                        );
+
                 bc.sendMessage( id, &overMsg );
+                i->second.markused();
 
         } // if( i == linkMapping.end() )
@@ -397 +428 @@
         if( i->second.interface != NULL )
                 i->second.interface->onLinkChanged( id, nodeId, i->second.node );
+
+        i->second.markused();
 }
 
@@ -412 +445 @@
         if( i->second.interface != NULL )
                 i->second.interface->onLinkFail( id, nodeId, i->second.node );
+
+        i->second.markused();
 }
 
@@ -427 +462 @@
         if( i->second.interface != NULL )
                 i->second.interface->onLinkQoSChanged( id, nodeId, i->second.node, qos );
+
+        i->second.markused();
 }
 
@@ -434 +471 @@
         OverlayMsg* overlayMsg = ((Message*)message)->decapsulate<OverlayMsg>();
         if( overlayMsg == NULL ) return false;
+
+        // mark the link as in action
+        LinkMapping::iterator item = linkMapping.find( link );
+        if( item != linkMapping.end() ) item->second.markused();
 
         //
@@ -683 +724 @@
                 i->second.interface = iface;
                 iface->onLinkUp( link, nodeId, sourcenode );
+                i->second.markused();
 
                 return true ;
@@ -828 +870 @@
 }
 
+void BaseOverlay::eventFunction(){
+
+        list<LinkID> oldlinks;
+        time_t now = time(NULL);
+
+        // first gather all the links from linkMapping that need droppin
+        // don't directly drop, as the dropLink function affects the
+        // linkMapping structure that we are traversing here.
+        // drop links after a timeout of 30s
+
+        // the macro gets confused if type is passed directly
+        // because of the comma in the pair definition
+        typedef pair<LinkID,LinkItem> pairitem; 
+
+        BOOST_FOREACH( pairitem item, linkMapping ){
+                if( item.second.autolink && difftime(now, item.second.lastuse) > 30)
+                        oldlinks.push_back( item.first );
+        }
+
+        BOOST_FOREACH( const LinkID lnk, oldlinks ){
+                dropLink( lnk );
+        }
+}
+
 }} // namespace ariba, overlay

source/ariba/overlay/BaseOverlay.h

@@ -43 +43 @@
 #include <iostream>
 #include <algorithm>
+#include <ctime>
 #include <boost/foreach.hpp>
 
@@ -50 +51 @@
 #include "ariba/utility/misc/Demultiplexer.hpp"
 #include "ariba/utility/logging/Logging.h"
+#include "ariba/utility/system/Timer.h"
 
 #include "ariba/communication/EndpointDescriptor.h"
@@ -67 +69 @@
 using std::map;
 using std::make_pair;
+using std::pair;
 
 using ariba::communication::EndpointDescriptor;
@@ -92 +95 @@
 using ariba::utility::MessageSender;
 using ariba::utility::seqnum_t;
+using ariba::utility::Timer;
 
 #define ovl OvlVis::instance()
@@ -109 +113 @@
         public MessageReceiver,
         public CommunicationEvents,
-        public OverlayStructureEvents {
+        public OverlayStructureEvents,
+        protected Timer {
 
         use_logging_h( BaseOverlay );
@@ -248 +253 @@
         virtual void onNodeJoin( const NodeID& node );
 
+        // for timer events
+        virtual void eventFunction();
+
 private:
 
@@ -316 +324 @@
                 static const LinkItem UNSPECIFIED;
 
-                LinkItem( const LinkID& _link, const NodeID& _node,
+                LinkItem( const LinkID& _link, const NodeID& _node, 
                                 const ServiceID& _service, ServiceInterface* _interface )
-                        : link( _link ), node( _node ), service( _service ), interface( _interface ){
+                        : link( _link ), node( _node ), service( _service ), interface( _interface ), 
+                                autolink( false ), lastuse( time(NULL) ) {
                 }
+
+                // general information about the link
 
                 const LinkID link;
@@ -325 +336 @@
                 ServiceID service;
                 ServiceInterface* interface;
+
+                // information needed for auto links
+
+                void markused(){
+                        lastuse = time(NULL);
+                }
+
+                bool autolink;
+                time_t lastuse;
         };