Index: /source/ariba/Makefile.am =================================================================== --- /source/ariba/Makefile.am (revision 5860) +++ /source/ariba/Makefile.am (revision 5870) @@ -144,6 +144,4 @@ overlay/messages/JoinReply.cpp \ overlay/messages/JoinRequest.cpp \ - overlay/messages/LinkRequest.cpp \ - overlay/messages/RelayMessage.cpp \ overlay/messages/OverlayMsg.cpp @@ -151,6 +149,4 @@ overlay/messages/JoinReply.h \ overlay/messages/JoinRequest.h \ - overlay/messages/LinkRequest.h \ - overlay/messages/RelayMessage.h \ overlay/messages/OverlayMsg.h @@ -185,10 +181,8 @@ libariba_la_SOURCES += \ overlay/modules/chord/Chord.cpp \ - overlay/modules/chord/messages/ChordMessage.cpp \ overlay/modules/chord/messages/Discovery.cpp nobase_libariba_la_HEADERS += \ overlay/modules/chord/Chord.h \ - overlay/modules/chord/messages/ChordMessage.h \ overlay/modules/chord/messages/Discovery.h \ overlay/modules/chord/detail/chord_routing_table.hpp \ Index: /source/ariba/SideportListener.cpp =================================================================== --- /source/ariba/SideportListener.cpp (revision 5860) +++ /source/ariba/SideportListener.cpp (revision 5870) @@ -85,4 +85,6 @@ bool SideportListener::isRelayedNode(const NodeID& node){ if( overlay == NULL ) return false; +/* + bool relay = false; bool relay = false; @@ -100,15 +102,22 @@ } +<<<<<<< .working return relay; +======= + return relay; + */ + + return false; +>>>>>>> .merge-rechts.r5869 } bool SideportListener::isRelayingNode(const NodeID& node){ if( overlay == NULL ) return false; - +/* BOOST_FOREACH( LinkDescriptor* link, overlay->links ){ if( link->relay && link->localRelay == node && link->up) return true; } - +*/ return false; } @@ -119,5 +128,5 @@ using namespace ariba::addressing; - +/* LinkDescriptor* link = overlay->getSendDescriptor(node); if (link==NULL) return (Protocol)ret; @@ -142,5 +151,5 @@ ret = SideportListener::rfcomm; } - +*/ return (Protocol)ret; } Index: /source/ariba/overlay/BaseOverlay.cpp =================================================================== --- /source/ariba/overlay/BaseOverlay.cpp (revision 5860) +++ /source/ariba/overlay/BaseOverlay.cpp (revision 5870) @@ -49,9 +49,8 @@ #include "ariba/overlay/LinkDescriptor.h" + #include "ariba/overlay/messages/OverlayMsg.h" #include "ariba/overlay/messages/JoinRequest.h" #include "ariba/overlay/messages/JoinReply.h" -#include "ariba/overlay/messages/LinkRequest.h" -#include "ariba/overlay/messages/RelayMessage.h" #include "ariba/utility/misc/OvlVis.h" @@ -59,4 +58,20 @@ namespace ariba { namespace overlay { + +/* ***************************************************************************** + * PREREQUESITES + * ****************************************************************************/ + +CommunicationListener* BaseOverlay::getListener( const ServiceID& service ) { + if( !communicationListeners.contains( service ) ) { + logging_error( "No listener found for service " << service.toString() ); + return NULL; + } + CommunicationListener* listener = communicationListeners.get( service ); + assert( listener != NULL ); + return listener; +} + +// link descriptor handling ---------------------------------------------------- LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) { @@ -91,5 +106,5 @@ if ( desc == NULL ) { desc = new LinkDescriptor(); - desc->overlayId = link; + if (!link.isUnspecified()) desc->overlayId = link; links.push_back(desc); } @@ -99,7 +114,9 @@ /// returns a auto-link descriptor 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) 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 ) @@ -108,7 +125,17 @@ } -/// returns a direct local link relay descriptor for the given remote node -LinkDescriptor* BaseOverlay::getRelayDescriptor( const NodeID& id ) { - +/// stabilizes link information +void BaseOverlay::stabilizeLinks() { + // send keep-alive messages over established links + BOOST_FOREACH( LinkDescriptor* ld, links ) { + if (!ld->up) continue; + OverlayMsg msg( OverlayMsg::typeLinkAlive, + OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode ); + msg.setRelayed(true); + if (ld->relayed) msg.setRouteRecord(true); + send_link( &msg, ld->overlayId ); + } + +<<<<<<< .working // get used next hop towards node LinkDescriptor* rld = NULL; @@ -118,5 +145,15 @@ // get descriptor of first hop rld = getDescriptor(rlid); - +======= + // iterate over all links and check for time boundaries + vector oldlinks; + time_t now = time(NULL); + BOOST_FOREACH( LinkDescriptor* ld, links ) { + // remote used as relay flag + if ( ld->relaying && difftime( now, ld->timeRelaying ) > 10) + ld->relaying = false; +>>>>>>> .merge-rechts.r5869 + +<<<<<<< .working // is first hop a relay path use local relay if ( rld->relay ) relayNode = rld->localRelay; @@ -125,21 +162,24 @@ else relayNode = rld->remoteNode; } - - // if first relay is unknown choose a arbitrary direct node as relay - if ( relayNode.isUnspecified() ) { - for (size_t i=0; iup && - links[i]->communicationUp && - !links[i]->relay && - links[i]->keepAliveMissed <= 1 && - links[i]->service == OverlayInterface::OVERLAY_SERVICE_ID) { - relayNode = links[i]->remoteNode; - break; +======= + // keep alives and not up? yes-> link connection request is stale! + if ( !ld->up && difftime( now, ld->keepAliveTime ) >= 2 ) { +>>>>>>> .merge-rechts.r5869 + + // 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 ); + ld->relaying = false; + oldlinks.push_back( ld ); + continue; } - } - - // no local relay found-> damn! - if (relayNode.isUnspecified()) return NULL; - + } + + if (!ld->up) continue; + +<<<<<<< .working // get descriptor BOOST_FOREACH( LinkDescriptor* lp, links ) @@ -149,156 +189,282 @@ lp->up) return lp; - +======= + // drop links that are dropped and not used as relay + if (ld->dropAfterRelaying && !ld->relaying && !ld->autolink) { + oldlinks.push_back( ld ); + continue; + } + + // auto-link time exceeded? + if ( ld->autolink && difftime( now, ld->lastuse ) > 30 ) { + oldlinks.push_back( ld ); + continue; +>>>>>>> .merge-rechts.r5869 + + // keep alives missed? yes-> + if ( difftime( now, ld->keepAliveTime ) > 2 ) { + + // increase counter + ld->keepAliveMissed++; + + // missed more than four keep-alive messages (4 sec)? -> drop link + if (ld->keepAliveMissed >= 4) { + logging_info( "Link is stale, closing: " << ld ); + ld->relaying = false; + oldlinks.push_back( ld ); + continue; + } + } + } + + // drop links + BOOST_FOREACH( const LinkDescriptor* ld, oldlinks ) { +/* + vector::iterator it = std::find( + bootstrapLinks.begin(), bootstrapLinks.end(), ld->communicationId); + + if (!ld->communicationId.isUnspecified() && it != bootstrapLinks.end() ){ + logging_info( "Not dropping initiator link: " << ld ); + continue; + } +*/ + logging_info( "Link timed out. Dropping " << ld ); + dropLink( ld->overlayId ); + } + + // show link state + counter++; + if (counter>=4) showLinks(); + if (counter>=4 || counter<0) counter = 0; +} + +void BaseOverlay::showLinks() { + int i=0; + logging_info("--- link state -------------------------------"); + BOOST_FOREACH( LinkDescriptor* ld, links ) { + logging_info("link " << i << ": " << ld); + i++; + } + logging_info("----------------------------------------------"); +} + +// internal message delivery --------------------------------------------------- + +/// routes a message to its destination node +void BaseOverlay::route( OverlayMsg* message, LinkDescriptor* incomingLink ) { + + // exceeded time-to-live? yes-> drop message + if (message->getNumHops() > message->getTimeToLive()) { + logging_warn("Message exceeded TTL -> drop message!"); + return; + + // no-> forward message + } else { + // destinastion myself? yes-> handle message + if (message->getDestinationNode() == nodeId) + handleMessage( message, incomingLink ); + else + // no->send message to next hop + send( message, message->getDestinationNode() ); + } +} + +/// sends a message to another node, delivers it to the base overlay class +seqnum_t BaseOverlay::send( OverlayMsg* message, const NodeID& destination ) { + 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; + } + + // relay path known? yes-> send over relay link + next_link = getRelayLinkTo( destination ); + if (next_link != NULL) { + next_link->setRelaying(); + return send(message, next_link); + } + + // no-> relay path! route over overlay path + LinkID next_id = overlayInterface->getNextLinkId( destination ); + if (next_id.isUnspecified()) { + logging_error("Could not send message. No next hop found to " << destination ); + return -1; + } + + // get link descriptor, up and running? yes-> send message + next_link = getDescriptor(next_id); + if (next_link != NULL && next_link->up) + return send(message, next_link); + + // no-> error, dropping message + else { + logging_error("Could not send message. Link not known or up"); + return -1; + } + + // not reached-> fail + return -1; +} + +/// send a message using a link descriptor, delivers it to the base overlay class +seqnum_t BaseOverlay::send( OverlayMsg* message, LinkDescriptor* ld, bool ignore_down ) { + assert(ld!=NULL); + + // check if up + if (!ld->up && !ignore_down) { + logging_error("Can not send message. Link not up:" << ld ); + return -1; + } + + // handle relayed link + if (ld->relayed) { + logging_debug("send(): Resolving direct link for relayed link to " + << ld->remoteNode); + ld = getRelayLinkTo( ld->remoteNode ); + if (ld==NULL) { + logging_error("Direct link not found."); + return -1; + } + message->setRelayed(); + ld->setRelaying(); + } + + // 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 mesage. " + "Not a relayed link and direct link is not up."); + return -1; + } + return -1; +} + +seqnum_t BaseOverlay::send_node( OverlayMsg* message, const NodeID& remote, + const ServiceID& service) { + message->setSourceNode(nodeId); + message->setService(service); + message->setDestinationNode(remote); + send( message, remote ); +} + +seqnum_t BaseOverlay::send_link( OverlayMsg* message, const LinkID& link,bool ignore_down ) { + LinkDescriptor* ld = getDescriptor(link); + if (ld==NULL) { + logging_error("Cannot find descriptor to link id=" << link.toString()); + return -1; + } + message->setSourceNode(nodeId); + message->setSourceLink(ld->overlayId); + message->setService(ld->service); + message->setDestinationNode(ld->remoteNode); + message->setDestinationLink(ld->remoteLink); + return send( message, ld, ignore_down ); +} + +// relay route management ------------------------------------------------------ + +/// stabilize relay information +void BaseOverlay::stabilizeRelays() { + vector::iterator i = relay_routes.begin(); + while (i!=relay_routes.end() ) { + relay_route& route = *i; + LinkDescriptor* ld = getDescriptor(route.link); + if (ld==NULL + || !ld->up + || difftime(route.used, time(NULL)) > 4) { + logging_info("Forgetting relay information to node " + << route.node.toString() ); + i = relay_routes.erase(i); + } else + i++; + } +} + +void BaseOverlay::removeRelayLink( const LinkID& link ) { + vector::iterator i = relay_routes.begin(); + while (i!=relay_routes.end() ) { + relay_route& route = *i; + if (route.link == link ) i = relay_routes.erase(i); else i++; + } +} + +void BaseOverlay::removeRelayNode( const NodeID& remote ) { + vector::iterator i = relay_routes.begin(); + while (i!=relay_routes.end() ) { + relay_route& route = *i; + if (route.node == remote ) i = relay_routes.erase(i); else i++; + } +} + +/// refreshes relay information +void BaseOverlay::refreshRelayInformation( const OverlayMsg* message, LinkDescriptor* ld ) { + + // handle relayed messages from real links only + if (ld == NULL + || ld->relayed + || !message->isRelayed() + || message->getSourceNode()==nodeId ) return; + + // check wheter this node is already part of the routing table + LinkID next_link = overlayInterface->getNextLinkId(message->getSourceNode()); + if (next_link == ld->overlayId) return; + ld->setRelaying(); + + // try to find source node + BOOST_FOREACH( relay_route& route, relay_routes ) { + + // relay route found? yes-> + if ( route.node == message->getSourceNode() ) { + + // refresh timer + route.used = time(NULL); + + // route has a shorter hop count? yes-> replace + if (route.hops > message->getNumHops() ) { + logging_info("Updating relay information to node " + << route.node.toString() + << " reducing to " << message->getNumHops() << " hops."); + route.hops = message->getNumHops(); + route.link = ld->overlayId; + } + return; + } + } + + // not found-> add new entry + relay_route route; + route.hops = message->getNumHops(); + route.link = ld->overlayId; + route.node = message->getSourceNode(); + route.used = time(NULL); + logging_info("Remembering relay information to node " << route.node.toString()); + relay_routes.push_back(route); +} + +/// returns a known "vital" relay link which is up and running +LinkDescriptor* BaseOverlay::getRelayLinkTo( const NodeID& remote ) { + // try to find source node + BOOST_FOREACH( relay_route& route, relay_routes ) { + if (route.node == remote ) { + LinkDescriptor* ld = getDescriptor( route.link ); + if (ld==NULL || !ld->up) return NULL; else return ld; + } + } return NULL; } -/// returns the link descriptor that is actually used for sending a message over the overöay -LinkDescriptor* BaseOverlay::getSendDescriptor( const NodeID& nodeid, bool follow ) { - for (size_t i=0; irelay && - links[i]->up && - links[i]->communicationUp && - links[i]->keepAliveMissed <= 1 && - links[i]->remoteNode == nodeid && - links[i]->service == OverlayInterface::OVERLAY_SERVICE_ID) { - return links[i]; - } - LinkDescriptor* ld = getDescriptor(overlayInterface->getNextLinkId(nodeid)); - if (ld != NULL && ld->relay && follow) - return getSendDescriptor(ld->localRelay, false); - return NULL; -} - -NodeID BaseOverlay::getRelayNode( const NodeID& remoteNode ) { - LinkDescriptor* rld = getRelayDescriptor(remoteNode); - return rld!=NULL ? rld->remoteNode : NodeID::UNSPECIFIED; -} - -/// routes a message over the overlay or directly sends it when a link is open -seqnum_t BaseOverlay::sendOverlay( Message* message, const NodeID& nodeid, const NodeID& remoteRelay ) { - /// send message directly to a neighbor - for (size_t i=0; irelay && - links[i]->up && - links[i]->communicationUp && - links[i]->keepAliveMissed <= 1 && - links[i]->remoteNode == nodeid && - links[i]->service == OverlayInterface::OVERLAY_SERVICE_ID) { - - // mark as relay and send message - links[i]->markAsRelay(); - return sendMessage( message, links[i] ); - } - - /// send relayed message over the overlay - if (!remoteRelay.isUnspecified()) { - // create a information relay message to inform the relay about - OverlayMsg overlay_msg( - OverlayMsg::typeRelay, OverlayInterface::OVERLAY_SERVICE_ID, nodeId); - RelayMessage relayMsg( RelayMessage::typeInform, remoteRelay, nodeid, LinkID::UNSPECIFIED ); - relayMsg.encapsulate( message ); - overlay_msg.encapsulate( &relayMsg ); - - // get local relay link - LinkDescriptor* rld = getRelayDescriptor(nodeid); - - // local relay available? send to local relay! - if (rld!=NULL) { - rld->markAsRelay(); - sendMessage(&overlay_msg, rld); - } else - overlayInterface->routeMessage(remoteRelay, &overlay_msg); - - // finished - return 0; - } - - // common case: send message over the overlay - overlayInterface->routeMessage(nodeid, message); - return 0; -} - -/// forwards a message over relays/directly using link descriptor -seqnum_t BaseOverlay::sendMessage( Message* message, const LinkDescriptor* ld ) { - - // directly send message - if ( !ld->communicationId.isUnspecified() && ld->communicationUp ) { - logging_debug("Send: Sending message via Base Communication"); - return bc->sendMessage( ld->communicationId, message ); - } - - // relay message - else if ( ld->relay ) { - - // sending a relayed message - logging_debug("Send: Relaying message to node " - << ld->remoteNode.toString() - << " using relay " << ld->localRelay - ); - - // create a information relay message to inform the relay about - OverlayMsg overlay_msg( OverlayMsg::typeRelay, ld->service, nodeId ); - RelayMessage relayMsg( RelayMessage::typeInform, ld->remoteRelay, ld->remoteNode, ld->remoteLinkId ); - relayMsg.encapsulate( message ); - overlay_msg.encapsulate( &relayMsg ); - - // route message to relay node in order to inform it! - logging_debug("sendMessage: Sending message over relayed link with" << ld ); - sendOverlay( &overlay_msg, ld->localRelay ); - return 0; - } - - // error - else { - logging_error( "Could not send message descriptor=" << ld ); - return -1; - } - return -1; -} - -/// creates a link descriptor, apply relay semantics if possible -LinkDescriptor* BaseOverlay::createLinkDescriptor( - const NodeID remoteNode, const ServiceID service, const LinkID link_id ) { - - // find listener - if( !communicationListeners.contains( service ) ) { - logging_error( "No listener found for service " << service.toString() ); - return NULL; - } - CommunicationListener* listener = communicationListeners.get( service ); - assert( listener != NULL ); - - // copy link id - LinkID linkid = link_id; - - // create link id if necessary - if ( linkid.isUnspecified() ) - linkid = LinkID::create(); - - // create relay link descriptor - NodeID relayNode = getRelayNode(remoteNode); - - // add descriptor - LinkDescriptor* ld = addDescriptor( linkid ); - ld->overlayId = linkid; - ld->service = service; - ld->listener = listener; - ld->remoteNode = remoteNode; - - // set relay node if available - ld->relay = !relayNode.isUnspecified(); - ld->localRelay = relayNode; - - if (!ld->relay) - logging_error("No relay found!"); - - // debug output - logging_debug( "Created link descriptor: " << ld ); - - return ld; -} - - -// ---------------------------------------------------------------------------- +/* ***************************************************************************** + * PUBLIC MEMBERS + * ****************************************************************************/ use_logging_cpp(BaseOverlay); @@ -472,28 +638,21 @@ // ---------------------------------------------------------------------------- -const LinkID BaseOverlay::establishLink( - const EndpointDescriptor& ep, const NodeID& nodeid, - const ServiceID& service, const NodeID& remoteRelay, const LinkID& linkid ) { - - LinkID link_id = linkid; +const LinkID BaseOverlay::establishLink( const EndpointDescriptor& remoteEp, + const NodeID& remoteId, const ServiceID& service ) { // establish link via overlay - if (!nodeid.isUnspecified()) - link_id = establishLink( nodeid, service, remoteRelay, link_id ); + if (!remoteId.isUnspecified()) + return establishLink( remoteId, service ); + else // establish link directly if only ep is known - if (nodeid.isUnspecified()) - establishDirectLink( ep, service, link_id ); - - return link_id; + if (remoteId.isUnspecified()) + return establishDirectLink(remoteEp, service ); + } /// call base communication's establish link and add link mapping const LinkID BaseOverlay::establishDirectLink( const EndpointDescriptor& ep, - const ServiceID& service, const LinkID& linkid ) { - - // create a new link id if necessary - LinkID link_id = linkid; - if (link_id.isUnspecified()) link_id = LinkID::create(); + const ServiceID& service ) { /// find a service listener @@ -505,51 +664,47 @@ assert( listener != NULL ); - /// establish link and add mapping - logging_info("Establishing direct link " << link_id.toString() - << " using " << ep.toString()); - // create descriptor - LinkDescriptor* ld = addDescriptor( link_id ); - ld->overlayId = link_id; - ld->communicationId = link_id; + LinkDescriptor* ld = addDescriptor(); + ld->relayed = false; ld->listener = listener; ld->service = service; - bc->establishLink( ep, link_id ); - - return link_id; + ld->communicationId = bc->establishLink( ep ); + + /// establish link and add mapping + logging_info("Establishing direct link " << ld->communicationId.toString() + << " using " << ep.toString()); + + return ld->communicationId; } /// establishes a link between two arbitrary nodes -const LinkID BaseOverlay::establishLink( const NodeID& node, - const ServiceID& service, const NodeID& remoteRelay, const LinkID& link_id ) { +const LinkID BaseOverlay::establishLink( const NodeID& remote, + const ServiceID& service ) { // do not establish a link to myself! - if (node == nodeId) return LinkID::UNSPECIFIED; + if (remote == nodeId) return LinkID::UNSPECIFIED; // create a link descriptor - LinkDescriptor* ld = createLinkDescriptor( node, service, link_id ); - ld->remoteRelay = remoteRelay; - - // create link request message with own link id - uint32_t nonce = (uint32_t)(rand() ^ (rand() << 16) ^ time(NULL)); - LinkRequest link_request_msg( - nonce, &bc->getEndpointDescriptor(), false, - ld->overlayId, ld->localRelay ); - OverlayMsg overlay_msg( OverlayMsg::typeLinkRequest, service, nodeId ); - overlay_msg.encapsulate( &link_request_msg ); - pendingLinks.insert( make_pair(nonce, ld->overlayId) ); + LinkDescriptor* ld = addDescriptor(); + ld->relayed = true; + ld->remoteNode = remote; + ld->service = service; + ld->listener = getListener(ld->service); + + // create link request message + OverlayMsg msg(OverlayMsg::typeLinkRequest, service, nodeId, remote ); + msg.setSourceLink(ld->overlayId); + msg.setRelayed(true); // debug message - logging_debug( + logging_info( "Sending link request with" - << " link id=" << ld->overlayId - << " node id=" << ld->remoteNode.toString() - << " service id=" << ld->service.toString() - << " local relay id=" << ld->localRelay.toString() - << " nonce= " << nonce + << " link=" << ld->overlayId.toString() + << " node=" << ld->remoteNode.toString() + << " serv=" << ld->service.toString() ); - // sending message through new link - sendMessage( &overlay_msg, ld ); + // sending message to node + send_node( &msg, ld->remoteNode, ld->service ); return ld->overlayId; @@ -558,5 +713,5 @@ /// drops an established link void BaseOverlay::dropLink(const LinkID& link) { - logging_debug( "Dropping link (initiated locally):" << link.toString() ); + logging_info( "Dropping link (initiated locally):" << link.toString() ); // find the link item to drop @@ -579,5 +734,5 @@ // do not drop relay links - if (!ld->usedAsRelay) { + if (!ld->relaying) { // drop the link in base communication if (ld->communicationUp) bc->dropLink( ld->communicationId ); @@ -586,5 +741,5 @@ eraseDescriptor( ld->overlayId ); } else - ld->dropWhenRelaysLeft = true; + ld->dropAfterRelaying = true; } @@ -605,5 +760,5 @@ // check if the link is up yet, if its an auto link queue message if( !ld->up ) { - ld->markAsUsed(); + ld->setAutoUsed(); if( ld->autolink ) { logging_info("Auto-link " << link.toString() << " not up, queue message"); @@ -618,9 +773,9 @@ // compile overlay message (has service and node id) - OverlayMsg overmsg( OverlayMsg::typeData, ld->service, nodeId ); + OverlayMsg overmsg( OverlayMsg::typeData ); overmsg.encapsulate( const_cast(message) ); // send message over relay/direct/overlay - return sendMessage( &overmsg, ld ); + return send_link( &overmsg, ld->overlayId ); } @@ -640,9 +795,6 @@ ); - // this will call onlinkup on us, if everything worked we now have a mapping - LinkID link = LinkID::create(); - // call base overlay to create a link - link = establishLink( node, service, NodeID::UNSPECIFIED, link ); + LinkID link = establishLink( node, service ); ld = getDescriptor( link ); if( ld == NULL ) { @@ -658,5 +810,5 @@ // mark the link as used, as we now send a message through it - ld->markAsUsed(); + ld->setAutoUsed(); // send / queue message @@ -794,5 +946,6 @@ // send join request message - OverlayMsg overlayMsg( OverlayMsg::typeJoinRequest, nodeId ); + OverlayMsg overlayMsg( OverlayMsg::typeJoinRequest, + OverlayInterface::OVERLAY_SERVICE_ID, nodeId ); JoinRequest joinRequest( spovnetId, nodeId ); overlayMsg.encapsulate( &joinRequest ); @@ -804,5 +957,5 @@ if (ld == NULL) { ld = addDescriptor( id ); - logging_debug( "onLinkUp (remote request) descriptor: " << ld ); + logging_info( "onLinkUp (remote request) descriptor: " << ld ); // update descriptor @@ -810,5 +963,6 @@ ld->communicationId = id; ld->communicationUp = true; - ld->markAsUsed(); + ld->setAutoUsed(); + ld->setAlive(); // in this case, do not inform listener, since service it unknown @@ -817,31 +971,34 @@ // link mapping found? -> send update message with node-id and service id } else { - logging_debug( "onLinkUp descriptor (initiated locally):" << ld ); - - // note: necessary to validate the link on the remote side! - logging_debug( "Sending out update" << - " for service " << ld->service.toString() << - " with local node id " << nodeId.toString() << - " on link " << ld->overlayId.toString() ); + logging_info( "onLinkUp descriptor (initiated locally):" << ld ); // update descriptor - ld->markAsUsed(); + ld->setAutoUsed(); + ld->setAlive(); ld->communicationUp = true; - - // if link is a relayed link ->convert to direct link - if (ld->relay && !ld->remoteLinkId.isUnspecified() ) { + ld->fromRemote = false; + + // if link is a relayed link->convert to direct link + if (ld->relayed) { logging_info( "Converting to direct link: " << ld ); ld->up = true; - ld->relay = false; - ld->localRelay = NodeID::UNSPECIFIED; - OverlayMsg overMsg( OverlayMsg::typeDirectLink, ld->service, nodeId ); - overMsg.setRelayLink( ld->remoteLinkId ); - bc->sendMessage( ld->communicationId, &overMsg ); - } - - // compile and send update message - OverlayMsg overlayMsg( OverlayMsg::typeUpdate, ld->service, nodeId ); - overlayMsg.setAutoLink( ld->autolink ); - bc->sendMessage( ld->communicationId, &overlayMsg ); + ld->relayed = false; + OverlayMsg overMsg( OverlayMsg::typeLinkDirect ); + overMsg.setSourceLink( ld->overlayId ); + overMsg.setDestinationLink( ld->remoteLink ); + send_link( &overMsg, ld->overlayId ); + } else { + // note: necessary to validate the link on the remote side! + logging_info( "Sending out update" << + " for service " << ld->service.toString() << + " with local node id " << nodeId.toString() << + " on link " << ld->overlayId.toString() ); + + // compile and send update message + OverlayMsg overlayMsg( OverlayMsg::typeLinkUpdate ); + overlayMsg.setSourceLink(ld->overlayId); + overlayMsg.setAutoLink( ld->autolink ); + send_link( &overlayMsg, ld->overlayId, true ); + } } } @@ -859,8 +1016,13 @@ logging_info( "onLinkDown descriptor: " << ld ); + // removing relay link information + removeRelayLink(ld->overlayId); + // inform listeners about link down ld->communicationUp = false; - ld->listener->onLinkDown( ld->overlayId, ld->remoteNode ); - sideport->onLinkDown( id, this->nodeId, ld->remoteNode, this->spovnetId ); + if (!ld->service.isUnspecified()) { + getListener(ld->service)->onLinkDown( ld->overlayId, ld->remoteNode ); + sideport->onLinkDown( id, this->nodeId, ld->remoteNode, this->spovnetId ); + } // delete all queued messages (auto links) @@ -889,5 +1051,5 @@ // autolinks: refresh timestamp - ld->markAsUsed(); + ld->setAutoUsed(); } @@ -908,7 +1070,4 @@ ld->listener->onLinkFail( ld->overlayId, ld->remoteNode ); sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId ); - - // autolinks: refresh timestamp - ld->markAsUsed(); } @@ -921,7 +1080,4 @@ if ( ld == NULL ) return; // not found? ->ignore! logging_debug( "Link quality changed id=" << ld->overlayId.toString() ); - - // autolinks: refresh timestamp - ld->markAsUsed(); } @@ -937,192 +1093,184 @@ // get descriptor for link LinkDescriptor* ld = getDescriptor( link, true ); - - // link known? - if (ld == NULL) { // no-> handle with unspecified params - logging_debug("Received message from base communication, link descriptor unknown" ); - return handleMessage( message, LinkID::UNSPECIFIED, link, NodeID::UNSPECIFIED ); - } else { // yes -> handle with overlay link id - logging_debug("Received message from base communication, link id=" << ld->overlayId.toString() ); - return handleMessage( message, ld->overlayId, link, NodeID::UNSPECIFIED ); - } + return handleMessage( message, ld, link ); } // ---------------------------------------------------------------------------- -/// handles a message from an overlay -void BaseOverlay::incomingRouteMessage( Message* msg, const LinkID& link, const NodeID& source ) { - logging_debug("Received message from overlay -- " - << " link id=" << link.toString() - << " node id=" << source.toString() ); - handleMessage( msg, link, LinkID::UNSPECIFIED, source ); -} - -// ---------------------------------------------------------------------------- - -/// handles an incoming message -bool BaseOverlay::handleMessage( const Message* message, - const LinkID& boLink, const LinkID& bcLink, const NodeID& remoteNode ) { - logging_debug( "Handling message: " << message->toString()); - - bool ret = false; - - // decapsulate overlay message - OverlayMsg* overlayMsg = - const_cast(message)->decapsulate(); - if( overlayMsg == NULL ) return false; - - // mark the link as in action - LinkDescriptor* ld = getDescriptor(boLink); - if (ld == NULL) ld = getDescriptor(bcLink, true); - if (ld != NULL) { - ld->markAsUsed(); - ld->markAlive(); - } - - switch ( overlayMsg->getType() ) { - // --------------------------------------------------------------------- - // Handle spovnet instance join requests - // --------------------------------------------------------------------- - case OverlayMsg::typeJoinRequest: { - - // decapsulate message - JoinRequest* joinReq = overlayMsg->decapsulate(); - logging_info( "Received join request for spovnet " << - joinReq->getSpoVNetID().toString() ); - - // check spovnet id - if( joinReq->getSpoVNetID() != spovnetId ) { - logging_error( - "Received join request for spovnet we don't handle " << - joinReq->getSpoVNetID().toString() ); - ret = false; - break; +/// Handle spovnet instance join requests +bool BaseOverlay::handleJoinRequest( OverlayMsg* overlayMsg, const LinkID& bcLink ) { + + // decapsulate message + JoinRequest* joinReq = overlayMsg->decapsulate(); + logging_info( "Received join request for spovnet " << + joinReq->getSpoVNetID().toString() ); + + // check spovnet id + if( joinReq->getSpoVNetID() != spovnetId ) { + logging_error( + "Received join request for spovnet we don't handle " << + joinReq->getSpoVNetID().toString() ); + return false; + } + + // TODO: here you can implement mechanisms to deny joining of a node + bool allow = true; + logging_info( "Sending join reply for spovnet " << + spovnetId.toString() << " to node " << + overlayMsg->getSourceNode().toString() << + ". Result: " << (allow ? "allowed" : "denied") ); + joiningNodes.push_back( overlayMsg->getSourceNode() ); + + // return overlay parameters + assert( overlayInterface != NULL ); + logging_debug( "Using bootstrap end-point " + << getEndpointDescriptor().toString() ) + OverlayParameterSet parameters = overlayInterface->getParameters(); + OverlayMsg retmsg( OverlayMsg::typeJoinReply, + OverlayInterface::OVERLAY_SERVICE_ID, nodeId ); + JoinReply replyMsg( spovnetId, parameters, + allow, getEndpointDescriptor() ); + retmsg.encapsulate(&replyMsg); + bc->sendMessage( bcLink, &retmsg ); + + return true; +} + +/// Handle replies to spovnet instance join requests +bool BaseOverlay::handleJoinReply( OverlayMsg* overlayMsg, const LinkID& bcLink ) { + // decapsulate message + logging_debug("received join reply message"); + JoinReply* replyMsg = overlayMsg->decapsulate(); + + // correct spovnet? + if( replyMsg->getSpoVNetID() != spovnetId ) { // no-> fail + logging_error( "Received SpoVNet join reply for " << + replyMsg->getSpoVNetID().toString() << + " != " << spovnetId.toString() ); + delete replyMsg; + return false; + } + + // access granted? no -> fail + if( !replyMsg->getJoinAllowed() ) { + logging_error( "Our join request has been denied" ); + + // drop initiator link + if( !bcLink.isUnspecified() ){ + bc->dropLink( bcLink ); + + vector::iterator it = std::find( + bootstrapLinks.begin(), bootstrapLinks.end(), bcLink); + if( it != bootstrapLinks.end() ) + bootstrapLinks.erase(it); + } + + // inform all registered services of the event + BOOST_FOREACH( NodeListener* i, nodeListeners ) + i->onJoinFailed( spovnetId ); + + delete replyMsg; + return true; + } + + // access has been granted -> continue! + logging_info("Join request has been accepted for spovnet " << + spovnetId.toString() ); + + logging_debug( "Using bootstrap end-point " + << replyMsg->getBootstrapEndpoint().toString() ); + + // create overlay structure from spovnet parameter set + // if we have not boostrapped yet against some other node + if( overlayInterface == NULL ){ + + logging_debug("first-time bootstrapping"); + + overlayInterface = OverlayFactory::create( + *this, replyMsg->getParam(), nodeId, this ); + + // overlay structure supported? no-> fail! + if( overlayInterface == NULL ) { + logging_error( "overlay structure not supported" ); + + if( !bcLink.isUnspecified() ){ + bc->dropLink( bcLink ); + + vector::iterator it = std::find( + bootstrapLinks.begin(), bootstrapLinks.end(), bcLink); + if( it != bootstrapLinks.end() ) + bootstrapLinks.erase(it); } - // TODO: here you can implement mechanisms to deny joining of a node - bool allow = true; - logging_info( "Sending join reply for spovnet " << - spovnetId.toString() << " to node " << - overlayMsg->getSourceNode().toString() << - ". Result: " << (allow ? "allowed" : "denied") ); - joiningNodes.push_back( overlayMsg->getSourceNode() ); - - // return overlay parameters - assert( overlayInterface != NULL ); - logging_debug( "Using bootstrap end-point " - << getEndpointDescriptor().toString() ) - OverlayParameterSet parameters = overlayInterface->getParameters(); - OverlayMsg retmsg( OverlayMsg::typeJoinReply, nodeId ); - JoinReply replyMsg( spovnetId, parameters, - allow, getEndpointDescriptor() ); - retmsg.encapsulate(&replyMsg); - bc->sendMessage( bcLink, &retmsg ); - ret = true; - break; - } - - // --------------------------------------------------------------------- - // handle replies to spovnet instance join requests - // --------------------------------------------------------------------- - case OverlayMsg::typeJoinReply: { - - // decapsulate message - logging_debug("received join reply message"); - JoinReply* replyMsg = overlayMsg->decapsulate(); - - // correct spovnet? - if( replyMsg->getSpoVNetID() != spovnetId ) { // no-> fail - logging_error( "Received SpoVNet join reply for " << - replyMsg->getSpoVNetID().toString() << - " != " << spovnetId.toString() ); - ret = false; - delete replyMsg; - break; - } - - // access granted? no -> fail - if( !replyMsg->getJoinAllowed() ) { - logging_error( "Our join request has been denied" ); - - // drop initiator link - if(bcLink != LinkID::UNSPECIFIED){ - bc->dropLink( bcLink ); - - vector::iterator it = std::find( - bootstrapLinks.begin(), bootstrapLinks.end(), bcLink); - if( it != bootstrapLinks.end() ) - bootstrapLinks.erase(it); - } - - // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ) - i->onJoinFailed( spovnetId ); - - ret = true; - delete replyMsg; - break; - } - - // access has been granted -> continue! - logging_info("Join request has been accepted for spovnet " << - spovnetId.toString() ); - - logging_debug( "Using bootstrap end-point " - << replyMsg->getBootstrapEndpoint().toString() ); - - // - // create overlay structure from spovnet parameter set - // if we have not boostrapped yet against some other node - // - - if( overlayInterface == NULL ){ - - logging_debug("first-time bootstrapping"); - - overlayInterface = OverlayFactory::create( - *this, replyMsg->getParam(), nodeId, this ); - - // overlay structure supported? no-> fail! - if( overlayInterface == NULL ) { - logging_error( "overlay structure not supported" ); - - if(bcLink != LinkID::UNSPECIFIED){ - bc->dropLink( bcLink ); - - vector::iterator it = std::find( - bootstrapLinks.begin(), bootstrapLinks.end(), bcLink); - if( it != bootstrapLinks.end() ) - bootstrapLinks.erase(it); - } - + // inform all registered services of the event + BOOST_FOREACH( NodeListener* i, nodeListeners ) + i->onJoinFailed( spovnetId ); + + delete replyMsg; + return true; + } + + // everything ok-> join the overlay! + state = BaseOverlayStateCompleted; + overlayInterface->createOverlay(); + + overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() ); + overlayBootstrap.recordJoin( replyMsg->getBootstrapEndpoint() ); + + // update ovlvis + //ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN); + + // inform all registered services of the event + BOOST_FOREACH( NodeListener* i, nodeListeners ) + i->onJoinCompleted( spovnetId ); + + delete replyMsg; + +<<<<<<< .working // inform all registered services of the event BOOST_FOREACH( NodeListener* i, nodeListeners ) i->onJoinFailed( spovnetId ); - - delete replyMsg; - ret = true; - break; - } - - // everything ok-> join the overlay! - state = BaseOverlayStateCompleted; - overlayInterface->createOverlay(); - overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() ); - - // update ovlvis - //ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN); - - // inform all registered services of the event - BOOST_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() ); - - } // if( overlayInterface == NULL ) - +======= + } else { +>>>>>>> .merge-rechts.r5869 + + // 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; + } // if( overlayInterface == NULL ) + + return true; +} + + +<<<<<<< .working +======= +bool BaseOverlay::handleData( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { + // get service + const ServiceID& service = overlayMsg->getService(); + logging_debug( "Received data for service " << service.toString() + << " on link " << overlayMsg->getDestinationLink().toString() ); + +>>>>>>> .merge-rechts.r5869 + // delegate data message + getListener(service)->onMessage( + overlayMsg, + overlayMsg->getSourceNode(), + overlayMsg->getDestinationLink() + ); + + return true; +} + +<<<<<<< .working +======= + +>>>>>>> .merge-rechts.r5869 +bool BaseOverlay::handleLinkUpdate( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { + +<<<<<<< .working //record bootstrap ep as good endpoint to join overlayBootstrap.recordJoin( replyMsg->getBootstrapEndpoint() ); @@ -1132,272 +1280,230 @@ break; } - - // --------------------------------------------------------------------- - // handle data forward messages - // --------------------------------------------------------------------- - case OverlayMsg::typeData: { - - // get service - const ServiceID& service = overlayMsg->getService(); - logging_debug( "received data for service " << service.toString() ); - - // find listener - CommunicationListener* listener = - communicationListeners.get( service ); - if( listener == NULL ) { - ret = true; - break; - } - +======= + if( ld == NULL ) { + logging_warn( "received overlay update message for link for " + << "which we have no mapping" ); + return false; + } + logging_info("Received type update message on link " << ld ); +>>>>>>> .merge-rechts.r5869 + + // update our link mapping information for this link + bool changed = + ( ld->remoteNode != overlayMsg->getSourceNode() ) + || ( ld->service != overlayMsg->getService() ); + + // set parameters + ld->up = true; + ld->remoteNode = overlayMsg->getSourceNode(); + ld->remoteLink = overlayMsg->getSourceLink(); + ld->service = overlayMsg->getService(); + ld->autolink = overlayMsg->isAutoLink(); + + // if our link information changed, we send out an update, too + if( changed ) { + overlayMsg->swapRoles(); + overlayMsg->setSourceNode(nodeId); + overlayMsg->setSourceLink(ld->overlayId); + overlayMsg->setService(ld->service); + send( overlayMsg, ld ); + } + +<<<<<<< .working // delegate data message listener->onMessage( overlayMsg, overlayMsg->getSourceNode(), ld->overlayId ); - - ret = true; - break; - } - - // --------------------------------------------------------------------- - // handle update messages for link establishment - // --------------------------------------------------------------------- - case OverlayMsg::typeUpdate: { - // get info - const NodeID& sourcenode = overlayMsg->getSourceNode(); - const ServiceID& service = overlayMsg->getService(); - - // no link descriptor available -> error! - if( ld == NULL ) { - logging_warn( "received overlay update message for link for " - << "which we have no mapping" ); - ret = false; - break; - } - logging_debug("Received type update message on link " << ld ); - - // update our link mapping information for this link - bool changed = - ( ld->remoteNode != sourcenode ) || ( ld->service != service ); - ld->remoteNode = sourcenode; - ld->service = service; - ld->autolink = overlayMsg->isAutoLink(); - - // if our link information changed, we send out an update, too - if( changed ) { - OverlayMsg overMsg( OverlayMsg::typeUpdate, ld->service, nodeId ); - overMsg.setAutoLink(ld->autolink); - bc->sendMessage( ld->communicationId, &overMsg ); - } - - // service registered? no-> error! - if( !communicationListeners.contains( service ) ) { - logging_warn( "Link up: event listener has not been registered" ); - ret = false; - break; - } - - // default or no service registered? - CommunicationListener* listener = communicationListeners.get( service ); - if( listener == NULL || listener == &CommunicationListener::DEFAULT ) { - logging_warn("Link up: event listener is default or null!" ); - ret = true; - break; - } - - // update descriptor - ld->listener = listener; - ld->markAsUsed(); - ld->markAlive(); - - // ask the service whether it wants to accept this link - if( !listener->onLinkRequest(sourcenode) ) { - - logging_debug("Link id=" << ld->overlayId.toString() << - " has been denied by service " << service.toString() << ", dropping link"); - - // prevent onLinkDown calls to the service - ld->listener = &CommunicationListener::DEFAULT; - - // drop the link - dropLink( ld->overlayId ); - ret = true; - break; - } - - // set link up - ld->up = true; - logging_debug( - "Link " << ld->overlayId.toString() - << " has been accepted by service " << service.toString() - << " and is now up" - ); - - // auto links: link has been accepted -> send queued messages - if( ld->messageQueue.size() > 0 ) { - logging_info( "sending out queued messages on link " << - ld->overlayId.toString() ); - BOOST_FOREACH( Message* msg, ld->messageQueue ) { - sendMessage( msg, ld->overlayId ); - delete msg; - } - ld->messageQueue.clear(); - } - - // call the notification functions - listener->onLinkUp( ld->overlayId, sourcenode ); - sideport->onLinkUp( ld->overlayId, nodeId, sourcenode, this->spovnetId ); - - ret = true; - break; - } - - // --------------------------------------------------------------------- - // handle link request forwarded through the overlay - // --------------------------------------------------------------------- - case OverlayMsg::typeLinkRequest: { - - logging_debug( "received link request on link" ); - - // decapsulate message - LinkRequest* linkReq = overlayMsg->decapsulate(); - const ServiceID& service = overlayMsg->getService(); - - // is request reply? - if ( linkReq->isReply() ) { - - // find link - PendingLinkMap::iterator i = pendingLinks.find( linkReq->getNonce() ); - if ( i == pendingLinks.end() ) { - logging_error( "Nonce not found in link request" ); - ret = true; - break; - } - - // debug message - logging_debug( "Link request reply received. Establishing link " - << i->second << " to " << (linkReq->getEndpoint()->toString()) - << " for service " << service.toString() - << " with nonce " << linkReq->getNonce() - << " using relay " << linkReq->getRelay().toString() - << " and remote link id=" << linkReq->getRemoteLinkId() - ); - - // get descriptor - LinkDescriptor* ldn = getDescriptor(i->second); - if (ldn==NULL) { - delete linkReq; - ret = true; - break; - } - - // check if link request reply has a relay node ... - if (!linkReq->getRelay().isUnspecified()) { // yes-> - ldn->up = true; - ldn->relay = true; - if (ldn->localRelay.isUnspecified()) { - logging_error("On LinkRequest reply: local relay is unspecifed on link " << ldn ); - showLinkState(); - } - ldn->remoteRelay = linkReq->getRelay(); - ldn->remoteLinkId = linkReq->getRemoteLinkId(); - ldn->remoteNode = overlayMsg->getSourceNode(); - - ldn->markAlive(); - - // compile and send update message - OverlayMsg _overlayMsg( OverlayMsg::typeUpdate, ldn->service, nodeId ); - _overlayMsg.setAutoLink(ldn->autolink); - sendMessage( &_overlayMsg, ldn ); - - // auto links: link has been accepted -> send queued messages - if( ldn->messageQueue.size() > 0 ) { - logging_info( "Sending out queued messages on link " << - ldn->overlayId.toString() ); - BOOST_FOREACH( Message* msg, ldn->messageQueue ) { - sendMessage( msg, ldn->overlayId ); - delete msg; - } - ldn->messageQueue.clear(); - } - - ldn->listener->onLinkUp( ldn->overlayId, ldn->remoteNode ); - - // try to establish a direct link - ldn->communicationId = - bc->establishLink( *linkReq->getEndpoint(), i->second ); - } - - // no relay node-> use overlay routing - else { - ldn->up = true; - - // establish direct link - ldn->communicationId = - bc->establishLink( *linkReq->getEndpoint(), i->second ); - } - } else { - logging_debug( "Link request received from node id=" - << overlayMsg->getSourceNode() ); - - // create link descriptor - LinkDescriptor* ldn = - createLinkDescriptor(overlayMsg->getSourceNode(), - overlayMsg->getService(), LinkID::UNSPECIFIED ); - assert(!ldn->overlayId.isUnspecified()); - - // create reply message - OverlayMsg overlay_msg( OverlayMsg::typeLinkRequest, service, nodeId ); - LinkRequest link_request_msg( - linkReq->getNonce(), - &bc->getEndpointDescriptor(), - true, ldn->overlayId, ldn->localRelay - ); - overlay_msg.encapsulate( &link_request_msg ); - - // debug message - logging_debug( "Sending LinkRequest reply for link with nonce " << - linkReq->getNonce() ); - - // if this is a relay link-> update information & inform listeners - if (!linkReq->getRelay().isUnspecified()) { - // set flags - ldn->up = true; - ldn->relay = true; - if (ldn->localRelay.isUnspecified()) { - logging_error("On LinkRequest request: local relay is unspecifed on link " << ldn ); - showLinkState(); - } - ldn->remoteRelay = linkReq->getRelay(); - ldn->remoteNode = overlayMsg->getSourceNode(); - ldn->remoteLinkId = linkReq->getRemoteLinkId(); - ldn->listener->onLinkUp( ldn->overlayId, ldn->remoteNode ); - } - - // route message back over overlay - sendMessage( &overlay_msg, ldn ); - } - delete linkReq; - ret = true; - break; - } - - // --------------------------------------------------------------------- - // handle relay message to forward messages - // --------------------------------------------------------------------- - case OverlayMsg::typeRelay: { - - logging_debug( "received relay request on link" ); - - // decapsulate message - RelayMessage* relayMsg = overlayMsg->decapsulate(); - - // is relay message informative? - switch (relayMsg->getType()) { - - // handle relay notification - case RelayMessage::typeInform: { - logging_info("Received relay information message with" - << " relay " << relayMsg->getRelayNode() - << " destination " << relayMsg->getDestNode() ); - +======= + // service registered? no-> error! + if( !communicationListeners.contains( ld->service ) ) { + logging_warn( "Link up: event listener has not been registered" ); + return false; + } +>>>>>>> .merge-rechts.r5869 + + // default or no service registered? + CommunicationListener* listener = communicationListeners.get( ld->service ); + if( listener == NULL || listener == &CommunicationListener::DEFAULT ) { + logging_warn("Link up: event listener is default or null!" ); + return true; + } + + // update descriptor + ld->listener = listener; + ld->setAutoUsed(); + ld->setAlive(); + + // ask the service whether it wants to accept this link + if( !listener->onLinkRequest(ld->remoteNode) ) { + + logging_debug("Link id=" << ld->overlayId.toString() << + " has been denied by service " << ld->service.toString() << ", dropping link"); + + // prevent onLinkDown calls to the service + ld->listener = &CommunicationListener::DEFAULT; + + // drop the link + dropLink( ld->overlayId ); + return true; + } + + // set link up + ld->up = true; + logging_info( "Link has been accepted by service and is up: " << ld ); + + // auto links: link has been accepted -> send queued messages + 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; + } + ld->messageQueue.clear(); + } + + // call the notification functions + listener->onLinkUp( ld->overlayId, ld->remoteNode ); + sideport->onLinkUp( ld->overlayId, nodeId, ld->remoteNode, this->spovnetId ); + + return true; +} + +/// 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() ... + + // create link descriptor + LinkDescriptor* ldn = addDescriptor(); + + // flags + ldn->up = true; + ldn->fromRemote = true; + ldn->relayed = true; + + // parameters + ldn->service = overlayMsg->getService(); + ldn->listener = getListener(ldn->service); + ldn->remoteNode = overlayMsg->getSourceNode(); + ldn->remoteLink = overlayMsg->getSourceLink(); + + // update time-stamps + ldn->setAlive(); + ldn->setAutoUsed(); + + // 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 + + // inform listener + ldn->listener->onLinkUp( ldn->overlayId, ldn->remoteNode ); + + return true; +} + +bool BaseOverlay::handleLinkReply( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { + + // find link request + LinkDescriptor* ldn = getDescriptor(overlayMsg->getDestinationLink()); + + // not found? yes-> drop with error! + if (ldn == NULL) { + logging_error( "No link request pending for " + << overlayMsg->getDestinationLink().toString() ); + return false; + } + logging_debug("Handling link reply for " << ldn ) + + // check if already up + if (ldn->up) { + logging_warn( "Link already up: " << ldn ); + return true; + } + + // debug message + logging_debug( "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() + ); + + // set local link descriptor data + ldn->up = true; + ldn->relayed = true; + ldn->service = overlayMsg->getService(); + ldn->listener = getListener(ldn->service); + ldn->remoteLink = overlayMsg->getSourceLink(); + ldn->remoteNode = overlayMsg->getSourceNode(); + + // update timestamps + ldn->setAlive(); + ldn->setAutoUsed(); + + // auto links: link has been accepted -> send queued messages + if( ldn->messageQueue.size() > 0 ) { + logging_info( "Sending out queued messages on link " << + ldn->overlayId.toString() ); + BOOST_FOREACH( Message* msg, ldn->messageQueue ) { + sendMessage( msg, ldn->overlayId ); + delete msg; + } + ldn->messageQueue.clear(); + } + + // inform listeners about new link + ldn->listener->onLinkUp( ldn->overlayId, ldn->remoteNode ); + + // try to replace relay link with direct link + ldn->communicationId = + bc->establishLink( overlayMsg->getSourceEndpoint() ); + + return true; +} + +<<<<<<< .working +======= +/// handle a keep-alive message for a link +bool BaseOverlay::handleLinkAlive( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { + LinkDescriptor* rld = getDescriptor(overlayMsg->getDestinationLink()); + if ( rld != NULL ) { + logging_debug("Keep-Alive for " << + overlayMsg->getDestinationLink() ); + if (overlayMsg->isRouteRecord()) + rld->routeRecord = overlayMsg->getRouteRecord(); + rld->setAlive(); + return true; + } else { + logging_error("Keep-Alive for " + << overlayMsg->getDestinationLink() << ": link unknown." ); + return false; + } +} + +/// handle a direct link message +bool BaseOverlay::handleLinkDirect( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { + logging_debug( "Received direct link replacement request" ); + +>>>>>>> .merge-rechts.r5869 + /// get destination overlay link + LinkDescriptor* rld = getDescriptor( overlayMsg->getDestinationLink() ); + if (rld == NULL || ld == NULL) { + logging_error("Direct link replacement: Link " + << overlayMsg->getDestinationLink() << "not found error." ); + return false; + } + logging_info( "Received direct link convert notification for " << rld ); + + // update information + rld->communicationId = ld->communicationId; + rld->communicationUp = true; + rld->relayed = false; + +<<<<<<< .working // mark incoming link as relay if (ld!=NULL) ld->markAsRelay(); @@ -1417,23 +1523,20 @@ _relayMsg.setType( RelayMessage::typeRoute ); _overMsg.encapsulate( &_relayMsg ); - - // forward message - if (relayMsg->getRelayNode() == nodeId || relayMsg->getRelayNode().isUnspecified()) { - logging_info("Routing relay message to " << relayMsg->getDestNode().toString() ); - sendOverlay( &_overMsg, relayMsg->getDestNode() ); - } else { - logging_info("Routing relay message to " << relayMsg->getRelayNode().toString() ); - sendOverlay( &_overMsg, relayMsg->getRelayNode() ); - } - ret = true; - break; - } - - // handle relay routing - case RelayMessage::typeRoute: { - logging_info("Received relay route message with" - << " relay " << relayMsg->getRelayNode() - << " destination " << relayMsg->getDestNode() ); - +======= + // mark used and alive! + rld->setAlive(); + rld->setAutoUsed(); +>>>>>>> .merge-rechts.r5869 + + // erase the original descriptor + eraseDescriptor(ld->overlayId); +} + +/// handles an incoming message +bool BaseOverlay::handleMessage( const Message* message, LinkDescriptor* ld, + const LinkID bcLink ) { + logging_debug( "Handling message: " << message->toString()); + +<<<<<<< .working // mark incoming link as relay if (ld!=NULL) ld->markAsRelay(); @@ -1454,85 +1557,77 @@ RelayMessage _relayMsg( *relayMsg ); _overMsg.encapsulate(&_relayMsg); - - /// this must be handled by using relay link! - sendOverlay(&_overMsg, relayMsg->getDestNode()); - ret = true; - break; - } - - // error: I'm not a relay or destination! - logging_error("This node is neither relay nor destination. Dropping Message!"); - ret = true; - break; - } - default: { - logging_error("RelayMessage Unknown!"); - ret = true; - break; - } - } - delete relayMsg; - break; - } - - // --------------------------------------------------------------------- - // handle keep-alive messages - // --------------------------------------------------------------------- - case OverlayMsg::typeKeepAlive: { - logging_debug( "received keep-alive on link" ); - if ( ld != NULL ) { - logging_info("Keep-Alive for "<< ld->overlayId); - ld->markAlive(); - } - break; - } - - // --------------------------------------------------------------------- - // handle direct link replacement messages - // --------------------------------------------------------------------- - case OverlayMsg::typeDirectLink: { - - logging_debug( "Received direct link replacement request" ); - - LinkDescriptor* rld = getDescriptor( overlayMsg->getRelayLink() ); - if (rld == NULL || ld == NULL) { - logging_error("Direct link replacement: Link " - << overlayMsg->getRelayLink() << "not found error." ); - break; - } - logging_info( "Received direct link convert notification for " << rld ); - - // set communcation link id and set it up - rld->communicationId = ld->communicationId; - - // this is neccessary since this link was a relay link before! - rld->communicationUp = true; - - // this is not a relay link anymore! - rld->relay = false; - rld->localRelay = NodeID::UNSPECIFIED; - rld->remoteRelay = NodeID::UNSPECIFIED; - - // mark used and alive! - rld->markAsUsed(); - rld->markAlive(); - - // erase the original descriptor - eraseDescriptor(ld->overlayId); - break; - } - - // --------------------------------------------------------------------- +======= + // decapsulate overlay message + OverlayMsg* overlayMsg = + const_cast(message)->decapsulate(); + if( overlayMsg == NULL ) return false; +>>>>>>> .merge-rechts.r5869 + + // refresh relay information + refreshRelayInformation( overlayMsg, ld ); + + // increase number of hops + overlayMsg->increaseNumHops(); + + // update route record + overlayMsg->addRouteRecord(nodeId); + + // handle signaling messages (do not route!) + if (overlayMsg->getType()>=OverlayMsg::typeSignalingStart && + overlayMsg->getType()<=OverlayMsg::typeSignalingEnd ) { + overlayInterface->onMessage(overlayMsg, NodeID::UNSPECIFIED, LinkID::UNSPECIFIED); + delete overlayMsg; + return true; + } + + // message for reached destination? no-> route message + if (!overlayMsg->getDestinationNode().isUnspecified() && + overlayMsg->getDestinationNode() != nodeId ) { + logging_debug("Routing message " + << " from " << overlayMsg->getSourceNode() + << " to " << overlayMsg->getDestinationNode() + ); + + route( overlayMsg, ld ); + delete overlayMsg; + return true; + } + + // 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() ); + "what to do with this message of type " << overlayMsg->getType() ); ret = false; break; } - } /* switch */ - + } + + // free overlay message and return value delete overlayMsg; return ret; @@ -1554,12 +1649,11 @@ } +/// return the overlay neighbors vector BaseOverlay::getOverlayNeighbors(bool deep) const { vector nodes = overlayInterface->getKnownNodes(deep); - // the known nodes _can_ also include our node, so we remove ourself vector::iterator i = find( nodes.begin(), nodes.end(), this->nodeId ); if( i != nodes.end() ) nodes.erase( i ); - return nodes; } @@ -1594,4 +1688,5 @@ void BaseOverlay::eventFunction() { +<<<<<<< .working // send keep-alive messages over established links @@ -1673,14 +1768,8 @@ if (counter>=4) showLinkState(); if (counter>=4 || counter<0) counter = 0; -} - -void BaseOverlay::showLinkState() { - int i=0; - logging_info("--- link state -------------------------------"); - BOOST_FOREACH( LinkDescriptor* ld, links ) { - logging_info("link " << i << ": " << ld); - i++; - } - logging_info("----------------------------------------------"); +======= + stabilizeRelays(); + stabilizeLinks(); +>>>>>>> .merge-rechts.r5869 } Index: /source/ariba/overlay/BaseOverlay.h =================================================================== --- /source/ariba/overlay/BaseOverlay.h (revision 5860) +++ /source/ariba/overlay/BaseOverlay.h (revision 5870) @@ -121,4 +121,5 @@ class LinkDescriptor; +class OverlayMsg; class BaseOverlay: public MessageReceiver, @@ -161,8 +162,6 @@ /// Tries to establish a direct or overlay link - const LinkID establishLink(const EndpointDescriptor& ep, - const NodeID& node, const ServiceID& service, - const NodeID& remoteRelay = NodeID::UNSPECIFIED, - const LinkID& linkid = LinkID::UNSPECIFIED); + const LinkID establishLink( const EndpointDescriptor& ep, + const NodeID& node, const ServiceID& service ); /** @@ -174,7 +173,6 @@ * @param linkid Link identifier to be used with this link */ - const LinkID establishLink(const NodeID& node, const ServiceID& service, - const NodeID& remoteRelay = NodeID::UNSPECIFIED, - const LinkID& linkid = LinkID::UNSPECIFIED); + const LinkID establishLink( const NodeID& remote, + const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID ); /** @@ -182,16 +180,16 @@ * endpoint and to the specified service */ - const LinkID establishDirectLink(const EndpointDescriptor& ep, - const ServiceID& service, const LinkID& linkid = LinkID::UNSPECIFIED); + const LinkID establishDirectLink( const EndpointDescriptor& endpoint, + const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID ); /// drops a link - void dropLink(const LinkID& link); + void dropLink( const LinkID& link ); /// sends a message over an existing link - seqnum_t sendMessage(const Message* message, const LinkID& link); + seqnum_t sendMessage(const Message* message, const LinkID& link ); /// sends a message to a node and a specific service - seqnum_t sendMessage(const Message* message, const NodeID& node, - const ServiceID& service); + seqnum_t sendMessage(const Message* message, const NodeID& remote, + const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID); /** @@ -207,6 +205,6 @@ * @return The end-point descriptor of the link's end-point */ - const EndpointDescriptor& getEndpointDescriptor(const LinkID& link = - LinkID::UNSPECIFIED) const; + const EndpointDescriptor& getEndpointDescriptor( + const LinkID& link = LinkID::UNSPECIFIED) const; /** @@ -332,37 +330,25 @@ * the node the message came from! */ - virtual bool receiveMessage(const Message* message, const LinkID& link, - const NodeID& source = NodeID::UNSPECIFIED); + virtual bool receiveMessage( const Message* message, const LinkID& link, + const NodeID& ); + + /** + * This method is called, when a new node joined the network + * + * @see OverlayStructureEvents.h + */ + virtual void onNodeJoin(const NodeID& node); + + /** + * Timer Event method + */ + virtual void eventFunction(); //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - /// handles an incoming message with link descriptor - bool handleMessage(const Message* message, - const LinkID& boLink, const LinkID& bcLink, const NodeID& remoteNode ); - - //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - /** - * This method is called, when a routed message arrives from the - * overlay. - * - * @see OverlayStructureEvents.h - */ - virtual void incomingRouteMessage(Message* msg, - const LinkID& link = LinkID::UNSPECIFIED, - const NodeID& source = NodeID::UNSPECIFIED); - - /** - * This method is called, when a new node joined the network - * - * @see OverlayStructureEvents.h - */ - virtual void onNodeJoin(const NodeID& node); - - /** - * Timer Event method - */ - virtual void eventFunction(); private: + /// is the base overlay started yet + bool started; /// The state of the BaseOverlay @@ -372,13 +358,14 @@ } BaseOverlayState; - BaseOverlayState state; ///< Current Base-Overlay state - BaseCommunication* bc; ///< reference to the base communication - NodeID nodeId; ///< the node id of this node - SpoVNetID spovnetId; ///< the spovnet id of the currently joined overlay - vector bootstrapLinks; ///< the link id of the link to the initiator node - NodeID spovnetInitiator; ///< The initiator node + BaseOverlayState state; ///< Current Base-Overlay state + BaseCommunication* bc; ///< reference to the base communication + NodeID nodeId; ///< the node id of this node + SpoVNetID spovnetId; ///< the spovnet id of the currently joined overlay + vector bootstrapLinks; ///< the link id of the link to the initiator node + NodeID spovnetInitiator; ///< The initiator node /// the service id communication listeners Demultiplexer communicationListeners; + CommunicationListener* getListener( const ServiceID& id ); /// the node listeners @@ -392,6 +379,37 @@ OverlayInterface* overlayInterface; + /// Bootstrapper for our spovnet + OverlayBootstrap overlayBootstrap; + + // message handlers -------------------------------------------------------- + + /// demultiplexes a incoming message with link descriptor + bool handleMessage( const Message* message, LinkDescriptor* ld, + const LinkID bcLink = LinkID::UNSPECIFIED ); + + // handle data and signalling messages + bool handleData( OverlayMsg* msg, LinkDescriptor* ld ); + 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 ); + + // handle link messages + bool handleLinkRequest( OverlayMsg* msg, LinkDescriptor* ld ); + bool handleLinkReply( OverlayMsg* msg, LinkDescriptor* ld ); + bool handleLinkUpdate( OverlayMsg* msg, LinkDescriptor* ld ); + bool handleLinkDirect( OverlayMsg* msg, LinkDescriptor* ld ); + bool handleLinkAlive( OverlayMsg* msg, LinkDescriptor* ld ); + + // link state handling ----------------------------------------------------- + + /// link state information counter + int counter; + /// The link mapping of the node vector links; + + /// erases a link descriptor void eraseDescriptor(const LinkID& link, bool communication = false); @@ -410,29 +428,60 @@ LinkDescriptor* addDescriptor(const LinkID& link = LinkID::UNSPECIFIED); - /// returns a direct link relay descriptor to the given relay node - LinkDescriptor* getRelayDescriptor( const NodeID& remoteNode ); - - /// returns the local relay node to a given remote node - NodeID getRelayNode( const NodeID& remoteNode ); - - /// returns the direct link the message to a neighbor is send to - LinkDescriptor* getSendDescriptor( const NodeID& nodeid, bool follow = true ); - - /// routes a message over the overlay or directly sends it when a link is open - seqnum_t sendOverlay( Message* message, const NodeID& nodeid, - const NodeID& remoteRelay = NodeID::UNSPECIFIED ); - - /// forwards a message over relays/overlay/directly using link descriptor - seqnum_t sendMessage( Message* message, const LinkDescriptor* ld ); - - /// creates a link descriptor, applys relay semantics if possible - LinkDescriptor* createLinkDescriptor( - const NodeID remoteNode, const ServiceID service, const LinkID link_id ); - - // map of a link request map a nonce to a LinkID - typedef map PendingLinkMap; - PendingLinkMap pendingLinks; - - void showLinkState(); + /// stabilizes link information + void stabilizeLinks(); + + /// print the currently known links + void showLinks(); + + // relay route management -------------------------------------------------- + + /// relay route definitions + class relay_route { + public: + NodeID node; + LinkID link; + uint8_t hops; + time_t used; + }; + vector relay_routes; + + /// stabilize relay information + void stabilizeRelays(); + + /// refreshes relay information + void refreshRelayInformation( const OverlayMsg* message, LinkDescriptor* ld ); + + /// returns a known relay link + LinkDescriptor* getRelayLinkTo( const NodeID& remote ); + + /// removes relay link information + void removeRelayLink( const LinkID& link ); + + /// removes relay node information + void removeRelayNode( const NodeID& link ); + + // internal message delivery ----------------------------------------------- + + /// routes a message to its destination node + void route( OverlayMsg* message, LinkDescriptor* incomingLink ); + + /// sends a raw message to another node, delivers it to the base overlay class + seqnum_t send( OverlayMsg* message, const NodeID& destination ); + + /// 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 ); + + /// 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); + + /// 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 ); + + // misc -------------------------------------------------------------------- /** @@ -442,14 +491,4 @@ typedef vector JoiningNodes; JoiningNodes joiningNodes; - - int counter; - - /** - * Bootstrapper for our spovnet - */ - OverlayBootstrap overlayBootstrap; - - /// is the base overlay started yet - bool started; }; Index: /source/ariba/overlay/LinkDescriptor.h =================================================================== --- /source/ariba/overlay/LinkDescriptor.h (revision 5860) +++ /source/ariba/overlay/LinkDescriptor.h (revision 5870) @@ -39,20 +39,18 @@ // default values this->up = false; - this->dropWhenRelaysLeft = false; this->fromRemote = false; this->remoteNode = NodeID::UNSPECIFIED; - this->overlayId = LinkID::UNSPECIFIED; + this->overlayId = LinkID::create(); this->communicationUp = false; this->communicationId = LinkID::UNSPECIFIED; this->keepAliveTime = time(NULL); this->keepAliveMissed = 0; - this->usedAsRelay = false; - this->timeUsedAsRelay = time(NULL); + this->relaying = false; + this->timeRelaying = time(NULL); + this->dropAfterRelaying = false; this->service = ServiceID::UNSPECIFIED; this->listener = &CommunicationListener::DEFAULT; - this->relay = false; - this->localRelay = NodeID::UNSPECIFIED; - this->remoteRelay = NodeID::UNSPECIFIED; - this->remoteLinkId = LinkID::UNSPECIFIED; + this->relayed = false; + this->remoteLink = LinkID::UNSPECIFIED; this->autolink = false; this->lastuse = time(NULL); @@ -64,78 +62,77 @@ } - // general information about the link --------------------------------- + // general information about the link -------------------------------------- + bool up; ///< flag whether this link is up and running + bool fromRemote; ///< flag, whether this link was requested from remote + NodeID remoteNode; ///< remote end-point node - bool up; ///< flag wheter this link is up and running - bool dropWhenRelaysLeft; - bool fromRemote; /// routeRecord; - bool usedAsRelay; ///< flag, wheter this link has been used as relay - time_t timeUsedAsRelay; ///< last time the link has been used as relay - - void markAsRelay() { - usedAsRelay = true; - timeUsedAsRelay = time(NULL); + // relay state ------------------------------------------------------------- + bool relaying; ///< flag, wheter this link has been used as relay + bool dropAfterRelaying; + time_t timeRelaying; ///< last time the link has been used as relay + void setRelaying() { + relaying = true; + timeRelaying = time(NULL); } - // owner -------------------------------------------------------------- + // owner ------------------------------------------------------------------- ServiceID service; ///< service using this link CommunicationListener* listener; ///< the listener using this node - // relay information -------------------------------------------------- - bool relay; ///< flag whether this link is a relay link - NodeID localRelay; ///< the local relay node - NodeID remoteRelay; ///< the remote relay node - LinkID remoteLinkId; ///< the remote link id - - // auto links --------------------------------------------------------- + // auto links -------------------------------------------------------------- bool autolink; ///< flag, whether this link is a auto-link time_t lastuse; ///< time, when the link was last used deque messageQueue; ///< waiting messages to be delivered - - /// updates the timestamp - void markAsUsed() { - lastuse = time(NULL); + void setAutoUsed() { + if (autolink) lastuse = time(NULL); } - - /// drops waiting messsages + /// drops waiting auto-link messages void flushQueue() { - BOOST_FOREACH( Message* msg, messageQueue ) - delete msg; + BOOST_FOREACH( Message* msg, messageQueue ) delete msg; messageQueue.clear(); } + // string representation --------------------------------------------------- std::string to_string() const { std::ostringstream s; s << "up=" << up << " "; - s << "fromRem=" << fromRemote << " "; - s << "remNode=" << remoteNode.toString().substr(0,6) << " "; + s << "init=" << !fromRemote << " "; + s << "id=" << overlayId.toString().substr(0,4) << " "; s << "serv=" << service.toString() << " "; - s << "overId=" << overlayId.toString().substr(0,6) << " "; - s << "commUp=" << communicationUp << " "; - s << "commId=" << communicationId.toString().substr(0,6) << " "; - s << "usedAsRel=" << usedAsRelay << " "; - s << "KeepAliveMiss=" << keepAliveMissed << " "; - if ( !localRelay.isUnspecified() ) { - s << "locRel=" << localRelay.toString().substr(0,6) << " "; - s << "remRel=" << remoteRelay.toString().substr(0,6) << " "; - s << "remLink=" << remoteLinkId.toString().substr(0,6) << " "; + s << "node=" << remoteNode.toString().substr(0,4) << " "; + s << "relaying=" << relaying << " "; + s << "miss=" << keepAliveMissed << " "; + s << "auto=" << autolink << " "; + if ( relayed ) { + s << "| Relayed: "; + s << "remote link=" << remoteLink.toString().substr(0,4) << " "; + if (routeRecord.size()>0) { + cout << "route record="; + for (size_t i=0; i - */ -class LinkRequest : public Message { - VSERIALIZEABLE; -private: - bool free_endpoint_; - uint8_t flags; - uint32_t nonce; - const EndpointDescriptor* endpoint; - LinkID remoteLinkId; - NodeID relay; - -public: - LinkRequest() { - - } - - LinkRequest( uint32_t nonce, const EndpointDescriptor* endpoint, - bool reply = false, const LinkID& remoteLinkId = LinkID::UNSPECIFIED, - const NodeID& relay = NodeID::UNSPECIFIED ) : - flags(reply&1), nonce(nonce), endpoint(endpoint), remoteLinkId(remoteLinkId), relay(relay) { - } - - virtual ~LinkRequest(); - - const EndpointDescriptor* getEndpoint() const { - return endpoint; - } - - const LinkID& getRemoteLinkId() const { - return remoteLinkId; - } - - const NodeID& getRelay() const { - return relay; - } - - bool isReply() const { - return flags & 1; - } - - uint32_t getNonce() const { - return nonce; - } -}; - -}} // ariba::overlay - -sznBeginDefault( ariba::overlay::LinkRequest, X ) { - if (X.isDeserializer()) endpoint = new EndpointDescriptor(); - X && flags && nonce; - X && const_cast(endpoint); - X && &relay && &remoteLinkId; -} sznEnd(); - -#endif /* ARIBA_OVERLAY_LINKREQUEST_H_ */ Index: /source/ariba/overlay/messages/OverlayMsg.h =================================================================== --- /source/ariba/overlay/messages/OverlayMsg.h (revision 5860) +++ /source/ariba/overlay/messages/OverlayMsg.h (revision 5870) @@ -41,4 +41,5 @@ #include + #include "ariba/utility/messages.h" #include "ariba/utility/serialization.h" @@ -46,4 +47,9 @@ #include "ariba/utility/types/NodeID.h" #include "ariba/utility/types/LinkID.h" +#include "ariba/communication/EndpointDescriptor.h" + + +namespace ariba { +namespace overlay { using ariba::utility::LinkID; @@ -51,47 +57,65 @@ using ariba::utility::ServiceID; using ariba::utility::Message; - -namespace ariba { -namespace overlay { - -using_serialization -; - -class OverlayMsg: public Message { -VSERIALIZEABLE - ; +using ariba::communication::EndpointDescriptor; +using_serialization; + +/** + * A general purpose overlay message that is used to exchange messages + * between nodes. + * + * @author Sebastian Mies + */ +class OverlayMsg: public Message { VSERIALIZEABLE; public: - - /// (payload-) message types - enum type_ { - typeInvalid = 0, ///< invalid type (no encapsulated messages) - typeData = 1, ///< message contains data for higher layers - typeJoinRequest = 2, ///< join request - typeJoinReply = 3, ///< join reply - typeUpdate = 4, ///< update message for link association - typeLinkRequest = 5, ///< link request (sent over the overlay) - typeRelay = 6, ///< relay message - typeKeepAlive = 7, ///< a keep-alive message - typeDirectLink = 8, - ///< a direct connection has been established + /// message types, is: uint8_t + enum type_ { // is: uint8_t + typeInvalid = 0x00, ///< invalid, unspecified type + + // data transfer + maskTransfer = 0x10, ///< bit mask for transfer messages + typeData = 0x11, ///< message contains data for higher layers + + // join signaling + maskJoin = 0x20, ///< bit mask for join messages + typeJoinRequest = 0x21, ///< join request + typeJoinReply = 0x22, ///< join reply + + // link messages + maskLink = 0x30, ///< bit mask for link messages + typeLinkRequest = 0x31, ///< request a new link + typeLinkReply = 0x32, ///< link request reply + typeLinkUpdate = 0x33, ///< update message for link association + typeLinkDirect = 0x34, ///< direct connection has been established + typeLinkAlive = 0x35, ///< keep-alive message + + // topology signaling + typeSignalingStart = 0x80, ///< start of the signaling types + typeSignalingEnd = 0xFF ///< end of the signaling types }; /// default constructor - OverlayMsg(type_ type = typeInvalid, const ServiceID _service = - ServiceID::UNSPECIFIED, const NodeID _sourceNode = - NodeID::UNSPECIFIED) : - type((uint8_t) type), service(_service), sourceNode(_sourceNode), - relayLink(LinkID::UNSPECIFIED), autoLink(false) { - } - - OverlayMsg(const OverlayMsg& rhs) : - type(rhs.type), service(rhs.service), sourceNode(rhs.sourceNode), - relayLink(rhs.relayLink), autoLink(rhs.autoLink) { - } - - /// type and source node constructor - OverlayMsg(type_ type, const NodeID _sourceNode) : - type((uint8_t) type), service(ServiceID::UNSPECIFIED), sourceNode( - _sourceNode), relayLink(LinkID::UNSPECIFIED), autoLink(false) { + OverlayMsg( + uint8_t type = typeInvalid, + const ServiceID& _service = ServiceID::UNSPECIFIED, + const NodeID& _sourceNode = NodeID::UNSPECIFIED, + const NodeID& _destinationNode = NodeID::UNSPECIFIED, + const LinkID& _sourceLink = LinkID::UNSPECIFIED, + const LinkID& _destinationLink = LinkID::UNSPECIFIED ) + : type(type), flags(0), hops(0), ttl(25), + service(_service), + sourceNode(_sourceNode), destinationNode(_destinationNode), + sourceLink(_sourceLink), destinationLink(_destinationLink), + routeRecord() { + if (!_sourceLink.isUnspecified() || !_destinationLink.isUnspecified()) + setLinkMessage(true); + } + + // copy constructor + OverlayMsg(const OverlayMsg& rhs) + : type(rhs.type), flags(rhs.flags), hops(rhs.hops), ttl(rhs.ttl), + service(rhs.service), + sourceNode(rhs.sourceNode), destinationNode(rhs.destinationNode), + sourceLink(rhs.sourceLink), destinationLink(rhs.destinationLink), + routeRecord(rhs.routeRecord) { } @@ -99,48 +123,196 @@ ~OverlayMsg(); + /// type ------------------------------------------------------------------- + type_ getType() const { return (type_) type; } + void setType( type_ type ) { + this->type = type; + } + + bool hasTypeMask( type_ mask ) const { + return (type & (uint8_t)mask) == (uint8_t)mask; + } + + /// flags ------------------------------------------------------------------ + + bool isRelayed() const { + return (flags & 0x01)!=0; + } + + void setRelayed( bool relayed = true ) { + if (relayed) flags |= 1; else flags &= ~1; + } + + bool isRouteRecord() const { + return (flags & 0x02)!=0; + } + + void setRouteRecord( bool route_record = true ) { + if (route_record) flags |= 0x02; else flags &= ~0x02; + } + + bool isAutoLink() const { + return (flags & 0x80) == 0x80; + } + + void setAutoLink(bool auto_link = true ) { + if (auto_link) flags |= 0x80; else flags &= ~0x80; + } + + bool isLinkMessage() const { + return (flags & 0x40)!=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; + } + + /// number of hops and time to live ---------------------------------------- + + uint8_t getNumHops() const { + return hops; + } + + void setNumHops( uint8_t hops ) { + this->hops = hops; + } + + uint8_t increaseNumHops() { + hops++; + } + + uint8_t getTimeToLive() const { + return ttl; + } + + void setTimeToLive( uint8_t ttl ) { + this->ttl = ttl; + } + + /// addresses and links ---------------------------------------------------- + const ServiceID& getService() const { return service; } + void setService( const ServiceID& service ) { + this->service = service; + } + const NodeID& getSourceNode() const { return sourceNode; } - const LinkID& getRelayLink() const { - return relayLink; - } - - void setRelayLink(const LinkID& relayLink) { - this->relayLink = relayLink; - } - - const bool isAutoLink() const { - return autoLink; - } - - void setAutoLink(bool autoLink) { - this->autoLink = autoLink; - } + void setSourceNode( const NodeID& node ) { + this->sourceNode = node; + } + + const NodeID& getDestinationNode() const { + return destinationNode; + } + + void setDestinationNode( const NodeID& node ) { + this->destinationNode = node; + } + + const LinkID& getSourceLink() const { + return sourceLink; + } + + void setSourceLink( const LinkID& link ) { + this->sourceLink = link; + setLinkMessage(); + } + + const LinkID& getDestinationLink() const { + return destinationLink; + } + + void setDestinationLink( const LinkID& link ) { + this->destinationLink = link; + setLinkMessage(); + } + + void setSourceEndpoint( const EndpointDescriptor& endpoint ) { + sourceEndpoint = endpoint; + setContainsSourceEndpoint(true); + } + + const EndpointDescriptor& getSourceEndpoint() const { + return sourceEndpoint; + } + + /// swaps source and destination + void swapRoles() { + NodeID dummyNode = sourceNode; + sourceNode = destinationNode; + destinationNode = dummyNode; + LinkID dummyLink = sourceLink; + sourceLink = destinationLink; + destinationLink = dummyLink; + hops = 0; + } + + const vector getRouteRecord() const { + return routeRecord; + } + + void addRouteRecord( const NodeID& node ) { + if (isRouteRecord()) + routeRecord.push_back(node); + } + private: - uint8_t type; + uint8_t type, flags, hops, ttl; ServiceID service; NodeID sourceNode; - LinkID relayLink; - uint8_t autoLink; + NodeID destinationNode; + LinkID sourceLink; + LinkID destinationLink; + EndpointDescriptor sourceEndpoint; + vector routeRecord; }; -} -} // ariba::overlay - +}} // ariba::overlay + +/// serialization sznBeginDefault( ariba::overlay::OverlayMsg, X ){ -X && type && &service && &sourceNode; -if (type == typeDirectLink) X && &relayLink; -if (type == typeUpdate) X && autoLink; -X && Payload(); -}sznEnd(); + // header + X && type && flags && hops && ttl; + + // addresses + X && &service && &sourceNode && &destinationNode; + + // message is associated with a end-to-end link + if (isLinkMessage()) + X && &sourceLink && &destinationLink; + + // message is associated with a source end-point + if (containsSourceEndpoint()) + X && sourceEndpoint; + + // message should record its route + if (isRouteRecord()) { + uint8_t size = routeRecord.size(); + X && size; + if (X.isDeserializer()) routeRecord.resize(size); + for (uint8_t i=0;i - */ -class RelayMessage : public Message { - VSERIALIZEABLE; -private: - uint8_t type; - NodeID relayNode; - NodeID destNode; - LinkID destLink; - -public: - enum type_ { - typeInvalid = 0, - typeInform = 1, - typeRoute = 2 - }; - - /// contructs a unspecified relay message - RelayMessage() : - type(typeInvalid), relayNode(NodeID::UNSPECIFIED), destNode(NodeID::UNSPECIFIED) { - } - - RelayMessage( type_ type, const NodeID& relayNode, const NodeID& destNode, const LinkID& destLink = LinkID::UNSPECIFIED ) : - type(type), relayNode(relayNode), destNode(destNode), destLink(destLink) { - } - - ~RelayMessage(); - - /// returns the type of this message - type_ getType() const { - return (type_)type; - } - - /// sets the type of this message - void setType( type_ type ) { - this->type = type; - } - - /// returns the remote (destination) node id - const NodeID& getDestNode() const { - return destNode; - } - - const LinkID& getDestLink() const { - return destLink; - } - - /// returns the relay node for the destination - const NodeID& getRelayNode() const { - return relayNode; - } -}; - -}} // ariba::overlay - -sznBeginDefault( ariba::overlay::RelayMessage, X ) { - X && type && &relayNode && &destNode && &destLink && Payload(); -} sznEnd(); - -#endif /* RELAYMESSAGE_H_ */ Index: /source/ariba/overlay/modules/OverlayInterface.h =================================================================== --- /source/ariba/overlay/modules/OverlayInterface.h (revision 5860) +++ /source/ariba/overlay/modules/OverlayInterface.h (revision 5870) @@ -74,10 +74,7 @@ * Constructs a new overlay. */ - OverlayInterface( - BaseOverlay& _baseoverlay, - const NodeID& _nodeid, - OverlayStructureEvents* _eventsReceiver, - OverlayParameterSet _parameters - ); + OverlayInterface( BaseOverlay& _baseoverlay, const NodeID& _nodeid, + OverlayStructureEvents* _eventsReceiver, OverlayParameterSet _parameters + ); /** @@ -103,5 +100,6 @@ * end-point, if this node is the initiator */ - virtual void joinOverlay(const EndpointDescriptor& bootstrap = EndpointDescriptor::UNSPECIFIED()) = 0; + virtual void joinOverlay( + const EndpointDescriptor& bootstrap = EndpointDescriptor::UNSPECIFIED()) = 0; /** @@ -117,24 +115,4 @@ */ virtual const EndpointDescriptor& resolveNode(const NodeID& node) = 0; - - /** - * Routes a message to a given node by using overlay routing. - * - * @param destnode The destination node. - * @param msg The message to be routed. - */ - virtual void routeMessage(const NodeID& destnode, Message* msg) = 0; - - /** - * Routes a message to a given node by using an existing link. - * - * TODO: This is a hack. This method allows the BaseOverlay class to - * use overlay signaling links to transfer data for relaying - * - * @param node The destination node. - * @param link An established link - * @param msg The message to be sent. - */ - virtual void routeMessage(const NodeID& node, const LinkID& link, Message* msg) = 0; /** @@ -183,5 +161,4 @@ const LinkID& lnk = LinkID::UNSPECIFIED); - const OverlayParameterSet& getParameters() const; Index: /source/ariba/overlay/modules/chord/Chord.cpp =================================================================== --- /source/ariba/overlay/modules/chord/Chord.cpp (revision 5860) +++ /source/ariba/overlay/modules/chord/Chord.cpp (revision 5870) @@ -38,13 +38,18 @@ #include "ariba/overlay/BaseOverlay.h" +#include "ariba/overlay/messages/OverlayMsg.h" #include "Chord.h" -#include "messages/ChordMessage.h" +#include "detail/chord_routing_table.hpp" + #include "messages/Discovery.h" - -#include "detail/chord_routing_table.hpp" namespace ariba { namespace overlay { + +enum signalMessageTypes { + typeDiscovery = OverlayMsg::typeSignalingStart + 0x01, + typeLeave = OverlayMsg::typeSignalingStart + 0x02, +}; typedef chord_routing_table::item route_item; @@ -57,5 +62,9 @@ // create routing table +<<<<<<< .working this->table = new chord_routing_table(_nodeid, 2); +======= + this->table = new chord_routing_table(_nodeid, 4); +>>>>>>> .merge-rechts.r5869 orphan_removal_counter = 0; discovery_count = 0; @@ -71,71 +80,81 @@ /// helper: sets up a link using the base overlay -LinkID Chord::setup(const EndpointDescriptor& endp, const NodeID& node, const NodeID& remoteRelay ) { - logging_debug("Request to setup link to " << endp.toString() ); +LinkID Chord::setup(const EndpointDescriptor& endpoint, const NodeID& remote ) { // check if we already have a connection for (int i=0; isize(); i++) - if ((*table)[i]->id == node && !((*table)[i]->info.isUnspecified())) + if ((*table)[i]->ref_count > 0 && (*table)[i]->id == remote && !((*table)[i]->info.isUnspecified())) return LinkID::UNSPECIFIED; // check if we are already trying to establish a link for (size_t i=0; i>>>>>> .merge-rechts.r5869 } /// helper: sends a message using the "base overlay" -seqnum_t Chord::send(Message* msg, const LinkID& link) { +seqnum_t Chord::send( OverlayMsg* msg, const LinkID& link ) { if (link.isUnspecified()) return 0; - return baseoverlay.sendMessage(msg, link); + msg->setRelayed(true); + return baseoverlay.send_link( msg, link ); } /// sends a discovery message -void Chord::send_discovery_to(const NodeID& destination, int ttl) { +void Chord::send_discovery_to(const NodeID& remote, int ttl) { + LinkID link = getNextLinkId(remote); + if ( remote == nodeid || link.isUnspecified()) return; if ( table->size() == 0 ) return; - ChordMessage cmsg(ChordMessage::discovery, nodeid, destination); - Discovery dmsg; - dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor()); - dmsg.setSourceRelay(baseoverlay.getRelayNode(destination)); - dmsg.setFollowType(Discovery::normal); - dmsg.setTTL((uint8_t) ttl); - cmsg.encapsulate(&dmsg); - + OverlayMsg msg( typeDiscovery ); + msg.setRelayed(true); + Discovery dmsg( Discovery::normal, (uint8_t)2, baseoverlay.getEndpointDescriptor() ); + msg.encapsulate(&dmsg); + +<<<<<<< .working // get next hop const route_item* item = table->get_next_hop(destination); if (item!=NULL && !item->info.isUnspecified()) send(&cmsg, item->info); -} - -void Chord::discover_neighbors( const LinkID& lnk ) { +======= + // send to node + baseoverlay.send_node( &msg, remote ); +>>>>>>> .merge-rechts.r5869 +} + +void Chord::discover_neighbors( const LinkID& link ) { + uint8_t ttl = 2; + { + // send predecessor discovery + OverlayMsg msg( typeDiscovery ); + msg.setRelayed(true); + Discovery dmsg( Discovery::predecessor, ttl, + baseoverlay.getEndpointDescriptor() ); + msg.encapsulate(&dmsg); + send(&msg, link); + } { // send successor discovery - ChordMessage cmsg(ChordMessage::discovery, nodeid, nodeid); - Discovery dmsg; - dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor()); - dmsg.setSourceRelay(baseoverlay.getRelayNode(nodeid)); - dmsg.setFollowType(Discovery::successor); - dmsg.setTTL((uint8_t)4); - cmsg.encapsulate(&dmsg); - send(&cmsg, lnk); - } - { - // send predecessor discovery - ChordMessage cmsg(ChordMessage::discovery, nodeid, nodeid); - Discovery dmsg; - dmsg.setSourceEndpoint(&baseoverlay.getEndpointDescriptor()); - dmsg.setSourceRelay(baseoverlay.getRelayNode(nodeid)); - dmsg.setFollowType(Discovery::predecessor); - dmsg.setTTL((uint8_t)4); - cmsg.encapsulate(&dmsg); - send(&cmsg, lnk); + OverlayMsg msg( typeDiscovery ); + msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() ); + msg.setRelayed(true); + Discovery dmsg( Discovery::successor, ttl, + baseoverlay.getEndpointDescriptor() ); + msg.encapsulate(&dmsg); + send(&msg, link); } } @@ -166,9 +185,10 @@ for (size_t i = 0; i < table->size(); i++) { route_item* it = (*table)[i]; - ChordMessage msg(ChordMessage::leave, nodeid, it->id); - send(&msg,it->info); - } -} - + OverlayMsg msg( typeLeave ); + send( &msg, it->info ); + } +} + +/// @see OverlayInterface.h const EndpointDescriptor& Chord::resolveNode(const NodeID& node) { const route_item* item = table->get(node); @@ -177,12 +197,24 @@ } -void Chord::routeMessage(const NodeID& destnode, Message* msg) { +/// @see OverlayInterface.h +const LinkID& Chord::getNextLinkId( const NodeID& id ) const { // get next hop +<<<<<<< .working const route_item* item = table->get_next_hop(destnode); - +======= + const route_item* item = table->get_next_hop(id); +>>>>>>> .merge-rechts.r5869 + +<<<<<<< .working // message for this node? yes-> delegate to base overlay if (item->id == nodeid || destnode == nodeid) baseoverlay.incomingRouteMessage( msg, LinkID::UNSPECIFIED, nodeid ); - +======= + // returns a unspecified id when this is itself + if (item == NULL || item->id == nodeid) + return LinkID::UNSPECIFIED; +>>>>>>> .merge-rechts.r5869 + +<<<<<<< .working else { // no-> send to next hop ChordMessage cmsg(ChordMessage::route, nodeid, destnode); @@ -190,6 +222,11 @@ send(&cmsg, item->info); } -} - +======= + /// return routing info + return item->info; +>>>>>>> .merge-rechts.r5869 +} + +<<<<<<< .working /// @see OverlayInterface.h void Chord::routeMessage(const NodeID& node, const LinkID& link, Message* msg) { @@ -214,4 +251,6 @@ } +======= +>>>>>>> .merge-rechts.r5869 OverlayInterface::NodeList Chord::getKnownNodes(bool deep) const { OverlayInterface::NodeList nodelist; @@ -243,5 +282,5 @@ /// @see OverlayInterface.h void Chord::onLinkUp(const LinkID& lnk, const NodeID& remote) { - logging_debug("link_up: link=" << lnk.toString() << " remote=" << + logging_info("link_up: link=" << lnk.toString() << " remote=" << remote.toString() ); for (vector::iterator i=pending.begin(); i!=pending.end(); i++) @@ -250,4 +289,17 @@ break; } +/* + // check if we already have a connection, yes-> do not handle duplicate! + for (int i=0; isize(); i++) + if ((*table)[i]->id == remote && !((*table)[i]->info.isUnspecified()) && (*table)[i]->info != lnk) { + + return; + } +*/ + if (remote==nodeid) { + baseoverlay.dropLink(lnk); + return; + } + route_item* item = table->insert(remote); @@ -257,7 +309,10 @@ << " with link " << lnk.toString()); // replace with new link - if (!item->info.isUnspecified()) + if (!item->info.isUnspecified() || item->info!=lnk) baseoverlay.dropLink(item->info); item->info = lnk; + // discover neighbors of new overlay neighbor + discover_neighbors( lnk ); + showLinks(); } else { // no-> add orphan entry to routing table logging_info("new orphan: " << remote.toString() @@ -266,11 +321,7 @@ } - discover_neighbors( lnk ); - + // erase bootstrap link vector::iterator it = std::find(bootstrapLinks.begin(), bootstrapLinks.end(), lnk); - if( it != bootstrapLinks.end() ) { - discover_neighbors( lnk ); - bootstrapLinks.erase( it ); - } + if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it ); } @@ -282,6 +333,8 @@ // remove link from routing table route_item* item = table->get(remote); - if (item!=NULL) item->info = LinkID::UNSPECIFIED; - table->remove(remote); + if (item!=NULL && item->info==lnk) { + item->info = LinkID::UNSPECIFIED; + table->remove(remote); + } } @@ -292,6 +345,5 @@ // decode message - typedef ChordMessage M; - M* m = msg.getMessage()->convert (); + OverlayMsg* m = dynamic_cast(msg.getMessage()); if (m == NULL) return; @@ -299,4 +351,5 @@ switch (m->getType()) { +<<<<<<< .working // invalid message case M::invalid: @@ -324,55 +377,63 @@ // discovery request case M::discovery: { +======= + // discovery request + case typeDiscovery: { +>>>>>>> .merge-rechts.r5869 // decapsulate message Discovery* dmsg = m->decapsulate (); - logging_debug("received discovery message with" - << " src=" << m->getSource().toString() - << " dst=" << m->getDestination().toString() + logging_debug("Received discovery message with" + << " src=" << m->getSourceNode().toString() + << " dst=" << m->getDestinationNode().toString() << " ttl=" << (int)dmsg->getTTL() - << " type=" << (int)dmsg->getFollowType() + << " type=" << (int)dmsg->getType() ); // check if source node can be added to routing table and setup link - if (m->getSource() != nodeid && table->is_insertable(m->getSource())) - setup(*dmsg->getSourceEndpoint(), m->getSource(), dmsg->getSourceRelay() ); + if (m->getSourceNode() != nodeid + && table->is_insertable(m->getSourceNode())) + setup( dmsg->getEndpoint(), m->getSourceNode() ); // delegate discovery message - switch (dmsg->getFollowType()) { + switch (dmsg->getType()) { // normal: route discovery message like every other message case Discovery::normal: { // closest node? yes-> split to follow successor and predecessor - if (table->is_closest_to(m->getDestination()) - || (table->get_successor()!=NULL && *table->get_successor() == m->getDestination()) - || (table->get_predesessor()!=NULL && *table->get_predesessor() == m->getDestination()) - ) { + if ( table->is_closest_to(m->getDestinationNode()) ) { if (table->get_successor() != NULL) { - // send successor message - ChordMessage cmsg_s(*m); - Discovery dmsg_s(*dmsg); - dmsg_s.setFollowType(Discovery::successor); - cmsg_s.encapsulate(&dmsg_s); + OverlayMsg omsg(*m); + dmsg->setType(Discovery::successor); + omsg.encapsulate(dmsg); route_item* succ_item = table->get(*table->get_successor()); logging_debug("Discovery split: routing discovery message to successor " << succ_item->id.toString() ); +<<<<<<< .working send(&cmsg_s, succ_item->info); +======= + send(&omsg, succ_item->info); +>>>>>>> .merge-rechts.r5869 } // send predecessor message if (table->get_predesessor() != NULL) { - ChordMessage cmsg_p(*m); - Discovery dmsg_p(*dmsg); - dmsg_p.setFollowType(Discovery::predecessor); - cmsg_p.encapsulate(&dmsg_p); + OverlayMsg omsg(*m); + dmsg->setType(Discovery::predecessor); + omsg.encapsulate(dmsg); route_item* pred_item = table->get( *table->get_predesessor()); logging_debug("Discovery split: routing discovery message to predecessor " << pred_item->id.toString() ); +<<<<<<< .working send(&cmsg_p, pred_item->info); +======= + send( &omsg, pred_item->info); +>>>>>>> .merge-rechts.r5869 } } // no-> route message else { +<<<<<<< .working // find next hop const route_item* item = table->get_next_hop(m->getDestination()); @@ -381,4 +442,7 @@ item->id.toString() ); send(m, item->info); +======= + baseoverlay.send( m, m->getDestinationNode() ); +>>>>>>> .merge-rechts.r5869 } break; @@ -395,5 +459,5 @@ const route_item* item = NULL; - if (dmsg->getFollowType() == Discovery::successor && + if (dmsg->getType() == Discovery::successor && table->get_successor() != NULL) { item = table->get(*table->get_successor()); @@ -405,8 +469,8 @@ logging_debug("routing discovery message to succ/pred " << item->id.toString() ); - ChordMessage cmsg(*m); - Discovery dmsg_p(*dmsg); - cmsg.encapsulate(&dmsg_p); - send(&cmsg, item->info); + OverlayMsg omsg(*m); + omsg.encapsulate(dmsg); + omsg.setDestinationNode(item->id); + baseoverlay.send(&omsg, omsg.getDestinationNode()); break; }} @@ -415,6 +479,6 @@ } - // leave - case M::leave: { + // leave + case typeLeave: { if (link!=LinkID::UNSPECIFIED) { route_item* item = table->get(remote); @@ -424,7 +488,5 @@ } break; - } - } - delete m; + }} } @@ -464,5 +526,9 @@ // remove orphan links orphan_removal_counter++; +<<<<<<< .working if (orphan_removal_counter <0 || orphan_removal_counter >= 4) { +======= + if (orphan_removal_counter <0 || orphan_removal_counter >= 2) { +>>>>>>> .merge-rechts.r5869 logging_info("Running orphan removal"); orphan_removal_counter = 0; @@ -473,6 +539,7 @@ table->insert(it->id); if (it->ref_count==0) { - baseoverlay.dropLink(it->info); + LinkID id = it->info; it->info = LinkID::UNSPECIFIED; + baseoverlay.dropLink(id); } } @@ -480,4 +547,7 @@ } } +} + +void Chord::showLinks() { logging_info("--- chord routing information ----------------------------------"); logging_info("predecessor: " << (table->get_predesessor()==NULL? "" : Index: /source/ariba/overlay/modules/chord/Chord.h =================================================================== --- /source/ariba/overlay/modules/chord/Chord.h (revision 5860) +++ /source/ariba/overlay/modules/chord/Chord.h (revision 5870) @@ -51,4 +51,6 @@ namespace overlay { +class OverlayMsg; + using ariba::communication::EndpointDescriptor; using ariba::utility::Timer; @@ -82,9 +84,8 @@ // helper: sets up a link using the "base overlay" LinkID setup( const EndpointDescriptor& endp, - const NodeID& node = NodeID::UNSPECIFIED, - const NodeID& remoteRelay = NodeID::UNSPECIFIED ); + const NodeID& node = NodeID::UNSPECIFIED ); // helper: sends a message using the "base overlay" - seqnum_t send( Message* msg, const LinkID& link ); + seqnum_t send( OverlayMsg* msg, const LinkID& link ); // stabilization: sends a discovery message to the specified neighborhood @@ -92,4 +93,6 @@ void discover_neighbors( const LinkID& lnk ); + + void showLinks(); public: @@ -119,10 +122,4 @@ /// @see OverlayInterface.h - virtual void routeMessage( const NodeID& destnode, Message* msg ); - - /// @see OverlayInterface.h - virtual void routeMessage(const NodeID& node, const LinkID& link, Message* msg); - - /// @see OverlayInterface.h virtual NodeList getKnownNodes(bool deep = true) const; @@ -137,8 +134,7 @@ const LinkID& lnk = LinkID::UNSPECIFIED); - - /// @see Timer.h virtual void eventFunction(); + }; Index: urce/ariba/overlay/modules/chord/messages/ChordMessage.cpp =================================================================== --- /source/ariba/overlay/modules/chord/messages/ChordMessage.cpp (revision 5860) +++ (revision ) @@ -1,49 +1,0 @@ -// [License] -// The Ariba-Underlay Copyright -// -// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH) -// -// Institute of Telematics -// UniversitÀt Karlsruhe (TH) -// Zirkel 2, 76128 Karlsruhe -// Germany -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are -// met: -// -// 1. Redistributions of source code must retain the above copyright -// notice, this list of conditions and the following disclaimer. -// 2. Redistributions in binary form must reproduce the above copyright -// notice, this list of conditions and the following disclaimer in the -// documentation and/or other materials provided with the distribution. -// -// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND -// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE -// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR -// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR -// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, -// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, -// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR -// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF -// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING -// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS -// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -// -// The views and conclusions contained in the software and documentation -// are those of the authors and should not be interpreted as representing -// official policies, either expressed or implied, of the Institute of -// Telematics. -// [License] - -#include "ChordMessage.h" - -namespace ariba { -namespace overlay { - -vsznDefault( ChordMessage ); - -ChordMessage::~ChordMessage(){ -} - -}} // ariba::overlay Index: urce/ariba/overlay/modules/chord/messages/ChordMessage.h =================================================================== --- /source/ariba/overlay/modules/chord/messages/ChordMessage.h (revision 5860) +++ (revision ) @@ -1,110 +1,0 @@ -// [License] -// The Ariba-Underlay Copyright -// -// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH) -// -// Institute of Telematics -// UniversitÀt Karlsruhe (TH) -// Zirkel 2, 76128 Karlsruhe -// Germany -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are -// met: -// -// 1. Redistributions of source code must retain the above copyright -// notice, this list of conditions and the following disclaimer. -// 2. Redistributions in binary form must reproduce the above copyright -// notice, this list of conditions and the following disclaimer in the -// documentation and/or other materials provided with the distribution. -// -// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND -// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE -// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR -// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR -// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, -// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, -// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR -// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF -// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING -// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS -// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -// -// The views and conclusions contained in the software and documentation -// are those of the authors and should not be interpreted as representing -// official policies, either expressed or implied, of the Institute of -// Telematics. -// [License] - -#ifndef __CHORD_MESSAGE_H -#define __CHORD_MESSAGE_H - -#include "ariba/utility/messages.h" -#include "ariba/utility/serialization.h" -#include "ariba/Identifiers.h" - -using ariba::utility::Message; - -namespace ariba { -namespace overlay { - -using_serialization -; - -class ChordMessage: public Message { VSERIALIZEABLE -public: - enum type_ { - invalid = 0, ///< invalid message - route = 1, ///< route message with payload - discovery = 2, ///< discover neighbors of a destination node - leave = 3, ///< leave message - }; - - inline ChordMessage( const ChordMessage& msg ) { - this->type = msg.type; - this->hop_count = msg.hop_count; - this->source = msg.source; - this->destination = msg.destination; - } - - inline explicit ChordMessage(type_ type = invalid, - const NodeID& source = NodeID::UNSPECIFIED, - const NodeID& destination = NodeID::UNSPECIFIED) : - type((uint8_t) type), source(source), destination(destination) { - } - - virtual ~ChordMessage(); - - inline uint8_t getHopCount() const { - return hop_count; - } - - inline void setHopCount( uint8_t hop_count ) { - this->hop_count = hop_count; - } - - inline type_ getType() const { - return (type_) this->type; - } - - const NodeID& getSource() const { - return source; - } - - inline const NodeID& getDestination() const { - return destination; - } - -private: - uint8_t type; - uint8_t hop_count; - NodeID source, destination; -}; - -}} // ariba::overlay - -sznBeginDefault( ariba::overlay::ChordMessage, X ){ - X && type && hop_count && &source && &destination && Payload(); -}sznEnd(); - -#endif // __CHORD_MESSAGE_H Index: /source/ariba/overlay/modules/chord/messages/Discovery.cpp =================================================================== --- /source/ariba/overlay/modules/chord/messages/Discovery.cpp (revision 5860) +++ /source/ariba/overlay/modules/chord/messages/Discovery.cpp (revision 5870) @@ -44,7 +44,4 @@ vsznDefault(Discovery); -Discovery::Discovery(){ -} - Discovery::~Discovery(){ } Index: /source/ariba/overlay/modules/chord/messages/Discovery.h =================================================================== --- /source/ariba/overlay/modules/chord/messages/Discovery.h (revision 5860) +++ /source/ariba/overlay/modules/chord/messages/Discovery.h (revision 5870) @@ -41,4 +41,5 @@ #include + #include "ariba/utility/messages.h" #include "ariba/utility/serialization.h" @@ -61,25 +62,26 @@ VSERIALIZEABLE; public: - enum follow_type_ { - normal = 0, - successor = 1, - predecessor = 2 + enum type_ { + invalid = 0, + normal = 1, + successor = 2, + predecessor = 3 }; - Discovery( const Discovery& msg ) { - this->follow_type = msg.follow_type; - this->ttl = msg.ttl; - this->source_relay = msg.source_relay; - this->source_endpoint = msg.source_endpoint; + Discovery( const Discovery& msg ) : type(msg.type), ttl(msg.ttl), + endpoint(msg.endpoint) { } - explicit Discovery(); + Discovery( type_ type = invalid, uint8_t ttl = 4, + const EndpointDescriptor& endpoint = EndpointDescriptor::UNSPECIFIED() ) + : type(type), ttl(ttl), endpoint(endpoint) { + } virtual ~Discovery(); - const EndpointDescriptor* getSourceEndpoint() const { - return &source_endpoint; + inline type_ getType() const { + return (type_)type; } - void setSourceEndpoint( const EndpointDescriptor* endpoint ) { - source_endpoint = *endpoint; + inline void setType( type_ type ) { + this->type = type; } @@ -92,24 +94,16 @@ } - inline follow_type_ getFollowType() const { - return (follow_type_)follow_type; + inline const EndpointDescriptor& getEndpoint() const { + return endpoint; } - inline void setFollowType( follow_type_ type ) { - follow_type = (uint8_t)type; + inline void setEndpoint( const EndpointDescriptor& endpoint ) { + this->endpoint = endpoint; } - inline void setSourceRelay( const NodeID& relay ) { - source_relay = relay; - } - - inline const NodeID& getSourceRelay() const { - return source_relay; - } private: - uint8_t follow_type; + uint8_t type; uint8_t ttl; - EndpointDescriptor source_endpoint; - NodeID source_relay; + EndpointDescriptor endpoint; }; @@ -117,9 +111,5 @@ sznBeginDefault( ariba::overlay::Discovery, X ) { - /// serialize follow-type and time-to-live - X && follow_type && ttl; - - // serialize end-point - X && &source_relay && source_endpoint; + X && type && ttl && endpoint; } sznEnd(); Index: /source/ariba/utility/messages/Message.cpp =================================================================== --- /source/ariba/utility/messages/Message.cpp (revision 5860) +++ /source/ariba/utility/messages/Message.cpp (revision 5870) @@ -52,4 +52,5 @@ Message::~Message() { + dropPayload(); // if ( srcAddr != NULL) delete srcAddr; // if ( destAddr != NULL) delete destAddr; Index: /source/ariba/utility/serialization/Data.hpp =================================================================== --- /source/ariba/utility/serialization/Data.hpp (revision 5860) +++ /source/ariba/utility/serialization/Data.hpp (revision 5870) @@ -420,5 +420,5 @@ finline void release() { - delete [] bufferPtr; + if (bufferPtr!=NULL && bufferLen>=0) delete [] bufferPtr; bufferPtr = NULL; bufferLen = -1; Index: /source/ariba/utility/system/StartupWrapper.cpp =================================================================== --- /source/ariba/utility/system/StartupWrapper.cpp (revision 5860) +++ /source/ariba/utility/system/StartupWrapper.cpp (revision 5870) @@ -119,13 +119,13 @@ log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BootstrapManager")); logger->setLevel(log4cxx::Level::getDebug()); - } + }*/ { log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("Chord")); - logger->setLevel(log4cxx::Level::getDebug()); + logger->setLevel(log4cxx::Level::getInfo()); } { log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BaseOverlay")); - logger->setLevel(log4cxx::Level::getDebug()); - }*/ + logger->setLevel(log4cxx::Level::getInfo()); + } //$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$