Index: source/ariba/overlay/modules/OverlayFactory.cpp =================================================================== --- source/ariba/overlay/modules/OverlayFactory.cpp (revision 10700) +++ source/ariba/overlay/modules/OverlayFactory.cpp (revision 12060) @@ -41,5 +41,5 @@ // structured overlays #include "chord/Chord.h" -#include "onehop/OneHop.h" +//#include "onehop/OneHop.h" //DEPRECATED namespace ariba { @@ -57,11 +57,17 @@ return new Chord( baseoverlay, nodeid, routeReceiver, param ); - case OverlayParameterSet::OverlayStructureOneHop: - return new OneHop( baseoverlay, nodeid, routeReceiver, param ); +// case OverlayParameterSet::OverlayStructureOneHop: +// return new OneHop( baseoverlay, nodeid, routeReceiver, param ); default: + // NEVER return "NULL" + assert(false); + throw 42; return NULL; } + // NEVER return "NULL" + assert(false); + throw 42; return NULL; } Index: source/ariba/overlay/modules/OverlayInterface.cpp =================================================================== --- source/ariba/overlay/modules/OverlayInterface.cpp (revision 10700) +++ source/ariba/overlay/modules/OverlayInterface.cpp (revision 12060) @@ -68,4 +68,5 @@ void OverlayInterface::onLinkFail(const LinkID& lnk, const NodeID& remote) { + onLinkDown(lnk, remote); } @@ -79,5 +80,7 @@ } -void OverlayInterface::onMessage(const DataMessage& msg, const NodeID& remote, +void OverlayInterface::onMessage(OverlayMsg* msg, + reboost::shared_buffer_t sub_msg, + const NodeID& remote, const LinkID& lnk) { } Index: source/ariba/overlay/modules/OverlayInterface.h =================================================================== --- source/ariba/overlay/modules/OverlayInterface.h (revision 10700) +++ source/ariba/overlay/modules/OverlayInterface.h (revision 12060) @@ -145,5 +145,26 @@ */ virtual const LinkID& getNextLinkId( const NodeID& id ) const = 0; - + + /** + * Returns link ids of possible next hops a route message could take, + * sorted by "quality" (e.g. overlay-distance). + * + * The »num« parameter can be used to specify the desired number of elements + * in the returned vector. This is intendet for optimizations. The + * implementation may choose to return a different number of elements than + * requested. + * + * NOTE: The returned vector may contain »unspecified« links. These refer to + * to the own node. (e.g. If there's no closer node, the top element in the + * returned vector is unsoecified.) + * + * @param id The destination node id + * @param num The desired number of elements in the returned vector. + * (0 means »not specified/max)« + * @return A sorted vector of link ids to possible next hops. + */ + virtual std::vector getSortedLinkIdsTowardsNode( + const NodeID& id, int num = 0 ) const = 0; + /** * Returns the NodeID of the next hop a route message would take. @@ -176,5 +197,7 @@ /// @see CommunicationListener - virtual void onMessage(const DataMessage& msg, const NodeID& remote, + virtual void onMessage(OverlayMsg* msg, + reboost::shared_buffer_t sub_msg, + const NodeID& remote, const LinkID& lnk = LinkID::UNSPECIFIED); Index: source/ariba/overlay/modules/OverlayStructureEvents.h =================================================================== --- source/ariba/overlay/modules/OverlayStructureEvents.h (revision 10700) +++ source/ariba/overlay/modules/OverlayStructureEvents.h (revision 12060) @@ -41,7 +41,9 @@ #include "ariba/utility/types/NodeID.h" -#include "ariba/utility/messages.h" +#include "ariba/utility/types/LinkID.h" +#include "ariba/utility/messages/Message.h" using ariba::utility::NodeID; +using ariba::utility::LinkID; using ariba::utility::Message; @@ -50,9 +52,9 @@ class OverlayInterface; -class OneHop; +//class OneHop; class OverlayStructureEvents { friend class ariba::overlay::OverlayInterface; - friend class ariba::overlay::OneHop; +// friend class ariba::overlay::OneHop; public: Index: source/ariba/overlay/modules/chord/Chord.cpp =================================================================== --- source/ariba/overlay/modules/chord/Chord.cpp (revision 10700) +++ source/ariba/overlay/modules/chord/Chord.cpp (revision 12060) @@ -43,5 +43,5 @@ #include "detail/chord_routing_table.hpp" -#include "messages/Discovery.h" +//#include "messages/Discovery.h" // XXX DEPRECATED namespace ariba { @@ -55,5 +55,60 @@ typedef chord_routing_table::item route_item; +using ariba::transport::system_priority; + use_logging_cpp( Chord ); + + +////// Messages +struct DiscoveryMessage +{ + /** + * DiscoveryMessage + * - type + * - data + * - Endpoint + */ + + // type enum + enum type_ { + invalid = 0, + normal = 1, + successor = 2, + predecessor = 3 + }; + + + // data + uint8_t type; + uint8_t ttl; + EndpointDescriptor endpoint; + + // serialize + reboost::message_t serialize() + { + // serialize endpoint + reboost::message_t msg = endpoint.serialize(); + + // serialize type and ttl + uint8_t* buff1 = msg.push_front(2*sizeof(uint8_t)).mutable_data(); + buff1[0] = type; + buff1[1] = ttl; + + return msg; + } + + //deserialize + reboost::shared_buffer_t deserialize(reboost::shared_buffer_t buff) + { + // deserialize type and ttl + const uint8_t* bytes = buff.data(); + type = bytes[0]; + ttl = bytes[1]; + + // deserialize endpoint + return endpoint.deserialize(buff(2*sizeof(uint8_t))); + } +}; + Chord::Chord(BaseOverlay& _baseoverlay, const NodeID& _nodeid, @@ -102,7 +157,22 @@ /// helper: sends a message using the "base overlay" -seqnum_t Chord::send( OverlayMsg* msg, const LinkID& link ) { - if (link.isUnspecified()) return 0; - return baseoverlay.send_link( msg, link ); +void Chord::send( OverlayMsg* msg, const LinkID& link ) { + if (link.isUnspecified()) + return; + + baseoverlay.send_link( msg, link, system_priority::OVERLAY ); +} + +void Chord::send_node( OverlayMsg* message, const NodeID& remote ) +{ + try + { + baseoverlay.send( message, remote, system_priority::OVERLAY ); + } + catch ( message_not_sent& e ) + { + logging_warn("Chord: Could not send message to " << remote + << ": " << e.what()); + } } @@ -116,20 +186,44 @@ OverlayMsg msg( typeDiscovery ); msg.setRegisterRelay(true); - Discovery dmsg( Discovery::normal, (uint8_t)ttl, baseoverlay.getEndpointDescriptor() ); - msg.encapsulate(&dmsg); + + // create DiscoveryMessage + DiscoveryMessage dmsg; + dmsg.type = DiscoveryMessage::normal; + dmsg.ttl = ttl; + dmsg.endpoint = baseoverlay.getEndpointDescriptor(); + + msg.set_payload_message(dmsg.serialize()); // send to node - baseoverlay.send_node( &msg, remote ); + try + { + baseoverlay.send_node( &msg, remote, system_priority::OVERLAY ); + } + catch ( message_not_sent& e ) + { + logging_warn("Chord: Could not send message to " << remote + << ": " << e.what()); + } } void Chord::discover_neighbors( const LinkID& link ) { uint8_t ttl = 1; + + // FIXME try-catch for the send operations + + // create DiscoveryMessage + DiscoveryMessage dmsg; + dmsg.ttl = ttl; + dmsg.endpoint = baseoverlay.getEndpointDescriptor(); { // send predecessor discovery OverlayMsg msg( typeDiscovery ); msg.setRegisterRelay(true); - Discovery dmsg( Discovery::predecessor, ttl, - baseoverlay.getEndpointDescriptor() ); - msg.encapsulate(&dmsg); + + // set type + dmsg.type = DiscoveryMessage::predecessor; + + // send + msg.set_payload_message(dmsg.serialize()); send(&msg, link); } @@ -137,9 +231,12 @@ // send successor discovery OverlayMsg msg( typeDiscovery ); - msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() ); +// msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() ); // XXX this was redundand, wasn't it? msg.setRegisterRelay(true); - Discovery dmsg( Discovery::successor, ttl, - baseoverlay.getEndpointDescriptor() ); - msg.encapsulate(&dmsg); + + // set type + dmsg.type = DiscoveryMessage::successor; + + // send + msg.set_payload_message(dmsg.serialize()); send(&msg, link); } @@ -163,5 +260,6 @@ // timer for stabilization management - Timer::setInterval(1000); +// Timer::setInterval(1000); // TODO find an appropriate interval! + Timer::setInterval(10000); // XXX testing... Timer::start(); } @@ -200,4 +298,48 @@ return item->info; } + +std::vector Chord::getSortedLinkIdsTowardsNode( + const NodeID& id, int num ) const +{ + std::vector ret; + + switch ( num ) + { + // special case: just call »getNextLinkId« + case 1: + { + ret.push_back(&getNextLinkId(id)); + + break; + } + + // * calculate top 2 * + case 0: + case 2: + { + std::vector items = table->get_next_2_hops(id); + + ret.reserve(items.size()); + + BOOST_FOREACH( const route_item* item, items ) + { + ret.push_back(&item->info); + } + + break; + } + + // NOTE: implement real sorting, if needed (and handle "case 0" properly, then) + default: + { + throw std::runtime_error("Not implemented. (Chord::getSortedLinkIdsTowardsNode with num != 2)"); + + break; + } + } + + return ret; +} + /// @see OverlayInterface.h @@ -253,4 +395,5 @@ if (remote==nodeid) { logging_warn("dropping link that has been established to myself (nodes have same nodeid?)"); + logging_warn("NodeID: " << remote); baseoverlay.dropLink(lnk); return; @@ -290,5 +433,6 @@ /// @see CommunicationListener.h or @see OverlayInterface.h void Chord::onLinkDown(const LinkID& lnk, const NodeID& remote) { - logging_debug("link_down: link=" << lnk.toString() << " remote=" << + // XXX logging_debug + logging_info("link_down (Chord): link=" << lnk.toString() << " remote=" << remote.toString() ); @@ -303,23 +447,24 @@ /// @see CommunicationListener.h /// @see OverlayInterface.h -void Chord::onMessage(const DataMessage& msg, const NodeID& remote, +void Chord::onMessage(OverlayMsg* msg, + reboost::shared_buffer_t sub_msg, + const NodeID& remote, const LinkID& link) { - // decode message - OverlayMsg* m = dynamic_cast(msg.getMessage()); - if (m == NULL) return; - // handle messages - switch ((signalMessageTypes)m->getType()) { + switch ((signalMessageTypes) msg->getType()) { // discovery request - case typeDiscovery: { - // decapsulate message - Discovery* dmsg = m->decapsulate (); + case typeDiscovery: + { + // deserialize discovery message + DiscoveryMessage dmsg; + dmsg.deserialize(sub_msg); + logging_debug("Received discovery message with" - << " src=" << m->getSourceNode().toString() - << " dst=" << m->getDestinationNode().toString() - << " ttl=" << (int)dmsg->getTTL() - << " type=" << (int)dmsg->getType() + << " src=" << msg->getSourceNode().toString() + << " dst=" << msg->getDestinationNode().toString() + << " ttl=" << (int)dmsg.ttl + << " type=" << (int)dmsg.type ); @@ -327,108 +472,125 @@ bool found = false; BOOST_FOREACH( NodeID& value, discovery ) - if (value == m->getSourceNode()) { + if (value == msg->getSourceNode()) { found = true; break; } - if (!found) discovery.push_back(m->getSourceNode()); + if (!found) discovery.push_back(msg->getSourceNode()); // check if source node can be added to routing table and setup link - if (m->getSourceNode() != nodeid) - setup( dmsg->getEndpoint(), m->getSourceNode() ); + if (msg->getSourceNode() != nodeid) + setup( dmsg.endpoint, msg->getSourceNode() ); // process discovery message -------------------------- switch start -- - switch (dmsg->getType()) { - - // normal: route discovery message like every other message - case Discovery::normal: { - // closest node? yes-> split to follow successor and predecessor - if ( table->is_closest_to(m->getDestinationNode()) ) { - logging_debug("Discovery split:"); - if (!table->get_successor()->isUnspecified()) { - OverlayMsg omsg(*m); - dmsg->setType(Discovery::successor); - omsg.encapsulate(dmsg); - logging_debug("* Routing to successor " - << table->get_successor()->toString() ); - baseoverlay.send( &omsg, *table->get_successor() ); - } - - // send predecessor message - if (!table->get_predesessor()->isUnspecified()) { - OverlayMsg omsg(*m); - dmsg->setType(Discovery::predecessor); - omsg.encapsulate(dmsg); - logging_debug("* Routing to predecessor " - << table->get_predesessor()->toString() ); - baseoverlay.send( &omsg, *table->get_predesessor() ); - } - } - // no-> route message - else { - baseoverlay.route( m ); - } - break; - } - - // successor mode: follow the successor until TTL is zero - case Discovery::successor: - case Discovery::predecessor: { - // reached destination? no->forward! - if (m->getDestinationNode() != nodeid) { - OverlayMsg omsg(*m); - omsg.encapsulate(dmsg); - omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID); - baseoverlay.route( &omsg ); - break; - } - - // time to live ended? yes-> stop routing - if (dmsg->getTTL() == 0 || dmsg->getTTL() > 10) break; - - // decrease time-to-live - dmsg->setTTL(dmsg->getTTL() - 1); - - const route_item* item = NULL; - if (dmsg->getType() == Discovery::successor && - table->get_successor() != NULL) { - item = table->get(*table->get_successor()); - } else { - if (table->get_predesessor()!=NULL) - item = table->get(*table->get_predesessor()); - } - if (item == NULL) - break; - - logging_debug("Routing discovery message to succ/pred " - << item->id.toString() ); - OverlayMsg omsg(*m); - omsg.encapsulate(dmsg); - omsg.setDestinationNode(item->id); - omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID); - baseoverlay.send(&omsg, omsg.getDestinationNode()); - break; - } - case Discovery::invalid: - break; - - default: - break; - } - // process discovery message ---------------------------- switch end -- - - delete dmsg; - break; - } - - // leave - case typeLeave: { - if (link!=LinkID::UNSPECIFIED) { - route_item* item = table->get(remote); - if (item!=NULL) item->info = LinkID::UNSPECIFIED; - table->remove(remote); - baseoverlay.dropLink(link); - } - break; - }} + switch ( dmsg.type ) + { + // normal: route discovery message like every other message + case DiscoveryMessage::normal: + { + // closest node? yes-> split to follow successor and predecessor + if ( table->is_closest_to(msg->getDestinationNode()) ) + { + logging_debug("Discovery split:"); + if (!table->get_successor()->isUnspecified()) + { + OverlayMsg omsg(*msg); + + dmsg.type = DiscoveryMessage::successor; + omsg.set_payload_message(dmsg.serialize()); + + logging_debug("* Routing to successor " + << table->get_successor()->toString() ); + send_node( &omsg, *table->get_successor() ); + } + + // send predecessor message + if (!table->get_predesessor()->isUnspecified()) + { + OverlayMsg omsg(*msg); + + dmsg.type = DiscoveryMessage::predecessor; + omsg.set_payload_message(dmsg.serialize()); + + logging_debug("* Routing to predecessor " + << table->get_predesessor()->toString() ); + send_node( &omsg, *table->get_predesessor() ); + } + } + // no-> route message + else + { + baseoverlay.route( msg ); + } + break; + } + + // successor mode: follow the successor until TTL is zero + case DiscoveryMessage::successor: + case DiscoveryMessage::predecessor: + { + // reached destination? no->forward! + if (msg->getDestinationNode() != nodeid) + { + OverlayMsg omsg(*msg); + omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID); + + omsg.set_payload_message(dmsg.serialize()); + + baseoverlay.route( &omsg ); + break; + } + + // time to live ended? yes-> stop routing + if (dmsg.ttl == 0 || dmsg.ttl > 10) break; + + // decrease time-to-live + dmsg.ttl--; + + const route_item* item = NULL; + if (dmsg.type == DiscoveryMessage::successor && + table->get_successor() != NULL) + { + item = table->get(*table->get_successor()); + } + else if (table->get_predesessor() != NULL) + { + item = table->get(*table->get_predesessor()); + } + if (item == NULL) + break; + + logging_debug("Routing discovery message to succ/pred " + << item->id.toString() ); + OverlayMsg omsg(*msg); + omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID); + omsg.setDestinationNode(item->id); + + omsg.set_payload_message(dmsg.serialize()); + + send_node( &omsg, omsg.getDestinationNode() ); + break; + } + case DiscoveryMessage::invalid: + break; + + default: + break; + } + // process discovery message ---------------------------- switch end -- + + break; + } + + // leave + case typeLeave: { + if (link!=LinkID::UNSPECIFIED) { + route_item* item = table->get(remote); + if (item!=NULL) item->info = LinkID::UNSPECIFIED; + table->remove(remote); + baseoverlay.dropLink(link); + } + break; + } + } } Index: source/ariba/overlay/modules/chord/Chord.h =================================================================== --- source/ariba/overlay/modules/chord/Chord.h (revision 10700) +++ source/ariba/overlay/modules/chord/Chord.h (revision 12060) @@ -45,4 +45,5 @@ #include "../OverlayInterface.h" #include +#include class chord_routing_table; @@ -87,7 +88,10 @@ const NodeID& node = NodeID::UNSPECIFIED ); - // helper: sends a message using the "base overlay" - seqnum_t send( OverlayMsg* msg, const LinkID& link ); + // helper: sends a message over a link using the "base overlay" + void send( OverlayMsg* msg, const LinkID& link ); + // helper: sends a message to a node using the "base overlay" + void send_node( OverlayMsg* message, const NodeID& remote ); + // stabilization: sends a discovery message to the specified neighborhood void send_discovery_to( const NodeID& destination, int ttl = 3 ); @@ -105,4 +109,9 @@ virtual const LinkID& getNextLinkId( const NodeID& id ) const; + /// @see OverlayInterface.h + /// NOTE: This implementation excepts num == 2 + virtual std::vector getSortedLinkIdsTowardsNode( + const NodeID& id, int num = 0 ) const; + /// @see OverlayInterface.h virtual const NodeID& getNextNodeId( const NodeID& id ) const; @@ -138,5 +147,7 @@ /// @see CommunicationListener.h or @see OverlayInterface.h - virtual void onMessage(const DataMessage& msg, const NodeID& remote, + virtual void onMessage(OverlayMsg* msg, + reboost::shared_buffer_t sub_msg, + const NodeID& remote, const LinkID& lnk = LinkID::UNSPECIFIED); Index: source/ariba/overlay/modules/chord/detail/chord_routing_table.hpp =================================================================== --- source/ariba/overlay/modules/chord/detail/chord_routing_table.hpp (revision 10700) +++ source/ariba/overlay/modules/chord/detail/chord_routing_table.hpp (revision 12060) @@ -96,4 +96,7 @@ // the own node id nodeid_t id; + + // the own node id as routing item + item own_id_item; // successor and predecessor tables @@ -168,6 +171,11 @@ /// constructs the reactive chord routing table explicit chord_routing_table( const nodeid_t& id, int redundancy = 4 ) : - id(id), succ( redundancy, succ_compare_type(this->id), *this ), - pred( redundancy, pred_compare_type(this->id), *this ) { + id(id), + succ( redundancy, succ_compare_type(this->id), *this ), + pred( redundancy, pred_compare_type(this->id), *this ) + { + // init reflexive item + own_id_item.id = id; + own_id_item.ref_count = 1; // create finger tables @@ -273,16 +281,68 @@ } } + if (best_item != NULL && distance(value, id)id)) return NULL; return best_item; } + + std::vector get_next_2_hops( const nodeid_t& value) + { + ring_distance distance; + item* best_item = &own_id_item; + item* second_best_item = NULL; + + // find best and second best item + for (size_t i=0; iref_count==0) continue; + + // check if we found a better item + // is best item + if ( distance(value, curr->id) < distance(value, best_item->id) ) + { + second_best_item = best_item; + best_item = curr; + } + // is second best item + else + { + if ( second_best_item == NULL ) + { + second_best_item = curr; + continue; + } + + if ( distance(value, curr->id) < distance(value, second_best_item->id) ) + { + second_best_item = curr; + } + } + } + + // prepare return vector + std::vector ret; + if ( best_item != NULL ) + { + ret.push_back(best_item); + } + if ( second_best_item != NULL ) + { + ret.push_back(second_best_item); + } + + return ret; + } const nodeid_t* get_successor() { - if (succ.size()==NULL) return NULL; + if (succ.size()==0) return NULL; return &succ.front(); } const nodeid_t* get_predesessor() { - if (pred.size()==NULL) return NULL; + if (pred.size()==0) return NULL; return &pred.front(); } Index: source/ariba/overlay/modules/chord/messages/CMakeLists.txt =================================================================== --- source/ariba/overlay/modules/chord/messages/CMakeLists.txt (revision 10700) +++ source/ariba/overlay/modules/chord/messages/CMakeLists.txt (revision 12060) @@ -37,5 +37,5 @@ # [License] -add_headers(Discovery.h) +#add_headers(Discovery.h) -add_sources(Discovery.cpp) +#add_sources(Discovery.cpp) Index: source/ariba/overlay/modules/chord/messages/Discovery.h =================================================================== --- source/ariba/overlay/modules/chord/messages/Discovery.h (revision 10700) +++ source/ariba/overlay/modules/chord/messages/Discovery.h (revision 12060) @@ -59,4 +59,5 @@ using_serialization; +// XXX This whole message is DEPRECATED class Discovery : public Message { VSERIALIZEABLE; Index: source/ariba/overlay/modules/onehop/CMakeLists.txt =================================================================== --- source/ariba/overlay/modules/onehop/CMakeLists.txt (revision 10700) +++ source/ariba/overlay/modules/onehop/CMakeLists.txt (revision 12060) @@ -37,7 +37,9 @@ # [License] -add_headers(OneHop.h) -add_sources(OneHop.cpp) +## DEPRECATED +#add_headers(OneHop.h) -add_subdir_sources(messages) +#add_sources(OneHop.cpp) + +#add_subdir_sources(messages)