Changeset 5870 for source/ariba/overlay/modules/chord/Chord.cpp

Timestamp:

Aug 11, 2009, 4:11:02 PM (15 years ago)

Author:

Christoph Mayer

Message:

merge noch nicht fertig

File:

• source/ariba/overlay/modules/chord/Chord.cpp (modified) (23 diffs)

source/ariba/overlay/modules/chord/Chord.cpp

@@ -38 +38 @@
 
 #include "ariba/overlay/BaseOverlay.h"
+#include "ariba/overlay/messages/OverlayMsg.h"
 
 #include "Chord.h"
-#include "messages/ChordMessage.h"
+#include "detail/chord_routing_table.hpp"
+
 #include "messages/Discovery.h"
-
-#include "detail/chord_routing_table.hpp"
 
 namespace ariba {
 namespace overlay {
+
+enum signalMessageTypes {
+        typeDiscovery = OverlayMsg::typeSignalingStart + 0x01,
+        typeLeave = OverlayMsg::typeSignalingStart + 0x02,
+};
 
 typedef chord_routing_table::item route_item;
@@ -57 +62 @@
 
         // create routing table
+<<<<<<< .working
         this->table = new chord_routing_table(_nodeid, 2);
+=======
+        this->table = new chord_routing_table(_nodeid, 4);
+>>>>>>> .merge-rechts.r5869
         orphan_removal_counter = 0;
         discovery_count = 0;
@@ -71 +80 @@
 
 /// helper: sets up a link using the base overlay
-LinkID Chord::setup(const EndpointDescriptor& endp, const NodeID& node, const NodeID& remoteRelay ) {
-        logging_debug("Request to setup link to " << endp.toString() );
+LinkID Chord::setup(const EndpointDescriptor& endpoint, const NodeID& remote ) {
 
         // check if we already have a connection
         for (int i=0; i<table->size(); i++)
-                if ((*table)[i]->id == node && !((*table)[i]->info.isUnspecified()))
+                if ((*table)[i]->ref_count > 0 && (*table)[i]->id == remote && !((*table)[i]->info.isUnspecified()))
                         return LinkID::UNSPECIFIED;
 
         // check if we are already trying to establish a link
         for (size_t i=0; i<pending.size(); i++)
-                if ( pending[i] == node ) {
-                        logging_debug("Already trying to establish a link to node " << node.toString() );
+                if ( pending[i] == remote ) {
+                        logging_debug("Already trying to establish a link to node "
+                                << remote.toString() );
                         return LinkID::UNSPECIFIED;
                 }
 
         // adding node to list of pending connections
-        pending.push_back( node );
+        pending.push_back( remote );
+
+        logging_info("Request to setup link to " << endpoint.toString() );
 
         // establish link via base overlay
+<<<<<<< .working
         return baseoverlay.establishLink(endp, node, OverlayInterface::OVERLAY_SERVICE_ID, remoteRelay );
+=======
+        return baseoverlay.establishLink( endpoint, remote,
+                        OverlayInterface::OVERLAY_SERVICE_ID );
+>>>>>>> .merge-rechts.r5869
 }
 
 /// helper: sends a message using the "base overlay"
-seqnum_t Chord::send(Message* msg, const LinkID& link) {
+seqnum_t Chord::send( OverlayMsg* msg, const LinkID& link ) {
         if (link.isUnspecified()) return 0;
-        return baseoverlay.sendMessage(msg, link);
+        msg->setRelayed(true);
+        return baseoverlay.send_link( msg, link );
 }
 
 /// sends a discovery message
-void Chord::send_discovery_to(const NodeID& destination, int ttl) {
+void Chord::send_discovery_to(const NodeID& remote, int ttl) {
+        LinkID link = getNextLinkId(remote);
+        if ( remote == nodeid || link.isUnspecified()) return;
         if ( table->size() == 0 ) return;
 
-        ChordMessage cmsg(ChordMessage::discovery, nodeid, destination);
-        Discovery dmsg;
-        dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor());
-        dmsg.setSourceRelay(baseoverlay.getRelayNode(destination));
-        dmsg.setFollowType(Discovery::normal);
-        dmsg.setTTL((uint8_t) ttl);
-        cmsg.encapsulate(&dmsg);
-
+        OverlayMsg msg( typeDiscovery );
+        msg.setRelayed(true);
+        Discovery dmsg( Discovery::normal, (uint8_t)2, baseoverlay.getEndpointDescriptor() );
+        msg.encapsulate(&dmsg);
+
+<<<<<<< .working
         // get next hop
         const route_item* item = table->get_next_hop(destination);
         if (item!=NULL && !item->info.isUnspecified()) send(&cmsg, item->info);
-}
-
-void Chord::discover_neighbors( const LinkID& lnk ) {
+=======
+        // send to node
+        baseoverlay.send_node( &msg, remote );
+>>>>>>> .merge-rechts.r5869
+}
+
+void Chord::discover_neighbors( const LinkID& link ) {
+        uint8_t ttl = 2;
+        {
+                // send predecessor discovery
+                OverlayMsg msg( typeDiscovery );
+                msg.setRelayed(true);
+                Discovery dmsg( Discovery::predecessor, ttl,
+                        baseoverlay.getEndpointDescriptor() );
+                msg.encapsulate(&dmsg);
+                send(&msg, link);
+        }
         {
                 // send successor discovery
-                ChordMessage cmsg(ChordMessage::discovery, nodeid, nodeid);
-                Discovery dmsg;
-                dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor());
-                dmsg.setSourceRelay(baseoverlay.getRelayNode(nodeid));
-                dmsg.setFollowType(Discovery::successor);
-                dmsg.setTTL((uint8_t)4);
-                cmsg.encapsulate(&dmsg);
-                send(&cmsg, lnk);
-        }
-        {
-                // send predecessor discovery
-                ChordMessage cmsg(ChordMessage::discovery, nodeid, nodeid);
-                Discovery dmsg;
-                dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor());
-                dmsg.setSourceRelay(baseoverlay.getRelayNode(nodeid));
-                dmsg.setFollowType(Discovery::predecessor);
-                dmsg.setTTL((uint8_t)4);
-                cmsg.encapsulate(&dmsg);
-                send(&cmsg, lnk);
+                OverlayMsg msg( typeDiscovery );
+                msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() );
+                msg.setRelayed(true);
+                Discovery dmsg( Discovery::successor, ttl,
+                        baseoverlay.getEndpointDescriptor() );
+                msg.encapsulate(&dmsg);
+                send(&msg, link);
         }
 }
@@ -166 +185 @@
         for (size_t i = 0; i < table->size(); i++) {
                 route_item* it = (*table)[i];
-                ChordMessage msg(ChordMessage::leave, nodeid, it->id);
-                send(&msg,it->info);
-        }
-}
-
+                OverlayMsg msg( typeLeave );
+                send( &msg, it->info );
+        }
+}
+
+/// @see OverlayInterface.h
 const EndpointDescriptor& Chord::resolveNode(const NodeID& node) {
         const route_item* item = table->get(node);
@@ -177 +197 @@
 }
 
