Index: source/ariba/overlay/BaseOverlay.cpp =================================================================== --- source/ariba/overlay/BaseOverlay.cpp (revision 10700) +++ source/ariba/overlay/BaseOverlay.cpp (revision 12060) @@ -57,11 +57,23 @@ #include "ariba/utility/visual/DddVis.h" #include "ariba/utility/visual/ServerVis.h" +#include namespace ariba { namespace overlay { + +using namespace std; +using ariba::transport::system_priority; #define visualInstance ariba::utility::DddVis::instance() #define visualIdOverlay ariba::utility::ServerVis::NETWORK_ID_BASE_OVERLAY #define visualIdBase ariba::utility::ServerVis::NETWORK_ID_BASE_COMMUNICATION + + +// time constants (in seconds) +#define KEEP_ALIVE_TIME 60 // send keep-alive message after link is not used for #s + +#define LINK_ESTABLISH_TIME_OUT 10 // timeout: link requested but not up +#define KEEP_ALIVE_TIME_OUT KEEP_ALIVE_TIME + LINK_ESTABLISH_TIME_OUT // timeout: no data received on this link (incl. keep-alive messages) +#define AUTO_LINK_TIME_OUT KEEP_ALIVE_TIME_OUT // timeout: auto link not used for #s @@ -85,5 +97,5 @@ LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) { - BOOST_FOREACH( LinkDescriptor* lp, links ) + foreach( LinkDescriptor* lp, links ) if ((communication ? lp->communicationId : lp->overlayId) == link) return lp; @@ -92,5 +104,5 @@ const LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) const { - BOOST_FOREACH( const LinkDescriptor* lp, links ) + foreach( const LinkDescriptor* lp, links ) if ((communication ? lp->communicationId : lp->overlayId) == link) return lp; @@ -122,13 +134,23 @@ /// returns a auto-link descriptor -LinkDescriptor* BaseOverlay::getAutoDescriptor( const NodeID& node, const ServiceID& service ) { +LinkDescriptor* BaseOverlay::getAutoDescriptor( const NodeID& node, const ServiceID& service ) +{ // search for a descriptor that is already up - BOOST_FOREACH( LinkDescriptor* lp, links ) - if (lp->autolink && lp->remoteNode == node && lp->service == service && lp->up && lp->keepAliveMissed == 0) - return lp; + foreach( LinkDescriptor* lp, links ) + { + if (lp->autolink && lp->remoteNode == node && lp->service == service && isLinkVital(lp) ) + return lp; + } + // if not found, search for one that is about to come up... - BOOST_FOREACH( LinkDescriptor* lp, links ) - if (lp->autolink && lp->remoteNode == node && lp->service == service && lp->keepAliveMissed == 0 ) - return lp; + foreach( LinkDescriptor* lp, links ) + { + time_t now = time(NULL); + + if (lp->autolink && lp->remoteNode == node && lp->service == service + && difftime( now, lp->keepAliveReceived ) <= LINK_ESTABLISH_TIME_OUT ) + return lp; + } + return NULL; } @@ -136,35 +158,47 @@ /// stabilizes link information void BaseOverlay::stabilizeLinks() { - // send keep-alive messages over established links - BOOST_FOREACH( LinkDescriptor* ld, links ) { + time_t now = time(NULL); + + // send keep-alive messages over established links + foreach( LinkDescriptor* ld, links ) + { if (!ld->up) continue; - OverlayMsg msg( OverlayMsg::typeLinkAlive, - OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode ); - if (ld->relayed) msg.setRouteRecord(true); - send_link( &msg, ld->overlayId ); + + if ( difftime( now, ld->keepAliveSent ) >= KEEP_ALIVE_TIME ) + { + logging_debug("[BaseOverlay] Sending KeepAlive over " + << ld->to_string() + << " after " + << difftime( now, ld->keepAliveSent ) + << "s"); + + OverlayMsg msg( OverlayMsg::typeKeepAlive, + OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode ); + msg.setRouteRecord(true); + ld->keepAliveSent = now; + send_link( &msg, ld->overlayId, system_priority::OVERLAY ); + } } // iterate over all links and check for time boundaries vector oldlinks; - time_t now = time(NULL); - BOOST_FOREACH( LinkDescriptor* ld, links ) { - - // keep alives and not up? yes-> link connection request is stale! - if ( !ld->up && difftime( now, ld->keepAliveTime ) >= 2 ) { - - // increase counter - ld->keepAliveMissed++; - - // missed more than four keep-alive messages (10 sec)? -> drop link - if (ld->keepAliveMissed > 4) { - logging_info( "Link connection request is stale, closing: " << ld ); - oldlinks.push_back( ld ); - continue; - } + foreach( LinkDescriptor* ld, links ) { + + // link connection request stale? + if ( !ld->up && difftime( now, ld->keepAliveReceived ) >= LINK_ESTABLISH_TIME_OUT ) // NOTE: keepAliveReceived == now, on connection request + { + logging_info( "Link connection request is stale, closing: " << ld ); + ld->failed = true; + oldlinks.push_back( ld ); + continue; } if (!ld->up) continue; + + + // check if link is relayed and retry connecting directly + // TODO Mario: What happens here? --> There are 3 attempts to replace a relayed link with a direct one. see: handleLinkReply if ( ld->relayed && !ld->communicationUp && ld->retryCounter > 0) { ld->retryCounter--; @@ -173,5 +207,5 @@ // remote used as relay flag - if ( ld->relaying && difftime( now, ld->timeRelaying ) > 10) + if ( ld->relaying && difftime( now, ld->timeRelaying ) > KEEP_ALIVE_TIME_OUT) // TODO is this a reasonable timeout ?? ld->relaying = false; @@ -183,5 +217,5 @@ // auto-link time exceeded? - if ( ld->autolink && difftime( now, ld->lastuse ) > 30 ) { + if ( ld->autolink && difftime( now, ld->lastuse ) > AUTO_LINK_TIME_OUT ) { oldlinks.push_back( ld ); continue; @@ -189,20 +223,15 @@ // keep alives missed? yes-> - if ( difftime( now, ld->keepAliveTime ) > 4 ) { - - // increase counter - ld->keepAliveMissed++; - - // missed more than four keep-alive messages (4 sec)? -> drop link - if (ld->keepAliveMissed >= 2) { - logging_info( "Link is stale, closing: " << ld ); - oldlinks.push_back( ld ); - continue; - } + if ( difftime( now, ld->keepAliveReceived ) >= KEEP_ALIVE_TIME_OUT ) + { + logging_info( "Link is stale, closing: " << ld ); + ld->failed = true; + oldlinks.push_back( ld ); + continue; } } // drop links - BOOST_FOREACH( LinkDescriptor* ld, oldlinks ) { + foreach( LinkDescriptor* ld, oldlinks ) { logging_info( "Link timed out. Dropping " << ld ); ld->relaying = false; @@ -210,8 +239,17 @@ } - // show link state - counter++; - if (counter>=4) showLinks(); - if (counter>=4 || counter<0) counter = 0; + + + + // show link state (debug output) + if (counter>=10 || counter<0) + { + showLinks(); + counter = 0; + } + else + { + counter++; + } } @@ -230,8 +268,8 @@ int i=0; - BOOST_FOREACH( LinkDescriptor* ld, links ) { - if (!ld->isVital() || ld->service != OverlayInterface::OVERLAY_SERVICE_ID) continue; + foreach( LinkDescriptor* ld, links ) { + if (!isLinkVital(ld) || ld->service != OverlayInterface::OVERLAY_SERVICE_ID) continue; bool found = false; - BOOST_FOREACH(NodeID& id, nodes) + foreach(NodeID& id, nodes) if (id == ld->remoteNode) found = true; if (found) continue; @@ -261,8 +299,15 @@ int i=0; logging_info("--- link state -------------------------------"); - BOOST_FOREACH( LinkDescriptor* ld, links ) { + foreach( LinkDescriptor* ld, links ) { string epd = ""; - if (ld->isDirectVital()) - epd = getEndpointDescriptor(ld->remoteNode).toString(); + if (isLinkDirectVital(ld)) + { +// epd = getEndpointDescriptor(ld->remoteNode).toString(); + + epd = "Connection: "; + epd += bc->get_local_endpoint_of_link(ld->communicationId)->to_string(); + epd += " <---> "; + epd += bc->get_remote_endpoint_of_link(ld->communicationId)->to_string(); + } logging_info("LINK_STATE: " << i << ": " << ld << " " << epd); @@ -289,11 +334,38 @@ // internal message delivery --------------------------------------------------- + +seqnum_t BaseOverlay::send_overlaymessage_down( OverlayMsg* message, const LinkID& bc_link, uint8_t priority ) +{ + // set priority + message->setPriority(priority); + + // wrap old OverlayMsg into reboost message + reboost::message_t wrapped_message = message->wrap_up_for_sending(); + + // send down to BaseCommunication + try + { + // * send * + return bc->sendMessage(bc_link, wrapped_message, priority, false); + } + catch ( communication::communication_message_not_sent& e ) + { + ostringstream out; + out << "Communication message not sent: " << e.what(); + throw message_not_sent(out.str()); + } + + throw logic_error("This should never happen!"); +} + + /// routes a message to its destination node -void BaseOverlay::route( OverlayMsg* message ) { - +void BaseOverlay::route( OverlayMsg* message, const NodeID& last_hop ) +{ // exceeded time-to-live? yes-> drop message - if (message->getNumHops() > message->getTimeToLive()) { - logging_warn("Message exceeded TTL. Dropping message and relay routes" - "for recovery."); + if (message->getNumHops() > message->getTimeToLive()) + { + logging_warn("Message exceeded TTL. Dropping message and relay routes " + << "for recovery. Hop count: " << (int) message->getNumHops()); removeRelayNode(message->getDestinationNode()); return; @@ -301,132 +373,279 @@ // no-> forward message - else { + else + { // destinastion myself? yes-> handle message - if (message->getDestinationNode() == nodeId) { - logging_warn("Usually I should not route messages to myself!"); - Message msg; - msg.encapsulate(message); - handleMessage( &msg, NULL ); - } else { - // no->send message to next hop - send( message, message->getDestinationNode() ); - } - } + if (message->getDestinationNode() == nodeId) + { + logging_warn("Usually I should not route messages to myself. And I won't!"); + } + + // no->send message to next hop + else + { + try + { + /* (deep) packet inspection to determine priority */ + // BRANCH: typeData --> send with low priority + if ( message->getType() == OverlayMsg::typeData ) + { + // TODO think about implementing explicit routing queue (with active queue management??) + send( message, + message->getDestinationNode(), + message->getPriority(), + last_hop ); + } + // BRANCH: internal message --> send with higher priority + else + { + send( message, + message->getDestinationNode(), + system_priority::HIGH, + last_hop ); + } + } + catch ( message_not_sent& e ) + { + logging_warn("Unable to route message of type " + << message->getType() + << " to " + << message->getDestinationNode() + << ". Reason: " + << e.what()); + + // inform sender + if ( message->getType() != OverlayMsg::typeMessageLost ) + { + report_lost_message(message); + } + } + } + } +} + +void BaseOverlay::report_lost_message( const OverlayMsg* message ) +{ + OverlayMsg reply(OverlayMsg::typeMessageLost); + reply.setSeqNum(message->getSeqNum()); + + /** + * MessageLost-Message + * + * - Type of lost message + * - Hop count of lost message + * - Source-LinkID of lost message + */ + reboost::shared_buffer_t b(sizeof(uint8_t)*2); + b.mutable_data()[0] = message->getType(); + b.mutable_data()[1] = message->getNumHops(); + reply.append_buffer(b); + reply.append_buffer(message->getSourceLink().serialize()); + + try + { + send_node(&reply, message->getSourceNode(), + system_priority::OVERLAY, + OverlayInterface::OVERLAY_SERVICE_ID); + } + catch ( message_not_sent& e ) + { + logging_warn("Tried to inform another node that we could'n route their message. But we were not able to send this error-message, too."); + } } /// sends a message to another node, delivers it to the base overlay class -seqnum_t BaseOverlay::send( OverlayMsg* message, const NodeID& destination ) { +seqnum_t BaseOverlay::send( OverlayMsg* message, + const NodeID& destination, + uint8_t priority, + const NodeID& last_hop ) throw(message_not_sent) +{ LinkDescriptor* next_link = NULL; // drop messages to unspecified destinations - if (destination.isUnspecified()) return -1; - - // send messages to myself -> handle message and drop warning! - if (destination == nodeId) { - logging_warn("Sent message to myself. Handling message.") - Message msg; - msg.encapsulate(message); - handleMessage( &msg, NULL ); - return -1; + if (destination.isUnspecified()) + throw message_not_sent("No destination specified. Drop!"); + + // send messages to myself -> drop! + // TODO maybe this is not what we want. why not just deliver this message? + // There is a similar check in the route function, there it should be okay. + if (destination == nodeId) + { + logging_warn("Sent message to myself. Drop!"); + + throw message_not_sent("Sent message to myself. Drop!"); } // use relay path? - if (message->isRelayed()) { + if (message->isRelayed()) + { next_link = getRelayLinkTo( destination ); - if (next_link != NULL) { + + if (next_link != NULL) + { next_link->setRelaying(); - return bc->sendMessage(next_link->communicationId, message); - } else { - logging_warn("Could not send message. No relay hop found to " - << destination << " -- trying to route over overlay paths ...") -// logging_error("ERROR: " << debugInformation() ); - // return -1; - } - } - + + // * send message over relayed link * + return send_overlaymessage_down(message, next_link->communicationId, priority); + } + else + { + logging_warn("No relay hop found to " << destination + << " -- trying to route over overlay paths ...") + } + } + + // last resort -> route over overlay path LinkID next_id = overlayInterface->getNextLinkId( destination ); - if (next_id.isUnspecified()) { - logging_warn("Could not send message. No next hop found to " << - destination ); - logging_error("ERROR: " << debugInformation() ); - return -1; - } - - // get link descriptor, up and running? yes-> send message + if ( next_id.isUnspecified() ) + { + // apperently we're the closest node --> try second best node + // NOTE: This is helpful if our routing table is not up-to-date, but + // may lead to circles. So we have to be careful. + std::vector next_ids = + overlayInterface->getSortedLinkIdsTowardsNode( destination ); + + for ( int i = 0; i < next_ids.size(); i++ ) + { + const LinkID& link = *next_ids[i]; + + if ( ! link.isUnspecified() ) + { + next_id = link; + + break; + } + } + + // still no next hop found. drop. + if ( next_id.isUnspecified() ) + { + logging_warn("Could not send message. No next hop found to " << + destination ); + logging_error("ERROR: " << debugInformation() ); + + throw message_not_sent("No next hop found."); + } + } + + + /* get link descriptor, do some checks and send message */ next_link = getDescriptor(next_id); - if (next_link != NULL && next_link->up) { - // send message over relayed link - return send(message, next_link); - } - - // no-> error, dropping message - else { - logging_warn("Could not send message. Link not known or up"); - logging_error("ERROR: " << debugInformation() ); - return -1; - } - - // not reached-> fail - return -1; -} + + // check pointer + if ( next_link == NULL ) + { + // NOTE: this shuldn't happen + throw message_not_sent("Could not send message. Link not known."); + } + + // avoid circles + if ( next_link->remoteNode == last_hop ) + { + // XXX logging_debug + logging_info("Possible next hop would create a circle: " + << next_link->remoteNode); + + throw message_not_sent("Could not send message. Possible next hop would create a circle."); + } + + // check if link is up + if ( ! next_link->up) + { + logging_warn("Could not send message. Link not up"); + logging_error("ERROR: " << debugInformation() ); + + throw message_not_sent("Could not send message. Link not up"); + } + + // * send message over overlay link * + return send(message, next_link, priority); +} + /// send a message using a link descriptor, delivers it to the base overlay class -seqnum_t BaseOverlay::send( OverlayMsg* message, LinkDescriptor* ldr, bool ignore_down ) { +seqnum_t BaseOverlay::send( OverlayMsg* message, + LinkDescriptor* ldr, + uint8_t priority ) throw(message_not_sent) +{ // check if null - if (ldr == NULL) { - logging_error("Can not send message to " << message->getDestinationAddress()); - return -1; + if (ldr == NULL) + { + ostringstream out; + out << "Can not send message to " << message->getDestinationAddress(); + throw message_not_sent(out.str()); } // check if up - if (!ldr->up && !ignore_down) { - logging_error("Can not send message. Link not up:" << ldr ); + if ( !ldr->up ) + { logging_error("DEBUG_INFO: " << debugInformation() ); - return -1; - } - LinkDescriptor* ld = NULL; - - // handle relayed link - if (ldr->relayed) { + + ostringstream out; + out << "Can not send message. Link not up:" << ldr->to_string(); + throw message_not_sent(out.str()); + } + + LinkDescriptor* next_hop_ld = NULL; + + // BRANCH: relayed link + if (ldr->relayed) + { logging_debug("Resolving direct link for relayed link to " << ldr->remoteNode); - ld = getRelayLinkTo( ldr->remoteNode ); - if (ld==NULL) { - logging_error("No relay path found to link " << ldr ); + + next_hop_ld = getRelayLinkTo( ldr->remoteNode ); + + if (next_hop_ld==NULL) + { logging_error("DEBUG_INFO: " << debugInformation() ); - return -1; - } - ld->setRelaying(); + + ostringstream out; + out << "No relay path found to link: " << ldr; + throw message_not_sent(out.str()); + } + + next_hop_ld->setRelaying(); message->setRelayed(true); - } else - ld = ldr; - - // handle direct link - if (ld->communicationUp) { - logging_debug("send(): Sending message over direct link."); - return bc->sendMessage( ld->communicationId, message ); - } else { - logging_error("send(): Could not send message. " - "Not a relayed link and direct link is not up."); - return -1; - } - return -1; + } + // BRANCH: direct link + else + { + next_hop_ld = ldr; + } + + + // check next hop-link + if ( ! next_hop_ld->communicationUp) + { + throw message_not_sent( "send(): Could not send message." + " Not a relayed link and direct link is not up." ); + } + + // send over next link + logging_debug("send(): Sending message over direct link."); + return send_overlaymessage_down(message, next_hop_ld->communicationId, priority); + } seqnum_t BaseOverlay::send_node( OverlayMsg* message, const NodeID& remote, - const ServiceID& service) { + uint8_t priority, const ServiceID& service) throw(message_not_sent) +{ message->setSourceNode(nodeId); message->setDestinationNode(remote); message->setService(service); - return send( message, remote ); -} - -seqnum_t BaseOverlay::send_link( OverlayMsg* message, const LinkID& link,bool ignore_down ) { + return send( message, remote, priority ); +} + +void BaseOverlay::send_link( OverlayMsg* message, + const LinkID& link, + uint8_t priority ) throw(message_not_sent) +{ LinkDescriptor* ld = getDescriptor(link); - if (ld==NULL) { - logging_error("Cannot find descriptor to link id=" << link.toString()); - return -1; - } + if (ld==NULL) + { + throw message_not_sent("Cannot find descriptor to link id=" + link.toString()); + } + message->setSourceNode(nodeId); message->setDestinationNode(ld->remoteNode); @@ -437,5 +656,17 @@ message->setService(ld->service); message->setRelayed(ld->relayed); - return send( message, ld, ignore_down ); + + + try + { + // * send message * + send( message, ld, priority ); + } + catch ( message_not_sent& e ) + { + // drop failed link + ld->failed = true; + dropLink(ld->overlayId); + } } @@ -451,6 +682,7 @@ // relay link still used and alive? if (ld==NULL - || !ld->isDirectVital() - || difftime(route.used, time(NULL)) > 8) { + || !isLinkDirectVital(ld) + || difftime(route.used, time(NULL)) > KEEP_ALIVE_TIME_OUT) // TODO this was set to 8 before.. Is the new timeout better? + { logging_info("Forgetting relay information to node " << route.node.toString() ); @@ -488,5 +720,5 @@ if (message->isRelayed()) { // try to find source node - BOOST_FOREACH( relay_route& route, relay_routes ) { + foreach( relay_route& route, relay_routes ) { // relay route found? yes-> if ( route.node == message->getDestinationNode() ) { @@ -504,5 +736,5 @@ // try to find source node - BOOST_FOREACH( relay_route& route, relay_routes ) { + foreach( relay_route& route, relay_routes ) { // relay route found? yes-> @@ -516,8 +748,8 @@ if (route.hops > message->getNumHops() || rld == NULL - || !rld->isDirectVital()) { + || !isLinkDirectVital(ld)) { logging_info("Updating relay information to node " << route.node.toString() - << " reducing to " << message->getNumHops() << " hops."); + << " reducing to " << (int) message->getNumHops() << " hops."); route.hops = message->getNumHops(); route.link = ld->overlayId; @@ -542,8 +774,8 @@ LinkDescriptor* BaseOverlay::getRelayLinkTo( const NodeID& remote ) { // try to find source node - BOOST_FOREACH( relay_route& route, relay_routes ) { + foreach( relay_route& route, relay_routes ) { if (route.node == remote ) { LinkDescriptor* ld = getDescriptor( route.link ); - if (ld==NULL || !ld->isDirectVital()) return NULL; else { + if (ld==NULL || !isLinkDirectVital(ld)) return NULL; else { route.used = time(NULL); return ld; @@ -575,9 +807,9 @@ // ---------------------------------------------------------------------------- -void BaseOverlay::start( BaseCommunication& _basecomm, const NodeID& _nodeid ) { +void BaseOverlay::start( BaseCommunication* _basecomm, const NodeID& _nodeid ) { logging_info("Starting..."); // set parameters - bc = &_basecomm; + bc = _basecomm; nodeId = _nodeid; @@ -587,5 +819,6 @@ // timer for auto link management - Timer::setInterval( 1000 ); + Timer::setInterval( 1000 ); // XXX +// Timer::setInterval( 10000 ); Timer::start(); @@ -641,5 +874,5 @@ overlayInterface->joinOverlay(); state = BaseOverlayStateCompleted; - BOOST_FOREACH( NodeListener* i, nodeListeners ) + foreach( NodeListener* i, nodeListeners ) i->onJoinCompleted( spovnetId ); @@ -682,5 +915,6 @@ // gather all service links vector servicelinks; - BOOST_FOREACH( LinkDescriptor* ld, links ) { + foreach( LinkDescriptor* ld, links ) + { if( ld->service != OverlayInterface::OVERLAY_SERVICE_ID ) servicelinks.push_back( ld->overlayId ); @@ -688,15 +922,23 @@ // drop all service links - BOOST_FOREACH( LinkID lnk, servicelinks ) - dropLink( lnk ); + foreach( LinkID lnk, servicelinks ) + { + logging_debug("Dropping service link " << lnk.toString()); + dropLink( lnk ); + } // let the node leave the spovnet overlay interface logging_debug( "Leaving overlay" ); if( overlayInterface != NULL ) + { overlayInterface->leaveOverlay(); + } // drop still open bootstrap links - BOOST_FOREACH( LinkID lnk, bootstrapLinks ) - bc->dropLink( lnk ); + foreach( LinkID lnk, bootstrapLinks ) + { + logging_debug("Dropping bootstrap link " << lnk.toString()); + bc->dropLink( lnk ); + } // change to inalid state @@ -708,5 +950,6 @@ // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ) { + foreach( NodeListener* i, nodeListeners ) + { if( ret ) i->onLeaveCompleted( spovnetId ); else i->onLeaveFailed( spovnetId ); @@ -731,5 +974,5 @@ state = BaseOverlayStateInvalid; - BOOST_FOREACH( NodeListener* i, nodeListeners ) + foreach( NodeListener* i, nodeListeners ) i->onJoinFailed( spovnetId ); @@ -783,7 +1026,11 @@ const ServiceID& service ) { + // TODO What if we already have a Link to this node and this service id? + // do not establish a link to myself! - if (remote == nodeId) return LinkID::UNSPECIFIED; - + if (remote == nodeId) return + LinkID::UNSPECIFIED; + + // create a link descriptor LinkDescriptor* ld = addDescriptor(); @@ -792,4 +1039,8 @@ ld->service = service; ld->listener = getListener(ld->service); + + // initialize sequence numbers + ld->last_sent_seqnum = SequenceNumber::createRandomSeqNum_Short(); + logging_debug("Creating new link with initial SeqNum: " << ld->last_sent_seqnum); // create link request message @@ -800,4 +1051,7 @@ msg.setRelayed(true); msg.setRegisterRelay(true); +// msg.setRouteRecord(true); + + msg.setSeqNum(ld->last_sent_seqnum); // debug message @@ -809,17 +1063,71 @@ ); + // sending message to node - send_node( &msg, ld->remoteNode, ld->service ); - + try + { + // * send * + seqnum_t seq = send_node( &msg, ld->remoteNode, system_priority::OVERLAY, ld->service ); + } + catch ( message_not_sent& e ) + { + logging_warn("Link request not sent: " << e.what()); + + // Message not sent. Cancel link request. + SystemQueue::instance().scheduleCall( + boost::bind( + &BaseOverlay::__onLinkEstablishmentFailed, + this, + ld->overlayId) + ); + } + return ld->overlayId; } +/// NOTE: "id" is an Overlay-LinkID +void BaseOverlay::__onLinkEstablishmentFailed(const LinkID& id) +{ + // TODO This code redundant. But also it's not easy to aggregate in one function. + + // get descriptor for link + LinkDescriptor* ld = getDescriptor(id, false); + if ( ld == NULL ) return; // not found? ->ignore! + + logging_debug( "__onLinkEstablishmentFaild: " << ld ); + + // removing relay link information + removeRelayLink(ld->overlayId); + + // inform listeners about link down + ld->communicationUp = false; + if (!ld->service.isUnspecified()) + { + CommunicationListener* lst = getListener(ld->service); + if(lst != NULL) lst->onLinkFail( ld->overlayId, ld->remoteNode ); + sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId ); + } + + // delete all queued messages (auto links) + if( ld->messageQueue.size() > 0 ) { + logging_warn( "Dropping link " << id.toString() << " that has " + << ld->messageQueue.size() << " waiting messages" ); + ld->flushQueue(); + } + + // erase mapping + eraseDescriptor(ld->overlayId); +} + + /// drops an established link -void BaseOverlay::dropLink(const LinkID& link) { - logging_info( "Dropping link (initiated locally):" << link.toString() ); +void BaseOverlay::dropLink(const LinkID& link) +{ + logging_info( "Dropping link: " << link.toString() ); // find the link item to drop LinkDescriptor* ld = getDescriptor(link); - if( ld == NULL ) { + if( ld == NULL ) + { logging_warn( "Can't drop link, link is unknown!"); return; @@ -827,25 +1135,73 @@ // delete all queued messages - if( ld->messageQueue.size() > 0 ) { + if( ld->messageQueue.size() > 0 ) + { logging_warn( "Dropping link " << ld->overlayId.toString() << " that has " << ld->messageQueue.size() << " waiting messages" ); ld->flushQueue(); } - - // inform sideport and listener - if(ld->listener != NULL) - ld->listener->onLinkDown( ld->overlayId, ld->remoteNode ); - sideport->onLinkDown(ld->overlayId, this->nodeId, ld->remoteNode, this->spovnetId ); - - // do not drop relay links - if (!ld->relaying) { - // drop the link in base communication - if (ld->communicationUp) bc->dropLink( ld->communicationId ); - - // erase descriptor - eraseDescriptor( ld->overlayId ); - } else { - ld->dropAfterRelaying = true; - } + + + // inform application and remote note (but only once) + // NOTE: If we initiated the drop, this function is called twice, but on + // the second call, there is noting to do. + if ( ld->up && ! ld->failed ) + { + // inform sideport and listener + if(ld->listener != NULL) + { + ld->listener->onLinkDown( ld->overlayId, ld->remoteNode ); + } + sideport->onLinkDown(ld->overlayId, this->nodeId, ld->remoteNode, this->spovnetId ); + + // send link-close to remote node + logging_info("Sending LinkClose message to remote node."); + OverlayMsg close_msg(OverlayMsg::typeLinkClose); + send_link(&close_msg, link, system_priority::OVERLAY); + + // deactivate link + ld->up = false; +// ld->closing = true; + } + + else if ( ld->failed ) + { + // inform listener + if( ld->listener != NULL ) + { + ld->listener->onLinkFail( ld->overlayId, ld->remoteNode ); + } + + ld->up = false; + __removeDroppedLink(ld->overlayId); + } +} + +/// called from typeLinkClose-handler +void BaseOverlay::__removeDroppedLink(const LinkID& link) +{ + // find the link item to drop + LinkDescriptor* ld = getDescriptor(link); + if( ld == NULL ) + { + return; + } + + // do not drop relay links + if (!ld->relaying) + { + // drop the link in base communication + if (ld->communicationUp) + { + bc->dropLink( ld->communicationId ); + } + + // erase descriptor + eraseDescriptor( ld->overlayId ); + } + else + { + ld->dropAfterRelaying = true; + } } @@ -853,40 +1209,86 @@ /// internal send message, always use this functions to send messages over links -seqnum_t BaseOverlay::sendMessage( const Message* message, const LinkID& link ) { +const SequenceNumber& BaseOverlay::sendMessage( reboost::message_t message, + const LinkID& link, + uint8_t priority ) throw(message_not_sent) +{ logging_debug( "Sending data message on link " << link.toString() ); // get the mapping for this link LinkDescriptor* ld = getDescriptor(link); - if( ld == NULL ) { - logging_error("Could not send message. " - << "Link not found id=" << link.toString()); - return -1; + if( ld == NULL ) + { + throw message_not_sent("Could not send message. Link not found id=" + link.toString()); } // check if the link is up yet, if its an auto link queue message - if( !ld->up ) { + if( !ld->up ) + { ld->setAutoUsed(); - if( ld->autolink ) { + if( ld->autolink ) + { logging_info("Auto-link " << link.toString() << " not up, queue message"); - Data data = data_serialize( message ); - const_cast(message)->dropPayload(); - ld->messageQueue.push_back( new Message(data) ); - } else { - logging_error("Link " << link.toString() << " not up, drop message"); - } - return -1; - } - - // compile overlay message (has service and node id) - OverlayMsg overmsg( OverlayMsg::typeData ); - overmsg.encapsulate( const_cast(message) ); - - // send message over relay/direct/overlay - return send_link( &overmsg, ld->overlayId ); -} - - -seqnum_t BaseOverlay::sendMessage(const Message* message, - const NodeID& node, const ServiceID& service) { + + // queue message + LinkDescriptor::message_queue_entry msg; + msg.message = message; + msg.priority = priority; + + ld->messageQueue.push_back( msg ); + + return SequenceNumber::DISABLED; // TODO what to return if message is queued? + } + else + { + throw message_not_sent("Link " + link.toString() + " not up, drop message"); + } + } + + // TODO AKTUELL: sequence numbers + // TODO seqnum on fast path ? + ld->last_sent_seqnum.increment(); + + /* choose fast-path for direct links; normal overlay-path otherwise */ + // BRANCH: direct link + if ( ld->communicationUp && !ld->relayed ) + { + // * send down to BaseCommunication * + try + { + bc->sendMessage(ld->communicationId, message, priority, true); + } + catch ( communication::communication_message_not_sent& e ) + { + ostringstream out; + out << "Communication message on fast-path not sent: " << e.what(); + throw message_not_sent(out.str()); + } + } + + // BRANCH: use (slow) overlay-path + else + { + // compile overlay message (has service and node id) + OverlayMsg overmsg( OverlayMsg::typeData ); + overmsg.set_payload_message(message); + + // set SeqNum + if ( ld->transmit_seqnums ) + { + overmsg.setSeqNum(ld->last_sent_seqnum); + } + logging_debug("Sending Message with SeqNum: " << overmsg.getSeqNum()); + + // send message over relay/direct/overlay + send_link( &overmsg, ld->overlayId, priority ); + } + + // return seqnum + return ld->last_sent_seqnum; +} + + +const SequenceNumber& BaseOverlay::sendMessage(reboost::message_t message, + const NodeID& node, uint8_t priority, const ServiceID& service) { // find link for node and service @@ -907,5 +1309,5 @@ if( ld == NULL ) { logging_error( "Failed to establish auto-link."); - return -1; + throw message_not_sent("Failed to establish auto-link."); } ld->autolink = true; @@ -920,10 +1322,10 @@ // send / queue message - return sendMessage( message, ld->overlayId ); -} - - -NodeID BaseOverlay::sendMessageCloserToNodeID(const Message* message, - const NodeID& address, const ServiceID& service) { + return sendMessage( message, ld->overlayId, priority ); +} + + +NodeID BaseOverlay::sendMessageCloserToNodeID(reboost::message_t message, + const NodeID& address, uint8_t priority, const ServiceID& service) { if ( overlayInterface->isClosestNodeTo(address) ) @@ -936,8 +1338,8 @@ if ( closest_node != NodeID::UNSPECIFIED ) { - seqnum_t seqnum = sendMessage(message, closest_node, service); + sendMessage(message, closest_node, priority, service); } - return closest_node; // XXX return seqnum ?? tuple? closest_node via (non const) reference? + return closest_node; // return seqnum ?? tuple? closest_node via (non const) reference? } // ---------------------------------------------------------------------------- @@ -978,5 +1380,5 @@ // see if we can find the node in our own table - BOOST_FOREACH(const LinkDescriptor* ld, links){ + foreach(const LinkDescriptor* ld, links){ if(ld->remoteNode != node) continue; if(!ld->communicationUp) continue; @@ -1079,5 +1481,6 @@ void BaseOverlay::onLinkUp(const LinkID& id, - const address_v* local, const address_v* remote) { + const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote) +{ logging_debug( "Link up with base communication link id=" << id ); @@ -1085,8 +1488,9 @@ LinkDescriptor* ld = getDescriptor(id, true); - // handle bootstrap link we initiated + // BRANCH: handle bootstrap link we initiated if( std::find(bootstrapLinks.begin(), bootstrapLinks.end(), id) != bootstrapLinks.end() ){ logging_info( "Join has been initiated by me and the link is now up. " << + "LinkID: " << id.toString() << "Sending out join request for SpoVNet " << spovnetId.toString() ); @@ -1096,10 +1500,12 @@ OverlayInterface::OVERLAY_SERVICE_ID, nodeId ); JoinRequest joinRequest( spovnetId, nodeId ); - overlayMsg.encapsulate( &joinRequest ); - bc->sendMessage( id, &overlayMsg ); + overlayMsg.append_buffer(joinRequest.serialize_into_shared_buffer()); + + send_overlaymessage_down(&overlayMsg, id, system_priority::OVERLAY); + return; } - // no link found? -> link establishment from remote, add one! + // BRANCH: link establishment from remote, add one! if (ld == NULL) { ld = addDescriptor( id ); @@ -1115,7 +1521,9 @@ // in this case, do not inform listener, since service it unknown // -> wait for update message! - - // link mapping found? -> send update message with node-id and service id - } else { + } + + // BRANCH: We requested this link in the first place + else + { logging_info( "onLinkUp descriptor (initiated locally):" << ld ); @@ -1126,14 +1534,34 @@ ld->fromRemote = false; - // if link is a relayed link->convert to direct link - if (ld->relayed) { - logging_info( "Converting to direct link: " << ld ); + // BRANCH: this was a relayed link before --> convert to direct link + // TODO do we really have to send a message here? + if (ld->relayed) + { ld->up = true; ld->relayed = false; + logging_info( "Converting to direct link: " << ld ); + + // send message OverlayMsg overMsg( OverlayMsg::typeLinkDirect ); overMsg.setSourceLink( ld->overlayId ); overMsg.setDestinationLink( ld->remoteLink ); - send_link( &overMsg, ld->overlayId ); - } else { + send_link( &overMsg, ld->overlayId, system_priority::OVERLAY ); + + // inform listener + if( ld->listener != NULL) + ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode ); + } + + + /* NOTE: Chord is opening direct-links in it's setup routine which are + * neither set to "relayed" nor to "up". To activate these links a + * typeLinkUpdate must be sent. + * + * This branch is would also be taken when we had a working link before + * (ld->up == true). I'm not sure if this case does actually happen + * and whether it's tested. + */ + else + { // note: necessary to validate the link on the remote side! logging_info( "Sending out update" << @@ -1144,7 +1572,15 @@ // compile and send update message OverlayMsg overlayMsg( OverlayMsg::typeLinkUpdate ); - overlayMsg.setSourceLink(ld->overlayId); overlayMsg.setAutoLink( ld->autolink ); - send_link( &overlayMsg, ld->overlayId, true ); + overlayMsg.setSourceNode(nodeId); + overlayMsg.setDestinationNode(ld->remoteNode); + overlayMsg.setSourceLink(ld->overlayId); + overlayMsg.setDestinationLink(ld->remoteLink); + overlayMsg.setService(ld->service); + overlayMsg.setRelayed(false); + + // TODO ld->communicationId = id ?? + + send_overlaymessage_down(&overlayMsg, id, system_priority::OVERLAY); } } @@ -1152,6 +1588,7 @@ void BaseOverlay::onLinkDown(const LinkID& id, - const address_v* local, const address_v* remote) { - + const addressing2::EndpointPtr local, + const addressing2::EndpointPtr remote) +{ // erase bootstrap links vector::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id ); @@ -1185,61 +1622,111 @@ } + +void BaseOverlay::onLinkFail(const LinkID& id, + const addressing2::EndpointPtr local, + const addressing2::EndpointPtr remote) +{ + logging_debug( "Link fail with base communication link id=" << id ); + +// // erase bootstrap links +// vector::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id ); +// if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it ); +// +// // get descriptor for link +// LinkDescriptor* ld = getDescriptor(id, true); +// if ( ld == NULL ) return; // not found? ->ignore! +// logging_debug( "Link failed id=" << ld->overlayId.toString() ); +// +// // inform listeners +// ld->listener->onLinkFail( ld->overlayId, ld->remoteNode ); +// sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId ); + + logging_debug( " ... calling onLinkDown ..." ); + onLinkDown(id, local, remote); +} + + void BaseOverlay::onLinkChanged(const LinkID& id, - const address_v* oldlocal, const address_v* newlocal, - const address_v* oldremote, const address_v* newremote) { - - // get descriptor for link - LinkDescriptor* ld = getDescriptor(id, true); - if ( ld == NULL ) return; // not found? ->ignore! - logging_debug( "onLinkChanged descriptor: " << ld ); - - // inform listeners - ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode ); - sideport->onLinkChanged( id, this->nodeId, ld->remoteNode, this->spovnetId ); - - // autolinks: refresh timestamp - ld->setAutoUsed(); -} - -void BaseOverlay::onLinkFail(const LinkID& id, - const address_v* local, const address_v* remote) { - logging_debug( "Link fail with base communication link id=" << id ); - - // erase bootstrap links - vector::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id ); - if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it ); - - // get descriptor for link - LinkDescriptor* ld = getDescriptor(id, true); - if ( ld == NULL ) return; // not found? ->ignore! - logging_debug( "Link failed id=" << ld->overlayId.toString() ); - - // inform listeners - ld->listener->onLinkFail( ld->overlayId, ld->remoteNode ); - sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId ); -} - -void BaseOverlay::onLinkQoSChanged(const LinkID& id, const address_v* local, - const address_v* remote, const QoSParameterSet& qos) { - logging_debug( "Link quality changed with base communication link id=" << id ); - - // get descriptor for link - LinkDescriptor* ld = getDescriptor(id, true); - if ( ld == NULL ) return; // not found? ->ignore! - logging_debug( "Link quality changed id=" << ld->overlayId.toString() ); -} - -bool BaseOverlay::onLinkRequest( const LinkID& id, const address_v* local, - const address_v* remote ) { + const addressing2::EndpointPtr oldlocal, const addressing2::EndpointPtr newlocal, + const addressing2::EndpointPtr oldremote, const addressing2::EndpointPtr newremote) +{ + // get descriptor for link + LinkDescriptor* ld = getDescriptor(id, true); + if ( ld == NULL ) return; // not found? ->ignore! + logging_debug( "onLinkChanged descriptor: " << ld ); + + // inform listeners + ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode ); + sideport->onLinkChanged( id, this->nodeId, ld->remoteNode, this->spovnetId ); + + // autolinks: refresh timestamp + ld->setAutoUsed(); +} + +//void BaseOverlay::onLinkQoSChanged(const LinkID& id, +// const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote, +// const QoSParameterSet& qos) +//{ +// logging_debug( "Link quality changed with base communication link id=" << id ); +// +// // get descriptor for link +// LinkDescriptor* ld = getDescriptor(id, true); +// if ( ld == NULL ) return; // not found? ->ignore! +// logging_debug( "Link quality changed id=" << ld->overlayId.toString() ); +//} + +bool BaseOverlay::onLinkRequest(const LinkID& id, + const addressing2::EndpointPtr local, + const addressing2::EndpointPtr remote) +{ logging_debug("Accepting link request from " << remote->to_string() ); + + // TODO ask application..? + return true; } + + + /// handles a message from base communication -bool BaseOverlay::receiveMessage(const Message* message, - const LinkID& link, const NodeID& ) { +bool BaseOverlay::receiveMessage( reboost::shared_buffer_t message, + const LinkID& link, + const NodeID&, + bool bypass_overlay ) +{ // get descriptor for link LinkDescriptor* ld = getDescriptor( link, true ); - return handleMessage( message, ld, link ); + + + /* choose fastpath for direct links; normal overlay-path otherwise */ + if ( bypass_overlay && ld ) + { + // message received --> link is alive + ld->keepAliveReceived = time(NULL); + // hop count on this link + ld->hops = 0; + + + // hand over to CommunicationListener (aka Application) + CommunicationListener* lst = getListener(ld->service); + if ( lst != NULL ) + { + lst->onMessage( + message, + ld->remoteNode, + ld->overlayId, + SequenceNumber::DISABLED, + NULL ); + + return true; + } + + return false; + } + else + { + return handleMessage( message, ld, link ); + } } @@ -1247,17 +1734,19 @@ /// Handle spovnet instance join requests -bool BaseOverlay::handleJoinRequest( OverlayMsg* overlayMsg, const LinkID& bcLink ) { - +bool BaseOverlay::handleJoinRequest( reboost::shared_buffer_t message, const NodeID& source, const LinkID& bcLink ) +{ // decapsulate message - JoinRequest* joinReq = overlayMsg->decapsulate(); + JoinRequest joinReq; + joinReq.deserialize_from_shared_buffer(message); + logging_info( "Received join request for spovnet " << - joinReq->getSpoVNetID().toString() ); + joinReq.getSpoVNetID().toString() ); // check spovnet id - if( joinReq->getSpoVNetID() != spovnetId ) { + if( joinReq.getSpoVNetID() != spovnetId ) { logging_error( "Received join request for spovnet we don't handle " << - joinReq->getSpoVNetID().toString() ); - delete joinReq; + joinReq.getSpoVNetID().toString() ); + return false; } @@ -1267,7 +1756,7 @@ logging_info( "Sending join reply for spovnet " << spovnetId.toString() << " to node " << - overlayMsg->getSourceNode().toString() << + source.toString() << ". Result: " << (allow ? "allowed" : "denied") ); - joiningNodes.push_back( overlayMsg->getSourceNode() ); + joiningNodes.push_back( source ); // return overlay parameters @@ -1276,32 +1765,44 @@ << getEndpointDescriptor().toString() ) OverlayParameterSet parameters = overlayInterface->getParameters(); + + + // create JoinReplay Message OverlayMsg retmsg( OverlayMsg::typeJoinReply, OverlayInterface::OVERLAY_SERVICE_ID, nodeId ); - JoinReply replyMsg( spovnetId, parameters, - allow, getEndpointDescriptor() ); - retmsg.encapsulate(&replyMsg); - bc->sendMessage( bcLink, &retmsg ); - - delete joinReq; + JoinReply replyMsg( spovnetId, parameters, allow ); + retmsg.append_buffer(replyMsg.serialize_into_shared_buffer()); + + // XXX This is unlovely clash between the old message system and the new one, + // but a.t.m. we can't migrate everything to the new system at once.. + // ---> Consider the EndpointDescriptor as part of the JoinReply.. + retmsg.append_buffer(getEndpointDescriptor().serialize()); + + // * send * + send_overlaymessage_down(&retmsg, bcLink, system_priority::OVERLAY); + return true; } /// Handle replies to spovnet instance join requests -bool BaseOverlay::handleJoinReply( OverlayMsg* overlayMsg, const LinkID& bcLink ) { +bool BaseOverlay::handleJoinReply( reboost::shared_buffer_t message, const LinkID& bcLink ) +{ // decapsulate message logging_debug("received join reply message"); - JoinReply* replyMsg = overlayMsg->decapsulate(); + JoinReply replyMsg; + EndpointDescriptor endpoints; + reboost::shared_buffer_t buff = replyMsg.deserialize_from_shared_buffer(message); + buff = endpoints.deserialize(buff); // correct spovnet? - if( replyMsg->getSpoVNetID() != spovnetId ) { // no-> fail + if( replyMsg.getSpoVNetID() != spovnetId ) { // no-> fail logging_error( "Received SpoVNet join reply for " << - replyMsg->getSpoVNetID().toString() << + replyMsg.getSpoVNetID().toString() << " != " << spovnetId.toString() ); - delete replyMsg; + return false; } // access granted? no -> fail - if( !replyMsg->getJoinAllowed() ) { + if( !replyMsg.getJoinAllowed() ) { logging_error( "Our join request has been denied" ); @@ -1317,8 +1818,7 @@ // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ) + foreach( NodeListener* i, nodeListeners ) i->onJoinFailed( spovnetId ); - delete replyMsg; return true; } @@ -1329,5 +1829,5 @@ logging_debug( "Using bootstrap end-point " - << replyMsg->getBootstrapEndpoint().toString() ); + << endpoints.toString() ); // create overlay structure from spovnet parameter set @@ -1338,5 +1838,5 @@ overlayInterface = OverlayFactory::create( - *this, replyMsg->getParam(), nodeId, this ); + *this, replyMsg.getParam(), nodeId, this ); // overlay structure supported? no-> fail! @@ -1354,8 +1854,7 @@ // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ) + foreach( NodeListener* i, nodeListeners ) i->onJoinFailed( spovnetId ); - delete replyMsg; return true; } @@ -1365,6 +1864,6 @@ overlayInterface->createOverlay(); - overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() ); - overlayBootstrap.recordJoin( replyMsg->getBootstrapEndpoint() ); + overlayInterface->joinOverlay( endpoints ); + overlayBootstrap.recordJoin( endpoints ); // update ovlvis @@ -1372,18 +1871,13 @@ // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ) - i->onJoinCompleted( spovnetId ); - - delete replyMsg; - - } else { - + foreach( NodeListener* i, nodeListeners ) + i->onJoinCompleted( spovnetId ); + } + else + { // this is not the first bootstrap, just join the additional node logging_debug("not first-time bootstrapping"); - overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() ); - overlayBootstrap.recordJoin( replyMsg->getBootstrapEndpoint() ); - - delete replyMsg; - + overlayInterface->joinOverlay( endpoints ); + overlayBootstrap.recordJoin( endpoints ); } // if( overlayInterface == NULL ) @@ -1392,7 +1886,9 @@ -bool BaseOverlay::handleData( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { +bool BaseOverlay::handleData( reboost::shared_buffer_t message, OverlayMsg* overlayMsg, LinkDescriptor* ld ) +{ // get service - const ServiceID& service = overlayMsg->getService(); + const ServiceID& service = ld->service; //overlayMsg->getService(); + logging_debug( "Received data for service " << service.toString() << " on link " << overlayMsg->getDestinationLink().toString() ); @@ -1402,7 +1898,11 @@ if(lst != NULL){ lst->onMessage( - overlayMsg, - overlayMsg->getSourceNode(), - overlayMsg->getDestinationLink() + message, +// overlayMsg->getSourceNode(), +// overlayMsg->getDestinationLink(), + ld->remoteNode, + ld->overlayId, + overlayMsg->getSeqNum(), + overlayMsg ); } @@ -1411,4 +1911,105 @@ } +bool BaseOverlay::handleLostMessage( reboost::shared_buffer_t message, OverlayMsg* msg ) +{ + /** + * Deserialize MessageLost-Message + * + * - Type of lost message + * - Hop count of lost message + * - Source-LinkID of lost message + */ + const uint8_t* buff = message(0, sizeof(uint8_t)*2).data(); + uint8_t type = buff[0]; + uint8_t hops = buff[1]; + LinkID linkid; + linkid.deserialize(message(sizeof(uint8_t)*2)); + + logging_warn("Node " << msg->getSourceNode() + << " informed us, that our message of type " << (int) type + << " is lost after traveling " << (int) hops << " hops." + << " (LinkID: " << linkid.toString()); + + + // TODO switch-case ? + + // BRANCH: LinkRequest --> link request failed + if ( type == OverlayMsg::typeLinkRequest ) + { + __onLinkEstablishmentFailed(linkid); + } + + // BRANCH: Data --> link disrupted. Drop link. + // (We could use something more advanced here. e.g. At least send a + // keep-alive message and wait for a keep-alive reply.) + if ( type == OverlayMsg::typeData ) + { + LinkDescriptor* link_desc = getDescriptor(linkid); + + if ( link_desc ) + { + link_desc->failed = true; + } + + dropLink(linkid); + } + + // BRANCH: ping lost + if ( type == OverlayMsg::typePing ) + { + CommunicationListener* lst = getListener(msg->getService()); + if( lst != NULL ) + { + lst->onPingLost(msg->getSourceNode()); + } + } + + return true; +} + +bool BaseOverlay::handlePing( OverlayMsg* overlayMsg, LinkDescriptor* ld ) +{ + // TODO AKTUELL: implement interfaces: Node::ping(node); BaseOverlay::ping(node) + + bool send_pong = false; + + // inform application and ask permission to send a pong message + CommunicationListener* lst = getListener(overlayMsg->getService()); + if( lst != NULL ) + { + send_pong = lst->onPing(overlayMsg->getSourceNode()); + } + + // send pong message if allowed + if ( send_pong ) + { + OverlayMsg pong_msg(OverlayMsg::typePong); + pong_msg.setSeqNum(overlayMsg->getSeqNum()); + + // send message + try + { + send_node( &pong_msg, + overlayMsg->getSourceNode(), + system_priority::OVERLAY, + overlayMsg->getService() ); + } + catch ( message_not_sent& e ) + { + logging_info("Could not send Pong-Message to node: " << + overlayMsg->getSourceNode()); + } + } +} + +bool BaseOverlay::handlePong( OverlayMsg* overlayMsg, LinkDescriptor* ld ) +{ + // inform application + CommunicationListener* lst = getListener(overlayMsg->getService()); + if( lst != NULL ) + { + lst->onPong(overlayMsg->getSourceNode()); + } +} bool BaseOverlay::handleLinkUpdate( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { @@ -1439,5 +2040,5 @@ overlayMsg->setSourceLink(ld->overlayId); overlayMsg->setService(ld->service); - send( overlayMsg, ld ); + send( overlayMsg, ld, system_priority::OVERLAY ); } @@ -1481,8 +2082,8 @@ if( ld->messageQueue.size() > 0 ) { logging_info( "Sending out queued messages on link " << ld ); - BOOST_FOREACH( Message* msg, ld->messageQueue ) { - sendMessage( msg, ld->overlayId ); - delete msg; - } + foreach( LinkDescriptor::message_queue_entry msg, ld->messageQueue ) + { + sendMessage( msg.message, ld->overlayId, msg.priority ); + } ld->messageQueue.clear(); } @@ -1497,5 +2098,4 @@ /// handle a link request and reply bool BaseOverlay::handleLinkRequest( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { - logging_info( "Link request received from node id=" << overlayMsg->getSourceNode() ); //TODO: Check if a request has already been sent using getSourceLink() ... @@ -1514,16 +2114,33 @@ ldn->remoteNode = overlayMsg->getSourceNode(); ldn->remoteLink = overlayMsg->getSourceLink(); - + ldn->hops = overlayMsg->getNumHops(); + + // initialize sequence numbers + ldn->last_sent_seqnum = SequenceNumber::createRandomSeqNum_Short(); + logging_debug("Creating new link with initial SeqNum: " << ldn->last_sent_seqnum); + + // update time-stamps ldn->setAlive(); ldn->setAutoUsed(); + logging_info( "Link request received from node id=" + << overlayMsg->getSourceNode() + << " LINK: " + << ldn); + // create reply message and send back! overlayMsg->swapRoles(); // swap source/destination overlayMsg->setType(OverlayMsg::typeLinkReply); overlayMsg->setSourceLink(ldn->overlayId); - overlayMsg->setSourceEndpoint( bc->getEndpointDescriptor() ); overlayMsg->setRelayed(true); - send( overlayMsg, ld ); // send back to link +// overlayMsg->setRouteRecord(true); + overlayMsg->setSeqNum(ld->last_sent_seqnum); + + // TODO aktuell do the same thing in the typeLinkRequest-Message, too. But be careful with race conditions!! + // append our endpoints (for creation of a direct link) + overlayMsg->set_payload_message(bc->getEndpointDescriptor().serialize()); + + send( overlayMsg, ld, system_priority::OVERLAY ); // send back to link // inform listener @@ -1534,6 +2151,13 @@ } -bool BaseOverlay::handleLinkReply( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { - +bool BaseOverlay::handleLinkReply( + OverlayMsg* overlayMsg, + reboost::shared_buffer_t sub_message, + LinkDescriptor* ld ) +{ + // deserialize EndpointDescriptor + EndpointDescriptor endpoints; + endpoints.deserialize(sub_message); + // find link request LinkDescriptor* ldn = getDescriptor(overlayMsg->getDestinationLink()); @@ -1554,9 +2178,10 @@ // debug message - logging_debug( "Link request reply received. Establishing link" + logging_info( "Link request reply received. Establishing link" << " for service " << overlayMsg->getService().toString() << " with local id=" << overlayMsg->getDestinationLink() << " and remote link id=" << overlayMsg->getSourceLink() - << " to " << overlayMsg->getSourceEndpoint().toString() + << " to " << endpoints.toString() + << " hop count: " << overlayMsg->getRouteRecord().size() ); @@ -1577,7 +2202,7 @@ logging_info( "Sending out queued messages on link " << ldn->overlayId.toString() ); - BOOST_FOREACH( Message* msg, ldn->messageQueue ) { - sendMessage( msg, ldn->overlayId ); - delete msg; + foreach( LinkDescriptor::message_queue_entry msg, ldn->messageQueue ) + { + sendMessage( msg.message, ldn->overlayId, msg.priority ); } ldn->messageQueue.clear(); @@ -1589,5 +2214,5 @@ // try to replace relay link with direct link ldn->retryCounter = 3; - ldn->endpoint = overlayMsg->getSourceEndpoint(); + ldn->endpoint = endpoints; ldn->communicationId = bc->establishLink( ldn->endpoint ); @@ -1596,15 +2221,42 @@ /// handle a keep-alive message for a link -bool BaseOverlay::handleLinkAlive( OverlayMsg* overlayMsg, LinkDescriptor* ld ) { +bool BaseOverlay::handleLinkAlive( OverlayMsg* overlayMsg, LinkDescriptor* ld ) +{ LinkDescriptor* rld = getDescriptor(overlayMsg->getDestinationLink()); - if ( rld != NULL ) { - logging_debug("Keep-Alive for " << - overlayMsg->getDestinationLink() ); + + if ( rld != NULL ) + { + logging_debug("Keep-Alive for " << overlayMsg->getDestinationLink() ); if (overlayMsg->isRouteRecord()) + { rld->routeRecord = overlayMsg->getRouteRecord(); + } + + // set alive rld->setAlive(); + + + /* answer keep alive */ + if ( overlayMsg->getType() == OverlayMsg::typeKeepAlive ) + { + time_t now = time(NULL); + logging_debug("[BaseOverlay] Answering KeepAlive over " + << ld->to_string() + << " after " + << difftime( now, ld->keepAliveSent ) + << "s"); + + OverlayMsg msg( OverlayMsg::typeKeepAliveReply, + OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode ); + msg.setRouteRecord(true); + ld->keepAliveSent = now; + send_link( &msg, ld->overlayId, system_priority::OVERLAY ); + } + return true; - } else { - logging_error("Keep-Alive for " + } + else + { + logging_error("No Keep-Alive for " << overlayMsg->getDestinationLink() << ": link unknown." ); return false; @@ -1636,17 +2288,28 @@ // erase the original descriptor eraseDescriptor(ld->overlayId); + + // inform listener + if( rld->listener != NULL) + rld->listener->onLinkChanged( rld->overlayId, rld->remoteNode ); + return true; } /// handles an incoming message -bool BaseOverlay::handleMessage( const Message* message, LinkDescriptor* ld, - const LinkID bcLink ) { - logging_debug( "Handling message: " << message->toString()); - +bool BaseOverlay::handleMessage( reboost::shared_buffer_t message, LinkDescriptor* ld, + const LinkID bcLink ) +{ // decapsulate overlay message - OverlayMsg* overlayMsg = - const_cast(message)->decapsulate(); - if( overlayMsg == NULL ) return false; - + OverlayMsg* overlayMsg = new OverlayMsg(); + reboost::shared_buffer_t sub_buff = overlayMsg->deserialize_from_shared_buffer(message); + +// // XXX debug +// logging_info( "Received overlay message." +// << " Hops: " << (int) overlayMsg->getNumHops() +// << " Type: " << (int) overlayMsg->getType() +// << " Payload size: " << sub_buff.size() +// << " SeqNum: " << overlayMsg->getSeqNum() ); + + // increase number of hops overlayMsg->increaseNumHops(); @@ -1660,6 +2323,7 @@ // handle signaling messages (do not route!) if (overlayMsg->getType()>=OverlayMsg::typeSignalingStart && - overlayMsg->getType()<=OverlayMsg::typeSignalingEnd ) { - overlayInterface->onMessage(overlayMsg, NodeID::UNSPECIFIED, LinkID::UNSPECIFIED); + overlayMsg->getType()<=OverlayMsg::typeSignalingEnd ) + { + overlayInterface->onMessage(overlayMsg, sub_buff, NodeID::UNSPECIFIED, LinkID::UNSPECIFIED); delete overlayMsg; return true; @@ -1673,42 +2337,110 @@ << " to " << overlayMsg->getDestinationNode() ); - route( overlayMsg ); + +// // XXX testing AKTUELL +// logging_info("MARIO: Routing message " +// << " from " << overlayMsg->getSourceNode() +// << " to " << overlayMsg->getDestinationNode() ); +// logging_info( "Type: " << overlayMsg->getType() << " Payload size: " << sub_buff.size()); + overlayMsg->append_buffer(sub_buff); + + route( overlayMsg, ld->remoteNode ); delete overlayMsg; return true; } - // handle base overlay message + + /* handle base overlay message */ bool ret = false; // return value - switch ( overlayMsg->getType() ) { - - // data transport messages - case OverlayMsg::typeData: - ret = handleData(overlayMsg, ld); break; - - // overlay setup messages - case OverlayMsg::typeJoinRequest: - ret = handleJoinRequest(overlayMsg, bcLink ); break; - case OverlayMsg::typeJoinReply: - ret = handleJoinReply(overlayMsg, bcLink ); break; - - // link specific messages - case OverlayMsg::typeLinkRequest: - ret = handleLinkRequest(overlayMsg, ld ); break; - case OverlayMsg::typeLinkReply: - ret = handleLinkReply(overlayMsg, ld ); break; - case OverlayMsg::typeLinkUpdate: - ret = handleLinkUpdate(overlayMsg, ld ); break; - case OverlayMsg::typeLinkAlive: - ret = handleLinkAlive(overlayMsg, ld ); break; - case OverlayMsg::typeLinkDirect: - ret = handleLinkDirect(overlayMsg, ld ); break; - - // handle unknown message type - default: { - logging_error( "received message in invalid state! don't know " << - "what to do with this message of type " << overlayMsg->getType() ); - ret = false; - break; - } + try + { + switch ( overlayMsg->getType() ) + { + // data transport messages + case OverlayMsg::typeData: + { + // NOTE: On relayed links, »ld« does not point to our link, but on the relay link. + LinkDescriptor* end_to_end_ld = getDescriptor(overlayMsg->getDestinationLink()); + + if ( ! end_to_end_ld ) + { + logging_warn("Error: Data-Message claims to belong to a link we don't know."); + + ret = false; + } + else + { + // message received --> link is alive + end_to_end_ld->keepAliveReceived = time(NULL); + // hop count on this link + end_to_end_ld->hops = overlayMsg->getNumHops(); + + // * call handler * + ret = handleData(sub_buff, overlayMsg, end_to_end_ld); + } + + break; + } + case OverlayMsg::typeMessageLost: + ret = handleLostMessage(sub_buff, overlayMsg); + + break; + + // overlay setup messages + case OverlayMsg::typeJoinRequest: + ret = handleJoinRequest(sub_buff, overlayMsg->getSourceNode(), bcLink ); break; + case OverlayMsg::typeJoinReply: + ret = handleJoinReply(sub_buff, bcLink ); break; + + // link specific messages + case OverlayMsg::typeLinkRequest: + ret = handleLinkRequest(overlayMsg, ld ); break; + case OverlayMsg::typeLinkReply: + ret = handleLinkReply(overlayMsg, sub_buff, ld ); break; + case OverlayMsg::typeLinkUpdate: + ret = handleLinkUpdate(overlayMsg, ld ); break; + case OverlayMsg::typeKeepAlive: + case OverlayMsg::typeKeepAliveReply: + ret = handleLinkAlive(overlayMsg, ld ); break; + case OverlayMsg::typeLinkDirect: + ret = handleLinkDirect(overlayMsg, ld ); break; + + case OverlayMsg::typeLinkClose: + { + dropLink(overlayMsg->getDestinationLink()); + __removeDroppedLink(overlayMsg->getDestinationLink()); + + break; + } + + /// ping over overlay path (or similar) + case OverlayMsg::typePing: + { + ret = handlePing(overlayMsg, ld); + break; + } + case OverlayMsg::typePong: + { + ret = handlePong(overlayMsg, ld); + break; + } + + // handle unknown message type + default: + { + logging_error( "received message in invalid state! don't know " << + "what to do with this message of type " << overlayMsg->getType() ); + ret = false; + break; + } + } + } + catch ( reboost::illegal_sub_buffer& e ) + { + logging_error( "Failed to create sub-buffer while reading message: »" + << e.what() + << "« Message too short? "); + + assert(false); // XXX } @@ -1720,11 +2452,8 @@ // ---------------------------------------------------------------------------- -void BaseOverlay::broadcastMessage(Message* message, const ServiceID& service) { +void BaseOverlay::broadcastMessage(reboost::message_t message, const ServiceID& service, uint8_t priority) { logging_debug( "broadcasting message to all known nodes " << "in the overlay from service " + service.toString() ); - - if(message == NULL) return; - message->setReleasePayload(false); OverlayInterface::NodeList nodes = overlayInterface->getKnownNodes(true); @@ -1732,6 +2461,6 @@ NodeID& id = nodes.at(i); if(id == this->nodeId) continue; // don't send to ourselfs - if(i+1 == nodes.size()) message->setReleasePayload(true); // release payload on last send - sendMessage( message, id, service ); + + sendMessage( message, id, priority, service ); } } @@ -1755,5 +2484,5 @@ vector BaseOverlay::getLinkIDs( const NodeID& nid ) const { vector linkvector; - BOOST_FOREACH( LinkDescriptor* ld, links ) { + foreach( LinkDescriptor* ld, links ) { if( ld->remoteNode == nid || nid == NodeID::UNSPECIFIED ) { linkvector.push_back( ld->overlayId ); @@ -1779,4 +2508,43 @@ updateVisual(); } + + + +/* link status */ +bool BaseOverlay::isLinkDirect(const ariba::LinkID& lnk) const +{ + const LinkDescriptor* ld = getDescriptor(lnk); + + if (!ld) + return false; + + return ld->communicationUp && !ld->relayed; +} + +int BaseOverlay::getHopCount(const ariba::LinkID& lnk) const +{ + const LinkDescriptor* ld = getDescriptor(lnk); + + if (!ld) + return -1; + + return ld->hops; +} + + +bool BaseOverlay::isLinkVital(const LinkDescriptor* link) const +{ + time_t now = time(NULL); + + return link->up && difftime( now, link->keepAliveReceived ) <= KEEP_ALIVE_TIME_OUT; // TODO is this too long for a "vital" link..? +} + +bool BaseOverlay::isLinkDirectVital(const LinkDescriptor* link) const +{ + return isLinkVital(link) && link->communicationUp && !link->relayed; +} + +/* [link status] */ + void BaseOverlay::updateVisual(){ @@ -1878,6 +2646,6 @@ static set linkset; set remotenodes; - BOOST_FOREACH( LinkDescriptor* ld, links ) { - if (!ld->isVital() || ld->service != OverlayInterface::OVERLAY_SERVICE_ID) + foreach( LinkDescriptor* ld, links ) { + if (!isLinkVital(ld) || ld->service != OverlayInterface::OVERLAY_SERVICE_ID) continue; @@ -1895,5 +2663,5 @@ do{ changed = false; - BOOST_FOREACH(NodeID n, linkset){ + foreach(NodeID n, linkset){ if(remotenodes.find(n) == remotenodes.end()){ visualInstance.visDisconnect(visualIdBase, this->nodeId, n, ""); @@ -1908,5 +2676,5 @@ do{ changed = false; - BOOST_FOREACH(NodeID n, remotenodes){ + foreach(NodeID n, remotenodes){ if(linkset.find(n) == linkset.end()){ visualInstance.visConnect(visualIdBase, this->nodeId, n, ""); @@ -1933,5 +2701,5 @@ // dump link state s << "--- link state -------------------------------" << endl; - BOOST_FOREACH( LinkDescriptor* ld, links ) { + foreach( LinkDescriptor* ld, links ) { s << "link " << i << ": " << ld << endl; i++; Index: source/ariba/overlay/BaseOverlay.h =================================================================== --- source/ariba/overlay/BaseOverlay.h (revision 10700) +++ source/ariba/overlay/BaseOverlay.h (revision 12060) @@ -47,5 +47,12 @@ #include #include +#include #include + +#ifdef ECLIPSE_PARSER + #define foreach(a, b) for(a : b) +#else + #define foreach(a, b) BOOST_FOREACH(a, b) +#endif #include "ariba/utility/messages.h" @@ -64,4 +71,5 @@ #include "ariba/overlay/modules/OverlayStructureEvents.h" #include "ariba/overlay/OverlayBootstrap.h" +#include "ariba/overlay/SequenceNumber.h" // forward declarations @@ -92,4 +100,7 @@ using ariba::communication::BaseCommunication; using ariba::communication::CommunicationEvents; + +// transport +//using ariba::transport::system_priority; // utilities @@ -103,5 +114,4 @@ using ariba::utility::Demultiplexer; using ariba::utility::MessageReceiver; -using ariba::utility::MessageSender; using ariba::utility::seqnum_t; using ariba::utility::Timer; @@ -110,5 +120,19 @@ namespace overlay { -using namespace ariba::addressing; + + +class message_not_sent: public std::runtime_error +{ +public: + /** Takes a character string describing the error. */ + explicit message_not_sent(const string& __arg) : + std::runtime_error(__arg) + { + } + + virtual ~message_not_sent() throw() {} +}; + + class LinkDescriptor; @@ -121,5 +145,5 @@ protected Timer { - friend class OneHop; +// friend class OneHop; // DEPRECATED friend class Chord; friend class ariba::SideportListener; @@ -128,5 +152,4 @@ public: - /** * Constructs an empty non-functional base overlay instance @@ -142,5 +165,5 @@ * Starts the Base Overlay instance */ - void start(BaseCommunication& _basecomm, const NodeID& _nodeid); + void start(BaseCommunication* _basecomm, const NodeID& _nodeid); /** @@ -161,5 +184,5 @@ * Starts a link establishment procedure to the specfied node * for the service with id service - * + * * @param node Destination node id * @param service Service to connect to @@ -179,10 +202,20 @@ void dropLink( const LinkID& link ); + + + /* +++++ Message sending +++++ */ + + /// sends a message over an existing link - seqnum_t sendMessage(const Message* message, const LinkID& link ); + const SequenceNumber& sendMessage(reboost::message_t message, + const LinkID& link, + uint8_t priority ) throw(message_not_sent); /// sends a message to a node and a specific service - seqnum_t sendMessage(const Message* message, const NodeID& remote, - const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID); + const SequenceNumber& sendMessage(reboost::message_t message, + const NodeID& remote, + uint8_t priority, + const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID); + /** @@ -191,6 +224,6 @@ * @return NodeID of the (closest) destination node; */ - NodeID sendMessageCloserToNodeID(const Message* message, const NodeID& address, - const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID); + NodeID sendMessageCloserToNodeID(reboost::message_t message, const NodeID& address, + uint8_t priority, const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID); /** @@ -198,6 +231,11 @@ * Depending on the structure of the overlay, this can be very different. */ - void broadcastMessage(Message* message, const ServiceID& service); - + void broadcastMessage(reboost::message_t message, const ServiceID& service, uint8_t priority); + + + /* +++++ [Message sending] +++++ */ + + + /** * Returns the end-point descriptor of a link. @@ -294,44 +332,59 @@ */ void leaveSpoVNet(); + + + /* link status */ + bool isLinkDirect(const ariba::LinkID& lnk) const; + int getHopCount(const ariba::LinkID& lnk) const; + + bool isLinkVital(const LinkDescriptor* link) const; + bool isLinkDirectVital(const LinkDescriptor* link) const; protected: - /** - * @see ariba::communication::CommunicationEvents.h - */ - virtual void onLinkUp(const LinkID& id, const address_v* local, - const address_v* remote); - - /** - * @see ariba::communication::CommunicationEvents.h - */ - virtual void onLinkDown(const LinkID& id, const address_v* local, - const address_v* remote); - - /** - * @see ariba::communication::CommunicationEvents.h - */ - virtual void onLinkChanged(const LinkID& id, - const address_v* oldlocal, const address_v* newlocal, - const address_v* oldremote, const address_v* newremote); - - /** - * @see ariba::communication::CommunicationEvents.h - */ - virtual void onLinkFail(const LinkID& id, const address_v* local, - const address_v* remote); - - /** - * @see ariba::communication::CommunicationEvents.h - */ - virtual void onLinkQoSChanged(const LinkID& id, - const address_v* local, const address_v* remote, - const QoSParameterSet& qos); - - /** - * @see ariba::communication::CommunicationEvents.h - */ - virtual bool onLinkRequest(const LinkID& id, const address_v* local, - const address_v* remote); - + + /** + * @see ariba::communication::CommunicationEvents.h + */ + virtual bool onLinkRequest(const LinkID& id, + const addressing2::EndpointPtr local, + const addressing2::EndpointPtr remote); + + /** + * @see ariba::communication::CommunicationEvents.h + */ + virtual void onLinkUp(const LinkID& id, + const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote); + + /** + * @see ariba::communication::CommunicationEvents.h + */ + virtual void onLinkDown(const LinkID& id, + const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote); + + /** + * @see ariba::communication::CommunicationEvents.h + */ + virtual void onLinkChanged(const LinkID& id, + const addressing2::EndpointPtr oldlocal, const addressing2::EndpointPtr newlocal, + const addressing2::EndpointPtr oldremote, const addressing2::EndpointPtr newremote); + + /** + * @see ariba::communication::CommunicationEvents.h + * + * NOTE: Just calls onLinkDown (at the moment..) + */ + virtual void onLinkFail(const LinkID& id, + const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote); + + /** + * @see ariba::communication::CommunicationEvents.h + */ +// virtual void onLinkQoSChanged(const LinkID& id, +// const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote, +// const QoSParameterSet& qos); + + + + /** * Processes a received message from BaseCommunication @@ -340,6 +393,8 @@ * the node the message came from! */ - virtual bool receiveMessage( const Message* message, const LinkID& link, - const NodeID& ); + virtual bool receiveMessage( reboost::shared_buffer_t message, + const LinkID& link, + const NodeID&, + bool bypass_overlay ); /** @@ -359,4 +414,12 @@ std::string getLinkHTMLInfo(); + +private: + /// NOTE: "id" is an Overlay-LinkID + void __onLinkEstablishmentFailed(const LinkID& id); + + /// called from typeLinkClose-handler + void __removeDroppedLink(const LinkID& link); + private: /// is the base overlay started yet @@ -396,21 +459,26 @@ /// demultiplexes a incoming message with link descriptor - bool handleMessage( const Message* message, LinkDescriptor* ld, + bool handleMessage( reboost::shared_buffer_t message, LinkDescriptor* ld, const LinkID bcLink = LinkID::UNSPECIFIED ); // handle data and signalling messages - bool handleData( OverlayMsg* msg, LinkDescriptor* ld ); + bool handleData( reboost::shared_buffer_t message, OverlayMsg* msg, LinkDescriptor* ld ); + bool handleLostMessage( reboost::shared_buffer_t message, OverlayMsg* msg ); bool handleSignaling( OverlayMsg* msg, LinkDescriptor* ld ); // handle join request / reply messages - bool handleJoinRequest( OverlayMsg* msg, const LinkID& bcLink ); - bool handleJoinReply( OverlayMsg* msg, const LinkID& bcLink ); + bool handleJoinRequest( reboost::shared_buffer_t message, const NodeID& source, const LinkID& bcLink ); + bool handleJoinReply( reboost::shared_buffer_t message, const LinkID& bcLink ); // handle link messages bool handleLinkRequest( OverlayMsg* msg, LinkDescriptor* ld ); - bool handleLinkReply( OverlayMsg* msg, LinkDescriptor* ld ); + bool handleLinkReply( OverlayMsg* msg, reboost::shared_buffer_t sub_message, LinkDescriptor* ld ); bool handleLinkUpdate( OverlayMsg* msg, LinkDescriptor* ld ); bool handleLinkDirect( OverlayMsg* msg, LinkDescriptor* ld ); bool handleLinkAlive( OverlayMsg* msg, LinkDescriptor* ld ); + + // ping-pong over overlaypath/routing + bool handlePing( OverlayMsg* overlayMsg, LinkDescriptor* ld ); + bool handlePong( OverlayMsg* overlayMsg, LinkDescriptor* ld ); @@ -478,24 +546,40 @@ // internal message delivery ----------------------------------------------- + // Convert OverlayMessage into new format and give it down to BaseCommunication + seqnum_t send_overlaymessage_down( OverlayMsg* message, const LinkID& bc_link, uint8_t priority ); + + /// routes a message to its destination node - void route( OverlayMsg* message ); - + void route( OverlayMsg* message, const NodeID& last_hop = NodeID::UNSPECIFIED ); + /// sends a raw message to another node, delivers it to the base overlay class - seqnum_t send( OverlayMsg* message, const NodeID& destination ); + /// may throw "message_not_sent"-exception + seqnum_t send( OverlayMsg* message, + const NodeID& destination, + uint8_t priority, + const NodeID& last_hop = NodeID::UNSPECIFIED ) + throw(message_not_sent); /// send a raw message using a link descriptor, delivers it to the base overlay class - seqnum_t send( OverlayMsg* message, LinkDescriptor* ld, - bool ignore_down = false ); + seqnum_t send( OverlayMsg* message, + LinkDescriptor* ld, + uint8_t priority ) throw(message_not_sent); /// send a message using a node id using overlay routing /// sets necessary fields in the overlay message! - seqnum_t send_node( OverlayMsg* message, const NodeID& remote, - const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID); + /// may throw "message_not_sent"-exception + seqnum_t send_node( OverlayMsg* message, const NodeID& remote, uint8_t priority, + const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID) throw(message_not_sent); /// send a message using a node id using overlay routing using a link /// sets necessary fields in the overlay message! - seqnum_t send_link( OverlayMsg* message, const LinkID& link, - bool ignore_down = false ); - + void send_link( OverlayMsg* message, + const LinkID& link, + uint8_t priority ) throw(message_not_sent); + + + /// sends a notification to a sender from whom we just dropped a message + void report_lost_message( const OverlayMsg* message ); + // misc -------------------------------------------------------------------- Index: source/ariba/overlay/CMakeLists.txt =================================================================== --- source/ariba/overlay/CMakeLists.txt (revision 10700) +++ source/ariba/overlay/CMakeLists.txt (revision 12060) @@ -41,4 +41,5 @@ LinkDescriptor.h OverlayBootstrap.h + SequenceNumber.h ) @@ -47,4 +48,5 @@ LinkDescriptor.cpp OverlayBootstrap.cpp + SequenceNumber.cpp ) Index: source/ariba/overlay/LinkDescriptor.h =================================================================== --- source/ariba/overlay/LinkDescriptor.h (revision 10700) +++ source/ariba/overlay/LinkDescriptor.h (revision 12060) @@ -12,4 +12,11 @@ #include "ariba/communication/EndpointDescriptor.h" #include "ariba/CommunicationListener.h" + +// reboost messages +#include "ariba/utility/transport/messages/message.hpp" +#include + +#include "ariba/overlay/SequenceNumber.h" + namespace ariba { @@ -37,8 +44,18 @@ class LinkDescriptor { public: + struct message_queue_entry + { + reboost::message_t message; + uint8_t priority; + }; + // ctor LinkDescriptor() { + time_t now = time(NULL); + // default values this->up = false; +// this->closing = false; + this->failed = false; this->fromRemote = false; this->remoteNode = NodeID::UNSPECIFIED; @@ -46,8 +63,8 @@ this->communicationUp = false; this->communicationId = LinkID::UNSPECIFIED; - this->keepAliveTime = time(NULL); - this->keepAliveMissed = 0; + this->keepAliveReceived = now; + this->keepAliveSent = now; this->relaying = false; - this->timeRelaying = time(NULL); + this->timeRelaying = now; this->dropAfterRelaying = false; this->service = ServiceID::UNSPECIFIED; @@ -56,6 +73,9 @@ this->remoteLink = LinkID::UNSPECIFIED; this->autolink = false; - this->lastuse = time(NULL); + this->lastuse = now; this->retryCounter = 0; + this->hops = -1; + + this->transmit_seqnums = false; // XXX } @@ -67,12 +87,8 @@ // general information about the link -------------------------------------- bool up; ///< flag whether this link is up and running +// bool closing; ///< flag, whether this link is in the regular process of closing + bool failed; ///< flag, whether communication is (assumed to be) not/no longer possible on this link bool fromRemote; ///< flag, whether this link was requested from remote NodeID remoteNode; ///< remote end-point node - bool isVital() { - return up && keepAliveMissed == 0; - } - bool isDirectVital() { - return isVital() && communicationUp && !relayed; - } @@ -82,4 +98,8 @@ bool communicationUp; ///< flag, whether the communication is up + // sequence numbers -------------------------------------------------------- + SequenceNumber last_sent_seqnum; + bool transmit_seqnums; + // direct link retries ----------------------------------------------------- EndpointDescriptor endpoint; @@ -87,9 +107,9 @@ // link alive information -------------------------------------------------- - time_t keepAliveTime; ///< the last time a keep-alive message was received - int keepAliveMissed; ///< the number of missed keep-alive messages + time_t keepAliveReceived; ///< the last time a keep-alive message was received + time_t keepAliveSent; ///< the number of missed keep-alive messages void setAlive() { - keepAliveMissed = 0; - keepAliveTime = time(NULL); +// keepAliveSent = time(NULL); + keepAliveReceived = time(NULL); } @@ -98,4 +118,5 @@ LinkID remoteLink; ///< the remote link id vector routeRecord; + int hops; // relay state ------------------------------------------------------------- @@ -114,6 +135,6 @@ // 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 + time_t lastuse; ///< time, when the link was last used XXX AUTO_LINK-ONLY + deque messageQueue; ///< waiting messages to be delivered void setAutoUsed() { if (autolink) lastuse = time(NULL); @@ -121,5 +142,5 @@ /// drops waiting auto-link messages void flushQueue() { - BOOST_FOREACH( Message* msg, messageQueue ) delete msg; +// BOOST_FOREACH( Message* msg, messageQueue ) delete msg; // XXX MARIO: shouldn't be necessary anymore, since we're using shared pointers messageQueue.clear(); } @@ -127,13 +148,21 @@ // string representation --------------------------------------------------- std::string to_string() const { + time_t now = time(NULL); + std::ostringstream s; + if ( relayed ) + s << "[RELAYED-"; + else + s << "[DIRECT-"; + s << "LINK] "; + s << "id=" << overlayId.toString().substr(0,4) << " "; + s << "serv=" << service.toString() << " "; s << "up=" << up << " "; s << "init=" << !fromRemote << " "; - s << "id=" << overlayId.toString().substr(0,4) << " "; - s << "serv=" << service.toString() << " "; s << "node=" << remoteNode.toString().substr(0,4) << " "; s << "relaying=" << relaying << " "; - s << "miss=" << keepAliveMissed << " "; + s << "last_received=" << now - keepAliveReceived << "s "; s << "auto=" << autolink << " "; + s << "hops=" << hops << " "; if ( relayed ) { s << "| Relayed: "; @@ -146,5 +175,5 @@ } else { s << "| Direct: "; - s << "using id=" << communicationId.toString().substr(0,4) << " "; + s << "using [COMMUNICATION-LINK] id=" << communicationId.toString().substr(0,4) << " "; s << "(up=" << communicationUp << ") "; } Index: source/ariba/overlay/SequenceNumber.cpp =================================================================== --- source/ariba/overlay/SequenceNumber.cpp (revision 12060) +++ source/ariba/overlay/SequenceNumber.cpp (revision 12060) @@ -0,0 +1,167 @@ + +#include "SequenceNumber.h" +#include +#include + +namespace ariba { +namespace overlay { + +/** static initializers **/ +// DISABLED_SEQNUM const +const SequenceNumber SequenceNumber::DISABLED; + +// seed the RNG +boost::mt19937 SequenceNumber::rng_32bit(std::time(NULL)); + +boost::uniform_int SequenceNumber::seqnum_distribution_32bit( + MIN_SEQ_NUM, std::numeric_limits::max()); + +boost::uniform_int SequenceNumber::distribution_full32bit( + 0, std::numeric_limits::max()); + + + +/** class implementation **/ +SequenceNumber::SequenceNumber(): + seqnum_32(SEQ_NUM_DISABLED), + seqnum_64(SEQ_NUM_DISABLED) +{ +} +SequenceNumber::SequenceNumber(uint32_t seqnum): + seqnum_32(seqnum), + seqnum_64(SEQ_NUM_DISABLED) +{ +} +SequenceNumber::SequenceNumber(uint64_t seqnum): + seqnum_32(SEQ_NUM_DISABLED), + seqnum_64(seqnum) +{ +} + + +SequenceNumber SequenceNumber::createRandomSeqNum_Short() +{ + uint32_t num = seqnum_distribution_32bit(rng_32bit); + return SequenceNumber(num); +} + +SequenceNumber SequenceNumber::createRandomSeqNum_Long() +{ + uint64_t num = distribution_full32bit(rng_32bit); + num << 32; + num += seqnum_distribution_32bit(rng_32bit); + + return SequenceNumber(num); +} + +bool SequenceNumber::isShortSeqNum() const +{ + return seqnum_32 >= MIN_SEQ_NUM && seqnum_64 == SEQ_NUM_DISABLED; +} + +bool SequenceNumber::isLongSeqNum() const +{ + return seqnum_32 == SEQ_NUM_DISABLED && seqnum_64 >= MIN_SEQ_NUM; +} + +bool SequenceNumber::isDisabled() const +{ + return seqnum_32 == SEQ_NUM_DISABLED && seqnum_64 == SEQ_NUM_DISABLED; +} + +bool SequenceNumber::isValid() const +{ + return isShortSeqNum() || isLongSeqNum(); +} + +void SequenceNumber::increment() +{ + // BRANCH: short seqnum + if ( isShortSeqNum() ) + { + seqnum_32++; + + // wrap overflow + if ( seqnum_32 < MIN_SEQ_NUM ) + { + seqnum_32 = MIN_SEQ_NUM; + } + } + + // BRANCH: long seqnum + else if ( isLongSeqNum() ) + { + seqnum_64++; + + // wrap overflow + if ( seqnum_64 < MIN_SEQ_NUM ) + { + seqnum_64 = MIN_SEQ_NUM; + } + } +} + +bool SequenceNumber::operator==(const SequenceNumber& rhs) const +{ + return seqnum_32 + seqnum_64 == rhs.seqnum_32 + rhs.seqnum_64; +} + +bool SequenceNumber::operator<(const SequenceNumber& rhs) const +{ + return seqnum_32 + seqnum_64 < rhs.seqnum_32 + rhs.seqnum_64; +} + +bool SequenceNumber::isSuccessor(const SequenceNumber& rhs) +{ + // TODO implement + + return false; +} + +bool SequenceNumber::isPredecessor(const SequenceNumber& rhs) +{ + // TODO implement + + return false; +} + +uint32_t SequenceNumber::getShortSeqNum() const +{ + return seqnum_32; +} + +uint64_t SequenceNumber::getLongSeqNum() const +{ + return seqnum_64; +} + + +std::ostream& operator<<(std::ostream& stream, const SequenceNumber& rhs) +{ + if ( rhs.isDisabled() ) + { + return stream << "DISABLED"; + } + + else if ( ! rhs.isValid() ) + { + return stream << "INVALID"; + } + + else if ( rhs.isShortSeqNum() ) + { + return stream << rhs.seqnum_32; + } + + else if ( rhs.isLongSeqNum() ) + { + return stream << rhs.seqnum_64; + } + + else + { + return stream << "ERROR"; + } +} + +}} // [namespace ariba::overlay] Index: source/ariba/overlay/SequenceNumber.h =================================================================== --- source/ariba/overlay/SequenceNumber.h (revision 12060) +++ source/ariba/overlay/SequenceNumber.h (revision 12060) @@ -0,0 +1,81 @@ + +#ifndef SEQUENCENUMBER_H +#define SEQUENCENUMBER_H + +#include +#include + +// random +#include +#include + +namespace ariba { +namespace overlay { + +#define MIN_SEQ_NUM 100 +#define SEQ_NUM_DISABLED 0 + +class SequenceNumber +{ +public: + static const SequenceNumber DISABLED; + + /// constructors + SequenceNumber(); + SequenceNumber(uint32_t seqnum); + SequenceNumber(uint64_t seqnum); + + /// factories + static SequenceNumber createRandomSeqNum_Short(); + static SequenceNumber createRandomSeqNum_Long(); + + + /// get type information + bool isShortSeqNum() const; + bool isLongSeqNum() const; + bool isDisabled() const; + bool isValid() const; + + + /// operators + void increment(); + + /// NOTE: this is also »true« if the long seq_num matches the short seq_num + bool operator== (const SequenceNumber& rhs) const; + + /// NOTE: return value is only meaningful if »isValid() == true« for both values + bool operator< (const SequenceNumber& rhs) const; + + bool isSuccessor(const SequenceNumber& rhs); + bool isPredecessor(const SequenceNumber& rhs); + + + /// getter + uint32_t getShortSeqNum() const; + uint64_t getLongSeqNum() const; + + + /// to string + friend std::ostream& operator<< (std::ostream& stream, const SequenceNumber& rhs); + +private: + uint32_t seqnum_32; + uint64_t seqnum_64; + + /// static random_number_generator + static boost::mt19937 rng_32bit; + static boost::uniform_int seqnum_distribution_32bit; + + // XXX NOTE: on later boost versions these are: + // boost::random::mt19937 + // boost::random::uniform_int_distribution<> + + // NOTE: since the mt19937_64 is not available in this boost version, we + // need an addition distribution + static boost::uniform_int distribution_full32bit; +}; + + +}} // [namespace ariba::overlay] + +#endif // SEQUENCENUMBER_H Index: source/ariba/overlay/messages/JoinReply.cpp =================================================================== --- source/ariba/overlay/messages/JoinReply.cpp (revision 10700) +++ source/ariba/overlay/messages/JoinReply.cpp (revision 12060) @@ -44,6 +44,7 @@ vsznDefault(JoinReply); -JoinReply::JoinReply(const SpoVNetID _spovnetid, const OverlayParameterSet _param, bool _joinAllowed, const EndpointDescriptor _bootstrapEp) - : spovnetid( _spovnetid ), param( _param ), joinAllowed( _joinAllowed ), bootstrapEp( _bootstrapEp ){ +JoinReply::JoinReply(const SpoVNetID _spovnetid, const OverlayParameterSet _param, bool _joinAllowed) + : spovnetid( _spovnetid ), param( _param ), joinAllowed( _joinAllowed ) +{ } @@ -64,7 +65,7 @@ } -const EndpointDescriptor& JoinReply::getBootstrapEndpoint(){ - return bootstrapEp; -} +//const EndpointDescriptor& JoinReply::getBootstrapEndpoint(){ +// return bootstrapEp; +//} }} // ariba::overlay Index: source/ariba/overlay/messages/JoinReply.h =================================================================== --- source/ariba/overlay/messages/JoinReply.h (revision 10700) +++ source/ariba/overlay/messages/JoinReply.h (revision 12060) @@ -40,10 +40,11 @@ #define JOIN_REPLY_H__ -#include "ariba/utility/messages.h" +//#include "ariba/utility/messages.h" +#include "ariba/utility/messages/Message.h" #include "ariba/utility/serialization.h" #include "ariba/utility/types/SpoVNetID.h" #include "ariba/utility/types/NodeID.h" #include "ariba/utility/types/OverlayParameterSet.h" -#include "ariba/communication/EndpointDescriptor.h" +//#include "ariba/communication/EndpointDescriptor.h" using ariba::utility::OverlayParameterSet; @@ -51,5 +52,5 @@ using ariba::utility::SpoVNetID; using ariba::utility::NodeID; -using ariba::communication::EndpointDescriptor; +//using ariba::communication::EndpointDescriptor; namespace ariba { @@ -64,5 +65,5 @@ OverlayParameterSet param; //< overlay parameters bool joinAllowed; //< join successfull or access denied - EndpointDescriptor bootstrapEp; //< the endpoint for bootstrapping the overlay interface +// EndpointDescriptor bootstrapEp; //< the endpoint for bootstrapping the overlay interface public: @@ -70,6 +71,6 @@ const SpoVNetID _spovnetid = SpoVNetID::UNSPECIFIED, const OverlayParameterSet _param = OverlayParameterSet::DEFAULT, - bool _joinAllowed = false, - const EndpointDescriptor _bootstrapEp = EndpointDescriptor::UNSPECIFIED() + bool _joinAllowed = false /*, + const EndpointDescriptor _bootstrapEp = EndpointDescriptor::UNSPECIFIED()*/ ); @@ -79,5 +80,5 @@ const OverlayParameterSet& getParam(); bool getJoinAllowed(); - const EndpointDescriptor& getBootstrapEndpoint(); +// const EndpointDescriptor& getBootstrapEndpoint(); }; @@ -86,5 +87,7 @@ sznBeginDefault( ariba::overlay::JoinReply, X ) { uint8_t ja = joinAllowed; - X && &spovnetid && param && bootstrapEp && ja; + X && &spovnetid && param; +// X && bootstrapEp; + X && ja; if (X.isDeserializer()) joinAllowed = ja; } sznEnd(); Index: source/ariba/overlay/messages/OverlayMsg.h =================================================================== --- source/ariba/overlay/messages/OverlayMsg.h (revision 10700) +++ source/ariba/overlay/messages/OverlayMsg.h (revision 12060) @@ -47,6 +47,6 @@ #include "ariba/utility/types/NodeID.h" #include "ariba/utility/types/LinkID.h" -#include "ariba/communication/EndpointDescriptor.h" - +// #include +#include "ariba/overlay/SequenceNumber.h" namespace ariba { @@ -57,5 +57,5 @@ using ariba::utility::ServiceID; using ariba::utility::Message; -using ariba::communication::EndpointDescriptor; +//using ariba::communication::EndpointDescriptor; using_serialization; @@ -64,5 +64,5 @@ * between nodes. * - * @author Sebastian Mies + * @author Sebastian Mies , Mario Hock */ class OverlayMsg: public Message { VSERIALIZEABLE; @@ -75,4 +75,5 @@ maskTransfer = 0x10, ///< bit mask for transfer messages typeData = 0x11, ///< message contains data for higher layers + typeMessageLost = 0x12, ///< message contains info about a dropped message // join signaling @@ -87,5 +88,7 @@ typeLinkUpdate = 0x33, ///< update message for link association typeLinkDirect = 0x34, ///< direct connection has been established - typeLinkAlive = 0x35, ///< keep-alive message + typeKeepAlive = 0x35, ///< keep-alive message + typeKeepAliveReply = 0x36, ///< keep-alive message (replay) + typeLinkClose = 0x37, /// DHT routed messages @@ -100,4 +103,8 @@ maskDHTResponse = 0x50, ///< bit mask for dht responses typeDHTData = 0x51, ///< DHT get data + + /// misc message types + typePing = 0x44, + typePong = 0x45, // topology signaling @@ -105,4 +112,17 @@ typeSignalingEnd = 0xFF ///< end of the signaling types }; + + /// message flags (uint8_t) + enum flags_ + { + flagRelayed = 1 << 0, + flagRegisterRelay = 1 << 1, + flagRouteRecord = 1 << 2, + flagSeqNum1 = 1 << 3, + flagSeqNum2 = 1 << 4, + flagAutoLink = 1 << 5, + flagLinkMessage = 1 << 6, + flagHasMoreFlags = 1 << 7 + }; /// default constructor @@ -114,5 +134,5 @@ const LinkID& _sourceLink = LinkID::UNSPECIFIED, const LinkID& _destinationLink = LinkID::UNSPECIFIED ) - : type(type), flags(0), hops(0), ttl(10), + : type(type), flags(0), extended_flags(0), hops(0), ttl(10), priority(0), service(_service), sourceNode(_sourceNode), destinationNode(_destinationNode), @@ -125,6 +145,7 @@ // copy constructor OverlayMsg(const OverlayMsg& rhs) - : type(rhs.type), flags(rhs.flags), hops(rhs.hops), ttl(rhs.ttl), - service(rhs.service), + : type(rhs.type), flags(rhs.flags), extended_flags(rhs.extended_flags), + hops(rhs.hops), ttl(rhs.ttl), + priority(rhs.priority), service(rhs.service), sourceNode(rhs.sourceNode), destinationNode(rhs.destinationNode), sourceLink(rhs.sourceLink), destinationLink(rhs.destinationLink), @@ -149,53 +170,59 @@ } + /// priority ------------------------------------------------------------------ + + uint8_t getPriority() const { + return priority; + } + + void setPriority(uint8_t priority) { + this->priority = priority; + } + /// flags ------------------------------------------------------------------ bool isRelayed() const { - return (flags & 0x01)!=0; + return (flags & flagRelayed)!=0; } void setRelayed( bool relayed = true ) { - if (relayed) flags |= 1; else flags &= ~1; + if (relayed) flags |= flagRelayed; else flags &= ~flagRelayed; } bool isRegisterRelay() const { - return (flags & 0x02)!=0; + return (flags & flagRegisterRelay)!=0; } void setRegisterRelay( bool relayed = true ) { - if (relayed) flags |= 0x02; else flags &= ~0x02; + if (relayed) flags |= flagRegisterRelay; else flags &= ~flagRegisterRelay; } bool isRouteRecord() const { - return (flags & 0x04)!=0; + return (flags & flagRouteRecord)!=0; } void setRouteRecord( bool route_record = true ) { - if (route_record) flags |= 0x04; else flags &= ~0x04; + if (route_record) flags |= flagRouteRecord; else flags &= ~flagRouteRecord; } bool isAutoLink() const { - return (flags & 0x80) == 0x80; + return (flags & flagAutoLink) == flagAutoLink; } void setAutoLink(bool auto_link = true ) { - if (auto_link) flags |= 0x80; else flags &= ~0x80; + if (auto_link) flags |= flagAutoLink; else flags &= ~flagAutoLink; } bool isLinkMessage() const { - return (flags & 0x40)!=0; + return (flags & flagLinkMessage)!=0; } void setLinkMessage(bool link_info = true ) { - if (link_info) flags |= 0x40; else flags &= ~0x40; - } - - bool containsSourceEndpoint() const { - return (flags & 0x20)!=0; - } - - void setContainsSourceEndpoint(bool contains_endpoint) { - if (contains_endpoint) flags |= 0x20; else flags &= ~0x20; - } + if (link_info) flags |= flagLinkMessage; else flags &= ~flagLinkMessage; + } + + bool hasExtendedFlags() const { + return (flags & flagHasMoreFlags) == flagHasMoreFlags; + } /// number of hops and time to live ---------------------------------------- @@ -264,13 +291,4 @@ this->destinationLink = link; setLinkMessage(); - } - - void setSourceEndpoint( const EndpointDescriptor& endpoint ) { - sourceEndpoint = endpoint; - setContainsSourceEndpoint(true); - } - - const EndpointDescriptor& getSourceEndpoint() const { - return sourceEndpoint; } @@ -284,4 +302,5 @@ destinationLink = dummyLink; hops = 0; + routeRecord.clear(); } @@ -294,7 +313,48 @@ routeRecord.push_back(node); } + + /// sequence numbers + bool hasShortSeqNum() const + { + return (flags & (flagSeqNum1 | flagSeqNum2)) == flagSeqNum1; + } + + bool hasLongSeqNum() const + { + return (flags & (flagSeqNum1 | flagSeqNum2)) == flagSeqNum2; + } + + void setSeqNum(const SequenceNumber& sequence_number) + { + this->seqnum = sequence_number; + + // short seqnum + if ( sequence_number.isShortSeqNum() ) + { + flags |= flagSeqNum1; + flags &= ~flagSeqNum2; + } + // longseqnum + else if ( sequence_number.isShortSeqNum() ) + { + flags &= ~flagSeqNum1; + flags |= flagSeqNum2; + } + // no seqnum + else + { + flags &= ~flagSeqNum1; + flags &= ~flagSeqNum2; + } + } + + const SequenceNumber& getSeqNum() const + { + return seqnum; + } + private: - uint8_t type, flags, hops, ttl; + uint8_t type, flags, extended_flags, hops, ttl, priority; ServiceID service; NodeID sourceNode; @@ -302,6 +362,7 @@ LinkID sourceLink; LinkID destinationLink; - EndpointDescriptor sourceEndpoint; +// EndpointDescriptor sourceEndpoint; vector routeRecord; + SequenceNumber seqnum; }; @@ -311,8 +372,16 @@ sznBeginDefault( ariba::overlay::OverlayMsg, X ){ // header - X && type && flags && hops && ttl; + X && type && flags; + + if ( hasExtendedFlags() ) + X && extended_flags; + + X && hops && ttl; // addresses X && &service && &sourceNode && &destinationNode; + + // priority + X && priority; // message is associated with a end-to-end link @@ -320,8 +389,40 @@ X && &sourceLink && &destinationLink; - // message is associated with a source end-point - if (containsSourceEndpoint()) - X && sourceEndpoint; - + + /* seqnum */ + // serialize + if ( X.isSerializer() ) + { + if ( hasShortSeqNum() ) + { + uint32_t short_seqnum; + short_seqnum = seqnum.getShortSeqNum(); + X && short_seqnum; + } + if ( hasLongSeqNum() ) + { + uint64_t long_seqnum; + long_seqnum = seqnum.getLongSeqNum(); + X && long_seqnum; + } + } + // deserialize + else + { + if ( hasShortSeqNum() ) + { + uint32_t short_seqnum; + X && short_seqnum; + seqnum = ariba::overlay::SequenceNumber(short_seqnum); + } + if ( hasLongSeqNum() ) + { + uint64_t long_seqnum; + X && long_seqnum; + seqnum = ariba::overlay::SequenceNumber(long_seqnum); + } + } + + // message should record its route if (isRouteRecord()) { @@ -333,5 +434,5 @@ // payload - X && Payload(); +// X && Payload(); } sznEnd(); Index: source/ariba/overlay/modules/OverlayFactory.cpp =================================================================== --- source/ariba/overlay/modules/OverlayFactory.cpp (revision 10700) +++ source/ariba/overlay/modules/OverlayFactory.cpp (revision 12060) @@ -41,5 +41,5 @@ // structured overlays #include "chord/Chord.h" -#include "onehop/OneHop.h" +//#include "onehop/OneHop.h" //DEPRECATED namespace ariba { @@ -57,11 +57,17 @@ return new Chord( baseoverlay, nodeid, routeReceiver, param ); - case OverlayParameterSet::OverlayStructureOneHop: - return new OneHop( baseoverlay, nodeid, routeReceiver, param ); +// case OverlayParameterSet::OverlayStructureOneHop: +// return new OneHop( baseoverlay, nodeid, routeReceiver, param ); default: + // NEVER return "NULL" + assert(false); + throw 42; return NULL; } + // NEVER return "NULL" + assert(false); + throw 42; return NULL; } Index: source/ariba/overlay/modules/OverlayInterface.cpp =================================================================== --- source/ariba/overlay/modules/OverlayInterface.cpp (revision 10700) +++ source/ariba/overlay/modules/OverlayInterface.cpp (revision 12060) @@ -68,4 +68,5 @@ void OverlayInterface::onLinkFail(const LinkID& lnk, const NodeID& remote) { + onLinkDown(lnk, remote); } @@ -79,5 +80,7 @@ } -void OverlayInterface::onMessage(const DataMessage& msg, const NodeID& remote, +void OverlayInterface::onMessage(OverlayMsg* msg, + reboost::shared_buffer_t sub_msg, + const NodeID& remote, const LinkID& lnk) { } Index: source/ariba/overlay/modules/OverlayInterface.h =================================================================== --- source/ariba/overlay/modules/OverlayInterface.h (revision 10700) +++ source/ariba/overlay/modules/OverlayInterface.h (revision 12060) @@ -145,5 +145,26 @@ */ virtual const LinkID& getNextLinkId( const NodeID& id ) const = 0; - + + /** + * Returns link ids of possible next hops a route message could take, + * sorted by "quality" (e.g. overlay-distance). + * + * The »num« parameter can be used to specify the desired number of elements + * in the returned vector. This is intendet for optimizations. The + * implementation may choose to return a different number of elements than + * requested. + * + * NOTE: The returned vector may contain »unspecified« links. These refer to + * to the own node. (e.g. If there's no closer node, the top element in the + * returned vector is unsoecified.) + * + * @param id The destination node id + * @param num The desired number of elements in the returned vector. + * (0 means »not specified/max)« + * @return A sorted vector of link ids to possible next hops. + */ + virtual std::vector getSortedLinkIdsTowardsNode( + const NodeID& id, int num = 0 ) const = 0; + /** * Returns the NodeID of the next hop a route message would take. @@ -176,5 +197,7 @@ /// @see CommunicationListener - virtual void onMessage(const DataMessage& msg, const NodeID& remote, + virtual void onMessage(OverlayMsg* msg, + reboost::shared_buffer_t sub_msg, + const NodeID& remote, const LinkID& lnk = LinkID::UNSPECIFIED); Index: source/ariba/overlay/modules/OverlayStructureEvents.h =================================================================== --- source/ariba/overlay/modules/OverlayStructureEvents.h (revision 10700) +++ source/ariba/overlay/modules/OverlayStructureEvents.h (revision 12060) @@ -41,7 +41,9 @@ #include "ariba/utility/types/NodeID.h" -#include "ariba/utility/messages.h" +#include "ariba/utility/types/LinkID.h" +#include "ariba/utility/messages/Message.h" using ariba::utility::NodeID; +using ariba::utility::LinkID; using ariba::utility::Message; @@ -50,9 +52,9 @@ class OverlayInterface; -class OneHop; +//class OneHop; class OverlayStructureEvents { friend class ariba::overlay::OverlayInterface; - friend class ariba::overlay::OneHop; +// friend class ariba::overlay::OneHop; public: Index: source/ariba/overlay/modules/chord/Chord.cpp =================================================================== --- source/ariba/overlay/modules/chord/Chord.cpp (revision 10700) +++ source/ariba/overlay/modules/chord/Chord.cpp (revision 12060) @@ -43,5 +43,5 @@ #include "detail/chord_routing_table.hpp" -#include "messages/Discovery.h" +//#include "messages/Discovery.h" // XXX DEPRECATED namespace ariba { @@ -55,5 +55,60 @@ typedef chord_routing_table::item route_item; +using ariba::transport::system_priority; + use_logging_cpp( Chord ); + + +////// Messages +struct DiscoveryMessage +{ + /** + * DiscoveryMessage + * - type + * - data + * - Endpoint + */ + + // type enum + enum type_ { + invalid = 0, + normal = 1, + successor = 2, + predecessor = 3 + }; + + + // data + uint8_t type; + uint8_t ttl; + EndpointDescriptor endpoint; + + // serialize + reboost::message_t serialize() + { + // serialize endpoint + reboost::message_t msg = endpoint.serialize(); + + // serialize type and ttl + uint8_t* buff1 = msg.push_front(2*sizeof(uint8_t)).mutable_data(); + buff1[0] = type; + buff1[1] = ttl; + + return msg; + } + + //deserialize + reboost::shared_buffer_t deserialize(reboost::shared_buffer_t buff) + { + // deserialize type and ttl + const uint8_t* bytes = buff.data(); + type = bytes[0]; + ttl = bytes[1]; + + // deserialize endpoint + return endpoint.deserialize(buff(2*sizeof(uint8_t))); + } +}; + Chord::Chord(BaseOverlay& _baseoverlay, const NodeID& _nodeid, @@ -102,7 +157,22 @@ /// helper: sends a message using the "base overlay" -seqnum_t Chord::send( OverlayMsg* msg, const LinkID& link ) { - if (link.isUnspecified()) return 0; - return baseoverlay.send_link( msg, link ); +void Chord::send( OverlayMsg* msg, const LinkID& link ) { + if (link.isUnspecified()) + return; + + baseoverlay.send_link( msg, link, system_priority::OVERLAY ); +} + +void Chord::send_node( OverlayMsg* message, const NodeID& remote ) +{ + try + { + baseoverlay.send( message, remote, system_priority::OVERLAY ); + } + catch ( message_not_sent& e ) + { + logging_warn("Chord: Could not send message to " << remote + << ": " << e.what()); + } } @@ -116,20 +186,44 @@ OverlayMsg msg( typeDiscovery ); msg.setRegisterRelay(true); - Discovery dmsg( Discovery::normal, (uint8_t)ttl, baseoverlay.getEndpointDescriptor() ); - msg.encapsulate(&dmsg); + + // create DiscoveryMessage + DiscoveryMessage dmsg; + dmsg.type = DiscoveryMessage::normal; + dmsg.ttl = ttl; + dmsg.endpoint = baseoverlay.getEndpointDescriptor(); + + msg.set_payload_message(dmsg.serialize()); // send to node - baseoverlay.send_node( &msg, remote ); + try + { + baseoverlay.send_node( &msg, remote, system_priority::OVERLAY ); + } + catch ( message_not_sent& e ) + { + logging_warn("Chord: Could not send message to " << remote + << ": " << e.what()); + } } void Chord::discover_neighbors( const LinkID& link ) { uint8_t ttl = 1; + + // FIXME try-catch for the send operations + + // create DiscoveryMessage + DiscoveryMessage dmsg; + dmsg.ttl = ttl; + dmsg.endpoint = baseoverlay.getEndpointDescriptor(); { // send predecessor discovery OverlayMsg msg( typeDiscovery ); msg.setRegisterRelay(true); - Discovery dmsg( Discovery::predecessor, ttl, - baseoverlay.getEndpointDescriptor() ); - msg.encapsulate(&dmsg); + + // set type + dmsg.type = DiscoveryMessage::predecessor; + + // send + msg.set_payload_message(dmsg.serialize()); send(&msg, link); } @@ -137,9 +231,12 @@ // send successor discovery OverlayMsg msg( typeDiscovery ); - msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() ); +// msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() ); // XXX this was redundand, wasn't it? msg.setRegisterRelay(true); - Discovery dmsg( Discovery::successor, ttl, - baseoverlay.getEndpointDescriptor() ); - msg.encapsulate(&dmsg); + + // set type + dmsg.type = DiscoveryMessage::successor; + + // send + msg.set_payload_message(dmsg.serialize()); send(&msg, link); } @@ -163,5 +260,6 @@ // timer for stabilization management - Timer::setInterval(1000); +// Timer::setInterval(1000); // TODO find an appropriate interval! + Timer::setInterval(10000); // XXX testing... Timer::start(); } @@ -200,4 +298,48 @@ return item->info; } + +std::vector Chord::getSortedLinkIdsTowardsNode( + const NodeID& id, int num ) const +{ + std::vector ret; + + switch ( num ) + { + // special case: just call »getNextLinkId« + case 1: + { + ret.push_back(&getNextLinkId(id)); + + break; + } + + // * calculate top 2 * + case 0: + case 2: + { + std::vector items = table->get_next_2_hops(id); + + ret.reserve(items.size()); + + BOOST_FOREACH( const route_item* item, items ) + { + ret.push_back(&item->info); + } + + break; + } + + // NOTE: implement real sorting, if needed (and handle "case 0" properly, then) + default: + { + throw std::runtime_error("Not implemented. (Chord::getSortedLinkIdsTowardsNode with num != 2)"); + + break; + } + } + + return ret; +} + /// @see OverlayInterface.h @@ -253,4 +395,5 @@ if (remote==nodeid) { logging_warn("dropping link that has been established to myself (nodes have same nodeid?)"); + logging_warn("NodeID: " << remote); baseoverlay.dropLink(lnk); return; @@ -290,5 +433,6 @@ /// @see CommunicationListener.h or @see OverlayInterface.h void Chord::onLinkDown(const LinkID& lnk, const NodeID& remote) { - logging_debug("link_down: link=" << lnk.toString() << " remote=" << + // XXX logging_debug + logging_info("link_down (Chord): link=" << lnk.toString() << " remote=" << remote.toString() ); @@ -303,23 +447,24 @@ /// @see CommunicationListener.h /// @see OverlayInterface.h -void Chord::onMessage(const DataMessage& msg, const NodeID& remote, +void Chord::onMessage(OverlayMsg* msg, + reboost::shared_buffer_t sub_msg, + const NodeID& remote, const LinkID& link) { - // decode message - OverlayMsg* m = dynamic_cast(msg.getMessage()); - if (m == NULL) return; - // handle messages - switch ((signalMessageTypes)m->getType()) { + switch ((signalMessageTypes) msg->getType()) { // discovery request - case typeDiscovery: { - // decapsulate message - Discovery* dmsg = m->decapsulate (); + case typeDiscovery: + { + // deserialize discovery message + DiscoveryMessage dmsg; + dmsg.deserialize(sub_msg); + logging_debug("Received discovery message with" - << " src=" << m->getSourceNode().toString() - << " dst=" << m->getDestinationNode().toString() - << " ttl=" << (int)dmsg->getTTL() - << " type=" << (int)dmsg->getType() + << " src=" << msg->getSourceNode().toString() + << " dst=" << msg->getDestinationNode().toString() + << " ttl=" << (int)dmsg.ttl + << " type=" << (int)dmsg.type ); @@ -327,108 +472,125 @@ bool found = false; BOOST_FOREACH( NodeID& value, discovery ) - if (value == m->getSourceNode()) { + if (value == msg->getSourceNode()) { found = true; break; } - if (!found) discovery.push_back(m->getSourceNode()); + if (!found) discovery.push_back(msg->getSourceNode()); // check if source node can be added to routing table and setup link - if (m->getSourceNode() != nodeid) - setup( dmsg->getEndpoint(), m->getSourceNode() ); + if (msg->getSourceNode() != nodeid) + setup( dmsg.endpoint, msg->getSourceNode() ); // process discovery message -------------------------- switch start -- - switch (dmsg->getType()) { - - // normal: route discovery message like every other message - case Discovery::normal: { - // closest node? yes-> split to follow successor and predecessor - if ( table->is_closest_to(m->getDestinationNode()) ) { - logging_debug("Discovery split:"); - if (!table->get_successor()->isUnspecified()) { - OverlayMsg omsg(*m); - dmsg->setType(Discovery::successor); - omsg.encapsulate(dmsg); - logging_debug("* Routing to successor " - << table->get_successor()->toString() ); - baseoverlay.send( &omsg, *table->get_successor() ); - } - - // send predecessor message - if (!table->get_predesessor()->isUnspecified()) { - OverlayMsg omsg(*m); - dmsg->setType(Discovery::predecessor); - omsg.encapsulate(dmsg); - logging_debug("* Routing to predecessor " - << table->get_predesessor()->toString() ); - baseoverlay.send( &omsg, *table->get_predesessor() ); - } - } - // no-> route message - else { - baseoverlay.route( m ); - } - break; - } - - // successor mode: follow the successor until TTL is zero - case Discovery::successor: - case Discovery::predecessor: { - // reached destination? no->forward! - if (m->getDestinationNode() != nodeid) { - OverlayMsg omsg(*m); - omsg.encapsulate(dmsg); - omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID); - baseoverlay.route( &omsg ); - break; - } - - // time to live ended? yes-> stop routing - if (dmsg->getTTL() == 0 || dmsg->getTTL() > 10) break; - - // decrease time-to-live - dmsg->setTTL(dmsg->getTTL() - 1); - - const route_item* item = NULL; - if (dmsg->getType() == Discovery::successor && - table->get_successor() != NULL) { - item = table->get(*table->get_successor()); - } else { - if (table->get_predesessor()!=NULL) - item = table->get(*table->get_predesessor()); - } - if (item == NULL) - break; - - logging_debug("Routing discovery message to succ/pred " - << item->id.toString() ); - OverlayMsg omsg(*m); - omsg.encapsulate(dmsg); - omsg.setDestinationNode(item->id); - omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID); - baseoverlay.send(&omsg, omsg.getDestinationNode()); - break; - } - case Discovery::invalid: - break; - - default: - break; - } - // process discovery message ---------------------------- switch end -- - - delete dmsg; - break; - } - - // leave - case typeLeave: { - if (link!=LinkID::UNSPECIFIED) { - route_item* item = table->get(remote); - if (item!=NULL) item->info = LinkID::UNSPECIFIED; - table->remove(remote); - baseoverlay.dropLink(link); - } - break; - }} + switch ( dmsg.type ) + { + // normal: route discovery message like every other message + case DiscoveryMessage::normal: + { + // closest node? yes-> split to follow successor and predecessor + if ( table->is_closest_to(msg->getDestinationNode()) ) + { + logging_debug("Discovery split:"); + if (!table->get_successor()->isUnspecified()) + { + OverlayMsg omsg(*msg); + + dmsg.type = DiscoveryMessage::successor; + omsg.set_payload_message(dmsg.serialize()); + + logging_debug("* Routing to successor " + << table->get_successor()->toString() ); + send_node( &omsg, *table->get_successor() ); + } + + // send predecessor message + if (!table->get_predesessor()->isUnspecified()) + { + OverlayMsg omsg(*msg); + + dmsg.type = DiscoveryMessage::predecessor; + omsg.set_payload_message(dmsg.serialize()); + + logging_debug("* Routing to predecessor " + << table->get_predesessor()->toString() ); + send_node( &omsg, *table->get_predesessor() ); + } + } + // no-> route message + else + { + baseoverlay.route( msg ); + } + break; + } + + // successor mode: follow the successor until TTL is zero + case DiscoveryMessage::successor: + case DiscoveryMessage::predecessor: + { + // reached destination? no->forward! + if (msg->getDestinationNode() != nodeid) + { + OverlayMsg omsg(*msg); + omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID); + + omsg.set_payload_message(dmsg.serialize()); + + baseoverlay.route( &omsg ); + break; + } + + // time to live ended? yes-> stop routing + if (dmsg.ttl == 0 || dmsg.ttl > 10) break; + + // decrease time-to-live + dmsg.ttl--; + + const route_item* item = NULL; + if (dmsg.type == DiscoveryMessage::successor && + table->get_successor() != NULL) + { + item = table->get(*table->get_successor()); + } + else if (table->get_predesessor() != NULL) + { + item = table->get(*table->get_predesessor()); + } + if (item == NULL) + break; + + logging_debug("Routing discovery message to succ/pred " + << item->id.toString() ); + OverlayMsg omsg(*msg); + omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID); + omsg.setDestinationNode(item->id); + + omsg.set_payload_message(dmsg.serialize()); + + send_node( &omsg, omsg.getDestinationNode() ); + break; + } + case DiscoveryMessage::invalid: + break; + + default: + break; + } + // process discovery message ---------------------------- switch end -- + + break; + } + + // leave + case typeLeave: { + if (link!=LinkID::UNSPECIFIED) { + route_item* item = table->get(remote); + if (item!=NULL) item->info = LinkID::UNSPECIFIED; + table->remove(remote); + baseoverlay.dropLink(link); + } + break; + } + } } Index: source/ariba/overlay/modules/chord/Chord.h =================================================================== --- source/ariba/overlay/modules/chord/Chord.h (revision 10700) +++ source/ariba/overlay/modules/chord/Chord.h (revision 12060) @@ -45,4 +45,5 @@ #include "../OverlayInterface.h" #include +#include class chord_routing_table; @@ -87,7 +88,10 @@ const NodeID& node = NodeID::UNSPECIFIED ); - // helper: sends a message using the "base overlay" - seqnum_t send( OverlayMsg* msg, const LinkID& link ); + // helper: sends a message over a link using the "base overlay" + void send( OverlayMsg* msg, const LinkID& link ); + // helper: sends a message to a node using the "base overlay" + void send_node( OverlayMsg* message, const NodeID& remote ); + // stabilization: sends a discovery message to the specified neighborhood void send_discovery_to( const NodeID& destination, int ttl = 3 ); @@ -105,4 +109,9 @@ virtual const LinkID& getNextLinkId( const NodeID& id ) const; + /// @see OverlayInterface.h + /// NOTE: This implementation excepts num == 2 + virtual std::vector getSortedLinkIdsTowardsNode( + const NodeID& id, int num = 0 ) const; + /// @see OverlayInterface.h virtual const NodeID& getNextNodeId( const NodeID& id ) const; @@ -138,5 +147,7 @@ /// @see CommunicationListener.h or @see OverlayInterface.h - virtual void onMessage(const DataMessage& msg, const NodeID& remote, + virtual void onMessage(OverlayMsg* msg, + reboost::shared_buffer_t sub_msg, + const NodeID& remote, const LinkID& lnk = LinkID::UNSPECIFIED); Index: source/ariba/overlay/modules/chord/detail/chord_routing_table.hpp =================================================================== --- source/ariba/overlay/modules/chord/detail/chord_routing_table.hpp (revision 10700) +++ source/ariba/overlay/modules/chord/detail/chord_routing_table.hpp (revision 12060) @@ -96,4 +96,7 @@ // the own node id nodeid_t id; + + // the own node id as routing item + item own_id_item; // successor and predecessor tables @@ -168,6 +171,11 @@ /// constructs the reactive chord routing table explicit chord_routing_table( const nodeid_t& id, int redundancy = 4 ) : - id(id), succ( redundancy, succ_compare_type(this->id), *this ), - pred( redundancy, pred_compare_type(this->id), *this ) { + id(id), + succ( redundancy, succ_compare_type(this->id), *this ), + pred( redundancy, pred_compare_type(this->id), *this ) + { + // init reflexive item + own_id_item.id = id; + own_id_item.ref_count = 1; // create finger tables @@ -273,16 +281,68 @@ } } + if (best_item != NULL && distance(value, id)id)) return NULL; return best_item; } + + std::vector get_next_2_hops( const nodeid_t& value) + { + ring_distance distance; + item* best_item = &own_id_item; + item* second_best_item = NULL; + + // find best and second best item + for (size_t i=0; iref_count==0) continue; + + // check if we found a better item + // is best item + if ( distance(value, curr->id) < distance(value, best_item->id) ) + { + second_best_item = best_item; + best_item = curr; + } + // is second best item + else + { + if ( second_best_item == NULL ) + { + second_best_item = curr; + continue; + } + + if ( distance(value, curr->id) < distance(value, second_best_item->id) ) + { + second_best_item = curr; + } + } + } + + // prepare return vector + std::vector ret; + if ( best_item != NULL ) + { + ret.push_back(best_item); + } + if ( second_best_item != NULL ) + { + ret.push_back(second_best_item); + } + + return ret; + } const nodeid_t* get_successor() { - if (succ.size()==NULL) return NULL; + if (succ.size()==0) return NULL; return &succ.front(); } const nodeid_t* get_predesessor() { - if (pred.size()==NULL) return NULL; + if (pred.size()==0) return NULL; return &pred.front(); } Index: source/ariba/overlay/modules/chord/messages/CMakeLists.txt =================================================================== --- source/ariba/overlay/modules/chord/messages/CMakeLists.txt (revision 10700) +++ source/ariba/overlay/modules/chord/messages/CMakeLists.txt (revision 12060) @@ -37,5 +37,5 @@ # [License] -add_headers(Discovery.h) +#add_headers(Discovery.h) -add_sources(Discovery.cpp) +#add_sources(Discovery.cpp) Index: source/ariba/overlay/modules/chord/messages/Discovery.h =================================================================== --- source/ariba/overlay/modules/chord/messages/Discovery.h (revision 10700) +++ source/ariba/overlay/modules/chord/messages/Discovery.h (revision 12060) @@ -59,4 +59,5 @@ using_serialization; +// XXX This whole message is DEPRECATED class Discovery : public Message { VSERIALIZEABLE; Index: source/ariba/overlay/modules/onehop/CMakeLists.txt =================================================================== --- source/ariba/overlay/modules/onehop/CMakeLists.txt (revision 10700) +++ source/ariba/overlay/modules/onehop/CMakeLists.txt (revision 12060) @@ -37,7 +37,9 @@ # [License] -add_headers(OneHop.h) -add_sources(OneHop.cpp) +## DEPRECATED +#add_headers(OneHop.h) -add_subdir_sources(messages) +#add_sources(OneHop.cpp) + +#add_subdir_sources(messages)