Changeset 12060 for source/ariba/overlay

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.

Location:

source/ariba/overlay

Files:

• BaseOverlay.cpp (modified) (76 diffs)
• BaseOverlay.h (modified) (17 diffs)
• CMakeLists.txt (modified) (2 diffs)
• LinkDescriptor.h (modified) (12 diffs)
• SequenceNumber.cpp (added)
• SequenceNumber.h (added)
• messages/JoinReply.cpp (modified) (2 diffs)
• messages/JoinReply.h (modified) (6 diffs)
• messages/OverlayMsg.h (modified) (17 diffs)
• modules/OverlayFactory.cpp (modified) (2 diffs)
• modules/OverlayInterface.cpp (modified) (2 diffs)
• modules/OverlayInterface.h (modified) (2 diffs)
• modules/OverlayStructureEvents.h (modified) (2 diffs)
• modules/chord/Chord.cpp (modified) (11 diffs)
• modules/chord/Chord.h (modified) (4 diffs)
• modules/chord/detail/chord_routing_table.hpp (modified) (3 diffs)
• modules/chord/messages/CMakeLists.txt (modified) (1 diff)
• modules/chord/messages/Discovery.h (modified) (1 diff)
• modules/onehop/CMakeLists.txt (modified) (1 diff)

source/ariba/overlay/BaseOverlay.cpp

@@ -57 +57 @@
 #include "ariba/utility/visual/DddVis.h"
 #include "ariba/utility/visual/ServerVis.h"
