tests: avoid use-after-free in proxytester

crypto: add certificate request support

.gitignore

@@ -46,6 +46,15 @@ Makefile.in
 .kdev4/
 *.kdev4
 
+# VSCode
+.vscode/*
+
+# Eclipse CDT
+.cproject
+.project
+.autotools
+language.settings.xml
+
 # Python
 *.pyc
 *.stamp

CMakeLists.txt

@@ -149,6 +149,9 @@ list (APPEND opendht_SOURCES
     src/securedht.cpp
     src/dhtrunner.cpp
     src/log.cpp
+    src/peer_discovery.cpp
+    src/network_utils.h
+    src/network_utils.cpp
 )
 
 list (APPEND opendht_HEADERS
@@ -172,9 +175,14 @@ list (APPEND opendht_HEADERS
     include/opendht/securedht.h
     include/opendht/log.h
     include/opendht/log_enable.h
+    include/opendht/peer_discovery.h
     include/opendht.h
 )
 
+if (OPENDHT_PYTHON)
+    message("Indexation enabled since it is required for Python support")
+    set(OPENDHT_INDEX ON)
+endif()
 if (OPENDHT_INDEX)
     list (APPEND opendht_SOURCES src/indexation/pht.cpp)
     list (APPEND opendht_HEADERS include/opendht/indexation/pht.h)
@@ -310,6 +318,8 @@ if (OPENDHT_TESTS)
       tests/cryptotester.cpp
       tests/dhtrunnertester.h
       tests/dhtrunnertester.cpp
+      tests/peerdiscoverytester.h
+      tests/peerdiscoverytester.cpp
     )
     if (OPENDHT_PROXY_SERVER AND OPENDHT_PROXY_CLIENT)
       list (APPEND test_FILES

include/opendht/dht.h

@@ -163,7 +163,7 @@ public:
     /**
      * Get locally stored data for the given hash.
      */
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const;
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, const Value::Filter& f = {}) const;
 
     /**
      * Get locally stored data for the given key and value id.
@@ -210,12 +210,12 @@ public:
     /**
      * Get data currently being put at the given hash.
      */
-    std::vector<Sp<Value>> getPut(const InfoHash&);
+    std::vector<Sp<Value>> getPut(const InfoHash&) const;
 
     /**
      * Get data currently being put at the given hash with the given id.
      */
-    Sp<Value> getPut(const InfoHash&, const Value::Id&);
+    Sp<Value> getPut(const InfoHash&, const Value::Id&) const;
 
     /**
      * Stop any put/announce operation at the given location,

include/opendht/dht_interface.h

@@ -106,7 +106,7 @@ public:
     /**
      * Get locally stored data for the given hash.
      */
-    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const = 0;
+    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, const Value::Filter& f = {}) const = 0;
 
     /**
      * Get locally stored data for the given key and value id.
@@ -143,12 +143,12 @@ public:
     /**
      * Get data currently being put at the given hash.
      */
-    virtual std::vector<Sp<Value>> getPut(const InfoHash&) = 0;
+    virtual std::vector<Sp<Value>> getPut(const InfoHash&) const = 0;
 
     /**
      * Get data currently being put at the given hash with the given id.
      */
-    virtual Sp<Value> getPut(const InfoHash&, const Value::Id&) = 0;
+    virtual Sp<Value> getPut(const InfoHash&, const Value::Id&) const = 0;
 
     /**
      * Stop any put/announce operation at the given location,

include/opendht/dht_proxy_client.h

@@ -202,12 +202,12 @@ public:
     /**
      * Get data currently being put at the given hash.
      */
-    std::vector<Sp<Value>> getPut(const InfoHash&);
+    std::vector<Sp<Value>> getPut(const InfoHash&) const;
 
     /**
      * Get data currently being put at the given hash with the given id.
      */
-    Sp<Value> getPut(const InfoHash&, const Value::Id&);
+    Sp<Value> getPut(const InfoHash&, const Value::Id&) const;
 
     /**
      * Stop any put/announce operation at the given location,
@@ -224,7 +224,7 @@ public:
         return types.getType(type_id);
     }
 
-    std::vector<Sp<Value>> getLocal(const InfoHash& k, Value::Filter filter) const;
+    std::vector<Sp<Value>> getLocal(const InfoHash& k, const Value::Filter& filter) const;
     Sp<Value> getLocalById(const InfoHash& k, Value::Id id) const;
 
     /**

include/opendht/dhtrunner.h

@@ -40,6 +40,7 @@ namespace dht {
 
 struct Node;
 class SecureDht;
+class PeerDiscovery;
 struct SecureDhtConfig;
 
 /**
@@ -58,6 +59,8 @@ public:
         bool threaded;
         std::string proxy_server;
         std::string push_node_id;
+        bool peer_discovery;
+        bool peer_publish;
     };
 
     DhtRunner();
@@ -105,7 +108,7 @@ public:
         getFilterSet<T>());
     }
 
-    std::future<std::vector<std::shared_ptr<dht::Value>>> get(InfoHash key, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    std::future<std::vector<std::shared_ptr<dht::Value>>> get(InfoHash key, Value::Filter f = {}, Where w = {}) {
         auto p = std::make_shared<std::promise<std::vector<std::shared_ptr< dht::Value >>>>();
         auto values = std::make_shared<std::vector<std::shared_ptr< dht::Value >>>();
         get(key, [=](const std::vector<std::shared_ptr<dht::Value>>& vlist) {
@@ -136,7 +139,7 @@ public:
         query(hash, cb, bindDoneCb(done_cb), q);
     }
 
-    std::future<size_t> listen(InfoHash key, ValueCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
+    std::future<size_t> listen(InfoHash key, ValueCallback vcb, Value::Filter f = {}, Where w = {});
 
     std::future<size_t> listen(InfoHash key, GetCallback cb, Value::Filter f={}, Where w={}) {
         return listen(key, [cb](const std::vector<Sp<Value>>& vals, bool expired){
@@ -145,8 +148,8 @@ public:
             return true;
         }, std::forward<Value::Filter>(f), std::forward<Where>(w));
     }
-    std::future<size_t> listen(const std::string& key, GetCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
-    std::future<size_t> listen(InfoHash key, GetCallbackSimple cb, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    std::future<size_t> listen(const std::string& key, GetCallback vcb, Value::Filter f = {}, Where w = {});
+    std::future<size_t> listen(InfoHash key, GetCallbackSimple cb, Value::Filter f = {}, Where w = {}) {
         return listen(key, bindGetCb(cb), f, w);
     }
 
@@ -168,7 +171,7 @@ public:
     }
 
     template <typename T>
-    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
+    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = {}, Where w = {})
     {
         return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals) {
             for (const auto& v : vals) {
@@ -184,7 +187,7 @@ public:
         getFilterSet<T>(f), w);
     }
     template <typename T>
-    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&, bool)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
+    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&, bool)> cb, Value::Filter f = {}, Where w = {})
     {
         return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
             for (const auto& v : vals) {
@@ -259,6 +262,12 @@ public:
      */
     void bootstrap(const std::string& host, const std::string& service);
 
