Changeset 3712

Timestamp:

May 26, 2009, 8:03:39 PM (16 years ago)

Author:

Christoph Mayer

Message:

-autolink message buffer von basecomm nach baseoverlay und komplette logik hochgezogen damit mit neuem overlay routing zurecht kommt, damit ist letzer fehler raus

Location:

source/ariba

Files:

• communication/BaseCommunication.cpp (modified) (5 diffs)
• communication/BaseCommunication.h (modified) (4 diffs)
• overlay/BaseOverlay.cpp (modified) (12 diffs)
• overlay/BaseOverlay.h (modified) (5 diffs)
• utility/system/StartupWrapper.cpp (modified) (1 diff)

source/ariba/communication/BaseCommunication.cpp

@@ -217 +217 @@
         }
 
-        // 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" );
-        }
-
         // create message to drop the link
         logging_debug( "sending out link close request. for us, the link is closed now" );
@@ -266 +260 @@
         msg.encapsulate( const_cast<Message*>(message) );
 
-        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 )
+        if( !linkDesc.linkup ){
+                logging_error("cant send message on link " << lid.toString() << ", link not up");
+                return -1;
+        }
+
+        // send message
+        transport->sendMessage( &msg );
+        return ++currentSeqnum;
 }
 
@@ -466 +453 @@
                 }
 
-                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
 
@@ -503 +479 @@
                         i->onLinkDown( linkDesc.localLink, linkDesc.localLocator, linkDesc.remoteLocator );
                 }
-
-                //
-                // delete all pending messages for the link that has been closed
-                //
-
-                BOOST_FOREACH( Message* msg, linkDesc.waitingmsg ){
-                        delete msg;
-                }
-
-                linkDesc.waitingmsg.clear();
 
                 //
@@ -579 +545 @@
                 if( (*i).localLink != localLink) continue;
 
-                BOOST_FOREACH( Message* msg, i->waitingmsg ){
-                        delete msg;
-                }
-
-                i->waitingmsg.clear();
                 linkSet.erase( i );
-
                 break;
         }

source/ariba/communication/BaseCommunication.h

@@ -46 +46 @@
 #include <vector>
 #include <iostream>
-#include <deque>
 #include <algorithm>
 #include <boost/foreach.hpp>
@@ -78 +77 @@
 using std::map;
 using std::vector;
-using std::deque;
 using std::pair;
 using std::make_pair;
@@ -256 +254 @@
                         remoteEndpoint(desc.remoteEndpoint),
                         linkup(desc.linkup) {
-
-                        BOOST_FOREACH( Message* msg, desc.waitingmsg ){
-                                waitingmsg.push_back( msg );
-                        }
                 }
 
@@ -266 +260 @@
                 }
 
-                LinkID                  localLink;
+                LinkID                                  localLink;
                 const NetworkLocator*   localLocator;
-                LinkID                  remoteLink;
+                LinkID                                  remoteLink;
                 const NetworkLocator*   remoteLocator;
-                EndpointDescriptor      remoteEndpoint;
-
-                bool                    linkup;
-                deque<Message*>         waitingmsg;
+                EndpointDescriptor              remoteEndpoint;
+
+                bool                                    linkup;
         };
 

source/ariba/overlay/BaseOverlay.cpp

@@ -270 +270 @@
         LinkMapping::iterator i = linkMapping.find( link );
 
+        // find the link item to drop
         if( i == linkMapping.end() ){
                 logging_warn( "can't drop link, mapping unknown " << link.toString() );
@@ -275 +276 @@
         }
 
+        LinkItem item = i->second;
+
+        // delete all queued messages
+        if( item.waitingmsg.size() > 0 ){
+
+                logging_warn( "dropping link " << link.toString() <<
+                        " that has " << item.waitingmsg.size() << " waiting messages" );
+
+                item.deleteWaiting();
+        }
+
+        // erase the mapping and drop the link
         linkMapping.erase( i );