-void Chord::routeMessage(const NodeID& destnode, Message* msg) {
+/// @see OverlayInterface.h
+const LinkID& Chord::getNextLinkId( const NodeID& id ) const {
         // get next hop
+<<<<<<< .working
         const route_item* item = table->get_next_hop(destnode);
-
+=======
+        const route_item* item = table->get_next_hop(id);
+>>>>>>> .merge-rechts.r5869
+
+<<<<<<< .working
         // message for this node? yes-> delegate to base overlay
         if (item->id == nodeid || destnode == nodeid)
                 baseoverlay.incomingRouteMessage( msg, LinkID::UNSPECIFIED, nodeid );
-
+=======
+        // returns a unspecified id when this is itself
+        if (item == NULL || item->id == nodeid)
+                return LinkID::UNSPECIFIED;
+>>>>>>> .merge-rechts.r5869
+
+<<<<<<< .working
         else { // no-> send to next hop
                 ChordMessage cmsg(ChordMessage::route, nodeid, destnode);
@@ -190 +222 @@
                 send(&cmsg, item->info);
         }
-}
-
+=======
+        /// return routing info
+        return item->info;
+>>>>>>> .merge-rechts.r5869
+}
+
+<<<<<<< .working
 /// @see OverlayInterface.h
 void Chord::routeMessage(const NodeID& node, const LinkID& link, Message* msg) {
@@ -214 +251 @@
 }
 
