Changeset 6266 for source/ariba

Timestamp:

Sep 25, 2009, 2:30:33 PM (15 years ago)

Author:

mies

Message:

added basic DHT functionality (untested)

Location:

source/ariba

Files:

• CommunicationListener.cpp (modified) (2 diffs)
• CommunicationListener.h (modified) (1 diff)
• Makefile.am (modified) (2 diffs)
• Node.cpp (modified) (1 diff)
• Node.h (modified) (1 diff)
• overlay/BaseOverlay.cpp (modified) (6 diffs)
• overlay/BaseOverlay.h (modified) (5 diffs)
• overlay/messages/DHTMessage.cpp (added)
• overlay/messages/DHTMessage.h (added)
• overlay/messages/OverlayMsg.h (modified) (3 diffs)
• overlay/modules/OverlayInterface.h (modified) (1 diff)
• overlay/modules/chord/Chord.cpp (modified) (1 diff)
• overlay/modules/chord/Chord.h (modified) (1 diff)
• overlay/modules/onehop/OneHop.cpp (modified) (1 diff)
• overlay/modules/onehop/OneHop.h (modified) (1 diff)

source/ariba/CommunicationListener.cpp

@@ -62 +62 @@
 
 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 +73 @@
         return true;
 }
+
+void CommunicationListener::onKeyValue( const Data& key, const vector<Data>& value ) {
+
+}
+
 
 // --- extended message functionality ---

source/ariba/CommunicationListener.h

@@ -86 +86 @@
 
         // --- sniffing related method ---
+
         virtual bool onEnableSideportListener();
 
+        // --- dht functionality ---
+
+        virtual void onKeyValue( const Data& key, const vector<Data>& 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 );
 };
 

source/ariba/Makefile.am

@@ -144 +144 @@
   overlay/messages/JoinReply.cpp \
   overlay/messages/JoinRequest.cpp \
+  overlay/messages/DHTMessage.cpp \
   overlay/messages/OverlayMsg.cpp
 
@@ -149 +150 @@
   overlay/messages/JoinReply.h \
   overlay/messages/JoinRequest.h \
+  overlay/messages/DHTMessage.h\
   overlay/messages/OverlayMsg.h
 

source/ariba/Node.cpp

@@ -176 +176 @@
 
 // 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

source/ariba/Node.h

@@ -282 +282 @@
         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 );
+
+
         //-------------------------------------------------------------------------
         //

source/ariba/overlay/BaseOverlay.cpp

@@ -51 +51 @@
 
 #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 +59 @@
 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<ValueEntry> values;