-
-        LinkItem item = i->second;
         bc->dropLink( link );
 
+        // tell sideports and listeners of the drop
         item.interface->onLinkDown( link, item.node );
         sideport->onLinkDown(link, this->nodeId, item.node, this->spovnetId );
@@ -287 +299 @@
 
         logging_debug( "baseoverlay is sending data message on link " << link.toString() );
+
+        //
+        // get the mapping for this link
+        //
 
         LinkMapping::iterator i = linkMapping.find( link );
@@ -294 +310 @@
         }
 
-        OverlayMsg overmsg(
-                OverlayMsg::OverlayMessageTypeData,     i->second.service, nodeId );
+        i->second.markused();
+
+        //
+        // check if the link is up yet, if its an autlink queue message
+        //
+
+        if( !i->second.linkup ){
+
+                if( i->second.autolink ){
+                        logging_info( "auto link " << link.toString() << " is not up yet, queueing message" );
+                        i->second.waitingmsg.push_back( new Message(*message) );
+                } else {
+                        logging_error("link " << link.toString() << " is not up yet, dropping message" );
+                }
+
+                return -1;
+        }
+
+        //
+        // send the message through the basecomm
+        //
+
+        OverlayMsg overmsg( OverlayMsg::OverlayMessageTypeData, i->second.service, nodeId );
         overmsg.encapsulate( const_cast<Message*>(message) );
 
-        i->second.markused();
         return bc->sendMessage( link, &overmsg );
 }
@@ -308 +344 @@
         LinkMapping::iterator i = linkMapping.begin();
         LinkMapping::iterator iend = linkMapping.end();
+
+        //
+        // see if we find a link for this node and service destination
+        //
 
         for( ; i != iend; i++ ){
@@ -316 +356 @@
         }
 
+        //
+        // if we found no link, create an auto link
+        //
+
         if( link == LinkID::UNSPECIFIED ){
 
@@ -322 +366 @@
                                 ". creating auto link ...");
 
+                // call basecomm to create a link
                 link = establishLink( node, service );
+
+                // this will call onlinkup on us, if everything worked we now have a mapping
                 LinkMapping::iterator i = linkMapping.find( link );
+                i->second.autolink = true;
 
                 if( i == linkMapping.end() || link == LinkID::UNSPECIFIED ){
@@ -330 +378 @@
                         return -1;
                 }
-
-                i->second.autolink = true;
 
                 logging_debug( "establishing autolink in progress to node "
@@ -531 +577 @@
         sideport->onLinkDown( id, this->nodeId, i->second.node, this->spovnetId );
 
+        // delete all queued messages
+        if( i->second.waitingmsg.size() > 0 ){
+
+                logging_warn( "dropping link " << id.toString() <<
+                        " that has " << i->second.waitingmsg.size() << " waiting messages" );
+
+                i->second.deleteWaiting();
+        }
+
         linkMapping.erase( i );
 }
@@ -788 +843 @@
 
                 // update our link mapping information for this link
-                bool changed =
-                        ( i->second.node != sourcenode ) ||
-                        ( i->second.service != service );
+                bool changed = ( i->second.node != sourcenode ) || ( i->second.service != service );
                 i->second.node = sourcenode;
                 i->second.service = service;
@@ -818 +871 @@
 
                 // ask the service whether it wants to accept this link
