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

Timestamp:

Jun 19, 2013, 11:05:49 AM (11 years ago)

Author:

hock@…

Message:

Reintegrate branch: 20130111-hock-message_classes

improvements:

• new message classes (reboost, zero-copy)
• "fast path" for direct links (skip overlay layer)
• link-properties accessible from the application
• SystemQueue can call boost::bind functions
• protlib compatibility removed (32bit overhead saved in every message)
• addressing2
• AddressDiscovery discoveres only addresses on which we're actually listening
• ariba serialization usage reduced (sill used in OverlayMsg)
• Node::connect, easier and cleaner interface to start-up ariba from the application
• ariba configs via JSON, XML, etc (boost::property_tree)
• keep-alive overhead greatly reduced
• (relayed) overlay links can actually be closed now
• lost messages are detected in most cases
• notification to the application when link is transformed into direct-link
• overlay routing: send message to second best hop if it would be dropped otherwise
• SequenceNumbers (only mechanisms, so for: upward compatibility)
• various small fixes

regressions:

• bluetooth is not yet working again
• bootstrap modules deactivated
• liblog4xx is not working (use cout-logging)

This patch brings great performance and stability improvements at cost of backward compatibility.
Also bluetooth and the bootstrap modules have not been ported to the new interfaces, yet.

File:

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

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

@@ -43 +43 @@
 #include "detail/chord_routing_table.hpp"
 
-#include "messages/Discovery.h"
+//#include "messages/Discovery.h"  // XXX DEPRECATED
 
 namespace ariba {
@@ -55 +55 @@
 typedef chord_routing_table::item route_item;
 
+using ariba::transport::system_priority;
+
 use_logging_cpp( Chord );
+
+
+////// Messages
+struct DiscoveryMessage
+{
+    /**
+     * DiscoveryMessage
+     * - type
+     * - data
+     * - Endpoint
+     */
+
+    // type enum
+    enum type_ {
+        invalid = 0,
+        normal = 1,
+        successor = 2,
+        predecessor = 3
+    };
+
+    
+    // data
+    uint8_t type;
+    uint8_t ttl;
+    EndpointDescriptor endpoint;
+
+    // serialize
+    reboost::message_t serialize()
+    {
+        // serialize endpoint
+        reboost::message_t msg = endpoint.serialize();
+        
+        // serialize type and ttl
+        uint8_t* buff1 = msg.push_front(2*sizeof(uint8_t)).mutable_data();
+        buff1[0] = type;
+        buff1[1] = ttl;
+        
+        return msg;
+    }
+    
+    //deserialize
+    reboost::shared_buffer_t deserialize(reboost::shared_buffer_t buff)
+    {
+        // deserialize type and ttl
+        const uint8_t* bytes = buff.data();
+        type = bytes[0];
+        ttl = bytes[1];
+        
+        // deserialize endpoint
+        return endpoint.deserialize(buff(2*sizeof(uint8_t)));
+    }
+};
+
 
 Chord::Chord(BaseOverlay& _baseoverlay, const NodeID& _nodeid,
@@ -102 +157 @@
 
 /// helper: sends a message using the "base overlay"
-seqnum_t Chord::send( OverlayMsg* msg, const LinkID& link ) {
-        if (link.isUnspecified()) return 0;
-        return baseoverlay.send_link( msg, link );
+void Chord::send( OverlayMsg* msg, const LinkID& link ) {
+        if (link.isUnspecified())
+        return;
+    
+        baseoverlay.send_link( msg, link, system_priority::OVERLAY );
+}
+
+void Chord::send_node( OverlayMsg* message, const NodeID& remote )
+{
+    try
+    {
+        baseoverlay.send( message, remote, system_priority::OVERLAY );
+    }
+    catch ( message_not_sent& e )
+    {
+        logging_warn("Chord: Could not send message to " << remote
+                << ": " << e.what());
+    }
 }
 
@@ -116 +186 @@
         OverlayMsg msg( typeDiscovery );
         msg.setRegisterRelay(true);
-        Discovery dmsg( Discovery::normal, (uint8_t)ttl, baseoverlay.getEndpointDescriptor() );
-        msg.encapsulate(&dmsg);
+        
+        // create DiscoveryMessage
+        DiscoveryMessage dmsg;
+        dmsg.type = DiscoveryMessage::normal;
+        dmsg.ttl = ttl;
+        dmsg.endpoint = baseoverlay.getEndpointDescriptor();
+
+        msg.set_payload_message(dmsg.serialize());
 
         // send to node
-        baseoverlay.send_node( &msg, remote );
+    try
+    {
+        baseoverlay.send_node( &msg, remote, system_priority::OVERLAY );
+    }
+    catch ( message_not_sent& e )
+    {
+        logging_warn("Chord: Could not send message to " << remote
+                << ": " << e.what());
+    }
 }
 
 void Chord::discover_neighbors( const LinkID& link ) {
         uint8_t ttl = 1;
+        
+    // FIXME try-catch for the send operations
+    
+    // create DiscoveryMessage
+    DiscoveryMessage dmsg;
+    dmsg.ttl = ttl;
+    dmsg.endpoint = baseoverlay.getEndpointDescriptor();
         {
                 // send predecessor discovery
                 OverlayMsg msg( typeDiscovery );
                 msg.setRegisterRelay(true);
-                Discovery dmsg( Discovery::predecessor, ttl,
-                        baseoverlay.getEndpointDescriptor() );
-                msg.encapsulate(&dmsg);
+
+                // set type
+            dmsg.type = DiscoveryMessage::predecessor;
+
+            // send
+            msg.set_payload_message(dmsg.serialize());
                 send(&msg, link);
         }
@@ -137 +231 @@
                 // send successor discovery
                 OverlayMsg msg( typeDiscovery );
-                msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() );
+//              msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() );  // XXX this was redundand, wasn't it?
                 msg.setRegisterRelay(true);
-                Discovery dmsg( Discovery::successor, ttl,
-                        baseoverlay.getEndpointDescriptor() );
-                msg.encapsulate(&dmsg);
+
+                // set type
+        dmsg.type = DiscoveryMessage::successor;
+
+        // send
+        msg.set_payload_message(dmsg.serialize());
                 send(&msg, link);
         }
@@ -163 +260 @@
 
         // timer for stabilization management
-        Timer::setInterval(1000);
+//      Timer::setInterval(1000);  // TODO find an appropriate interval!
+        Timer::setInterval(10000);  // XXX testing...
         Timer::start();
 }
