Changeset 6266 for source/ariba/overlay/BaseOverlay.cpp

Timestamp:

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

Author:

mies

Message:

added basic DHT functionality (untested)

File:

• source/ariba/overlay/BaseOverlay.cpp (modified) (6 diffs)

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