-                if( iface->onLinkRequest(sourcenode) ){
-
-                        logging_debug("link " << link.toString() <<
-                                        " has been accepted by service " << service.toString());
-
-                        // call the notification functions
-                        iface->onLinkUp( link, sourcenode );
-                        sideport->onLinkUp( link, nodeId, sourcenode, this->spovnetId );
-
-                } else {
+                if( !iface->onLinkRequest(sourcenode) ){
 
                         logging_debug("link " << link.toString() <<
@@ -836 +880 @@
                         // drop the link
                         dropLink( link );
-                }
+
+                        return true;
+                }
+
+                //
+                // link has been accepted, link is now up, send messages out first
+                //
+
+                i->second.linkup = true;
+                logging_debug("link " << link.toString() <<
+                                                " has been accepted by service " << service.toString() << " and is now up");
+
+                if( i->second.waitingmsg.size() > 0 ){
+                        logging_info( "sending out queued messages on link " << link.toString() );
+
+                        BOOST_FOREACH( Message* msg, i->second.waitingmsg ){
+                                sendMessage( msg, link );
+                                delete msg;
+                        }
+
+                        i->second.waitingmsg.clear();
+                }
+
+                // call the notification functions
+                iface->onLinkUp( link, sourcenode );
+                sideport->onLinkUp( link, nodeId, sourcenode, this->spovnetId );
 
                 return true;

source/ariba/overlay/BaseOverlay.h

@@ -46 +46 @@
 #include <list>
 #include <vector>
+#include <deque>
 #include <boost/foreach.hpp>
 
@@ -80 +81 @@
 using std::pair;
 using std::find;
+using std::deque;
 
 using ariba::NodeListener;
@@ -420 +422 @@
 
                 LinkItem() :
-                        link(LinkID::UNSPECIFIED), node(NodeID::UNSPECIFIED), service(
-                                        ServiceID::UNSPECIFIED), interface(
-                                        &CommunicationListener::DEFAULT), autolink(false), lastuse(
-                                        0) {
+                        link(LinkID::UNSPECIFIED),
+                        node(NodeID::UNSPECIFIED),
+                        service(ServiceID::UNSPECIFIED),
+                        interface(&CommunicationListener::DEFAULT),
+                        autolink(false),
+                        lastuse(0),
+                        linkup(false){
                 }
 
                 LinkItem(const LinkID& _link, const NodeID& _node,
                                 const ServiceID& _service, CommunicationListener* _interface) :
-                        link(_link), node(_node), service(_service), interface(_interface),
-                                        autolink(false), lastuse(time(NULL)) {
+                        link(_link),
+                        node(_node),
+                        service(_service),
+                        interface(_interface),
+                        autolink(false),
+                        lastuse(time(NULL)),
+                        linkup(false){
 
                         assert( _interface != NULL );
                 }
 
+                LinkItem(const LinkItem& rh) :
+                        link(rh.link),
+                        node(rh.node),
+                        service(rh.service),
+                        interface(rh.interface),
+                        autolink(rh.autolink),
+                        lastuse(rh.lastuse),
+                        linkup(rh.linkup){
+
+                        BOOST_FOREACH( Message* msg, rh.waitingmsg ){
+                                waitingmsg.push_back( msg );
+                        }
+                }
+
+                void deleteWaiting(){
+                        BOOST_FOREACH( Message* msg, waitingmsg ){
+                                delete msg;
+                        }
+                        waitingmsg.clear();
+                }
+
                 // general information about the link
-
                 const LinkID link;
                 NodeID node;
                 ServiceID service;
                 CommunicationListener* interface;
+                bool linkup;
 
                 // information needed for auto links
@@ -448 +479 @@
                 bool autolink;
                 time_t lastuse;
+                deque<Message*> waitingmsg;
         };
 
@@ -467 +499 @@
 };
 
-}
-} // namespace ariba, overlay
+}} // namespace ariba, overlay
 
 #endif /*BASEOVERLAY_H_*/

source/ariba/utility/system/StartupWrapper.cpp

@@ -108 +108 @@
         // set up again an individual level if you like
         {
-                // log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BaseOverlay"));
-                // logger->setLevel(log4cxx::Level::getInfo());
+//              log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BaseOverlay"));
+//              logger->setLevel(log4cxx::Level::getDebug());
+        }
+        {
+//              log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BaseCommunication"));
+//              logger->setLevel(log4cxx::Level::getDebug());
         }