@@ -200 +298 @@
         return item->info;
 }
+
+std::vector<const LinkID*> Chord::getSortedLinkIdsTowardsNode(
+    const NodeID& id, int num ) const
+{
+    std::vector<const LinkID*> ret;
+    
+    switch ( num )
+    {
+        // special case: just call »getNextLinkId«
+        case 1:
+        {
+            ret.push_back(&getNextLinkId(id));
+            
+            break;
+        }
+        
+        // * calculate top 2 *
+        case 0:
+        case 2:
+        {
+            std::vector<const route_item*> items = table->get_next_2_hops(id);
+            
+            ret.reserve(items.size());
+            
+            BOOST_FOREACH( const route_item* item, items )
+            {
+                ret.push_back(&item->info);
+            }
+            
+            break;
+        }
+        
+        // NOTE: implement real sorting, if needed (and handle "case 0" properly, then)
+        default:
+        {
+            throw std::runtime_error("Not implemented. (Chord::getSortedLinkIdsTowardsNode with num != 2)");
+            
+            break;
+        }
+    }
+    
+    return ret;
+}
+
 
 /// @see OverlayInterface.h
@@ -253 +395 @@
         if (remote==nodeid) {
                 logging_warn("dropping link that has been established to myself (nodes have same nodeid?)");
+                logging_warn("NodeID: " << remote);
                 baseoverlay.dropLink(lnk);
                 return;
@@ -290 +433 @@
 /// @see CommunicationListener.h or @see OverlayInterface.h
 void Chord::onLinkDown(const LinkID& lnk, const NodeID& remote) {
-        logging_debug("link_down: link=" << lnk.toString() << " remote=" <<
+    // XXX logging_debug
+        logging_info("link_down (Chord): link=" << lnk.toString() << " remote=" <<
                         remote.toString() );
 
@@ -303 +447 @@
 /// @see CommunicationListener.h
 /// @see OverlayInterface.h
-void Chord::onMessage(const DataMessage& msg, const NodeID& remote,
+void Chord::onMessage(OverlayMsg* msg,
+        reboost::shared_buffer_t sub_msg,
+        const NodeID& remote,
                 const LinkID& link) {
 
-        // decode message
-        OverlayMsg* m = dynamic_cast<OverlayMsg*>(msg.getMessage());
-        if (m == NULL) return;
-
         // handle messages
-        switch ((signalMessageTypes)m->getType()) {
+        switch ((signalMessageTypes) msg->getType()) {
 
         // discovery request
-        case typeDiscovery: {
-                // decapsulate message
-                Discovery* dmsg = m->decapsulate<Discovery> ();
+        case typeDiscovery:
+        {
+                // deserialize discovery message
+                DiscoveryMessage dmsg;
+                dmsg.deserialize(sub_msg);
+                
                 logging_debug("Received discovery message with"
-                            << " src=" << m->getSourceNode().toString()
-                                << " dst=" << m->getDestinationNode().toString()
-                                << " ttl=" << (int)dmsg->getTTL()
-                                << " type=" << (int)dmsg->getType()
+                            << " src=" << msg->getSourceNode().toString()
+                                << " dst=" << msg->getDestinationNode().toString()
+                                << " ttl=" << (int)dmsg.ttl
+                                << " type=" << (int)dmsg.type
                 );
 
@@ -327 +472 @@
                 bool found = false;
                 BOOST_FOREACH( NodeID& value, discovery )
-                        if (value == m->getSourceNode()) {
+                        if (value == msg->getSourceNode()) {
                                 found = true;
                                 break;
                         }
-                if (!found) discovery.push_back(m->getSourceNode());
+                if (!found) discovery.push_back(msg->getSourceNode());
 
                 // check if source node can be added to routing table and setup link
-                if (m->getSourceNode() != nodeid)
-                        setup( dmsg->getEndpoint(), m->getSourceNode() );
+                if (msg->getSourceNode() != nodeid)
+                        setup( dmsg.endpoint, msg->getSourceNode() );
 
                 // process discovery message -------------------------- switch start --
-                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->getDestinationNode()) ) {
-                                logging_debug("Discovery split:");
-                                if (!table->get_successor()->isUnspecified()) {
-                                        OverlayMsg omsg(*m);
-                                        dmsg->setType(Discovery::successor);
-                                        omsg.encapsulate(dmsg);
-                                        logging_debug("* Routing to successor "
-                                                        << table->get_successor()->toString() );
-                                        baseoverlay.send( &omsg, *table->get_successor() );
-                                }
-
-                                // send predecessor message
-                                if (!table->get_predesessor()->isUnspecified()) {
-                                        OverlayMsg omsg(*m);
-                                        dmsg->setType(Discovery::predecessor);
-                                        omsg.encapsulate(dmsg);
-                                        logging_debug("* Routing to predecessor "
-                                                        << table->get_predesessor()->toString() );
-                                        baseoverlay.send( &omsg, *table->get_predesessor() );
-                                }
-                        }
-                        // no-> route message
-                        else {
-                                baseoverlay.route( m );
-                        }
-                        break;
-                }
-
-                // successor mode: follow the successor until TTL is zero
-                case Discovery::successor:
-                case Discovery::predecessor: {
-                        // reached destination? no->forward!
-                        if (m->getDestinationNode() != nodeid) {
-                                OverlayMsg omsg(*m);
-                                omsg.encapsulate(dmsg);
-                                omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
-                                baseoverlay.route( &omsg );
-                                break;
-                        }
-
-                        // time to live ended? yes-> stop routing
-                        if (dmsg->getTTL() == 0 || dmsg->getTTL() > 10) break;
-
-                        // decrease time-to-live
-                        dmsg->setTTL(dmsg->getTTL() - 1);
-
-                        const route_item* item = NULL;
-                        if (dmsg->getType() == Discovery::successor &&
-                                        table->get_successor() != NULL) {
-                                item = table->get(*table->get_successor());
-                        } else {
-                                if (table->get_predesessor()!=NULL)
-                                        item = table->get(*table->get_predesessor());
-                        }
-                        if (item == NULL)
-                                break;
-
-                        logging_debug("Routing discovery message to succ/pred "
-                                << item->id.toString() );
-                        OverlayMsg omsg(*m);
-                        omsg.encapsulate(dmsg);
-                        omsg.setDestinationNode(item->id);
-                        omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
-                        baseoverlay.send(&omsg, omsg.getDestinationNode());
-                        break;
-                }
-                case Discovery::invalid:
-                        break;
-
-                default:
-                        break;
-                }
-                // process discovery message ---------------------------- switch end --
-
-                delete dmsg;
-                break;
-        }
-
-        // leave
-        case typeLeave: {
-                if (link!=LinkID::UNSPECIFIED) {
-                        route_item* item = table->get(remote);
-                        if (item!=NULL) item->info = LinkID::UNSPECIFIED;
-                        table->remove(remote);
-                        baseoverlay.dropLink(link);
-                }
-                break;
-        }}
+                switch ( dmsg.type )
+                {
+            // normal: route discovery message like every other message
+            case DiscoveryMessage::normal:
+            {
+                // closest node? yes-> split to follow successor and predecessor
+                if ( table->is_closest_to(msg->getDestinationNode()) )
+                {
+                    logging_debug("Discovery split:");
+                    if (!table->get_successor()->isUnspecified())
+                    {
+                        OverlayMsg omsg(*msg);
+                        
+                        dmsg.type = DiscoveryMessage::successor;
+                        omsg.set_payload_message(dmsg.serialize());
+
+                        logging_debug("* Routing to successor "
+                                << table->get_successor()->toString() );
+                        send_node( &omsg, *table->get_successor() );
+                    }
+    
+                    // send predecessor message
+                    if (!table->get_predesessor()->isUnspecified())
+                    {
+                        OverlayMsg omsg(*msg);
+                        
+                        dmsg.type = DiscoveryMessage::predecessor;
+                        omsg.set_payload_message(dmsg.serialize());
+
+                        logging_debug("* Routing to predecessor "
+                                << table->get_predesessor()->toString() );
+                        send_node( &omsg, *table->get_predesessor() );
+                    }
+                }
+                // no-> route message
+                else
+                {
+                    baseoverlay.route( msg );
+                }
+                break;
+            }
+    
+            // successor mode: follow the successor until TTL is zero
+            case DiscoveryMessage::successor:
+            case DiscoveryMessage::predecessor:
+            {
+                // reached destination? no->forward!
+                if (msg->getDestinationNode() != nodeid)
+                {
+                    OverlayMsg omsg(*msg);
+                    omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
+                    
+                    omsg.set_payload_message(dmsg.serialize());
+                    
+                    baseoverlay.route( &omsg );
+                    break;
+                }
+    
+                // time to live ended? yes-> stop routing
+                if (dmsg.ttl == 0 || dmsg.ttl > 10) break;
+    
+                // decrease time-to-live
+                dmsg.ttl--;
+    
+                const route_item* item = NULL;
+                if (dmsg.type == DiscoveryMessage::successor &&
+                        table->get_successor() != NULL)
+                {
+                    item = table->get(*table->get_successor());
+                }
+                else if (table->get_predesessor() != NULL)
+                {
+                        item = table->get(*table->get_predesessor());
+                }
+                if (item == NULL)
+                    break;
+    
+                logging_debug("Routing discovery message to succ/pred "
+                    << item->id.toString() );
+                OverlayMsg omsg(*msg);
+                omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
+                omsg.setDestinationNode(item->id);
+                
+                omsg.set_payload_message(dmsg.serialize());
+                
+                send_node( &omsg, omsg.getDestinationNode() );
+                break;
+            }
+            case DiscoveryMessage::invalid:
+                break;
+    
+            default:
+                break;
+            }
+            // process discovery message ---------------------------- switch end --
+    
+            break;
+        }
+    
+        // leave
+        case typeLeave: {
+            if (link!=LinkID::UNSPECIFIED) {
+                route_item* item = table->get(remote);
+                if (item!=NULL) item->info = LinkID::UNSPECIFIED;
+                table->remove(remote);
+                baseoverlay.dropLink(link);
+            }
+            break;
+        }
+        }
 }