Changeset 6266 for source/ariba/overlay

Timestamp:

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

Author:

mies

Message:

added basic DHT functionality (untested)

Location:

source/ariba/overlay

Files:

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

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;