Index: /source/ariba/CommunicationListener.cpp =================================================================== --- /source/ariba/CommunicationListener.cpp (revision 6211) +++ /source/ariba/CommunicationListener.cpp (revision 6266) @@ -62,10 +62,10 @@ bool CommunicationListener::onLinkRequest(const NodeID& remote, - const DataMessage& msg) { + const DataMessage& msg) { return true; } void CommunicationListener::onMessage(const DataMessage& msg, - const NodeID& remote, const LinkID& lnk) { + const NodeID& remote, const LinkID& lnk) { } @@ -73,4 +73,9 @@ return true; } + +void CommunicationListener::onKeyValue( const Data& key, const vector& value ) { + +} + // --- extended message functionality --- Index: /source/ariba/CommunicationListener.h =================================================================== --- /source/ariba/CommunicationListener.h (revision 6211) +++ /source/ariba/CommunicationListener.h (revision 6266) @@ -86,17 +86,20 @@ // --- sniffing related method --- + virtual bool onEnableSideportListener(); + // --- dht functionality --- + + virtual void onKeyValue( const Data& key, const vector& value ); + // --- extended message functionality --- + // virtual void onLinkQoSChanged(const LinkID& lnk, const NodeID& remote, // const LinkProperties& prop); // --- extended message functionality --- + // virtual void onMessageSent(seqnum_t seq_num, bool failed, // const DataMessage& msg = DataMessage::UNSPECIFIED); - - // --- dht functionality --- - // virtual void onGetResponse( const Identifier<> id, const Message* msg ); - // virtual void onPutResponse( const Identifier<> id, const Message* msg ); }; Index: /source/ariba/Makefile.am =================================================================== --- /source/ariba/Makefile.am (revision 6211) +++ /source/ariba/Makefile.am (revision 6266) @@ -144,4 +144,5 @@ overlay/messages/JoinReply.cpp \ overlay/messages/JoinRequest.cpp \ + overlay/messages/DHTMessage.cpp \ overlay/messages/OverlayMsg.cpp @@ -149,4 +150,5 @@ overlay/messages/JoinReply.h \ overlay/messages/JoinRequest.h \ + overlay/messages/DHTMessage.h\ overlay/messages/OverlayMsg.h Index: /source/ariba/Node.cpp =================================================================== --- /source/ariba/Node.cpp (revision 6211) +++ /source/ariba/Node.cpp (revision 6266) @@ -176,12 +176,12 @@ // service directory -/* - void Node::put(const Identifier<>& key, Message* value) { - } - - void Node::get(const Identifier<>& key) { - - } - */ + +void Node::put( const Data& key, const Data& value, uint16_t ttl ) { + base_overlay->dhtPut(key,value,ttl); +} + +void Node::get( const Data& key, const ServiceID& sid ) { + base_overlay->dhtGet(key,sid); +} // @see Module.h Index: /source/ariba/Node.h =================================================================== --- /source/ariba/Node.h (revision 6211) +++ /source/ariba/Node.h (revision 6266) @@ -282,10 +282,23 @@ bool unbind(CommunicationListener* listener, const ServiceID& sid); - // --- extension proposal: service directory -- - // main-idea: use this methods to register groups/rendevous points inside - // the base overlay via a distributed storage service. - // - //void put(const KeyID& key, Message* value); - //void get(const KeyID& key); + /** + * Adds a key value pair to the DHT + * + * @param key The key data + * @param value The value data + * @param ttl The time to live in seconds + */ + void put( const Data& key, const Data& value, uint16_t ttl ); + + /** + * Queries for values stored in the DHT. Fires an communication event when + * values arrive. + * + * @param key The key data + * @param sid The service that is requesting the values + */ + void get( const Data& key, const ServiceID& sid ); + + //------------------------------------------------------------------------- // Index: /source/ariba/overlay/BaseOverlay.cpp =================================================================== --- /source/ariba/overlay/BaseOverlay.cpp (revision 6211) +++ /source/ariba/overlay/BaseOverlay.cpp (revision 6266) @@ -51,4 +51,5 @@ #include "ariba/overlay/messages/OverlayMsg.h" +#include "ariba/overlay/messages/DHTMessage.h" #include "ariba/overlay/messages/JoinRequest.h" #include "ariba/overlay/messages/JoinReply.h" @@ -58,4 +59,184 @@ namespace ariba { namespace overlay { + +class ValueEntry { +public: + ValueEntry( const Data& value ) : ttl(0), last_update(time(NULL)), + last_change(time(NULL)), value(value.clone()) { + } + + ~ValueEntry() { + value.release(); + } + + void refresh() { + last_update = time(NULL); + } + + void set_value( const Data& value ) { + this->value.release(); + this->value = value.clone(); + this->last_change = time(NULL); + this->last_update = time(NULL); + } + + Data get_value() const { + return value; + } + + uint16_t get_ttl() const { + return ttl; + } + + void set_ttl( uint16_t ttl ) { + this->ttl = ttl; + } + + bool is_ttl_elapsed() const { + // is persistent? yes-> always return false + if (ttl==0) return false; + + // return true, if ttl is elapsed + return ( difftime( time(NULL), this->last_update ) >= ttl ); + } + +private: + uint16_t ttl; + time_t last_update; + time_t last_change; + Data value; +}; + +class DHTEntry { +public: + Data key; + vector values; + + vector get_values() { + vector vect; + BOOST_FOREACH( ValueEntry& e, values ) + vect.push_back( e.get_value() ); + return vect; + } +}; + +class DHT { +public: + typedef vector Entries; + typedef vector Values; + Entries entries; + + static bool equals( const Data& lhs, const Data& rhs ) { + if (rhs.getLength()!=lhs.getLength()) return false; + for (int i=0; i refresh ttl + if ( equals(entry.values[j].get_value(), value) ) { + entry.values[j].refresh(); + std::cout << "DHT: Republished value. Refreshing TTL." + << std::endl; + return; + } + } + + // new value-> add to entry + std::cout << "DHT: Added value to " + << " key=" << key << " with value=" << value << std::endl; + entry.values.push_back( ValueEntry( value ) ); + entry.values.back().set_ttl(ttl); + return; + } + } + + // key is unknown-> add key value pair + std::cout << "DHT: New key value pair " + << " key=" << key << " with value=" << value << std::endl; + + // add new entry + entries.push_back( DHTEntry() ); + DHTEntry& entry = entries.back(); + entry.key = key.clone(); + entry.values.push_back( ValueEntry(value) ); + entry.values.back().set_ttl(ttl); + } + + vector get( const Data& key ) { + // find entry + for (size_t i=0; i(); + } + + bool remove( const Data& key ) { + // find entry + for (Entries::iterator i = entries.begin(); i != entries.end(); i++) { + DHTEntry& entry = *i; + + // found? yes-> delete entry + if ( equals(entry.key, key) ) { + i = entries.erase(i); + return true; + } + } + return false; + } + + bool remove( const Data& key, const Data& value ) { + // find entry + for (Entries::iterator i = entries.begin(); i != entries.end(); i++) { + DHTEntry& entry = *i; + + // found? yes-> try to find value + if ( equals(entry.key, key) ) { + for (Values::iterator j = entry.values.begin(); + j != entry.values.end(); j++) { + + // value found? yes-> delete + if (equals(j->get_value(), value)) { + j = entry.values.erase(j); + return true; + } + } + } + } + return false; + } + + void cleanup() { + // find entry + for (Entries::iterator i = entries.begin(); i != entries.end(); i++) { + DHTEntry& entry = *i; + + for (Values::iterator j = entry.values.begin(); + j != entry.values.end(); j++) { + + // value found? yes-> delete + if (j->is_ttl_elapsed()) + j = entry.values.erase(j); + } + + if (entry.values.size()==0) i = entries.erase(i); + } + } +}; + +// ---------------------------------------------------------------------------- /* ***************************************************************************** @@ -545,7 +726,9 @@ spovnetId(SpoVNetID::UNSPECIFIED), state(BaseOverlayStateInvalid), sideport(&SideportListener::DEFAULT), started(false), counter(0) { + dht = new DHT(); } BaseOverlay::~BaseOverlay() { + delete dht; } @@ -1582,4 +1765,8 @@ overlayMsg->addRouteRecord(nodeId); + // handle dht messages (do not route) + if (overlayMsg->isDHTMessage()) + return handleDHTMessage(overlayMsg); + // handle signaling messages (do not route!) if (overlayMsg->getType()>=OverlayMsg::typeSignalingStart && @@ -1601,4 +1788,12 @@ return true; } + + // handle DHT response messages + if (overlayMsg->hasTypeMask( OverlayMsg::maskDHTResponse )) { + bool ret = handleDHTMessage(overlayMsg); + delete overlayMsg; + return ret; + } + // handle base overlay message @@ -1699,3 +1894,124 @@ } + +// ---------------------------------------------------------------------------- + + + +/// stabilize DHT state +void BaseOverlay::stabilizeDHT() { + +} + +// handle DHT messages +bool BaseOverlay::handleDHTMessage( OverlayMsg* msg ) { + + // decapsulate message + logging_debug("received DHT message"); + DHTMessage* dhtMsg = msg->decapsulate(); + + // handle DHT data message + if (msg->getType()==OverlayMsg::typeDHTData) { + const ServiceID& service = msg->getService(); + logging_debug( "Received DHT data for service " << service.toString() ); + + // delegate data message + getListener(service)->onKeyValue(dhtMsg->getKey(), dhtMsg->getValues() ); + return true; + } + + // route message to closest node + if (!overlayInterface->isClosestNodeTo(msg->getDestinationNode())) { + logging_debug("Routing DHT message to closest node " + << " from " << msg->getSourceNode() + << " to " << msg->getDestinationNode() + ); + route( msg ); + delete msg; + return true; + } + + // now, we are the closest node... + switch (msg->getType()) { + case OverlayMsg::typeDHTPut: { + BOOST_FOREACH( Data value, dhtMsg->getValues() ) + dht->put(dhtMsg->getKey(), value, dhtMsg->getTTL() ); + break; + } + + case OverlayMsg::typeDHTGet: { + vector vect = dht->get(dhtMsg->getKey()); + OverlayMsg omsg(*msg); + omsg.swapRoles(); + omsg.setType(OverlayMsg::typeDHTData); + DHTMessage dhtmsg(dhtMsg->getKey(), vect); + omsg.encapsulate(&dhtmsg); + this->send(&omsg, omsg.getDestinationNode()); + break; + } + + case OverlayMsg::typeDHTRemove: { + if (dhtMsg->hasValues()) { + BOOST_FOREACH( Data value, dhtMsg->getValues() ) + dht->remove(dhtMsg->getKey(), value ); + } else + dht->remove( dhtMsg->getKey() ); + break; + } + + default: + logging_error("DHT Message type unknown."); + return false; + } + delete msg; + return true; +} + +/// put a value to the DHT with a ttl given in seconds +void BaseOverlay::dhtPut( const Data& key, const Data& value, int ttl ) { + // calculate hash + NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8); + OverlayMsg msg(OverlayMsg::typeDHTPut); + DHTMessage dhtmsg(key,value); + dhtmsg.setTTL(ttl); + msg.setDestinationNode(dest); + msg.encapsulate( &dhtmsg ); + send(&msg, dest); +} + +/// removes a key value pair from the DHT +void BaseOverlay::dhtRemove( const Data& key, const Data& value ) { + // calculate hash + NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8); + OverlayMsg msg(OverlayMsg::typeDHTRemove); + DHTMessage dhtmsg(key,value); + msg.setDestinationNode(dest); + msg.encapsulate( &dhtmsg ); + send(&msg, dest); +} + +/// removes all data stored at the given key +void BaseOverlay::dhtRemove( const Data& key ) { + // calculate hash + NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8); + OverlayMsg msg(OverlayMsg::typeDHTRemove); + DHTMessage dhtmsg(key); + msg.setDestinationNode(dest); + msg.encapsulate( &dhtmsg ); + send(&msg, dest); +} + +/// requests data stored using key +void BaseOverlay::dhtGet( const Data& key, const ServiceID& service ) { + // calculate hash + NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8); + OverlayMsg msg(OverlayMsg::typeDHTRemove); + DHTMessage dhtmsg(key); + msg.setDestinationNode(dest); + msg.setService(service); + msg.encapsulate( &dhtmsg ); + send(&msg, dest); +} + + }} // namespace ariba, overlay Index: /source/ariba/overlay/BaseOverlay.h =================================================================== --- /source/ariba/overlay/BaseOverlay.h (revision 6211) +++ /source/ariba/overlay/BaseOverlay.h (revision 6266) @@ -122,4 +122,5 @@ class LinkDescriptor; class OverlayMsg; +class DHT; class BaseOverlay: public MessageReceiver, @@ -284,6 +285,17 @@ void leaveSpoVNet(); + /// put a value to the DHT with a ttl given in seconds + void dhtPut( const Data& key, const Data& value, int ttl = 0); + + /// removes a key value pair from the DHT + void dhtRemove( const Data& key, const Data& value ); + + /// removes all data stored at the given key + void dhtRemove( const Data& key ); + + /// requests data stored using key + void dhtGet( const Data& key, const ServiceID& service ); + protected: - /** * @see ariba::communication::CommunicationEvents.h @@ -397,4 +409,7 @@ bool handleJoinReply( OverlayMsg* msg, const LinkID& bcLink ); + // handle DHT messages + bool handleDHTMessage( OverlayMsg* msg ); + // handle link messages bool handleLinkRequest( OverlayMsg* msg, LinkDescriptor* ld ); @@ -403,4 +418,5 @@ bool handleLinkDirect( OverlayMsg* msg, LinkDescriptor* ld ); bool handleLinkAlive( OverlayMsg* msg, LinkDescriptor* ld ); + // link state handling ----------------------------------------------------- @@ -487,4 +503,10 @@ bool ignore_down = false ); + // distributed hashtable handling ------------------------------------------ + + DHT* dht; + + void stabilizeDHT(); + // misc -------------------------------------------------------------------- Index: /source/ariba/overlay/messages/DHTMessage.cpp =================================================================== --- /source/ariba/overlay/messages/DHTMessage.cpp (revision 6266) +++ /source/ariba/overlay/messages/DHTMessage.cpp (revision 6266) @@ -0,0 +1,40 @@ +#include "DHTMessage.h" + +#include + +namespace ariba { +namespace overlay { + +vsznDefault(DHTMessage); + +DHTMessage::DHTMessage() { + this->key.setLength(0); + this->ttl =0 ; +} + +DHTMessage::DHTMessage( const Data& key ) { + // calculate hash of key + this->hash = NodeID::sha1( key.getBuffer(), key.getLength() / 8 ); + this->key = key.clone(); + this->ttl =0 ; +} + +DHTMessage::DHTMessage( const Data& key, const Data& value ) { + // calculate hash of key + this->hash = NodeID::sha1( key.getBuffer(), key.getLength() / 8 ); + this->key = key.clone(); + this->values.push_back( value.clone() ); + this->ttl =0 ; +} + +DHTMessage::DHTMessage( const Data& key, const vector& values ) { + BOOST_FOREACH(const Data value, values ) + this->values.push_back( value.clone() ); +} + +DHTMessage::~DHTMessage() { + this->key.release(); + BOOST_FOREACH( Data& value, values ) value.release(); +} + +}} Index: /source/ariba/overlay/messages/DHTMessage.h =================================================================== --- /source/ariba/overlay/messages/DHTMessage.h (revision 6266) +++ /source/ariba/overlay/messages/DHTMessage.h (revision 6266) @@ -0,0 +1,85 @@ +#ifndef DHTMESSAGE_H_ +#define DHTMESSAGE_H_ + +#include "ariba/utility/messages.h" +#include "ariba/utility/serialization.h" + +namespace ariba { +namespace overlay { + +using ariba::utility::Message; +using_serialization; + +class DHTMessage : public Message { VSERIALIZEABLE +public: + DHTMessage(); + DHTMessage( const Data& key ); + DHTMessage( const Data& key, const Data& value ); + DHTMessage( const Data& key, const vector& values ); + virtual ~DHTMessage(); + + const NodeID& getHashedKey() const { + return hash; + } + + const Data& getKey() const { + return key; + } + + /// returns the first element of the key vector + const Data& getValue() const { + return values.at(0); + } + + bool hasValues() const { + values.size() != 0; + } + + uint16_t getTTL() const { + return ttl; + } + + void setTTL( uint16_t ttl ) { + this->ttl = ttl; + } + + /// return alles values for the key + const vector& getValues() const { + return values; + } + +private: + NodeID hash; + uint16_t ttl; + Data key; + vector values; +}; + +}} // namespace ariba::overlay + +sznBeginDefault( ariba::overlay::DHTMessage, X ) { + + X && ttl; + + // key serialization + uint16_t key_length = key.getLength(); + X && key_length; + if (X.isDeserializer()) key.setLength( key_length ); + X && key; + + // store number of values + uint16_t num_values = values.size(); + X && num_values; + + // value serialization + for (size_t i=0; iinfo.isUnspecified()) return EndpointDescriptor::UNSPECIFIED(); return baseoverlay.getEndpointDescriptor(item->info); +} + +/// @see OverlayInterface.h +bool Chord::isClosestNodeTo( const NodeID& node ) { + return table->is_closest_to(node); } Index: /source/ariba/overlay/modules/chord/Chord.h =================================================================== --- /source/ariba/overlay/modules/chord/Chord.h (revision 6211) +++ /source/ariba/overlay/modules/chord/Chord.h (revision 6266) @@ -123,4 +123,7 @@ /// @see OverlayInterface.h + virtual bool isClosestNodeTo( const NodeID& node ); + + /// @see OverlayInterface.h virtual NodeList getKnownNodes(bool deep = true) const; Index: /source/ariba/overlay/modules/onehop/OneHop.cpp =================================================================== --- /source/ariba/overlay/modules/onehop/OneHop.cpp (revision 6211) +++ /source/ariba/overlay/modules/onehop/OneHop.cpp (revision 6266) @@ -79,4 +79,11 @@ } + +/// @see OverlayInterface.h +bool OneHop::isClosestNodeTo( const NodeID& node ) { + throw "NOT IMPLEMENTED!"; + return false; +} + void OneHop::routeMessage(const NodeID& destnode, Message* msg){ Index: /source/ariba/overlay/modules/onehop/OneHop.h =================================================================== --- /source/ariba/overlay/modules/onehop/OneHop.h (revision 6211) +++ /source/ariba/overlay/modules/onehop/OneHop.h (revision 6266) @@ -81,4 +81,7 @@ /// @see OverlayInterface.h + virtual bool isClosestNodeTo( const NodeID& node ); + + /// @see OverlayInterface.h virtual const LinkID& getNextLinkId( const NodeID& id ) const;