Merge pull request #116 from kaldoran/filter_pht_values

pht : Filter of duplicate values

.gitignore

@@ -45,3 +45,13 @@ Makefile.in
 # KDevelop
 .kdev4/
 *.kdev4
+
+# Python
+*.pyc
+
+# git backup files
+*.orig
+
+# vim swap files
+*.swp
+*.swo

.travis.yml

@@ -1,44 +1,18 @@
-sudo: false
+sudo: required
 
 services:
   - docker
 
 language: cpp
 
-compiler:
-  - gcc
-  - clang
-
-addons:
-  apt:
-    sources:
-    - ubuntu-toolchain-r-test
-    packages:
-    - gcc-6
-    - g++-6
-    - libgnutls28-dev
-    - cmake
-
 before_install:
-  - if [ $TRAVIS_OS_NAME == linux ] && [ $CXX == g++ ]; then export CXX=/usr/bin/g++-6 && export CC=/usr/bin/gcc-6; fi
-  - if [ $TRAVIS_OS_NAME == linux ]; then git clone https://github.com/msgpack/msgpack-c.git; fi
-  - if [ $TRAVIS_OS_NAME == linux ]; then cd msgpack-c && git checkout cpp-1.2.0 && mkdir build && mkdir -p $HOME/usr && cd build; fi
-  - if [ $TRAVIS_OS_NAME == linux ]; then cmake -DCMAKE_INSTALL_PREFIX=$HOME/usr -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF .. && make -j && make install && cd ../.. && pwd; fi
-  - if [ $TRAVIS_OS_NAME == osx ]; then brew update && brew install llvm36 && brew install msgpack && brew install gnutls && brew link --force llvm36; fi
-
-before_script:
-  - if [ $TRAVIS_OS_NAME == linux ]; then travis_retry ./autogen.sh && ./configure --prefix=$HOME/usr; fi
-  - if [ $TRAVIS_OS_NAME == osx ]; then travis_retry ./autogen.sh && ./configure; fi
+- docker pull aberaud/opendht-deps
+- docker pull aberaud/opendht-deps-llvm
 
 script:
-  - make -j
+- docker build -f docker/DockerfileTravis .
+- docker build -f docker/DockerfileTravisLlvm .
 
 notifications:
   email:
       - adrien.beraud@savoirfairelinux.com
-
-env:
-  - LANG="en_US.UTF-8"
-
-os:
-  - osx

CMakeLists.txt

@@ -23,11 +23,12 @@ option (OPENDHT_TOOLS "Build DHT tools" ON)
 set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11 -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor")
 set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
 if (OPENDHT_LTO)
-    set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -flto -fuse-linker-plugin")
+    set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -flto")
     if (CMAKE_COMPILER_IS_GNUCC)
-        SET (CMAKE_AR        "gcc-ar")
-        SET (CMAKE_NM        "gcc-nm")
-        SET (CMAKE_RANLIB    "gcc-ranlib")
+        set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fuse-linker-plugin")
+        set (CMAKE_AR        "gcc-ar")
+        set (CMAKE_NM        "gcc-nm")
+        set (CMAKE_RANLIB    "gcc-ranlib")
     endif ()
 endif ()
 
@@ -51,6 +52,7 @@ list (APPEND opendht_SOURCES
     src/network_engine.cpp
     src/securedht.cpp
     src/dhtrunner.cpp
+	src/indexation/pht.cpp
     src/argon2/argon2.c
     src/argon2/core.c
     src/argon2/blake2/blake2b.c
@@ -76,6 +78,7 @@ list (APPEND opendht_HEADERS
     include/opendht/securedht.h
     include/opendht/log.h
     include/opendht.h
+	include/opendht/indexation/pht.h
 )
 
 configure_file (

docker/Dockerfile

@@ -0,0 +1,6 @@
+FROM aberaud/opendht-deps
+MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+RUN git clone https://github.com/savoirfairelinux/opendht.git \
+	&& cd opendht && mkdir build && cd build \
+	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On && make -j8 && make install \
+	&& cd ../.. && rm -rf opendht

docker/DockerfileDeps

@@ -0,0 +1,9 @@
+FROM ubuntu:16.04
+MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+RUN apt-get update && apt-get install -y build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls-dev cython3 python3-dev python3-setuptools && apt-get clean
+RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.0.0/msgpack-2.0.0.tar.gz \
+	&& tar -xzf msgpack-2.0.0.tar.gz \
+	&& cd msgpack-2.0.0 && mkdir build && cd build \
+	&& cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
+	&& make -j8 && make install \
+	&& cd ../.. && rm -rf msgpack-2.0.0 msgpack-2.0.0.tar.gz

docker/DockerfileDepsLlvm

@@ -0,0 +1,13 @@
+FROM ubuntu:16.04
+MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+RUN apt-get update \
+	&& apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls-dev cython3 python3-dev python3-setuptools \
+	&& apt-get remove -y gcc g++ && apt-get autoremove -y && apt-get clean
+ENV CC cc
+ENV CXX c++
+RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.0.0/msgpack-2.0.0.tar.gz \
+	&& tar -xzf msgpack-2.0.0.tar.gz \
+	&& cd msgpack-2.0.0 && mkdir build && cd build \
+	&& cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
+	&& make -j8 && make install \
+	&& cd ../.. && rm -rf msgpack-2.0.0 msgpack-2.0.0.tar.gz

docker/DockerfileLlvm

@@ -0,0 +1,6 @@
+FROM aberaud/opendht-deps-llvm
+MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+RUN git clone https://github.com/savoirfairelinux/opendht.git \
+	&& cd opendht && mkdir build && cd build \
+	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On && make -j8 && make install \
+	&& cd ../.. && rm -rf opendht

docker/DockerfileTravis

@@ -0,0 +1,6 @@
+FROM aberaud/opendht-deps
+MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+COPY . /root/opendht
+RUN cd /root/opendht && mkdir build && cd build \
+	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On .. && make -j8 && make install \
+	&& cd ../.. && rm -rf opendht

docker/DockerfileTravisLlvm

@@ -0,0 +1,6 @@
+FROM aberaud/opendht-deps-llvm
+MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+COPY . /root/opendht
+RUN cd /root/opendht && mkdir build && cd build \
+	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On .. && make -j8 && make install \
+	&& cd ../.. && rm -rf opendht

include/opendht.h

@@ -22,3 +22,4 @@
 #include "opendht/dhtrunner.h"
 #include "opendht/log.h"
 #include "opendht/default_types.h"
+#include "opendht/indexation/pht.h"

include/opendht/callbacks.h

@@ -65,12 +65,13 @@ GetCallback bindGetCb(GetCallbackSimple cb);
 using DoneCallback = std::function<void(bool success, const std::vector<std::shared_ptr<Node>>& nodes)>;
 typedef void (*DoneCallbackRaw)(bool, std::vector<std::shared_ptr<Node>>*, void *user_data);
 typedef void (*ShutdownCallbackRaw)(void *user_data);
+typedef void (*DoneCallbackSimpleRaw)(bool, void *user_data);
 
 using DoneCallbackSimple = std::function<void(bool success)>;
 
 ShutdownCallback bindShutdownCb(ShutdownCallbackRaw shutdown_cb_raw, void* user_data);
 DoneCallback bindDoneCb(DoneCallbackSimple donecb);
 DoneCallback bindDoneCb(DoneCallbackRaw raw_cb, void* user_data);
-
+DoneCallbackSimple bindDoneCbSimple(DoneCallbackSimpleRaw raw_cb, void* user_data);
 
 }

include/opendht/dhtrunner.h

@@ -156,16 +156,16 @@ public:
     void cancelListen(InfoHash h, size_t token);
     void cancelListen(InfoHash h, std::shared_future<size_t> token);
 
-    void put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb={}, bool permanent = false);
-    void put(InfoHash hash, std::shared_ptr<Value> value, DoneCallbackSimple cb, bool permanent = false) {
-        put(hash, value, bindDoneCb(cb), permanent);
+    void put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb={}, time_point created=time_point::max(), bool permanent = false);
+    void put(InfoHash hash, std::shared_ptr<Value> value, DoneCallbackSimple cb, time_point created=time_point::max(), bool permanent = false) {
+        put(hash, value, bindDoneCb(cb), created, permanent);
     }
 