+    /**
+     * Insert known nodes to the routing table, without necessarly ping them.
+     * Usefull to restart a node and get things running fast without putting load on the network.
+     */
+    void bootstrap(const InfoHash& id, const SockAddr& address);
+
     /**
      * Clear the list of bootstrap added using bootstrap(const std::string&, const std::string&).
      */
@@ -361,7 +370,9 @@ public:
             },
             /*.threaded = */threaded,
             /*.proxy_server = */"",
-            /*.push_node_id = */""
+            /*.push_node_id = */"",
+            /*.peer_discovery = */true,
+            /*.peer_publish = */true,
         });
     }
     void run(in_port_t port, const Config& config);
@@ -526,6 +537,12 @@ private:
 
     /** Push notification token */
     std::string pushToken_;
+
+    /** PeerDiscovery Parameters */
+    std::unique_ptr<PeerDiscovery> peerDiscovery4_;
+    std::unique_ptr<PeerDiscovery> peerDiscovery6_;
+    const in_port_t port_multicast = 8888;
+
 };
 
 }

include/opendht/infohash.h

@@ -65,7 +65,7 @@ public:
     typedef typename T::iterator iterator;
     typedef typename T::const_iterator const_iterator;
 
-    Hash () {
+    Hash () noexcept {
         data_.fill(0);
     }
     Hash (const uint8_t* h, size_t data_len) {
@@ -85,7 +85,7 @@ public:
         msgpack_unpack(o);
     }
 
-    size_t size() const { return data_.size(); }
+    static constexpr size_t size() noexcept { return N; }
     const uint8_t* data() const { return data_.data(); }
     uint8_t* data() { return data_.data(); }
     iterator begin() { return data_.begin(); }

include/opendht/peer_discovery.h

@@ -0,0 +1,137 @@
+/*
+ *  Copyright (C) 2014-2019 Savoir-faire Linux Inc.
+ *  Author(s) : Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include "sockaddr.h"
+#include "infohash.h"
+
+#include <string.h>
+#include <stdio.h>
+#include <unistd.h>
+
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+
+namespace dht {
+
+class OPENDHT_PUBLIC PeerDiscovery
+{
+public:
+
+    using PeerDiscoveredCallback = std::function<void(const InfoHash&, const SockAddr&)>;
+
+    PeerDiscovery(sa_family_t domain, in_port_t port);
+    ~PeerDiscovery();
+
+    /**
+     * startDiscovery - Keep Listening data from the sender until node is joinned or stop is called
+    */
+    void startDiscovery(PeerDiscoveredCallback callback);
+
+    /**
+     * startPublish - Keeping sending data until node is joinned or stop is called
+    */
+    void startPublish(const dht::InfoHash &nodeId, in_port_t port_to_send);
+
+    /**
+     * Thread Stopper
+    */
+    void stop();
+
+    /**
+     * Configure the sockopt to be able to listen multicast group
+    */
+    static void socketJoinMulticast(int sockfd, sa_family_t family);
+
+    /**
+     * Join the threads
+    */
+    void join() {
+        if(running_listen.joinable()) running_listen.join();
+        if(running_send.joinable()) running_send.join();
+    }
+
+private:
+    std::mutex mtx_;
+    std::condition_variable cv_;
+    bool running_ {true};
+    sa_family_t domain_ {AF_UNSPEC};
+    int port_;
+    int sockfd_ {-1};
+    int stop_writefd_ {-1};
+
+    SockAddr sockAddrSend_;
+    std::array<uint8_t,dht::InfoHash::size() + sizeof(in_port_t)> data_send_;
+
+    //Thread export to be joined
+    std::thread running_listen;
+    std::thread running_send;
+    dht::InfoHash nodeId_;
+
+    /**
+     * Multicast Socket Initialization, accept IPV4, IPV6
+    */
+    static int initialize_socket(sa_family_t domain);
+
+    /**
+     * Send messages
+    */
+    void sendTo(uint8_t *buf,size_t buf_size);
+
+    /**
+     * Receive messages
+    */
+    SockAddr recvFrom(uint8_t *buf, size_t &buf_size);
+
+    /**
+     * Send thread loop
+    */
+    void sender_thread();
+
+    /**
+     * Listener thread loop
+    */
+    void listener_thread(PeerDiscoveredCallback callback);
+
+    /**
+     * Listener Parameters Setup
+    */
+    void listener_setup();
+
+    /**
+     * Sender Parameters Setup
+    */
+    void sender_setup(const dht::InfoHash& nodeId, in_port_t port_to_send);
+
+    /**
+     * Binary Converters
+    */
+    static void inttolitend(uint32_t x, uint8_t *lit_int) {
+        lit_int[0] = (uint8_t)(x >>  0);
+        lit_int[1] = (uint8_t)(x >>  8);
+    }
+
+    static uint16_t litendtoint(uint8_t *lit_int) {
+        return (uint32_t)lit_int[0] <<  0
+            |  (uint32_t)lit_int[1] <<  8;
+    }
+};
+
+}

include/opendht/securedht.h