+=======
+>>>>>>> .merge-rechts.r5869
 OverlayInterface::NodeList Chord::getKnownNodes(bool deep) const {
         OverlayInterface::NodeList nodelist;
@@ -243 +282 @@
 /// @see OverlayInterface.h
 void Chord::onLinkUp(const LinkID& lnk, const NodeID& remote) {
-        logging_debug("link_up: link=" << lnk.toString() << " remote=" <<
+        logging_info("link_up: link=" << lnk.toString() << " remote=" <<
                         remote.toString() );
         for (vector<NodeID>::iterator i=pending.begin(); i!=pending.end(); i++)
@@ -250 +289 @@
                         break;
                 }
+/*
+        // check if we already have a connection, yes-> do not handle duplicate!
+        for (int i=0; i<table->size(); i++)
+                if ((*table)[i]->id == remote && !((*table)[i]->info.isUnspecified()) && (*table)[i]->info != lnk) {
+
+                        return;
+                }
+*/
+        if (remote==nodeid) {
+                baseoverlay.dropLink(lnk);
+                return;
+        }
+
         route_item* item = table->insert(remote);
 
@@ -257 +309 @@
                                 << " with link " << lnk.toString());
                 // replace with new link
-                if (!item->info.isUnspecified())
+                if (!item->info.isUnspecified() || item->info!=lnk)
                         baseoverlay.dropLink(item->info);
                 item->info = lnk;
+                // discover neighbors of new overlay neighbor
+                discover_neighbors( lnk );
+                showLinks();
         } else { // no-> add orphan entry to routing table
                 logging_info("new orphan: " << remote.toString()
@@ -266 +321 @@
         }
 
-        discover_neighbors( lnk );
-
+        // erase bootstrap link
         vector<LinkID>::iterator it = std::find(bootstrapLinks.begin(), bootstrapLinks.end(), lnk);
-        if( it != bootstrapLinks.end() ) {
-                discover_neighbors( lnk );
-                bootstrapLinks.erase( it );
-        }
+        if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it );
 }
 
@@ -282 +333 @@
         // remove link from routing table
         route_item* item = table->get(remote);
-        if (item!=NULL) item->info = LinkID::UNSPECIFIED;
-        table->remove(remote);
+        if (item!=NULL && item->info==lnk) {
+                item->info = LinkID::UNSPECIFIED;
+                table->remove(remote);
+        }
 }
 
@@ -292 +345 @@
 
         // decode message
-        typedef ChordMessage M;
-        M* m = msg.getMessage()->convert<ChordMessage> ();
+        OverlayMsg* m = dynamic_cast<OverlayMsg*>(msg.getMessage());
         if (m == NULL) return;
 