-    void put(InfoHash hash, Value&& value, DoneCallback cb={}, bool permanent = false);
-    void put(InfoHash hash, Value&& value, DoneCallbackSimple cb, bool permanent = false) {
-        put(hash, std::forward<Value>(value), bindDoneCb(cb), permanent);
+    void put(InfoHash hash, Value&& value, DoneCallback cb={}, time_point created=time_point::max(), bool permanent = false);
+    void put(InfoHash hash, Value&& value, DoneCallbackSimple cb, time_point created=time_point::max(), bool permanent = false) {
+        put(hash, std::forward<Value>(value), bindDoneCb(cb), created, permanent);
     }
-    void put(const std::string& key, Value&& value, DoneCallbackSimple cb={}, bool permanent = false);
+    void put(const std::string& key, Value&& value, DoneCallbackSimple cb={}, time_point created=time_point::max(), bool permanent = false);
 
     void cancelPut(const InfoHash& h, const Value::Id& id);
 

include/opendht/indexation/pht.h

@@ -0,0 +1,514 @@
+#pragma once
+
+#include <string>
+#include <vector>
+#include <memory>
+#include <map>
+#include <functional>
+#include <stdexcept>
+#include <bitset>
+#include <iostream>
+#include <sstream>
+
+#include "../value.h"
+#include "../dhtrunner.h"
+#include "../infohash.h"
+
+namespace dht {
+namespace indexation {
+
+/*!
+ * @class   Prefix
+ * @brief   A blob structure which prefixes a Key in the PHT.
+ * @details
+ * Since the PHT structure is a "trie", every node in this structure have a
+ * label which is defined by the path from the root of the trie to the node. If
+ * the node in question is a leaf, *the label is a prefix of all the keys
+ * contained in the leaf*.
+ */
+struct Prefix {
+    Prefix() {}
+    Prefix(InfoHash h) : size_(h.size() * 8), content_(h.begin(), h.end()) { }
+    Prefix(const Blob& d, const Blob& f={}) : size_(d.size()*8), flags_(f), content_(d) { }
+
+    Prefix(const Prefix& p, size_t first) :
+        size_(std::min(first, p.content_.size()*8)),
+        content_(Blob(p.content_.begin(), p.content_.begin()+size_/8))
+    {
+
+        auto rem = size_ % 8;
+        if ( not p.flags_.empty() ) {
+            flags_ = Blob(p.flags_.begin(), p.flags_.begin()+size_/8);
+            if (rem)
+                flags_.push_back(p.flags_[size_/8] & (0xFF << (8 - rem)));
+        }
+
+        if (rem)
+            content_.push_back(p.content_[size_/8] & (0xFF << (8 - rem)));
+    }
+
+    /**
+     * Get a sub prefix of the Prefix
+     *
+     * @param len lenght of the prefix to get, could be negative
+     *            if len is negativ then you will get the prefix
+     *            of size of the previous prefix minus len
+     *
+     * @return Sub-prefix of size len or if len is negative sub-prefix of size
+     *         of prefix minus len
+     *
+     * @throw out_of_range if len is larger than size of the content
+     */
+    Prefix getPrefix(ssize_t len) const {
+        if ((size_t)std::abs(len) >= content_.size() * 8)
+            throw std::out_of_range("len larger than prefix size.");
+        if (len < 0)
+            len += size_;
+
+        return Prefix(*this, len);
+    }
+
+    /**
+     * Flags are considered as active if flag is empty or if the flag
+     * at pos 'pos' is active
+ee     *
+     * @see isActiveBit in private function
+     */
+    bool isFlagActive(size_t pos) const {
+        return flags_.empty() or isActiveBit(flags_, pos);
+    }
+
+    /**
+     * @see isActiveBit in private function
+     */
+    bool isContentBitActive(size_t pos) const {
+        return isActiveBit(content_, pos);
+    }
+
+    Prefix getFullSize() { return Prefix(*this, content_.size()*8); }
+
+    /**
+     * This methods gets the prefix of its sibling in the PHT structure.
+     *
+     * @return The prefix of this sibling.
+     */
+    Prefix getSibling() const {
+        Prefix copy = *this;
+        if ( size_ )
+            copy.swapContentBit(size_ - 1);
+
+        return copy;
+    }
+
+    InfoHash hash() const {
+        Blob copy(content_);
+        copy.push_back(size_);
+        return InfoHash::get(copy);
+    }
+
+    /**
+     * This method count total of bit in common between 2 prefix
+     *
+     * @param p1 first prefix to compared
+     * @param p2 second prefix to compared
+     * @return Lenght of the larger common prefix between both
+     */
+    static inline unsigned commonBits(const Prefix& p1, const Prefix& p2) {
+        unsigned i, j;
+        uint8_t x;
+        auto longest_prefix_size = std::min(p1.size_, p2.size_);
+
+        for (i = 0; i < longest_prefix_size; i++) {
+            if (p1.content_.data()[i] != p2.content_.data()[i]
+                or not p1.isFlagActive(i)
+                or not p2.isFlagActive(i) ) {
+
+                    break;
+            }
+        }
+
+        if (i == longest_prefix_size)
+            return 8*longest_prefix_size;
+
+        x = p1.content_.data()[i] ^ p2.content_.data()[i];
+
+        j = 0;
+        while ((x & 0x80) == 0) {
+            x <<= 1;
+            j++;
+        }
+
+        return 8 * i + j;
+    }
+
+    /**
+     * @see doc of swap private function
+     */
+    void swapContentBit(size_t bit) {
+        swapBit(content_, bit);
+    }
+
+    /**
+     * @see doc of swap private function
+     */
+    void swapFlagBit(size_t bit) {
+        swapBit(flags_, bit);
+    }
+
+    /**
+     * @see doc of addPadding private function
+     */
+    void addPaddingContent(size_t size) {
+        content_ = addPadding(content_, size);
+    }
+
+    void updateFlags() {
+        /* Fill first known bit */
+        auto csize = size_ - flags_.size() * 8;
+        while(csize >= 8) {
+            flags_.push_back(0xFF);
+            csize -= 8;
+        }
+
+        /* if needed fill remaining bit */
+        if ( csize )
+            flags_.push_back(0xFF << (8 - csize));
+
+        /* Complet vector space missing */
+        for ( auto i = flags_.size(); i < content_.size(); i++ )
+            flags_.push_back(0xFF);
+    }
+
+    std::string toString() const;
+
+    size_t size_ {0};
+
+    /* Will contain flags according to content_.
+       If flags_[i] == 0, then content_[i] is unknown
+       else if flags_[i] == 1, then content_[i] is known */
+    Blob flags_ {};
+    Blob content_ {};
+
+private:
+
+    /**
+     * Add a padding to the input blob
+     *
+     * @param toP  : Prefix where to add a padding
+     * @param size : Final size of the prefix with padding
+     *
+     * @return Copy of the input Blob but with a padding
+     */
+    Blob addPadding(Blob toP, size_t size) {
+        Blob copy = toP;
+        for ( auto i = copy.size(); i < size; i++ )
+            copy.push_back(0);
+
+        swapBit(copy, size_ + 1);
+        return copy;
+    }
+
+    /**
+     * Check if the bit a pos 'pos' is active, i.e. equal to 1
+     *
+     * @param b   : Blob to check
+     * @param pos : Position to check
+     *
+     * @return true if the bit is equal to 1, false otherwise
+     *
+     * @throw out_of_range if bit is superior to blob size * 8
+     */
+    bool isActiveBit(const Blob &b, size_t pos) const {
+        if ( pos >= content_.size() * 8 )
+            throw std::out_of_range("Can't detect active bit at pos, pos larger than prefix size or empty prefix");
+
+        return ((b[pos / 8] >> (7 - (pos % 8)) ) & 1) == 1;
+    }
+
+    /**
+     * Swap bit at position bit [from 0 to 1 and vice-versa]
+     *
+     * @param b   : Blob to swap
+     * @param bit : Bit to swap on b
+     *
+     * @return the input prefix with the bit at pos 'bit' swapped
+     *
+     * @throw out_of_range if bit is superior to blob size * 8
+     */
+    void swapBit(Blob &b, size_t bit) {
+        if ( bit >= b.size() * 8 )
+            throw std::out_of_range("bit larger than prefix size.");
+
+        size_t offset_bit = (8 - bit) % 8;
+        b[bit / 8] ^= (1 << offset_bit);
+    }
+};
+
+using Value = std::pair<InfoHash, dht::Value::Id>;
+struct IndexEntry : public dht::Value::Serializable<IndexEntry> {
+    static const ValueType TYPE;
+
+    virtual void unpackValue(const dht::Value& v) {
+        Serializable<IndexEntry>::unpackValue(v);
+        name = v.user_type;
+    }
+
+    virtual dht::Value packValue() const {
+        auto pack = Serializable<IndexEntry>::packValue();
+        pack.user_type = name;
+        return pack;
+    }
+
+    Blob prefix;
+    Value value;
+    std::string name;
+    MSGPACK_DEFINE_MAP(prefix, value);
+};
+
+class Pht {
+    static constexpr const char* INVALID_KEY = "Key does not match the PHT key spec.";
+
+    /* Prefixes the user_type for all dht values put on the DHT */
+    static constexpr const char* INDEX_PREFIX = "index.pht.";
+
+public:
+
+    /* This is the maximum number of entries per node. This parameter is
+     * critical and influences the traffic a lot during a lookup operation.
+     */
+    static constexpr const size_t MAX_NODE_ENTRY_COUNT {16};
+
+    /* A key for a an index entry */
+    using Key = std::map<std::string, Blob>;
+
+    /* Specifications of the keys. It defines the number, the length and the
+     * serialization order of fields. */
+    using KeySpec = std::map<std::string, size_t>;
+    using LookupCallback = std::function<void(std::vector<std::shared_ptr<Value>>& values, Prefix p)>;
+
+    typedef void (*LookupCallbackRaw)(std::vector<std::shared_ptr<Value>>* values, Prefix* p, void *user_data);
+    static LookupCallback
+    bindLookupCb(LookupCallbackRaw raw_cb, void* user_data) {
+        if (not raw_cb) return {};
+        return [=](std::vector<std::shared_ptr<Value>>& values, Prefix p) {
+            raw_cb((std::vector<std::shared_ptr<Value>>*) &values, (Prefix*) &p, user_data);
+        };
+    }
+    using LookupCallbackSimple = std::function<void(std::vector<std::shared_ptr<Value>>& values)>;
+    typedef void (*LookupCallbackSimpleRaw)(std::vector<std::shared_ptr<Value>>* values, void *user_data);
+    static LookupCallbackSimple
+    bindLookupCbSimple(LookupCallbackSimpleRaw raw_cb, void* user_data) {
+        if (not raw_cb) return {};
+        return [=](std::vector<std::shared_ptr<Value>>& values) {
+            raw_cb((std::vector<std::shared_ptr<Value>>*) &values, user_data);
+        };
+    }
+
+    Pht(std::string name, KeySpec k_spec, std::shared_ptr<DhtRunner> dht)
+        : name_(INDEX_PREFIX + name), canary_(name_ + ".canary"), keySpec_(k_spec), dht_(dht) {}
+
+    virtual ~Pht () { }
+
+    /**
+     * Lookup a key for a value.
+     */
+    void lookup(Key k, LookupCallback cb = {}, DoneCallbackSimple done_cb = {}, bool exact_match = true);
+    void lookup(Key k, LookupCallbackSimple cb = {}, DoneCallbackSimple done_cb = {}, bool exact_match = true)
+    {
+        lookup(k, [=](std::vector<std::shared_ptr<Value>>& values, Prefix) { cb(values); }, done_cb, exact_match);
+    }
+
+    /**
+     * Wrapper function which call the private one.
+     *
+     * @param k : Key to insert [i.e map of string, blob]
+     * @param v : Value to insert
+     * @param done_cb : Callbakc which going to be call when all the insert is done
+     */
+    void insert(Key k, Value v, DoneCallbackSimple done_cb = {}) {
+        Prefix p = linearize(k);
+
+        auto lo = std::make_shared<int>(0);
+        auto hi = std::make_shared<int>(p.size_);
+
+        IndexEntry entry;
+        entry.value = v;
+        entry.prefix = p.content_;
+        entry.name = name_;
+
+        Pht::insert(p, entry, lo, hi, clock::now(), true, done_cb);
+    }
+
+private:
+
+    /**
+     * Insert function which really insert onto the pht
+     *
+     * @param kp          : Prefix to insert (linearize the the key)
+     * @param entry       : Entry created from the value
+     * @param lo          : Lowest point to start in the prefix
+     * @param hi          : Highest point to end in the prefix
+     * @param time_p      : Timepoint to use for the insertion into the dht (must be < now)
+     * @param check_split : If this flag is true then the algoritm will not use the merge algorithm
+     * @param done_cb     : Callback to call when the insert is done
+     */
+
+    void insert(Prefix kp, IndexEntry entry, std::shared_ptr<int> lo, std::shared_ptr<int> hi, time_point time_p,
+                bool check_split, DoneCallbackSimple done_cb = {});
+
+    class Cache {
+    public:
+        /**
+         * Insert all needed node into the tree according to a prefix
+         * @param p : Prefix that we need to insert
+         */
+        void insert(const Prefix& p);
+
+        /**
+         * Lookup into the tree to return the maximum prefix length in the cache tree
+         *
+         * @param p : Prefix that we are looking for
+         *
+         * @return  : The size of the longest prefix known in the cache between 0 and p.size_
+         */
+
+        int lookup(const Prefix& p);
+
+    private:
+        static constexpr const size_t MAX_ELEMENT {1024};
+        static constexpr const std::chrono::minutes NODE_EXPIRE_TIME {5};
+
+        struct Node {
+            time_point last_reply;           /* Made the assocation between leaves and leaves multimap */
+            std::shared_ptr<Node> parent;    /* Share_ptr to the parent, it allow the self destruction of tree */
+            std::weak_ptr<Node> left_child;  /* Left child, for bit equal to 1 */
+            std::weak_ptr<Node> right_child; /* Right child, for bit equal to 0 */
+        };
+
+        std::weak_ptr<Node> root_;                         /* Root of the tree */
+
+        /**
+         * This mutlimap contains all prefix insert in the tree in time order
+         * We could then delete the last one if there is too much node
+         * The tree will self destroy is branch ( thanks to share_ptr )
+         */
+        std::multimap<time_point, std::shared_ptr<Node>> leaves_;
+    };
+
+    /* Callback used for insert value by using the pht */
+    using RealInsertCallback = std::function<void(std::shared_ptr<Prefix> p, IndexEntry entry)>;
+    using LookupCallbackWrapper = std::function<void(std::vector<std::shared_ptr<IndexEntry>>& values, Prefix p)>;
+
+    /**
+     * Performs a step in the lookup operation. Each steps are performed
+     * asynchronously.
+     *
+     * @param k          : Prefix on which the lookup is performed
+     * @param lo         : lowest bound on the prefix (where to start)
+     * @param hi         : highest bound on the prefix (where to stop)
+     * @param vals       : Shared ptr to a vector of IndexEntry (going to contains all values found)
+     * @param cb         : Callback to use at the end of the lookupStep (call on the value of vals)
+     * @param done_cb    : Callback at the end of the lookupStep
+     * @param max_common_prefix_len: used in the inexacte lookup match case, indicate the longest common prefix found
+     * @param start      : If start is set then lo and hi will be ignore for the first step, if the step fail lo and hi will be used
+     * @param all_values : If all value is true then all value met during the lookupstep will be in the vector vals
+     */
+    void lookupStep(Prefix k, std::shared_ptr<int> lo, std::shared_ptr<int> hi,
+            std::shared_ptr<std::vector<std::shared_ptr<IndexEntry>>> vals,
+            LookupCallbackWrapper cb, DoneCallbackSimple done_cb,
+            std::shared_ptr<unsigned> max_common_prefix_len,
+            int start = -1, bool all_values = false);
+
+    /**
+     * Apply the zcurve algorithm on the list of input prefix
+     *
+     * @param all_prefix : Vector of prefix to interleave
+     *
+     * @return The output prefix where all flags and content are interleaves
+     */
+    Prefix zcurve(const std::vector<Prefix>& all_prefix) const;
+
+    /**
+     * Linearizes the key into a unidimensional key. A pht only takes
+     * unidimensional key.
+     *
+     * @param Key  The initial key.
+     *
+     * @return the prefix of the linearized key.
+     */
+    virtual Prefix linearize(Key k) const;
+
+    /**
+     * Looking where to put the data cause if there is free space on the node
+     * above then this node will became the real leave.
+     *
+     * @param p       Share_ptr on the Prefix to check
+     * @param entry   The entry to put at the prefix p
+     * @param end_cb  Callback to use at the end of counting
+     */
+    void getRealPrefix(std::shared_ptr<Prefix> p, IndexEntry entry, RealInsertCallback end_cb );
+
+    /**
+     * Looking where to put the data cause if there is free space on the node
+     * above then this node will became the real leave.
+     *
+     * @param p       Share_ptr on the Prefix to check
+     * @param entry   The entry to put at the prefix p
+     * @param end_cb  Callback to use at the end of counting
+     */
+    void checkPhtUpdate(Prefix p, IndexEntry entry, time_point time_p);
+
+    /**
+     * Search for the split location by comparing 'compared' to all values in vals.
+     *
+     * @param compared : Value which going to be compared
+     * @param vals     : The vector of values to compare with comapred
+     * @return position compared diverge from all others
+     */
+    size_t foundSplitLocation(Prefix compared, std::shared_ptr<std::vector<std::shared_ptr<IndexEntry>>> vals) {
+        for ( size_t i = 0; i < compared.content_.size() * 8 - 1; i++ )
+            for ( auto const& v : *vals)
+                if ( Prefix(v->prefix).isContentBitActive(i) != compared.isContentBitActive(i) )
+                    return i + 1;
+
+        return compared.content_.size() * 8 - 1;
+    }
+
+    /**
+     * Put canary from the split point until the last known canary and add the prefix at the good place
+     *
+     * @param insert : Prefix to insertm but also prefix which going to check where we need to split
+     * @param vals   : Vector of vals for the comparaison
+     * @param entry  : Entry to put on the pht
+     * @param end_cb : Callback to apply to the insert prefi (here does the insert)
+     */
+    void split(Prefix insert, std::shared_ptr<std::vector<std::shared_ptr<IndexEntry>>> vals, IndexEntry entry, RealInsertCallback end_cb);
+
+    /**
+     * Tells if the key is valid according to the key spec.
+     */
+    bool validKey(const Key& k) const {
+        return k.size() == keySpec_.size() and
+            std::equal(k.begin(), k.end(), keySpec_.begin(),
+                [&](const Key::value_type& key, const KeySpec::value_type& key_spec) {
+                    return key.first == key_spec.first and key.second.size() <= key_spec.second;
+                }
+            );
+    }
+
+    /**
+     * Updates the canary token on the node responsible for the specified
+     * Prefix.
+     */
+    void updateCanary(Prefix p);
+
+    const std::string name_;
+    const std::string canary_;
+    const KeySpec keySpec_;
+    Cache cache_;
+    std::shared_ptr<DhtRunner> dht_;
+};
+
+} /* indexation  */
+} /* dht */
+

python/Makefile.am

@@ -13,7 +13,7 @@ clean-local:
 	rm -rf $(builddir)/build $(builddir)/*.so $(PYTHON_INSTALL_RECORD)
 
 install-exec-local:
-	$(PYTHON) setup.py install --root=$(DESTDIR) --record $(PYTHON_INSTALL_RECORD)
+	$(PYTHON) setup.py install --root=$(DESTDIR)/ --record $(PYTHON_INSTALL_RECORD)
 	rm -rf $(builddir)/build
 
 if HAVE_PIP

python/opendht.pyx

@@ -40,6 +40,16 @@ cimport opendht_cpp as cpp
 
 import threading
 
+cdef inline void lookup_callback(cpp.vector[cpp.shared_ptr[cpp.IndexValue]]* values, cpp.Prefix* p, void *user_data) with gil:
+    cbs = <object>user_data
+    if 'lookup' in cbs and cbs['lookup']:
+        vals = []
+        for val in deref(values):
+            v = IndexValue()
+            v._value = val
+            vals.append(v)
+        cbs['lookup'](vals, p.toString())
+
 cdef inline void shutdown_callback(void* user_data) with gil:
     cbs = <object>user_data
     if 'shutdown' in cbs and cbs['shutdown']:
@@ -64,6 +74,12 @@ cdef inline void done_callback(bool done, cpp.vector[cpp.shared_ptr[cpp.Node]]*
         cbs['done'](done, node_ids)
     ref.Py_DECREF(cbs)
 
+cdef inline void done_callback_simple(bool done, void *user_data) with gil:
+    cbs = <object>user_data
+    if 'done' in cbs and cbs['done']:
+        cbs['done'](done)
+    ref.Py_DECREF(cbs)
+
 cdef class _WithID(object):
     def __repr__(self):
         return "<%s '%s'>" % (self.__class__.__name__, str(self))
@@ -72,7 +88,7 @@ cdef class _WithID(object):
 
 cdef class InfoHash(_WithID):
     cdef cpp.InfoHash _infohash
-    def __init__(self, bytes str=b''):
+    def __cinit__(self, bytes str=b''):
         self._infohash = cpp.InfoHash(str)
     def __bool__(InfoHash self):
         return not (self._infohash == cpp.InfoHash())
@@ -254,49 +270,49 @@ cdef class DhtConfig(object):
         self._config.dht_config.node_config.node_id = id._infohash
 
 cdef class DhtRunner(_WithID):
-    cdef cpp.DhtRunner* thisptr
+    cdef cpp.shared_ptr[cpp.DhtRunner] thisptr
     def __cinit__(self):
-        self.thisptr = new cpp.DhtRunner()
+        self.thisptr.reset(new cpp.DhtRunner())
     def getId(self):
         h = InfoHash()
-        h._infohash = self.thisptr.getId()
+        h._infohash = self.thisptr.get().getId()
         return h
     def getNodeId(self):
-        return self.thisptr.getNodeId().toString()
+        return self.thisptr.get().getNodeId().toString()
     def bootstrap(self, str host, str port):
-        self.thisptr.bootstrap(host.encode(), port.encode())
+        self.thisptr.get().bootstrap(host.encode(), port.encode())
     def run(self, Identity id=None, is_bootstrap=False, cpp.in_port_t port=0, str ipv4="", str ipv6="", DhtConfig config=DhtConfig()):
         if id:
             config.setIdentity(id)
         if ipv4 or ipv6:
             bind4 = ipv4.encode() if ipv4 else b''
             bind6 = ipv6.encode() if ipv6 else b''
-            self.thisptr.run(bind4, bind6, str(port).encode(), config._config)
+            self.thisptr.get().run(bind4, bind6, str(port).encode(), config._config)
         else:
-            self.thisptr.run(port, config._config)
+            self.thisptr.get().run(port, config._config)
     def join(self):
-        self.thisptr.join()
+        self.thisptr.get().join()
     def shutdown(self, shutdown_cb=None):
         cb_obj = {'shutdown':shutdown_cb}
         ref.Py_INCREF(cb_obj)
-        self.thisptr.shutdown(cpp.bindShutdownCb(shutdown_callback, <void*>cb_obj))
+        self.thisptr.get().shutdown(cpp.bindShutdownCb(shutdown_callback, <void*>cb_obj))
     def enableLogging(self):
-        cpp.enableLogging(self.thisptr[0])
+        cpp.enableLogging(self.thisptr.get()[0])
     def disableLogging(self):
-        cpp.disableLogging(self.thisptr[0])
+        cpp.disableLogging(self.thisptr.get()[0])
     def enableFileLogging(self, str path):
-        cpp.enableFileLogging(self.thisptr[0], path)
+        cpp.enableFileLogging(self.thisptr.get()[0], path)
     def isRunning(self):
-        return self.thisptr.isRunning()
+        return self.thisptr.get().isRunning()
     def getStorageLog(self):
-        return self.thisptr.getStorageLog().decode()
+        return self.thisptr.get().getStorageLog().decode()
     def getRoutingTablesLog(self, cpp.sa_family_t af):
-        return self.thisptr.getRoutingTablesLog(af).decode()
+        return self.thisptr.get().getRoutingTablesLog(af).decode()
     def getSearchesLog(self, cpp.sa_family_t af):
-        return self.thisptr.getSearchesLog(af).decode()
+        return self.thisptr.get().getSearchesLog(af).decode()
     def getNodeMessageStats(self):
         stats = []
-        cdef cpp.vector[unsigned] res = self.thisptr.getNodeMessageStats(False)
+        cdef cpp.vector[unsigned] res = self.thisptr.get().getNodeMessageStats(False)
         for n in res:
             stats.append(n)
         return stats
@@ -304,14 +320,16 @@ cdef class DhtRunner(_WithID):
     def get(self, InfoHash key, get_cb=None, done_cb=None):
         """Retreive values associated with a key on the DHT.
 
-        key -- the key for which to search
-        get_cb -- is set, makes the operation non-blocking. Called when a value is found on the DHT.
-        done_cb -- optional callback used when get_cb is set. Called when the operation is completed.
+        key     -- the key for which to search
+        get_cb  -- is set, makes the operation non-blocking. Called when a value
+                   is found on the DHT.
+        done_cb -- optional callback used when get_cb is set. Called when the
+                   operation is completed.
         """
         if get_cb:
             cb_obj = {'get':get_cb, 'done':done_cb}
             ref.Py_INCREF(cb_obj)
-            self.thisptr.get(key._infohash, cpp.bindGetCb(get_callback, <void*>cb_obj), cpp.bindDoneCb(done_callback, <void*>cb_obj))
+            self.thisptr.get().get(key._infohash, cpp.bindGetCb(get_callback, <void*>cb_obj), cpp.bindDoneCb(done_callback, <void*>cb_obj))
         else:
             lock = threading.Condition()
             pending = 0
@@ -334,14 +352,14 @@ cdef class DhtRunner(_WithID):
     def put(self, InfoHash key, Value val, done_cb=None):
         """Publish a new value on the DHT at key.
 
-        key -- the DHT key where to put the value
-        val -- the value to put on the DHT
+        key     -- the DHT key where to put the value
+        val     -- the value to put on the DHT
         done_cb -- optional callback called when the operation is completed.
         """
         if done_cb:
             cb_obj = {'done':done_cb}
             ref.Py_INCREF(cb_obj)
-            self.thisptr.put(key._infohash, val._value, cpp.bindDoneCb(done_callback, <void*>cb_obj))
+            self.thisptr.get().put(key._infohash, val._value, cpp.bindDoneCb(done_callback, <void*>cb_obj))
         else:
             lock = threading.Condition()
             pending = 0
@@ -365,9 +383,74 @@ cdef class DhtRunner(_WithID):
         t._cb['cb'] = cb_obj
         # avoid the callback being destructed if the token is destroyed
         ref.Py_INCREF(cb_obj)
-        t._t = self.thisptr.listen(t._h, cpp.bindGetCb(get_callback, <void*>cb_obj)).share()
+        t._t = self.thisptr.get().listen(t._h, cpp.bindGetCb(get_callback, <void*>cb_obj)).share()
         return t
     def cancelListen(self, ListenToken token):
-        self.thisptr.cancelListen(token._h, token._t)
+        self.thisptr.get().cancelListen(token._h, token._t)
         ref.Py_DECREF(<object>token._cb['cb'])
         # fixme: not thread safe
+
+cdef class IndexValue(object):
+    cdef cpp.shared_ptr[cpp.IndexValue] _value
+    def __init__(self, InfoHash h=None, cpp.uint64_t vid=0):
+       cdef cpp.InfoHash hh = h._infohash
+       self._value.reset(new cpp.IndexValue(hh, vid))
+    def __str__(self):
+        return "(" + self.getKey().toString().decode() +", "+ str(self.getValueId()) +")"
+    def getKey(self):
+        h = InfoHash()
+        h._infohash = self._value.get().first
+        return h
+    def getValueId(self):
+        return self._value.get().second
+
+cdef class Pht(object):
+    cdef cpp.Pht* thisptr
+    def __cinit__(self, bytes name, key_spec, DhtRunner dht):
+        cdef cpp.IndexKeySpec cpp_key_spec
+        for kk, size in key_spec.items():
+            cpp_key_spec[bytes(kk, 'utf-8')] = size
+        self.thisptr = new cpp.Pht(name, cpp_key_spec, dht.thisptr)
+    property MAX_NODE_ENTRY_COUNT:
+        def __get__(self):
+            return cpp.PHT_MAX_NODE_ENTRY_COUNT
+    def lookup(self, key, lookup_cb=None, done_cb=None):
+        """Query the Index with a specified key.
+
+        key       -- the key for to the entry in the index.
+        lookup_cb -- function called when the operation is completed. This
+                     function takes a list of IndexValue objects and a string
+                     representation of the prefix where the value was indexed in
+                     the PHT.
+        """
+        cb_obj = {'lookup':lookup_cb, 'done':done_cb} # TODO: donecallback is to be removed
+        ref.Py_INCREF(cb_obj)
+        cdef cpp.IndexKey cppk
+        for kk, v in key.items():
+            cppk[bytes(kk, 'utf-8')] = bytes(v)
+        self.thisptr.lookup(
+                cppk,
+                cpp.Pht.bindLookupCb(lookup_callback, <void*>cb_obj),
+                cpp.bindDoneCbSimple(done_callback_simple, <void*>cb_obj)
+        )
+    def insert(self, key, IndexValue value, done_cb=None):
+        """Add an index entry to the Index.
+
+        key     -- the key for to the entry in the index.
+        value   -- an IndexValue object describing the indexed value.
+        done_cb -- Called when the operation is completed.
+        """
+        cb_obj = {'done':done_cb}
+        ref.Py_INCREF(cb_obj)
+        cdef cpp.IndexKey cppk
+        for kk, v in key.items():
+            cppk[bytes(kk, 'utf-8')] = bytes(v)
+        cdef cpp.IndexValue val
+        val.first = (<InfoHash>value.getKey())._infohash
+        val.second = value.getValueId()
+        self.thisptr.insert(
+                cppk,
+                val,
+                cpp.bindDoneCbSimple(done_callback_simple, <void*>cb_obj)
+        )
+

python/opendht_cpp.pxd

@@ -20,6 +20,7 @@ from libcpp cimport bool
 from libcpp.string cimport string
 from libcpp.vector cimport vector
 from libcpp.utility cimport pair
+from libcpp.map cimport map
 
 ctypedef uint16_t in_port_t
 ctypedef unsigned short int sa_family_t;
@@ -27,6 +28,7 @@ ctypedef unsigned short int sa_family_t;
 cdef extern from "<memory>" namespace "std" nogil:
     cdef cppclass shared_ptr[T]:
         shared_ptr() except +
+        shared_ptr(T*) except +
         T* get()
         T operator*()
         void reset(T*)
@@ -103,19 +105,21 @@ cdef extern from "opendht/callbacks.h" namespace "dht":
     ctypedef void (*ShutdownCallbackRaw)(void *user_data)
     ctypedef bool (*GetCallbackRaw)(shared_ptr[Value] values, void *user_data)
     ctypedef void (*DoneCallbackRaw)(bool done, vector[shared_ptr[Node]]* nodes, void *user_data)
+    ctypedef void (*DoneCallbackSimpleRaw)(bool done, void *user_data)
 
     cppclass ShutdownCallback:
         ShutdownCallback() except +
     cppclass GetCallback:
         GetCallback() except +
-        #GetCallback(GetCallbackRaw cb, void *user_data) except +
     cppclass DoneCallback:
         DoneCallback() except +
-        #DoneCallback(DoneCallbackRaw, void *user_data) except +
+    cppclass DoneCallbackSimple:
+        DoneCallbackSimple() except +
 
     cdef ShutdownCallback bindShutdownCb(ShutdownCallbackRaw cb, void *user_data)
     cdef GetCallback bindGetCb(GetCallbackRaw cb, void *user_data)
     cdef DoneCallback bindDoneCb(DoneCallbackRaw cb, void *user_data)
+    cdef DoneCallbackSimple bindDoneCbSimple(DoneCallbackSimpleRaw cb, void *user_data)
 
     cppclass Config:
         InfoHash node_id
@@ -156,3 +160,22 @@ cdef extern from "opendht/log.h" namespace "dht::log":
     void disableLogging(DhtRunner& dht)
     void enableFileLogging(DhtRunner& dht, const string& path)
 
+cdef extern from "opendht/indexation/pht.h" namespace "dht::indexation":
+    size_t PHT_MAX_NODE_ENTRY_COUNT "dht::indexation::Pht::MAX_NODE_ENTRY_COUNT"
+    cdef cppclass Prefix:
+        Prefix() except +
+        Prefix(vector[uint8_t]) except +
+        string toString() const
+    ctypedef pair[InfoHash, uint64_t] IndexValue "dht::indexation::Value"
+    ctypedef map[string, vector[uint8_t]] IndexKey "dht::indexation::Pht::Key"
+    ctypedef map[string, uint32_t] IndexKeySpec "dht::indexation::Pht::KeySpec"
+    ctypedef void (*LookupCallbackRaw)(vector[shared_ptr[IndexValue]]* values, Prefix* p, void* user_data);
+    cdef cppclass Pht:
+        cppclass LookupCallback:
+            LookupCallback() except +
+        Pht(string, IndexKeySpec, shared_ptr[DhtRunner]) except +
+        void lookup(IndexKey k, LookupCallback cb, DoneCallbackSimple doneCb);
+        void insert(IndexKey k, IndexValue v, DoneCallbackSimple cb)
+        @staticmethod
+        LookupCallback bindLookupCb(LookupCallbackRaw cb, void *user_data)
+

python/tools/benchmark.py

@@ -27,7 +27,7 @@ import random
 
 from dht.network import DhtNetwork
 from dht.network import DhtNetworkSubProcess
-from dht.tests import PerformanceTest, PersistenceTest
+from dht.tests import PerformanceTest, PersistenceTest, PhtTest
 from dht import virtual_network_builder
 from dht import network as dhtnetwork
 
@@ -157,6 +157,11 @@ if __name__ == '__main__':
     featureArgs = parser.add_mutually_exclusive_group(required=True)
     featureArgs.add_argument('--performance', action='store_true', default=False,
             help='Launches performance benchmark test. Available args for "-t" are: gets.')
+    featureArgs.add_argument('--pht', action='store_true', default=False,
+            help='Launches PHT benchmark test. '\
+                    'Available args for "-t" are: insert. '\
+                    'Timer available by adding "timer" to "-o" args'\
+                    'Use "-m" option for fixing number of keys to create during the test.')
     featureArgs.add_argument('--data-persistence', action='store_true', default=0,
             help='Launches data persistence benchmark test. '\
                  'Available args for "-t" are: delete, replace, mult_time. '\
@@ -206,6 +211,10 @@ if __name__ == '__main__':
             PerformanceTest(args.test, wb, test_opt).run()
         elif args.data_persistence:
             PersistenceTest(args.test, wb, test_opt).run()
+        elif args.pht:
+            if args.m:
+                test_opt.update({ 'num_keys' : args.m })
+            PhtTest(args.test, wb, test_opt).run()
 
     except Exception as e:
         print(e)

python/tools/dht/tests.py

@@ -10,10 +10,16 @@ import string
 import time
 import subprocess
 import re
+import collections
+
+from matplotlib.ticker import FuncFormatter
+import math
 
 import numpy as np
 import matplotlib.pyplot as plt
-from matplotlib.ticker import FuncFormatter
+import networkx as nx
+from networkx.drawing.nx_agraph import graphviz_layout
+
 
 from opendht import *
 from dht.network import DhtNetwork, DhtNetworkSubProcess
@@ -30,6 +36,24 @@ Mbit_format = FuncFormatter(lambda x, pos: '%1.1f' % (x*1024**-2) + 'Mb')
 def random_hash():
     return InfoHash(''.join(random.SystemRandom().choice(string.hexdigits) for _ in range(40)).encode())
 
+def timer(f, *args):
+    """
+    Start a timer which count time taken for execute function f
+
+    @param f : Function to time
+    @type  f : function
+
+    @param args : Arguments of the function f
+    @type  args : list
+
+    @rtype : timer
+    @return : Time taken by the function f
+    """
+    start = time.time()
+    f(*args)
+
+    return time.time() - start
+
 def reset_before_test(featureTestMethod):
     """
     This is a decorator for all test methods needing reset().
@@ -125,13 +149,13 @@ def iftop_traffic_data(ifname, interval=2, rate_type='send_receive'):
 ###########
 
 class FeatureTest(object):
-    done = 0
-    lock = None
-
     """
     This is a base test.
     """
 
+    done = 0
+    lock = None
+
     def __init__(self, test, workbench):
         """
         @param test: The test string indicating the test to run. This string is
@@ -143,6 +167,7 @@ class FeatureTest(object):
         """
         self._test = test
         self._workbench = workbench
+        self._bootstrap = self._workbench.get_bootstrap()
 
     def _reset(self):
         """
@@ -156,7 +181,155 @@ class FeatureTest(object):
     def run(self):
         raise NotImplementedError('This method must be implemented.')
 
+##################################
+#               PHT              #
+##################################
 
+class PhtTest(FeatureTest):
+    """TODO
+    """
+
+    indexEntries = None
+    prefix       = None
+    key          = None
+
+    def __init__(self, test, workbench, opts):
+        """
+        @param test: is one of the following:
+                     - 'insert': indexes a considerable amount of data in
+                       the PHT structure.
+                       TODO
+        @type  test: string
+
+        @param opts: Dictionnary containing options for the test. Allowed
+                     options are:
+                     - 'num_keys': this specifies the number of keys to insert
+                                   in the PHT during the test.
+        @type  opts: dict
+        """
+        super(PhtTest, self).__init__(test, workbench)
+        self._num_keys = opts['num_keys'] if 'num_keys' in opts else 32
+        self._timer = True if 'timer' in opts else False
+
+    def _reset(self):
+        super(PhtTest, self)._reset()
+        PhtTest.indexEntries = []
+
+    @staticmethod
+    def lookupCb(vals, prefix):
+        PhtTest.indexEntries = list(vals)
+        PhtTest.prefix = prefix.decode()
+        DhtNetwork.log('Index name: <todo>')
+        DhtNetwork.log('Leaf prefix:', prefix)
+        for v in vals:
+            DhtNetwork.log('[ENTRY]:', v)
+
+    @staticmethod
+    def lookupDoneCb(ok):
+        DhtNetwork.log('[LOOKUP]:', PhtTest.key, "--", "success!" if ok else "Fail...")
+        with FeatureTest.lock:
+            FeatureTest.lock.notify()
+
+    @staticmethod
+    def insertDoneCb(ok):
+        DhtNetwork.log('[INSERT]:', PhtTest.key, "--", "success!" if ok else "Fail...")
+        with FeatureTest.lock:
+            FeatureTest.lock.notify()
+
+    @staticmethod
+    def drawTrie(trie_dict):
+        """
+        Draws the trie structure of the PHT from dictionnary.
+
+        @param trie_dict: Dictionnary of index entries (prefix -> entry).
+        @type  trie_dict: dict
+        """
+        prefixes = list(trie_dict.keys())
+        if len(prefixes) == 0:
+            return
+
+        edges = list([])
+        for prefix in prefixes:
+            for i in range(-1, len(prefix)-1):
+                u = prefix[:i+1]
+                x = ("." if i == -1 else u, u+"0")
+                y = ("." if i == -1 else u, u+"1")
+                if x not in edges:
+                    edges.append(x)
+                if y not in edges:
+                    edges.append(y)
+
+        # TODO: use a binary tree position layout...
+        #   UPDATE : In a better way [change lib]
+        G = nx.Graph(sorted(edges, key=lambda x: len(x[0])))
+        plt.title("PHT: Tree")
+        pos=graphviz_layout(G,prog='dot')
+        nx.draw(G, pos, with_labels=True, node_color='white')
+        plt.show()
+
+    def run(self):
+        try:
+            if self._test == 'insert':
+                self._insertTest()
+        except Exception as e:
+            print(e)
+        finally:
+            self._bootstrap.resize(1)
+
+    ###########
+    #  Tests  #
+    ###########
+
+    @reset_before_test
+    def _insertTest(self):
+        """TODO: Docstring for _massIndexTest.
+        """
+        bootstrap = self._bootstrap
+        bootstrap.resize(2)
+        dht = bootstrap.get(1)
+
+        NUM_DIG  = max(math.log(self._num_keys, 2)/4, 5) # at least 5 digit keys.
+        keyspec = collections.OrderedDict([('foo', NUM_DIG)])
+        pht = Pht(b'foo_index', keyspec, dht)
+
+        DhtNetwork.log('PHT has',
+                       pht.MAX_NODE_ENTRY_COUNT,
+                       'node'+ ('s' if pht.MAX_NODE_ENTRY_COUNT > 1 else ''),
+                       'per leaf bucket.')
+        keys = [{
+            [_ for _ in keyspec.keys()][0] :
+            ''.join(random.SystemRandom().choice(string.hexdigits)
+                for _ in range(NUM_DIG)).encode()
+            } for n in range(self._num_keys)]
+        all_entries = {}
+
+        # Index all entries.
+        for key in keys:
+            PhtTest.key = key
+            with FeatureTest.lock:
+                time_taken = timer(pht.insert, key, IndexValue(random_hash()), PhtTest.insertDoneCb)
+                if self._timer:
+                    DhtNetwork.log('This insert step took : ', time_taken, 'second')
+                FeatureTest.lock.wait()
+
+        time.sleep(1)
+
+        # Recover entries now that the trie is complete.
+        for key in keys:
+            PhtTest.key = key
+            with FeatureTest.lock:
+                time_taken = timer(pht.lookup, key, PhtTest.lookupCb, PhtTest.lookupDoneCb)
+                if self._timer:
+                    DhtNetwork.log('This lookup step took : ', time_taken, 'second')
+                FeatureTest.lock.wait()
+
+            all_entries[PhtTest.prefix] = [e.__str__()
+                                           for e in PhtTest.indexEntries]
+
+        for p in all_entries.keys():
+            DhtNetwork.log('All entries under prefix', p, ':')
+            DhtNetwork.log(all_entries[p])
+        PhtTest.drawTrie(all_entries)
 
 ##################################
 #               DHT              #
@@ -164,7 +337,7 @@ class FeatureTest(object):
 
 class DhtFeatureTest(FeatureTest):
     """
-    This is base test. A method run() implementation is required.
+    This is a base dht test.
     """
     #static variables used by class callbacks
     successfullTransfer = lambda lv,fv: len(lv) == len(fv)
@@ -173,7 +346,6 @@ class DhtFeatureTest(FeatureTest):
 
     def __init__(self, test, workbench):
         super(DhtFeatureTest, self).__init__(test, workbench)
-        self.bootstrap = self._workbench.get_bootstrap()
 
     def _reset(self):
         super(DhtFeatureTest, self)._reset()
@@ -234,7 +406,6 @@ class DhtFeatureTest(FeatureTest):
                 for n in DhtFeatureTest.foreignNodes:
                     nodes.add(n)
 
-
 class PersistenceTest(DhtFeatureTest):
     """
     This tests persistence of data on the network.
@@ -292,8 +463,8 @@ class PersistenceTest(DhtFeatureTest):
             config.setNodeId(InfoHash(_hash_str.encode()))
             n = DhtRunner()
             n.run(config=config)
-            n.bootstrap(self.bootstrap.ip4,
-                        str(self.bootstrap.port))
+            n.bootstrap(self._bootstrap.ip4,
+                        str(self._bootstrap.port))
             DhtNetwork.log('Node','['+_hash_str+']',
                            'started around', _hash.toString().decode()
                            if n.isRunning() else
@@ -302,7 +473,7 @@ class PersistenceTest(DhtFeatureTest):
             trigger_nodes.append(n)
 
     def _result(self, local_values, new_nodes):
-        bootstrap = self.bootstrap
+        bootstrap = self._bootstrap
         if not DhtFeatureTest.successfullTransfer(local_values, DhtFeatureTest.foreignValues):
             DhtNetwork.log('[GET]: Only %s on %s values persisted.' %
                     (len(DhtFeatureTest.foreignValues), len(local_values)))
@@ -340,7 +511,7 @@ class PersistenceTest(DhtFeatureTest):
                 plot_fname = "traffic-plot"
                 print('plot saved to', plot_fname)
                 plt.savefig(plot_fname)
-            self.bootstrap.resize(1)
+            self._bootstrap.resize(1)
 
     ###########
     #  Tests  #
@@ -353,7 +524,7 @@ class PersistenceTest(DhtFeatureTest):
         """
         trigger_nodes = []
         wb = self._workbench
-        bootstrap = self.bootstrap
+        bootstrap = self._bootstrap
         # Value representing an ICE packet. Each ICE packet is around 1KB.
         VALUE_SIZE = 1024
         NUM_VALUES = self._num_values/wb.node_num if self._num_values else 5
@@ -430,7 +601,7 @@ class PersistenceTest(DhtFeatureTest):
         It uses Dht shutdown call from the API to gracefuly finish the nodes one
         after the other.
         """
-        bootstrap = self.bootstrap
+        bootstrap = self._bootstrap
 
         ops_count = []
 
@@ -494,7 +665,7 @@ class PersistenceTest(DhtFeatureTest):
         """
         clusters = 8
 
-        bootstrap = self.bootstrap
+        bootstrap = self._bootstrap
 
         bootstrap.resize(3)
         consumer = bootstrap.get(1)
@@ -531,14 +702,14 @@ class PersistenceTest(DhtFeatureTest):
         minutes for the nodes to trigger storage maintenance.
         """
         trigger_nodes = []
-        bootstrap = self.bootstrap
+        bootstrap = self._bootstrap
 
         N_PRODUCERS = self._num_producers if self._num_values else 16
         DP_TIMEOUT = 1
 
         hashes = []
 
-        # Generating considerable amount of values of size 1KB.
+    # Generating considerable amount of values of size 1KB.
         VALUE_SIZE = 1024
         NUM_VALUES = self._num_values if self._num_values else 50
         values = [Value(''.join(random.choice(string.hexdigits) for _ in range(VALUE_SIZE)).encode()) for _ in range(NUM_VALUES)]
@@ -612,7 +783,7 @@ class PerformanceTest(DhtFeatureTest):
         except Exception as e:
             print(e)
         finally:
-            self.bootstrap.resize(1)
+            self._bootstrap.resize(1)
 
 
     ###########
@@ -624,7 +795,7 @@ class PerformanceTest(DhtFeatureTest):
         """
         Tests for performance of the DHT doing multiple get() operation.
         """
-        bootstrap = self.bootstrap
+        bootstrap = self._bootstrap
 
         plt.ion()
 
@@ -713,8 +884,7 @@ class PerformanceTest(DhtFeatureTest):
         deleting around the target hash.
         """
 
-
-        bootstrap = self.bootstrap
+        bootstrap = self._bootstrap
 
         bootstrap.resize(3)
         consumer = bootstrap.get(1)

src/Makefile.am

@@ -20,7 +20,8 @@ libopendht_la_SOURCES = \
         crypto.cpp \
         securedht.cpp \
         dhtrunner.cpp \
-        default_types.cpp
+        default_types.cpp \
+        indexation/pht.cpp
 
 if WIN32
 libopendht_la_SOURCES += rng.cpp
@@ -31,7 +32,7 @@ nobase_include_HEADERS = \
         ../include/opendht/dht.h \
         ../include/opendht/callbacks.h \
         ../include/opendht/node_cache.h \
-        ../include/opendht/routing_table.h \        
+        ../include/opendht/routing_table.h \
         ../include/opendht/network_engine.h \
         ../include/opendht/scheduler.h \
         ../include/opendht/utils.h \
@@ -43,4 +44,5 @@ nobase_include_HEADERS = \
         ../include/opendht/dhtrunner.h \
         ../include/opendht/default_types.h \
         ../include/opendht/log.h \
-        ../include/opendht/rng.h
+        ../include/opendht/rng.h \
+        ../include/opendht/indexation/pht.h

src/callbacks.cpp

@@ -47,4 +47,12 @@ bindDoneCb(DoneCallbackRaw raw_cb, void* user_data)
     };
 }
 
+DoneCallbackSimple
+bindDoneCbSimple(DoneCallbackSimpleRaw raw_cb, void* user_data) {
+    if (not raw_cb) return {};
+    return [=](bool success) {
+        raw_cb(success, user_data);
+    };
+}
+
 }

src/dht.cpp

@@ -841,7 +841,7 @@ Dht::searchStep(std::shared_ptr<Search> sr)
                     continue;
                 }
             }
-            if (in) storageStore(sr->id, a.value, a.created);
+
             unsigned i = 0;
             for (auto& n : sr->nodes) {
                 if (not n.isSynced(now))
@@ -1243,6 +1243,9 @@ Dht::announce(const InfoHash& id, sa_family_t af, std::shared_ptr<Value> value,
             callback(false, {});
         return;
     }
+
+    storageStore(id, value, created);
+
     auto& srs = af == AF_INET ? searches4 : searches6;
     auto srp = srs.find(id);
     auto sr = srp == srs.end() ? search(id, af, nullptr, nullptr) : srp->second;

src/dhtrunner.cpp

@@ -484,30 +484,30 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> token)
 }
 
 void
-DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, bool permanent)
+DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created, bool permanent)
 {
     std::lock_guard<std::mutex> lck(storage_mtx);
     auto sv = std::make_shared<Value>(std::move(value));
     pending_ops.emplace([=](SecureDht& dht) {
-        dht.put(hash, sv, cb, {}, permanent);
+        dht.put(hash, sv, cb, created, permanent);
     });
     cv.notify_all();
 }
 
 void
-DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
+DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, time_point created, bool permanent)
 {
     std::lock_guard<std::mutex> lck(storage_mtx);
     pending_ops.emplace([=](SecureDht& dht) {
-        dht.put(hash, value, cb, {}, permanent);
+        dht.put(hash, value, cb, created, permanent);
     });
     cv.notify_all();
 }
 
 void
-DhtRunner::put(const std::string& key, Value&& value, DoneCallbackSimple cb, bool permanent)
+DhtRunner::put(const std::string& key, Value&& value, DoneCallbackSimple cb, time_point created, bool permanent)
 {
-    put(InfoHash::get(key), std::forward<Value>(value), cb, permanent);
+    put(InfoHash::get(key), std::forward<Value>(value), cb, created, permanent);
 }
 
 void

src/indexation/pht.cpp

@@ -0,0 +1,518 @@
+#include "indexation/pht.h"
+#include "rng.h"
+
+namespace dht {
+namespace indexation {
+
+/**
+ * Output the blob into string and readable way
+ *
+ * @param bl   : Blob to print
+ *
+ * @return string that represent the blob into a readable way
+ */
+static std::string blobToString(const Blob &bl) {
+    std::stringstream ss;
+    auto bn = bl.size() % 8;
+    auto n = bl.size() / 8;
+
+    for (size_t i = 0; i < bl.size(); i++)
+        ss << std::bitset<8>(bl[i]) << " ";
+    if (bn)
+        for (unsigned b=0; b < bn; b++)
+            ss << (char)((bl[n] & (1 << (7 - b))) ? '1':'0');
+
+    return ss.str();
+}
+
+std::string Prefix::toString() const {
+    std::stringstream ss;
+
+    ss << "Prefix : " << std::endl << "\tContent_ : \"";
+    ss << blobToString(content_);
+    ss << "\"" << std::endl;
+
+    ss << "\tFlags_   : \"";
+    ss << blobToString(flags_);
+    ss << "\"" << std::endl;
+
+    return ss.str();
+}
+
+void Pht::Cache::insert(const Prefix& p) {
+    size_t i = 0;
+    auto now = clock::now();
+
+    std::shared_ptr<Node> curr_node;
+
+    while ((leaves_.size() > 0
+        and leaves_.begin()->first + NODE_EXPIRE_TIME < now)
+        or  leaves_.size() > MAX_ELEMENT) {
+
+        leaves_.erase(leaves_.begin());
+    }
+
+    if (not (curr_node = root_.lock()) ) {
+        /* Root does not exist, need to create one*/
+        curr_node = std::make_shared<Node>();
+        root_ = curr_node;
+    }
+
+    curr_node->last_reply = now;
+
+    /* Iterate through all bit of the Blob */
+    for ( i = 0; i < p.size_; i++ ) {
+
+        /* According to the bit define which node is the next one */
+        auto& next = ( p.isContentBitActive(i) ) ? curr_node->right_child : curr_node->left_child;
+
+        /**
+         * If lock, node exists
+         * else create it
+         */
+        if (auto n = next.lock()) {
+            curr_node = std::move(n);
+        } else {
+            /* Create the next node if doesn't exist*/
+            auto tmp_curr_node = std::make_shared<Node>();
+            tmp_curr_node->parent = curr_node;
+            next = tmp_curr_node;
+            curr_node = std::move(tmp_curr_node);
+        }
+
+        curr_node->last_reply = now;
+    }
+
+    /* Insert the leaf (curr_node) into the multimap */
+    leaves_.emplace(std::move(now), std::move(curr_node) );
+}
+
+int Pht::Cache::lookup(const Prefix& p) {
+    int pos = -1;
+    auto now = clock::now(), last_node_time = now;
+
+    /* Before lookup remove the useless one [i.e. too old] */
+    while ( leaves_.size() > 0
+        and leaves_.begin()->first + NODE_EXPIRE_TIME < now ) {
+
+        leaves_.erase(leaves_.begin());
+    }
+
+    auto next = root_;
+    std::shared_ptr<Node> curr_node;
+
+    while ( auto n = next.lock() ) {
+        ++pos;
+        /* Safe since pos is equal to 0 until here */
+        if ( (unsigned) pos >= p.content_.size() * 8) break;
+
+        curr_node = n;
+        last_node_time = curr_node->last_reply;
+        curr_node->last_reply = now;
+
+        /* Get the Prefix bit by bit, starting from left */
+        next = ( p.isContentBitActive(pos) ) ? curr_node->right_child : curr_node->left_child;
+    }
+
+    if ( pos >= 0 ) {
+        auto to_erase = leaves_.find(last_node_time);
+        if ( to_erase != leaves_.end() )
+            leaves_.erase( to_erase );
+
+        leaves_.emplace( std::move(now), std::move(curr_node) );
+    }
+
+    return pos;
+}
+
+const ValueType IndexEntry::TYPE = ValueType::USER_DATA;
+constexpr std::chrono::minutes Pht::Cache::NODE_EXPIRE_TIME;
+
+void Pht::lookupStep(Prefix p, std::shared_ptr<int> lo, std::shared_ptr<int> hi,
+        std::shared_ptr<std::vector<std::shared_ptr<IndexEntry>>> vals,
+        LookupCallbackWrapper cb, DoneCallbackSimple done_cb,
+        std::shared_ptr<unsigned> max_common_prefix_len,
+        int start, bool all_values)
+{
+    struct node_lookup_result {
+        bool done {false};
+        bool is_pht {false};
+    };
+
+    /* start could be under 0 but after the compare it to 0 it always will be unsigned, so we can cast it*/
+    auto mid = (start >= 0) ? (unsigned) start : (*lo + *hi)/2;
+
+    auto first_res = std::make_shared<node_lookup_result>();
+    auto second_res = std::make_shared<node_lookup_result>();
+
+    auto on_done = [=](bool ok) {
+        bool is_leaf = first_res->is_pht and not second_res->is_pht;
+        if (not ok) {
+            if (done_cb)
+                done_cb(false);
+        }
+        else if (is_leaf or *lo > *hi) {
+            // leaf node
+            Prefix to_insert = p.getPrefix(mid);
+            cache_.insert(to_insert);
+
+            if (cb) {
+                if (vals->size() == 0 and max_common_prefix_len and mid > 0) {
+                    auto p_ = (p.getPrefix(mid)).getSibling().getFullSize();
+                    *lo = mid;
+                    *hi = p_.size_;
+                    lookupStep(p_, lo, hi, vals, cb, done_cb, max_common_prefix_len, -1, all_values);
+                }
+
+                cb(*vals, to_insert);
+            }
+
+            if (done_cb)
+                done_cb(true);
+        } else if (first_res->is_pht) {
+            // internal node
+            *lo = mid+1;
+            lookupStep(p, lo, hi, vals, cb, done_cb, max_common_prefix_len, -1, all_values);
+        } else {
+            // first get failed before second.
+            if (done_cb)
+                done_cb(false);
+        }
+    };
+
+    if (*lo <= *hi) {
+        auto pht_filter = [&](const dht::Value& v) {
+            return v.user_type.compare(0, name_.size(), name_) == 0;
+        };
+
+        auto on_get = [=](const std::shared_ptr<dht::Value>& value, std::shared_ptr<node_lookup_result> res) {
+            if (value->user_type == canary_) {
+                res->is_pht = true;
+            }
+            else {
+                IndexEntry entry;
+                entry.unpackValue(*value);
+
+                auto it = std::find_if(vals->cbegin(), vals->cend(), [&](const std::shared_ptr<IndexEntry>& ie) {
+                    return ie->value == entry.value;
+                });
+
+                /* If we already got the value then get the next one */
+                if (it != vals->cend())
+                    return true;
+
+                if (max_common_prefix_len) { /* inexact match case */
+                    auto common_bits = Prefix::commonBits(p, entry.prefix);
+
+                    if (vals->empty()) {
+                        vals->emplace_back(std::make_shared<IndexEntry>(entry));
+                        *max_common_prefix_len = common_bits;
+                    }
+                    else {
+                        if (common_bits == *max_common_prefix_len) /* this is the max so far */
+                            vals->emplace_back(std::make_shared<IndexEntry>(entry));
+                        else if (common_bits > *max_common_prefix_len) { /* new max found! */
+                            vals->clear();
+                            vals->emplace_back(std::make_shared<IndexEntry>(entry));
+                            *max_common_prefix_len = common_bits;
+                        }
+                    }
+                } else if (all_values or entry.prefix == p.content_) /* exact match case */
+                    vals->emplace_back(std::make_shared<IndexEntry>(entry));
+            }
+
+            return true;
+        };
+
+        dht_->get(p.getPrefix(mid).hash(),
+                std::bind(on_get, std::placeholders::_1, first_res),
+                [=](bool ok) {
+                    if (not ok) {
+                        // DHT failed
+                        first_res->done = true;
+                        if (done_cb and second_res->done)
+                            on_done(false);
+                    }
+                    else {
+                        if (not first_res->is_pht) {
+                            // Not a PHT node.
+                            *hi = mid-1;
+                            lookupStep(p, lo, hi, vals, cb, done_cb, max_common_prefix_len, -1, all_values);
+                        } else {
+                            first_res->done = true;
+                            if (second_res->done or mid >= p.size_ - 1)
+                                on_done(true);
+                        }
+                    }
+                }, pht_filter);
+
+        if (mid < p.size_ - 1)
+           dht_->get(p.getPrefix(mid+1).hash(),
+                    std::bind(on_get, std::placeholders::_1, second_res),
+                    [=](bool ok) {
+                        if (not ok) {
+                            // DHT failed
+                            second_res->done = true;
+                            if (done_cb and first_res->done)
+                                on_done(false);
+                        }
+                        else {
+                            second_res->done = true;
+                            if (first_res->done)
+                                on_done(true);
+                        }
+                }, pht_filter);
+    } else {
+        on_done(true);
+    }
+}
+
+void Pht::lookup(Key k, Pht::LookupCallback cb, DoneCallbackSimple done_cb, bool exact_match) {
+    auto prefix = linearize(k);
+    auto values = std::make_shared<std::vector<std::shared_ptr<IndexEntry>>>();
+
+    auto lo = std::make_shared<int>(0);
+    auto hi = std::make_shared<int>(prefix.size_);
+    std::shared_ptr<unsigned> max_common_prefix_len = not exact_match ? std::make_shared<unsigned>(0) : nullptr;
+
+    lookupStep(prefix, lo, hi, values,
+        [=](std::vector<std::shared_ptr<IndexEntry>>& entries, Prefix p) {
+            std::vector<std::shared_ptr<Value>> vals(entries.size());
+
+            std::transform(entries.begin(), entries.end(), vals.begin(),
+                [](const std::shared_ptr<IndexEntry>& ie) {
+                    return std::make_shared<Value>(ie->value);
+            });
+
+            cb(vals, p);
+        }, done_cb, max_common_prefix_len, cache_.lookup(prefix));
+}
+
+void Pht::updateCanary(Prefix p) {
+    // TODO: change this... copy value
+    dht::Value canary_value;
+    canary_value.user_type = canary_;
+
+    dht_->put(p.hash(), std::move(canary_value),
+        [=](bool){
+            static std::bernoulli_distribution d(0.5);
+            crypto::random_device rd;
+            if (p.size_ and d(rd))
+                updateCanary(p.getPrefix(-1));
+        }
+    );
+
+    if (p.size_) {
+        dht::Value canary_second_value;
+        canary_second_value.user_type = canary_;
+        dht_->put(p.getSibling().hash(), std::move(canary_second_value));
+    }
+}
+
+void Pht::insert(Prefix kp, IndexEntry entry, std::shared_ptr<int> lo, std::shared_ptr<int> hi, time_point time_p,
+                 bool check_split, DoneCallbackSimple done_cb) {
+
+    if (time_p + ValueType::USER_DATA.expiration < clock::now()) return;
+
+    auto vals = std::make_shared<std::vector<std::shared_ptr<IndexEntry>>>();
+    auto final_prefix = std::make_shared<Prefix>();
+
+    lookupStep(kp, lo, hi, vals,
+        [=](std::vector<std::shared_ptr<IndexEntry>>&, Prefix p) {
+            *final_prefix = Prefix(p);
+        },
+        [=](bool ok){
+            if (not ok) {
+                if (done_cb)
+                    done_cb(false);
+            } else {
+
+                RealInsertCallback real_insert = [=]( std::shared_ptr<Prefix> p, IndexEntry entry) {
+                    updateCanary(*p);
+                    checkPhtUpdate(*p, entry, time_p);
+                    cache_.insert(*p);
+                    dht_->put(p->hash(), std::move(entry), done_cb , time_p);
+                };
+
+                if ( not check_split or final_prefix->size_ == kp.size_ ) {
+                    real_insert(final_prefix, std::move(entry));
+                } else {
+                    if ( vals->size() < MAX_NODE_ENTRY_COUNT ) {
+                        getRealPrefix(final_prefix, std::move(entry), real_insert);
+                    }
+                    else {
+                        split(*final_prefix, vals, entry, real_insert);
+                    }
+                }
+            }
+        }, nullptr, cache_.lookup(kp), true
+    );
+}
+
+Prefix Pht::zcurve(const std::vector<Prefix>& all_prefix) const {
+    Prefix p;
+
+    if ( all_prefix.size() == 1 )
+        return all_prefix[0];
+
+    /* All prefix got the same size (thanks to padding) */
+    size_t prefix_size = all_prefix[0].content_.size();
+
+    /* Loop on all uint8_t of the input prefix */
+    for ( size_t j = 0, bit = 0; j < prefix_size; j++) {
+
+        uint8_t mask = 0x80;
+        /* For each of the 8 bits of the input uint8_t */
+        for ( int i = 0; i < 8; ) {
+
+            uint8_t flags = 0;
+            uint8_t content = 0;
+
+            /* For each bit of the output uint8_t */
+            for ( int k = 0 ; k < 8; k++ ) {
+
+                auto diff = k - i;
+
+                /*get the content 'c', and the flag 'f' of the input prefix */
+                auto c = all_prefix[bit].content_[j] & mask;
+                auto f = all_prefix[bit].flags_[j] & mask;
+
+                /* Move this bit at the right position according to the diff
+                   and merge it into content and flags in the same way */
+                content |= ( diff >= 0 ) ? c >> diff : c << std::abs(diff);
+                flags   |= ( diff >= 0 ) ? f >> diff : f << std::abs(diff);
+
+                /* If we are on the last prefix of the vector get back to the first and
+                ,move the mask in order to get the n + 1nth bit */
+                if ( ++bit == all_prefix.size() ) { bit = 0; ++i; mask >>= 1; }
+            }
+
+            /* Add the next flags + content to the output prefix */
+            p.content_.push_back(content);
+            p.flags_.push_back(flags);
+            p.size_ += 8;
+        }
+    }
+
+    return p;
+}
+
+Prefix Pht::linearize(Key k) const {
+    if (not validKey(k)) { throw std::invalid_argument(INVALID_KEY); }
+
+    std::vector<Prefix> all_prefix;
+    all_prefix.reserve(k.size());
+
+    /* Get the max size of the keyspec and take it for size limit (for padding) */
+    auto max = std::max_element(keySpec_.begin(), keySpec_.end(),
+        [](const std::pair<std::string, size_t>& a, const std::pair<std::string, size_t>& b) {
+            return a.second < b.second;
+        })->second + 1;
+
+    for ( auto const& it : k ) {
+        Prefix p = Blob {it.second.begin(), it.second.end()};
+        p.addPaddingContent(max);
+        p.updateFlags();
+
+        all_prefix.emplace_back(std::move(p));
+    }
+
+    return zcurve(all_prefix);
+}
+
+void Pht::getRealPrefix(std::shared_ptr<Prefix> p, IndexEntry entry, RealInsertCallback end_cb ) {
+
+    if ( p->size_ == 0 ) {
+        end_cb(p, std::move(entry));
+        return;
+    }
+
+    auto total = std::make_shared<unsigned int>(0); /* Will contains the total number of data on 3 nodes */
+    auto ended = std::make_shared<unsigned int>(0); /* Just indicate how many have end */
+
+    auto parent = std::make_shared<Prefix>(p->getPrefix(-1));
+    auto sibling = std::make_shared<Prefix>(p->getSibling());
+
+    auto pht_filter = [&](const dht::Value& v) {
+        return v.user_type.compare(0, name_.size(), name_) == 0;
+    };
+
+    /* Lambda will count total number of data node */
+    auto count = [=]( const std::shared_ptr<dht::Value> value ) {
+        if ( value->user_type != canary_)
+            (*total)++;
+
+        return true;
+    };
+
+    auto on_done = [=] ( bool ) {
+        (*ended)++;
+        /* Only the last one do the CallBack*/
+        if  ( *ended == 3 ) {
+            if ( *total < MAX_NODE_ENTRY_COUNT )
+                end_cb(parent, std::move(entry));
+            else
+                end_cb(p, std::move(entry));
+        }
+    };
+
+    dht_->get(parent->hash(),
+        count,
+        on_done,
+        pht_filter
+    );
+
+    dht_->get(p->hash(),
+        count,
+        on_done,
+        pht_filter
+    );
+
+    dht_->get(sibling->hash(),
+        count,
+        on_done,
+        pht_filter
+    );
+}
+
+void Pht::checkPhtUpdate(Prefix p, IndexEntry entry, time_point time_p) {
+
+    Prefix full = entry.prefix;
+    if ( p.content_.size() * 8 >= full.content_.size() * 8 ) return;
+
+    auto next_prefix = full.getPrefix( p.size_ + 1 );
+
+    dht_->listen(next_prefix.hash(),
+        [=](const std::shared_ptr<dht::Value> &value) {
+            if (value->user_type == canary_) {
+                insert(full, entry, std::make_shared<int>(0), std::make_shared<int>(full.size_), time_p, false, nullptr);
+
+                /* Cancel listen since we found where we need to update*/
+                return false;
+            }
+
+            return true;
+        },
+        [=](const dht::Value& v) {
+            /* Filter value v thats start with the same name as ours */
+            return v.user_type.compare(0, name_.size(), name_) == 0;
+        }
+    );
+}
+
+void Pht::split(Prefix insert, std::shared_ptr<std::vector<std::shared_ptr<IndexEntry>>> vals, IndexEntry entry, RealInsertCallback end_cb ) {
+    auto full = Prefix(entry.prefix);
+
+    auto loc = foundSplitLocation(full, vals);
+    auto prefix_to_insert = std::make_shared<Prefix>(full.getPrefix(loc));
+
+    for(;loc != insert.size_ - 1; loc--) {
+        updateCanary(full.getPrefix(loc));
+    }
+
+    end_cb(prefix_to_insert, entry);
+}
+
+} /* indexation  */
+
+} /* dht */