@@ -231,7 +231,7 @@ public:
     void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) override {
         dht_->query(key, cb, done_cb, std::move(q));
     }
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const override {
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, const Value::Filter& f = {}) const override {
         return dht_->getLocal(key, f);
     }
     Sp<Value> getLocalById(const InfoHash& key, Value::Id vid) const override {
@@ -270,10 +270,10 @@ public:
     {
         dht_->put(key, std::move(v), cb, created, permanent);
     }
-    std::vector<Sp<Value>> getPut(const InfoHash& h) override {
+    std::vector<Sp<Value>> getPut(const InfoHash& h) const override {
         return dht_->getPut(h);
     }
-    Sp<Value> getPut(const InfoHash& h, const Value::Id& vid) override {
+    Sp<Value> getPut(const InfoHash& h, const Value::Id& vid) const override {
         return dht_->getPut(h, vid);
     }
     bool cancelPut(const InfoHash& h, const Value::Id& vid) override {

include/opendht/sockaddr.h

@@ -23,6 +23,7 @@
 #ifndef _WIN32
 #include <sys/socket.h>
 #include <netinet/in.h>
+#include <arpa/inet.h>
 #ifdef __ANDROID__
 typedef uint16_t in_port_t;
 #endif
@@ -146,6 +147,21 @@ public:
             addr->sa_family = af;
     }
 
+    /**
+     * Set Network Interface to any
+     */
+    void setAny() {
+        auto family = getFamily();
+        switch(family) {
+        case AF_INET:
+            getIPv4().sin_addr.s_addr = htonl(INADDR_ANY);
+            break;
+        case AF_INET6:
+            getIPv6().sin6_addr = in6addr_any;
+            break;
+        }
+    }
+
     /**
      * Retreive the port (in host byte order) or 0 if the address is not
      * of a supported family.
@@ -175,6 +191,12 @@ public:
         }
     }
 
+    /**
+     * Set the address part of the socket address from a numeric IP address (string representation).
+     * Family must be already set. Throws in case of parse failue.
+     */
+    void setAddress(const char* address);
+
     /**
      * Returns the accessible byte length at the pointer returned by #get().
      * If zero, #get() returns null.

src/Makefile.am

@@ -26,7 +26,9 @@ libopendht_la_SOURCES  = \
         securedht.cpp \
         dhtrunner.cpp \
         default_types.cpp \
-        log.cpp
+        log.cpp \
+        network_utils.h \
+        network_utils.cpp
 
 if WIN32
 libopendht_la_SOURCES += rng.cpp

src/dht.cpp

@@ -986,19 +986,15 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
 {
     scheduler.syncTime();
 
-    auto q = std::make_shared<Query>(Select {}, std::move(where));
     auto op = std::make_shared<GetStatus<std::map<Value::Id, Sp<Value>>>>();
-    auto f = filter.chain(q->where.getFilter());
-
-    auto gcb = [getcb, donecb, op, f](const std::vector<Sp<Value>>& vals) {
+    auto gcb = [getcb, donecb, op](const std::vector<Sp<Value>>& vals) {
         auto& o = *op;
-        return callbackWrapper(getcb, donecb, vals, [&o,&f](const std::vector<Sp<Value>>& values) {
+        return callbackWrapper(getcb, donecb, vals, [&o](const std::vector<Sp<Value>>& values) {
             std::vector<Sp<Value>> newvals {};
             for (const auto& v : values) {
                 auto it = o.values.find(v->id);
                 if (it == o.values.cend() or (it->second != v && !(*it->second == *v))) {
-                    if (not f or f(*v))
-                        newvals.push_back(v);
+                    newvals.push_back(v);
                 }
             }
             return newvals;
@@ -1008,6 +1004,9 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
         }, o);
     };
 
+    auto q = std::make_shared<Query>(Select {}, std::move(where));
+    auto f = filter.chain(q->where.getFilter());
+
     /* Try to answer this search locally. */
     gcb(getLocal(id, f));
 
@@ -1075,7 +1074,7 @@ void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Quer
 }
 
 std::vector<Sp<Value>>
-Dht::getLocal(const InfoHash& id, Value::Filter f) const
+Dht::getLocal(const InfoHash& id, const Value::Filter& f) const
 {
     auto s = store.find(id);
     if (s == store.end()) return {};
@@ -1092,17 +1091,14 @@ Dht::getLocalById(const InfoHash& id, Value::Id vid) const
 }
 
 std::vector<Sp<Value>>
-Dht::getPut(const InfoHash& id)
+Dht::getPut(const InfoHash& id) const
 {
     std::vector<Sp<Value>> ret;
-    auto find_values = [&](std::map<InfoHash, Sp<Search>> srs) {
+    auto find_values = [&](const std::map<InfoHash, Sp<Search>>& srs) {
         auto srp = srs.find(id);
-        if (srp == srs.end())
-            return;
-        auto& search = srp->second;
-        ret.reserve(ret.size() + search->announce.size());
-        for (const auto& a : search->announce)
-            ret.push_back(a.value);
+        if (srp == srs.end()) return;
+        auto vals = srp->second->getPut();
+        ret.insert(ret.end(), vals.begin(), vals.end());
     };
     find_values(searches4);
     find_values(searches6);
@@ -1110,18 +1106,11 @@ Dht::getPut(const InfoHash& id)
 }
 
 Sp<Value>
-Dht::getPut(const InfoHash& id, const Value::Id& vid)
+Dht::getPut(const InfoHash& id, const Value::Id& vid) const
 {
-    auto find_value = [&](std::map<InfoHash, Sp<Search>> srs) {
+    auto find_value = [&](const std::map<InfoHash, Sp<Search>>& srs) {
         auto srp = srs.find(id);
-        if (srp == srs.end())
-            return Sp<Value> {};
-        auto& search = srp->second;
-        for (auto& a : search->announce) {
-            if (a.value->id == vid)
-                return a.value;
-        }
-        return Sp<Value> {};
+        return (srp != srs.end()) ? srp->second->getPut(vid) : Sp<Value> {};
     };
     if (auto v4 = find_value(searches4))
         return v4;
@@ -1134,29 +1123,17 @@ bool
 Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
 {
     bool canceled {false};
-    if (storageErase(id, vid))
-        canceled = true;
-    auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>> srs) {
+    auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>>& srs) {
         auto srp = srs.find(id);
-        if (srp == srs.end())
-            return;
-
-        auto& sr = srp->second;
-        for (auto it = sr->announce.begin(); it != sr->announce.end();) {
-            if (it->value->id == vid) {
-                canceled = true;
-                it = sr->announce.erase(it);
-            }
-            else
-                ++it;
-        }
+        return (srp != srs.end()) ? srp->second->cancelPut(vid) : false;
     };
-    sr_cancel_put(searches4);
-    sr_cancel_put(searches6);
+    canceled |= sr_cancel_put(searches4);
+    canceled |= sr_cancel_put(searches6);
+    if (canceled)
+        storageErase(id, vid);
     return canceled;
 }
 
-
 // Storage
 
 void
@@ -2016,14 +1993,15 @@ Dht::exportValues() const
 void
 Dht::importValues(const std::vector<ValuesExport>& import)
 {
-    for (const auto& h : import) {
-        if (h.second.empty())
+    const auto& now = scheduler.time();
+
+    for (const auto& node : import) {
+        if (node.second.empty())
             continue;
 
-        const auto& now = scheduler.time();
         try {
             msgpack::unpacked msg;
-            msgpack::unpack(msg, (const char*)h.second.data(), h.second.size());
+            msgpack::unpack(msg, (const char*)node.second.data(), node.second.size());
             auto valarr = msg.get();
             if (valarr.type != msgpack::type::ARRAY)
                 throw msgpack::type_error();
@@ -2037,14 +2015,14 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                     val_time = time_point{time_point::duration{valel.via.array.ptr[0].as<time_point::duration::rep>()}};
                     tmp_val.msgpack_unpack(valel.via.array.ptr[1]);
                 } catch (const std::exception&) {
-                    DHT_LOG.e(h.first, "Error reading value at %s", h.first.toString().c_str());
+                    DHT_LOG.e(node.first, "Error reading value at %s", node.first.toString().c_str());
                     continue;
                 }
                 val_time = std::min(val_time, now);
-                storageStore(h.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
+                storageStore(node.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
             }
         } catch (const std::exception&) {
-            DHT_LOG.e(h.first, "Error reading values at %s", h.first.toString().c_str());
+            DHT_LOG.e(node.first, "Error reading values at %s", node.first.toString().c_str());
             continue;
         }
     }

src/dht_proxy_client.cpp

@@ -110,7 +110,7 @@ DhtProxyClient::~DhtProxyClient()
 }
 
 std::vector<Sp<Value>>
-DhtProxyClient::getLocal(const InfoHash& k, Value::Filter filter) const {
+DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const {
     std::lock_guard<std::mutex> lock(searchLock_);
     auto s = searches_.find(k);
     if (s == searches_.end())
@@ -445,7 +445,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_
  * Get data currently being put at the given hash.
  */
 std::vector<Sp<Value>>
-DhtProxyClient::getPut(const InfoHash& key) {
+DhtProxyClient::getPut(const InfoHash& key) const {
     std::vector<Sp<Value>> ret;
     auto search = searches_.find(key);
     if (search != searches_.end()) {
@@ -460,7 +460,7 @@ DhtProxyClient::getPut(const InfoHash& key) {
  * Get data currently being put at the given hash with the given id.
  */
 Sp<Value>
-DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) {
+DhtProxyClient::getPut(const InfoHash& key, const Value::Id& id) const {
     auto search = searches_.find(key);
     if (search == searches_.end())
         return {};

src/dhtrunner.cpp

@@ -20,26 +20,13 @@
 
 #include "dhtrunner.h"
 #include "securedht.h"
+#include "peer_discovery.h"
+#include "network_utils.h"
 
 #ifdef OPENDHT_PROXY_CLIENT
 #include "dht_proxy_client.h"
 #endif
 
-#ifndef _WIN32
-#include <unistd.h>
-#else
-#include <io.h>
-#endif
-
-#ifndef _WIN32
-#include <sys/socket.h>
-#else
-#include <winsock2.h>
-#include <ws2tcpip.h>
-#define close(x) closesocket(x)
-#define write(s, b, f) send(s, b, (int)strlen(b), 0)
-#endif
-
 namespace dht {
 
 constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
@@ -147,6 +134,37 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const DhtRunner::
                 cv.wait_until(lk, wakeup, hasJobToDo);
         }
     });
+
+    if (config.peer_discovery or config.peer_publish) {
+        try {
+            peerDiscovery4_.reset(new PeerDiscovery(AF_INET, port_multicast));
+        } catch(std::exception &exception){
+            perror(exception.what());
+        }
+        try{
+            peerDiscovery6_.reset(new PeerDiscovery(AF_INET6, port_multicast));
+        } catch(std::exception &exception){
+            perror(exception.what());
+        }
+    }
+    if (config.peer_discovery) {
+        using sig = void (DhtRunner::*)(const InfoHash&, const SockAddr&);
+        if (peerDiscovery4_)
+            peerDiscovery4_->startDiscovery(std::bind(static_cast<sig>(&DhtRunner::bootstrap),
+                                                   this,
+                                                   std::placeholders::_1,std::placeholders::_2));
+
+        if (peerDiscovery6_)
+            peerDiscovery6_->startDiscovery(std::bind(static_cast<sig>(&DhtRunner::bootstrap),
+                                                    this,
+                                                    std::placeholders::_1,std::placeholders::_2));
+    }
+    if (config.peer_publish) {
+        if (peerDiscovery4_)
+            peerDiscovery4_->startPublish(getNodeId(), getBoundPort(AF_INET));
+        if (peerDiscovery6_)
+            peerDiscovery6_->startPublish(getNodeId(), getBoundPort(AF_INET6));
+    }
 }
 
 void
@@ -169,12 +187,21 @@ DhtRunner::join()
     running = false;
     cv.notify_all();
     bootstrap_cv.notify_all();
+    if (peerDiscovery4_)
+        peerDiscovery4_->stop();
+    if (peerDiscovery6_)
+        peerDiscovery6_->stop();
+
     if (dht_thread.joinable())
         dht_thread.join();
     if (bootstrap_thread.joinable())
         bootstrap_thread.join();
     if (rcv_thread.joinable())
         rcv_thread.join();
+    if (peerDiscovery4_)
+        peerDiscovery4_->join();
+    if (peerDiscovery6_)
+        peerDiscovery6_->join();
 
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
@@ -471,34 +498,6 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
     return sock;
 }
 
-#ifdef _WIN32
-inline void udpPipe(int fds[2])
-{
-    int lst = socket(AF_INET, SOCK_DGRAM, 0);
-    if (lst < 0)
-        throw DhtException(std::string("Can't open socket: ") + strerror(lst));
-    sockaddr_in inaddr;
-    sockaddr addr;
-    memset(&inaddr, 0, sizeof(inaddr));
-    memset(&addr, 0, sizeof(addr));
-    inaddr.sin_family = AF_INET;
-    inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
-    inaddr.sin_port = 0;
-    int yes=1;
-    setsockopt(lst, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
-    int rc = bind(lst, (sockaddr*)&inaddr, sizeof(inaddr));
-    if (rc < 0) {
-        close(lst);
-        throw DhtException("Can't bind socket on " + print_addr((sockaddr*)&addr, sizeof(inaddr)) + " " + strerror(rc));
-    }
-    socklen_t len = sizeof(inaddr);
-    getsockname(lst, &addr, &len);
-    fds[0] = lst;
-    fds[1] = socket(AF_INET, SOCK_DGRAM, 0);
-    connect(fds[1], &addr, len);
-}
-#endif
-
 void
 DhtRunner::stopNetwork()
 {
@@ -524,7 +523,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
         throw DhtException(std::string("Can't open pipe: ") + strerror(errno));
     }
 #else
-    udpPipe(stopfds);
+    net::udpPipe(stopfds);
 #endif
     int stop_readfd = stopfds[0];
     stop_writefd = stopfds[1];
@@ -569,7 +568,8 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                 if(s6 >= 0)
                     FD_SET(s6, &readfds);
 
-                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, nullptr);
+                int selectFd = std::max({s4, s6, stop_readfd}) + 1;
+                int rc = select(selectFd, &readfds, nullptr, nullptr, nullptr);
                 if(rc < 0) {
                     if(errno != EINTR) {
                         perror("select");
@@ -591,6 +591,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                         rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
                     else
                         continue;
+
                     if (rc > 0) {
                         {
                             std::lock_guard<std::mutex> lck(sock_mtx);
@@ -934,6 +935,18 @@ DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
     cv.notify_all();
 }
 
+void
+DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
+{
+    {
+        std::unique_lock<std::mutex> lck(storage_mtx);
+        pending_ops_prio.emplace([id, address](SecureDht& dht) mutable {
+            dht.insertNode(id, address);
+        });
+    }
+    cv.notify_all();
+}
+
 void
 DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
 {
@@ -973,6 +986,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
 void
 DhtRunner::resetDht()
 {
+    peerDiscovery4_.reset();
+    peerDiscovery6_.reset();
 #ifdef OPENDHT_PROXY_CLIENT
     listeners_.clear();
     dht_via_proxy_.reset();

src/network_engine.cpp

@@ -23,6 +23,7 @@
 #include "default_types.h"
 #include "log_enable.h"
 #include "parsed_message.h"
+#include "network_utils.h"
 
 #include <msgpack.hpp>
 
@@ -61,32 +62,6 @@ static const uint8_t v4prefix[16] = {
 
 constexpr unsigned SEND_NODES {8};
 
-#ifdef _WIN32
-
-static bool
-set_nonblocking(int fd, int nonblocking)
-{
-    unsigned long mode = !!nonblocking;
-    int rc = ioctlsocket(fd, FIONBIO, &mode);
-    return rc == 0;
-}
-
-extern const char *inet_ntop(int, const void *, char *, socklen_t);
-
-#else
-
-static bool
-set_nonblocking(int fd, int nonblocking)
-{
-    int rc = fcntl(fd, F_GETFL, 0);
-    if (rc < 0)
-        return false;
-    rc = fcntl(fd, F_SETFL, nonblocking?(rc | O_NONBLOCK):(rc & ~O_NONBLOCK));
-    return rc >= 0;
-}
-
-#endif
-
 
 /* Transaction-ids are 4-bytes long, with the first two bytes identifying
  * the kind of request, and the remaining two a sequence number in
@@ -165,11 +140,11 @@ NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, const int& s, const int&
     myid(myid), network(net), dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
 {
     if (dht_socket >= 0) {
-        if (!set_nonblocking(dht_socket, 1))
+        if (!set_nonblocking(dht_socket))
             throw DhtException("Can't set socket to non-blocking mode");
     }
     if (dht_socket6 >= 0) {
-        if (!set_nonblocking(dht_socket6, 1))
+        if (!set_nonblocking(dht_socket6))
             throw DhtException("Can't set socket to non-blocking mode");
     }
 }

src/network_utils.cpp

@@ -0,0 +1,82 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "network_utils.h"
+
+#ifndef _WIN32
+#include <arpa/inet.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#else
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#include <io.h>
+#define close(x) closesocket(x)
+#define write(s, b, f) send(s, b, (int)strlen(b), 0)
+#endif
+#include <fcntl.h>
+
+namespace dht {
+namespace net {
+
+bool
+set_nonblocking(int fd, bool nonblocking)
+{
+#ifdef _WIN32
+    unsigned long mode = !!nonblocking;
+    int rc = ioctlsocket(fd, FIONBIO, &mode);
+    return rc == 0;
+#else
+    int rc = fcntl(fd, F_GETFL, 0);
+    if (rc < 0)
+        return false;
+    rc = fcntl(fd, F_SETFL, nonblocking ? (rc | O_NONBLOCK) : (rc & ~O_NONBLOCK));
+    return rc >= 0;
+#endif
+}
+
+#ifdef _WIN32
+void udpPipe(int fds[2])
+{
+    int lst = socket(AF_INET, SOCK_DGRAM, 0);
+    if (lst < 0)
+        throw DhtException(std::string("Can't open socket: ") + strerror(lst));
+    sockaddr_in inaddr;
+    sockaddr addr;
+    memset(&inaddr, 0, sizeof(inaddr));
+    memset(&addr, 0, sizeof(addr));
+    inaddr.sin_family = AF_INET;
+    inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    inaddr.sin_port = 0;
+    int yes=1;
+    setsockopt(lst, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
+    int rc = bind(lst, (sockaddr*)&inaddr, sizeof(inaddr));
+    if (rc < 0) {
+        close(lst);
+        throw DhtException("Can't bind socket on " + print_addr((sockaddr*)&addr, sizeof(inaddr)) + " " + strerror(rc));
+    }
+    socklen_t len = sizeof(inaddr);
+    getsockname(lst, &addr, &len);
+    fds[0] = lst;
+    fds[1] = socket(AF_INET, SOCK_DGRAM, 0);
+    connect(fds[1], &addr, len);
+}
+#endif
+
+}
+}

src/network_utils.h

@@ -0,0 +1,37 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+#pragma once
+
+#ifdef _WIN32
+#include <winsock2.h>
+#define close(x) closesocket(x)
+#define write(s, b, f) send(s, b, (int)strlen(b), 0)
+#endif
+
+namespace dht {
+namespace net {
+
+bool set_nonblocking(int fd, bool nonblocking = true);
+
+#ifdef _WIN32
+void udpPipe(int fds[2]);
+// extern const char *inet_ntop(int, const void *, char *, socklen_t);
+#endif
+
+}
+}

src/peer_discovery.cpp

@@ -0,0 +1,321 @@
+/*
+ *  Copyright (C) 2014-2019 Savoir-faire Linux Inc.
+ *  Author(s) : Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "peer_discovery.h"
+#include "network_utils.h"
+
+#ifdef _WIN32
+#include <Ws2tcpip.h> // needed for ip_mreq definition for multicast
+#include <Windows.h>
+#else
+#include <sys/types.h>
+#endif
+#include <fcntl.h>
+
+namespace dht {
+
+constexpr char MULTICAST_ADDRESS_IPV4[10] = "224.0.0.1";
+constexpr char MULTICAST_ADDRESS_IPV6[8] = "ff05::2"; // Site-local multicast
+
+PeerDiscovery::PeerDiscovery(sa_family_t domain, in_port_t port)
+    : domain_(domain), port_(port), sockfd_(initialize_socket(domain))
+{
+    socketJoinMulticast(sockfd_, domain);
+}
+
+PeerDiscovery::~PeerDiscovery()
+{
+    if (sockfd_ != -1)
+        close(sockfd_);
+
+#ifdef _WIN32
+    WSACleanup();
+#endif
+}
+
+int
+PeerDiscovery::initialize_socket(sa_family_t domain)
+{
+#ifdef _WIN32
+    WSADATA wsaData;
+    if (WSAStartup(0x0101, &wsaData)) {
+        throw std::runtime_error(std::string("Can't initialize Winsock2 ") + strerror(errno));
+    }
+#endif
+
+    int sockfd = socket(domain, SOCK_DGRAM, 0);
+    if (sockfd < 0) {
+        throw std::runtime_error(std::string("Socket Creation Error: ") + strerror(errno));
+    }
+    net::set_nonblocking(sockfd);
+    return sockfd;
+}
+
+void
+PeerDiscovery::listener_setup()
+{
+    SockAddr sockAddrListen_;
+    sockAddrListen_.setFamily(domain_);
+    sockAddrListen_.setPort(port_);
+    sockAddrListen_.setAny();
+
+    unsigned int opt = 1;
+    if (setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR|SO_REUSEPORT, (char*) &opt, sizeof(opt)) < 0){
+       throw std::runtime_error(std::string("Reusing ADDR failed: ") + strerror(errno));
+    }
+
+    // bind to receive address
+    if (bind(sockfd_, sockAddrListen_.get(), sockAddrListen_.getLength()) < 0){
+        throw std::runtime_error(std::string("Bind Socket For Listener Error: ") + strerror(errno));
+    }
+}
+
+void
+PeerDiscovery::socketJoinMulticast(int sockfd, sa_family_t family)
+{
+    switch (family)
+    {
+    case AF_INET:{
+        ip_mreq config_ipv4;
+
+        //This option can be used to set the interface for sending outbound
+        //multicast datagrams from the sockets application.
+        config_ipv4.imr_interface.s_addr = htonl(INADDR_ANY);
+        if( setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_IF, &config_ipv4.imr_interface, sizeof( struct in_addr )) < 0 ) {
+            throw std::runtime_error(std::string("Bound Network Interface IPV4 Error: ") + strerror(errno));
+        }
+
+        //The IP_MULTICAST_TTL socket option allows the application to primarily
+        //limit the lifetime of the packet in the Internet and prevent it from circulating indefinitely
+        unsigned char ttl4 = 20;
+        if( setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl4, sizeof( ttl4 )) < 0 ) {
+            throw std::runtime_error(std::string(" TTL Sockopt Error: ") + strerror(errno));
+        }
+
+        // config the listener to be interested in joining in the multicast group
+        config_ipv4.imr_multiaddr.s_addr = inet_addr(MULTICAST_ADDRESS_IPV4);
+        config_ipv4.imr_interface.s_addr = htonl(INADDR_ANY);
+        if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char*)&config_ipv4, sizeof(config_ipv4)) < 0){
+            throw std::runtime_error(std::string(" Member Addition IPV4 Error: ") + strerror(errno));
+        }
+        break;
+    }
+    case AF_INET6: {
+        ipv6_mreq config_ipv6;
+
+        unsigned int outif = 0;
+        if( setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_IF, &outif, sizeof( outif )) < 0 ) {
+            throw std::runtime_error(std::string("Bound Network Interface IPV6 Error: ") + strerror(errno));
+        }
+
+        unsigned int ttl6 = 20;
+        if( setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &ttl6, sizeof( ttl6 )) < 0 ) {
+            throw std::runtime_error(std::string("Hop Count Set Error: ") + strerror(errno));
+        }
+
+        config_ipv6.ipv6mr_interface = 0;
+        inet_pton(AF_INET6, MULTICAST_ADDRESS_IPV6, &config_ipv6.ipv6mr_multiaddr);
+        if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &config_ipv6, sizeof(config_ipv6)) < 0){
+            throw std::runtime_error(std::string("Member Addition IPV6 Error: ") + strerror(errno));
+        }
+        break;
+    }
+    }
+}
+
+void
+PeerDiscovery::sendTo(uint8_t *buf, size_t buf_size)
+{
+    ssize_t nbytes = sendto(
+        sockfd_,
+        buf,
+        buf_size,
+        0,
+        sockAddrSend_.get(),
+        sockAddrSend_.getLength()
+    );
+    if (nbytes < 0) {
+        throw std::runtime_error(std::string("Error sending packet: ") + strerror(errno));
+    }
+}
+
+SockAddr
+PeerDiscovery::recvFrom(uint8_t *buf, size_t& buf_size)
+{
+    sockaddr_storage storeage_recv;
+    socklen_t sa_len = sizeof(storeage_recv);
+
+    ssize_t nbytes = recvfrom(
+        sockfd_,
+        buf,
+        buf_size,
+        0,
+        (sockaddr*)&storeage_recv,
+        &sa_len
+    );
+    if (nbytes < 0) {
+        throw std::runtime_error(std::string("Error receiving packet: ") + strerror(errno));
+    }
+
+    buf_size = nbytes;
+    SockAddr ret {storeage_recv, sa_len};
+    return ret;
+}
+
+void
+PeerDiscovery::sender_setup(const dht::InfoHash& nodeId, in_port_t port_to_send)
+{
+    nodeId_ = nodeId;
+    //Set up for Sender
+    sockAddrSend_.setFamily(domain_);
+    sockAddrSend_.setAddress(domain_ == AF_INET ? MULTICAST_ADDRESS_IPV4 : MULTICAST_ADDRESS_IPV6);
+    sockAddrSend_.setPort(port_);
+
+    //Setup for send data
+    int port_node = port_to_send;
+    uint8_t port_node_binary[2];
+    PeerDiscovery::inttolitend(port_node,port_node_binary);
+
+    //Copy Node id and node port
+    memcpy (data_send_.data(), nodeId.data(), nodeId.size());
+    data_send_[InfoHash::size()] = port_node_binary[0];
+    data_send_[InfoHash::size() + 1] = port_node_binary[1];
+}
+
+void
+PeerDiscovery::sender_thread()
+{
+    while(true) {
+        try {
+            sendTo(data_send_.data(), data_send_.size());
+        } catch (const std::exception& e) {
+            std::cerr << e.what() << std::endl;
+        }
+        {
+            std::unique_lock<std::mutex> lck(mtx_);
+            if (cv_.wait_for(lck,std::chrono::seconds(3),[&]{ return !running_; }))
+                break;
+        }
+    }
+}
+
+void
+PeerDiscovery::listener_thread(PeerDiscoveredCallback callback)
+{
+    int stopfds_pipe[2];
+#ifndef _WIN32
+    auto status = pipe(stopfds_pipe);
+    if (status == -1) {
+        throw std::runtime_error(std::string("Can't open pipe: ") + strerror(errno));
+    }
+#else
+    net::udpPipe(stopfds_pipe);
+#endif
+    int stop_readfd = stopfds_pipe[0];
+    stop_writefd_ = stopfds_pipe[1];
+
+    while(true) {
+        fd_set readfds;
+
+        FD_ZERO(&readfds);
+        FD_SET(stop_readfd, &readfds);
+        FD_SET(sockfd_, &readfds);
+
+        int data_coming = select(sockfd_ > stop_readfd ? sockfd_ + 1 : stop_readfd + 1, &readfds, nullptr, nullptr, nullptr);
+
+        {
+            std::unique_lock<std::mutex> lck(mtx_);
+            if (not running_)
+                break;
+        }
+
+        if (data_coming < 0) {
+            if(errno != EINTR) {
+                perror("Select Error");
+                std::this_thread::sleep_for( std::chrono::seconds(1) );
+            }
+        }
+
+        if (data_coming > 0) {
+
+            if(FD_ISSET(stop_readfd, &readfds)){ break; }
+
+            std::array<uint8_t,dht::InfoHash::size() + sizeof(in_port_t)> data_receive;
+            size_t data_receive_size = data_receive.size();
+            auto from = recvFrom(data_receive.data(), data_receive_size);
+
+            //Data_receive_size as a value-result member will hlep to filter packs
+            if(data_receive_size != data_receive.size()){
+                perror("Data Received Unmatch");
+                continue;
+            }
+
+            std::array<uint8_t,dht::InfoHash::size()> data_infohash;
+            uint8_t data_port[2];
+
+            memcpy (data_infohash.data(), data_receive.data(), dht::InfoHash::size());
+            data_port[0] = data_receive[dht::InfoHash::size()];
+            data_port[1] = data_receive[dht::InfoHash::size() + 1];
+
+            auto port = PeerDiscovery::litendtoint(data_port);
+            auto nodeId = dht::InfoHash(data_infohash.data(), dht::InfoHash::size());
+
+            if (nodeId != nodeId_){
+                from.setPort(port);
+                callback(nodeId, from);
+            }
+        }
+    }
+    if (stop_readfd != -1)
+        close(stop_readfd);
+    if (stop_writefd_ != -1) {
+        close(stop_writefd_);
+        stop_writefd_ = -1;
+    }
+}
+
+void
+PeerDiscovery::startDiscovery(PeerDiscoveredCallback callback)
+{
+    listener_setup();
+    running_listen = std::thread(&PeerDiscovery::listener_thread, this, callback);
+}
+
+void
+PeerDiscovery::startPublish(const dht::InfoHash &nodeId, in_port_t port_to_send)
+{
+    sender_setup(nodeId, port_to_send);
+    running_send = std::thread(&PeerDiscovery::sender_thread, this);
+}
+
+void
+PeerDiscovery::stop()
+{
+    {
+        std::unique_lock<std::mutex> lck(mtx_);
+        running_ = false;
+    }
+    cv_.notify_one();
+    if (stop_writefd_ != -1) {
+        if (write(stop_writefd_, "\0", 1) == -1) {
+            perror("write");
+        }
+    }
+}
+
+}

src/search.h

@@ -532,6 +532,34 @@ struct Dht::Search {
         scheduler.edit(opExpirationJob, cache.getExpiration());
     }
 
+    std::vector<Sp<Value>> getPut() const {
+        std::vector<Sp<Value>> ret;
+        ret.reserve(announce.size());
+        for (const auto& a : announce)
+            ret.push_back(a.value);
+        return ret;
+    }
+
+    Sp<Value> getPut(Value::Id vid) const {
+        for (auto& a : announce) {
+            if (a.value->id == vid)
+                return a.value;
+        }
+    }
+
+    bool cancelPut(Value::Id vid) {
+        bool canceled {false};
+        for (auto it = announce.begin(); it != announce.end();) {
+            if (it->value->id == vid) {
+                canceled = true;
+                it = announce.erase(it);
+            }
+            else
+                ++it;
+        }
+        return canceled;
+    }
+
     /**
      * @return The number of non-good search nodes.
      */

src/storage.h

@@ -115,7 +115,7 @@ struct Storage {
         return {};
     }
 
-    std::vector<Sp<Value>> get(Value::Filter f = {}) const {
+    std::vector<Sp<Value>> get(const Value::Filter& f = {}) const {
         std::vector<Sp<Value>> newvals {};
         if (not f) newvals.reserve(values.size());
         for (auto& v : values) {

src/utils.cpp

@@ -73,6 +73,22 @@ SockAddr::resolve(const std::string& host, const std::string& service)
     return ips;
 }
 
+void
+SockAddr::setAddress(const char* address)
+{
+    auto family = getFamily();
+    void* addr = nullptr;
+    switch (family) {
+    case AF_INET:
+        addr = &getIPv4().sin_addr;
+        break;
+    case AF_INET6:
+        addr = &getIPv6().sin6_addr;
+        break;
+    }
+    if (not addr or inet_pton(family, address, addr) <= 0)
+        throw std::runtime_error("Can't parse IP address");
+}
 
 std::string
 print_addr(const sockaddr* sa, socklen_t slen)

tests/dhtproxytester.cpp

@@ -160,20 +160,25 @@ DhtProxyTester::testResubscribeGetValues() {
 
     // For the second subscribe, the proxy will return the value in the body
     auto values = std::vector<dht::Blob>();
-    nodeClient->listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool) {
-        for (const auto& value : v)
-            values.emplace_back(value->data);
-        done = true;
+    auto ftoken = nodeClient->listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool expired) {
+        if (not expired) {
+            std::lock_guard<std::mutex> lk(cv_m);
+            for (const auto& value : v)
+                values.emplace_back(value->data);
+            done = true;
+        }
         cv.notify_all();
         return true;
     });
 
     cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
+    auto token = ftoken.get();
+    CPPUNIT_ASSERT(token);
+    nodeClient->cancelListen(key, token);
     done = false;
     // Here values should still contains 1 values
-    CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 1);
-    CPPUNIT_ASSERT(values.front() == firstVal_data);
-
+    CPPUNIT_ASSERT_EQUAL((size_t)1u, values.size());
+    CPPUNIT_ASSERT(firstVal_data == values.front());
 }
 
 }  // namespace test

tests/dhtrunnertester.cpp

@@ -19,9 +19,9 @@
 
 #include "dhtrunnertester.h"
 
-// std
-#include <iostream>
-#include <string>
+#include <chrono>
+#include <mutex>
+#include <condition_variable>
 
 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
@@ -62,35 +62,58 @@ DhtRunnerTester::testGetPut() {
 
 void
 DhtRunnerTester::testListen() {
+    std::mutex mutex;
+    std::condition_variable cv;
     std::atomic_uint valueCount(0);
-    std::atomic_uint putCount(0);
+    unsigned putCount(0);
+    unsigned putOkCount(0);
 
     auto a = dht::InfoHash::get("234");
     auto b = dht::InfoHash::get("2345");
     auto c = dht::InfoHash::get("23456");
-    constexpr unsigned N = 32;
+    constexpr unsigned N = 64;
 
-    auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&){
+    auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&) {
         valueCount++;
         return true;
     });
 
-    auto ftokenb = node1.listen(b, [&](const std::shared_ptr<dht::Value>&){
+    auto ftokenb = node1.listen(b, [&](const std::shared_ptr<dht::Value>&) {
         valueCount++;
         return false;
     });
 
-    auto ftokenc = node1.listen(c, [&](const std::shared_ptr<dht::Value>&){
+    auto ftokenc = node1.listen(c, [&](const std::shared_ptr<dht::Value>&) {
         valueCount++;
         return true;
     });
 
     for (unsigned i=0; i<N; i++) {
-        node2.put(a, dht::Value("v1"), [&](bool ok) { if (ok) putCount++; });
-        node2.put(b, dht::Value("v2"), [&](bool ok) { if (ok) putCount++; });
+        node2.put(a, dht::Value("v1"), [&](bool ok) {
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                putCount++;
+                if (ok) putOkCount++;
+            }
+            cv.notify_all();
+        });
+        node2.put(b, dht::Value("v2"), [&](bool ok) {
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                putCount++;
+                if (ok) putOkCount++;
+            }
+            cv.notify_all();
+        });
+    }
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        cv.wait_for(lk, std::chrono::seconds(30), [&]{ return putCount == N * 2u; });
+        CPPUNIT_ASSERT_EQUAL(N * 2u, putCount);
+        CPPUNIT_ASSERT_EQUAL(N * 2u, putOkCount);
     }
 
-    std::this_thread::sleep_for(std::chrono::milliseconds(100));
     auto tokena = ftokena.get();
     auto tokenb = ftokenb.get();
     auto tokenc = ftokenc.get();
@@ -98,8 +121,11 @@ DhtRunnerTester::testListen() {
     CPPUNIT_ASSERT(tokena);
     CPPUNIT_ASSERT(tokenb);
     CPPUNIT_ASSERT(tokenc);
-    CPPUNIT_ASSERT_EQUAL(N * 2u, putCount.load());
     CPPUNIT_ASSERT_EQUAL(N + 1u, valueCount.load());
+
+    node1.cancelListen(a, tokena);
+    node1.cancelListen(b, tokena);
+    node1.cancelListen(c, tokena);
 }
 
 }  // namespace test

tests/peerdiscoverytester.cpp

@@ -0,0 +1,94 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *
+ *  Author: Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "peerdiscoverytester.h"
+
+namespace test {
+
+CPPUNIT_TEST_SUITE_REGISTRATION(PeerDiscoveryTester);
+
+void PeerDiscoveryTester::setUp(){}
+
+void PeerDiscoveryTester::testTransmission_ipv4(){
+
+    // Node for getnode id
+    dht::InfoHash data_n = dht::InfoHash::get("applepin");
+    int port = 2222;
+    in_port_t port_n = 50000;
+    try{
+        dht::PeerDiscovery test_n(AF_INET, port);
+        dht::PeerDiscovery test_s(AF_INET, port);
+        try{
+            test_s.startDiscovery([&](const dht::InfoHash& node, const dht::SockAddr& addr){
+                CPPUNIT_ASSERT_EQUAL(data_n, node);
+                CPPUNIT_ASSERT_EQUAL(port_n, addr.getPort());
+            });
+
+            test_n.startPublish(data_n,port_n);
+
+            sleep(5);
+            test_n.stop();
+            test_s.stop();
+            test_n.join();
+            test_s.join();
+        } catch(std::exception &exception){
+            perror(exception.what());
+            CPPUNIT_ASSERT(false);
+        }
+    } catch(std::exception &exception){
+            perror(exception.what());
+    }
+}
+
+void PeerDiscoveryTester::testTransmission_ipv6(){
+
+    // Node for getnode id
+    dht::InfoHash data_n = dht::InfoHash::get("applepin");
+    int port = 3333;
+    in_port_t port_n = 50001;
+    try{
+        dht::PeerDiscovery test_n(AF_INET6,port);
+        dht::PeerDiscovery test_s(AF_INET6,port);
+
+        try{
+            test_s.startDiscovery([&](const dht::InfoHash& node, const dht::SockAddr& addr){
+                CPPUNIT_ASSERT_EQUAL(data_n, node);
+                CPPUNIT_ASSERT_EQUAL(port_n, addr.getPort());
+            });
+
+            test_n.startPublish(data_n,port_n);
+
+            sleep(5);
+            test_n.stop();
+            test_s.stop();
+            test_n.join();
+            test_s.join();
+        } catch(std::exception &exception){
+            perror(exception.what());
+            CPPUNIT_ASSERT(false);
+        }
+    } catch(std::exception &exception){
+            perror(exception.what());
+    }
+
+}
+
+void PeerDiscoveryTester::tearDown(){}
+
+}  // namespace test

tests/peerdiscoverytester.h

@@ -0,0 +1,56 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *
+ *  Author: Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include "opendht/peer_discovery.h"
+
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+namespace test {
+
+class PeerDiscoveryTester : public CppUnit::TestFixture {
+
+    CPPUNIT_TEST_SUITE(PeerDiscoveryTester);
+    CPPUNIT_TEST(testTransmission_ipv4);
+    CPPUNIT_TEST(testTransmission_ipv6);
+    CPPUNIT_TEST_SUITE_END();
+
+ public:
+    /**
+     * Method automatically called before each test by CppUnit
+     */
+    void setUp();
+    /**
+     * Method automatically called after each test CppUnit
+     */
+    void tearDown();
+    /**
+     * Test Multicast Transmission Ipv4
+     */
+    void testTransmission_ipv4();
+    /**
+     * Test Multicast Transmission Ipv6
+     */
+    void testTransmission_ipv6();
+
+};
+
+}  // namespace test

tools/dhtnode.cpp

@@ -505,6 +505,8 @@ main(int argc, char **argv)
         config.threaded = true;
         config.proxy_server = params.proxyclient;
         config.push_node_id = "dhtnode";
+        config.peer_discovery = params.peer_discovery;
+        config.peer_publish = params.peer_discovery;
         if (not params.proxyclient.empty())
             dht->setPushNotificationToken(params.devicekey);
 

tools/tools_common.h

@@ -98,6 +98,7 @@ struct dht_params {
     bool generate_identity {false};
     bool daemonize {false};
     bool service {false};
+    bool peer_discovery {false};
     std::pair<std::string, std::string> bootstrap {};
     in_port_t proxyserver {0};
     std::string proxyclient {};
@@ -107,29 +108,30 @@ struct dht_params {
 };
 
 static const constexpr struct option long_options[] = {
-   {"help",       no_argument      , nullptr, 'h'},
-   {"port",       required_argument, nullptr, 'p'},
-   {"net",        required_argument, nullptr, 'n'},
-   {"bootstrap",  required_argument, nullptr, 'b'},
-   {"identity",   no_argument      , nullptr, 'i'},
-   {"verbose",    no_argument      , nullptr, 'v'},
-   {"daemonize",  no_argument      , nullptr, 'd'},
-   {"service",    no_argument      , nullptr, 's'},
-   {"persist",    required_argument, nullptr, 'f'},
-   {"logfile",    required_argument, nullptr, 'l'},
-   {"syslog",     no_argument      , nullptr, 'L'},
-   {"proxyserver",required_argument, nullptr, 'S'},
-   {"proxyclient",required_argument, nullptr, 'C'},
-   {"pushserver", required_argument, nullptr, 'P'},
-   {"devicekey",  required_argument, nullptr, 'D'},
-   {nullptr,      0                , nullptr,  0}
+   {"help",             no_argument      , nullptr, 'h'},
+   {"port",             required_argument, nullptr, 'p'},
+   {"net",              required_argument, nullptr, 'n'},
+   {"bootstrap",        required_argument, nullptr, 'b'},
+   {"identity",         no_argument      , nullptr, 'i'},
+   {"verbose",          no_argument      , nullptr, 'v'},
+   {"daemonize",        no_argument      , nullptr, 'd'},
+   {"service",          no_argument      , nullptr, 's'},
+   {"peer-discovery",   no_argument      , nullptr, 'D'},
+   {"persist",          required_argument, nullptr, 'f'},
+   {"logfile",          required_argument, nullptr, 'l'},
+   {"syslog",           no_argument      , nullptr, 'L'},
+   {"proxyserver",      required_argument, nullptr, 'S'},
+   {"proxyclient",      required_argument, nullptr, 'C'},
+   {"pushserver",       required_argument, nullptr, 'y'},
+   {"devicekey",        required_argument, nullptr, 'z'},
+   {nullptr,            0                , nullptr,  0}
 };
 
 dht_params
 parseArgs(int argc, char **argv) {
     dht_params params;
     int opt;
-    while ((opt = getopt_long(argc, argv, "hidsvp:n:b:f:l:", long_options, nullptr)) != -1) {
+    while ((opt = getopt_long(argc, argv, "hidsvDp:n:b:f:l:", long_options, nullptr)) != -1) {
         switch (opt) {
         case 'p': {
                 int port_arg = atoi(optarg);
@@ -147,13 +149,16 @@ parseArgs(int argc, char **argv) {
                     std::cout << "Invalid port: " << port_arg << std::endl;
             }
             break;
-        case 'P':
+        case 'D':
+            params.peer_discovery = true;
+            break;
+        case 'y':
             params.pushserver = optarg;
             break;
         case 'C':
             params.proxyclient = optarg;
             break;
-        case 'D':
+        case 'z':
             params.devicekey = optarg;
             break;
         case 'f':