+
+        vector<Data> get_values() {
+                vector<Data> vect;
+                BOOST_FOREACH( ValueEntry& e, values )
+                        vect.push_back( e.get_value() );
+                return vect;
+        }
+};
+
+class DHT {
+public:
+        typedef vector<DHTEntry> Entries;
+        typedef vector<ValueEntry> Values;
+        Entries entries;
+
+        static bool equals( const Data& lhs, const Data& rhs ) {
+                if (rhs.getLength()!=lhs.getLength()) return false;
+                for (int i=0; i<lhs.getLength()/8; i++)
+                        if (lhs.getBuffer()[i] != rhs.getBuffer()[i]) return false;
+                return true;
+        }
+
+        void put( const Data& key, const Data& value, uint16_t ttl = 0 ) {
+
+                // find entry
+                for (size_t i=0; i<entries.size(); i++) {
+                        DHTEntry& entry = entries.at(i);
+
+                        // check if key is already known
+                        if ( equals(entry.key, key) ) {
+
+                                // check if value is already in values list
+                                for (size_t j=0; j<entry.values.size(); j++) {
+                                        // found value already? yes-> 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<Data> get( const Data& key ) {
+                // find entry
+                for (size_t i=0; i<entries.size(); i++) {
+                        DHTEntry& entry = entries.at(i);
+                        if ( equals(entry.key,key) )
+                                return entry.get_values();
+                }
+                return vector<Data>();
+        }
+
+        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 +726 @@
         spovnetId(SpoVNetID::UNSPECIFIED), state(BaseOverlayStateInvalid),
         sideport(&SideportListener::DEFAULT), started(false), counter(0) {
+        dht = new DHT();
 }
 
 BaseOverlay::~BaseOverlay() {
+        delete dht;
 }
 
@@ -1582 +1765 @@
         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 +1788 @@
                 return true;
         }
+
+        // handle DHT response messages
+        if (overlayMsg->hasTypeMask( OverlayMsg::maskDHTResponse )) {
+                bool ret = handleDHTMessage(overlayMsg);
+                delete overlayMsg;
+                return ret;
+        }
+
 
         // handle base overlay message
@@ -1699 +1894 @@
 }
 
+
+// ----------------------------------------------------------------------------
+
+
+
+/// 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<DHTMessage>();
+
+        // 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<Data> 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

source/ariba/overlay/BaseOverlay.h

@@ -122 +122 @@
 class LinkDescriptor;
 class OverlayMsg;
+class DHT;
 
 class BaseOverlay: public MessageReceiver,
@@ -284 +285 @@
         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 +409 @@
         bool handleJoinReply( OverlayMsg* msg, const LinkID& bcLink );
 
+        // handle DHT messages
+        bool handleDHTMessage( OverlayMsg* msg );
+
         // handle link messages
         bool handleLinkRequest( OverlayMsg* msg, LinkDescriptor* ld );
@@ -403 +418 @@
         bool handleLinkDirect( OverlayMsg* msg, LinkDescriptor* ld );
         bool handleLinkAlive( OverlayMsg* msg, LinkDescriptor* ld );
+
 
         // link state handling -----------------------------------------------------
@@ -487 +503 @@
                 bool ignore_down = false );
 
+        // distributed hashtable handling ------------------------------------------
+
+        DHT* dht;
+
+        void stabilizeDHT();
+
         // misc --------------------------------------------------------------------
 

source/ariba/overlay/messages/OverlayMsg.h

@@ -88 +88 @@
                 typeLinkDirect  = 0x34, ///< direct connection has been established
                 typeLinkAlive   = 0x35, ///< keep-alive message
+
+                // DHT routed messages
+                maskDHT                 = 0x40, ///< bit mask for dht messages
+                typeDHTPut      = 0x41, ///< DHT put operation
+                typeDHTGet      = 0x42, ///< DHT get operation
+                typeDHTRemove   = 0x43, ///< DHT remove operation
+
+                /// DHT response messages
+                maskDHTResponse = 0x50, ///< bit mask for dht responses
+                typeDHTData     = 0x51, ///< DHT get data
 
                 // topology signaling
@@ -179 +189 @@
         }
 
-
         bool containsSourceEndpoint() const {
                 return (flags & 0x20)!=0;
@@ -186 +195 @@
         void setContainsSourceEndpoint(bool contains_endpoint) {
                 if (contains_endpoint) flags |= 0x20; else flags &= ~0x20;
+        }
+
+        bool isDHTMessage() const {
+                return hasTypeMask(maskDHT);
         }
 

source/ariba/overlay/modules/OverlayInterface.h

@@ -116 +116 @@
         virtual const EndpointDescriptor& resolveNode(const NodeID& node) = 0;
 
+
+        /**
+         * Returns true if this is the closest node to the given node
+         * identifier.
+         *
+         * @param node The node identifier to compare with
+         * @return True if this is the closest node to the given node identifier
+         */
+        virtual bool isClosestNodeTo( const NodeID& node ) = 0;
+
         /**
          * Returns the nodes known to this overlay.

source/ariba/overlay/modules/chord/Chord.cpp

@@ -181 +181 @@
         if (item == NULL || item->info.isUnspecified()) return EndpointDescriptor::UNSPECIFIED();
         return baseoverlay.getEndpointDescriptor(item->info);
+}
+
+/// @see OverlayInterface.h
+bool Chord::isClosestNodeTo( const NodeID& node ) {
+        return table->is_closest_to(node);
 }
 

source/ariba/overlay/modules/chord/Chord.h

@@ -123 +123 @@
 
         /// @see OverlayInterface.h
+        virtual bool isClosestNodeTo( const NodeID& node );
+
+        /// @see OverlayInterface.h
         virtual NodeList getKnownNodes(bool deep = true) const;
 

source/ariba/overlay/modules/onehop/OneHop.cpp

@@ -79 +79 @@
 }
 
+
+/// @see OverlayInterface.h
+bool OneHop::isClosestNodeTo( const NodeID& node ) {
+        throw "NOT IMPLEMENTED!";
+        return false;
+}
+
 void OneHop::routeMessage(const NodeID& destnode, Message* msg){
 

source/ariba/overlay/modules/onehop/OneHop.h

@@ -81 +81 @@
 
         /// @see OverlayInterface.h
+        virtual bool isClosestNodeTo( const NodeID& node );
+
+        /// @see OverlayInterface.h
         virtual const LinkID& getNextLinkId( const NodeID& id ) const;