Changeset 6786 for source/ariba

Timestamp:

Nov 2, 2009, 1:31:19 AM (15 years ago)

Author:

mies

Message:

Changed Data to Message conversion constructor in Message to explicit
Fixed some general bugs in Data: operator<<
Fixed bug in DHTMessage: allow unspecified key/values
Added local DHT message delivery
Adapted sources to work with gcc 4.4.1

Location:

source/ariba

Files:

• overlay/BaseOverlay.cpp (modified) (6 diffs)
• overlay/BaseOverlay.h (modified) (1 diff)
• overlay/messages/DHTMessage.h (modified) (2 diffs)
• utility/addressing/facades/address_v.hpp (modified) (1 diff)
• utility/addressing/facades/to_bytes_v.hpp (modified) (1 diff)
• utility/addressing/port_address.hpp (modified) (1 diff)
• utility/messages/Message.h (modified) (1 diff)
• utility/serialization/Data.hpp (modified) (4 diffs)
• utility/transport/tcpip/protlib/address.cpp (modified) (1 diff)

source/ariba/overlay/BaseOverlay.cpp

@@ -727 +727 @@
         sideport(&SideportListener::DEFAULT), started(false), counter(0) {
         dht = new DHT();
+        localDHT = new DHT();
 }
 
@@ -1913 +1914 @@
         if (msg->getType()==OverlayMsg::typeDHTData) {
                 const ServiceID& service = msg->getService();
-                logging_debug( "Received DHT data for service " << service.toString() );
+                logging_info( "Received DHT data for service " << service.toString() );
 
                 // delegate data message
@@ -1940 +1941 @@
 
         case OverlayMsg::typeDHTGet: {
+                logging_info("DHT-Get: key=" << dhtMsg->getKey() );
                 vector<Data> vect = dht->get(dhtMsg->getKey());
+                BOOST_FOREACH(const Data& d, vect)
+                        logging_info("DHT-Get: value=" << d);
                 OverlayMsg omsg(*msg);
                 omsg.swapRoles();
@@ -1946 +1950 @@
                 DHTMessage dhtmsg(dhtMsg->getKey(), vect);
                 omsg.encapsulate(&dhtmsg);
-                this->send(&omsg, omsg.getDestinationNode());
+                dhtSend(&omsg, omsg.getDestinationNode());
                 break;
         }
@@ -1969 +1973 @@
 /// put a value to the DHT with a ttl given in seconds
 void BaseOverlay::dhtPut( const Data& key, const Data& value, int ttl ) {
+
+        logging_info("DHT: putting key=" <<  key
+                        << " value=" << value
+                        << " ttl=" << ttl
+        );
+
+        // put into local data store (for refreshes)
+        localDHT->put(key,value,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);
+
+        OverlayMsg msg(OverlayMsg::typeDHTPut);
         msg.encapsulate( &dhtmsg );
-        send(&msg, dest);
+        dhtSend(&msg, dest);
 }
 
 /// removes a key value pair from the DHT
 void BaseOverlay::dhtRemove( const Data& key, const Data& value ) {
+        // remove from local data store
+        localDHT->remove(key,value);
+
         // calculate hash
         NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8);
+        DHTMessage dhtmsg(key,value);
+
+        // send message
         OverlayMsg msg(OverlayMsg::typeDHTRemove);
-        DHTMessage dhtmsg(key,value);
-        msg.setDestinationNode(dest);
         msg.encapsulate( &dhtmsg );
-        send(&msg, dest);
+        dhtSend(&msg, dest);
 }
 
@@ -1994 +2011 @@
         // calculate hash
         NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8);
+        DHTMessage dhtmsg(key);
+
+        // send message
         OverlayMsg msg(OverlayMsg::typeDHTRemove);
-        DHTMessage dhtmsg(key);
-        msg.setDestinationNode(dest);
         msg.encapsulate( &dhtmsg );
-        send(&msg, dest);
+        dhtSend(&msg, dest);
 }
 
 /// requests data stored using key
 void BaseOverlay::dhtGet( const Data& key, const ServiceID& service ) {
+        logging_info("DHT: trying to resolve key=" <<
+                        key << " for service=" << service.toString() );
+
         // calculate hash
         NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8);
+        DHTMessage dhtmsg(key);
+
+        // send message
         OverlayMsg msg(OverlayMsg::typeDHTGet);
-        DHTMessage dhtmsg(key);
-        msg.setDestinationNode(dest);
         msg.setService(service);
         msg.encapsulate( &dhtmsg );
-        send(&msg, dest);
-}
+        dhtSend(&msg, dest);
+}
+
+void BaseOverlay::dhtSend( OverlayMsg* msg, const NodeID& dest ) {
+        logging_info("DHT: sending message with key=" << dest.toString() );
+        msg->setSourceNode(this->nodeId);
+        msg->setDestinationNode(dest);
+
+        // local storage? yes-> put into DHT directly
+        if (overlayInterface->isClosestNodeTo(msg->getDestinationNode())) {
+                Data d = data_serialize(msg);
+                Message* m2 = new Message(d);
+                OverlayMsg* m3 = m2->decapsulate<OverlayMsg>();
+                handleDHTMessage(m3);
+                delete m2;
+                return;
+        }
+
+        // send message "normally"
+        send(msg, dest);
+}
+
 
 }} // namespace ariba, overlay