@@ -299 +351 @@
         switch (m->getType()) {
 
+<<<<<<< .working
         // invalid message
         case M::invalid:
@@ -324 +377 @@
                 // discovery request
         case M::discovery: {
+=======
+        // discovery request
+        case typeDiscovery: {
+>>>>>>> .merge-rechts.r5869
                 // decapsulate message
                 Discovery* dmsg = m->decapsulate<Discovery> ();
-                logging_debug("received discovery message with"
-                            << " src=" << m->getSource().toString()
-                                << " dst=" << m->getDestination().toString()
+                logging_debug("Received discovery message with"
+                            << " src=" << m->getSourceNode().toString()
+                                << " dst=" << m->getDestinationNode().toString()
                                 << " ttl=" << (int)dmsg->getTTL()
-                                << " type=" << (int)dmsg->getFollowType()
+                                << " type=" << (int)dmsg->getType()
                 );
 
                 // check if source node can be added to routing table and setup link
-                if (m->getSource() != nodeid && table->is_insertable(m->getSource()))
-                        setup(*dmsg->getSourceEndpoint(), m->getSource(), dmsg->getSourceRelay() );
+                if (m->getSourceNode() != nodeid
+                        && table->is_insertable(m->getSourceNode()))
+                        setup( dmsg->getEndpoint(), m->getSourceNode() );
 
                 // delegate discovery message
-                switch (dmsg->getFollowType()) {
+                switch (dmsg->getType()) {
 
                 // normal: route discovery message like every other message
                 case Discovery::normal: {
                         // closest node? yes-> split to follow successor and predecessor
-                        if (table->is_closest_to(m->getDestination())
-                                || (table->get_successor()!=NULL && *table->get_successor() == m->getDestination())
-                                || (table->get_predesessor()!=NULL && *table->get_predesessor() == m->getDestination())
-                        ) {
+                        if ( table->is_closest_to(m->getDestinationNode()) ) {
 
                                 if (table->get_successor() != NULL) {
-                                        // send successor message
-                                        ChordMessage cmsg_s(*m);
-                                        Discovery dmsg_s(*dmsg);
-                                        dmsg_s.setFollowType(Discovery::successor);
-                                        cmsg_s.encapsulate(&dmsg_s);
+                                        OverlayMsg omsg(*m);
+                                        dmsg->setType(Discovery::successor);
+                                        omsg.encapsulate(dmsg);
                                         route_item* succ_item = table->get(*table->get_successor());
                                         logging_debug("Discovery split: routing discovery message to successor "
                                                         << succ_item->id.toString() );
+<<<<<<< .working
                                         send(&cmsg_s, succ_item->info);
+=======
+                                        send(&omsg, succ_item->info);
+>>>>>>> .merge-rechts.r5869
                                 }
 
                                 // send predecessor message
                                 if (table->get_predesessor() != NULL) {
-                                        ChordMessage cmsg_p(*m);
-                                        Discovery dmsg_p(*dmsg);
-                                        dmsg_p.setFollowType(Discovery::predecessor);
-                                        cmsg_p.encapsulate(&dmsg_p);
+                                        OverlayMsg omsg(*m);
+                                        dmsg->setType(Discovery::predecessor);
+                                        omsg.encapsulate(dmsg);
                                         route_item* pred_item = table->get(
                                                         *table->get_predesessor());
                                         logging_debug("Discovery split: routing discovery message to predecessor "
                                                         << pred_item->id.toString() );
+<<<<<<< .working
                                         send(&cmsg_p, pred_item->info);
+=======
+                                        send( &omsg, pred_item->info);
+>>>>>>> .merge-rechts.r5869
                                 }
                         }
                         // no-> route message
                         else {
+<<<<<<< .working
                                 // find next hop
                                 const route_item* item = table->get_next_hop(m->getDestination());
@@ -381 +442 @@
                                                 item->id.toString() );
                                 send(m, item->info);
+=======
+                                baseoverlay.send( m, m->getDestinationNode() );
+>>>>>>> .merge-rechts.r5869
                         }
                         break;
@@ -395 +459 @@
 
                         const route_item* item = NULL;
-                        if (dmsg->getFollowType() == Discovery::successor &&
+                        if (dmsg->getType() == Discovery::successor &&
                                         table->get_successor() != NULL) {
                                 item = table->get(*table->get_successor());
@@ -405 +469 @@
                         logging_debug("routing discovery message to succ/pred "
                                 << item->id.toString() );
-                        ChordMessage cmsg(*m);
-                        Discovery dmsg_p(*dmsg);
-                        cmsg.encapsulate(&dmsg_p);
-                        send(&cmsg, item->info);
+                        OverlayMsg omsg(*m);
+                        omsg.encapsulate(dmsg);
+                        omsg.setDestinationNode(item->id);
+                        baseoverlay.send(&omsg, omsg.getDestinationNode());
                         break;
                 }}
@@ -415 +479 @@
         }
 
-                // leave
-        case M::leave: {
+        // leave
+        case typeLeave: {
                 if (link!=LinkID::UNSPECIFIED) {
                         route_item* item = table->get(remote);
@@ -424 +488 @@
                 }
                 break;
-        }
-        }
-        delete m;
+        }}
 }
 
@@ -464 +526 @@
                 // remove orphan links
                 orphan_removal_counter++;
+<<<<<<< .working
                 if (orphan_removal_counter <0 || orphan_removal_counter >= 4) {
+=======
+                if (orphan_removal_counter <0 || orphan_removal_counter >= 2) {
+>>>>>>> .merge-rechts.r5869
                         logging_info("Running orphan removal");
                         orphan_removal_counter = 0;
@@ -473 +539 @@
                                         table->insert(it->id);
                                         if (it->ref_count==0) {
-                                                baseoverlay.dropLink(it->info);
+                                                LinkID id = it->info;
                                                 it->info = LinkID::UNSPECIFIED;
+                                                baseoverlay.dropLink(id);
                                         }
                                 }
@@ -480 +547 @@
                 }
         }
+}
+
+void Chord::showLinks() {
         logging_info("--- chord routing information ----------------------------------");
         logging_info("predecessor: " << (table->get_predesessor()==NULL? "<none>" :