+#include <ariba/utility/misc/sha1.h>
 
 namespace ariba {
 namespace overlay {
+
+using namespace std;
+using ariba::transport::system_priority;
 
 #define visualInstance          ariba::utility::DddVis::instance()
 #define visualIdOverlay         ariba::utility::ServerVis::NETWORK_ID_BASE_OVERLAY
 #define visualIdBase            ariba::utility::ServerVis::NETWORK_ID_BASE_COMMUNICATION
+
+
+// time constants (in seconds)
+#define KEEP_ALIVE_TIME         60                      // send keep-alive message after link is not used for #s  
+
+#define LINK_ESTABLISH_TIME_OUT 10                      // timeout: link requested but not up
+#define KEEP_ALIVE_TIME_OUT     KEEP_ALIVE_TIME + LINK_ESTABLISH_TIME_OUT     // timeout: no data received on this link (incl. keep-alive messages)
+#define AUTO_LINK_TIME_OUT      KEEP_ALIVE_TIME_OUT                    // timeout: auto link not used for #s
 
 
@@ -85 +97 @@
 
 LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) {
-        BOOST_FOREACH( LinkDescriptor* lp, links )
+        foreach( LinkDescriptor* lp, links )
                                 if ((communication ? lp->communicationId : lp->overlayId) == link)
                                         return lp;
@@ -92 +104 @@
 
 const LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) const {
-        BOOST_FOREACH( const LinkDescriptor* lp, links )
+        foreach( const LinkDescriptor* lp, links )
                                 if ((communication ? lp->communicationId : lp->overlayId) == link)
                                         return lp;
@@ -122 +134 @@
 
 /// returns a auto-link descriptor
-LinkDescriptor* BaseOverlay::getAutoDescriptor( const NodeID& node, const ServiceID& service ) {
+LinkDescriptor* BaseOverlay::getAutoDescriptor( const NodeID& node, const ServiceID& service )
+{
         // search for a descriptor that is already up
-        BOOST_FOREACH( LinkDescriptor* lp, links )
-                                if (lp->autolink && lp->remoteNode == node && lp->service == service && lp->up && lp->keepAliveMissed == 0)
-                                        return lp;
+        foreach( LinkDescriptor* lp, links )
+    {
+        if (lp->autolink && lp->remoteNode == node && lp->service == service && isLinkVital(lp) )
+            return lp;
+    }
+        
         // if not found, search for one that is about to come up...
-        BOOST_FOREACH( LinkDescriptor* lp, links )
-        if (lp->autolink && lp->remoteNode == node && lp->service == service && lp->keepAliveMissed == 0 )
-                return lp;
+        foreach( LinkDescriptor* lp, links )
+        {
+            time_t now = time(NULL);
+            
+        if (lp->autolink && lp->remoteNode == node && lp->service == service
+                && difftime( now, lp->keepAliveReceived ) <= LINK_ESTABLISH_TIME_OUT )
+            return lp;
+        }
+        
         return NULL;
 }
@@ -136 +158 @@
 /// stabilizes link information
 void BaseOverlay::stabilizeLinks() {
-        // send keep-alive messages over established links
-        BOOST_FOREACH( LinkDescriptor* ld, links ) {
+    time_t now = time(NULL);
+    
+    // send keep-alive messages over established links
+        foreach( LinkDescriptor* ld, links )
+    {
                 if (!ld->up) continue;
-                OverlayMsg msg( OverlayMsg::typeLinkAlive,
-                                OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode );
-                if (ld->relayed) msg.setRouteRecord(true);
-                send_link( &msg, ld->overlayId );
+                
+                if ( difftime( now, ld->keepAliveSent ) >= KEEP_ALIVE_TIME )
+                {
+                    logging_debug("[BaseOverlay] Sending KeepAlive over "
+                            << ld->to_string()
+                            << " after "
+                            << difftime( now, ld->keepAliveSent )
+                            << "s");
+                    
+            OverlayMsg msg( OverlayMsg::typeKeepAlive,
+                    OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode );
+            msg.setRouteRecord(true);
+            ld->keepAliveSent = now;
+            send_link( &msg, ld->overlayId, system_priority::OVERLAY );
+                }
         }
 
         // iterate over all links and check for time boundaries
         vector<LinkDescriptor*> oldlinks;
-        time_t now = time(NULL);
-        BOOST_FOREACH( LinkDescriptor* ld, links ) {
-
-                // keep alives and not up? yes-> link connection request is stale!
-                if ( !ld->up && difftime( now, ld->keepAliveTime ) >= 2 ) {
-
-                        // increase counter
-                        ld->keepAliveMissed++;
-
-                        // missed more than four keep-alive messages (10 sec)? -> drop link
-                        if (ld->keepAliveMissed > 4) {
-                                logging_info( "Link connection request is stale, closing: " << ld );
-                                oldlinks.push_back( ld );
-                                continue;
-                        }
+        foreach( LinkDescriptor* ld, links ) {
+
+                // link connection request stale?
+                if ( !ld->up && difftime( now, ld->keepAliveReceived ) >= LINK_ESTABLISH_TIME_OUT )  // NOTE: keepAliveReceived == now, on connection request
+                {
+            logging_info( "Link connection request is stale, closing: " << ld );
+            ld->failed = true;
+            oldlinks.push_back( ld );
+            continue;
                 }
 
                 if (!ld->up) continue;
 
+                
+                
+                
                 // check if link is relayed and retry connecting directly
+                // TODO Mario: What happens here?  --> There are 3 attempts to replace a relayed link with a direct one. see: handleLinkReply
                 if ( ld->relayed && !ld->communicationUp && ld->retryCounter > 0) {
                         ld->retryCounter--;
@@ -173 +207 @@
 
                 // remote used as relay flag
-                if ( ld->relaying && difftime( now, ld->timeRelaying ) > 10)
+                if ( ld->relaying && difftime( now, ld->timeRelaying ) > KEEP_ALIVE_TIME_OUT)  // TODO is this a reasonable timeout ??
                         ld->relaying = false;
 
@@ -183 +217 @@
 
                 // auto-link time exceeded?
-                if ( ld->autolink && difftime( now, ld->lastuse ) > 30 ) {
+                if ( ld->autolink && difftime( now, ld->lastuse ) > AUTO_LINK_TIME_OUT ) {
                         oldlinks.push_back( ld );
                         continue;
@@ -189 +223 @@
 
                 // keep alives missed? yes->
-                if ( difftime( now, ld->keepAliveTime ) > 4 ) {
-
-                        // increase counter
-                        ld->keepAliveMissed++;
-
-                        // missed more than four keep-alive messages (4 sec)? -> drop link
-                        if (ld->keepAliveMissed >= 2) {
-                                logging_info( "Link is stale, closing: " << ld );
-                                oldlinks.push_back( ld );
-                                continue;
-                        }
+                if ( difftime( now, ld->keepAliveReceived ) >= KEEP_ALIVE_TIME_OUT )
+                {
+            logging_info( "Link is stale, closing: " << ld );
+            ld->failed = true;
+            oldlinks.push_back( ld );
+            continue;
                 }
         }
 
         // drop links
-        BOOST_FOREACH( LinkDescriptor* ld, oldlinks ) {
+        foreach( LinkDescriptor* ld, oldlinks ) {
                 logging_info( "Link timed out. Dropping " << ld );
                 ld->relaying = false;
@@ -210 +239 @@
         }
 
-        // show link state
-        counter++;
-        if (counter>=4) showLinks();
-        if (counter>=4 || counter<0) counter = 0;
+        
+        
+        
+        // show link state  (debug output)
+        if (counter>=10 || counter<0)
+    {
+            showLinks();
+            counter = 0;
+    }
+        else
+        {
+            counter++;
+        }
 }
 
@@ -230 +268 @@
 
                 int i=0;
-                BOOST_FOREACH( LinkDescriptor* ld, links ) {
-                        if (!ld->isVital() || ld->service != OverlayInterface::OVERLAY_SERVICE_ID) continue;
+                foreach( LinkDescriptor* ld, links ) {
+                        if (!isLinkVital(ld) || ld->service != OverlayInterface::OVERLAY_SERVICE_ID) continue;
                         bool found = false;
-                        BOOST_FOREACH(NodeID& id, nodes)
+                        foreach(NodeID& id, nodes)
                         if (id  == ld->remoteNode) found = true;
                         if (found) continue;
@@ -261 +299 @@
         int i=0;
         logging_info("--- link state -------------------------------");
-        BOOST_FOREACH( LinkDescriptor* ld, links ) {
+        foreach( LinkDescriptor* ld, links ) {
                 string epd = "";
-                if (ld->isDirectVital())
-                        epd = getEndpointDescriptor(ld->remoteNode).toString();
+                if (isLinkDirectVital(ld))
+                {
+//                      epd = getEndpointDescriptor(ld->remoteNode).toString();
+                    
+                    epd = "Connection: ";
+                    epd += bc->get_local_endpoint_of_link(ld->communicationId)->to_string();
+                    epd += " <---> ";
+                    epd += bc->get_remote_endpoint_of_link(ld->communicationId)->to_string();
+                }
 
                 logging_info("LINK_STATE: " << i << ": " << ld << " " << epd);
@@ -289 +334 @@
 // internal message delivery ---------------------------------------------------
 
+
+seqnum_t BaseOverlay::send_overlaymessage_down( OverlayMsg* message, const LinkID& bc_link, uint8_t priority )
+{
+    // set priority
+    message->setPriority(priority);
+    
+    // wrap old OverlayMsg into reboost message
+    reboost::message_t wrapped_message = message->wrap_up_for_sending();
+    
+    // send down to BaseCommunication
+    try
+    {
+        // * send *
+        return bc->sendMessage(bc_link, wrapped_message, priority, false);
+    }
+    catch ( communication::communication_message_not_sent& e )
+    {
+        ostringstream out;
+        out << "Communication message not sent: " << e.what();
+        throw message_not_sent(out.str());
+    }
+    
+    throw logic_error("This should never happen!");
+}
+
+
 /// routes a message to its destination node
-void BaseOverlay::route( OverlayMsg* message ) {
-
+void BaseOverlay::route( OverlayMsg* message, const NodeID& last_hop )
+{
         // exceeded time-to-live? yes-> drop message
-        if (message->getNumHops() > message->getTimeToLive()) {
-                logging_warn("Message exceeded TTL. Dropping message and relay routes"
-                                "for recovery.");
+        if (message->getNumHops() > message->getTimeToLive())
+        {
+                logging_warn("Message exceeded TTL. Dropping message and relay routes "
+            << "for recovery. Hop count: " << (int) message->getNumHops());
                 removeRelayNode(message->getDestinationNode());
                 return;
@@ -301 +373 @@
 
         // no-> forward message
-        else {
+        else
+        {
                 // destinastion myself? yes-> handle message
-                if (message->getDestinationNode() == nodeId) {
-                        logging_warn("Usually I should not route messages to myself!");
-                        Message msg;
-                        msg.encapsulate(message);
-                        handleMessage( &msg, NULL );
-                } else {
-                        // no->send message to next hop
-                        send( message, message->getDestinationNode() );
-                }
-        }
+                if (message->getDestinationNode() == nodeId)
+                {
+                        logging_warn("Usually I should not route messages to myself. And I won't!");
+                }
+
+                // no->send message to next hop
+                else
+                {
+                    try
+                    {
+                /* (deep) packet inspection to determine priority */
+                // BRANCH: typeData  -->  send with low priority
+                if ( message->getType() == OverlayMsg::typeData )
+                {
+                    // TODO think about implementing explicit routing queue (with active queue management??)
+                    send( message,
+                          message->getDestinationNode(),
+                          message->getPriority(),
+                          last_hop );
+                }
+                // BRANCH: internal message  -->  send with higher priority
+                else
+                {
+                    send( message,
+                          message->getDestinationNode(),
+                          system_priority::HIGH,
+                          last_hop );
+                }
+                    }
+                    catch ( message_not_sent& e )
+                    {
+                        logging_warn("Unable to route message of type "
+                                << message->getType()
+                                << " to "
+                                << message->getDestinationNode()
+                                << ". Reason: "
+                                << e.what());
+                        
+                        // inform sender
+                if ( message->getType() != OverlayMsg::typeMessageLost )
+                {
+                    report_lost_message(message);
+                }
+                    }
+                }
+        }
+}
+
+void BaseOverlay::report_lost_message( const OverlayMsg* message )
+{
+    OverlayMsg reply(OverlayMsg::typeMessageLost);
+    reply.setSeqNum(message->getSeqNum());
+    
+    /**
+     * MessageLost-Message
+     * 
+     * - Type of lost message
+     * - Hop count of lost message
+     * - Source-LinkID  of lost message
+     */
+    reboost::shared_buffer_t b(sizeof(uint8_t)*2);
+    b.mutable_data()[0] = message->getType();
+    b.mutable_data()[1] = message->getNumHops();
+    reply.append_buffer(b);
+    reply.append_buffer(message->getSourceLink().serialize());
+    
+    try
+    {
+        send_node(&reply, message->getSourceNode(),
+                system_priority::OVERLAY,
+                OverlayInterface::OVERLAY_SERVICE_ID);
+    }
+    catch ( message_not_sent& e )
+    {
+        logging_warn("Tried to inform another node that we could'n route their message. But we were not able to send this error-message, too.");
+    }
 }
 
 /// sends a message to another node, delivers it to the base overlay class
-seqnum_t BaseOverlay::send( OverlayMsg* message, const NodeID& destination ) {
+seqnum_t BaseOverlay::send( OverlayMsg* message,
+        const NodeID& destination,
+        uint8_t priority,
+        const NodeID& last_hop ) throw(message_not_sent)
+{
         LinkDescriptor* next_link = NULL;
 
         // drop messages to unspecified destinations
-        if (destination.isUnspecified()) return -1;
-
-        // send messages to myself -> handle message and drop warning!
-        if (destination == nodeId) {
-                logging_warn("Sent message to myself. Handling message.")
-                Message msg;
-                msg.encapsulate(message);
-                handleMessage( &msg, NULL );
-                return -1;
+        if (destination.isUnspecified())
+            throw message_not_sent("No destination specified. Drop!");
+
+        // send messages to myself -> drop!
+    // TODO maybe this is not what we want. why not just deliver this message?
+    //   There is a similar check in the route function, there it should be okay.
+        if (destination == nodeId)
+        {
+            logging_warn("Sent message to myself. Drop!");
+            
+            throw message_not_sent("Sent message to myself. Drop!");
         }
 
         // use relay path?
-        if (message->isRelayed()) {
+        if (message->isRelayed())
+        {
                 next_link = getRelayLinkTo( destination );
-                if (next_link != NULL) {
+                
+                if (next_link != NULL)
+                {
                         next_link->setRelaying();
-                        return bc->sendMessage(next_link->communicationId, message);
-                } else {
-                        logging_warn("Could not send message. No relay hop found to "
-                                        << destination << " -- trying to route over overlay paths ...")
-//                      logging_error("ERROR: " << debugInformation() );
-                //                      return -1;
-                }
-        }
-
+
+                        // * send message over relayed link *
+                        return send_overlaymessage_down(message, next_link->communicationId, priority);
+                }
+                else
+                {
+                        logging_warn("No relay hop found to " << destination
+                                << " -- trying to route over overlay paths ...")
+                }
+        }
+
+        
         // last resort -> route over overlay path
         LinkID next_id = overlayInterface->getNextLinkId( destination );
-        if (next_id.isUnspecified()) {
-                logging_warn("Could not send message. No next hop found to " <<
-                                destination );
-                logging_error("ERROR: " << debugInformation() );
-                return -1;
-        }
-
-        // get link descriptor, up and running? yes-> send message
+        if ( next_id.isUnspecified() )
+        {               
+        // apperently we're the closest node --> try second best node
+        //   NOTE: This is helpful if our routing table is not up-to-date, but 
+        //   may lead to circles. So we have to be careful.
+        std::vector<const LinkID*> next_ids = 
+            overlayInterface->getSortedLinkIdsTowardsNode( destination );
+            
+        for ( int i = 0; i < next_ids.size(); i++ )
+        {
+            const LinkID& link = *next_ids[i];
+            
+            if ( ! link.isUnspecified() )
+            {
+                next_id = link;
+                
+                break;
+            }
+        }
+     
+        // still no next hop found. drop.
+        if ( next_id.isUnspecified() )
+        {
+            logging_warn("Could not send message. No next hop found to " <<
+                destination );
+            logging_error("ERROR: " << debugInformation() );
+            
+            throw message_not_sent("No next hop found.");
+        }
+        }
+
+        
+        /* get link descriptor, do some checks and send message */
         next_link = getDescriptor(next_id);
-        if (next_link != NULL && next_link->up) {
-                // send message over relayed link
-                return send(message, next_link);
-        }
-
-        // no-> error, dropping message
-        else {
-                logging_warn("Could not send message. Link not known or up");
-                logging_error("ERROR: " << debugInformation() );
-                return -1;
-        }
-
-        // not reached-> fail
-        return -1;
-}
+    
+    // check pointer
+    if ( next_link == NULL )
+    {
+        // NOTE: this shuldn't happen
+        throw message_not_sent("Could not send message. Link not known.");
+    }
+    
+    // avoid circles
+    if ( next_link->remoteNode == last_hop )
+    {
+        // XXX logging_debug
+        logging_info("Possible next hop would create a circle: "
+            << next_link->remoteNode);
+        
+        throw message_not_sent("Could not send message. Possible next hop would create a circle.");
+    }
+    
+    // check if link is up
+        if ( ! next_link->up)
+        {
+        logging_warn("Could not send message. Link not up");
+        logging_error("ERROR: " << debugInformation() );
+        
+        throw message_not_sent("Could not send message. Link not up");
+        }
+
+        // * send message over overlay link *
+        return send(message, next_link, priority);
+}
+
 
 /// send a message using a link descriptor, delivers it to the base overlay class
-seqnum_t BaseOverlay::send( OverlayMsg* message, LinkDescriptor* ldr, bool ignore_down ) {
+seqnum_t BaseOverlay::send( OverlayMsg* message,
+        LinkDescriptor* ldr,
+        uint8_t priority ) throw(message_not_sent)
+{
         // check if null
-        if (ldr == NULL) {
-                logging_error("Can not send message to " << message->getDestinationAddress());
-                return -1;
+        if (ldr == NULL)
+        {
+        ostringstream out;
+        out << "Can not send message to " << message->getDestinationAddress();
+        throw message_not_sent(out.str());
         }
 
         // check if up
-        if (!ldr->up && !ignore_down) {
-                logging_error("Can not send message. Link not up:" << ldr );
+        if ( !ldr->up )
+        {
                 logging_error("DEBUG_INFO: " << debugInformation() );
-                return -1;
-        }
-        LinkDescriptor* ld = NULL;
-
-        // handle relayed link
-        if (ldr->relayed) {
+
+        ostringstream out;
+        out << "Can not send message. Link not up:" << ldr->to_string();
+        throw message_not_sent(out.str());
+        }
+        
+        LinkDescriptor* next_hop_ld = NULL;
+
+        // BRANCH: relayed link
+        if (ldr->relayed)
+        {
                 logging_debug("Resolving direct link for relayed link to "
                                 << ldr->remoteNode);
-                ld = getRelayLinkTo( ldr->remoteNode );
-                if (ld==NULL) {
-                        logging_error("No relay path found to link " << ldr );
+                
+                next_hop_ld = getRelayLinkTo( ldr->remoteNode );
+                
+                if (next_hop_ld==NULL)
+                {
                         logging_error("DEBUG_INFO: " << debugInformation() );
-                        return -1;
-                }
-                ld->setRelaying();
+                        
+                ostringstream out;
+                out << "No relay path found to link: " << ldr;
+                throw message_not_sent(out.str());
+                }
+                
+                next_hop_ld->setRelaying();
                 message->setRelayed(true);
-        } else
-                ld = ldr;
-
-        // handle direct link
-        if (ld->communicationUp) {
-                logging_debug("send(): Sending message over direct link.");
-                return bc->sendMessage( ld->communicationId, message );
-        } else {
-                logging_error("send(): Could not send message. "
-                                "Not a relayed link and direct link is not up.");
-                return -1;
-        }
-        return -1;
+        }
+        // BRANCH: direct link
+        else
+        {
+                next_hop_ld = ldr;
+        }
+
+        
+        // check next hop-link
+        if ( ! next_hop_ld->communicationUp)
+        {
+            throw message_not_sent( "send(): Could not send message."
+                    " Not a relayed link and direct link is not up." );
+        }
+
+        // send over next link
+    logging_debug("send(): Sending message over direct link.");
+    return send_overlaymessage_down(message, next_hop_ld->communicationId, priority);
+
 }
 
 seqnum_t BaseOverlay::send_node( OverlayMsg* message, const NodeID& remote,
-                const ServiceID& service) {
+        uint8_t priority, const ServiceID& service) throw(message_not_sent)
+{
         message->setSourceNode(nodeId);
         message->setDestinationNode(remote);
         message->setService(service);
-        return send( message, remote );
-}
-
-seqnum_t BaseOverlay::send_link( OverlayMsg* message, const LinkID& link,bool ignore_down ) {
+        return send( message, remote, priority );
+}
+
+void BaseOverlay::send_link( OverlayMsg* message,
+        const LinkID& link,
+        uint8_t priority ) throw(message_not_sent)
+{
         LinkDescriptor* ld = getDescriptor(link);
-        if (ld==NULL) {
-                logging_error("Cannot find descriptor to link id=" << link.toString());
-                return -1;
-        }
+        if (ld==NULL)
+        {
+            throw message_not_sent("Cannot find descriptor to link id=" + link.toString());
+        }
+        
         message->setSourceNode(nodeId);
         message->setDestinationNode(ld->remoteNode);
@@ -437 +656 @@
         message->setService(ld->service);
         message->setRelayed(ld->relayed);
-        return send( message, ld, ignore_down );
+    
+    
+    try
+    {
+        // * send message *
+        send( message, ld, priority );
+    }
+    catch ( message_not_sent& e )
+    {
+        // drop failed link
+        ld->failed = true;
+        dropLink(ld->overlayId);
+    }
 }
 
@@ -451 +682 @@
                 // relay link still used and alive?
                 if (ld==NULL
-                                || !ld->isDirectVital()
-                                || difftime(route.used, time(NULL)) > 8) {
+                                || !isLinkDirectVital(ld)
+                                || difftime(route.used, time(NULL)) > KEEP_ALIVE_TIME_OUT)  // TODO this was set to 8 before.. Is the new timeout better?
+                {
                         logging_info("Forgetting relay information to node "
                                         << route.node.toString() );
@@ -488 +720 @@
         if (message->isRelayed()) {
                 // try to find source node
-                BOOST_FOREACH( relay_route& route, relay_routes ) {
+                foreach( relay_route& route, relay_routes ) {
                         // relay route found? yes->
                         if ( route.node == message->getDestinationNode() ) {
@@ -504 +736 @@
 
                 // try to find source node
-                BOOST_FOREACH( relay_route& route, relay_routes ) {
+                foreach( relay_route& route, relay_routes ) {
 
                         // relay route found? yes->
@@ -516 +748 @@
                                 if (route.hops > message->getNumHops()
                                                 || rld == NULL
-                                                || !rld->isDirectVital()) {
+                                                || !isLinkDirectVital(ld)) {
                                         logging_info("Updating relay information to node "
                                                         << route.node.toString()
-                                                        << " reducing to " << message->getNumHops() << " hops.");
+                                                        << " reducing to " << (int) message->getNumHops() << " hops.");
                                         route.hops = message->getNumHops();
                                         route.link = ld->overlayId;
@@ -542 +774 @@
 LinkDescriptor* BaseOverlay::getRelayLinkTo( const NodeID& remote ) {
         // try to find source node
-        BOOST_FOREACH( relay_route& route, relay_routes ) {
+        foreach( relay_route& route, relay_routes ) {
                 if (route.node == remote ) {
                         LinkDescriptor* ld = getDescriptor( route.link );
-                        if (ld==NULL || !ld->isDirectVital()) return NULL; else {
+                        if (ld==NULL || !isLinkDirectVital(ld)) return NULL; else {
                                 route.used = time(NULL);
                                 return ld;
@@ -575 +807 @@
 // ----------------------------------------------------------------------------
 
-void BaseOverlay::start( BaseCommunication& _basecomm, const NodeID& _nodeid ) {
+void BaseOverlay::start( BaseCommunication* _basecomm, const NodeID& _nodeid ) {
         logging_info("Starting...");
 
         // set parameters
-        bc = &_basecomm;
+        bc = _basecomm;
         nodeId = _nodeid;
 
@@ -587 +819 @@
 
         // timer for auto link management
-        Timer::setInterval( 1000 );
+        Timer::setInterval( 1000 ); // XXX
+//      Timer::setInterval( 10000 );
         Timer::start();
 
@@ -641 +874 @@
                 overlayInterface->joinOverlay();
                 state = BaseOverlayStateCompleted;
-                BOOST_FOREACH( NodeListener* i, nodeListeners )
+                foreach( NodeListener* i, nodeListeners )
                         i->onJoinCompleted( spovnetId );
 
@@ -682 +915 @@
         // gather all service links
         vector<LinkID> servicelinks;
-        BOOST_FOREACH( LinkDescriptor* ld, links ) {
+        foreach( LinkDescriptor* ld, links )
+        {
                 if( ld->service != OverlayInterface::OVERLAY_SERVICE_ID )
                         servicelinks.push_back( ld->overlayId );
@@ -688 +922 @@
 
         // drop all service links
-        BOOST_FOREACH( LinkID lnk, servicelinks )
-        dropLink( lnk );
+        foreach( LinkID lnk, servicelinks )
+        {
+            logging_debug("Dropping service link " << lnk.toString());
+            dropLink( lnk );
+        }
 
         // let the node leave the spovnet overlay interface
         logging_debug( "Leaving overlay" );
         if( overlayInterface != NULL )
+        {
                 overlayInterface->leaveOverlay();
+        }
 
         // drop still open bootstrap links
-        BOOST_FOREACH( LinkID lnk, bootstrapLinks )
-        bc->dropLink( lnk );
+        foreach( LinkID lnk, bootstrapLinks )
+        {
+            logging_debug("Dropping bootstrap link " << lnk.toString());
+            bc->dropLink( lnk );
+        }
 
         // change to inalid state
@@ -708 +950 @@
 
         // inform all registered services of the event
-        BOOST_FOREACH( NodeListener* i, nodeListeners ) {
+        foreach( NodeListener* i, nodeListeners )
+        {
                 if( ret ) i->onLeaveCompleted( spovnetId );
                 else i->onLeaveFailed( spovnetId );
@@ -731 +974 @@
                 state = BaseOverlayStateInvalid;
 
-                BOOST_FOREACH( NodeListener* i, nodeListeners )
+                foreach( NodeListener* i, nodeListeners )
                 i->onJoinFailed( spovnetId );
 
@@ -783 +1026 @@
                 const ServiceID& service ) {
 
+    // TODO What if we already have a Link to this node and this service id?
+    
         // do not establish a link to myself!
-        if (remote == nodeId) return LinkID::UNSPECIFIED;
-
+        if (remote == nodeId) return 
+                LinkID::UNSPECIFIED;
+
+        
         // create a link descriptor
         LinkDescriptor* ld = addDescriptor();
@@ -792 +1039 @@
         ld->service = service;
         ld->listener = getListener(ld->service);
+    
+    // initialize sequence numbers
+    ld->last_sent_seqnum = SequenceNumber::createRandomSeqNum_Short();
+    logging_debug("Creating new link with initial SeqNum: " << ld->last_sent_seqnum);
 
         // create link request message
@@ -800 +1051 @@
         msg.setRelayed(true);
         msg.setRegisterRelay(true);
+//      msg.setRouteRecord(true);
+    
+    msg.setSeqNum(ld->last_sent_seqnum);
 
         // debug message
@@ -809 +1063 @@
         );
 
+        
         // sending message to node
-        send_node( &msg, ld->remoteNode, ld->service );
-
+        try
+        {
+            // * send *
+            seqnum_t seq = send_node( &msg, ld->remoteNode, system_priority::OVERLAY, ld->service );
+        }
+        catch ( message_not_sent& e )
+        {
+            logging_warn("Link request not sent: " << e.what());
+            
+            // Message not sent. Cancel link request.
+            SystemQueue::instance().scheduleCall(
+                    boost::bind(
+                            &BaseOverlay::__onLinkEstablishmentFailed,
+                            this,
+                            ld->overlayId)
+                );
+        }
+        
         return ld->overlayId;
 }
 
+/// NOTE: "id" is an Overlay-LinkID
+void BaseOverlay::__onLinkEstablishmentFailed(const LinkID& id)
+{
+    // TODO This code redundant. But also it's not easy to aggregate in one function.
+    
+    // get descriptor for link
+    LinkDescriptor* ld = getDescriptor(id, false);
+    if ( ld == NULL ) return; // not found? ->ignore!
+
+    logging_debug( "__onLinkEstablishmentFaild: " << ld );
+
+    // removing relay link information
+    removeRelayLink(ld->overlayId);
+
+    // inform listeners about link down
+    ld->communicationUp = false;
+    if (!ld->service.isUnspecified())
+    {
+        CommunicationListener* lst = getListener(ld->service);
+        if(lst != NULL) lst->onLinkFail( ld->overlayId, ld->remoteNode );
+        sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId );
+    }
+
+    // delete all queued messages (auto links)
+    if( ld->messageQueue.size() > 0 ) {
+        logging_warn( "Dropping link " << id.toString() << " that has "
+                << ld->messageQueue.size() << " waiting messages" );
+        ld->flushQueue();
+    }
+
+    // erase mapping
+    eraseDescriptor(ld->overlayId);
+}
+
+
 /// drops an established link
-void BaseOverlay::dropLink(const LinkID& link) {
-        logging_info( "Dropping link (initiated locally):" << link.toString() );
+void BaseOverlay::dropLink(const LinkID& link)
+{
+        logging_info( "Dropping link: " << link.toString() );
 
         // find the link item to drop
         LinkDescriptor* ld = getDescriptor(link);
-        if( ld == NULL ) {
+        if( ld == NULL )
+        {
                 logging_warn( "Can't drop link, link is unknown!");
                 return;
@@ -827 +1135 @@
 
         // delete all queued messages
-        if( ld->messageQueue.size() > 0 ) {
+        if( ld->messageQueue.size() > 0 )
+        {
                 logging_warn( "Dropping link " << ld->overlayId.toString() << " that has "
                                 << ld->messageQueue.size() << " waiting messages" );
                 ld->flushQueue();
         }
-
-        // inform sideport and listener
-        if(ld->listener != NULL)
-                ld->listener->onLinkDown( ld->overlayId, ld->remoteNode );
-        sideport->onLinkDown(ld->overlayId, this->nodeId, ld->remoteNode, this->spovnetId );
-
-        // do not drop relay links
-        if (!ld->relaying) {
-                // drop the link in base communication
-                if (ld->communicationUp) bc->dropLink( ld->communicationId );
-
-                // erase descriptor
-                eraseDescriptor( ld->overlayId );
-        } else {
-                ld->dropAfterRelaying = true;
-        }
+        
+            
+        // inform application and remote note (but only once)
+        //   NOTE: If we initiated the drop, this function is called twice, but on 
+        //   the second call, there is noting to do.
+        if ( ld->up && ! ld->failed )
+        {
+        // inform sideport and listener
+        if(ld->listener != NULL)
+        {
+            ld->listener->onLinkDown( ld->overlayId, ld->remoteNode );
+        }
+        sideport->onLinkDown(ld->overlayId, this->nodeId, ld->remoteNode, this->spovnetId );
+        
+        // send link-close to remote node
+        logging_info("Sending LinkClose message to remote node.");
+        OverlayMsg close_msg(OverlayMsg::typeLinkClose);
+        send_link(&close_msg, link, system_priority::OVERLAY);
+    
+        // deactivate link
+        ld->up = false;
+//         ld->closing = true;
+        }
+        
+        else if ( ld->failed )
+    {
+        // inform listener
+        if( ld->listener != NULL )
+        {
+            ld->listener->onLinkFail( ld->overlayId, ld->remoteNode );
+        }
+        
+        ld->up = false;
+        __removeDroppedLink(ld->overlayId);
+    }
+}
+
+/// called from typeLinkClose-handler
+void BaseOverlay::__removeDroppedLink(const LinkID& link)
+{
+    // find the link item to drop
+    LinkDescriptor* ld = getDescriptor(link);
+    if( ld == NULL )
+    {
+        return;
+    }
+
+    // do not drop relay links
+    if (!ld->relaying)
+    {
+        // drop the link in base communication
+        if (ld->communicationUp)
+        {
+            bc->dropLink( ld->communicationId );
+        }
+
+        // erase descriptor
+        eraseDescriptor( ld->overlayId );
+    }
+    else
+    {
+        ld->dropAfterRelaying = true;
+    }
 }
 
@@ -853 +1209 @@
 
 /// internal send message, always use this functions to send messages over links
-seqnum_t BaseOverlay::sendMessage( const Message* message, const LinkID& link ) {
+const SequenceNumber& BaseOverlay::sendMessage( reboost::message_t message,
+        const LinkID& link,
+        uint8_t priority ) throw(message_not_sent)
+{
         logging_debug( "Sending data message on link " << link.toString() );
 
         // get the mapping for this link
         LinkDescriptor* ld = getDescriptor(link);
-        if( ld == NULL ) {
-                logging_error("Could not send message. "
-                                << "Link not found id=" << link.toString());
-                return -1;
+        if( ld == NULL )
+        {
+            throw message_not_sent("Could not send message. Link not found id=" + link.toString());
         }
 
         // check if the link is up yet, if its an auto link queue message
-        if( !ld->up ) {
+        if( !ld->up )
+        {
                 ld->setAutoUsed();
-                if( ld->autolink ) {
+                if( ld->autolink )
+                {
                         logging_info("Auto-link " << link.toString() << " not up, queue message");
-                        Data data = data_serialize( message );
-                        const_cast<Message*>(message)->dropPayload();
-                        ld->messageQueue.push_back( new Message(data) );
-                } else {
-                        logging_error("Link " << link.toString() << " not up, drop message");
-                }
-                return -1;
-        }
-
-        // compile overlay message (has service and node id)
-        OverlayMsg overmsg( OverlayMsg::typeData );
-        overmsg.encapsulate( const_cast<Message*>(message) );
-
-        // send message over relay/direct/overlay
-        return send_link( &overmsg, ld->overlayId );
-}
-
-
-seqnum_t BaseOverlay::sendMessage(const Message* message,
-                const NodeID& node, const ServiceID& service) {
+                        
+                        // queue message
+                LinkDescriptor::message_queue_entry msg;
+                msg.message = message;
+                msg.priority = priority;
+
+                        ld->messageQueue.push_back( msg );
+                        
+                        return SequenceNumber::DISABLED;  // TODO what to return if message is queued?
+                }
+                else
+                {
+                    throw message_not_sent("Link " + link.toString() + " not up, drop message");
+                }
+        }
+        
+        // TODO AKTUELL: sequence numbers
+        // TODO seqnum on fast path ?
+        ld->last_sent_seqnum.increment();
+        
+        /* choose fast-path for direct links; normal overlay-path otherwise */
+        // BRANCH: direct link
+        if ( ld->communicationUp && !ld->relayed )
+        {
+            // * send down to BaseCommunication *
+            try
+            {
+                bc->sendMessage(ld->communicationId, message, priority, true);
+        }
+        catch ( communication::communication_message_not_sent& e )
+        {
+            ostringstream out;
+            out << "Communication message on fast-path not sent: " << e.what();
+            throw message_not_sent(out.str());
+        }
+        }
+
+        // BRANCH: use (slow) overlay-path
+        else
+        {
+        // compile overlay message (has service and node id)
+        OverlayMsg overmsg( OverlayMsg::typeData );
+        overmsg.set_payload_message(message);
+        
+        // set SeqNum
+        if ( ld->transmit_seqnums )
+        {
+            overmsg.setSeqNum(ld->last_sent_seqnum);
+        }
+        logging_debug("Sending Message with SeqNum: " << overmsg.getSeqNum());
+    
+        // send message over relay/direct/overlay
+        send_link( &overmsg, ld->overlayId, priority );
+        }
+        
+        // return seqnum
+        return ld->last_sent_seqnum;
+}
+
+
+const SequenceNumber& BaseOverlay::sendMessage(reboost::message_t message,
+                const NodeID& node, uint8_t priority, const ServiceID& service) {
 
         // find link for node and service
@@ -907 +1309 @@
                 if( ld == NULL ) {
                         logging_error( "Failed to establish auto-link.");
-                        return -1;
+            throw message_not_sent("Failed to establish auto-link.");
                 }
                 ld->autolink = true;
@@ -920 +1322 @@
 
         // send / queue message
-        return sendMessage( message, ld->overlayId );
-}
-
-
-NodeID BaseOverlay::sendMessageCloserToNodeID(const Message* message,
-        const NodeID& address, const ServiceID& service) {
+        return sendMessage( message, ld->overlayId, priority );
+}
+
+
+NodeID BaseOverlay::sendMessageCloserToNodeID(reboost::message_t message,
+        const NodeID& address, uint8_t priority, const ServiceID& service) {
     
     if ( overlayInterface->isClosestNodeTo(address) )
@@ -936 +1338 @@
     if ( closest_node != NodeID::UNSPECIFIED )
     {
-        seqnum_t seqnum = sendMessage(message, closest_node, service);
+        sendMessage(message, closest_node, priority, service);
     }
     
-    return closest_node;  // XXX return seqnum ?? tuple? closest_node via (non const) reference?
+    return closest_node;  // return seqnum ?? tuple? closest_node via (non const) reference?
 }
 // ----------------------------------------------------------------------------
@@ -978 +1380 @@
 
         // see if we can find the node in our own table
-        BOOST_FOREACH(const LinkDescriptor* ld, links){
+        foreach(const LinkDescriptor* ld, links){
                 if(ld->remoteNode != node) continue;
                 if(!ld->communicationUp) continue;
@@ -1079 +1481 @@
 
 void BaseOverlay::onLinkUp(const LinkID& id,
-                const address_v* local, const address_v* remote) {
+        const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote)
+{
         logging_debug( "Link up with base communication link id=" << id );
 
@@ -1085 +1488 @@
         LinkDescriptor* ld = getDescriptor(id, true);
 
-        // handle bootstrap link we initiated
+        // BRANCH: handle bootstrap link we initiated
         if( std::find(bootstrapLinks.begin(), bootstrapLinks.end(), id) != bootstrapLinks.end() ){
                 logging_info(
                                 "Join has been initiated by me and the link is now up. " <<
+                                "LinkID: " << id.toString() <<
                                 "Sending out join request for SpoVNet " << spovnetId.toString()
                 );
@@ -1096 +1500 @@
                                 OverlayInterface::OVERLAY_SERVICE_ID, nodeId );
                 JoinRequest joinRequest( spovnetId, nodeId );
-                overlayMsg.encapsulate( &joinRequest );
-                bc->sendMessage( id, &overlayMsg );
+                overlayMsg.append_buffer(joinRequest.serialize_into_shared_buffer());
+
+                send_overlaymessage_down(&overlayMsg, id, system_priority::OVERLAY);
+                
                 return;
         }
 
-        // no link found? -> link establishment from remote, add one!
+        // BRANCH: link establishment from remote, add one!
         if (ld == NULL) {
                 ld = addDescriptor( id );
@@ -1115 +1521 @@
                 // in this case, do not inform listener, since service it unknown
                 // -> wait for update message!
-
-                // link mapping found? -> send update message with node-id and service id
-        } else {
+        }
+        
+        // BRANCH: We requested this link in the first place
+        else
+        {
                 logging_info( "onLinkUp descriptor (initiated locally):" << ld );
 
@@ -1126 +1534 @@
                 ld->fromRemote = false;
 
-                // if link is a relayed link->convert to direct link
-                if (ld->relayed) {
-                        logging_info( "Converting to direct link: " << ld );
+                // BRANCH: this was a relayed link before --> convert to direct link
+                //   TODO do we really have to send a message here?
+                if (ld->relayed)
+                {
                         ld->up = true;
                         ld->relayed = false;
+                        logging_info( "Converting to direct link: " << ld );
+                        
+                        // send message
                         OverlayMsg overMsg( OverlayMsg::typeLinkDirect );
                         overMsg.setSourceLink( ld->overlayId );
                         overMsg.setDestinationLink( ld->remoteLink );
-                        send_link( &overMsg, ld->overlayId );
-                } else {
+                        send_link( &overMsg, ld->overlayId, system_priority::OVERLAY );
+                        
+                    // inform listener
+                    if( ld->listener != NULL)
+                        ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode );
+                }
+                
+
+        /* NOTE: Chord is opening direct-links in it's setup routine which are
+         *   neither set to "relayed" nor to "up". To activate these links a
+         *   typeLinkUpdate must be sent.
+         *   
+         * This branch is would also be taken when we had a working link before 
+         *   (ld->up == true). I'm not sure if this case does actually happen 
+         *   and whether it's tested.
+         */
+                else
+                {
                         // note: necessary to validate the link on the remote side!
                         logging_info( "Sending out update" <<
@@ -1144 +1572 @@
                         // compile and send update message
                         OverlayMsg overlayMsg( OverlayMsg::typeLinkUpdate );
-                        overlayMsg.setSourceLink(ld->overlayId);
                         overlayMsg.setAutoLink( ld->autolink );
-                        send_link( &overlayMsg, ld->overlayId, true );
+                    overlayMsg.setSourceNode(nodeId);
+                    overlayMsg.setDestinationNode(ld->remoteNode);
+                    overlayMsg.setSourceLink(ld->overlayId);
+                    overlayMsg.setDestinationLink(ld->remoteLink);
+                    overlayMsg.setService(ld->service);
+                    overlayMsg.setRelayed(false);
+
+                    // TODO ld->communicationId = id ??
+                    
+                    send_overlaymessage_down(&overlayMsg, id, system_priority::OVERLAY);
                 }
         }
@@ -1152 +1588 @@
 
 void BaseOverlay::onLinkDown(const LinkID& id,
-                const address_v* local, const address_v* remote) {
-
+        const addressing2::EndpointPtr local,
+        const addressing2::EndpointPtr remote)
+{
         // erase bootstrap links
         vector<LinkID>::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id );
@@ -1185 +1622 @@
 }
 
+
+void BaseOverlay::onLinkFail(const LinkID& id,
+        const addressing2::EndpointPtr local,
+        const addressing2::EndpointPtr remote)
+{
+        logging_debug( "Link fail with base communication link id=" << id );
+
+//      // erase bootstrap links
+//      vector<LinkID>::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id );
+//      if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it );
+//
+//      // get descriptor for link
+//      LinkDescriptor* ld = getDescriptor(id, true);
+//      if ( ld == NULL ) return; // not found? ->ignore!
+//      logging_debug( "Link failed id=" << ld->overlayId.toString() );
+//
+//      // inform listeners
+//      ld->listener->onLinkFail( ld->overlayId, ld->remoteNode );
+//      sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId );
+        
+        logging_debug( "  ... calling onLinkDown ..." );
+        onLinkDown(id, local, remote);
+}
+
+
 void BaseOverlay::onLinkChanged(const LinkID& id,
-                const address_v* oldlocal, const address_v* newlocal,
-                const address_v* oldremote, const address_v* newremote) {
-
-        // get descriptor for link
-        LinkDescriptor* ld = getDescriptor(id, true);
-        if ( ld == NULL ) return; // not found? ->ignore!
-        logging_debug( "onLinkChanged descriptor: " << ld );
-
-        // inform listeners
-        ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode );
-        sideport->onLinkChanged( id, this->nodeId, ld->remoteNode, this->spovnetId );
-
-        // autolinks: refresh timestamp
-        ld->setAutoUsed();
-}
-
-void BaseOverlay::onLinkFail(const LinkID& id,
-                const address_v* local, const address_v* remote) {
-        logging_debug( "Link fail with base communication link id=" << id );
-
-        // erase bootstrap links
-        vector<LinkID>::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id );
-        if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it );
-
-        // get descriptor for link
-        LinkDescriptor* ld = getDescriptor(id, true);
-        if ( ld == NULL ) return; // not found? ->ignore!
-        logging_debug( "Link failed id=" << ld->overlayId.toString() );
-
-        // inform listeners
-        ld->listener->onLinkFail( ld->overlayId, ld->remoteNode );
-        sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId );
-}
-
-void BaseOverlay::onLinkQoSChanged(const LinkID& id, const address_v* local,
-                const address_v* remote, const QoSParameterSet& qos) {
-        logging_debug( "Link quality changed with base communication link id=" << id );
-
-        // get descriptor for link
-        LinkDescriptor* ld = getDescriptor(id, true);
-        if ( ld == NULL ) return; // not found? ->ignore!
-        logging_debug( "Link quality changed id=" << ld->overlayId.toString() );
-}
-
-bool BaseOverlay::onLinkRequest( const LinkID& id, const address_v* local,
-                const address_v* remote ) {
+        const addressing2::EndpointPtr oldlocal,  const addressing2::EndpointPtr newlocal,
+        const addressing2::EndpointPtr oldremote, const addressing2::EndpointPtr newremote)
+{
+    // get descriptor for link
+    LinkDescriptor* ld = getDescriptor(id, true);
+    if ( ld == NULL ) return; // not found? ->ignore!
+    logging_debug( "onLinkChanged descriptor: " << ld );
+
+    // inform listeners
+    ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode );
+    sideport->onLinkChanged( id, this->nodeId, ld->remoteNode, this->spovnetId );
+
+    // autolinks: refresh timestamp
+    ld->setAutoUsed();
+}
+
+//void BaseOverlay::onLinkQoSChanged(const LinkID& id,
+//        const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote,
+//        const QoSParameterSet& qos)
+//{
+//      logging_debug( "Link quality changed with base communication link id=" << id );
+//
+//      // get descriptor for link
+//      LinkDescriptor* ld = getDescriptor(id, true);
+//      if ( ld == NULL ) return; // not found? ->ignore!
+//      logging_debug( "Link quality changed id=" << ld->overlayId.toString() );
+//}
+
+bool BaseOverlay::onLinkRequest(const LinkID& id,
+        const addressing2::EndpointPtr local,
+        const addressing2::EndpointPtr remote)
+{
         logging_debug("Accepting link request from " << remote->to_string() );
+        
+        // TODO ask application..?
+        
         return true;
 }
 
+
+
+
 /// handles a message from base communication
-bool BaseOverlay::receiveMessage(const Message* message,
-                const LinkID& link, const NodeID& ) {
+bool BaseOverlay::receiveMessage( reboost::shared_buffer_t message,
+                const LinkID& link,
+                const NodeID&,
+                bool bypass_overlay )
+{
         // get descriptor for link
         LinkDescriptor* ld = getDescriptor( link, true );
-        return handleMessage( message, ld, link );
+
+        
+        /* choose fastpath for direct links; normal overlay-path otherwise */   
+        if ( bypass_overlay && ld )
+        {
+        // message received --> link is alive
+        ld->keepAliveReceived = time(NULL);
+        // hop count on this link
+        ld->hops = 0;
+
+        
+        // hand over to CommunicationListener (aka Application) 
+            CommunicationListener* lst = getListener(ld->service);
+            if ( lst != NULL )
+            {
+                lst->onMessage(
+                        message,
+                        ld->remoteNode,
+                        ld->overlayId,
+                    SequenceNumber::DISABLED,
+                        NULL );
+                
+                return true;
+            }
+
+            return false;
+        }
+        else
+        {
+            return handleMessage( message, ld, link );      
+        }
 }
 
@@ -1247 +1734 @@
 
 /// Handle spovnet instance join requests
-bool BaseOverlay::handleJoinRequest( OverlayMsg* overlayMsg, const LinkID& bcLink ) {
-
+bool BaseOverlay::handleJoinRequest( reboost::shared_buffer_t message, const NodeID& source, const LinkID& bcLink )
+{
         // decapsulate message
-        JoinRequest* joinReq = overlayMsg->decapsulate<JoinRequest>();
+        JoinRequest joinReq;
+        joinReq.deserialize_from_shared_buffer(message);
+        
         logging_info( "Received join request for spovnet " <<
-                        joinReq->getSpoVNetID().toString() );
+                        joinReq.getSpoVNetID().toString() );
 
         // check spovnet id
-        if( joinReq->getSpoVNetID() != spovnetId ) {
+        if( joinReq.getSpoVNetID() != spovnetId ) {
                 logging_error(
                                 "Received join request for spovnet we don't handle " <<
-                                joinReq->getSpoVNetID().toString() );
-                delete joinReq;
+                                joinReq.getSpoVNetID().toString() );
+
                 return false;
         }
@@ -1267 +1756 @@
         logging_info( "Sending join reply for spovnet " <<
                         spovnetId.toString() << " to node " <<
-                        overlayMsg->getSourceNode().toString() <<
+                        source.toString() <<
                         ". Result: " << (allow ? "allowed" : "denied") );
-        joiningNodes.push_back( overlayMsg->getSourceNode() );
+        joiningNodes.push_back( source );
 
         // return overlay parameters
@@ -1276 +1765 @@
                         << getEndpointDescriptor().toString() )
         OverlayParameterSet parameters = overlayInterface->getParameters();
+        
+        
+        // create JoinReplay Message
         OverlayMsg retmsg( OverlayMsg::typeJoinReply,
                         OverlayInterface::OVERLAY_SERVICE_ID, nodeId );
-        JoinReply replyMsg( spovnetId, parameters,
-                        allow, getEndpointDescriptor() );
-        retmsg.encapsulate(&replyMsg);
-        bc->sendMessage( bcLink, &retmsg );
-
-        delete joinReq;
+        JoinReply replyMsg( spovnetId, parameters, allow );
+        retmsg.append_buffer(replyMsg.serialize_into_shared_buffer());
+
+        // XXX This is unlovely clash between the old message system and the new one,
+        // but a.t.m. we can't migrate everything to the new system at once..
+        // ---> Consider the EndpointDescriptor as part of the JoinReply..
+        retmsg.append_buffer(getEndpointDescriptor().serialize());
+        
+        // * send *
+        send_overlaymessage_down(&retmsg, bcLink, system_priority::OVERLAY);
+
         return true;
 }
 
 /// Handle replies to spovnet instance join requests
-bool BaseOverlay::handleJoinReply( OverlayMsg* overlayMsg, const LinkID& bcLink ) {
+bool BaseOverlay::handleJoinReply( reboost::shared_buffer_t message, const LinkID& bcLink )
+{
         // decapsulate message
         logging_debug("received join reply message");
-        JoinReply* replyMsg = overlayMsg->decapsulate<JoinReply>();
+        JoinReply replyMsg;
+        EndpointDescriptor endpoints;
+        reboost::shared_buffer_t buff = replyMsg.deserialize_from_shared_buffer(message);
+        buff = endpoints.deserialize(buff);
 
         // correct spovnet?
-        if( replyMsg->getSpoVNetID() != spovnetId ) { // no-> fail
+        if( replyMsg.getSpoVNetID() != spovnetId ) { // no-> fail
                 logging_error( "Received SpoVNet join reply for " <<
-                                replyMsg->getSpoVNetID().toString() <<
+                                replyMsg.getSpoVNetID().toString() <<
                                 " != " << spovnetId.toString() );
-                delete replyMsg;
+
                 return false;
         }
 
         // access granted? no -> fail
-        if( !replyMsg->getJoinAllowed() ) {
+        if( !replyMsg.getJoinAllowed() ) {
                 logging_error( "Our join request has been denied" );
 
@@ -1317 +1818 @@
 
                 // inform all registered services of the event
-                BOOST_FOREACH( NodeListener* i, nodeListeners )
+                foreach( NodeListener* i, nodeListeners )
                 i->onJoinFailed( spovnetId );
 
-                delete replyMsg;
                 return true;
         }
@@ -1329 +1829 @@
 
         logging_debug( "Using bootstrap end-point "
-                        << replyMsg->getBootstrapEndpoint().toString() );
+                        << endpoints.toString() );
 
         // create overlay structure from spovnet parameter set
@@ -1338 +1838 @@
 
                 overlayInterface = OverlayFactory::create(
-                                *this, replyMsg->getParam(), nodeId, this );
+                                *this, replyMsg.getParam(), nodeId, this );
 
                 // overlay structure supported? no-> fail!
@@ -1354 +1854 @@
 
                         // inform all registered services of the event
-                        BOOST_FOREACH( NodeListener* i, nodeListeners )
+                        foreach( NodeListener* i, nodeListeners )
                         i->onJoinFailed( spovnetId );
 
-                        delete replyMsg;
                         return true;
                 }
@@ -1365 +1864 @@
                 overlayInterface->createOverlay();
 
-                overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() );
-                overlayBootstrap.recordJoin( replyMsg->getBootstrapEndpoint() );
+                overlayInterface->joinOverlay( endpoints );
+                overlayBootstrap.recordJoin( endpoints );
 
                 // update ovlvis
@@ -1372 +1871 @@
 
                 // inform all registered services of the event
-                BOOST_FOREACH( NodeListener* i, nodeListeners )
-                i->onJoinCompleted( spovnetId );
-
-                delete replyMsg;
-
-        } else {
-
+                foreach( NodeListener* i, nodeListeners )
+                    i->onJoinCompleted( spovnetId );
+        }
+        else
+        {
                 // this is not the first bootstrap, just join the additional node
                 logging_debug("not first-time bootstrapping");
-                overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() );
-                overlayBootstrap.recordJoin( replyMsg->getBootstrapEndpoint() );
-
-                delete replyMsg;
-
+                overlayInterface->joinOverlay( endpoints );
+                overlayBootstrap.recordJoin( endpoints );
         } // if( overlayInterface == NULL )
 
@@ -1392 +1886 @@
 
 
-bool BaseOverlay::handleData( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
+bool BaseOverlay::handleData( reboost::shared_buffer_t message, OverlayMsg* overlayMsg, LinkDescriptor* ld )
+{
         // get service
-        const ServiceID& service = overlayMsg->getService();
+        const ServiceID& service = ld->service; //overlayMsg->getService();
+
         logging_debug( "Received data for service " << service.toString()
                         << " on link " << overlayMsg->getDestinationLink().toString() );
@@ -1402 +1898 @@
         if(lst != NULL){
                 lst->onMessage(
-                                overlayMsg,
-                                overlayMsg->getSourceNode(),
-                                overlayMsg->getDestinationLink()
+                                message,
+//                              overlayMsg->getSourceNode(),
+//                              overlayMsg->getDestinationLink(),
+                                ld->remoteNode,
+                                ld->overlayId,
+                overlayMsg->getSeqNum(),
+                                overlayMsg
                 );
         }
@@ -1411 +1911 @@
 }
 
+bool BaseOverlay::handleLostMessage( reboost::shared_buffer_t message, OverlayMsg* msg )
+{
+    /**
+     * Deserialize MessageLost-Message
+     * 
+     * - Type of lost message
+     * - Hop count of lost message
+     * - Source-LinkID  of lost message
+     */
+    const uint8_t* buff = message(0, sizeof(uint8_t)*2).data();
+    uint8_t type = buff[0];
+    uint8_t hops = buff[1];
+    LinkID linkid;
+    linkid.deserialize(message(sizeof(uint8_t)*2));
+    
+    logging_warn("Node " << msg->getSourceNode()
+            << " informed us, that our message of type " << (int) type
+            << " is lost after traveling " << (int) hops << " hops."
+            << " (LinkID: " << linkid.toString());
+
+    
+    // TODO switch-case ?
+    
+    // BRANCH: LinkRequest --> link request failed
+    if ( type == OverlayMsg::typeLinkRequest )
+    {
+        __onLinkEstablishmentFailed(linkid);
+    }
+    
+    // BRANCH: Data --> link disrupted. Drop link.
+    //   (We could use something more advanced here. e.g. At least send a 
+    //    keep-alive message and wait for a keep-alive reply.)
+    if ( type == OverlayMsg::typeData )
+    {
+        LinkDescriptor* link_desc = getDescriptor(linkid);
+        
+        if ( link_desc )
+        {
+            link_desc->failed = true;
+        }
+        
+        dropLink(linkid);
+    }
+    
+    // BRANCH: ping lost
+    if ( type == OverlayMsg::typePing )
+    {
+        CommunicationListener* lst = getListener(msg->getService());
+        if( lst != NULL )
+        {
+            lst->onPingLost(msg->getSourceNode());
+        }
+    }
+    
+    return true;
+}
+
+bool BaseOverlay::handlePing( OverlayMsg* overlayMsg, LinkDescriptor* ld )
+{
+    // TODO AKTUELL: implement interfaces: Node::ping(node); BaseOverlay::ping(node)
+    
+    bool send_pong = false;
+    
+    // inform application and ask permission to send a pong message
+    CommunicationListener* lst = getListener(overlayMsg->getService());
+    if( lst != NULL )
+    {
+        send_pong = lst->onPing(overlayMsg->getSourceNode());
+    }
+    
+    // send pong message if allowed
+    if ( send_pong )
+    {
+        OverlayMsg pong_msg(OverlayMsg::typePong);
+        pong_msg.setSeqNum(overlayMsg->getSeqNum());
+        
+        // send message
+        try
+        {
+            send_node( &pong_msg, 
+                overlayMsg->getSourceNode(), 
+                system_priority::OVERLAY,
+                overlayMsg->getService() );
+        }
+        catch ( message_not_sent& e )
+        {
+            logging_info("Could not send Pong-Message to node: " << 
+                overlayMsg->getSourceNode());
+        }
+    }
+}
+
+bool BaseOverlay::handlePong( OverlayMsg* overlayMsg, LinkDescriptor* ld )
+{
+    // inform application
+    CommunicationListener* lst = getListener(overlayMsg->getService());
+    if( lst != NULL )
+    {
+        lst->onPong(overlayMsg->getSourceNode());
+    }
+}
 
 bool BaseOverlay::handleLinkUpdate( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
@@ -1439 +2040 @@
                 overlayMsg->setSourceLink(ld->overlayId);
                 overlayMsg->setService(ld->service);
-                send( overlayMsg, ld );
+                send( overlayMsg, ld, system_priority::OVERLAY );
         }
 
@@ -1481 +2082 @@
         if( ld->messageQueue.size() > 0 ) {
                 logging_info( "Sending out queued messages on link " << ld );
-                BOOST_FOREACH( Message* msg, ld->messageQueue ) {
-                        sendMessage( msg, ld->overlayId );
-                        delete msg;
-                }
+        foreach( LinkDescriptor::message_queue_entry msg, ld->messageQueue )
+        {
+            sendMessage( msg.message, ld->overlayId, msg.priority );
+        }
                 ld->messageQueue.clear();
         }
@@ -1497 +2098 @@
 /// handle a link request and reply
 bool BaseOverlay::handleLinkRequest( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
-        logging_info( "Link request received from node id=" << overlayMsg->getSourceNode() );
 
         //TODO: Check if a request has already been sent using getSourceLink() ...
@@ -1514 +2114 @@
         ldn->remoteNode = overlayMsg->getSourceNode();
         ldn->remoteLink = overlayMsg->getSourceLink();
-
+        ldn->hops = overlayMsg->getNumHops();
+        
+    // initialize sequence numbers
+    ldn->last_sent_seqnum = SequenceNumber::createRandomSeqNum_Short();
+    logging_debug("Creating new link with initial SeqNum: " << ldn->last_sent_seqnum);
+    
+    
         // update time-stamps
         ldn->setAlive();
         ldn->setAutoUsed();
 
+        logging_info( "Link request received from node id="
+                << overlayMsg->getSourceNode()
+                << " LINK: "
+                << ldn);
+        
         // create reply message and send back!
         overlayMsg->swapRoles(); // swap source/destination
         overlayMsg->setType(OverlayMsg::typeLinkReply);
         overlayMsg->setSourceLink(ldn->overlayId);
-        overlayMsg->setSourceEndpoint( bc->getEndpointDescriptor() );
         overlayMsg->setRelayed(true);
-        send( overlayMsg, ld ); // send back to link
+//      overlayMsg->setRouteRecord(true);
+    overlayMsg->setSeqNum(ld->last_sent_seqnum);
+        
+        // TODO aktuell do the same thing in the typeLinkRequest-Message, too. But be careful with race conditions!!
+        // append our endpoints (for creation of a direct link)
+        overlayMsg->set_payload_message(bc->getEndpointDescriptor().serialize());
+        
+        send( overlayMsg, ld, system_priority::OVERLAY ); // send back to link
 
         // inform listener
@@ -1534 +2151 @@
 }
 
-bool BaseOverlay::handleLinkReply( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
-
+bool BaseOverlay::handleLinkReply(
+        OverlayMsg* overlayMsg,
+        reboost::shared_buffer_t sub_message,
+        LinkDescriptor* ld )
+{
+    // deserialize EndpointDescriptor
+    EndpointDescriptor endpoints;
+    endpoints.deserialize(sub_message);
+    
         // find link request
         LinkDescriptor* ldn = getDescriptor(overlayMsg->getDestinationLink());
@@ -1554 +2178 @@
 
         // debug message
-        logging_debug( "Link request reply received. Establishing link"
+        logging_info( "Link request reply received. Establishing link"
                         << " for service " << overlayMsg->getService().toString()
                         << " with local id=" << overlayMsg->getDestinationLink()
                         << " and remote link id=" << overlayMsg->getSourceLink()
-                        << " to " << overlayMsg->getSourceEndpoint().toString()
+                        << " to " << endpoints.toString()
+                        << " hop count: " << overlayMsg->getRouteRecord().size()
         );
 
@@ -1577 +2202 @@
                 logging_info( "Sending out queued messages on link " <<
                                 ldn->overlayId.toString() );
-                BOOST_FOREACH( Message* msg, ldn->messageQueue ) {
-                        sendMessage( msg, ldn->overlayId );
-                        delete msg;
+                foreach( LinkDescriptor::message_queue_entry msg, ldn->messageQueue )
+                {
+                        sendMessage( msg.message, ldn->overlayId, msg.priority );
                 }
                 ldn->messageQueue.clear();
@@ -1589 +2214 @@
         // try to replace relay link with direct link
         ldn->retryCounter = 3;
-        ldn->endpoint = overlayMsg->getSourceEndpoint();
+        ldn->endpoint = endpoints;
         ldn->communicationId =  bc->establishLink( ldn->endpoint );
 
@@ -1596 +2221 @@
 
 /// handle a keep-alive message for a link
-bool BaseOverlay::handleLinkAlive( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
+bool BaseOverlay::handleLinkAlive( OverlayMsg* overlayMsg, LinkDescriptor* ld )
+{
         LinkDescriptor* rld = getDescriptor(overlayMsg->getDestinationLink());
-        if ( rld != NULL ) {
-                logging_debug("Keep-Alive for " <<
-                                overlayMsg->getDestinationLink() );
+        
+        if ( rld != NULL )
+        {
+                logging_debug("Keep-Alive for " << overlayMsg->getDestinationLink() );
                 if (overlayMsg->isRouteRecord())
+                {
                         rld->routeRecord = overlayMsg->getRouteRecord();
+                }
+                
+                // set alive
                 rld->setAlive();
+                
+                
+                /* answer keep alive */
+                if ( overlayMsg->getType() == OverlayMsg::typeKeepAlive )
+                {
+            time_t now = time(NULL);
+            logging_debug("[BaseOverlay] Answering KeepAlive over "
+                    << ld->to_string()
+                    << " after "
+                    << difftime( now, ld->keepAliveSent )
+                    << "s");
+            
+            OverlayMsg msg( OverlayMsg::typeKeepAliveReply,
+                    OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode );
+            msg.setRouteRecord(true);
+            ld->keepAliveSent = now;
+            send_link( &msg, ld->overlayId, system_priority::OVERLAY );
+                }
+
                 return true;
-        } else {
-                logging_error("Keep-Alive for "
+        }
+        else
+        {
+                logging_error("No Keep-Alive for "
                                 << overlayMsg->getDestinationLink() << ": link unknown." );
                 return false;
@@ -1636 +2288 @@
         // erase the original descriptor
         eraseDescriptor(ld->overlayId);
+        
+    // inform listener
+    if( rld->listener != NULL)
+        rld->listener->onLinkChanged( rld->overlayId, rld->remoteNode );
+        
         return true;
 }
 
 /// handles an incoming message
-bool BaseOverlay::handleMessage( const Message* message, LinkDescriptor* ld,
-                const LinkID bcLink ) {
-        logging_debug( "Handling message: " << message->toString());
-
+bool BaseOverlay::handleMessage( reboost::shared_buffer_t message, LinkDescriptor* ld,
+                const LinkID bcLink )
+{
         // decapsulate overlay message
-        OverlayMsg* overlayMsg =
-                        const_cast<Message*>(message)->decapsulate<OverlayMsg>();
-        if( overlayMsg == NULL ) return false;
-
+        OverlayMsg* overlayMsg = new OverlayMsg();
+        reboost::shared_buffer_t sub_buff = overlayMsg->deserialize_from_shared_buffer(message);
+
+//      // XXX debug
+//      logging_info( "Received overlay message."
+//              << " Hops: " << (int) overlayMsg->getNumHops()
+//              << " Type: " << (int) overlayMsg->getType()
+//              << " Payload size: " << sub_buff.size()
+//             << " SeqNum: " << overlayMsg->getSeqNum() );
+        
+        
         // increase number of hops
         overlayMsg->increaseNumHops();
@@ -1660 +2323 @@
         // handle signaling messages (do not route!)
         if (overlayMsg->getType()>=OverlayMsg::typeSignalingStart &&
-                        overlayMsg->getType()<=OverlayMsg::typeSignalingEnd ) {
-                overlayInterface->onMessage(overlayMsg, NodeID::UNSPECIFIED, LinkID::UNSPECIFIED);
+                        overlayMsg->getType()<=OverlayMsg::typeSignalingEnd )
+        {
+                overlayInterface->onMessage(overlayMsg, sub_buff, NodeID::UNSPECIFIED, LinkID::UNSPECIFIED);
                 delete overlayMsg;
                 return true;
@@ -1673 +2337 @@
                                 << " to " << overlayMsg->getDestinationNode()
                 );
-                route( overlayMsg );
+                
+//              // XXX testing AKTUELL
+//        logging_info("MARIO: Routing message "
+//                << " from " << overlayMsg->getSourceNode()
+//                << " to " << overlayMsg->getDestinationNode() );
+//        logging_info( "Type: " << overlayMsg->getType() << " Payload size: " << sub_buff.size());
+                overlayMsg->append_buffer(sub_buff);
+                
+                route( overlayMsg, ld->remoteNode );
                 delete overlayMsg;
                 return true;
         }
 
-        // handle base overlay message
+        
+        /* handle base overlay message */
         bool ret = false; // return value
-        switch ( overlayMsg->getType() ) {
-
-        // data transport messages
-        case OverlayMsg::typeData:
-                ret = handleData(overlayMsg, ld);                       break;
-
-                // overlay setup messages
-        case OverlayMsg::typeJoinRequest:
-                ret = handleJoinRequest(overlayMsg, bcLink );   break;
-        case OverlayMsg::typeJoinReply:
-                ret = handleJoinReply(overlayMsg, bcLink );     break;
-
-                // link specific messages
-        case OverlayMsg::typeLinkRequest:
-                ret = handleLinkRequest(overlayMsg, ld );       break;
-        case OverlayMsg::typeLinkReply:
-                ret = handleLinkReply(overlayMsg, ld );         break;
-        case OverlayMsg::typeLinkUpdate:
-                ret = handleLinkUpdate(overlayMsg, ld );        break;
-        case OverlayMsg::typeLinkAlive:
-                ret = handleLinkAlive(overlayMsg, ld );         break;
-        case OverlayMsg::typeLinkDirect:
-                ret = handleLinkDirect(overlayMsg, ld );        break;
-
-                // handle unknown message type
-        default: {
-                logging_error( "received message in invalid state! don't know " <<
-                                "what to do with this message of type " << overlayMsg->getType() );
-                ret = false;
-                break;
-        }
+        try
+        {
+        switch ( overlayMsg->getType() ) 
+        {
+            // data transport messages
+            case OverlayMsg::typeData:
+            {
+                // NOTE: On relayed links, »ld« does not point to our link, but on the relay link. 
+                LinkDescriptor* end_to_end_ld = getDescriptor(overlayMsg->getDestinationLink());
+                
+                if ( ! end_to_end_ld )
+                {
+                    logging_warn("Error: Data-Message claims to belong to a link we don't know.");
+                    
+                    ret = false;
+                }
+                else
+                {
+                    // message received --> link is alive
+                    end_to_end_ld->keepAliveReceived = time(NULL);
+                    // hop count on this link
+                    end_to_end_ld->hops = overlayMsg->getNumHops();
+                    
+                    // * call handler *
+                    ret = handleData(sub_buff, overlayMsg, end_to_end_ld);
+                }
+                
+                break;
+            }
+            case OverlayMsg::typeMessageLost:
+                ret = handleLostMessage(sub_buff, overlayMsg);
+                
+                break;
+        
+                // overlay setup messages
+            case OverlayMsg::typeJoinRequest:
+                ret = handleJoinRequest(sub_buff, overlayMsg->getSourceNode(), bcLink );        break;
+            case OverlayMsg::typeJoinReply:
+                ret = handleJoinReply(sub_buff, bcLink );       break;
+        
+                // link specific messages
+            case OverlayMsg::typeLinkRequest:
+                ret = handleLinkRequest(overlayMsg, ld );       break;
+            case OverlayMsg::typeLinkReply:
+                ret = handleLinkReply(overlayMsg, sub_buff, ld );       break;
+            case OverlayMsg::typeLinkUpdate:
+                ret = handleLinkUpdate(overlayMsg, ld );        break;
+            case OverlayMsg::typeKeepAlive:
+            case OverlayMsg::typeKeepAliveReply:
+                ret = handleLinkAlive(overlayMsg, ld );         break;
+            case OverlayMsg::typeLinkDirect:
+                ret = handleLinkDirect(overlayMsg, ld );        break;
+                
+            case OverlayMsg::typeLinkClose:
+            {
+                dropLink(overlayMsg->getDestinationLink());
+                __removeDroppedLink(overlayMsg->getDestinationLink());
+                
+                break;
+            }
+            
+            /// ping over overlay path (or similar)
+            case OverlayMsg::typePing:
+            {
+                ret = handlePing(overlayMsg, ld);
+                break;
+            }
+            case OverlayMsg::typePong:
+            {
+                ret = handlePong(overlayMsg, ld);
+                break;
+            }
+            
+                // handle unknown message type
+            default:
+            {
+                logging_error( "received message in invalid state! don't know " <<
+                        "what to do with this message of type " << overlayMsg->getType() );
+                ret = false;
+                break;
+            }
+        }
+        }
+        catch ( reboost::illegal_sub_buffer& e )
+        {
+            logging_error( "Failed to create sub-buffer while reading message: »"
+                    << e.what()
+                    << "« Message too short? ");
+            
+            assert(false); // XXX
         }
 
@@ -1720 +2452 @@
 // ----------------------------------------------------------------------------
 
-void BaseOverlay::broadcastMessage(Message* message, const ServiceID& service) {
+void BaseOverlay::broadcastMessage(reboost::message_t message, const ServiceID& service, uint8_t priority) {
 
         logging_debug( "broadcasting message to all known nodes " <<
                         "in the overlay from service " + service.toString() );
-
-        if(message == NULL) return;
-        message->setReleasePayload(false);
 
         OverlayInterface::NodeList nodes = overlayInterface->getKnownNodes(true);
@@ -1732 +2461 @@
                 NodeID& id = nodes.at(i);
                 if(id == this->nodeId) continue; // don't send to ourselfs
-                if(i+1 == nodes.size()) message->setReleasePayload(true); // release payload on last send
-                sendMessage( message, id, service );
+
+                sendMessage( message, id, priority, service );
         }
 }
@@ -1755 +2484 @@
 vector<LinkID> BaseOverlay::getLinkIDs( const NodeID& nid ) const {
         vector<LinkID> linkvector;
-        BOOST_FOREACH( LinkDescriptor* ld, links ) {
+        foreach( LinkDescriptor* ld, links ) {
                 if( ld->remoteNode == nid || nid == NodeID::UNSPECIFIED ) {
                         linkvector.push_back( ld->overlayId );
@@ -1779 +2508 @@
         updateVisual();
 }
+
+
+
+/* link status */
+bool BaseOverlay::isLinkDirect(const ariba::LinkID& lnk) const
+{
+    const LinkDescriptor* ld = getDescriptor(lnk);
+    
+    if (!ld)
+        return false;
+    
+    return ld->communicationUp && !ld->relayed;
+}
+
+int BaseOverlay::getHopCount(const ariba::LinkID& lnk) const
+{
+    const LinkDescriptor* ld = getDescriptor(lnk);
+    
+    if (!ld)
+        return -1;
+    
+    return ld->hops;    
+}
+
+
+bool BaseOverlay::isLinkVital(const LinkDescriptor* link) const
+{
+    time_t now = time(NULL);
+
+    return link->up && difftime( now, link->keepAliveReceived ) <= KEEP_ALIVE_TIME_OUT; // TODO is this too long for a "vital" link..? 
+}
+
+bool BaseOverlay::isLinkDirectVital(const LinkDescriptor* link) const
+{
+    return isLinkVital(link) && link->communicationUp && !link->relayed;
+}
+
+/* [link status] */
+
 
 void BaseOverlay::updateVisual(){
@@ -1878 +2646 @@
         static set<NodeID> linkset;
         set<NodeID> remotenodes;
-        BOOST_FOREACH( LinkDescriptor* ld, links ) {
-                if (!ld->isVital() || ld->service != OverlayInterface::OVERLAY_SERVICE_ID)
+        foreach( LinkDescriptor* ld, links ) {
+                if (!isLinkVital(ld) || ld->service != OverlayInterface::OVERLAY_SERVICE_ID)
                         continue;
 
@@ -1895 +2663 @@
         do{
                 changed = false;
-                BOOST_FOREACH(NodeID n, linkset){
+                foreach(NodeID n, linkset){
                         if(remotenodes.find(n) == remotenodes.end()){
                                 visualInstance.visDisconnect(visualIdBase, this->nodeId, n, "");
@@ -1908 +2676 @@
         do{
                 changed = false;
-                BOOST_FOREACH(NodeID n, remotenodes){
+                foreach(NodeID n, remotenodes){
                         if(linkset.find(n) == linkset.end()){
                                 visualInstance.visConnect(visualIdBase, this->nodeId, n, "");
@@ -1933 +2701 @@
         // dump link state
         s << "--- link state -------------------------------" << endl;
-        BOOST_FOREACH( LinkDescriptor* ld, links ) {
+        foreach( LinkDescriptor* ld, links ) {
                 s << "link " << i << ": " << ld << endl;
                 i++;

source/ariba/overlay/BaseOverlay.h

@@ -47 +47 @@
 #include <vector>
 #include <deque>
+#include <stdexcept>
 #include <boost/foreach.hpp>
+
+#ifdef ECLIPSE_PARSER
+    #define foreach(a, b) for(a : b)
+#else
+    #define foreach(a, b) BOOST_FOREACH(a, b)
+#endif
 
 #include "ariba/utility/messages.h"
@@ -64 +71 @@
 #include "ariba/overlay/modules/OverlayStructureEvents.h"
 #include "ariba/overlay/OverlayBootstrap.h"
+#include "ariba/overlay/SequenceNumber.h"
 
 // forward declarations
@@ -92 +100 @@
 using ariba::communication::BaseCommunication;
 using ariba::communication::CommunicationEvents;
+
+// transport
+//using ariba::transport::system_priority;
 
 // utilities
@@ -103 +114 @@
 using ariba::utility::Demultiplexer;
 using ariba::utility::MessageReceiver;
-using ariba::utility::MessageSender;
 using ariba::utility::seqnum_t;
 using ariba::utility::Timer;
@@ -110 +120 @@
 namespace overlay {
 
-using namespace ariba::addressing;
+
+
+class message_not_sent: public std::runtime_error
+{
+public:
+    /** Takes a character string describing the error.  */
+    explicit message_not_sent(const string& __arg)  :
+        std::runtime_error(__arg)
+    {
+    }
+    
+    virtual ~message_not_sent() throw() {}
+};
+
+
 
 class LinkDescriptor;
@@ -121 +145 @@
                 protected Timer {
 
-        friend class OneHop;
+//      friend class OneHop;  // DEPRECATED
         friend class Chord;
         friend class ariba::SideportListener;
@@ -128 +152 @@
 
 public:
-
         /**
          * Constructs an empty non-functional base overlay instance
@@ -142 +165 @@
          * Starts the Base Overlay instance
          */
-        void start(BaseCommunication& _basecomm, const NodeID& _nodeid);
+        void start(BaseCommunication* _basecomm, const NodeID& _nodeid);
 
         /**
@@ -161 +184 @@
          * Starts a link establishment procedure to the specfied node
          * for the service with id service
-         *
+         * 
          * @param node Destination node id
          * @param service Service to connect to
@@ -179 +202 @@
         void dropLink( const LinkID& link );
 
+        
+        
+        /* +++++ Message sending +++++ */
+        
+        
         /// sends a message over an existing link
-        seqnum_t sendMessage(const Message* message, const LinkID& link );
+        const SequenceNumber& sendMessage(reboost::message_t message,
+                const LinkID& link,
+                uint8_t priority ) throw(message_not_sent);
 
         /// sends a message to a node and a specific service
-        seqnum_t sendMessage(const Message* message, const NodeID& remote,
-                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
+    const SequenceNumber& sendMessage(reboost::message_t message,
+                const NodeID& remote,
+                uint8_t priority,
+                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
+
 
     /**
@@ -191 +224 @@
      *  @return NodeID of the (closest) destination node; 
      */
-        NodeID sendMessageCloserToNodeID(const Message* message, const NodeID& address,
-                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
+        NodeID sendMessageCloserToNodeID(reboost::message_t message, const NodeID& address,
+                uint8_t priority, const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
         
         /**
@@ -198 +231 @@
          * Depending on the structure of the overlay, this can be very different.
          */
-        void broadcastMessage(Message* message, const ServiceID& service);
-
+        void broadcastMessage(reboost::message_t message, const ServiceID& service, uint8_t priority);
+
+        
+        /* +++++ [Message sending] +++++ */
+        
+        
+        
         /**
          * Returns the end-point descriptor of a link.
@@ -294 +332 @@
          */
         void leaveSpoVNet();
+        
+        
+        /* link status */
+        bool isLinkDirect(const ariba::LinkID& lnk) const;
+        int getHopCount(const ariba::LinkID& lnk) const;
+
+    bool isLinkVital(const LinkDescriptor* link) const;
+    bool isLinkDirectVital(const LinkDescriptor* link) const;
 
 protected:
-        /**
-         * @see ariba::communication::CommunicationEvents.h
-         */
-        virtual void onLinkUp(const LinkID& id, const address_v* local,
-                const address_v* remote);
-
-        /**
-         * @see ariba::communication::CommunicationEvents.h
-         */
-        virtual void onLinkDown(const LinkID& id, const address_v* local,
-                const address_v* remote);
-
-        /**
-         * @see ariba::communication::CommunicationEvents.h
-         */
-        virtual void onLinkChanged(const LinkID& id,
-                const address_v* oldlocal, const address_v* newlocal,
-                const address_v* oldremote, const address_v* newremote);
-
-        /**
-         * @see ariba::communication::CommunicationEvents.h
-         */
-        virtual void onLinkFail(const LinkID& id, const address_v* local,
-                const address_v* remote);
-
-        /**
-         * @see ariba::communication::CommunicationEvents.h
-         */
-        virtual void onLinkQoSChanged(const LinkID& id,
-                const address_v* local, const address_v* remote,
-                const QoSParameterSet& qos);
-
-        /**
-         * @see ariba::communication::CommunicationEvents.h
-         */
-        virtual bool onLinkRequest(const LinkID& id, const address_v* local,
-                const address_v* remote);
-
+
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+    virtual bool onLinkRequest(const LinkID& id,
+            const addressing2::EndpointPtr local,
+            const addressing2::EndpointPtr remote);
+
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+    virtual void onLinkUp(const LinkID& id,
+            const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
+
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+    virtual void onLinkDown(const LinkID& id,
+            const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
+
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+    virtual void onLinkChanged(const LinkID& id,
+        const addressing2::EndpointPtr oldlocal,  const addressing2::EndpointPtr newlocal,
+        const addressing2::EndpointPtr oldremote, const addressing2::EndpointPtr newremote);
+
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     * 
+     * NOTE: Just calls onLinkDown (at the moment..)
+     */
+    virtual void onLinkFail(const LinkID& id,
+            const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
+
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+//    virtual void onLinkQoSChanged(const LinkID& id,
+//            const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote,
+//            const QoSParameterSet& qos);
+
+        
+        
+        
         /**
          * Processes a received message from BaseCommunication
@@ -340 +393 @@
          * the node the message came from!
          */
-        virtual bool receiveMessage( const Message* message, const LinkID& link,
-                const NodeID& );
+        virtual bool receiveMessage( reboost::shared_buffer_t message,
+                const LinkID& link,
+                const NodeID&,
+                bool bypass_overlay );
 
         /**
@@ -359 +414 @@
         std::string getLinkHTMLInfo();
 
+
+private:
+        /// NOTE: "id" is an Overlay-LinkID
+        void __onLinkEstablishmentFailed(const LinkID& id);
+        
+        /// called from typeLinkClose-handler
+    void __removeDroppedLink(const LinkID& link);
+        
 private:
         /// is the base overlay started yet
@@ -396 +459 @@
 
         /// demultiplexes a incoming message with link descriptor
-        bool handleMessage( const Message* message, LinkDescriptor* ld,
+        bool handleMessage( reboost::shared_buffer_t message, LinkDescriptor* ld,
                 const LinkID bcLink = LinkID::UNSPECIFIED );
 
         // handle data and signalling messages
-        bool handleData( OverlayMsg* msg, LinkDescriptor* ld );
+        bool handleData( reboost::shared_buffer_t message, OverlayMsg* msg, LinkDescriptor* ld );
+        bool handleLostMessage( reboost::shared_buffer_t message, OverlayMsg* msg );
         bool handleSignaling( OverlayMsg* msg, LinkDescriptor* ld );
 
         // handle join request / reply messages
-        bool handleJoinRequest( OverlayMsg* msg, const LinkID& bcLink );
-        bool handleJoinReply( OverlayMsg* msg, const LinkID& bcLink );
+        bool handleJoinRequest( reboost::shared_buffer_t message, const NodeID& source, const LinkID& bcLink );
+        bool handleJoinReply( reboost::shared_buffer_t message, const LinkID& bcLink );
 
         // handle link messages
         bool handleLinkRequest( OverlayMsg* msg, LinkDescriptor* ld );
-        bool handleLinkReply( OverlayMsg* msg, LinkDescriptor* ld );
+        bool handleLinkReply( OverlayMsg* msg, reboost::shared_buffer_t sub_message, LinkDescriptor* ld );
         bool handleLinkUpdate( OverlayMsg* msg, LinkDescriptor* ld );
         bool handleLinkDirect( OverlayMsg* msg, LinkDescriptor* ld );
         bool handleLinkAlive( OverlayMsg* msg, LinkDescriptor* ld );
+    
+    // ping-pong over overlaypath/routing
+    bool handlePing( OverlayMsg* overlayMsg, LinkDescriptor* ld );
+    bool handlePong( OverlayMsg* overlayMsg, LinkDescriptor* ld );
 
 
@@ -478 +546 @@
         // internal message delivery -----------------------------------------------
 
+    // Convert OverlayMessage into new format and give it down to BaseCommunication
+    seqnum_t send_overlaymessage_down( OverlayMsg* message, const LinkID& bc_link, uint8_t priority );
+
+    
         /// routes a message to its destination node
-        void route( OverlayMsg* message );
-
+        void route( OverlayMsg* message, const NodeID& last_hop = NodeID::UNSPECIFIED );
+        
         /// sends a raw message to another node, delivers it to the base overlay class
-        seqnum_t send( OverlayMsg* message, const NodeID& destination );
+        /// may throw "message_not_sent"-exception
+        seqnum_t send( OverlayMsg* message,
+                   const NodeID& destination,
+                   uint8_t priority,
+                const NodeID& last_hop = NodeID::UNSPECIFIED )
+    throw(message_not_sent);
 
         /// send a raw message using a link descriptor, delivers it to the base overlay class
-        seqnum_t send( OverlayMsg* message, LinkDescriptor* ld,
-                bool ignore_down = false );
+        seqnum_t send( OverlayMsg* message,
+                LinkDescriptor* ld,
+                uint8_t priority ) throw(message_not_sent);
 
         /// send a message using a node id using overlay routing
         /// sets necessary fields in the overlay message!
-        seqnum_t send_node( OverlayMsg* message, const NodeID& remote,
-                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
+        /// may throw "message_not_sent"-exception
+        seqnum_t send_node( OverlayMsg* message, const NodeID& remote, uint8_t priority,
+                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID) throw(message_not_sent);
 
         /// send a message using a node id using overlay routing using a link
         /// sets necessary fields in the overlay message!
-        seqnum_t send_link( OverlayMsg* message, const LinkID& link,
-                bool ignore_down = false );
-
+        void send_link( OverlayMsg* message,
+                const LinkID& link,
+                uint8_t priority ) throw(message_not_sent);
+
+        
+        /// sends a notification to a sender from whom we just dropped a message
+        void report_lost_message( const OverlayMsg* message );
+        
         // misc --------------------------------------------------------------------
 

source/ariba/overlay/CMakeLists.txt

@@ -41 +41 @@
     LinkDescriptor.h
     OverlayBootstrap.h
+    SequenceNumber.h
     )
 
@@ -47 +48 @@
     LinkDescriptor.cpp
     OverlayBootstrap.cpp
+    SequenceNumber.cpp
     )
 

source/ariba/overlay/LinkDescriptor.h

@@ -12 +12 @@
 #include "ariba/communication/EndpointDescriptor.h"
 #include "ariba/CommunicationListener.h"
+
+// reboost messages
+#include "ariba/utility/transport/messages/message.hpp"
+#include <ariba/utility/misc/sha1.h>
+
+#include "ariba/overlay/SequenceNumber.h"
+
 
 namespace ariba {
@@ -37 +44 @@
 class LinkDescriptor {
 public:
+    struct message_queue_entry
+    {
+        reboost::message_t message;
+        uint8_t priority;
+    };
+    
         // ctor
         LinkDescriptor() {
+            time_t now = time(NULL);
+            
                 // default values
                 this->up = false;
+//         this->closing = false;
+        this->failed = false;
                 this->fromRemote = false;
                 this->remoteNode = NodeID::UNSPECIFIED;
@@ -46 +63 @@
                 this->communicationUp = false;
                 this->communicationId = LinkID::UNSPECIFIED;
-                this->keepAliveTime = time(NULL);
-                this->keepAliveMissed = 0;
+                this->keepAliveReceived = now;
+                this->keepAliveSent = now;
                 this->relaying     = false;
-                this->timeRelaying = time(NULL);
+                this->timeRelaying = now;
                 this->dropAfterRelaying = false;
                 this->service  = ServiceID::UNSPECIFIED;
@@ -56 +73 @@
                 this->remoteLink = LinkID::UNSPECIFIED;
                 this->autolink = false;
-                this->lastuse = time(NULL);
+                this->lastuse = now;
                 this->retryCounter = 0;
+                this->hops = -1;
+        
+        this->transmit_seqnums = false; // XXX
         }
 
@@ -67 +87 @@
         // general information about the link --------------------------------------
         bool up;           ///< flag whether this link is up and running
+//      bool closing;      ///< flag, whether this link is in the regular process of closing
+        bool failed;       ///< flag, whether communication is (assumed to be) not/no longer possible on this link
         bool fromRemote;   ///< flag, whether this link was requested from remote
         NodeID remoteNode; ///< remote end-point node
-        bool isVital() {
-                return up && keepAliveMissed == 0;
-        }
-        bool isDirectVital() {
-                return isVital() && communicationUp && !relayed;
-        }
 
 
@@ -82 +98 @@
         bool   communicationUp;   ///< flag, whether the communication is up
 
+        // sequence numbers --------------------------------------------------------
+        SequenceNumber last_sent_seqnum;
+    bool transmit_seqnums;
+    
         // direct link retries -----------------------------------------------------
         EndpointDescriptor endpoint;
@@ -87 +107 @@
 
         // link alive information --------------------------------------------------
-        time_t keepAliveTime; ///< the last time a keep-alive message was received
-        int keepAliveMissed;  ///< the number of missed keep-alive messages
+        time_t keepAliveReceived; ///< the last time a keep-alive message was received
+        time_t keepAliveSent;  ///< the number of missed keep-alive messages
         void setAlive() {
-                keepAliveMissed = 0;
-                keepAliveTime = time(NULL);
+//              keepAliveSent = time(NULL);
+                keepAliveReceived = time(NULL);
         }
 
@@ -98 +118 @@
         LinkID remoteLink; ///< the remote link id
         vector<NodeID> routeRecord;
+        int hops;
 
         // relay state -------------------------------------------------------------
@@ -114 +135 @@
         // auto links --------------------------------------------------------------
         bool autolink;  ///< flag, whether this link is a auto-link
-        time_t lastuse; ///< time, when the link was last used
-        deque<Message*> messageQueue; ///< waiting messages to be delivered
+        time_t lastuse; ///< time, when the link was last used XXX AUTO_LINK-ONLY
+        deque<message_queue_entry> messageQueue; ///< waiting messages to be delivered
         void setAutoUsed() {
                 if (autolink) lastuse = time(NULL);
@@ -121 +142 @@
         /// drops waiting auto-link messages
         void flushQueue() {
-                BOOST_FOREACH( Message* msg, messageQueue )     delete msg;
+//              BOOST_FOREACH( Message* msg, messageQueue )     delete msg;  // XXX MARIO: shouldn't be necessary anymore, since we're using shared pointers
                 messageQueue.clear();
         }
@@ -127 +148 @@
         // string representation ---------------------------------------------------
         std::string to_string() const {
+            time_t now = time(NULL);
+            
                 std::ostringstream s;
+        if ( relayed )
+            s << "[RELAYED-";
+        else
+            s << "[DIRECT-";
+        s << "LINK] ";
+        s << "id=" << overlayId.toString().substr(0,4) << " ";
+        s << "serv=" << service.toString() << " ";
                 s << "up=" << up << " ";
                 s << "init=" << !fromRemote << " ";
-                s << "id=" << overlayId.toString().substr(0,4) << " ";
-                s << "serv=" << service.toString() << " ";
                 s << "node=" << remoteNode.toString().substr(0,4) << " ";
                 s << "relaying=" << relaying << " ";
-                s << "miss=" << keepAliveMissed << " ";
+                s << "last_received=" << now - keepAliveReceived << "s ";
                 s << "auto=" << autolink << " ";
+                s << "hops=" << hops << " ";
                 if ( relayed ) {
                         s << "| Relayed: ";
@@ -146 +175 @@
                 } else {
                         s << "| Direct: ";
-                        s << "using id=" << communicationId.toString().substr(0,4) << " ";
+                        s << "using [COMMUNICATION-LINK] id=" << communicationId.toString().substr(0,4) << " ";
                         s << "(up=" << communicationUp << ") ";
                 }

source/ariba/overlay/messages/JoinReply.cpp

@@ -44 +44 @@
 vsznDefault(JoinReply);
 
-JoinReply::JoinReply(const SpoVNetID _spovnetid, const OverlayParameterSet _param, bool _joinAllowed, const EndpointDescriptor _bootstrapEp)
-        : spovnetid( _spovnetid ), param( _param ), joinAllowed( _joinAllowed ), bootstrapEp( _bootstrapEp ){
+JoinReply::JoinReply(const SpoVNetID _spovnetid, const OverlayParameterSet _param, bool _joinAllowed)
+        : spovnetid( _spovnetid ), param( _param ), joinAllowed( _joinAllowed )
+{
 }
 
@@ -64 +65 @@
 }
 
-const EndpointDescriptor& JoinReply::getBootstrapEndpoint(){
-        return bootstrapEp;
-}
+//const EndpointDescriptor& JoinReply::getBootstrapEndpoint(){
+//      return bootstrapEp;
+//}
 
 }} // ariba::overlay

source/ariba/overlay/messages/JoinReply.h

@@ -40 +40 @@
 #define JOIN_REPLY_H__
 
-#include "ariba/utility/messages.h"
+//#include "ariba/utility/messages.h"
+#include "ariba/utility/messages/Message.h"
 #include "ariba/utility/serialization.h"
 #include "ariba/utility/types/SpoVNetID.h"
 #include "ariba/utility/types/NodeID.h"
 #include "ariba/utility/types/OverlayParameterSet.h"
-#include "ariba/communication/EndpointDescriptor.h"
+//#include "ariba/communication/EndpointDescriptor.h"
 
 using ariba::utility::OverlayParameterSet;
@@ -51 +52 @@
 using ariba::utility::SpoVNetID;
 using ariba::utility::NodeID;
-using ariba::communication::EndpointDescriptor;
+//using ariba::communication::EndpointDescriptor;
 
 namespace ariba {
@@ -64 +65 @@
         OverlayParameterSet param; //< overlay parameters
         bool joinAllowed; //< join successfull or access denied
-        EndpointDescriptor bootstrapEp; //< the endpoint for bootstrapping the overlay interface
+//      EndpointDescriptor bootstrapEp; //< the endpoint for bootstrapping the overlay interface
 
 public:
@@ -70 +71 @@
                 const SpoVNetID _spovnetid = SpoVNetID::UNSPECIFIED,
                 const OverlayParameterSet _param = OverlayParameterSet::DEFAULT,
-                bool _joinAllowed = false,
-                const EndpointDescriptor _bootstrapEp = EndpointDescriptor::UNSPECIFIED()
+                bool _joinAllowed = false  /*,
+                const EndpointDescriptor _bootstrapEp = EndpointDescriptor::UNSPECIFIED()*/
         );
 
@@ -79 +80 @@
         const OverlayParameterSet& getParam();
         bool getJoinAllowed();
-        const EndpointDescriptor& getBootstrapEndpoint();
+//      const EndpointDescriptor& getBootstrapEndpoint();
 };
 
@@ -86 +87 @@
 sznBeginDefault( ariba::overlay::JoinReply, X ) {
         uint8_t ja = joinAllowed;
-        X && &spovnetid && param && bootstrapEp && ja;
+        X && &spovnetid && param;
+//      X && bootstrapEp;
+        X && ja;
         if (X.isDeserializer()) joinAllowed = ja;
 } sznEnd();

source/ariba/overlay/messages/OverlayMsg.h

@@ -47 +47 @@
 #include "ariba/utility/types/NodeID.h"
 #include "ariba/utility/types/LinkID.h"
-#include "ariba/communication/EndpointDescriptor.h"
-
+// #include <ariba/utility/misc/sha1.h>
+#include "ariba/overlay/SequenceNumber.h"
 
 namespace ariba {
@@ -57 +57 @@
 using ariba::utility::ServiceID;
 using ariba::utility::Message;
-using ariba::communication::EndpointDescriptor;
+//using ariba::communication::EndpointDescriptor;
 using_serialization;
 
@@ -64 +64 @@
  * between nodes.
  *
- * @author Sebastian Mies <mies@tm.uka.de>
+ * @author Sebastian Mies <mies@tm.uka.de>, Mario Hock
  */
 class OverlayMsg: public Message { VSERIALIZEABLE;
@@ -75 +75 @@
                 maskTransfer    = 0x10, ///< bit mask for transfer messages
                 typeData        = 0x11, ///< message contains data for higher layers
+                typeMessageLost = 0x12, ///< message contains info about a dropped message
 
                 // join signaling
@@ -87 +88 @@
                 typeLinkUpdate  = 0x33, ///< update message for link association
                 typeLinkDirect  = 0x34, ///< direct connection has been established
-                typeLinkAlive   = 0x35, ///< keep-alive message
+                typeKeepAlive   = 0x35, ///< keep-alive message
+                typeKeepAliveReply   = 0x36, ///< keep-alive message (replay)
+                typeLinkClose   = 0x37,
 
                 /// DHT routed messages
@@ -100 +103 @@
                 maskDHTResponse = 0x50, ///< bit mask for dht responses
                 typeDHTData     = 0x51, ///< DHT get data
+        
+        /// misc message types
+        typePing        = 0x44,
+        typePong        = 0x45,
 
                 // topology signaling
@@ -105 +112 @@
                 typeSignalingEnd = 0xFF    ///< end of the signaling types
         };
+    
+    /// message flags (uint8_t)
+    enum flags_
+    {
+        flagRelayed         = 1 << 0,
+        flagRegisterRelay   = 1 << 1,
+        flagRouteRecord     = 1 << 2,
+        flagSeqNum1         = 1 << 3,
+        flagSeqNum2         = 1 << 4,
+        flagAutoLink        = 1 << 5,
+        flagLinkMessage     = 1 << 6,
+        flagHasMoreFlags    = 1 << 7
+    };
 
         /// default constructor
@@ -114 +134 @@
                 const LinkID& _sourceLink      = LinkID::UNSPECIFIED,
                 const LinkID& _destinationLink = LinkID::UNSPECIFIED )
-        :       type(type), flags(0), hops(0), ttl(10),
+    :   type(type), flags(0), extended_flags(0), hops(0), ttl(10), priority(0),
                 service(_service),
                 sourceNode(_sourceNode), destinationNode(_destinationNode),
@@ -125 +145 @@
         // copy constructor
         OverlayMsg(const OverlayMsg& rhs)
-        :       type(rhs.type), flags(rhs.flags), hops(rhs.hops), ttl(rhs.ttl),
-                service(rhs.service),
+    :   type(rhs.type), flags(rhs.flags), extended_flags(rhs.extended_flags), 
+        hops(rhs.hops), ttl(rhs.ttl),
+                priority(rhs.priority), service(rhs.service),
                 sourceNode(rhs.sourceNode), destinationNode(rhs.destinationNode),
                 sourceLink(rhs.sourceLink), destinationLink(rhs.destinationLink),
@@ -149 +170 @@
         }
 
+        /// priority ------------------------------------------------------------------
+        
+        uint8_t getPriority() const {
+            return priority;
+        }
+        
+        void setPriority(uint8_t priority) {
+            this->priority = priority;
+        }
+        
         /// flags ------------------------------------------------------------------
 
         bool isRelayed() const {
-                return (flags & 0x01)!=0;
+        return (flags & flagRelayed)!=0;
         }
 
         void setRelayed( bool relayed = true ) {
-                if (relayed) flags |= 1; else flags &= ~1;
+        if (relayed) flags |= flagRelayed; else flags &= ~flagRelayed;
         }
 
         bool isRegisterRelay() const {
-                return (flags & 0x02)!=0;
+                return (flags & flagRegisterRelay)!=0;
         }
 
         void setRegisterRelay( bool relayed = true ) {
-                if (relayed) flags |= 0x02; else flags &= ~0x02;
+        if (relayed) flags |= flagRegisterRelay; else flags &= ~flagRegisterRelay;
         }
 
         bool isRouteRecord() const {
-                return (flags & 0x04)!=0;
+                return (flags & flagRouteRecord)!=0;
         }
 
         void setRouteRecord( bool route_record = true ) {
-                if (route_record) flags |= 0x04; else flags &= ~0x04;
+        if (route_record) flags |= flagRouteRecord; else flags &= ~flagRouteRecord;
         }
 
         bool isAutoLink() const {
-                return (flags & 0x80) == 0x80;
+        return (flags & flagAutoLink) == flagAutoLink;
         }
 
         void setAutoLink(bool auto_link = true ) {
-                if (auto_link) flags |= 0x80; else flags &= ~0x80;
+        if (auto_link) flags |= flagAutoLink; else flags &= ~flagAutoLink;
         }
 
         bool isLinkMessage() const {
-                return (flags & 0x40)!=0;
+                return (flags & flagLinkMessage)!=0;
         }
 
         void setLinkMessage(bool link_info = true ) {
-                if (link_info) flags |= 0x40; else flags &= ~0x40;
-        }
-
-        bool containsSourceEndpoint() const {
-                return (flags & 0x20)!=0;
-        }
-
-        void setContainsSourceEndpoint(bool contains_endpoint) {
-                if (contains_endpoint) flags |= 0x20; else flags &= ~0x20;
-        }
+        if (link_info) flags |= flagLinkMessage; else flags &= ~flagLinkMessage;
+        }
+        
+        bool hasExtendedFlags() const {
+        return (flags & flagHasMoreFlags) == flagHasMoreFlags;
+    }
 
         /// number of hops and time to live ----------------------------------------
@@ -264 +291 @@
                 this->destinationLink = link;
                 setLinkMessage();
-        }
-
-        void setSourceEndpoint( const EndpointDescriptor& endpoint ) {
-                sourceEndpoint = endpoint;
-                setContainsSourceEndpoint(true);
-        }
-
-        const EndpointDescriptor& getSourceEndpoint() const {
-                return sourceEndpoint;
         }
 
@@ -284 +302 @@
                 destinationLink = dummyLink;
                 hops = 0;
+                routeRecord.clear();
         }
 
@@ -294 +313 @@
                         routeRecord.push_back(node);
         }
+        
+        /// sequence numbers
+        bool hasShortSeqNum() const
+    {
+        return (flags & (flagSeqNum1 | flagSeqNum2)) == flagSeqNum1;
+    }
+    
+    bool hasLongSeqNum() const
+    {
+        return (flags & (flagSeqNum1 | flagSeqNum2)) == flagSeqNum2;
+    }
+    
+    void setSeqNum(const SequenceNumber& sequence_number)
+    {
+        this->seqnum = sequence_number;
+        
+        // short seqnum
+        if ( sequence_number.isShortSeqNum() )
+        {
+            flags |= flagSeqNum1;
+            flags &= ~flagSeqNum2;
+        }
+        // longseqnum
+        else if ( sequence_number.isShortSeqNum() )
+        {
+            flags &= ~flagSeqNum1;
+            flags |= flagSeqNum2;
+        }
+        // no seqnum
+        else
+        {
+            flags &= ~flagSeqNum1;
+            flags &= ~flagSeqNum2;
+        }
+    }
+    
+    const SequenceNumber& getSeqNum() const
+    {
+        return seqnum;
+    }
+    
 
 private:
-        uint8_t type, flags, hops, ttl;
+        uint8_t type, flags, extended_flags, hops, ttl, priority;
         ServiceID service;
         NodeID sourceNode;
@@ -302 +362 @@
         LinkID sourceLink;
         LinkID destinationLink;
-        EndpointDescriptor sourceEndpoint;
+//      EndpointDescriptor sourceEndpoint;
         vector<NodeID> routeRecord;
+    SequenceNumber seqnum;
 };
 
@@ -311 +372 @@
 sznBeginDefault( ariba::overlay::OverlayMsg, X ){
         // header
-        X && type && flags && hops && ttl;
+        X && type && flags;
+    
+    if ( hasExtendedFlags() )
+        X && extended_flags;
+    
+    X && hops && ttl;
 
         // addresses
         X && &service && &sourceNode && &destinationNode;
+        
+        // priority
+        X && priority;
 
         // message is associated with a end-to-end link
@@ -320 +389 @@
                 X && &sourceLink && &destinationLink;
 
-        // message is associated with a source end-point
-        if (containsSourceEndpoint())
-                X && sourceEndpoint;
-
+    
+    /* seqnum */
+    // serialize
+    if ( X.isSerializer() )
+    {
+        if ( hasShortSeqNum() )
+        {
+            uint32_t short_seqnum;
+            short_seqnum = seqnum.getShortSeqNum();
+            X && short_seqnum;
+        }
+        if ( hasLongSeqNum() )
+        {
+            uint64_t long_seqnum;
+            long_seqnum = seqnum.getLongSeqNum();
+            X && long_seqnum;
+        }
+    }
+    // deserialize
+    else
+    {
+        if ( hasShortSeqNum() )
+        {
+            uint32_t short_seqnum;
+            X && short_seqnum;
+            seqnum = ariba::overlay::SequenceNumber(short_seqnum);
+        }
+        if ( hasLongSeqNum() )
+        {
+            uint64_t long_seqnum;
+            X && long_seqnum;
+            seqnum = ariba::overlay::SequenceNumber(long_seqnum);
+        }        
+    }
+    
+    
         // message should record its route
         if (isRouteRecord()) {
@@ -333 +434 @@
 
         // payload
-        X && Payload();
+//      X && Payload();
 } sznEnd();
 

source/ariba/overlay/modules/OverlayFactory.cpp

@@ -41 +41 @@
 // structured overlays
 #include "chord/Chord.h"
-#include "onehop/OneHop.h"
+//#include "onehop/OneHop.h"  //DEPRECATED
 
 namespace ariba {
@@ -57 +57 @@
                         return new Chord( baseoverlay, nodeid, routeReceiver, param );
 
-                case OverlayParameterSet::OverlayStructureOneHop:
-                        return new OneHop( baseoverlay, nodeid, routeReceiver, param );
+//              case OverlayParameterSet::OverlayStructureOneHop:
+//                      return new OneHop( baseoverlay, nodeid, routeReceiver, param );
 
                 default:
+                    // NEVER return "NULL"
+                    assert(false);
+                    throw 42;
                         return NULL;
         }
 
+        // NEVER return "NULL"
+        assert(false);
+        throw 42;
         return NULL;
 }

source/ariba/overlay/modules/OverlayInterface.cpp

@@ -68 +68 @@
 
 void OverlayInterface::onLinkFail(const LinkID& lnk, const NodeID& remote) {
+    onLinkDown(lnk, remote);
 }
 
@@ -79 +80 @@
 }
 
-void OverlayInterface::onMessage(const DataMessage& msg, const NodeID& remote,
+void OverlayInterface::onMessage(OverlayMsg* msg,
+        reboost::shared_buffer_t sub_msg,
+        const NodeID& remote,
                 const LinkID& lnk) {
 }

source/ariba/overlay/modules/OverlayInterface.h

@@ -145 +145 @@
          */
         virtual const LinkID& getNextLinkId( const NodeID& id ) const = 0;
-        
+
+    /**
+     * Returns link ids of possible next hops a route message could take,
+     * sorted by "quality" (e.g. overlay-distance).
+     *
+     * The »num« parameter can be used to specify the desired number of elements 
+     * in the returned vector. This is intendet for optimizations. The
+     * implementation may choose to return a different number of elements than
+     * requested.
+     * 
+     * NOTE: The returned vector may contain »unspecified« links. These refer to 
+     * to the own node. (e.g. If there's no closer node, the top element in the 
+     * returned vector is unsoecified.)
+     * 
+     * @param id The destination node id
+     * @param num The desired number of elements in the returned vector.
+     *            (0 means »not specified/max)«
+     * @return A sorted vector of link ids to possible next hops.
+     */
+    virtual std::vector<const LinkID*> getSortedLinkIdsTowardsNode(
+        const NodeID& id, int num = 0 ) const = 0;
+
         /**
          * Returns the NodeID of the next hop a route message would take.
@@ -176 +197 @@
 
         /// @see CommunicationListener
-        virtual void onMessage(const DataMessage& msg, const NodeID& remote,
+        virtual void onMessage(OverlayMsg* msg,
+                reboost::shared_buffer_t sub_msg,
+                const NodeID& remote,
                         const LinkID& lnk = LinkID::UNSPECIFIED);
 

source/ariba/overlay/modules/OverlayStructureEvents.h

@@ -41 +41 @@
 
 #include "ariba/utility/types/NodeID.h"
-#include "ariba/utility/messages.h"
+#include "ariba/utility/types/LinkID.h"
+#include "ariba/utility/messages/Message.h"
 
 using ariba::utility::NodeID;
+using ariba::utility::LinkID;
 using ariba::utility::Message;
 
@@ -50 +52 @@
 
 class OverlayInterface;
-class OneHop;
+//class OneHop;
 
 class OverlayStructureEvents {
         friend class ariba::overlay::OverlayInterface;
-        friend class ariba::overlay::OneHop;
+//      friend class ariba::overlay::OneHop;
 
 public:

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;
+        }
+        }
 }
 

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

@@ -45 +45 @@
 #include "../OverlayInterface.h"
 #include <vector>
+#include <stdexcept>
 
 class chord_routing_table;
@@ -87 +88 @@
                 const NodeID& node = NodeID::UNSPECIFIED );
 
-        // helper: sends a message using the "base overlay"
-        seqnum_t send( OverlayMsg* msg, const LinkID& link );
+        // helper: sends a message over a link using the "base overlay"
+        void send( OverlayMsg* msg, const LinkID& link );
 
+        // helper: sends a message to a node using the "base overlay"
+        void send_node( OverlayMsg* message, const NodeID& remote );
+                
         // stabilization: sends a discovery message to the specified neighborhood
         void send_discovery_to( const NodeID& destination, int ttl = 3 );
@@ -105 +109 @@
         virtual const LinkID& getNextLinkId( const NodeID& id ) const;
         
+    /// @see OverlayInterface.h
+    /// NOTE: This implementation excepts num == 2
+    virtual std::vector<const LinkID*> getSortedLinkIdsTowardsNode(
+        const NodeID& id, int num = 0 ) const;
+
         /// @see OverlayInterface.h
         virtual const NodeID& getNextNodeId( const NodeID& id ) const;
@@ -138 +147 @@
 
         /// @see CommunicationListener.h or @see OverlayInterface.h
-        virtual void onMessage(const DataMessage& msg, const NodeID& remote,
+        virtual void onMessage(OverlayMsg* msg,
+                reboost::shared_buffer_t sub_msg,
+                const NodeID& remote,
                         const LinkID& lnk = LinkID::UNSPECIFIED);
 

source/ariba/overlay/modules/chord/detail/chord_routing_table.hpp

@@ -96 +96 @@
         // the own node id
         nodeid_t id;
+    
+    // the own node id as routing item
+    item own_id_item;
 
         // successor and predecessor tables
@@ -168 +171 @@
         /// constructs the reactive chord routing table
         explicit chord_routing_table( const nodeid_t& id, int redundancy = 4 ) :
-                id(id), succ( redundancy, succ_compare_type(this->id), *this ),
-                pred( redundancy, pred_compare_type(this->id), *this ) {
+                id(id),
+                succ( redundancy, succ_compare_type(this->id), *this ),
+                pred( redundancy, pred_compare_type(this->id), *this )
+    {
+        // init reflexive item
+        own_id_item.id = id;
+        own_id_item.ref_count = 1;
 
                 // create finger tables
@@ -273 +281 @@
                         }
                 }
+
                 if (best_item != NULL && distance(value, id)<distance(value, best_item->id))
                         return NULL;
                 return best_item;
         }
+        
+        std::vector<const item*> get_next_2_hops( const nodeid_t& value)
+    {
+        ring_distance distance;
+        item* best_item = &own_id_item;
+        item* second_best_item = NULL;
+        
+        // find best and second best item
+        for (size_t i=0; i<table.size(); i++)
+        {
+            item* curr = &table[i];
+
+            // not not include orphans into routing!
+            if (curr->ref_count==0) continue;
+
+            // check if we found a better item
+            // is best item
+            if ( distance(value, curr->id) < distance(value, best_item->id) )
+            {
+                second_best_item = best_item;
+                best_item = curr;
+            }
+            // is second best item
+            else
+            {
+                if ( second_best_item == NULL )
+                {
+                    second_best_item = curr;
+                    continue;
+                }
+                
+                if ( distance(value, curr->id) < distance(value, second_best_item->id) )
+                {
+                    second_best_item = curr;
+                }
+            }
+        }
+
+        // prepare return vector
+        std::vector<const item*> ret;
+        if ( best_item != NULL )
+        {
+            ret.push_back(best_item);
+        }
+        if ( second_best_item != NULL )
+        {
+            ret.push_back(second_best_item);
+        }
+        
+        return ret;
+    }
 
         const nodeid_t* get_successor() {
-                if (succ.size()==NULL) return NULL;
+                if (succ.size()==0) return NULL;
                 return &succ.front();
         }
 
         const nodeid_t* get_predesessor() {
-                if (pred.size()==NULL) return NULL;
+                if (pred.size()==0) return NULL;
                 return &pred.front();
         }

source/ariba/overlay/modules/chord/messages/CMakeLists.txt

@@ -37 +37 @@
 # [License]
 
-add_headers(Discovery.h)
+#add_headers(Discovery.h)
 
-add_sources(Discovery.cpp)
+#add_sources(Discovery.cpp)

source/ariba/overlay/modules/chord/messages/Discovery.h

@@ -59 +59 @@
 using_serialization;
 
+// XXX This whole message is DEPRECATED
 class Discovery : public Message {
         VSERIALIZEABLE;

source/ariba/overlay/modules/onehop/CMakeLists.txt

@@ -37 +37 @@
 # [License]
 
-add_headers(OneHop.h)
 
-add_sources(OneHop.cpp)
+## DEPRECATED
+#add_headers(OneHop.h)
 
-add_subdir_sources(messages)
+#add_sources(OneHop.cpp)
+
+#add_subdir_sources(messages)