source/ariba/overlay/BaseOverlay.h

@@ -506 +506 @@
 
         DHT* dht;
+        DHT* localDHT;
 
         void stabilizeDHT();
+        void dhtSend( OverlayMsg* msg, const NodeID& dest );
 
         // misc --------------------------------------------------------------------

source/ariba/overlay/messages/DHTMessage.h

@@ -62 +62 @@
         X && ttl;
 
+//      std::cout << "1" << std::endl;
+
         // key serialization
-        uint16_t key_length = key.getLength();
+        uint16_t key_length = key.isUnspecified() ? 0 : key.getLength();
         X && key_length;
         if (X.isDeserializer()) key.setLength( key_length );
-        X && key;
+
+//      std::cout << "2 - " << key_length << std::endl;
+        X && this->key;
+//      std::cout << "2b" << std::endl;
 
         // store number of values
@@ -72 +77 @@
         X && num_values;
 
+//      std::cout << "3" << std::endl;
+
         // value serialization
         for (size_t i=0; i<num_values; i++) {
-                Data value = values[i];
-                uint16_t value_length = value.getLength();
+//              std::cout << "4" << std::endl;
+                Data value;
+                if (X.isSerializer()) value = values[i];
+                uint16_t value_length = value.isUnspecified() ? 0 : value.getLength();
                 X && value_length;
+//              std::cout << "5" << std::endl;
                 if (X.isDeserializer()) value.setLength( value_length );
                 X && value;
+//              std::cout << "6" << std::endl;
                 if (X.isDeserializer()) values.push_back(value);
         }

source/ariba/utility/addressing/facades/address_v.hpp

@@ -2 +2 @@
 #define ADDRESS_V_H_
 
+#include <stdint.h>
 #include <string>
 #include <iostream>

source/ariba/utility/addressing/facades/to_bytes_v.hpp

@@ -5 +5 @@
 
 #include <memory>
+#include <stdint.h>
 #include "vfacade.hpp"
 

source/ariba/utility/addressing/port_address.hpp

@@ -5 +5 @@
 
 #include<string>
+#include<cstdio>
+
 #include<boost/tr1/functional.hpp>
 

source/ariba/utility/messages/Message.h

@@ -105 +105 @@
          * data.
          */
-        inline Message( const Data& data ) :
+        explicit inline Message( const Data& data ) :
                 srcAddr(NULL),destAddr(NULL), releasePayload(true) {
                 this->payload = data.clone();

source/ariba/utility/serialization/Data.hpp

@@ -49 +49 @@
 template<typename T = uint8_t, typename DataModel = DefaultDataModel<uint8_t> > class DataTpl;
 typedef DataTpl<> Data;
-template<typename T, typename DataModel> std::ostream& operator<<(std::ostream& stream, DataTpl<T, DataModel>& data);
+template<typename T, typename DataModel> std::ostream& operator<<(std::ostream& stream, const DataTpl<T, DataModel>& data);
 
 //== internal includes ==
@@ -324 +324 @@
 };
 
-/* default human readable text output */
-template<typename T, typename DataModel>
-std::ostream& operator<<(std::ostream& stream, DataTpl<T, DataModel>& data) {
-        stream << "[" << bitstr(data.getBuffer(), data.getLength(), 4)
-                        << "|'";
-        char* buffer = (char*) data.getBuffer();
-        for (size_t i = 0; i < data.getLength() / 8; i++) {
-                char c = buffer[i] < 32 ? '.' : buffer[i];
-                stream << c;
-        }
-        stream << "']";
-        return stream;
-}
-
 /* unspecified type */
 template<typename T, typename DataModel>
@@ -397 +383 @@
 
         finline int32_t length() const {
-                return bufferLen;
+                return (bufferLen == -1) ? 0 : bufferLen;
         }
 
@@ -442 +428 @@
 }sznEnd();
 
+/* default human readable text output */
+template<typename T, typename DataModel>
+std::ostream& operator<<(std::ostream& stream, const DataTpl<T, DataModel>& data) {
+        stream << "[" << bitstr(data.getBuffer(), data.getLength(), 4)
+                        << "|'";
+        const char* buffer = (const char*) data.getBuffer();
+        for (size_t i = 0; i < data.getLength() / 8; i++) {
+                char c = buffer[i] < 32 ? '.' : buffer[i];
+                stream << c;
+        }
+        stream << "']";
+        return stream;
+}
 #endif /* DATA_HPP_ */

source/ariba/utility/transport/tcpip/protlib/address.cpp

@@ -534 +534 @@
         char ipstr[INET6_ADDRSTRLEN] = {0};
         // look for /
-        i = strchr(str,'/');
+        i = strchr((char*)str,'/');
         if (i) {
                 iplen = i-str;