tools/dhtnode.cpp

@@ -38,12 +38,12 @@ void print_id_req() {
     std::cout << "An identity is required to perform this operation (run with -i)" << std::endl;
 }
 
-void print_node_info(const DhtRunner& dht, const dht_params& params) {
-    std::cout << "OpenDht node " << dht.getNodeId() << " running on port " <<  dht.getBoundPort() << std::endl;
+void print_node_info(const std::shared_ptr<DhtRunner>& dht, const dht_params& params) {
+    std::cout << "OpenDht node " << dht->getNodeId() << " running on port " <<  dht->getBoundPort() << std::endl;
     if (params.is_bootstrap_node)
         std::cout << "Running in bootstrap mode (discouraged)." << std::endl;
     if (params.generate_identity)
-        std::cout << "Public key ID " << dht.getId() << std::endl;
+        std::cout << "Public key ID " << dht->getId() << std::endl;
 }
 
 void print_help() {
@@ -60,16 +60,20 @@ void print_help() {
               << "  lr         Print the full current routing table of this node" << std::endl;
 
     std::cout << std::endl << "Operations on the DHT:" << std::endl
-              << "  b ip:port             Ping potential node at given IP address/port." << std::endl
-              << "  g [key]               Get values at [key]." << std::endl
-              << "  l [key]               Listen for value changes at [key]." << std::endl
-              << "  p [key] [str]         Put string value at [key]." << std::endl
-              << "  s [key] [str]         Put string value at [key], signed with our generated private key." << std::endl
-              << "  e [key] [dest] [str]  Put string value at [key], encrypted for [dest] with its public key (if found)." << std::endl
+              << "  b <ip:port>             Ping potential node at given IP address/port." << std::endl
+              << "  g <key>               Get values at <key>." << std::endl
+              << "  l <key>               Listen for value changes at <key>." << std::endl
+              << "  p <key> <str>         Put string value at <key>." << std::endl
+              << "  s <key> <str>         Put string value at <key>, signed with our generated private key." << std::endl
+              << "  e <key> <dest> <str>  Put string value at <key>, encrypted for <dest> with its public key (if found)." << std::endl;
+    std::cout << std::endl << "Indexation operations on the DHT:" << std::endl
+              << "  il <name> <key> [exact match]   Lookup the index named <name> with the key <key>." << std::endl
+              << "                                  Set [exact match] to 'false' for inexact match lookup." << std::endl
+              << "  ii <name> <key> <value>         Inserts the value <value> under the key <key> in the index named <name>." << std::endl
               << std::endl;
 }
 
-void cmd_loop(DhtRunner& dht, dht_params& params)
+void cmd_loop(std::shared_ptr<DhtRunner>& dht, std::map<std::string, indexation::Pht> indexes, dht_params& params)
 {
     print_node_info(dht, params);
     std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl << std::endl;
@@ -85,8 +89,8 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
             break;
 
         std::istringstream iss(line);
-        std::string op, idstr, value;
-        iss >> op >> idstr;
+        std::string op, idstr, value, index, keystr;
+        iss >> op;
 
         if (op == "x" || op == "q" || op == "exit" || op == "quit") {
             break;
@@ -97,31 +101,32 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
             print_node_info(dht, params);
             unsigned good4, dubious4, cached4, incoming4;
             unsigned good6, dubious6, cached6, incoming6;
-            dht.getNodesStats(AF_INET, &good4, &dubious4, &cached4, &incoming4);
-            dht.getNodesStats(AF_INET6, &good6, &dubious6, &cached6, &incoming6);
+            dht->getNodesStats(AF_INET, &good4, &dubious4, &cached4, &incoming4);
+            dht->getNodesStats(AF_INET6, &good6, &dubious6, &cached6, &incoming6);
             std::cout << "IPv4 nodes : " << good4 << " good, " << dubious4 << " dubious, " << incoming4 << " incoming." << std::endl;
             std::cout << "IPv6 nodes : " << good6 << " good, " << dubious6 << " dubious, " << incoming6 << " incoming." << std::endl;
             continue;
         } else if (op == "lr") {
             std::cout << "IPv4 routing table:" << std::endl;
-            std::cout << dht.getRoutingTablesLog(AF_INET) << std::endl;
+            std::cout << dht->getRoutingTablesLog(AF_INET) << std::endl;
             std::cout << "IPv6 routing table:" << std::endl;
-            std::cout << dht.getRoutingTablesLog(AF_INET6) << std::endl;
+            std::cout << dht->getRoutingTablesLog(AF_INET6) << std::endl;
             continue;
         } else if (op == "ld") {
-            std::cout << dht.getStorageLog() << std::endl;
+            std::cout << dht->getStorageLog() << std::endl;
             continue;
         } else if (op == "ls") {
             std::cout << "Searches:" << std::endl;
-            std::cout << dht.getSearchesLog() << std::endl;
+            std::cout << dht->getSearchesLog() << std::endl;
             continue;
         } else if (op == "la")  {
             std::cout << "Reported public addresses:" << std::endl;
-            auto addrs = dht.getPublicAddressStr();
+            auto addrs = dht->getPublicAddressStr();
             for (const auto& addr : addrs)
                 std::cout << addr << std::endl;
             continue;
         } else if (op == "b") {
+            iss >> idstr;
             try {
                 auto addr = splitPort(idstr);
                 if (not addr.first.empty() and addr.second.empty()){
@@ -129,7 +134,7 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
                     ss << DHT_DEFAULT_PORT;
                     addr.second = ss.str();
                 }
-                dht.bootstrap(addr.first.c_str(), addr.second.c_str());
+                dht->bootstrap(addr.first.c_str(), addr.second.c_str());
             } catch (const std::exception& e) {
                 std::cerr << e.what() << std::endl;
             }
@@ -137,31 +142,57 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
         } else if (op == "log") {
             params.log = !params.log;
             if (params.log)
-                log::enableLogging(dht);
+                log::enableLogging(*dht);
             else
-                log::disableLogging(dht);
+                log::disableLogging(*dht);
             continue;
         }
 
         if (op.empty())
             continue;
 
-        dht::InfoHash id {idstr};
-        static const std::set<std::string> VALID_OPS {"g", "l", "p", "s", "e", "a"};
+        static const std::set<std::string> VALID_OPS {"g", "l", "il", "ii", "p", "s", "e", "a"};
         if (VALID_OPS.find(op) == VALID_OPS.cend()) {
             std::cout << "Unknown command: " << op << std::endl;
             std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl;
             continue;
         }
-        static constexpr dht::InfoHash INVALID_ID {};
-        if (id == INVALID_ID) {
-            std::cout << "Syntax error: invalid InfoHash." << std::endl;
-            continue;
+
+        if (op == "il" or op == "ii") {
+            // Pht syntax
+            iss >> index >> keystr;
+            auto new_index = std::find_if(indexes.begin(), indexes.end(),
+                    [&](std::pair<const std::string, indexation::Pht>& i) {
+                        return i.first == index;
+                    }) == indexes.end();
+            if (not index.size()) {
+                std::cout << "You must enter the index name." << std::endl;
+                continue;
+            } else if (new_index) {
+                using namespace dht::indexation;
+                try {
+                    auto key = createPhtKey(parseStringMap(keystr));
+                    Pht::KeySpec ks;
+                    std::transform(key.begin(), key.end(), std::inserter(ks, ks.end()), [](Pht::Key::value_type& f) {
+                        return std::make_pair(f.first, f.second.size());
+                    });
+                    indexes.emplace(index, Pht {index, std::move(ks), dht});
+                } catch (std::invalid_argument& e) { std::cout << e.what() << std::endl; }
+            }
+        }
+        else {
+            // Dht syntax
+            iss >> idstr;
+            InfoHash h {idstr};
+            if (not isInfoHash(h))
+                continue;
         }
 
+        dht::InfoHash id {idstr};
+        // Dht
         auto start = std::chrono::high_resolution_clock::now();
         if (op == "g") {
-            dht.get(id, [start](std::shared_ptr<Value> value) {
+            dht->get(id, [start](std::shared_ptr<Value> value) {
                 auto now = std::chrono::high_resolution_clock::now();
                 std::cout << "Get: found value (after " << print_dt(now-start) << "s)" << std::endl;
                 std::cout << "\t" << *value << std::endl;
@@ -172,7 +203,7 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
             });
         }
         else if (op == "l") {
-            dht.listen(id, [](std::shared_ptr<Value> value) {
+            dht->listen(id, [](std::shared_ptr<Value> value) {
                 std::cout << "Listen: found value:" << std::endl;
                 std::cout << "\t" << *value << std::endl;
                 return true;
@@ -181,7 +212,7 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
         else if (op == "p") {
             std::string v;
             iss >> v;
-            dht.put(id, dht::Value {
+            dht->put(id, dht::Value {
                 dht::ValueType::USER_DATA.id,
                 std::vector<uint8_t> {v.begin(), v.end()}
             }, [start](bool ok) {
@@ -196,7 +227,7 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
             }
             std::string v;
             iss >> v;
-            dht.putSigned(id, dht::Value {
+            dht->putSigned(id, dht::Value {
                 dht::ValueType::USER_DATA.id,
                 std::vector<uint8_t> {v.begin(), v.end()}
             }, [start](bool ok) {
@@ -212,7 +243,7 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
             std::string tostr;
             std::string v;
             iss >> tostr >> v;
-            dht.putEncrypted(id, InfoHash(tostr), dht::Value {
+            dht->putEncrypted(id, InfoHash(tostr), dht::Value {
                 dht::ValueType::USER_DATA.id,
                 std::vector<uint8_t> {v.begin(), v.end()}
             }, [start](bool ok) {
@@ -223,11 +254,56 @@ void cmd_loop(DhtRunner& dht, dht_params& params)
         else if (op == "a") {
             in_port_t port;
             iss >> port;
-            dht.put(id, dht::Value {dht::IpServiceAnnouncement::TYPE.id, dht::IpServiceAnnouncement(port)}, [start](bool ok) {
+            dht->put(id, dht::Value {dht::IpServiceAnnouncement::TYPE.id, dht::IpServiceAnnouncement(port)}, [start](bool ok) {
                 auto end = std::chrono::high_resolution_clock::now();
                 std::cout << "Announce: " << (ok ? "success" : "failure") << " (took " << print_dt(end-start) << "s)" << std::endl;
             });
         }
+        else if (op == "il") {
+            std::string exact_match;
+            iss >> exact_match;
+            try {
+                auto key = createPhtKey(parseStringMap(keystr));
+                indexes.at(index).lookup(key,
+                    [=](std::vector<std::shared_ptr<indexation::Value>>& vals, indexation::Prefix p) {
+                        if (vals.empty())
+                            return;
+                        std::cout << "Pht::lookup: found entries!" << std::endl
+                                  << p.toString() << std::endl
+                                  << "   hash: " << p.hash() << std::endl;
+                        std::cout << "   entries:" << std::endl;
+                        for (auto v : vals)
+                             std::cout << "      " << v->first.toString() << "[vid: " << v->second << "]" << std::endl;
+                    },
+                    [start](bool ok) {
+                        auto end = std::chrono::high_resolution_clock::now();
+                        std::cout << "Pht::lookup: " << (ok ? "done." : "failed.")
+                                  << " took " << print_dt(end-start) << "s)" << std::endl;
+
+                    }, exact_match.size() != 0 and exact_match == "false" ? false : true
+                );
+            }
+            catch (std::invalid_argument& e) { std::cout << e.what() << std::endl; }
+            catch (std::out_of_range& e) { }
+        }
+        else if (op == "ii") {
+            iss >> idstr;
+            InfoHash h {idstr};
+            if (not isInfoHash(h))
+                continue;
+
+            indexation::Value v {h, 0};
+            try {
+                auto key = createPhtKey(parseStringMap(keystr));
+                indexes.at(index).insert(key, v,
+                    [=](bool ok) {
+                        std::cout << "Pht::insert: " << (ok ? "done." : "failed.") << std::endl;
+                    }
+                );
+            }
+            catch (std::invalid_argument& e) { std::cout << e.what() << std::endl; }
+            catch (std::out_of_range& e) { }
+        }
     }
 
     std::cout << std::endl <<  "Stopping node..." << std::endl;
@@ -239,7 +315,9 @@ main(int argc, char **argv)
     if (int rc = gnutls_global_init())  // TODO: remove with GnuTLS >= 3.3
         throw std::runtime_error(std::string("Error initializing GnuTLS: ")+gnutls_strerror(rc));
 
-    DhtRunner dht;
+    auto dht = std::make_shared<DhtRunner>();
+    std::map<std::string, indexation::Pht> indexes;
+
     try {
         auto params = parseArgs(argc, argv);
         if (params.help) {
@@ -256,25 +334,25 @@ main(int argc, char **argv)
             crt = dht::crypto::generateIdentity("DHT Node", ca_tmp);
         }
 
-        dht.run(params.port, crt, true, params.is_bootstrap_node);
+        dht->run(params.port, crt, true, params.is_bootstrap_node);
 
         if (params.log) {
             if (not params.logfile.empty())
-                log::enableFileLogging(dht, params.logfile);
+                log::enableFileLogging(*dht, params.logfile);
             else
-                log::enableLogging(dht);
+                log::enableLogging(*dht);
         }
 
         if (not params.bootstrap.first.empty()) {
             //std::cout << "Bootstrap: " << params.bootstrap.first << ":" << params.bootstrap.second << std::endl;
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+            dht->bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
         }
 
         if (params.daemonize) {
             while (true)
                 std::this_thread::sleep_for(std::chrono::seconds(30));
         } else {
-            cmd_loop(dht, params);
+            cmd_loop(dht, indexes, params);
         }
 
     } catch(const std::exception&e) {
@@ -285,7 +363,7 @@ main(int argc, char **argv)
     std::mutex m;
     std::atomic_bool done {false};
 
-    dht.shutdown([&]()
+    dht->shutdown([&]()
     {
         std::lock_guard<std::mutex> lk(m);
         done = true;
@@ -296,7 +374,7 @@ main(int argc, char **argv)
     std::unique_lock<std::mutex> lk(m);
     cv.wait(lk, [&](){ return done.load(); });
 
-    dht.join();
+    dht->join();
     gnutls_global_deinit();
 
     return 0;

tools/tools_common.h

@@ -135,6 +135,47 @@ splitPort(const std::string& s) {
     return {s.substr(0,found), s.substr(found+1)};
 }
 
+/*
+ * The mapString shall have the following format:
+ *
+ *      k1:v1[,k2:v2[,...]]
+ */
+std::map<std::string, std::string> parseStringMap(std::string mapString) {
+    std::istringstream keySs(mapString);
+    std::string mapStr;
+    std::map<std::string, std::string> map;
+
+    while (std::getline(keySs, mapStr, ',')) {
+        std::istringstream mapSs(mapStr);
+        std::string key, value;
+
+        while (std::getline(mapSs, key, ':')) {
+            std::getline(mapSs, value, ':');
+            map[key] = { value };
+        }
+    }
+    return map;
+}
+
+dht::indexation::Pht::Key createPhtKey(std::map<std::string, std::string> pht_key_str_map) {
+    dht::indexation::Pht::Key pht_key;
+    for (auto f : pht_key_str_map) {
+        dht::Blob prefix {f.second.begin(), f.second.end()};
+        pht_key.emplace(f.first, std::move(prefix));
+    }
+    return pht_key;
+}
+
+bool isInfoHash(dht::InfoHash& h) {
+    static constexpr dht::InfoHash INVALID_ID {};
+
+    if (h == INVALID_ID) {
+        std::cout << "Syntax error: invalid InfoHash." << std::endl;
+        return false;
+    }
+    return true;
+}
+
 static const constexpr in_port_t DHT_DEFAULT_PORT = 4222;
 
 struct dht_params {