Index: source/ariba/Makefile.am =================================================================== --- source/ariba/Makefile.am (revision 5002) +++ source/ariba/Makefile.am (revision 5151) @@ -259,9 +259,11 @@ libariba_la_SOURCES += \ overlay/BaseOverlay.cpp \ - overlay/OverlayBootstrap.cpp + overlay/OverlayBootstrap.cpp \ + overlay/LinkDescriptor.cpp nobase_libariba_la_HEADERS += \ overlay/BaseOverlay.h \ - overlay/OverlayBootstrap.h + overlay/OverlayBootstrap.h \ + overlay/LinkDescriptor.h #------------> overlay :: messages @@ -271,4 +273,5 @@ overlay/messages/JoinRequest.cpp \ overlay/messages/LinkRequest.cpp \ + overlay/messages/RelayMessage.cpp \ overlay/messages/OverlayMsg.cpp @@ -277,4 +280,5 @@ overlay/messages/JoinRequest.h \ overlay/messages/LinkRequest.h \ + overlay/messages/RelayMessage.h \ overlay/messages/OverlayMsg.h Index: source/ariba/SideportListener.cpp =================================================================== --- source/ariba/SideportListener.cpp (revision 5002) +++ source/ariba/SideportListener.cpp (revision 5151) @@ -40,4 +40,7 @@ #include "ariba/overlay/BaseOverlay.h" +#include "ariba/overlay/LinkDescriptor.h" + +using ariba::overlay::LinkDescriptor; namespace ariba { @@ -71,59 +74,33 @@ } -//****************************************************** -// -// JUST EXPERIMENTAL WITH RANDOM RESULTS -// +bool SideportListener::isRelayedNode(const NodeID& node){ -// 0 = normal node --> directly accessible -// 1 = relaying node --> node that is relaying for us (must also be 0) -// 2 = relayed node --> node that we cannot access directly -static map relayingnodes; -static map protocolnodes; + BOOST_FOREACH( LinkDescriptor* link, overlay->links ){ + + if( (!link->localRelay.isUnspecified()) && link->remoteRelay == node && link->up) { + return true; + } + + } + + return false; +} bool SideportListener::isRelayingNode(const NodeID& node){ - map::iterator i = relayingnodes.find(node); - if(i != relayingnodes.end()) return (i->second == 1); + BOOST_FOREACH( LinkDescriptor* link, overlay->links ){ - relayingnodes.insert( std::make_pair( node, rand()%3 ) ); -} + if( (!link->localRelay.isUnspecified()) && link->localRelay == node && link->up) { + return true; + } -bool SideportListener::isRelayedNode(const NodeID& node){ - - map::iterator i = relayingnodes.find(node); - if(i != relayingnodes.end()) return (i->second == 2); - - relayingnodes.insert( std::make_pair( node, rand()%3 ) ); -} - -SideportListener::Protocol generateRandomProtocol(){ - - if( (rand() % 3) == 0 ){ - return SideportListener::rfcomm; } - int ret = SideportListener::undefined; - - if( (rand() % 2) == 0 ) ret |= SideportListener::ipv4; - else ret |= SideportListener::ipv6; - - if( (rand() % 2) == 0 ) ret |= SideportListener::udp; - else ret |= SideportListener::tcp; - - return (SideportListener::Protocol)ret; + return false; } SideportListener::Protocol SideportListener::getReachabilityProtocol(const NodeID& node){ - - map::iterator i = protocolnodes.find(node); - if(i != protocolnodes.end()) return i->second; - - protocolnodes.insert( std::make_pair(node, generateRandomProtocol()) ); + return (SideportListener::Protocol)(ipv4 | tcp); } - -// -// -//****************************************************** void SideportListener::configure( overlay::BaseOverlay* _overlay ) { Index: source/ariba/SideportListener.h =================================================================== --- source/ariba/SideportListener.h (revision 5002) +++ source/ariba/SideportListener.h (revision 5151) @@ -42,4 +42,5 @@ #include #include +#include #include "Identifiers.h" #include "CommunicationListener.h" Index: source/ariba/communication/BaseCommunication.cpp =================================================================== --- source/ariba/communication/BaseCommunication.cpp (revision 5002) +++ source/ariba/communication/BaseCommunication.cpp (revision 5151) @@ -97,5 +97,5 @@ IPv4Locator* ipv4locator = dynamic_cast(*i); - // TODO: which locators can we find to bind to? + // TODO: which locators are find to bind to? // localhost is not too bad, works when testing locally // with several instances. the manual override currently Index: source/ariba/communication/modules/transport/protlib/connectionmap.cpp =================================================================== --- source/ariba/communication/modules/transport/protlib/connectionmap.cpp (revision 5002) +++ source/ariba/communication/modules/transport/protlib/connectionmap.cpp (revision 5151) @@ -101,7 +101,16 @@ AssocData* ConnectionMap::lookup(const appladdress& addr) const { - const_addr2data_it_t hit= addr2data.find(addr); - if (hit!=addr2data.end()) return hit->second; - else return NULL; + // hack: only search for ip addresses! + const char* addr_ip = addr.get_ip_str(); + for (const_addr2data_it_t::const_iterator i=addr2data.begin(); i!=addr2data.end(); i++) { + const appladdress& map_addr = i->first; + const char* map_ip = map_addr.get_ip_str(); + if (strcmp(map_ip, addr_ip)==0) return i->second; + } + return NULL; + +// const_addr2data_it_t hit= addr2data.find(addr); +// if (hit!=addr2data.end()) return hit->second; +// else return NULL; } // end lookup Index: source/ariba/communication/modules/transport/tcp/TCPTransport.cpp =================================================================== --- source/ariba/communication/modules/transport/tcp/TCPTransport.cpp (revision 5002) +++ source/ariba/communication/modules/transport/tcp/TCPTransport.cpp (revision 5151) @@ -242,5 +242,4 @@ //std::cout << "XXXXXXXXXXXXXsending out data using tcp transport: " << data << std::endl; - const_cast(message)->dropPayload(); // prepare netmsg and send it @@ -258,4 +257,5 @@ " to address " + address->toString() << ": " + message->toString() ); + const_cast(message)->dropPayload(); string s = address->toString(); Index: source/ariba/overlay/BaseOverlay.cpp =================================================================== --- source/ariba/overlay/BaseOverlay.cpp (revision 5002) +++ source/ariba/overlay/BaseOverlay.cpp (revision 5151) @@ -39,72 +39,320 @@ #include "BaseOverlay.h" -#include "ariba/utility/misc/OvlVis.h" +#include +#include +#include +#include + #include "ariba/NodeListener.h" #include "ariba/CommunicationListener.h" #include "ariba/SideportListener.h" +#include "ariba/overlay/LinkDescriptor.h" #include "ariba/overlay/messages/OverlayMsg.h" #include "ariba/overlay/messages/JoinRequest.h" #include "ariba/overlay/messages/JoinReply.h" #include "ariba/overlay/messages/LinkRequest.h" +#include "ariba/overlay/messages/RelayMessage.h" + +#include "ariba/utility/misc/OvlVis.h" namespace ariba { namespace overlay { +#define logging_force(x) std::cout << x << std::endl; +#define logging_force1(x) std::cout << x << std::endl; + +LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) { + BOOST_FOREACH( LinkDescriptor* lp, links ) + if ((communication ? lp->communicationId : lp->overlayId) == link) + return lp; + return NULL; +} + +const LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) const { + BOOST_FOREACH( const LinkDescriptor* lp, links ) + if ((communication ? lp->communicationId : lp->overlayId) == link) + return lp; + return NULL; +} + +LinkDescriptor* BaseOverlay::getAutoDescriptor( const NodeID& node, const ServiceID& service ) { + BOOST_FOREACH( LinkDescriptor* lp, links ) + if (lp->autolink && lp->remoteNode == node && lp->service == service) + return lp; + return NULL; +} + +void BaseOverlay::eraseDescriptor( const LinkID& link, bool communication ) { + for ( vector::iterator i = links.begin(); i!= links.end(); i++) { + LinkDescriptor* ld = *i; + if ((communication ? ld->communicationId : ld->overlayId) == link) { + delete ld; + links.erase(i); + break; + } + } +} + +LinkDescriptor* BaseOverlay::addDescriptor( const LinkID& link ) { + LinkDescriptor* desc = getDescriptor( link ); + if ( desc == NULL ) { + desc = new LinkDescriptor(); + desc->overlayId = link; + links.push_back(desc); + } + return desc; +} + +/// returns a direct link relay descriptor to the given relay node +LinkDescriptor* BaseOverlay::getRelayDescriptor( const NodeID& relayNode ) { + BOOST_FOREACH( LinkDescriptor* lp, links ) + if (lp->remoteNode == relayNode && + lp->service == OverlayInterface::OVERLAY_SERVICE_ID && + lp->relay == false && + lp->up) + return lp; + return NULL; +} + +/// find a proper relay node +const NodeID BaseOverlay::findRelayNode( const NodeID& id ) { + LinkDescriptor* rld = NULL; + NodeID relayNode = NodeID::UNSPECIFIED; + + // get used next hop towards node + LinkID rlid = overlayInterface->getNextLinkId(id); + while ( relayNode.isUnspecified() && !rlid.isUnspecified() && rld == NULL ) { + + // get descriptor of first hop + rld = getDescriptor(rlid); + logging_force1( rld ); + + // is first hop a relay path? yes-> try to find real link! + if ( rld->relay ) + relayNode = getRelayDescriptor(rld->localRelay)->remoteNode; + + // no-> a proper relay node has been found + else relayNode = rld->remoteNode; + } + logging_force1( "Potential relay node " << relayNode.toString() ); + // do not return myself or use the node as relay node + if (relayNode == nodeId) + return NodeID::UNSPECIFIED; + else { + logging_force1( "Returning relay node " << relayNode.toString() ); + return relayNode; + } +} + +/// forwards a message over relays/overlay/directly using link descriptor +seqnum_t BaseOverlay::sendMessage( Message* message, const LinkDescriptor* ld ) { + + // directly send message + if ( !ld->communicationId.isUnspecified() && ld->communicationUp ) { + logging_debug("sendMessage: Sending message via Base Communication"); + return bc->sendMessage( ld->communicationId, message ); + } + + // relay message + else if ( ld->relay ) { + logging_debug("sendMessage: Relaying message to node " + << ld->remoteNode.toString() + << " using relay " << ld->localRelay + ); + + // get local relay link descriptor and mark as used for relaying + LinkDescriptor* rld = getRelayDescriptor(ld->localRelay); + if (rld==NULL) { + logging_error("sendMessage: Relay descriptor for relay " << + ld->localRelay.toString() << " unknown."); + return -1; + } + rld->markAsRelay(); + + // create a information relay message to inform the relay about + OverlayMsg overlay_msg( OverlayMsg::typeRelay, ld->service, nodeId ); + RelayMessage relayMsg( RelayMessage::typeInform, ld->remoteRelay, ld->remoteNode, ld->remoteLinkId ); + relayMsg.encapsulate( message ); + overlay_msg.encapsulate( &relayMsg ); + + // route message to relay node in order to inform it! + logging_debug("sendMessage: Sending message over relayed link with" << ld ); + overlayInterface->routeMessage(rld->remoteNode, rld->overlayId, &overlay_msg); + return 0; + } + + // route message using overlay + else { + logging_error("Could not send message"); + logging_debug( "sendMessage: Routing message to node " << ld->remoteNode.toString() ); + overlayInterface->routeMessage( ld->remoteNode, message ); + return 0; + } + + return -1; +} + +/// creates a link descriptor, apply relay semantics if possible +LinkDescriptor* BaseOverlay::createLinkDescriptor( + const NodeID& remoteNode, const ServiceID& service, const LinkID& link_id ) { + + // find listener + if( !communicationListeners.contains( service ) ) { + logging_error( "No listener found for service " << service.toString() ); + return NULL; + } + CommunicationListener* listener = communicationListeners.get( service ); + assert( listener != NULL ); + + // copy link id + LinkID linkid = link_id; + + // create link id if necessary + if ( linkid.isUnspecified() ) + linkid = LinkID::create(); + + // create relay link descriptor + NodeID relayNode = findRelayNode(remoteNode); + + // add descriptor + LinkDescriptor* ld = addDescriptor( linkid ); + ld->overlayId = linkid; + ld->service = service; + ld->listener = listener; + ld->remoteNode = remoteNode; + + // set relay node if available + ld->relay = !relayNode.isUnspecified(); + ld->localRelay = relayNode; + + if (!ld->relay) + logging_error("No relay found!"); + + // debug output + logging_debug( "Created link descriptor: " << ld ); + + return ld; +} + + +// ---------------------------------------------------------------------------- + use_logging_cpp(BaseOverlay); -BaseOverlay::BaseOverlay() - : bc(NULL), overlayInterface(NULL), nodeId(NodeID::UNSPECIFIED), - spovnetId(SpoVNetID::UNSPECIFIED), initiatorLink(LinkID::UNSPECIFIED), - state(BaseOverlayStateInvalid), sideport(&SideportListener::DEFAULT){ -} - -BaseOverlay::~BaseOverlay(){ -} - -void BaseOverlay::start( BaseCommunication& _basecomm, const NodeID& _nodeid ){ - +// ---------------------------------------------------------------------------- + +BaseOverlay::BaseOverlay() : + bc(NULL), overlayInterface(NULL), nodeId(NodeID::UNSPECIFIED), + spovnetId(SpoVNetID::UNSPECIFIED), initiatorLink(LinkID::UNSPECIFIED), + state(BaseOverlayStateInvalid), sideport(&SideportListener::DEFAULT) { +} + +BaseOverlay::~BaseOverlay() { +} + +// ---------------------------------------------------------------------------- + +void BaseOverlay::start( BaseCommunication& _basecomm, const NodeID& _nodeid ) { + logging_info("Starting..."); + + // set parameters bc = &_basecomm; nodeId = _nodeid; - logging_info("creating base overlay"); - + // register at base communication bc->registerMessageReceiver( this ); bc->registerEventListener( this ); - ovl.visCreate( ovlId, nodeId, string(""), string("") ); - ovl.visChangeNodeColor(ovlId, nodeId, OvlVis::NODE_COLORS_GREY); - -// if (Identifier(Configuration::instance().read("BASE_nodeid")) == -// Identifier(Configuration::instance().read("SOURCE"))) { -// ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CAMERA); -// } else if (Identifier(Configuration::instance().read("BASE_nodeid")) == -// Identifier(Configuration::instance().read("MR_A"))) { -// ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CHARACTER_A); -// } else if (Identifier(Configuration::instance().read("BASE_nodeid")) == -// Identifier(Configuration::instance().read("MR_W"))) { -// ovl.visChangeNodeIcon(ovlId, nodeId, OvlVis::ICON_ID_CHARACTER_W); -// } - // timer for auto link management - Timer::setInterval( 5000 ); + Timer::setInterval( 500 ); Timer::start(); } void BaseOverlay::stop() { - - logging_info("deleting base overlay"); - + logging_info("Stopping..."); + + // stop timer Timer::stop(); + + // delete oberlay interface + if(overlayInterface != NULL) { + delete overlayInterface; + overlayInterface = NULL; + } + + // unregister at base communication bc->unregisterMessageReceiver( this ); bc->unregisterEventListener( this ); - - if(overlayInterface != NULL){ - delete overlayInterface; - overlayInterface = NULL; - } -} - -void BaseOverlay::createSpoVNet(const SpoVNetID& id, const OverlayParameterSet& param, const SecurityParameterSet& sec, const QoSParameterSet& qos){ +} + +// ---------------------------------------------------------------------------- + +void BaseOverlay::joinSpoVNet(const SpoVNetID& id, + const EndpointDescriptor& bootstrapEp) { + + ovl.visShowNodeBubble ( ovlId, nodeId, "joining..." ); + logging_info( "Starting to join spovnet " << id.toString() << + " with nodeid " << nodeId.toString()); + + // contact the spovnet initiator and request to join. if the join is granted we will + // receive further information on the structure of the overlay that is used in the spovnet + // but first, we have to establish a link to the initiator... + spovnetId = id; + state = BaseOverlayStateJoinInitiated; + + initiatorLink = bc->establishLink( bootstrapEp ); + logging_info("join process initiated for " << id.toString() << "..."); +} + +void BaseOverlay::leaveSpoVNet() { + + logging_info( "Leaving spovnet " << spovnetId ); + bool ret = ( state != this->BaseOverlayStateInvalid ); + + logging_debug( "Dropping all auto-links" ); + + // gather all service links + vector servicelinks; + BOOST_FOREACH( LinkDescriptor* ld, links ) { + if( ld->service != OverlayInterface::OVERLAY_SERVICE_ID ) + servicelinks.push_back( ld->overlayId ); + } + + // drop all service links + BOOST_FOREACH( LinkID lnk, servicelinks ) + dropLink( lnk ); + + // let the node leave the spovnet overlay interface + logging_debug( "Leaving overlay" ); + if( overlayInterface != NULL ) + overlayInterface->leaveOverlay(); + + // leave spovnet + if( state != BaseOverlayStateInitiator ) { + // then, leave the spovnet baseoverlay + OverlayMsg overMsg( OverlayMsg::typeBye, nodeId ); + bc->sendMessage( initiatorLink, &overMsg ); + + // drop the link and set to correct state + bc->dropLink( initiatorLink ); + initiatorLink = LinkID::UNSPECIFIED; + } + + // change to inalid state + state = BaseOverlayStateInvalid; + ovl.visShutdown( ovlId, nodeId, string("") ); + + // inform all registered services of the event + BOOST_FOREACH( NodeListener* i, nodeListeners ) { + if( ret ) i->onLeaveCompleted( spovnetId ); + else i->onLeaveFailed( spovnetId ); + } +} + +void BaseOverlay::createSpoVNet(const SpoVNetID& id, + const OverlayParameterSet& param, + const SecurityParameterSet& sec, + const QoSParameterSet& qos) { // set the state that we are an initiator, this way incoming messages are @@ -117,5 +365,5 @@ overlayInterface = OverlayFactory::create( *this, param, nodeId, this ); - if( overlayInterface == NULL ){ + if( overlayInterface == NULL ) { logging_fatal( "overlay structure not supported" ); state = BaseOverlayStateInvalid; @@ -129,89 +377,55 @@ ovl.visChangeNodeIcon ( ovlId, nodeId, OvlVis::ICON_ID_CAMERA ); - ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN); -} - -void BaseOverlay::joinSpoVNet(const SpoVNetID& id, const EndpointDescriptor& bootstrapEp){ - - ovl.visShowNodeBubble ( ovlId, nodeId, "joining..." ); - logging_info( "starting to join spovnet " << id.toString() << - " with nodeid " << nodeId.toString()); - - spovnetId = id; - state = BaseOverlayStateJoinInitiated; - - // - // start bootstrapping for spovnets and publish our information - // - - overlayBootstrap.start( this, spovnetId, nodeId ); - overlayBootstrap.publish( bc->getEndpointDescriptor() ); - - // - // contact the spovnet initiator and request - // to join. if the join is granted we will - // receive further information on the structure - // of the overlay that is used in the spovnet - // - // but first, we have to establish a link to the initiator... - // - - initiatorLink = bc->establishLink( bootstrapEp ); - logging_info("join process initiated for " << id.toString() << "..."); -} - -void BaseOverlay::leaveSpoVNet(){ - - logging_info( "leaving spovnet " << spovnetId ); - bool ret = ( state != this->BaseOverlayStateInvalid ); - - logging_debug( "dropping all auto-links ..." ); - - // now we start leaving the spovnet: fist delete all links - // that we still have in the baseoverlay initiated by - // some services, the leave the actual overlay structure, - // then leave the spovnet - - // --> drop all service links - - vector servicelinks; - BOOST_FOREACH( LinkPair item, linkMapping ){ - if( item.second.service != OverlayInterface::OVERLAY_SERVICE_ID ) - servicelinks.push_back( item.first ); - } - BOOST_FOREACH( LinkID lnk, servicelinks ){ - // the dropLink function will remove - // the item from the linkMapping - dropLink( lnk ); - } - - // --> leave overlay structure - - logging_debug( "leaving overlay" ); - // first, leave the overlay interface - if( overlayInterface != NULL ) - overlayInterface->leaveOverlay(); - - // --> leave spovnet - - if( state != BaseOverlayStateInitiator ){ - - // then, leave the spovnet baseoverlay - OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeBye, nodeId ); - bc->sendMessage( initiatorLink, &overMsg ); - - // drop the link and set to correct state - bc->dropLink( initiatorLink ); - initiatorLink = LinkID::UNSPECIFIED; - } - - state = BaseOverlayStateInvalid; - ovl.visShutdown( ovlId, nodeId, string("") ); - - // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ){ - if( ret ) i->onLeaveCompleted( spovnetId ); - else i->onLeaveFailed( spovnetId ); - } + ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN ); +} + +// ---------------------------------------------------------------------------- + +const LinkID BaseOverlay::establishLink( + const EndpointDescriptor& ep, const NodeID& nodeid, + const ServiceID& service, const LinkID& linkid ) { + + LinkID link_id = linkid; + + // establish link via overlay + if (!nodeid.isUnspecified()) + link_id = establishLink( nodeid, service, link_id ); + + // establish link directly if only ep is known + if (nodeid.isUnspecified()) + establishLink( ep, service, link_id ); + + return link_id; +} + +/// call base communication's establish link and add link mapping +const LinkID BaseOverlay::establishLink( const EndpointDescriptor& ep, + const ServiceID& service, const LinkID& linkid ) { + + // create a new link id if necessary + LinkID link_id = linkid; + if (link_id.isUnspecified()) link_id = LinkID::create(); + + /// find a service listener + if( !communicationListeners.contains( service ) ) { + logging_error( "No listener registered for service id=" << service.toString() ); + return LinkID::UNSPECIFIED; + } + CommunicationListener* listener = communicationListeners.get( service ); + assert( listener != NULL ); + + /// establish link and add mapping + logging_info("Establishing direct link " << link_id.toString() + << " using " << ep.toString()); + + // create descriptor + LinkDescriptor* ld = addDescriptor( link_id ); + ld->overlayId = link_id; + ld->communicationId = link_id; + ld->listener = listener; + ld->service = service; + bc->establishLink( ep, link_id ); + + return link_id; } @@ -220,221 +434,197 @@ const ServiceID& service, const LinkID& link_id ) { - if( !communicationListeners.contains( service ) ){ - logging_error( "no registered listener on serviceid " << service.toString() ); - return LinkID::UNSPECIFIED; - } - - // copy link id - LinkID linkid = link_id; - - // create link id if necessary - if (linkid.isUnspecified()) linkid = LinkID::create(); + // do not establish a link to myself! + if (node == nodeId) return LinkID::UNSPECIFIED; + + // create a link descriptor + LinkDescriptor* ld = createLinkDescriptor( node, service, link_id ); + + // create link request message with own link id + uint32_t nonce = (uint32_t)(rand() ^ (rand() << 16) ^ time(NULL)); + LinkRequest link_request_msg( + nonce, &bc->getEndpointDescriptor(), false, + ld->overlayId, ld->localRelay ); + OverlayMsg overlay_msg( OverlayMsg::typeLinkRequest, service, nodeId ); + overlay_msg.encapsulate( &link_request_msg ); + pendingLinks.insert( make_pair(nonce, ld->overlayId) ); // debug message - logging_debug( "BaseOverlay called to establish link between node " << - node.toString() << " for service " << service.toString() ); - - // create link request message with own link id - OverlayMsg overlay_msg( OverlayMsg::OverlayMessageTypeLinkRequest, service, nodeId ); - uint32_t nonce = (uint32_t)(rand() ^ (rand() << 16) ^ time(NULL)); - LinkRequest link_request_msg( nonce, &bc->getEndpointDescriptor() ); - overlay_msg.encapsulate( &link_request_msg ); - pendingLinks.insert( make_pair(nonce, linkid) ); - - // debug message - logging_debug( "BaseOverlay routes LinkRequest message to node " << node.toString() ); - - // route message to overlay node - overlayInterface->routeMessage( node, &overlay_msg ); - - CommunicationListener* receiver = communicationListeners.get( service ); - assert( receiver != NULL ); - - LinkItem item (linkid, NodeID::UNSPECIFIED, service, receiver); - linkMapping.insert( make_pair(linkid, item) ); - - return linkid; -} - -const LinkID BaseOverlay::establishLink( const EndpointDescriptor& ep, - const ServiceID& service, const LinkID& linkid ){ - - if( !communicationListeners.contains( service ) ){ - logging_error( "no registered listener on serviceid " << service.toString() ); - return LinkID::UNSPECIFIED; - } - - const LinkID link = bc->establishLink( ep, linkid ); - - CommunicationListener* receiver = communicationListeners.get( service ); - assert( receiver != NULL ); - - LinkItem item (link, NodeID::UNSPECIFIED, service, receiver); - linkMapping.insert( make_pair(link, item) ); - - return link; -} - -void BaseOverlay::dropLink(const LinkID& link){ - - logging_debug( "baseoverlay dropping link " << link.toString() ); - LinkMapping::iterator i = linkMapping.find( link ); + logging_debug( + "Sending link request with" + << " link id=" << ld->overlayId + << " node id=" << ld->remoteNode.toString() + << " service id=" << ld->service.toString() + << " local relay id=" << ld->localRelay.toString() + << " nonce= " << nonce + ); + + // sending message through new link + sendMessage( &overlay_msg, ld ); + + return ld->overlayId; +} + +/// drops an established link +void BaseOverlay::dropLink(const LinkID& link) { + logging_debug( "Dropping link (initiated locally):" << link.toString() ); // find the link item to drop - if( i == linkMapping.end() ){ - logging_warn( "can't drop link, mapping unknown " << link.toString() ); + LinkDescriptor* ld = getDescriptor(link); + if( ld == NULL ) { + logging_warn( "Can't drop link, link is unknown!"); return; } - LinkItem item = i->second; - // delete all queued messages - if( item.waitingmsg.size() > 0 ){ - - logging_warn( "dropping link " << link.toString() << - " that has " << item.waitingmsg.size() << " waiting messages" ); - - item.deleteWaiting(); - } - - // erase the mapping and drop the link - linkMapping.erase( i ); - bc->dropLink( link ); - - // tell sideports and listeners of the drop - item.interface->onLinkDown( link, item.node ); - sideport->onLinkDown(link, this->nodeId, item.node, this->spovnetId ); -} - -seqnum_t BaseOverlay::sendMessage(const Message* message, const LinkID& link ){ - - logging_debug( "baseoverlay is sending data message on link " << link.toString() ); - - // + 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 + ld->listener->onLinkDown( ld->overlayId, ld->remoteNode ); + sideport->onLinkDown(ld->overlayId, this->nodeId, ld->remoteNode, this->spovnetId ); + + // do not drop relay links + if (!ld->usedAsRelay) { + // drop the link in base communication + if (ld->communicationUp) bc->dropLink( ld->communicationId ); + + // erase descriptor + eraseDescriptor( ld->overlayId ); + } else + ld->dropWhenRelaysLeft = true; +} + +// ---------------------------------------------------------------------------- + +/// internal send message, always use this functions to send messages over links +seqnum_t BaseOverlay::sendMessage( const Message* message, const LinkID& link ) { + logging_debug( "Sending data message on link " << link.toString() ); + // get the mapping for this link - // - - LinkMapping::iterator i = linkMapping.find( link ); - if( i == linkMapping.end() ){ - logging_error( "could not send message. link not found " << link.toString() ); + LinkDescriptor* ld = getDescriptor(link); + if( ld == NULL ) { + logging_error("Could not send message. " + << "Link not found id=" << link.toString()); return -1; } - i->second.markused(); - - // - // check if the link is up yet, if its an autlink queue message - // - - if( !i->second.linkup ){ - - if( i->second.autolink ){ - logging_info( "auto link " << link.toString() << " is not up yet, queueing message" ); + // check if the link is up yet, if its an auto link queue message + if( !ld->up ) { + ld->markAsUsed(); + if( ld->autolink ) { + logging_info("Auto-link " << link.toString() << " not up, queue message"); Data data = data_serialize( message ); const_cast(message)->dropPayload(); - i->second.waitingmsg.push_back( new Message(data) ); + ld->messageQueue.push_back( new Message(data) ); } else { - logging_error("link " << link.toString() << " is not up yet, dropping message" ); - } - + logging_error("Link " << link.toString() << " not up, drop message"); + } return -1; } - // - // send the message through the basecomm - // - - OverlayMsg overmsg( OverlayMsg::OverlayMessageTypeData, i->second.service, nodeId ); + // compile overlay message (has service and node id) + OverlayMsg overmsg( OverlayMsg::typeData, ld->service, nodeId ); overmsg.encapsulate( const_cast(message) ); - return bc->sendMessage( link, &overmsg ); -} - -seqnum_t BaseOverlay::sendMessage(const Message* message, const NodeID& node, const ServiceID& service){ - - LinkID link = LinkID::UNSPECIFIED; - - LinkMapping::iterator i = linkMapping.begin(); - LinkMapping::iterator iend = linkMapping.end(); - - // - // see if we find a link for this node and service destination - // - - for( ; i != iend; i++ ){ - if( i->second.node == node && i->second.service == service ){ - link = i->second.link; - break; - } - } - - // + // send message over relay/direct/overlay + return sendMessage( &overmsg, ld ); +} + +seqnum_t BaseOverlay::sendMessage(const Message* message, + const NodeID& node, const ServiceID& service) { + + // find link for node and service + LinkDescriptor* ld = getAutoDescriptor( node, service ); + // if we found no link, create an auto link - // - - if( link == LinkID::UNSPECIFIED ){ - - logging_info( "no link could be found to send message to node " << - node.toString() << " for service " << service.toString() << - ". creating auto link ..."); - - // call basecomm to create a link - link = establishLink( node, service ); + if( ld == NULL ) { + + // debug output + logging_info( "No link to send message to node " + << node.toString() << " found for service " + << service.toString() << ". Creating auto link ..." + ); // this will call onlinkup on us, if everything worked we now have a mapping - LinkMapping::iterator i = linkMapping.find( link ); - i->second.autolink = true; - - if( i == linkMapping.end() || link == LinkID::UNSPECIFIED ){ - logging_error( "failed to establish auto link to node " << node.toString() << - " for service " << service.toString() ); + LinkID link = LinkID::create(); + + // call base overlay to create a link + link = establishLink( node, service, link ); + ld = getDescriptor( link ); + if( ld == NULL ) { + logging_error( "Failed to establish auto-link."); return -1; } - - logging_debug( "establishing autolink in progress to node " - << node.toString() << " with new link-id " << link.toString() ); - - } // if( link != LinkID::UNSPECIFIED ) - - assert( link != LinkID::UNSPECIFIED ); - - // mark the link as used, as we - // now send a message through it - i->second.markused(); - - // send the message through the new link. the link may not be functional, - // but for us there is a link-id so we can send messages through it. if - // the link is not yet up and the message needs to be cached, this is the - // task of the BaseCommunication, it will cache and send it later. - return sendMessage( message, link ); -} - -const EndpointDescriptor& BaseOverlay::getEndpointDescriptor(const LinkID& link) const { - - return bc->getEndpointDescriptor( link ); -} - -const EndpointDescriptor& BaseOverlay::getEndpointDescriptor(const NodeID& node) const { - + ld->autolink = true; + + logging_debug( "Auto-link establishment in progress to node " + << node.toString() << " with link id=" << link.toString() ); + } + assert(ld != NULL); + + // mark the link as used, as we now send a message through it + ld->markAsUsed(); + + // send / queue message + return sendMessage( message, ld->overlayId ); +} + +// ---------------------------------------------------------------------------- + +const EndpointDescriptor& BaseOverlay::getEndpointDescriptor( + const LinkID& link) const { + + // return own end-point descriptor + if( link == LinkID::UNSPECIFIED ) + return bc->getEndpointDescriptor(); + + // find link descriptor. not found -> return unspecified + const LinkDescriptor* ld = getDescriptor(link); + if (ld==NULL) return EndpointDescriptor::UNSPECIFIED; + + // return endpoint-descriptor from base communication + return bc->getEndpointDescriptor( ld->communicationId ); +} + +const EndpointDescriptor& BaseOverlay::getEndpointDescriptor( + const NodeID& node) const { + + // return own end-point descriptor if( node == nodeId || node == NodeID::UNSPECIFIED ) return bc->getEndpointDescriptor(); - if( overlayInterface == NULL ){ + // no joined and request remote descriptor? -> fail! + if( overlayInterface == NULL ) { logging_error( "overlay interface not set, cannot resolve endpoint" ); return EndpointDescriptor::UNSPECIFIED; } - // TODO: if this is not a onehop overlay the operation will go asynchronously + // resolve end-point descriptor from the base-overlay routing table return overlayInterface->resolveNode( node ); } - -bool BaseOverlay::bind(CommunicationListener* listener, const ServiceID& sid){ +// ---------------------------------------------------------------------------- + +bool BaseOverlay::registerSidePort(SideportListener* _sideport) { + sideport = _sideport; + _sideport->configure( this ); +} + +bool BaseOverlay::unregisterSidePort(SideportListener* _sideport) { + sideport = &SideportListener::DEFAULT; +} + +// ---------------------------------------------------------------------------- + +bool BaseOverlay::bind(CommunicationListener* listener, const ServiceID& sid) { logging_debug( "binding communication listener " << listener - << " on serviceid " << sid.toString() ); - - if( communicationListeners.contains( sid ) ){ + << " on serviceid " << sid.toString() ); + + if( communicationListeners.contains( sid ) ) { logging_error( "some listener already registered for service id " - << sid.toString() ); + << sid.toString() ); return false; } @@ -444,25 +634,16 @@ } -bool BaseOverlay::registerSidePort(SideportListener* _sideport){ - sideport = _sideport; - _sideport->configure( this ); -} - -bool BaseOverlay::unregisterSidePort(SideportListener* _sideport){ - sideport = &SideportListener::DEFAULT; -} - -bool BaseOverlay::unbind(CommunicationListener* listener, const ServiceID& sid){ - logging_debug( "unbinding listener " << listener - << " from serviceid " << sid.toString() ); - - if( !communicationListeners.contains( sid ) ){ + +bool BaseOverlay::unbind(CommunicationListener* listener, const ServiceID& sid) { + logging_debug( "unbinding listener " << listener << " from serviceid " << sid.toString() ); + + if( !communicationListeners.contains( sid ) ) { logging_warn( "cannot unbind listener. no listener registered on service id " << sid.toString() ); return false; } - if( communicationListeners.get(sid) != listener ){ + if( communicationListeners.get(sid) != listener ) { logging_warn( "listener bound to service id " << sid.toString() - << " is different than listener trying to unbind" ); + << " is different than listener trying to unbind" ); return false; } @@ -472,554 +653,734 @@ } -bool BaseOverlay::bind(NodeListener* listener){ - logging_debug( "binding node listener " << listener ); - - NodeListenerVector::iterator i = find( nodeListeners.begin(), nodeListeners.end(), listener ); - if( i != nodeListeners.end() ){ - logging_warn( "node listener " << listener << " is already bound, cannot bind" ); +// ---------------------------------------------------------------------------- + +bool BaseOverlay::bind(NodeListener* listener) { + logging_debug( "Binding node listener " << listener ); + + // already bound? yes-> warning + NodeListenerVector::iterator i = + find( nodeListeners.begin(), nodeListeners.end(), listener ); + if( i != nodeListeners.end() ) { + logging_warn("Node listener " << listener << " is already bound!" ); return false; } + // no-> add nodeListeners.push_back( listener ); return true; } -bool BaseOverlay::unbind(NodeListener* listener){ - logging_debug( "unbinding node listener " << listener ); - +bool BaseOverlay::unbind(NodeListener* listener) { + logging_debug( "Unbinding node listener " << listener ); + + // already unbound? yes-> warning NodeListenerVector::iterator i = find( nodeListeners.begin(), nodeListeners.end(), listener ); - if( i == nodeListeners.end() ){ - logging_warn( "node listener " << listener << " is not bound, cannot unbind" ); + if( i == nodeListeners.end() ) { + logging_warn( "Node listener " << listener << " is not bound!" ); return false; } + // no-> remove nodeListeners.erase( i ); return true; } -void BaseOverlay::onLinkUp(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){ - - logging_debug( "base overlay received linkup event " + id.toString() ); - // TODO: updateOvlVis( getNodeID(id) ); - - // - // if we get up a link while we are in the - // join phase and this is the link that - // we have initiated towards the spovnet owner - // continue the join process by sending - // a join request message through the link - // - - if( state == BaseOverlayStateJoinInitiated && id == initiatorLink){ - +// ---------------------------------------------------------------------------- + +void BaseOverlay::onLinkUp(const LinkID& id, + const NetworkLocator* local, const NetworkLocator* remote) { + logging_debug( "Link up with base communication link id=" << id ); + + // get descriptor for link + LinkDescriptor* ld = getDescriptor(id, true); + + // handle initiator link + if(state == BaseOverlayStateJoinInitiated && id == initiatorLink) { logging_info( "Join has been initiated by me and the link is now up. " << - "sending out join request for SpoVNet " << spovnetId.toString() + "Sending out join request for SpoVNet " << spovnetId.toString() ); - OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeJoinRequest, nodeId ); - JoinRequest joinmsg( spovnetId, nodeId ); - overMsg.encapsulate( &joinmsg ); - - state = BaseOverlayStateJoinInitiated; // state remains in JoinInitiated - bc->sendMessage( id, &overMsg ); - + // send join request message + OverlayMsg overlayMsg( OverlayMsg::typeJoinRequest, nodeId ); + JoinRequest joinRequest( spovnetId, nodeId ); + overlayMsg.encapsulate( &joinRequest ); + bc->sendMessage( id, &overlayMsg ); return; - - } // if( state == BaseOverlayStateJoinInitiated && id == initiatorLink) - - // - // otherwise this is a link initiated by a service - // then we exchange update messages to exchange the - // service id and node id for the link. in this case - // we should have a link mapping for this link. if - // we have no link mapping this link was initiated by - // the remote side. - // - - LinkMapping::iterator i = linkMapping.find( id ); - - if( i == linkMapping.end() ){ - - LinkItem item (id, NodeID::UNSPECIFIED, ServiceID::UNSPECIFIED, &CommunicationListener::DEFAULT ); - linkMapping.insert( make_pair(id, item) ); - + } + + // no link found? -> link establishment from remote, add one! + if (ld == NULL) { + ld = addDescriptor( id ); + logging_debug( "onLinkUp (remote request) descriptor: " << ld ); + + // update descriptor + ld->fromRemote = true; + ld->communicationId = id; + ld->communicationUp = true; + ld->markAsUsed(); + + // 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 { - - logging_debug( "sending out OverlayMessageTypeUpdate" << - " for service " << i->second.service.toString() << - " with local node id " << nodeId.toString() << - " on link " << id.toString() ); - - OverlayMsg overMsg( - OverlayMsg::OverlayMessageTypeUpdate, - i->second.service, - nodeId - ); - - bc->sendMessage( id, &overMsg ); - i->second.markused(); - - } // if( i == linkMapping.end() ) - - // the link is only valid for the service when we receive - // the OverlayMessageTypeUpdate from the remote node and - // have the nodeid and serviceid for the link! -} - -void BaseOverlay::onLinkDown(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){ - - logging_debug( "link went down " << id.toString() ); - - // - // tell the service that the link went - // down and remove the mapping - // - - LinkMapping::iterator i = linkMapping.find( id ); - if( i == linkMapping.end() ) { - // this can also be one of the baseoverlay links that - // no mapping is stored for. therefore we issue no warning. - // it can also be a link that has been dropped and the - // mapping is already deleted in the dropLink function. - // also, the service notification is issued then in dropLink - return; - } - - i->second.interface->onLinkDown( id, i->second.node ); - sideport->onLinkDown( id, this->nodeId, i->second.node, this->spovnetId ); - - // delete all queued messages - if( i->second.waitingmsg.size() > 0 ){ - - logging_warn( "dropping link " << id.toString() << - " that has " << i->second.waitingmsg.size() << " waiting messages" ); - - i->second.deleteWaiting(); - } - - linkMapping.erase( i ); -} - -void BaseOverlay::onLinkChanged(const LinkID& id, const NetworkLocator* oldlocal, const NetworkLocator* newlocal, const NetworkLocator* oldremote, const NetworkLocator* newremote){ - - logging_debug( "link changed " << id.toString() ); - - // - // tell the service that the link changed - // - - LinkMapping::iterator i = linkMapping.find( id ); - if( i == linkMapping.end() ) return; - - i->second.interface->onLinkChanged( id, i->second.node ); - sideport->onLinkChanged( id, this->nodeId, i->second.node, this->spovnetId ); - - // TODO call onLinkQoSChanged? - - i->second.markused(); -} - -void BaseOverlay::onLinkFail(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote){ - - logging_debug( "link failed " << id.toString() ); - - // - // tell the service that the link failed - // - - LinkMapping::iterator i = linkMapping.find( id ); - if( i == linkMapping.end() ) return; - - i->second.interface->onLinkFail( id, i->second.node ); - sideport->onLinkFail( id, this->nodeId, i->second.node, this->spovnetId ); - - i->second.markused(); -} - -void BaseOverlay::onLinkQoSChanged(const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote, const QoSParameterSet& qos) { - - logging_debug( "link qos changed " << id.toString() ); - - // - // tell the service that the link qos has changed - // - - LinkMapping::iterator i = linkMapping.find( id ); - if( i == linkMapping.end() ) return; - - // TODO: convert QoSParameterSet to the LinkProperties properties - // TODO: currently not in the interface: i->second.interface->onLinkQoSChanged( id, i->second.node, LinkProperties::DEFAULT ); - - i->second.markused(); -} - -bool BaseOverlay::onLinkRequest( const LinkID& id, const NetworkLocator* local, const NetworkLocator* remote ){ - - // also see in the receiveMessage function. there the higher layer service - // is asked whether to accept link requests, but there a basic link association is - // already built up, so we know the node id - logging_debug("received link request from " << remote->toString() << ", accepting"); + logging_debug( "onLinkUp descriptor (initiated locally):" << ld ); + + // note: necessary to validate the link on the remote side! + logging_debug( "Sending out update" << + " for service " << ld->service.toString() << + " with local node id " << nodeId.toString() << + " on link " << ld->overlayId.toString() ); + + // update descriptor + ld->markAsUsed(); + ld->communicationUp = true; + + // if link is a relayed link ->convert to direct link + if (ld->relay) { + ld->up = true; + ld->relay = false; + ld->localRelay = NodeID::UNSPECIFIED; + OverlayMsg overMsg( OverlayMsg::typeDirectLink, ld->service, nodeId ); + overMsg.setRelayLink( ld->remoteLinkId ); + bc->sendMessage( ld->communicationId, &overMsg ); + } + + // compile and send update message + OverlayMsg overlayMsg( OverlayMsg::typeUpdate, ld->service, nodeId ); + overlayMsg.setAutoLink( ld->autolink ); + bc->sendMessage( ld->communicationId, &overlayMsg ); + } +} + +void BaseOverlay::onLinkDown(const LinkID& id, + const NetworkLocator* local, const NetworkLocator* remote) { + + // get descriptor for link + LinkDescriptor* ld = getDescriptor(id, true); + if ( ld == NULL ) return; // not found? ->ignore! + logging_force( "onLinkDown descriptor: " << ld ); + + // inform listeners about link down + ld->communicationUp = false; + ld->listener->onLinkDown( ld->overlayId, ld->remoteNode ); + sideport->onLinkDown( 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); +} + +void BaseOverlay::onLinkChanged(const LinkID& id, + const NetworkLocator* oldlocal, const NetworkLocator* newlocal, + const NetworkLocator* oldremote, const NetworkLocator* 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->markAsUsed(); +} + +void BaseOverlay::onLinkFail(const LinkID& id, + const NetworkLocator* local, const NetworkLocator* remote) { + logging_debug( "Link fail 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 failed id=" << ld->overlayId.toString() ); + + // inform listeners + ld->listener->onLinkFail( ld->overlayId, ld->remoteNode ); + sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId ); + + // autolinks: refresh timestamp + ld->markAsUsed(); +} + +void BaseOverlay::onLinkQoSChanged(const LinkID& id, const NetworkLocator* local, + const NetworkLocator* 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() ); + + // autolinks: refresh timestamp + ld->markAsUsed(); +} + +bool BaseOverlay::onLinkRequest( const LinkID& id, const NetworkLocator* local, + const NetworkLocator* remote ) { + logging_debug("Accepting link request from " << remote->toString() ); return true; } - +/// handles a message from base communication bool BaseOverlay::receiveMessage(const Message* message, - const LinkID& link, const NodeID& - /*the nodeid is invalid in this case! removed var to prevent errors*/ ){ + const LinkID& link, const NodeID& ) { + // get descriptor for link + LinkDescriptor* ld = getDescriptor( link, true ); + + // link known? + if (ld == NULL) { // no-> handle with unspecified params + logging_debug("Received message from base communication, link descriptor unknown" ); + return handleMessage( message, LinkID::UNSPECIFIED, link, NodeID::UNSPECIFIED ); + } else { // yes -> handle with overlay link id + logging_debug("Received message from base communication, link id=" << ld->overlayId.toString() ); + return handleMessage( message, ld->overlayId, link, NodeID::UNSPECIFIED ); + } +} + +// ---------------------------------------------------------------------------- + +/// handles a message from an overlay +void BaseOverlay::incomingRouteMessage( Message* msg, const LinkID& link, const NodeID& source ) { + logging_debug("Received message from overlay -- " + << " link id=" << link.toString() + << " node id=" << source.toString() ); + handleMessage( msg, link, LinkID::UNSPECIFIED, source ); +} + +// ---------------------------------------------------------------------------- + +/// handles an incoming message +bool BaseOverlay::handleMessage( const Message* message, + const LinkID& boLink, const LinkID& bcLink, const NodeID& remoteNode ) { + logging_debug( "Handling message: " << message->toString()); // decapsulate overlay message - logging_debug( "receiveMessage: " << message->toString()); - OverlayMsg* overlayMsg = const_cast(message)->decapsulate(); + OverlayMsg* overlayMsg = + const_cast(message)->decapsulate(); if( overlayMsg == NULL ) return false; // mark the link as in action - LinkMapping::iterator item = linkMapping.find( link ); - if( item != linkMapping.end() ) item->second.markused(); - - /* ************************************************************************ - /* handle user date that we forward to the appropriate service using the - * service id in the message. as we don't know the class of message that - * the service handles, we forward it as a pure Message - */ - if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeData) ) { - - logging_debug( "baseoverlay received message of type OverlayMessageTypeData" ); - - const ServiceID& service = overlayMsg->getService(); - CommunicationListener* serviceListener = communicationListeners.get( service ); - - logging_debug( "received data for service " << service.toString() ); - - if( serviceListener != NULL ) - serviceListener->onMessage( overlayMsg, overlayMsg->getSourceNode(), link ); - - return true; - - } // if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeData) ) - - /* ************************************************************************ - /* Handle spovnet instance join requests - */ - else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinRequest) ){ - - logging_debug( - "baseoverlay received message of type OverlayMessageTypeJoinRequest" - ); - - JoinRequest* joinReq = overlayMsg->decapsulate(); - logging_info( "received join request for spovnet " << - joinReq->getSpoVNetID().toString() ); - - /* make sure that the node actually wants to join - * the correct spovnet id that we administrate */ - if( joinReq->getSpoVNetID() != spovnetId ){ - logging_error( "received join request for spovnet we don't handle " << + LinkDescriptor* ld = getDescriptor(boLink); + if (ld == NULL) ld = getDescriptor(bcLink, true); + if (ld != NULL) { + ld->markAsUsed(); + ld->markAlive(); + } + + switch ( overlayMsg->getType() ) { + // --------------------------------------------------------------------- + // Handle spovnet instance join requests + // --------------------------------------------------------------------- + case OverlayMsg::typeJoinRequest: { + + // decapsulate message + JoinRequest* joinReq = overlayMsg->decapsulate(); + logging_info( "Received join request for spovnet " << joinReq->getSpoVNetID().toString() ); - return false; - } - - // - // only if all services allow the node to join it is allowed - // using the isJoinAllowed interface security policies can be - // implemented by higher layer services - // - - // TODO: here you can implement mechanisms to deny joining of a node - bool allow = true; - - logging_info( "sending back join reply for spovnet " << - spovnetId.toString() << " to node " << - overlayMsg->getSourceNode().toString() << - ". result: " << (allow ? "allowed" : "denied") ); - - joiningNodes.push_back( overlayMsg->getSourceNode() ); - - // - // send back our spovnetid, default overlay parameters, join allow - // result, and ourself as the end-point to bootstrap the overlay against - // - - assert( overlayInterface != NULL ); - OverlayParameterSet parameters = overlayInterface->getParameters(); - - OverlayMsg retmsg( OverlayMsg::OverlayMessageTypeJoinReply, nodeId ); - JoinReply replyMsg( spovnetId, parameters, + + // check spovnet id + if( joinReq->getSpoVNetID() != spovnetId ) { + logging_error( + "Received join request for spovnet we don't handle " << + joinReq->getSpoVNetID().toString() ); + return false; + } + + // TODO: here you can implement mechanisms to deny joining of a node + bool allow = true; + logging_info( "Sending join reply for spovnet " << + spovnetId.toString() << " to node " << + overlayMsg->getSourceNode().toString() << + ". Result: " << (allow ? "allowed" : "denied") ); + joiningNodes.push_back( overlayMsg->getSourceNode() ); + + // return overlay parameters + assert( overlayInterface != NULL ); + logging_debug( "Using bootstrap end-point " + << getEndpointDescriptor().toString() ) + OverlayParameterSet parameters = overlayInterface->getParameters(); + OverlayMsg retmsg( OverlayMsg::typeJoinReply, nodeId ); + JoinReply replyMsg( spovnetId, parameters, allow, getEndpointDescriptor() ); - - retmsg.encapsulate(&replyMsg); - bc->sendMessage( link, &retmsg ); - - return true; - - } // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinRequest)) - - /* ************************************************************************ - * handle replies to spovnet instance join requests - */ - else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinReply) && - state == BaseOverlayStateJoinInitiated){ - - logging_debug( - "baseoverlay received message of type OverlayMessageTypeJoinReply"); - - JoinReply* replyMsg = overlayMsg->decapsulate(); - logging_info( "received spovnet join reply" ); - - // ensure that we actually wanted to get into the spovnet whose id is - // in the message - if( replyMsg->getSpoVNetID() != spovnetId ){ - logging_error( "received spovnet join reply for spovnet " << - replyMsg->getSpoVNetID().toString() << - " but we wanted to join spovnet " << + retmsg.encapsulate(&replyMsg); + bc->sendMessage( bcLink, &retmsg ); + return true; + } + + // --------------------------------------------------------------------- + // handle replies to spovnet instance join requests + // --------------------------------------------------------------------- + case OverlayMsg::typeJoinReply: { + + // decapsulate message + logging_debug("received join reply message"); + JoinReply* replyMsg = overlayMsg->decapsulate(); + assert(state == BaseOverlayStateJoinInitiated); + + // correct spovnet? + if( replyMsg->getSpoVNetID() != spovnetId ) { // no-> fail + logging_error( "Received SpoVNet join reply for " << + replyMsg->getSpoVNetID().toString() << + " != " << spovnetId.toString() ); + return false; + } + + // access granted? no -> fail + if( !replyMsg->getJoinAllowed() ) { + logging_error( "Our join request has been denied" ); + + // drop initiator link + bc->dropLink( initiatorLink ); + initiatorLink = LinkID::UNSPECIFIED; + state = BaseOverlayStateInvalid; + + // inform all registered services of the event + BOOST_FOREACH( NodeListener* i, nodeListeners ) + i->onJoinFailed( spovnetId ); + return true; + } + + // access has been granted -> continue! + logging_info("Join request has been accepted for spovnet " << spovnetId.toString() ); - // state does not change here, maybe the reply does come in later - return false; - } - - // if we did not get access to the spovnet notify of the failure and - // close the link to the initiator - if( ! replyMsg->getJoinAllowed() ){ - - logging_error( "our join request has been denied" ); - - bc->dropLink( initiatorLink ); - initiatorLink = LinkID::UNSPECIFIED; - state = BaseOverlayStateInvalid; - - // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ){ + // create overlay structure from spovnet parameter set + overlayInterface = OverlayFactory::create( + *this, replyMsg->getParam(), nodeId, this ); + + // overlay structure supported? no-> fail! + if( overlayInterface == NULL ) { + logging_error( "overlay structure not supported" ); + + bc->dropLink( initiatorLink ); + initiatorLink = LinkID::UNSPECIFIED; + state = BaseOverlayStateInvalid; + + // inform all registered services of the event + BOOST_FOREACH( NodeListener* i, nodeListeners ) i->onJoinFailed( spovnetId ); + + return true; } - return true; - } - - logging_info( "join request has been accepted for spovnet " << - spovnetId.toString() ); - - // if we did get access to the spovnet we try to create the overlay - // structure as given in the reply message - overlayInterface = OverlayFactory::create( *this, - replyMsg->getParam(), nodeId, this ); - - if( overlayInterface == NULL ){ - logging_error( "overlay structure not supported" ); - - bc->dropLink( initiatorLink ); - initiatorLink = LinkID::UNSPECIFIED; - state = BaseOverlayStateInvalid; + // everything ok-> join the overlay! + state = BaseOverlayStateCompleted; + overlayInterface->createOverlay(); + logging_debug( "Using bootstrap end-point " + << replyMsg->getBootstrapEndpoint().toString() ); + overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() ); + + // update ovlvis + ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN); // inform all registered services of the event BOOST_FOREACH( NodeListener* i, nodeListeners ) - i->onJoinFailed( spovnetId ); + i->onJoinCompleted( spovnetId ); return true; } - /* now start the join process for the overlay. the join process for the - * spovnet baseoverlay is now complete. we use the endpoint for overlay - * structure bootstrapping that the initiator provided in his reply - * message */ - state = BaseOverlayStateCompleted; - ovl.visChangeNodeColor( ovlId, nodeId, OvlVis::NODE_COLORS_GREEN); - - overlayInterface->createOverlay(); - overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() ); - - // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ){ - i->onJoinCompleted( spovnetId ); - } - - return true; - - } // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeJoinReply) && state == BaseOverlayStateJoinInitiated) - - - /* ************************************************************************ - * handle update messages for link establishment - */ - else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeUpdate) ){ - - logging_debug( - "baseoverlay received message of type OverlayMessageTypeUpdate" - ); - - const NodeID& sourcenode = overlayMsg->getSourceNode(); - const ServiceID& service = overlayMsg->getService(); - - // linkmapping for the link available? no-> ignore - LinkMapping::iterator i = linkMapping.find( link ); - if( i == linkMapping.end() ) { - logging_warn( "received overlay update message for link " << - link.toString() << " for which we have no mapping" ); - return false; - } - - // update our link mapping information for this link - bool changed = ( i->second.node != sourcenode ) || ( i->second.service != service ); - i->second.node = sourcenode; - i->second.service = service; - - // if our link information changed, we send out an update, too - if( changed ){ - OverlayMsg overMsg( OverlayMsg::OverlayMessageTypeUpdate, i->second.service, nodeId ); - bc->sendMessage( link, &overMsg ); - } - - // set the correct listener service for the linkitem - // now we can tell the registered service of the linkup event - if( !communicationListeners.contains( service ) ){ - logging_warn( "linkup event for service that has not been registered" ); - return false; - } - - CommunicationListener* iface = communicationListeners.get( service ); - if( iface == NULL || iface == &CommunicationListener::DEFAULT ){ - logging_warn( "linkup event for service that has been registered " - "with a NULL interface" ); + // --------------------------------------------------------------------- + // handle data forward messages + // --------------------------------------------------------------------- + case OverlayMsg::typeData: { + + // get service + const ServiceID& service = overlayMsg->getService(); + logging_debug( "received data for service " << service.toString() ); + + // find listener + CommunicationListener* listener = + communicationListeners.get( service ); + if( listener == NULL ) return true; + + // delegate data message + listener->onMessage( overlayMsg, + overlayMsg->getSourceNode(), ld->overlayId ); + return true; } - i->second.interface = iface; - i->second.markused(); - - // ask the service whether it wants to accept this link - if( !iface->onLinkRequest(sourcenode) ){ - - logging_debug("link " << link.toString() << - " has been denied by service " << service.toString() << ", dropping link"); - - // prevent onLinkDown calls to the service - i->second.interface = &CommunicationListener::DEFAULT; - // drop the link - dropLink( link ); - - return true; - } - - // - // link has been accepted, link is now up, send messages out first - // - - i->second.linkup = true; - logging_debug("link " << link.toString() << - " has been accepted by service " << service.toString() << " and is now up"); - - if( i->second.waitingmsg.size() > 0 ){ - logging_info( "sending out queued messages on link " << link.toString() ); - - BOOST_FOREACH( Message* msg, i->second.waitingmsg ){ - sendMessage( msg, link ); - delete msg; + // --------------------------------------------------------------------- + // handle update messages for link establishment + // --------------------------------------------------------------------- + case OverlayMsg::typeUpdate: { + logging_debug("Received type update message on link " << ld ); + + // get info + const NodeID& sourcenode = overlayMsg->getSourceNode(); + const ServiceID& service = overlayMsg->getService(); + + // no link descriptor available -> error! + if( ld == NULL ) { + logging_warn( "received overlay update message for link " << + ld->overlayId.toString() << " for which we have no mapping" ); + return false; } - i->second.waitingmsg.clear(); - } - - // call the notification functions - iface->onLinkUp( link, sourcenode ); - sideport->onLinkUp( link, nodeId, sourcenode, this->spovnetId ); - - return true; - - } // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeUpdate) ) - - /* ************************************************************************ - * handle bye messages - */ - else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeBye) ) { - - logging_debug( "BaseOverlay received message of type OverlayMessageTypeBye" ); - logging_debug( "Received bye message from " << - overlayMsg->getSourceNode().toString() ); - - /* if we are the initiator and receive a bye from a node - * the node just left. if we are a node and receive a bye - * from the initiator, we have to close, too. - */ - if( overlayMsg->getSourceNode() == spovnetInitiator ){ - - bc->dropLink( initiatorLink ); - initiatorLink = LinkID::UNSPECIFIED; - state = BaseOverlayStateInvalid; - - logging_fatal( "initiator ended spovnet" ); - - // inform all registered services of the event - BOOST_FOREACH( NodeListener* i, nodeListeners ){ - i->onLeaveFailed( spovnetId ); + // update our link mapping information for this link + bool changed = + ( ld->remoteNode != sourcenode ) || ( ld->service != service ); + ld->remoteNode = sourcenode; + ld->service = service; + ld->autolink = overlayMsg->isAutoLink(); + + // if our link information changed, we send out an update, too + if( changed ) { + OverlayMsg overMsg( OverlayMsg::typeUpdate, ld->service, nodeId ); + overMsg.setAutoLink(ld->autolink); + bc->sendMessage( ld->communicationId, &overMsg ); } - } else { - // a node that said goodbye and we are the initiator don't have to - // do much here, as the node also will go out of the overlay - // structure - logging_info( "node left " << overlayMsg->getSourceNode() ); - } - - return true; - - } // else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeBye)) - - /* ************************************************************************ - * handle link request forwarded through the overlay - */ - else if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeLinkRequest)) { - LinkRequest* linkReq = overlayMsg->decapsulate(); - const ServiceID& service = overlayMsg->getService(); - if (linkReq->isReply()) { - - // find link - PendingLinkMap::iterator i = pendingLinks.find( linkReq->getNonce() ); - if ( i == pendingLinks.end() ) { - logging_error( "Nonce not found in link request" ); + // service registered? no-> error! + if( !communicationListeners.contains( service ) ) { + logging_warn( "Link up: event listener has not been registered" ); + return false; + } + + // default or no service registered? + CommunicationListener* listener = communicationListeners.get( service ); + if( listener == NULL || listener == &CommunicationListener::DEFAULT ) { + logging_warn("Link up: event listener is default or null!" ); return true; } - // debug message - logging_debug( "LinkRequest reply received. Establishing link " - << i->second << " to " << (linkReq->getEndpoint()->toString()) - << " for service " << service.toString() - << " with nonce " << linkReq->getNonce() + // update descriptor + ld->listener = listener; + ld->markAsUsed(); + ld->markAlive(); + + // ask the service whether it wants to accept this link + if( !listener->onLinkRequest(sourcenode) ) { + + logging_debug("Link id=" << ld->overlayId.toString() << + " has been denied by service " << service.toString() << ", dropping link"); + + // prevent onLinkDown calls to the service + ld->listener = &CommunicationListener::DEFAULT; + + // drop the link + dropLink( ld->overlayId ); + return true; + } + + // set link up + ld->up = true; + logging_debug( + "Link " << ld->overlayId.toString() + << " has been accepted by service " << service.toString() + << " and is now up" ); - // establishing link - bc->establishLink( *linkReq->getEndpoint(), i->second ); - } else { - OverlayMsg overlay_msg( OverlayMsg::OverlayMessageTypeLinkRequest, service, nodeId ); - LinkRequest link_request_msg( - linkReq->getNonce(), &bc->getEndpointDescriptor(), true ); - overlay_msg.encapsulate( &link_request_msg ); - - // debug message - logging_debug( "Sending LinkRequest reply for link with nonce " << - linkReq->getNonce() ); - - // route message back over overlay - overlayInterface->routeMessage( - overlayMsg->getSourceNode(), &overlay_msg - ); - } - } // if( overlayMsg->isType(OverlayMsg::OverlayMessageTypeLinkRequest)) - - /* ************************************************************************ - * unknown message type ... error! - */ - else { - - logging_error( "received message in invalid state! don't know " << - "what to do with this message of type " << - overlayMsg->getType() ); - return false; - - } // else - + // auto links: link has been accepted -> send queued messages + if( ld->messageQueue.size() > 0 ) { + logging_info( "sending out queued messages on link " << + ld->overlayId.toString() ); + BOOST_FOREACH( Message* msg, ld->messageQueue ) { + sendMessage( msg, ld->overlayId ); + delete msg; + } + ld->messageQueue.clear(); + } + + // call the notification functions + listener->onLinkUp( ld->overlayId, sourcenode ); + sideport->onLinkUp( ld->overlayId, nodeId, sourcenode, this->spovnetId ); + + return true; + } + + // --------------------------------------------------------------------- + // handle bye messages + // --------------------------------------------------------------------- + case OverlayMsg::typeBye: { + logging_debug( "received bye message from " << + overlayMsg->getSourceNode().toString() ); + + /* if we are the initiator and receive a bye from a node + * the node just left. if we are a node and receive a bye + * from the initiator, we have to close, too. + */ + if( overlayMsg->getSourceNode() == spovnetInitiator ) { + + bc->dropLink( initiatorLink ); + initiatorLink = LinkID::UNSPECIFIED; + state = BaseOverlayStateInvalid; + + logging_fatal( "initiator ended spovnet" ); + + // inform all registered services of the event + BOOST_FOREACH( NodeListener* i, nodeListeners ) + i->onLeaveFailed( spovnetId ); + + } else { + // a node that said goodbye and we are the initiator don't have to + // do much here, as the node also will go out of the overlay + // structure + logging_info( "node left " << overlayMsg->getSourceNode() ); + } + + return true; + + } + + // --------------------------------------------------------------------- + // handle link request forwarded through the overlay + // --------------------------------------------------------------------- + case OverlayMsg::typeLinkRequest: { + + // decapsulate message + LinkRequest* linkReq = overlayMsg->decapsulate(); + const ServiceID& service = overlayMsg->getService(); + + // is request reply? + if ( linkReq->isReply() ) { + + // find link + PendingLinkMap::iterator i = pendingLinks.find( linkReq->getNonce() ); + if ( i == pendingLinks.end() ) { + logging_error( "Nonce not found in link request" ); + return true; + } + + // debug message + logging_debug( "Link request reply received. Establishing link " + << i->second << " to " << (linkReq->getEndpoint()->toString()) + << " for service " << service.toString() + << " with nonce " << linkReq->getNonce() + << " using relay " << linkReq->getRelay().toString() + << " and remote link id=" << linkReq->getRemoteLinkId() + ); + + // get descriptor + LinkDescriptor* ldn = getDescriptor(i->second); + + // check if link request reply has a relay node ... + if (!linkReq->getRelay().isUnspecified()) { // yes-> + ldn->up = true; + ldn->relay = true; + if (ldn->localRelay.isUnspecified()) { + logging_error("On LinkRequest reply: local relay is unspecifed on link " << ldn ); + showLinkState(); + } + ldn->remoteRelay = linkReq->getRelay(); + ldn->remoteLinkId = linkReq->getRemoteLinkId(); + ldn->remoteNode = overlayMsg->getSourceNode(); + + ldn->markAlive(); + + // compile and send update message + OverlayMsg _overlayMsg( OverlayMsg::typeUpdate, ldn->service, nodeId ); + _overlayMsg.setAutoLink(ldn->autolink); + sendMessage( &_overlayMsg, ldn ); + + // auto links: link has been accepted -> send queued messages + if( ldn->messageQueue.size() > 0 ) { + logging_info( "Sending out queued messages on link " << + ldn->overlayId.toString() ); + BOOST_FOREACH( Message* msg, ldn->messageQueue ) { + sendMessage( msg, ldn->overlayId ); + delete msg; + } + ldn->messageQueue.clear(); + } + + ldn->listener->onLinkUp( ldn->overlayId, ldn->remoteNode ); + + // try to establish a direct link + ldn->communicationId = + bc->establishLink( *linkReq->getEndpoint(), i->second ); + } + + // no relay node-> use overlay routing + else { + ldn->up = true; + + // establish direct link + ldn->communicationId = + bc->establishLink( *linkReq->getEndpoint(), i->second ); + } + } else { + logging_debug( "Link request received from node id=" + << overlayMsg->getSourceNode() ); + + // create link descriptor + LinkDescriptor* ldn = + createLinkDescriptor(overlayMsg->getSourceNode(), + overlayMsg->getService(), LinkID::UNSPECIFIED ); + assert(!ldn->overlayId.isUnspecified()); + + // create reply message + OverlayMsg overlay_msg( OverlayMsg::typeLinkRequest, service, nodeId ); + LinkRequest link_request_msg( + linkReq->getNonce(), + &bc->getEndpointDescriptor(), + true, ldn->overlayId, ldn->localRelay + ); + overlay_msg.encapsulate( &link_request_msg ); + + // debug message + logging_debug( "Sending LinkRequest reply for link with nonce " << + linkReq->getNonce() ); + + // if this is a relay link-> update information & inform listeners + if (!linkReq->getRelay().isUnspecified()) { + // set flags + ldn->up = true; + ldn->relay = true; + if (ldn->localRelay.isUnspecified()) { + logging_error("On LinkRequest request: local relay is unspecifed on link " << ldn ); + showLinkState(); + } + ldn->remoteRelay = linkReq->getRelay(); + ldn->remoteNode = overlayMsg->getSourceNode(); + ldn->remoteLinkId = linkReq->getRemoteLinkId(); + ldn->listener->onLinkUp( ldn->overlayId, ldn->remoteNode ); + } + + // route message back over overlay + sendMessage( &overlay_msg, ldn ); + } + return true; + } + + // --------------------------------------------------------------------- + // handle relay message to forward messages + // --------------------------------------------------------------------- + case OverlayMsg::typeRelay: { + + // decapsulate message + RelayMessage* relayMsg = overlayMsg->decapsulate(); + + // is relay message informative? + switch (relayMsg->getType()) { + + // handle relay notification + case RelayMessage::typeInform: { + logging_info("Received relay information message with" + << " relay " << relayMsg->getRelayNode() + << " destination " << relayMsg->getDestNode() ); + + // mark incoming link as relay + if (ld!=NULL) ld->markAsRelay(); + + // am I the destination of this message? yes-> + if (relayMsg->getDestNode() == nodeId ) { + // deliver relay message locally! + logging_debug("Relay message reached destination. Handling the message."); + handleMessage( relayMsg, relayMsg->getDestLink(), LinkID::UNSPECIFIED, remoteNode ); + return true; + } + + // create route message + OverlayMsg _overMsg( *overlayMsg ); + RelayMessage _relayMsg( *relayMsg ); + _relayMsg.setType( RelayMessage::typeRoute ); + _overMsg.encapsulate( &_relayMsg ); + + // forward message + if (relayMsg->getRelayNode() == nodeId || relayMsg->getRelayNode().isUnspecified()) { + logging_info("Routing relay message to " << relayMsg->getDestNode().toString() ); + overlayInterface->routeMessage(relayMsg->getDestNode(), &_overMsg ); + } else { + logging_info("Routing relay message to " << relayMsg->getRelayNode().toString() ); + overlayInterface->routeMessage(relayMsg->getRelayNode(), &_overMsg ); + } + return true; + } + + // handle relay routing + case RelayMessage::typeRoute: { + logging_info("Received relay route message with" + << " relay " << relayMsg->getRelayNode() + << " destination " << relayMsg->getDestNode() ); + + // mark incoming link as relay + if (ld!=NULL) ld->markAsRelay(); + + // am I the destination of this message? yes-> + if (relayMsg->getDestNode() == nodeId ) { + // deliver relay message locally! + logging_debug("Relay message reached destination. Handling the message."); + handleMessage( relayMsg, relayMsg->getDestLink(), LinkID::UNSPECIFIED, remoteNode ); + return true; + } + + // am I the relay for this message? yes-> + if (relayMsg->getRelayNode() == nodeId ) { + logging_debug("I'm the relay for this message. Sending to destination."); + OverlayMsg _overMsg( *overlayMsg ); + RelayMessage _relayMsg( *relayMsg ); + _overMsg.encapsulate(&_relayMsg); + + /// this must be handled by using relay link! + overlayInterface->routeMessage(relayMsg->getDestNode(), &_overMsg ); + return true; + } + + // error: I'm not a relay or destination! + logging_error("This node is neither relay nor destination. Dropping Message!"); + return true; + } + default: { + logging_error("RelayMessage Unknown!"); + return true; + } + } + + break; + } + + // --------------------------------------------------------------------- + // handle keep-alive messages + // --------------------------------------------------------------------- + case OverlayMsg::typeKeepAlive: { + if ( ld != NULL ) { + //logging_force("Keep-Alive for "<< ld->overlayId); + ld->markAlive(); + } + break; + } + + // --------------------------------------------------------------------- + // handle direct link replacement messages + // --------------------------------------------------------------------- + case OverlayMsg::typeDirectLink: { + LinkDescriptor* rld = getDescriptor( overlayMsg->getRelayLink() ); + rld->communicationId = ld->communicationId; + rld->relay = false; + rld->localRelay = NodeID::UNSPECIFIED; + rld->remoteRelay = NodeID::UNSPECIFIED; + eraseDescriptor(ld->overlayId); + 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() ); + return false; + } + + } /* switch */ return false; } -void BaseOverlay::broadcastMessage(Message* message, const ServiceID& service){ +// ---------------------------------------------------------------------------- + +void BaseOverlay::broadcastMessage(Message* message, const ServiceID& service) { logging_debug( "broadcasting message to all known nodes " << @@ -1027,9 +1388,6 @@ OverlayInterface::NodeList nodes = overlayInterface->getKnownNodes(); - OverlayInterface::NodeList::iterator i = nodes.begin(); - OverlayInterface::NodeList::iterator iend = nodes.end(); - - for( ; i != iend; i++ ){ + for(; i != nodes.end(); i++ ) { if( *i == nodeId) continue; // don't send to ourselfs sendMessage( message, *i, service ); @@ -1038,102 +1396,121 @@ vector BaseOverlay::getOverlayNeighbors() const { - // the known nodes _can_ also include our - // node, so we remove ourselfs - + // the known nodes _can_ also include our node, so we remove ourself vector nodes = overlayInterface->getKnownNodes(); vector::iterator i = find( nodes.begin(), nodes.end(), this->nodeId ); if( i != nodes.end() ) nodes.erase( i ); - return nodes; } -void BaseOverlay::updateOvlVis( const NodeID& n ) { - NodeID node = n; -/* void visShowNodeBubble ( - NETWORK_ID network, - NodeID& node, - string label - ); -*/ - using namespace std; - - if (node == nodeId || node.isUnspecified()) return; - - // min/max - if ( node < min || min.isUnspecified() ) min = node; - if ( node > max || max.isUnspecified() ) max = node; - - // successor - if ( succ.isUnspecified() || (node > nodeId && (succ < nodeId || (node-nodeId) < (succ-nodeId))) ) { - if (!succ.isUnspecified() && node != succ) - ovl.visDisconnect(ovlId, nodeId, succ, string("")); - succ = node; - ovl.visConnect(ovlId, nodeId, succ, string("")); - } - - // set successor (circle-wrap) - if (succ.isUnspecified() && !min.isUnspecified()) { - succ = min; - ovl.visConnect(ovlId, nodeId, succ, string("")); - } -} - const NodeID& BaseOverlay::getNodeID(const LinkID& lid) const { - if( lid == LinkID::UNSPECIFIED ) return nodeId; - - LinkMapping::const_iterator i = linkMapping.find( lid ); - if( i == linkMapping.end() ) return NodeID::UNSPECIFIED; - else return i->second.node; + const LinkDescriptor* ld = getDescriptor(lid); + if( ld == NULL ) return NodeID::UNSPECIFIED; + else return ld->remoteNode; } vector BaseOverlay::getLinkIDs( const NodeID& nid ) const { - vector linkvector; - - BOOST_FOREACH( LinkPair item, linkMapping ){ - if( item.second.node == nid || nid == NodeID::UNSPECIFIED ){ - linkvector.push_back( item.second.link ); - } - } - + BOOST_FOREACH( LinkDescriptor* ld, links ) { + if( ld->remoteNode == nid || nid == NodeID::UNSPECIFIED ) { + linkvector.push_back( ld->overlayId ); + } + } return linkvector; } -void BaseOverlay::incomingRouteMessage(Message* msg){ - // gets handled as normal data message - receiveMessage( msg, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED ); -} - -void BaseOverlay::onNodeJoin(const NodeID& node){ - + +void BaseOverlay::onNodeJoin(const NodeID& node) { JoiningNodes::iterator i = std::find( joiningNodes.begin(), joiningNodes.end(), node ); if( i == joiningNodes.end() ) return; logging_info( "node has successfully joined baseoverlay and overlay structure " - << node.toString() ); + << node.toString() ); joiningNodes.erase( i ); } -void BaseOverlay::eventFunction(){ - - list oldlinks; +void BaseOverlay::eventFunction() { + + // send keep-alive messages over established links + BOOST_FOREACH( LinkDescriptor* ld, links ) { + if (!ld->up) continue; + OverlayMsg overMsg( OverlayMsg::typeKeepAlive, + OverlayInterface::OVERLAY_SERVICE_ID, nodeId ); + sendMessage( &overMsg, ld ); + } + + // iterate over all links and check for time boundaries + vector oldlinks; time_t now = time(NULL); - - // first gather all the links from linkMapping that need droppin - // don't directly drop, as the dropLink function affects the - // linkMapping structure that we are traversing here. - // drop links after a timeout of 30s - - BOOST_FOREACH( LinkPair item, linkMapping ){ - if( item.second.autolink && difftime(now, item.second.lastuse) > 30) - oldlinks.push_back( item.first ); - } - - BOOST_FOREACH( const LinkID lnk, oldlinks ) { - logging_debug( "auto-link " << lnk.toString() << " timed out and is getting dropped" ); - dropLink( lnk ); - } + BOOST_FOREACH( LinkDescriptor* ld, links ) { + // remote used as relay flag + if ( ld->usedAsRelay && difftime( now, ld->timeUsedAsRelay ) > 10) + ld->usedAsRelay = false; + + // keep alives missed? yes-> + if ( !ld->up && difftime( now, ld->keepAliveTime ) > 2 ) { + + // increase counter + ld->keepAliveMissed++; + + // missed more than four keep-alive messages (4 sec)? -> drop link + if (ld->keepAliveMissed > 10) { + logging_force( "Link connection request is stale, closing: " << ld ); + oldlinks.push_back( ld ); + } + } + + if (!ld->up) continue; + + // drop links that are dropped and not used as relay + if (ld->dropWhenRelaysLeft && !ld->usedAsRelay && !ld->autolink) + oldlinks.push_back( ld ); + else + + // auto-link time exceeded? + if ( ld->autolink && difftime( now, ld->lastuse ) > 30 ) + oldlinks.push_back( ld ); + + else + + // keep alives missed? yes-> + if ( !ld->autolink && difftime( now, ld->keepAliveTime ) > 2 ) { + + // increase counter + ld->keepAliveMissed++; + + // missed more than four keep-alive messages (4 sec)? -> drop link + if (ld->keepAliveMissed >= 8) { + logging_force( "Link is stale, closing: " << ld ); + oldlinks.push_back( ld ); + } + } + } + + // show link state + counter++; + if (counter>=4) showLinkState(); + if (counter>=4 || counter<0) counter = 0; + + // drop links + BOOST_FOREACH( const LinkDescriptor* ld, oldlinks ) { + if (!ld->communicationId.isUnspecified() && ld->communicationId == initiatorLink) { + logging_force( "Not dropping initiator link: " << ld ); + continue; + } + logging_force( "Link timed out. Dropping " << ld ); + dropLink( ld->overlayId ); + } +} + +void BaseOverlay::showLinkState() { + int i=0; + logging_force("--- link state -------------------------------"); + BOOST_FOREACH( LinkDescriptor* ld, links ) { + logging_force("link " << i << ": " << ld); + i++; + } + logging_force("----------------------------------------------"); } Index: source/ariba/overlay/BaseOverlay.h =================================================================== --- source/ariba/overlay/BaseOverlay.h (revision 5002) +++ source/ariba/overlay/BaseOverlay.h (revision 5151) @@ -86,4 +86,5 @@ // ariba interface using ariba::NodeListener; +using ariba::SideportListener; using ariba::CommunicationListener; @@ -117,10 +118,14 @@ namespace overlay { -class BaseOverlay: public MessageReceiver, public CommunicationEvents, - public OverlayStructureEvents, protected Timer { - -private: +class LinkDescriptor; + +class BaseOverlay: public MessageReceiver, + public CommunicationEvents, + public OverlayStructureEvents, + protected Timer { + friend class OneHop; friend class Chord; + friend class ariba::SideportListener; use_logging_h( BaseOverlay ); @@ -157,14 +162,20 @@ */ const LinkID establishLink(const NodeID& node, const ServiceID& service, - const LinkID& linkid = LinkID::UNSPECIFIED); + const LinkID& linkid = LinkID::UNSPECIFIED); /** * Starts a link establishment procedure to the specified - * + * endpoint and to the specified service. Concurrently it tries to + * establish a relay link over the overlay using the nodeid + */ + const LinkID establishLink(const EndpointDescriptor& ep, const NodeID& nodeid, + const ServiceID& service, const LinkID& linkid = LinkID::UNSPECIFIED); + + /** + * Starts a link establishment procedure to the specified * endpoint and to the specified service */ const LinkID establishLink(const EndpointDescriptor& ep, - const ServiceID& service, const LinkID& linkid = - LinkID::UNSPECIFIED); + const ServiceID& service, const LinkID& linkid = LinkID::UNSPECIFIED); /// drops a link @@ -176,5 +187,5 @@ /// sends a message to a node and a specific service seqnum_t sendMessage(const Message* message, const NodeID& node, - const ServiceID& service); + const ServiceID& service); /** @@ -251,5 +262,5 @@ * @param boot A bootstrap node */ - void joinSpoVNet(const SpoVNetID& id, const EndpointDescriptor& boot ); + void joinSpoVNet(const SpoVNetID& id, const EndpointDescriptor& boot); /** @@ -275,5 +286,5 @@ */ virtual void onLinkUp(const LinkID& id, const NetworkLocator* local, - const NetworkLocator* remote); + const NetworkLocator* remote); /** @@ -281,5 +292,5 @@ */ virtual void onLinkDown(const LinkID& id, const NetworkLocator* local, - const NetworkLocator* remote); + const NetworkLocator* remote); /** @@ -287,6 +298,6 @@ */ virtual void onLinkChanged(const LinkID& id, - const NetworkLocator* oldlocal, const NetworkLocator* newlocal, - const NetworkLocator* oldremote, const NetworkLocator* newremote); + const NetworkLocator* oldlocal, const NetworkLocator* newlocal, + const NetworkLocator* oldremote, const NetworkLocator* newremote); /** @@ -294,5 +305,5 @@ */ virtual void onLinkFail(const LinkID& id, const NetworkLocator* local, - const NetworkLocator* remote); + const NetworkLocator* remote); /** @@ -300,6 +311,6 @@ */ virtual void onLinkQoSChanged(const LinkID& id, - const NetworkLocator* local, const NetworkLocator* remote, - const QoSParameterSet& qos); + const NetworkLocator* local, const NetworkLocator* remote, + const QoSParameterSet& qos); /** @@ -307,19 +318,22 @@ */ virtual bool onLinkRequest(const LinkID& id, const NetworkLocator* local, - const NetworkLocator* remote); + const NetworkLocator* remote); + + /** + * Processes a received message from BaseCommunication + * + * In case of a message routed by the overlay the source identifies + * the node the message came from! + */ + virtual bool receiveMessage(const Message* message, const LinkID& link, + const NodeID& source = NodeID::UNSPECIFIED); //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - /** - * Processes a received message. - * - * Beware: nodeid is not valid in this case! (since this class implements - * nodeid's in the first place *g*) - */ - virtual bool receiveMessage( - const Message* message, const LinkID& link, const NodeID&); + /// handles an incoming message with link descriptor + bool handleMessage(const Message* message, + const LinkID& boLink, const LinkID& bcLink, const NodeID& remoteNode ); //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - /** * This method is called, when a routed message arrives from the @@ -328,5 +342,7 @@ * @see OverlayStructureEvents.h */ - virtual void incomingRouteMessage(Message* msg); + virtual void incomingRouteMessage(Message* msg, + const LinkID& link = LinkID::UNSPECIFIED, + const NodeID& source = NodeID::UNSPECIFIED); /** @@ -343,56 +359,6 @@ private: - /** - * The BaseCommunication the BaseOverlay - * communicates over - */ - BaseCommunication* bc; - - /** - * The nodeid of this BaseOverlay instance. - */ - NodeID nodeId; - - /** - * The SpoVNetID that we are joined to - * or that we have created. - */ - SpoVNetID spovnetId; - - /** - * TODO - */ - Demultiplexer communicationListeners; - - /** - * TODO - */ - typedef vector NodeListenerVector; - - /** - * TODO - */ - NodeListenerVector nodeListeners; - - /** - * TODO - */ - SideportListener* sideport; - - /** - * The abstract overlay interface that implements - * the overlay specific functionality. - */ - OverlayInterface* overlayInterface; - - /** - * The special link to the Initiator of the SpoVNet - * or LinkID::UNDEFINED if we are the Initiator - */ - LinkID initiatorLink; - - /** - * The state of the BaseOverlay - */ + + /// The state of the BaseOverlay typedef enum _BaseOverlayState { BaseOverlayStateInvalid = 0, @@ -402,92 +368,54 @@ } BaseOverlayState; - /** - * TODO - */ - BaseOverlayState state; - - /** - * The initiator node - */ - NodeID spovnetInitiator; - - /** - * OvlVis - */ - NodeID min, max; - NodeID succ, pred; - void updateOvlVis(const NodeID& node); - - /** - * Link management - */ - class LinkItem { - public: - static const LinkItem UNSPECIFIED; - - LinkItem() : - link(LinkID::UNSPECIFIED), - node(NodeID::UNSPECIFIED), - service(ServiceID::UNSPECIFIED), - interface(&CommunicationListener::DEFAULT), - autolink(false), - lastuse(0), - linkup(false){ - } - - LinkItem(const LinkID& _link, const NodeID& _node, - const ServiceID& _service, CommunicationListener* _interface) : - link(_link), - node(_node), - service(_service), - interface(_interface), - autolink(false), - lastuse(time(NULL)), - linkup(false){ - - assert( _interface != NULL ); - } - - LinkItem(const LinkItem& rh) : - link(rh.link), - node(rh.node), - service(rh.service), - interface(rh.interface), - autolink(rh.autolink), - lastuse(rh.lastuse), - linkup(rh.linkup){ - - BOOST_FOREACH( Message* msg, rh.waitingmsg ){ - waitingmsg.push_back( msg ); - } - } - - void deleteWaiting(){ - BOOST_FOREACH( Message* msg, waitingmsg ){ - delete msg; - } - waitingmsg.clear(); - } - - // general information about the link - const LinkID link; - NodeID node; - ServiceID service; - CommunicationListener* interface; - bool linkup; - - // information needed for auto links - void markused() { - lastuse = time(NULL); - } - - bool autolink; - time_t lastuse; - deque waitingmsg; - }; - - typedef map LinkMapping; - typedef pair LinkPair; - LinkMapping linkMapping; + BaseOverlayState state; ///< Current Base-Overlay state + BaseCommunication* bc; ///< reference to the base communication + NodeID nodeId; ///< the node id of this node + SpoVNetID spovnetId; ///< the spovnet id of the currently joined overlay + LinkID initiatorLink; ///< the link id of the link to the initiator node + NodeID spovnetInitiator;///< The initiator node + + /// the service id communication listeners + Demultiplexer communicationListeners; + + /// the node listeners + typedef vector NodeListenerVector; + NodeListenerVector nodeListeners; + + /// the sideport listener + SideportListener* sideport; + + /// the used overlay structure + OverlayInterface* overlayInterface; + + /// The link mapping of the node + vector links; + void eraseDescriptor(const LinkID& link, bool communication = false); + + /// returns a link descriptor for the given id + LinkDescriptor* getDescriptor(const LinkID& link, + bool communication = false); + + /// returns a link descriptor for the given id + const LinkDescriptor* getDescriptor(const LinkID& link, + bool communication = false) const; + + /// returns a auto-link descriptor for the given node and service id + LinkDescriptor* getAutoDescriptor(const NodeID& node, const ServiceID& service); + + /// adds a new link descriptor or uses an existing one + LinkDescriptor* addDescriptor(const LinkID& link = LinkID::UNSPECIFIED); + + /// returns a direct link relay descriptor to the given relay node + LinkDescriptor* getRelayDescriptor( const NodeID& relayNode ); + + /// find a proper relay node that is directly connected to this node + const NodeID findRelayNode( const NodeID& id ); + + /// forwards a message over relays/overlay/directly using link descriptor + seqnum_t sendMessage( Message* message, const LinkDescriptor* ld ); + + /// creates a link descriptor, applys relay semantics if possible + LinkDescriptor* createLinkDescriptor( + const NodeID& remoteNode, const ServiceID& service, const LinkID& link_id ); // map of a link request map a nonce to a LinkID @@ -495,4 +423,6 @@ PendingLinkMap pendingLinks; + void showLinkState(); + /** * nodes with pending joines. TODO: should be cleaned every @@ -502,4 +432,6 @@ JoiningNodes joiningNodes; + int counter; + /** * Bootstrapper for our spovnet Index: source/ariba/overlay/LinkDescriptor.cpp =================================================================== --- source/ariba/overlay/LinkDescriptor.cpp (revision 5151) +++ source/ariba/overlay/LinkDescriptor.cpp (revision 5151) @@ -0,0 +1,14 @@ +#include "LinkDescriptor.h" + +namespace ariba { +namespace overlay { + +std::ostream& operator<<(std::ostream& s, const LinkDescriptor* ld ) { + return s << ld->to_string(); +} + +std::ostream& operator<<(std::ostream& s, const LinkDescriptor& ld ) { + return s << ld.to_string(); +} + +}} // ariba, overlay Index: source/ariba/overlay/LinkDescriptor.h =================================================================== --- source/ariba/overlay/LinkDescriptor.h (revision 5151) +++ source/ariba/overlay/LinkDescriptor.h (revision 5151) @@ -0,0 +1,146 @@ +#ifndef __LINK_DESCRIPTOR_H +#define __LINK_DESCRIPTOR_H + +#include +#include +#include +#include +#include + +#include "ariba/utility/messages.h" +#include "ariba/utility/types.h" +#include "ariba/CommunicationListener.h" + +namespace ariba { + class CommunicationListener; +} + +using std::deque; +using ariba::utility::Message; +using ariba::utility::NodeID; +using ariba::utility::SpoVNetID; +using ariba::utility::ServiceID; +using ariba::utility::LinkID; +using ariba::CommunicationListener; + +namespace ariba { +namespace overlay { + +class LinkDescriptor; + +std::ostream& operator<<(std::ostream& s, const LinkDescriptor* ld ); +std::ostream& operator<<(std::ostream& s, const LinkDescriptor& ld ); + +/// LinkDescriptor +class LinkDescriptor { +public: + // ctor + LinkDescriptor() { + // default values + this->up = false; + this->dropWhenRelaysLeft = false; + this->fromRemote = false; + this->remoteNode = NodeID::UNSPECIFIED; + this->overlayId = LinkID::UNSPECIFIED; + this->communicationUp = false; + this->communicationId = LinkID::UNSPECIFIED; + this->keepAliveTime = time(NULL); + this->keepAliveMissed = 0; + this->usedAsRelay = false; + this->timeUsedAsRelay = time(NULL); + this->service = ServiceID::UNSPECIFIED; + this->listener = &CommunicationListener::DEFAULT; + this->relay = false; + this->localRelay = NodeID::UNSPECIFIED; + this->remoteRelay = NodeID::UNSPECIFIED; + this->remoteLinkId = LinkID::UNSPECIFIED; + this->autolink = false; + this->lastuse = time(NULL); + } + + // dtor + ~LinkDescriptor() { + flushQueue(); + } + + // general information about the link --------------------------------- + + bool up; ///< flag wheter this link is up and running + bool dropWhenRelaysLeft; + bool fromRemote; /// messageQueue; ///< waiting messages to be delivered + + /// updates the timestamp + void markAsUsed() { + lastuse = time(NULL); + } + + /// drops waiting messsages + void flushQueue() { + BOOST_FOREACH( Message* msg, messageQueue ) + delete msg; + messageQueue.clear(); + } + + std::string to_string() const { + std::ostringstream s; + s << "up=" << up << " "; + s << "fromRem=" << fromRemote << " "; + s << "remNode=" << remoteNode.toString().substr(0,6) << " "; + s << "serv=" << service.toString() << " "; + s << "overId=" << overlayId.toString().substr(0,6) << " "; + s << "commId=" << communicationId.toString().substr(0,6) << " "; + s << "usedAsRel=" << usedAsRelay << " "; + s << "KeepAliveMiss=" << keepAliveMissed << " "; + if ( !localRelay.isUnspecified() ) { + s << "locRel=" << localRelay.toString().substr(0,6) << " "; + s << "remRel=" << remoteRelay.toString().substr(0,6) << " "; + s << "remLink=" << remoteLinkId.toString().substr(0,6) << " "; + } + s << "auto=" << autolink; + return s.str(); + } +}; + +}} // namespace ariba, overlay + +#endif // __LINK_DESCRIPTOR_H + Index: source/ariba/overlay/messages/JoinReply.h =================================================================== --- source/ariba/overlay/messages/JoinReply.h (revision 5002) +++ source/ariba/overlay/messages/JoinReply.h (revision 5151) @@ -85,5 +85,7 @@ sznBeginDefault( ariba::overlay::JoinReply, X ) { - X && &spovnetid && param && bootstrapEp && joinAllowed && cI(0,7); + uint8_t ja = joinAllowed; + X && &spovnetid && param && &bootstrapEp && ja; + if (X.isDeserializer()) joinAllowed = ja; } sznEnd(); Index: source/ariba/overlay/messages/LinkRequest.cpp =================================================================== --- source/ariba/overlay/messages/LinkRequest.cpp (revision 5002) +++ source/ariba/overlay/messages/LinkRequest.cpp (revision 5151) @@ -7,12 +7,4 @@ vsznDefault(LinkRequest); -LinkRequest::LinkRequest() { - -} - -LinkRequest::LinkRequest( uint32_t nonce, const EndpointDescriptor* endpoint, bool reply ) : - flags(reply&1), nonce(nonce), endpoint(endpoint) { -} - LinkRequest::~LinkRequest() { } Index: source/ariba/overlay/messages/LinkRequest.h =================================================================== --- source/ariba/overlay/messages/LinkRequest.h (revision 5002) +++ source/ariba/overlay/messages/LinkRequest.h (revision 5151) @@ -27,10 +27,17 @@ uint32_t nonce; const EndpointDescriptor* endpoint; + LinkID remoteLinkId; + NodeID relay; public: - LinkRequest(); + LinkRequest() { + + } LinkRequest( uint32_t nonce, const EndpointDescriptor* endpoint, - bool reply = false ); + bool reply = false, const LinkID& remoteLinkId = LinkID::UNSPECIFIED, + const NodeID& relay = NodeID::UNSPECIFIED ) : + flags(reply&1), nonce(nonce), endpoint(endpoint), remoteLinkId(remoteLinkId), relay(relay) { + } virtual ~LinkRequest(); @@ -38,4 +45,12 @@ const EndpointDescriptor* getEndpoint() const { return endpoint; + } + + const LinkID& getRemoteLinkId() const { + return remoteLinkId; + } + + const NodeID& getRelay() const { + return relay; } @@ -53,5 +68,7 @@ sznBeginDefault( ariba::overlay::LinkRequest, X ) { if (X.isDeserializer()) endpoint = new EndpointDescriptor(); - X && flags && nonce && reinterpret_cast(const_cast(endpoint)); + X && flags && nonce; + X && const_cast(endpoint); + X && &relay && &remoteLinkId; } sznEnd(); Index: source/ariba/overlay/messages/OverlayMsg.cpp =================================================================== --- source/ariba/overlay/messages/OverlayMsg.cpp (revision 5002) +++ source/ariba/overlay/messages/OverlayMsg.cpp (revision 5151) @@ -44,30 +44,6 @@ vsznDefault(OverlayMsg); -OverlayMsg::OverlayMsg(OverlayMessageType _type, const ServiceID _service, const NodeID _sourceNode) - : type( (uint8_t)_type), service( _service ), sourceNode( _sourceNode ) { -} - -OverlayMsg::OverlayMsg(OverlayMessageType _type, const NodeID _sourceNode) - : type( (uint8_t)_type), service( ServiceID::UNSPECIFIED ), sourceNode( _sourceNode ){ -} - OverlayMsg::~OverlayMsg(){ } -OverlayMsg::OverlayMessageType OverlayMsg::getType(){ - return (OverlayMessageType)type; -} - -const ServiceID& OverlayMsg::getService(){ - return service; -} - -bool OverlayMsg::isType(OverlayMessageType _type){ - return (OverlayMessageType)type == _type; -} - -const NodeID& OverlayMsg::getSourceNode(){ - return sourceNode; -} - }} // ariba::overlay Index: source/ariba/overlay/messages/OverlayMsg.h =================================================================== --- source/ariba/overlay/messages/OverlayMsg.h (revision 5002) +++ source/ariba/overlay/messages/OverlayMsg.h (revision 5151) @@ -55,49 +55,93 @@ namespace overlay { -using_serialization; +using_serialization +; -class OverlayMsg : public Message { - VSERIALIZEABLE; +class OverlayMsg: public Message { +VSERIALIZEABLE + ; public: - typedef enum _OverlayMessageType { - OverlayMessageTypeInvalid = 0, // invalid type (no encapsulated messages) - OverlayMessageTypeData = 1, // message contains data for higher layers - OverlayMessageTypeJoinRequest = 2, // spovnet join request - OverlayMessageTypeJoinReply = 3, // spovnet join reply - OverlayMessageTypeUpdate = 4, // update message for link association - OverlayMessageTypeBye = 5, // spovnet leave (no encapsulated messages) - OverlayMessageTypeLinkRequest = 6, // link request (sent over the overlay) - } OverlayMessageType; + /// (payload-) message types + enum type_ { + typeInvalid = 0, ///< invalid type (no encapsulated messages) + typeData = 1, ///< message contains data for higher layers + typeJoinRequest = 2, ///< join request + typeJoinReply = 3, ///< join reply + typeUpdate = 4, ///< update message for link association + typeBye = 5, ///< leave (no encapsulated messages) + typeLinkRequest = 6, ///< link request (sent over the overlay) + typeRelay = 7, ///< relay message + typeKeepAlive = 8, ///< a keep-alive message + typeDirectLink = 9, + ///< a direct connection has been established + }; - OverlayMsg( - OverlayMessageType _type = OverlayMessageTypeInvalid, - const ServiceID _service = ServiceID::UNSPECIFIED, - const NodeID _sourceNode = NodeID::UNSPECIFIED - ); + /// default constructor + OverlayMsg(type_ type = typeInvalid, const ServiceID _service = + ServiceID::UNSPECIFIED, const NodeID _sourceNode = + NodeID::UNSPECIFIED) : + type((uint8_t) type), service(_service), sourceNode(_sourceNode), + relayLink(LinkID::UNSPECIFIED), autoLink(false) { + } - OverlayMsg( - OverlayMessageType _type, - const NodeID _sourceNode - ); + OverlayMsg(const OverlayMsg& rhs) : + type(rhs.type), service(rhs.service), sourceNode(rhs.sourceNode), + relayLink(rhs.relayLink), autoLink(rhs.autoLink) { + } - virtual ~OverlayMsg(); + /// type and source node constructor + OverlayMsg(type_ type, const NodeID _sourceNode) : + type((uint8_t) type), service(ServiceID::UNSPECIFIED), sourceNode( + _sourceNode), relayLink(LinkID::UNSPECIFIED), autoLink(false) { + } - bool isType(OverlayMessageType _type); - OverlayMessageType getType(); - const ServiceID& getService(); - const NodeID& getSourceNode(); + /// destructor + ~OverlayMsg(); + type_ getType() const { + return (type_) type; + } + + const ServiceID& getService() const { + return service; + } + + const NodeID& getSourceNode() const { + return sourceNode; + } + + const LinkID& getRelayLink() const { + return relayLink; + } + + void setRelayLink(const LinkID& relayLink) { + this->relayLink = relayLink; + } + + const bool isAutoLink() const { + return autoLink; + } + + void setAutoLink(bool autoLink) { + this->autoLink = autoLink; + } private: uint8_t type; ServiceID service; NodeID sourceNode; + LinkID relayLink; + uint8_t autoLink; }; -}} // ariba::overlay +} +} // ariba::overlay -sznBeginDefault( ariba::overlay::OverlayMsg, X ) { - X && type && &service && &sourceNode && Payload(); -} sznEnd(); +sznBeginDefault( ariba::overlay::OverlayMsg, X ){ +X && type && &service && &sourceNode; +if (type == typeDirectLink) X && &relayLink; +if (type == typeUpdate) X && autoLink; +X && Payload(); +}sznEnd(); #endif // OVERLAY_MSG_H__ Index: source/ariba/overlay/messages/RelayMessage.cpp =================================================================== --- source/ariba/overlay/messages/RelayMessage.cpp (revision 5151) +++ source/ariba/overlay/messages/RelayMessage.cpp (revision 5151) @@ -0,0 +1,12 @@ +#include "RelayMessage.h" + +namespace ariba { +namespace overlay { + +vsznDefault(RelayMessage); + +RelayMessage::~RelayMessage() { +} + +}} // ariba::overlay + Index: source/ariba/overlay/messages/RelayMessage.h =================================================================== --- source/ariba/overlay/messages/RelayMessage.h (revision 5151) +++ source/ariba/overlay/messages/RelayMessage.h (revision 5151) @@ -0,0 +1,79 @@ +#ifndef RELAYMESSAGE_H_ +#define RELAYMESSAGE_H_ + +#include "ariba/utility/messages.h" +#include "ariba/utility/serialization.h" +#include "ariba/communication/EndpointDescriptor.h" + +using ariba::communication::EndpointDescriptor; + +namespace ariba { +namespace overlay { + +using_serialization; + +using ariba::utility::Message; + +/** + * This message is sent to another overlay node to request a new link. + * + * @author Sebastian Mies + */ +class RelayMessage : public Message { + VSERIALIZEABLE; +private: + uint8_t type; + NodeID relayNode; + NodeID destNode; + LinkID destLink; + +public: + enum type_ { + typeInvalid = 0, + typeInform = 1, + typeRoute = 2 + }; + + /// contructs a unspecified relay message + RelayMessage() : + type(typeInvalid), relayNode(NodeID::UNSPECIFIED), destNode(NodeID::UNSPECIFIED) { + } + + RelayMessage( type_ type, const NodeID& relayNode, const NodeID& destNode, const LinkID& destLink = LinkID::UNSPECIFIED ) : + type(type), relayNode(relayNode), destNode(destNode), destLink(destLink) { + } + + ~RelayMessage(); + + /// returns the type of this message + type_ getType() const { + return (type_)type; + } + + /// sets the type of this message + void setType( type_ type ) { + this->type = type; + } + + /// returns the remote (destination) node id + const NodeID& getDestNode() const { + return destNode; + } + + const LinkID& getDestLink() const { + return destLink; + } + + /// returns the relay node for the destination + const NodeID& getRelayNode() const { + return relayNode; + } +}; + +}} // ariba::overlay + +sznBeginDefault( ariba::overlay::RelayMessage, X ) { + X && type && &relayNode && &destNode && &destLink && Payload(); +} sznEnd(); + +#endif /* RELAYMESSAGE_H_ */ Index: source/ariba/overlay/modules/OverlayInterface.h =================================================================== --- source/ariba/overlay/modules/OverlayInterface.h (revision 5002) +++ source/ariba/overlay/modules/OverlayInterface.h (revision 5151) @@ -128,4 +128,16 @@ /** + * Routes a message to a given node by using an existing link. + * + * TODO: This is a hack. This method allows the BaseOverlay class to + * use overlay signaling links to transfer data for relaying + * + * @param node The destination node. + * @param link An established link + * @param msg The message to be sent. + */ + virtual void routeMessage(const NodeID& node, const LinkID& link, Message* msg) = 0; + + /** * Returns the nodes known to this overlay. * @@ -139,19 +151,33 @@ virtual NodeList getKnownNodes() const = 0; + /** + * Returns the link id of the next hop a route message would take. + * + * @param id The destination node id + * @return The link id of the next hop + */ + virtual const LinkID& getNextLinkId( const NodeID& id ) const = 0; + //--- functions from CommunicationListener that we _can_ use as overlay --- /// @see CommunicationListener virtual void onLinkUp(const LinkID& lnk, const NodeID& remote); + /// @see CommunicationListener virtual void onLinkDown(const LinkID& lnk, const NodeID& remote); + /// @see CommunicationListener virtual void onLinkChanged(const LinkID& lnk, const NodeID& remote); + /// @see CommunicationListener virtual void onLinkFail(const LinkID& lnk, const NodeID& remote); + /// @see CommunicationListener virtual void onLinkQoSChanged(const LinkID& lnk, const NodeID& remote, const LinkProperties& prop); + /// @see CommunicationListener virtual bool onLinkRequest(const NodeID& remote, const DataMessage& msg); + /// @see CommunicationListener virtual void onMessage(const DataMessage& msg, const NodeID& remote, @@ -162,5 +188,4 @@ protected: - /// Reference to an active base overlay BaseOverlay& baseoverlay; Index: source/ariba/overlay/modules/OverlayStructureEvents.cpp =================================================================== --- source/ariba/overlay/modules/OverlayStructureEvents.cpp (revision 5002) +++ source/ariba/overlay/modules/OverlayStructureEvents.cpp (revision 5151) @@ -47,5 +47,5 @@ } -void OverlayStructureEvents::incomingRouteMessage(Message* msg){ +void OverlayStructureEvents::incomingRouteMessage(Message* msg, const LinkID& link, const NodeID& source ){ } Index: source/ariba/overlay/modules/OverlayStructureEvents.h =================================================================== --- source/ariba/overlay/modules/OverlayStructureEvents.h (revision 5002) +++ source/ariba/overlay/modules/OverlayStructureEvents.h (revision 5151) @@ -61,5 +61,5 @@ protected: - virtual void incomingRouteMessage( Message* msg ); + virtual void incomingRouteMessage( Message* msg, const LinkID& link = LinkID::UNSPECIFIED, const NodeID& source = NodeID::UNSPECIFIED ); virtual void onNodeJoin( const NodeID& node ); }; Index: source/ariba/overlay/modules/chord/Chord.cpp =================================================================== --- source/ariba/overlay/modules/chord/Chord.cpp (revision 5002) +++ source/ariba/overlay/modules/chord/Chord.cpp (revision 5151) @@ -71,9 +71,14 @@ /// helper: sets up a link using the base overlay -LinkID Chord::setup(const EndpointDescriptor& endp) { +LinkID Chord::setup(const EndpointDescriptor& endp, const NodeID& node) { logging_debug("request to setup link to " << endp.toString() ); + + for (size_t i=0; isetup first link - if (!(boot == EndpointDescriptor::UNSPECIFIED)) bootstrapLink = setup(boot); + if (!(boot == EndpointDescriptor::UNSPECIFIED)) + bootstrapLink = setup(boot); // timer for stabilization management @@ -138,5 +144,7 @@ // message for this node? yes-> delegate to base overlay - if (item->id == nodeid) baseoverlay.incomingRouteMessage(msg); + if (item->id == nodeid || destnode == nodeid) + baseoverlay.incomingRouteMessage( msg, LinkID::UNSPECIFIED, nodeid ); + else { // no-> send to next hop ChordMessage cmsg(ChordMessage::route, nodeid, destnode); @@ -144,4 +152,26 @@ send(&cmsg, item->info); } +} + +/// @see OverlayInterface.h +void Chord::routeMessage(const NodeID& node, const LinkID& link, Message* msg) { + logging_debug("Redirect over Chord to node id=" << node.toString() + << " link id=" << link.toString() ); + ChordMessage cmsg(ChordMessage::route, nodeid, node); + cmsg.encapsulate(msg); + send(&cmsg, link); +} + +/// @see OverlayInterface.h +const LinkID& Chord::getNextLinkId( const NodeID& id ) const { + // get next hop + const route_item* item = table->get_next_hop(id); + + // returns a unspecified id when this is itself + if (item == NULL || item->id == nodeid) + return LinkID::UNSPECIFIED; + + /// return routing info + return item->info; } @@ -160,4 +190,9 @@ logging_debug("link_up: link=" << lnk.toString() << " remote=" << remote.toString() ); + for (vector::iterator i=pending.begin(); i!=pending.end(); i++) + if (*i == remote) { + pending.erase(i); + break; + } route_item* item = table->insert(remote); @@ -207,5 +242,5 @@ break; - // route message with payload + // route message with payload case M::route: { // find next hop @@ -214,10 +249,11 @@ // next hop == myself? if (m->getDestination() == nodeid) { // yes-> route to base overlay - logging_debug("send message to baseoverlay"); - baseoverlay.incomingRouteMessage(m); + logging_debug("Send message to baseoverlay"); + baseoverlay.incomingRouteMessage( m, item->info, remote ); } // no-> route to next hop else { - logging_debug("route chord message to " << item->id.toString() ); + logging_debug("Route chord message to " + << item->id.toString() << " (destination=" << m->getDestination() << ")"); send(m, item->info); } @@ -236,6 +272,6 @@ // check if source node can be added to routing table and setup link - if (m->getSource() != nodeid && table->is_insertable(m->getSource())) setup( - *dmsg->getSourceEndpoint()); + if (m->getSource() != nodeid && table->is_insertable(m->getSource())) + setup(*dmsg->getSourceEndpoint(), m->getSource() ); // delegate discovery message @@ -302,5 +338,5 @@ } if (item == NULL) break; - logging_debug("routing discovery message to succ/pred " + logging_debug("routing discovery message to succ/pred " << item->id.toString() ); ChordMessage cmsg(*m); @@ -327,8 +363,7 @@ void Chord::eventFunction() { - if (!LinkID::UNSPECIFIED.isUnspecified()) - logging_error("LinkID::UNSPECIFIED not unspecified!!!!"); stabilize_counter++; if (stabilize_counter == 3) { + pending.clear(); size_t numNeighbors = 0; for (size_t i = 0; i < table->size(); i++) { @@ -368,4 +403,3 @@ } -} -} // namespace ariba, overlay +}} // namespace ariba, overlay Index: source/ariba/overlay/modules/chord/Chord.h =================================================================== --- source/ariba/overlay/modules/chord/Chord.h (revision 5002) +++ source/ariba/overlay/modules/chord/Chord.h (revision 5151) @@ -77,7 +77,8 @@ int stabilize_finger; LinkID bootstrapLink; + vector pending; // helper: sets up a link using the "base overlay" - LinkID setup( const EndpointDescriptor& endp ); + LinkID setup( const EndpointDescriptor& endp, const NodeID& node = NodeID::UNSPECIFIED ); // helper: sends a message using the "base overlay" @@ -91,4 +92,7 @@ OverlayStructureEvents* _eventsReceiver, const OverlayParameterSet& param); virtual ~Chord(); + + /// @see OverlayInterface.h + virtual const LinkID& getNextLinkId( const NodeID& id ) const; /// @see OverlayInterface.h @@ -113,4 +117,7 @@ /// @see OverlayInterface.h + virtual void routeMessage(const NodeID& node, const LinkID& link, Message* msg); + + /// @see OverlayInterface.h virtual NodeList getKnownNodes() const; Index: source/ariba/overlay/modules/onehop/OneHop.cpp =================================================================== --- source/ariba/overlay/modules/onehop/OneHop.cpp (revision 5002) +++ source/ariba/overlay/modules/onehop/OneHop.cpp (revision 5151) @@ -99,4 +99,17 @@ } +void OneHop::routeMessage(const NodeID& node, const LinkID& link, Message* msg) { + OneHopMessage onehopRoute( OneHopMessage::OneHopMessageTypeRoute ); + onehopRoute.encapsulate(msg); + baseoverlay.sendMessage( &onehopRoute, link ); +} + +/// @see OverlayInterface.h +const LinkID& OneHop::getNextLinkId( const NodeID& id ) const { + OverlayNodeMapping::const_iterator i = overlayNodes.find( id ); + if (i == overlayNodes.end()) return LinkID::UNSPECIFIED; + return i->second; +} + void OneHop::createOverlay() { // don't need to bootstrap against ourselfs. Index: source/ariba/overlay/modules/onehop/OneHop.h =================================================================== --- source/ariba/overlay/modules/onehop/OneHop.h (revision 5002) +++ source/ariba/overlay/modules/onehop/OneHop.h (revision 5151) @@ -81,5 +81,11 @@ /// @see OverlayInterface.h + virtual const LinkID& getNextLinkId( const NodeID& id ) const; + + /// @see OverlayInterface.h virtual void routeMessage(const NodeID& destnode, Message* msg); + + /// @see OverlayInterface.h + virtual void routeMessage(const NodeID& node, const LinkID& link, Message* msg); /// @see OverlayInterface.h Index: source/ariba/utility/logging/Logging.h =================================================================== --- source/ariba/utility/logging/Logging.h (revision 5002) +++ source/ariba/utility/logging/Logging.h (revision 5151) @@ -69,10 +69,23 @@ log4cxx::LoggerPtr x::logger(log4cxx::Logger::getLogger(#x)); -#define logging_trace(x) { LOG4CXX_TRACE(logger,x); } -#define logging_debug(x) {colorDebug; LOG4CXX_DEBUG(logger,x); colorDefault; } -#define logging_info(x) {colorInfo; LOG4CXX_INFO(logger,x); colorDefault; } -#define logging_warn(x) {colorWarn; LOG4CXX_WARN(logger,x); colorDefault; } -#define logging_error(x) {colorError; LOG4CXX_ERROR(logger,x); colorDefault; } -#define logging_fatal(x) {colorError; LOG4CXX_FATAL(logger,x); colorDefault; exit(-1); } +#ifdef HAVE_MAEMO + + #define logging_trace(x) { } + #define logging_debug(x) { } + #define logging_info(x) { } + #define logging_warn(x) {colorWarn; LOG4CXX_WARN(logger,x); colorDefault; } + #define logging_error(x) {colorError; LOG4CXX_ERROR(logger,x); colorDefault; } + #define logging_fatal(x) {colorError; LOG4CXX_FATAL(logger,x); colorDefault; exit(-1); } + +#else + + #define logging_trace(x) { LOG4CXX_TRACE(logger,x); } + #define logging_debug(x) {colorDebug; LOG4CXX_DEBUG(logger,x); colorDefault; } + #define logging_info(x) {colorInfo; LOG4CXX_INFO(logger,x); colorDefault; } + #define logging_warn(x) {colorWarn; LOG4CXX_WARN(logger,x); colorDefault; } + #define logging_error(x) {colorError; LOG4CXX_ERROR(logger,x); colorDefault; } + #define logging_fatal(x) {colorError; LOG4CXX_FATAL(logger,x); colorDefault; exit(-1); } + +#endif // HAVE_MAEMO #endif //LOGGING_H__ Index: source/ariba/utility/system/StartupWrapper.cpp =================================================================== --- source/ariba/utility/system/StartupWrapper.cpp (revision 5002) +++ source/ariba/utility/system/StartupWrapper.cpp (revision 5151) @@ -103,5 +103,5 @@ // set up again an individual level if you like { -// log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("BaseOverlay")); +// log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("PingPong")); // logger->setLevel(log4cxx::Level::getDebug()); }