update to version 1.4.1

* updated MSVC project * bug fixes
Merge pull request #215 from AmarOk1412/typoPeriod
2026-11-10 13:27:12 +00:00 · 2017-10-21 01:30:12 +02:00 · 2017-10-20 23:56:02 +02:00 · 2017-10-20 14:51:35 -04:00 · 2017-10-20 17:46:27 +02:00 · 2017-10-20 17:42:28 +02:00
35 changed files with 826 additions and 1721 deletions
@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 4.0)
+set (opendht_VERSION_MINOR 4.1)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -24,8 +24,6 @@ option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentati
 list (APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake")
 find_package (Threads)
 find_package (PkgConfig)
-pkg_search_module(libuv libuv)
-pkg_search_module(libsodium libsodium)
 find_package (GnuTLS 3.3 REQUIRED)
 pkg_search_module (Nettle nettle)
 find_package (Msgpack 1.2 REQUIRED)
@@ -33,6 +31,7 @@ if (OPENDHT_TOOLS)
    find_package (Readline 6 REQUIRED)
 endif ()
 if (NOT OPENDHT_ARGON2)
+    find_package(PkgConfig)
    pkg_search_module(argon2 libargon2)
    if (argon2_FOUND)
        message("Argon2 found " ${argon2_req})
@@ -44,7 +43,7 @@ if (NOT OPENDHT_ARGON2)
 endif ()

 # Build flags
-set (CMAKE_CXX_STANDARD 14)
+set (CMAKE_CXX_STANDARD 11)
 set (CMAKE_CXX_STANDARD_REQUIRED on)
 set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors -fvisibility=hidden")
 if (OPENDHT_SANITIZE)
@@ -78,13 +77,7 @@ endif ()
 if (Nettle_INCLUDE_DIRS)
    include_directories (SYSTEM "${Nettle_INCLUDE_DIRS}")
 endif ()
-if (libuv_INCLUDE_DIRS)
-    include_directories (SYSTEM "${libuv_INCLUDE_DIRS}")
-endif ()
-if (libsodium_INCLUDE_DIRS)
-    include_directories (SYSTEM "${libsodium_INCLUDE_DIRS}")
-endif ()
-link_directories (${libuv_LIBRARY_DIRS} ${libsodium_LIBRARY_DIRS} ${Nettle_LIBRARY_DIRS})
+link_directories (${Nettle_LIBRARY_DIRS})
 include_directories (
    ./
    include/
@@ -114,6 +107,7 @@ list (APPEND opendht_SOURCES
    src/storage.h
    src/listener.h
    src/search.h
+    src/net.h
    src/parsed_message.h
    src/request.h
    src/callbacks.cpp
@@ -141,7 +135,6 @@ list (APPEND opendht_HEADERS
    include/opendht/routing_table.h
    include/opendht/node_cache.h
    include/opendht/network_engine.h
-    include/opendht/net.h
    include/opendht/scheduler.h
    include/opendht/rate_limiter.h
    include/opendht/securedht.h
@@ -184,7 +177,7 @@ if (OPENDHT_STATIC)
        target_link_libraries(opendht-static ${argon2_LIBRARIES})
        target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
    endif ()
-    target_link_libraries(opendht-static ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${libuv_LIBRARIES} ${libsodium_LIBRARIES})
+    target_link_libraries(opendht-static ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES})
    install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -205,7 +198,7 @@ if (OPENDHT_SHARED)
        target_link_libraries(opendht PRIVATE ${argon2_LIBRARIES})
        target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
    endif ()
-    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${libuv_LIBRARIES} ${libsodium_LIBRARIES})
+    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES})
    install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -39,6 +39,7 @@
    <ClInclude Include="..\include\opendht.h" />
    <ClInclude Include="..\include\opendht\callbacks.h" />
    <ClInclude Include="..\include\opendht\crypto.h" />
+    <ClInclude Include="..\include\opendht\def.h" />
    <ClInclude Include="..\include\opendht\default_types.h" />
    <ClInclude Include="..\include\opendht\dht.h" />
    <ClInclude Include="..\include\opendht\dhtrunner.h" />
@@ -55,9 +56,12 @@
    <ClInclude Include="..\include\opendht\routing_table.h" />
    <ClInclude Include="..\include\opendht\scheduler.h" />
    <ClInclude Include="..\include\opendht\securedht.h" />
+    <ClInclude Include="..\include\opendht\sockaddr.h" />
    <ClInclude Include="..\include\opendht\utils.h" />
    <ClInclude Include="..\include\opendht\value.h" />
+    <ClInclude Include="..\src\net.h" />
    <ClInclude Include="..\src\listener.h" />
+    <ClInclude Include="..\src\parsed_message.h" />
    <ClInclude Include="..\src\search.h" />
    <ClInclude Include="..\src\storage.h" />
  </ItemGroup>
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 4)
-m4_define([opendht_patch_version], 0)
+m4_define([opendht_patch_version], 1)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -49,12 +49,6 @@ struct OPENDHT_PUBLIC NodeStats {
    std::string toString() const;
 };

-
-struct OPENDHT_PUBLIC NetworkConfig {
-    std::string bind_addr {"::0"};
-    in_port_t bind_port {0};
-};
-
 /**
 * Dht configuration.
 */
@@ -74,8 +68,6 @@ struct OPENDHT_PUBLIC Config {

    /** Makes the DHT responsible to maintain its stored values. Consumes more ressources. */
    bool maintain_storage;
-
-    NetworkConfig network_config;
 };

 /**
@@ -95,7 +87,6 @@ using QueryCallback = std::function<bool(const std::vector<std::shared_ptr<Field
 using GetCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values)>;
 using GetCallbackSimple = std::function<bool(std::shared_ptr<Value> value)>;
 using ShutdownCallback = std::function<void()>;
-using StatusCallback = std::function<void(NodeStatus, NodeStatus)>;

 using CertificateStoreQuery = std::function<std::vector<std::shared_ptr<crypto::Certificate>>(const InfoHash& pk_id)>;

@@ -104,7 +95,7 @@ typedef bool (*GetCallbackRaw)(std::shared_ptr<Value>, void *user_data);
 OPENDHT_PUBLIC GetCallbackSimple bindGetCb(GetCallbackRaw raw_cb, void* user_data);
 OPENDHT_PUBLIC GetCallback bindGetCb(GetCallbackSimple cb);

-using DoneCallback = std::function<void(bool success, std::vector<std::shared_ptr<Node>>&& nodes)>;
+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);
@@ -28,8 +28,6 @@ extern "C" {
 #include <gnutls/x509.h>
 }

-#include <sodium.h>
-
 #include <vector>
 #include <memory>

@@ -37,8 +35,6 @@ extern "C" {
 #include <iso646.h>
 #endif

-#include <sys/mman.h>
-
 namespace dht {

 /**
@@ -504,19 +500,11 @@ public:
    secure_vector() {}
    secure_vector(secure_vector<T> const&) = default;
    secure_vector(secure_vector<T> &&) = default;
-    explicit secure_vector(unsigned size): data_(size) {
-        mlock(data_.data(), data_.size()*sizeof(T));
-    }
-    explicit secure_vector(unsigned size, T _item): data_(size, _item) {
-        mlock(data_.data(), data_.size()*sizeof(T));
-    }
-    explicit secure_vector(const std::vector<T>& c): data_(c) {
-        mlock(data_.data(), data_.size()*sizeof(T));
-    }
-    secure_vector(std::vector<T>&& c): data_(std::move(c)) {
-        mlock(data_.data(), data_.size()*sizeof(T));
-    }
-    ~secure_vector() { clean(); munlock(data_.data(), data_.size()*sizeof(T)); }
+    explicit secure_vector(unsigned size): data_(size) {}
+    explicit secure_vector(unsigned size, T _item): data_(size, _item) {}
+    explicit secure_vector(const std::vector<T>& c): data_(c) {}
+    secure_vector(std::vector<T>&& c): data_(std::move(c)) {}
+    ~secure_vector() { clean(); }

    static secure_vector<T> getRandom(size_t size) {
        secure_vector<T> ret(size/sizeof(T));
@@ -529,30 +517,24 @@ public:
        if (&c == this)
            return *this;
        clean();
-        munlock(data_.data(), data_.size()*sizeof(T));
        data_ = c.data_;
-        mlock(data_.data(), data_.size()*sizeof(T));
        return *this;
    }
    secure_vector<T>& operator=(secure_vector<T>&& c) {
        if (&c == this)
            return *this;
        clean();
-        munlock(data_.data(), data_.size()*sizeof(T));
        data_ = std::move(c.data_);
        return *this;
    }
    secure_vector<T>& operator=(std::vector<T>&& c) {
        clean();
-        munlock(data_.data(), data_.size()*sizeof(T));
        data_ = std::move(c);
-        mlock(data_.data(), data_.size()*sizeof(T));
        return *this;
    }
    std::vector<T>& writable() { clean(); return data_; }
    const std::vector<T>& makeInsecure() const { return data_; }
    const uint8_t* data() const { return data_.data(); }
-    uint8_t* data() { return data_.data(); }

    void clean() {
        clean(data_.begin(), data_.end());
@@ -569,17 +551,14 @@ public:
        if (s < data_.size()) {
            //shrink
            clean(data_.begin()+s, data_.end());
-            munlock(data_.data()+s, (data_.size()-s)*sizeof(T));
            data_.resize(s);
        } else {
            //grow
            auto data = std::move(data_); // move protected data
            clear();
            data_.resize(s);
-            mlock(data_.data(), data_.size()*sizeof(T));
            std::copy(data.begin(), data.end(), data_.begin());
            clean(data.begin(), data.end());
-            munlock(data.data(), data.size()*sizeof(T));
        }
    }

@@ -596,154 +575,8 @@ private:
    std::vector<T> data_;
 };

-
-template <class T, size_t N>
-class OPENDHT_PUBLIC secure_array
-{
-public:
-    secure_array() {
-        mlock(data_.data(), data_.size()*sizeof(T));
-    }
-    explicit secure_array(T _item): data_( _item) {
-        mlock(data_.data(), data_.size()*sizeof(T));
-    }
-    explicit secure_array(const std::array<T,N>& c): data_(c) {
-        mlock(data_.data(), data_.size()*sizeof(T));
-    }
-    secure_array(secure_array<T,N> const&) = default;
-    secure_array(secure_array<T,N> &&) = delete;
-    ~secure_array() { clean(); munlock(data_.data(), data_.size()*sizeof(T)); }
-
-    static secure_array<T,N> getRandom() {
-        secure_array<T,N> ret;
-        crypto::random_device rdev;
-        std::uniform_int_distribution<uint8_t> rand_byte;
-        std::generate_n((uint8_t*)ret.data_.data(), ret.size()*sizeof(T), std::bind(rand_byte, std::ref(rdev)));
-        return ret;
-    }
-    secure_array<T,N>& operator=(const secure_array<T,N>& c) {
-        if (&c == this)
-            return *this;
-        clean();
-        data_ = c.data_;
-        return *this;
-    }
-    secure_array<T,N>& operator=(secure_array<T,N>&& c) = delete;
-    std::array<T,N>& writable() { clean(); return data_; }
-    const std::array<T,N>& makeInsecure() const { return data_; }
-    const uint8_t* data() const { return data_.data(); }
-    uint8_t* data() { return data_.data(); }
-
-    void clean() {
-        clean(data_.begin(), data_.end());
-    }
-    size_t size() const { return data_.size(); }
-
-private:
-    /**
-     * Securely wipe memory
-     */
-    static void clean(const typename std::array<T,N>::iterator& i, const typename std::array<T,N>::iterator& j) {
-        volatile uint8_t* b = reinterpret_cast<uint8_t*>(&*i);
-        volatile uint8_t* e = reinterpret_cast<uint8_t*>(&*j);
-        std::fill(b, e, 0);
-    }
-
-    std::array<T, N> data_;
-};
-
-
 using SecureBlob = secure_vector<uint8_t>;

-/*class EcPublicKey : public std::array<uint8_t, crypto_box_PUBLICKEYBYTES> {
-public:
-    Blob encrypt(const Blob& message) const {
-        Blob ret(crypto_box_SEALBYTES+message.size());
-        crypto_box_seal(ret.data(), message.data(), message.size(), data());
-        return ret;
-    }
-
-    const InfoHash& getId() const {
-        return *reinterpret_cast<const InfoHash*>(this);
-    }
-
-};*/
-
-class EcSecretKey {
-public:
-    using KeyData = secure_array<uint8_t, crypto_box_SECRETKEYBYTES>;
-    using EcPublicKey = InfoHash;
-
-    static EcSecretKey generate() {
-        EcSecretKey ret;
-        crypto_box_keypair(ret.pk.data(), ret.key.data());
-        return ret;
-    }
-
-    const EcPublicKey& getPublicKey() const {
-        return pk;
-    }
-
-    Blob encrypt(const uint8_t* data, size_t size, const EcPublicKey& pub) const {
-        Blob ret(crypto_box_NONCEBYTES+crypto_box_MACBYTES+size);
-        randombytes_buf(ret.data(), crypto_box_NONCEBYTES);
-        if (crypto_box_easy(ret.data()+crypto_box_NONCEBYTES, data, size, ret.data(),
-                            pub.data(), key.data()) != 0) {
-            throw CryptoException("Can't encrypt data");
-        }
-        return ret;
-    }
-
-    Blob decrypt(const uint8_t* cypher, size_t cypher_size) const {
-        if (cypher_size <= crypto_box_SEALBYTES)
-            throw DecryptError("Unexpected cipher length");
-        Blob ret(cypher_size - crypto_box_SEALBYTES);
-        if (crypto_box_seal_open(ret.data(), cypher, cypher_size, pk.data(), key.data()) != 0) {
-            throw DecryptError("Can't decrypt data");
-        }
-        return ret;
-    }
-
-    Blob decrypt(const uint8_t* cypher, size_t cypher_size, const EcPublicKey& pub) const {
-        if (cypher_size <= crypto_box_NONCEBYTES+crypto_box_MACBYTES)
-            throw DecryptError("Unexpected cipher length");
-        Blob ret(cypher_size - crypto_box_NONCEBYTES - crypto_box_MACBYTES);
-        if (crypto_box_open_easy(ret.data(),
-                cypher+crypto_box_NONCEBYTES,
-                cypher_size-crypto_box_NONCEBYTES,
-                cypher,
-                pub.data(), key.data()) != 0) {
-            throw DecryptError("Can't decrypt data");
-        }
-        return ret;
-    }
-
-    SecureBlob computeSecret(const EcPublicKey& pub, bool sending) const {
-        // shared key = h(q ‖ client_publickey ‖ server_publickey)
-        crypto_generichash_state h;
-        crypto_generichash_init(&h, nullptr, 0U, crypto_generichash_BYTES);
-        {
-            std::array<uint8_t, crypto_scalarmult_BYTES> q;
-            if (auto err = crypto_scalarmult(q.data(), key.data(), pub.data()))
-                throw CryptoException("Can't compute secret: " + std::to_string(err));
-            crypto_generichash_update(&h, q.data(), q.size());
-        }
-        {
-            const auto& sender_pk = sending ? pk : pub;
-            const auto& receiver_pk = sending ? pub : pk;
-            crypto_generichash_update(&h, sender_pk.data(), sender_pk.size());
-            crypto_generichash_update(&h, receiver_pk.data(), receiver_pk.size());
-        }
-        SecureBlob ret(crypto_generichash_BYTES);
-        crypto_generichash_final(&h, ret.data(), ret.size());
-        return ret;
-    }
-
-private:
-    KeyData key;
-    EcPublicKey pk;
-};
-
 /**
 * Generate an RSA key pair (4096 bits) and a certificate.
 * @param name the name used in the generated certificate
@@ -775,18 +608,18 @@ OPENDHT_PUBLIC Blob hash(const Blob& data, size_t hash_length = 512/8);
 * that can be transmitted in clear, and will be generated if
 * not provided (32 bytes).
 */
-OPENDHT_PUBLIC SecureBlob stretchKey(const std::string& password, Blob& salt, size_t key_length = 512/8);
+OPENDHT_PUBLIC Blob stretchKey(const std::string& password, Blob& salt, size_t key_length = 512/8);

 /**
 * AES-GCM encryption. Key must be 128, 192 or 256 bits long (16, 24 or 32 bytes).
 */
-OPENDHT_PUBLIC Blob aesEncrypt(const Blob& data, const SecureBlob& key);
+OPENDHT_PUBLIC Blob aesEncrypt(const Blob& data, const Blob& key);
 OPENDHT_PUBLIC Blob aesEncrypt(const Blob& data, const std::string& password);

 /**
 * AES-GCM decryption.
 */
-OPENDHT_PUBLIC Blob aesDecrypt(const Blob& data, const SecureBlob& key);
+OPENDHT_PUBLIC Blob aesDecrypt(const Blob& data, const Blob& key);
 OPENDHT_PUBLIC Blob aesDecrypt(const Blob& data, const std::string& password);

 }
@@ -54,6 +54,9 @@ struct LocalListener;
 /**
 * Main Dht class.
 * Provides a Distributed Hash Table node.
+ *
+ * Must be given open UDP sockets and ::periodic must be
+ * called regularly.
 */
 class OPENDHT_PUBLIC Dht {
 public:
@@ -70,13 +73,13 @@ public:
     * Initialise the Dht with two open sockets (for IPv4 and IP6)
     * and an ID for the node.
     */
-    Dht(uv_loop_t* loop, Config config);
+    Dht(int s, int s6, Config config);
    virtual ~Dht();

    /**
     * Get the ID of the node.
     */
-    inline const InfoHash& getNodeId() const { return network_engine.getNodeId(); }
+    inline const InfoHash& getNodeId() const { return myid; }

    /**
     * Get the current status of the node for the given family.
@@ -86,9 +89,6 @@ public:
    NodeStatus getStatus() const {
        return std::max(getStatus(AF_INET), getStatus(AF_INET6));
    }
-    void setOnStatusChanged(StatusCallback&& cb) {
-        statusCb = std::move(cb);
-    }

    /**
     * Performs final operations before quitting.
@@ -136,12 +136,12 @@ public:
        insertNode(n.id, SockAddr(n.ss, n.sslen));
    }

-    void bootstrap(const std::string& host, const std::string& service);
-    void clearBootstrap() {
-        bootstrap_nodes.clear();
-    }
+    void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& cb={});

-    void pingNode(const SockAddr&, DoneCallback&& cb={});
+    time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr&);
+    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) {
+        return periodic(buf, buflen, SockAddr(from, fromlen));
+    }

    /**
     * Get a value by searching on all available protocols (IPv4, IPv6),
@@ -307,10 +307,6 @@ public:

    std::vector<SockAddr> getPublicAddress(sa_family_t family = 0);

-    SockAddr getBoundAddr() {
-        return network_engine.getBoundAddr();
-    }
-
 protected:
    Logger DHT_LOG;
    bool logFilerEnable_ {};
@@ -382,8 +378,6 @@ private:
    const bool is_bootstrap {false};
    const bool maintain_storage {false};

-    std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
-
    // the stuff
    RoutingTable buckets4 {};
    RoutingTable buckets6 {};
@@ -406,7 +400,8 @@ private:

    // timing
    Scheduler scheduler;
-    Sp<Job> nextNodesConfirmation {};
+    Sp<Scheduler::Job> nextNodesConfirmation {};
+    Sp<Scheduler::Job> nextStorageMaintenance {};

    net::NetworkEngine network_engine;
    unsigned pending_pings4 {0};
@@ -414,13 +409,9 @@ private:

    using ReportedAddr = std::pair<unsigned, SockAddr>;
    std::vector<ReportedAddr> reported_addr;
-    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
-    
-    NodeStatus status4 {NodeStatus::Disconnected},
-               status6 {NodeStatus::Disconnected};
-    StatusCallback statusCb {nullptr};

-    void updateStatus();
+    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
+
    void rotateSecrets();

    Blob makeToken(const SockAddr&, bool old) const;
@@ -26,7 +26,6 @@
 #include "sockaddr.h"
 #include "log_enable.h"
 #include "def.h"
-#include "uv_utils.h"

 #include <thread>
 #include <mutex>
@@ -52,6 +51,8 @@ struct SecureDhtConfig;
 class OPENDHT_PUBLIC DhtRunner {

 public:
+    typedef std::function<void(NodeStatus, NodeStatus)> StatusCallback;
+
    DhtRunner();
    virtual ~DhtRunner();

@@ -201,11 +202,8 @@ public:
     * 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 ping(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes, DoneCallback&& cb={});
-    void ping(const SockAddr& addr, DoneCallback&& cb={});
-    void ping(const SockAddr& addr, DoneCallbackSimple&& cb={}) {
-        ping(addr, bindDoneCb(cb));
-    }
+    void bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes, DoneCallbackSimple&& cb={});
+    void bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb={});

    /**
     * Insert known nodes to the routing table, without necessarly ping them.
@@ -243,14 +241,16 @@ public:
     * Returns the currently bound address.
     * @param f: address family of the bound address to retreive.
     */
-    SockAddr getBound() const;
+    const SockAddr& getBound(sa_family_t f = AF_INET) const {
+        return (f == AF_INET) ? bound4 : bound6;
+    }

    /**
     * Returns the currently bound port, in host byte order.
     * @param f: address family of the bound port to retreive.
     */
-    in_port_t getBoundPort() const {
-        return getBound().getPort();
+    in_port_t getBoundPort(sa_family_t f = AF_INET) const {
+        return getBound(f).getPort();
    }

    std::pair<size_t, size_t> getStoreSize() const;
@@ -306,24 +306,20 @@ public:
     * @param cb: Optional callback to receive general state information.
     */
    void run(in_port_t port, const crypto::Identity identity, bool threaded = false, NetId network = 0) {
-        run({
+        run(port, {
            /*.dht_config = */{
                /*.node_config = */{
                    /*.node_id = */{},
                    /*.network = */network,
                    /*.is_bootstrap = */false,
-                    /*.maintain_storage = */false,
-                    /* .network_config = */{
-                        /*.addr = */ "::0",
-                        /*.port = */ port
-                    }
+                    /*.maintain_storage*/false
                },
                /*.id = */identity
            },
            /*.threaded = */threaded
        });
    }
-    void run(Config config);
+    void run(in_port_t port, Config config);

    /**
     * @param local4: Local IPv4 address and port to bind. Can be null.
@@ -333,22 +329,25 @@ public:
     * @param threaded: If false, loop() must be called periodically. Otherwise a thread is launched.
     * @param cb: Optional callback to receive general state information.
     */
-    //void run(const sockaddr_in* local4, const sockaddr_in6* local6, Config config);
+    void run(const sockaddr_in* local4, const sockaddr_in6* local6, Config config);

    /**
     * Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port).
     */
-    //void run(const char* ip4, const char* ip6, const char* service, Config config);
+    void run(const char* ip4, const char* ip6, const char* service, Config config);

-    void setOnStatusChanged(StatusCallback&& cb);
+    void setOnStatusChanged(StatusCallback&& cb) {
+        statusCb = std::move(cb);
+    }

    /**
     * In non-threaded mode, the user should call this method
     * regularly and everytime a new packet is received.
     * @return the next op
     */
-    void loop() {
-        uv_loop_.runOnce();
+    time_point loop() {
+        std::lock_guard<std::mutex> lck(dht_mtx);
+        return loop_();
    }

    /**
@@ -374,41 +373,33 @@ private:
     */
    void tryBootstrapContinuously();

-    void doRun(SecureDhtConfig config);
-    void loop_();
+    void doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDhtConfig config);
+    time_point loop_();

    static std::vector<std::pair<sockaddr_storage, socklen_t>> getAddrInfo(const std::string& host, const std::string& service);

-    void loopCallback() {
-        std::lock_guard<std::mutex> lck(dht_mtx);
-        if (running)
-            loop_();
-        else {
-            uv_close((uv_handle_t*)&uv_async_, nullptr);
-            uv_loop_.stop();
-        }
-    }
-
-    static void loop_callback(uv_async_t* handle) {
-        if (handle->data)
-            static_cast<DhtRunner*>(handle->data)->loopCallback();
+    NodeStatus getStatus() const {
+        return std::max(status4, status6);
    }

    std::unique_ptr<SecureDht> dht_;
-    EventLoop uv_loop_;
-    uv_async_t uv_async_;
    mutable std::mutex dht_mtx {};
    std::thread dht_thread {};
+    std::condition_variable cv {};
+
+    std::thread rcv_thread {};
+    std::mutex sock_mtx {};
+    std::vector<std::pair<Blob, SockAddr>> rcv {};

    /** true if currently actively boostraping */
    std::atomic_bool bootstraping {false};
    /* bootstrap nodes given as (host, service) pairs */
-    /*std::vector<std::pair<std::string,std::string>> bootstrap_nodes_all {};
+    std::vector<std::pair<std::string,std::string>> bootstrap_nodes_all {};
    std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
-    std::thread bootstrap_thread {};*/
+    std::thread bootstrap_thread {};
    /** protects bootstrap_nodes, bootstrap_thread */
-    /*std::mutex bootstrap_mtx {};
-    std::condition_variable bootstrap_cv {};*/
+    std::mutex bootstrap_mtx {};
+    std::condition_variable bootstrap_cv {};

    std::queue<std::function<void(SecureDht&)>> pending_ops_prio {};
    std::queue<std::function<void(SecureDht&)>> pending_ops {};
@@ -416,12 +407,12 @@ private:

    std::atomic<bool> running {false};

-    /*NodeStatus status4 {NodeStatus::Disconnected},
+    NodeStatus status4 {NodeStatus::Disconnected},
               status6 {NodeStatus::Disconnected};
    StatusCallback statusCb {nullptr};

    SockAddr bound4 {};
-    SockAddr bound6 {};*/
+    SockAddr bound6 {};
 };

 }
@@ -22,8 +22,6 @@

 #include <msgpack.hpp>

-#include <sodium.h>
-
 #ifndef _WIN32
 #include <netinet/in.h>
 #include <netdb.h>
@@ -48,7 +46,7 @@ typedef uint16_t in_port_t;


 // bytes
-#define HASH_LEN crypto_box_PUBLICKEYBYTES
+#define HASH_LEN 20u

 namespace dht {

@@ -149,12 +147,6 @@ public:
            std::copy_n(h, HASH_LEN, begin());
    }

-    std::vector<uint8_t> encrypt(const uint8_t* msg, size_t msg_size) const {
-        std::vector<uint8_t> ret(crypto_box_SEALBYTES+msg_size);
-        crypto_box_seal(ret.data(), msg, msg_size, data());
-        return ret;
-    }
-
    /**
     * Constructor from an hexadecimal string (without "0x").
     * hex must be at least 2.HASH_LEN characters long.
@@ -1,67 +0,0 @@
-/*
- *  Copyright (C) 2014-2017 Savoir-faire Linux Inc.
- *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
- *              Simon Désaulniers <simon.desaulniers@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 <array>
-#include <cstdint>
-
-namespace dht {
-namespace net {
-
-struct TransPrefix : public std::array<uint8_t, 2> {
-    TransPrefix(const std::string& str) : std::array<uint8_t, 2>({{(uint8_t)str[0], (uint8_t)str[1]}}) {}
-    static const TransPrefix PING;
-    static const TransPrefix FIND_NODE;
-    static const TransPrefix GET_VALUES;
-    static const TransPrefix ANNOUNCE_VALUES;
-    static const TransPrefix REFRESH;
-    static const TransPrefix LISTEN;
-};
-
-/* Transaction-ids are 4-bytes long, with the first two bytes identifying
- * the kind of request, and the remaining two a sequence number in
- * host order.
- */
-struct TransId final : public std::array<uint8_t, 4> {
-    static const constexpr uint16_t INVALID {0};
-
-    TransId() { std::fill(begin(), end(), 0); }
-    TransId(const std::array<char, 4>& o) { std::copy(o.begin(), o.end(), begin()); }
-    TransId(const TransPrefix prefix, uint16_t seqno = 0) {
-        std::copy_n(prefix.begin(), prefix.size(), begin());
-        *reinterpret_cast<uint16_t*>(data()+prefix.size()) = seqno;
-    }
-    TransId(uint32_t id) {
-        *reinterpret_cast<uint32_t*>(data()) = htonl(id);
-    }
-
-    uint32_t toInt() const {
-        return ntohl(*reinterpret_cast<const uint32_t*>(&(*this)[0]));
-    }
-
-    bool matches(const TransPrefix prefix) const {
-        return std::equal(begin(), begin()+2, prefix.begin());
-    }
-};
-
-TransId unpackTid(msgpack::object& o);
-
-} /* namespace net */
-} /* dht */
@@ -27,9 +27,6 @@
 #include "utils.h"
 #include "rng.h"
 #include "rate_limiter.h"
-#include "net.h"
-#include "callbacks.h"
-#include "uv_utils.h"

 #include <vector>
 #include <string>
@@ -43,6 +40,7 @@ namespace net {

 struct Request;
 struct Socket;
+struct TransId;

 #ifndef MSG_CONFIRM
 #define MSG_CONFIRM 0
@@ -171,7 +169,7 @@ private:
    std::function<RequestAnswer(Sp<Node>,
            const InfoHash&,
            const Blob&,
-            uint32_t,
+            Tid,
            const Query&)> onListen {};
    /**
     * Called on announce request.
@@ -204,8 +202,8 @@ public:
    using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
    using RequestExpiredCb = std::function<void(const Request&, bool)>;

-    NetworkEngine(uv_loop_t*, Logger& log, Scheduler& scheduler);
-    NetworkEngine(uv_loop_t*, const NetworkConfig& config, NetId net, Logger& log, Scheduler& scheduler,
+    NetworkEngine(Logger& log, Scheduler& scheduler);
+    NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& log, Scheduler& scheduler,
            decltype(NetworkEngine::onError) onError,
            decltype(NetworkEngine::onNewNode) onNewNode,
            decltype(NetworkEngine::onReportedAddr) onReportedAddr,
@@ -218,16 +216,7 @@ public:

    virtual ~NetworkEngine();

-    const InfoHash& getNodeId() const {
-        return id_key.getPublicKey();
-    }
-
    void clear();
-    void close(OnClose cb) {
-        sock->close(cb);
-        tcp_sock->close();
-        cache.closeAll();
-    }

    /**
     * Sends values (with closest nodes) to a listenner.
@@ -242,12 +231,12 @@ public:
     * @param nodes6      The ipv6 closest nodes.
     * @param values      The values to send.
     */
-    void tellListener(Sp<Node> n, uint32_t socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
+    void tellListener(Sp<Node> n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
            std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
            std::vector<Sp<Value>>&& values, const Query& q);

    bool isRunning(sa_family_t af) const;
-    inline want_t want () const { /*return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1;*/return WANT4 | WANT6; }
+    inline want_t want () const { return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1; }

    void connectivityChanged(sa_family_t);

@@ -265,7 +254,7 @@ public:
     * @return the request with information concerning its success.
     */
    Sp<Request>
-        sendPing(const Sp<Node>& n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
+        sendPing(Sp<Node> n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
    /**
     * Send a "ping" request to a given node.
     *
@@ -277,8 +266,8 @@ public:
     * @return the request with information concerning its success.
     */
    Sp<Request>
-        sendPing(const SockAddr& sa, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-            return sendPing(std::make_shared<Node>(zeroes, sa),
+        sendPing(const sockaddr* sa, socklen_t salen, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
+            return sendPing(std::make_shared<Node>(zeroes, sa, salen),
                    std::forward<RequestCb>(on_done),
                    std::forward<RequestExpiredCb>(on_expired));
        }
@@ -294,7 +283,7 @@ public:
     *
     * @return the request with information concerning its success.
     */
-    Sp<Request> sendFindNode(const Sp<Node>& n,
+    Sp<Request> sendFindNode(Sp<Node> n,
                             const InfoHash& hash,
                             want_t want,
                             RequestCb&& on_done,
@@ -313,7 +302,7 @@ public:
     *
     * @return the request with information concerning its success.
     */
-    Sp<Request> sendGetValues(const Sp<Node>& n,
+    Sp<Request> sendGetValues(Sp<Node> n,
                              const InfoHash& hash,
                              const Query& query,
                              want_t want,
@@ -342,7 +331,7 @@ public:
     *
     * @return the request with information concerning its success.
     */
-    Sp<Request> sendListen(const Sp<Node>& n,
+    Sp<Request> sendListen(Sp<Node> n,
                           const InfoHash& hash,
                           const Query& query,
                           const Blob& token,
@@ -363,7 +352,7 @@ public:
     *
     * @return the request with information concerning its success.
     */
-    Sp<Request> sendAnnounceValue(const Sp<Node>& n,
+    Sp<Request> sendAnnounceValue(Sp<Node> n,
                                  const InfoHash& hash,
                                  const Sp<Value>& v,
                                  time_point created,
@@ -383,15 +372,26 @@ public:
     *
     * @return the request with information concerning its success.
     */
-    Sp<Request> sendRefreshValue(const Sp<Node>& n,
+    Sp<Request> sendRefreshValue(Sp<Node> n,
                                 const InfoHash& hash,
                                 const Value::Id& vid,
                                 const Blob& token,
                                 RequestCb&& on_done,
                                 RequestExpiredCb&& on_expired);

+    /**
+     * Parses a message and calls appropriate callbacks.
+     *
+     * @param buf  The buffer containing the binary message.
+     * @param buflen  The length of the buffer.
+     * @param from  The address info of the sender.
+     * @param fromlen  The length of the corresponding sockaddr structure.
+     * @param now  The time to adjust the clock in the network engine.
+     */
+    void processMessage(const uint8_t *buf, size_t buflen, const SockAddr& addr);
+
    Sp<Node> insertNode(const InfoHash& myid, const SockAddr& addr) {
-        auto n = cache.getNode(myid, addr, {}, scheduler.time(), 0);
+        auto n = cache.getNode(myid, addr, scheduler.time(), 0);
        onNewNode(n, 0);
        return n;
    }
@@ -409,10 +409,6 @@ public:
        return cache.getCachedNodes(id, sa_f, count);
    }

-    SockAddr getBoundAddr() {
-        return sock->getBoundAddr();
-    }
-
 private:

    struct PartialMessage;
@@ -441,20 +437,7 @@ private:

    static const std::string my_v;

-    /**
-     * Parses a message and calls appropriate callbacks.
-     *
-     * @param buf  The buffer containing the binary message.
-     * @param buflen  The length of the buffer.
-     * @param from  The address info of the sender.
-     * @param fromlen  The length of the corresponding sockaddr structure.
-     * @param now  The time to adjust the clock in the network engine.
-     */
-    void processMessage(const uint8_t* data, size_t size, const SockAddr& addr);
-    Sp<Node> processEncrypted(const msgpack::object& o, const SockAddr& from);
-    Sp<Node> process(const msgpack::object& o, const SockAddr& addr);
-
-    Sp<Node> process(std::unique_ptr<ParsedMessage>&&, const SockAddr& from, const Sp<TcpSocket>& s = {});
+    void process(std::unique_ptr<ParsedMessage>&&, const SockAddr& from);

    bool rateLimit(const SockAddr& addr);

@@ -479,25 +462,21 @@ private:
    };


-    int send(msgpack::sbuffer& msg, const Node& node, const Sp<TcpSocket>& sock = {});
-    int sendUDP(msgpack::sbuffer& msg, const SockAddr& addr);
+    // basic wrapper for socket sendto function
+    int send(const char *buf, size_t len, int flags, const SockAddr& addr);

-    int send(const Blob& msg, int flags, const Node& node, UdpSocket::OnSent&& cb);
-
-    void startTcp(const Sp<TcpSocket>& sock, const Sp<Node>& node = {});
-
-    void sendValueParts(TransId tid, const std::vector<Blob>& svals, const SockAddr& addr);
-    std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&, bool stream);
-    void maintainRxBuffer(const TransId& tid);
+    void sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr);
+    std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
+    void maintainRxBuffer(Tid tid);

    /*************
     *  Answers  *
     *************/
    /* answer to a ping  request */
-    void sendPong(const Sp<Node>& node, TransId tid);
+    void sendPong(const SockAddr& addr, Tid tid);
    /* answer to findnodes/getvalues request */
-    void sendNodesValues(const Sp<Node>& node,
-            TransId tid,
+    void sendNodesValues(const SockAddr& addr,
+            Tid tid,
            const Blob& nodes,
            const Blob& nodes6,
            const std::vector<Sp<Value>>& st,
@@ -511,32 +490,27 @@ private:
            std::vector<Sp<Node>>& nodes,
            std::vector<Sp<Node>>& nodes6);
    /* answer to a listen request */
-    void sendListenConfirmation(const Sp<Node>& node, TransId tid);
+    void sendListenConfirmation(const SockAddr& addr, Tid tid);
    /* answer to put request */
-    void sendValueAnnounced(const Sp<Node>& node, TransId, Value::Id);
+    void sendValueAnnounced(const SockAddr& addr, Tid, Value::Id);
    /* answer in case of error */
-    void sendError(const Sp<Node>& node,
-            TransId tid,
+    void sendError(const SockAddr& addr,
+            Tid tid,
            uint16_t code,
            const std::string& message,
-            bool include_id=false,
-            const Sp<TcpSocket>& sock= {});
+            bool include_id=false);

    void deserializeNodes(ParsedMessage& msg, const SockAddr& from);

    /* DHT info */
-    //const InfoHash& myid;
+    const InfoHash& myid;
    const NetId network {0};
+    const int dht_socket {-1};
+    const int dht_socket6 {-1};
    const Logger& DHT_LOG;

-    Sp<UdpSocket> sock;
-    Sp<TcpSocket> tcp_sock;
-    std::set<Sp<TcpSocket>> pending_connect;
    NodeCache cache {};

-    crypto::EcSecretKey id_key;
-    //crypto::EcSecretKey id_key;
-
    // global limiting should be triggered by at least 8 different IPs
    using IpLimiter = RateLimiter<MAX_REQUESTS_PER_SEC/8>;
    using IpLimiterMap = std::map<SockAddr, IpLimiter, SockAddr::ipCmp>;
@@ -545,8 +519,8 @@ private:
    size_t limiter_maintenance {0};

    // requests handling
-    std::map<TransId, Sp<Request>> requests {};
-    std::map<TransId, PartialMessage> partial_messages;
+    std::map<Tid, Sp<Request>> requests {};
+    std::map<Tid, PartialMessage> partial_messages;

    MessageStats in_stats {}, out_stats {};
    std::set<SockAddr> blacklist {};
@@ -22,9 +22,6 @@
 #include "infohash.h" // includes socket structures
 #include "utils.h"
 #include "sockaddr.h"
-#include "net.h"
-#include "uv_utils.h"
-#include "crypto.h"

 #include <list>
 #include <map>
@@ -38,8 +35,8 @@ struct Socket;
 struct RequestAnswer;
 } /* namespace net */

+using Tid = uint32_t;
 using SocketCb = std::function<void(const Sp<Node>&, net::RequestAnswer&&)>;
-using SocketId = uint32_t;
 struct Socket {
    Socket() {}
    Socket(SocketCb&& on_receive) :
@@ -50,18 +47,9 @@ struct Socket {
 struct Node {
    const InfoHash id;

-    Sp<TcpSocket> sock;
-
-    InfoHash last_known_pk;
-    //std::list<crypto::EcSecretKey>
-
-    Node(const InfoHash& id, const SockAddr& addr, const Sp<TcpSocket>& s = {}, bool client=false);
+    Node(const InfoHash& id, const SockAddr& addr, bool client=false);
    Node(const InfoHash& id, const sockaddr* sa, socklen_t salen)
        : Node(id, SockAddr(sa, salen)) {}
-    ~Node() {
-        if (sock)
-            sock->close();
-    }

    InfoHash getId() const {
        return id;
@@ -70,7 +58,6 @@ struct Node {
    std::string getAddrStr() const {
        return addr.toString();
    }
-
    bool isClient() const { return is_client; }
    bool isIncoming() { return time > reply_time; }

@@ -78,14 +65,6 @@ struct Node {
    const time_point& getReplyTime() const { return reply_time; }
    void setTime(const time_point& t) { time = t; }

-    SockAddr getAddress() const {
-        return canStream() ? sock->getPeerAddr() : addr;
-    }
-
-    void startTCP() {
-
-    }
-
    /**
     * Makes notice about an additionnal authentication error with this node. Up
     * to MAX_AUTH_ERRORS errors are accepted in order to let the node recover.
@@ -118,15 +97,11 @@ struct Node {
    }
    sa_family_t getFamily() const { return addr.getFamily(); }

-    void update(const SockAddr&, const Sp<TcpSocket>&);
-
-    bool canStream() const {
-        return sock and not sock->isClosed() and sock->canWrite();
-    }
+    void update(const SockAddr&);

    void requested(const Sp<net::Request>& req);
    void received(time_point now, const Sp<net::Request>& req);
-    Sp<net::Request> getRequest(const net::TransId& tid);
+    Sp<net::Request> getRequest(Tid tid);
    void cancelRequest(const Sp<net::Request>& req);

    void setExpired();
@@ -140,16 +115,16 @@ struct Node {
     *
     * @return the socket.
     */
-    SocketId openSocket(SocketCb&& cb);
+    Tid openSocket(SocketCb&& cb);

-    Socket* getSocket(SocketId id);
+    Socket* getSocket(Tid id);

    /**
     * Closes a socket so that no further data will be red on that socket.
     *
     * @param socket  The socket to close.
     */
-    void closeSocket(SocketId id);
+    void closeSocket(Tid id);

    /**
     * Resets the state of the node so it's not expired anymore.
@@ -161,9 +136,9 @@ struct Node {
     *
     * @return the new id.
     */
-    uint16_t getNewTid() {
+    Tid getNewTid() {
        ++transaction_id;
-        return transaction_id == net::TransId::INVALID ? ++transaction_id : transaction_id;
+        return transaction_id ? ++transaction_id : transaction_id;
    }

    std::string toString() const;
@@ -188,11 +163,11 @@ private:
    time_point reply_time {time_point::min()};      /* time of last correct reply received */
    unsigned auth_errors {0};
    bool expired_ {false};
-    SocketId transaction_id;
+    Tid transaction_id;
    using TransactionDist = std::uniform_int_distribution<decltype(transaction_id)>;

-    std::map<net::TransId, Sp<net::Request>> requests_ {};
-    std::map<SocketId, Socket> sockets_;
+    std::map<Tid, Sp<net::Request>> requests_ {};
+    std::map<Tid, Socket> sockets_;
 };

 }
@@ -27,7 +27,7 @@ namespace dht {

 struct NodeCache {
    Sp<Node> getNode(const InfoHash& id, sa_family_t family);
-    Sp<Node> getNode(const InfoHash& id, const SockAddr&, const Sp<TcpSocket>&, time_point now, bool confirmed, bool client=false);
+    Sp<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed, bool client=false);
    std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) const;

    /**
@@ -36,15 +36,13 @@ struct NodeCache {
     * To use in case of connectivity change etc.
     */
    void clearBadNodes(sa_family_t family = 0);
-    void closeAll(sa_family_t family = 0);

 private:
    class NodeMap : public std::map<InfoHash, std::weak_ptr<Node>> {
    public:
        Sp<Node> getNode(const InfoHash& id);
-        Sp<Node> getNode(const InfoHash& id, const SockAddr&, const Sp<TcpSocket>&, time_point now, bool confirmed, bool client);
+        Sp<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed, bool client);
        void clearBadNodes();
-        void closeAll();
    };

    const NodeMap& cache(sa_family_t af) const { return af == AF_INET ? cache_4 : cache_6; }
@@ -23,12 +23,12 @@

 namespace dht {

-template<size_t Quota, unsigned long Perdiod=1>
+template<size_t Quota, unsigned long Period=1>
 class RateLimiter {
 public:
    /** Clear outdated records and return current quota usage */
    size_t maintain(const time_point& now) {
-        auto limit = now - std::chrono::seconds(Perdiod);
+        auto limit = now - std::chrono::seconds(Period);
        while (not records.empty() and records.front() < limit)
            records.pop();
        return records.size();
@@ -42,14 +42,6 @@ struct Bucket {
    Sp<Node> randomNode();

    void sendCachedPing(net::NetworkEngine& ne);
-
-    /** Return false if there is a single non-expired node in this bucket. */
-    bool expired() const {
-        for (const auto& n : nodes)
-            if (not n->isExpired())
-                return false;
-        return true;
-    }
 };

 class RoutingTable : public std::list<Bucket> {
@@ -22,9 +22,6 @@

 #include "utils.h"
 #include "log_enable.h"
-#include "uv_utils.h"
-
-#include <uv.h>

 #include <functional>
 #include <map>
@@ -39,7 +36,12 @@ namespace dht {
 */
 class Scheduler {
 public:
-    Scheduler(uv_loop_t* loop, const Logger& l) : loop_(loop), DHT_LOG(l) {}
+    Scheduler(const Logger& l) : DHT_LOG(l) {}
+
+    struct Job {
+        Job(std::function<void()>&& f) : do_(std::move(f)) {}
+        std::function<void()> do_;
+    };

    /**
     * Adds another job to the queue.
@@ -49,32 +51,10 @@ public:
     *
     * @return pointer to the newly scheduled job.
     */
-    Sp<Job> add(const time_point& t, std::function<void()>&& job_func) {
-        if (not loop_)
-            return {};
-        auto job = Job::make(loop_, std::move(job_func));
-        if (t != time_point::max()) {
-            garbageCollection();
-            pending_jobs.emplace(job);
-            job->run(t);
-        } else
-            job->cancel();
-        return job;
-    }
-    Sp<Job> add(const duration& t, std::function<void()>&& job_func) {
-        if (not loop_)
-            return {};
-        auto job = Job::make(loop_, std::move(job_func));
-        garbageCollection();
-        pending_jobs.emplace(job);
-        job->run(t);
-        return job;
-    }
-    Sp<Job> run(std::function<void()>&& job_func) {
-        auto job = Job::make(loop_, std::move(job_func));
-        garbageCollection();
-        pending_jobs.emplace(job);
-        job->run();
+    Sp<Scheduler::Job> add(time_point t, std::function<void()>&& job_func) {
+        auto job = std::make_shared<Job>(std::move(job_func));
+        if (t != time_point::max())
+            timers.emplace(std::move(t), job);
        return job;
    }

@@ -86,9 +66,7 @@ public:
     *
     * @return pointer to the newly scheduled job.
     */
-    void edit(Sp<Job>& job, const duration& t = {}) {
-        if (not loop_)
-            return;
+    void edit(Sp<Scheduler::Job>& job, time_point t) {
        if (not job) {
            DHT_LOG.ERR("editing an empty job");
            return;
@@ -97,73 +75,50 @@ public:
        // Force clearing old value.
        auto task = std::move(job->do_);
        job->do_ = {};
-        job->cancel();
        job = add(t, std::move(task));
    }
-    void edit(Sp<Job>& job, const time_point& t) {
-        if (not loop_)
-            return;
-        if (not job) {
-            DHT_LOG.ERR("editing an empty job");
-            return;
+
+    /**
+     * Runs the jobs to do up to now.
+     *
+     * @return The time for the next job to run.
+     */
+    time_point run() {
+        syncTime();
+        while (not timers.empty()) {
+            auto timer = timers.begin();
+            /*
+             * Running jobs scheduled before "now" prevents run+rescheduling
+             * loops before this method ends. It is garanteed by the fact that a
+             * job will at least be scheduled for "now" and not before.
+             */
+            if (timer->first > now)
+                break;
+
+            auto job = std::move(timer->second);
+            timers.erase(timer);
+
+            if (job->do_)
+                job->do_();
        }
-        auto task = std::move(job->do_);
-        job->do_ = {};
-        job->cancel();
-        job = add(t, std::move(task));
+        return getNextJobTime();
    }

-    void stop() {
-        loop_ = nullptr;
-        size_t del {0};
-        for (auto& w : pending_jobs)
-            if (auto job = w.lock()) {
-                job->cancel();
-                del++;
-            }
-        pending_jobs.clear();
-        DHT_LOG.DEBUG("stopping scheduler, %lu events canceled", del);
-    }
-
-    ~Scheduler(){
-        stop();
+    inline time_point getNextJobTime() const {
+        return timers.empty() ? time_point::max() : timers.begin()->first;
    }

    /**
     * Accessors for the common time reference used for synchronizing
     * operations.
     */
-    inline time_point time() const {
-        return time_point(std::chrono::duration_cast<duration>(std::chrono::milliseconds(uv_now(loop_))));
-    }
-    inline time_point syncTime() {
-        return time();
-    }
-    uv_loop_t* getLoop() const {
-        return loop_;
-    }
+    inline const time_point& time() const { return now; }
+    inline time_point syncTime() { return (now = clock::now()); }

 private:
-    std::set<std::weak_ptr<Job>, std::owner_less<std::weak_ptr<Job>>> pending_jobs {};
-    size_t lastSize_ {0};
-    uv_loop_t* loop_;
+    time_point now {clock::now()};
+    std::multimap<time_point, Sp<Job>> timers {}; /* the jobs ordered by time */
    const Logger& DHT_LOG;
-
-    void garbageCollection() {
-        if (pending_jobs.size() > lastSize_ * 2) {
-            unsigned del {0};
-            for (auto it = pending_jobs.begin(); it != pending_jobs.end();) {
-                if (it->expired()) {
-                    it = pending_jobs.erase(it);
-                    del++;
-                } else 
-                    ++it;
-            }
-            //DHT_LOG.DEBUG("pending_jobs cleanup: %u removed, %zu remaining", del, pending_jobs.size());
-            lastSize_ = pending_jobs.size();
-        }
-    }
-
 };

 }
@@ -44,7 +44,7 @@ public:
     * id:    the identity to use for the crypto layer and to compute
     *        our own hash on the Dht.
     */
-    SecureDht(uv_loop_t* loop, Config config);
+    SecureDht(int s, int s6, Config config);

    virtual ~SecureDht();

@@ -66,19 +66,6 @@ public:
            throw std::runtime_error("Socket address length is too large");
        set(sa, length);
    }
-    SockAddr(const sockaddr* sa) {
-        socklen_t len = 0;
-        if (sa) {
-            if (sa->sa_family == AF_INET)
-                len = sizeof(sockaddr_in);
-            else if(sa->sa_family == AF_INET6)
-                len = sizeof(sockaddr_in6);
-            else
-                throw std::runtime_error("Unknown address family");
-        }
-        set(sa, len);
-    }
-
    /**
     * Build from an existing sockaddr_storage structure.
     */
@@ -219,9 +206,6 @@ public:

    bool isUnspecified() const;

-    bool isMappedIPv4() const;
-    SockAddr getMappedIPv4() const;
-
    /**
     * A comparator to classify IP addresses, only considering the
     * first 64 bits in IPv6.
@@ -1,419 +0,0 @@
-#pragma once
-
-#include "utils.h"
-
-#include <uv.h>
-
-namespace dht {
-
-struct EventLoop {
-    using OnSignal = std::function<void(int)>;
-
-    EventLoop() {
-        uv_loop_init(&loop);
-        uv_signal_init(&loop, &sig_handler);
-        sig_handler.data = this;
-        uv_signal_start(&sig_handler, &EventLoop::signal_handler, SIGINT);
-    }
-    ~EventLoop() {
-        uv_loop_close(&loop);
-    }
-    auto run() {
-        return uv_run(&loop, UV_RUN_DEFAULT);
-    }
-    auto runOnce() {
-        return uv_run(&loop, UV_RUN_NOWAIT);
-    }
-    void stop() {
-        uv_signal_stop(&sig_handler);
-        uv_walk(&loop, &EventLoop::close_walk_cb, this);
-    }
-    uv_loop_t* get() { return &loop; }
-    void setOnSignal(OnSignal cb) {
-        onSignal_ = cb;
-    }
-private:
-    uv_loop_t loop;
-    uv_signal_t sig_handler;
-    OnSignal onSignal_;
-    void onSignal(int sig) {
-        if (onSignal_)
-            onSignal_(sig);
-    }
-    static void close_walk_cb(uv_handle_t* handle, void* /*arg*/) {
-        if (not uv_is_closing(handle))
-            uv_close(handle, [](uv_handle_t* handle) {
-                std::cout << "handle " << handle << " of type " << handle->type << " closed" << std::endl;
-            });
-    }
-    static void signal_handler(uv_signal_t* handle, int signum) {
-        if (handle->data) {
-            static_cast<EventLoop*>(handle->data)->onSignal(signum);
-        }
-    }
-};
-
-using GetAddrInfoCb = std::function<void(std::vector<SockAddr>&&)>;
-
-void getAddrInfo(uv_loop_t* loop, const char* node, const char* service, GetAddrInfoCb&& cb);
-
-struct Job : std::enable_shared_from_this<Job> {
-    Job(uv_loop_t* loop, std::function<void()>&& f = {}) : do_(std::move(f)) {
-        uv_timer_init(loop, &timer_);
-        timer_.data = nullptr;
-    }
-    void run(uint64_t t = 0) {
-        uv_timer_start(&timer_, &Job::timer_cb, t, 0);
-    }
-    inline void run(const duration& t) {
-        run(getTimeout(t));
-    }
-    inline void run(const time_point& t) {
-        run(getTimeout(t));
-    }
-    void cancel() {
-        if (timer_.data and not uv_is_closing((uv_handle_t*)&timer_)) {
-            uv_timer_stop(&timer_);
-            uv_close((uv_handle_t*)&timer_, &Job::close_cb);
-        }
-    }
-    static Sp<Job> make(uv_loop_t* loop, std::function<void()>&& f = {}) {
-        auto job = std::make_shared<Job>(loop, std::move(f));
-        job->timer_.data = new Sp<Job>(job);
-        return job;
-    }
-    static uint64_t getTimeout(const duration& d) {
-        return std::chrono::duration_cast<std::chrono::milliseconds>(d).count();
-    }
-    uint64_t getTimeout(const time_point& tp) const {
-        uint64_t t = getTimeout(tp.time_since_epoch());
-        uint64_t now = uv_now(timer_.loop);
-        return (t < now) ? 0 : t - now;
-    }
-    std::function<void()> do_;
-private:
-    uv_timer_t timer_;
-    void onTimer() {
-        if (do_)
-            do_();
-        cancel();
-    }
-    void onClosed() {
-        if (timer_.data) {
-            delete static_cast<Sp<Job>*>(timer_.data);
-            timer_.data = nullptr;
-        }
-    }
-    static void timer_cb(uv_timer_t* handle) {
-        if (handle->data) {
-            auto timer = *static_cast<Sp<Job>*>(handle->data);
-            timer->onTimer();
-        }
-    }
-    static void close_cb(uv_handle_t* handle) {
-        if (handle->data) {
-            auto timer = *static_cast<Sp<Job>*>(handle->data);
-            timer->onClosed();
-        } else {
-            std::cerr << "Job::onClosed(): no data! " << handle << std::endl;
-        }
-    }
-};
-
-
-static inline void alloc_buffer(uv_handle_t* /*handle*/, size_t suggested_size, uv_buf_t *buf) {
-    buf->base = (char*)malloc(suggested_size);
-    buf->len = suggested_size;
-}
-
-
-using OnClose = std::function<void()>;
-
-struct UdpSocket : public std::enable_shared_from_this<UdpSocket>
-{
-    using OnReceive = std::function<void(const uint8_t* data, size_t size, const SockAddr& addr)>;
-    using OnSent = std::function<void(int status)>;
-
-    UdpSocket(uv_loop_t* loop) {
-        uv_udp_init(loop, &sock);
-        sock.data = nullptr;
-    }
-    void open(const char* bind_addr, in_port_t port, OnReceive cb) {
-        if (not sock.data)
-            sock.data = new std::shared_ptr<UdpSocket>(shared_from_this());
-        receive_cb = std::move(cb);
-        struct sockaddr_in6 addr;
-        uv_ip6_addr(bind_addr, port, &addr);
-        uv_udp_bind(&sock, (const struct sockaddr*)&addr, 0);
-        uv_udp_recv_start(&sock, &UdpSocket::on_alloc_buffer, &UdpSocket::on_receive);
-    }
-
-    int send(uint8_t* data, size_t size, const SockAddr& addr, OnSent&& cb = {}) {
-        auto req = new SendReq;
-        req->msg = data;
-        req->req.data = req;
-        req->cb = std::move(cb);
-        const uv_buf_t buf {
-            /*.base =*/ (char*)data,
-            /*.len =*/ size
-        };
-        return uv_udp_send(&req->req, &sock, &buf, 1, addr.get(), &UdpSocket::on_sent);
-    }
-
-    void close(OnClose cb) {
-        close_cb = cb;
-        uv_close((uv_handle_t*) &sock, &UdpSocket::on_close);
-    }
-    bool isRunning() const {
-        return uv_is_active((uv_handle_t*) &sock);
-    }
-    SockAddr getBoundAddr() {
-        sockaddr_storage addr;
-        int size = sizeof(addr);
-        uv_udp_getsockname(&sock, (sockaddr*)&addr, &size);
-        return {(const sockaddr*)&addr, (socklen_t)size};
-    }
-private:
-    struct SendReq {
-        uv_udp_send_t req;
-        uint8_t* msg;
-        OnSent cb;
-    };
-
-    uv_udp_t sock;
-    OnReceive receive_cb;
-    OnClose close_cb;
-
-    void onAllocBuffer(size_t suggested_size, uv_buf_t* buf) {
-        buf->base = (char*)malloc(suggested_size);
-        buf->len = suggested_size;
-    }
-    void onReceive(ssize_t nread, const uv_buf_t* buf, const struct sockaddr* addr, unsigned /*flags*/) {
-        if (nread > 0 and addr and receive_cb) {
-            receive_cb((const uint8_t*)buf->base, nread, SockAddr(addr));
-        }
-        if (buf->base)
-            free(buf->base);
-    }
-    void onClosed() {
-        if (close_cb)
-            close_cb();
-        if (auto s = static_cast<std::shared_ptr<UdpSocket>*>(sock.data)) {
-            sock.data = nullptr;
-            delete s;
-        }
-    }
-    void onSent(SendReq& req, int status) {
-        if (req.cb) {
-            req.cb(status);
-        }
-        free(req.msg);
-    }
-    static std::shared_ptr<UdpSocket> get(uv_handle_t* handle) {
-        return handle->data ? *static_cast<std::shared_ptr<UdpSocket>*>(handle->data) : std::shared_ptr<UdpSocket>();
-    }
-    static void on_close(uv_handle_t* handle) {
-        if (auto node = get(handle))
-            node->onClosed();
-    }
-    static void on_sent(uv_udp_send_t* req, int status) {
-        if (auto data = static_cast<SendReq*>(req->data)) {
-            if (auto node = get((uv_handle_t*)req->handle))
-                node->onSent(*data, status);
-            delete data;
-        }
-    }
-    static void on_receive(uv_udp_t* handle, ssize_t nread, const uv_buf_t* buf, const struct sockaddr* addr, unsigned flags) {
-        if (auto node = get((uv_handle_t*)handle))
-            node->onReceive(nread, buf, addr, flags);
-    }
-    static inline void on_alloc_buffer(uv_handle_t* handle, size_t suggested_size, uv_buf_t *buf) {
-        if (auto node = get((uv_handle_t*)handle))
-            node->onAllocBuffer(suggested_size, buf);
-    }
-};
-
-struct TcpSocket : public std::enable_shared_from_this<TcpSocket>
-{
-    using OnConnect = std::function<void(std::shared_ptr<TcpSocket>&& new_socket)>;
-    using OnConnected = std::function<void(int status)>;
-    using OnRead = std::function<void(const uint8_t* data, size_t size)>;
-    using OnWrite = std::function<void(int status)>;
-
-    TcpSocket(uv_loop_t* loop) {
-        uv_tcp_init(loop, &sock);
-        sock.data = nullptr;
-    }
-    void listen(const char* bind_addr, in_port_t port, OnConnect cb) {
-        if (not sock.data)
-            sock.data = new std::shared_ptr<TcpSocket>(shared_from_this());
-        connect_cb = cb;
-        struct sockaddr_in6 addr;
-        uv_ip6_addr(bind_addr, port, &addr);
-        uv_tcp_bind(&sock, (const struct sockaddr*)&addr, 0);
-        if (int r = uv_listen((uv_stream_t*)&sock, 128, &TcpSocket::on_new_connection)) {
-            fprintf(stderr, "Listen error %s\n", uv_strerror(r));
-        }
-    }
-    void connect(const sockaddr* addr, OnConnected&& cb) {
-        if (not sock.data)
-            sock.data = new std::shared_ptr<TcpSocket>(shared_from_this());
-        auto req = new ConnectReq;
-        req->connect.data = req;
-        req->sock = shared_from_this();
-        req->callback = std::move(cb);
-        uv_tcp_connect(&req->connect, &sock, addr, &TcpSocket::on_connect);
-    }
-    void start(OnRead&& cb) {
-        std::cerr << "TCP start" << std::endl;
-        read_cb = std::move(cb);
-        uv_read_start((uv_stream_t*)&sock, &TcpSocket::on_alloc_buffer, &TcpSocket::on_read);
-    }
-    int write(uint8_t* msg, size_t len, OnWrite&& cb = {}) {
-        std::cerr << "TCP write " << len << std::endl;
-        auto req = new WriteReq;
-        req->write.data = req;
-        req->msg = msg;
-        req->sock = shared_from_this();
-        req->callback = std::move(cb);
-        uv_buf_t buf;
-        buf.base = (char*)msg;
-        buf.len = len;
-        return uv_write(&req->write, (uv_stream_t *)&sock, &buf, 1, &TcpSocket::on_write_end);
-    }
-    void close(OnClose onclose = {}) {
-        std::cerr << "TCP close" << std::endl;
-        if (not isClosed()) {
-            close_cb = onclose;
-            uv_close((uv_handle_t*) &sock, &TcpSocket::on_close);
-        }
-    }
-    bool isClosed() {
-        return not sock.data or uv_is_closing((const uv_handle_t*) &sock);
-    }
-    bool canWrite() const {
-        return uv_is_writable((const uv_stream_t*)&sock);
-    }
-    SockAddr getBoundAddr() {
-        sockaddr_storage addr;
-        int size = sizeof(addr);
-        uv_tcp_getsockname(&sock, (sockaddr*)&addr, &size);
-        return {(const sockaddr*)&addr, (socklen_t)size};
-    }
-    SockAddr getPeerAddr() {
-        sockaddr_storage addr;
-        int size = sizeof(addr);
-        uv_tcp_getpeername(&sock, (sockaddr*)&addr, &size);
-        return {(const sockaddr*)&addr, (socklen_t)size};
-    }
-private:
-    uv_tcp_t sock;
-    OnConnect connect_cb;
-    OnRead read_cb;
-    OnClose close_cb;
-    struct ConnectReq {
-        uv_connect_t connect;
-        std::shared_ptr<TcpSocket> sock;
-        OnConnected callback;
-    };
-    struct WriteReq {
-        uv_write_t write;
-        uint8_t* msg;
-        std::shared_ptr<TcpSocket> sock;
-        OnWrite callback;
-    };
-    std::shared_ptr<TcpSocket> accept() {
-        auto s = std::make_shared<TcpSocket>(sock.loop);
-        s->sock.data = new std::shared_ptr<TcpSocket>(s);
-        if (uv_accept((uv_stream_t*)&sock, (uv_stream_t*) &s->sock) == 0) {
-            return s;
-        } else {
-            s->close();
-            return {};
-        }
-    }
-    void onNewConnection(int status) {
-        if (status < 0) {
-            fprintf(stderr, "New connection error %s\n", uv_strerror(status));
-            return;
-        }
-        if (auto s = accept())
-            connect_cb(std::move(s));
-        fprintf(stderr, "on_new_connection end\n");
-    }
-    void onConnected(int status, OnConnected&& cb) {
-        if (cb)
-            cb(status);
-    }
-    void onAllocBuffer(size_t suggested_size, uv_buf_t* buf) {
-        buf->base = (char*)malloc(suggested_size);
-        buf->len = suggested_size;
-    }
-    void onRead(ssize_t nread, const uv_buf_t *buf) {
-        if (nread < 0) {
-            if (nread != UV_EOF) {
-                fprintf(stderr, "Read error %s\n", uv_err_name(nread));
-            } else {
-                fprintf(stderr, "EOF\n");
-            }
-            close();
-        } else if (nread > 0) {
-            if (read_cb)
-                read_cb((const uint8_t*)buf->base, nread);
-        }
-        if (buf->base)
-            free(buf->base);
-    }
-    void onWrite(int status, OnWrite&& cb) {
-        if (cb)
-            cb(status);
-    }
-    void onClosed() {
-        read_cb = {};
-        connect_cb = {};
-        if (close_cb) {
-            close_cb();
-            close_cb = {};
-        }
-        if (auto s = static_cast<std::shared_ptr<TcpSocket>*>(sock.data)) {
-            sock.data = nullptr;
-            delete s;
-        }
-    }
-    static std::shared_ptr<TcpSocket> get(uv_handle_t* handle) {
-        return handle->data ? *static_cast<std::shared_ptr<TcpSocket>*>(handle->data) : std::shared_ptr<TcpSocket>();
-    }
-    static void on_new_connection(uv_stream_t* handle, int status) {
-        if (auto node = get((uv_handle_t*)handle))
-            node->onNewConnection(status);
-    }
-    static void on_close(uv_handle_t* handle) {
-        if (auto node = get(handle))
-            node->onClosed();
-    }
-    static inline void on_alloc_buffer(uv_handle_t* handle, size_t suggested_size, uv_buf_t *buf) {
-        if (auto node = get((uv_handle_t*)handle))
-            node->onAllocBuffer(suggested_size, buf);
-    }
-    static void on_read(uv_stream_t* client, ssize_t nread, const uv_buf_t *buf) {
-        if (auto node = get((uv_handle_t*)client))
-            node->onRead(nread, buf);
-    }
-    static void on_write_end(uv_write_t* req, int status) {
-        if (auto write = static_cast<WriteReq*>(req->data)) {
-            write->sock->onWrite(status, std::move(write->callback));
-            if (write->msg)
-                free(write->msg);
-            delete write;
-        }
-    }
-    static void on_connect(uv_connect_t* req, int status) {
-        if (auto conn = static_cast<ConnectReq*>(req->data)) {
-            conn->sock->onConnected(status, std::move(conn->callback));
-            delete conn;
-        }
-    }
-};
-
-
-}
@@ -300,8 +300,6 @@ cdef class NodeSet(object):
        return s
    def __iter__(self):
        return NodeSetIter(self)
-    def __len__(self):
-        return self.size()

 cdef class PrivateKey(_WithID):
    cdef shared_ptr[cpp.PrivateKey] _key
@@ -444,10 +442,6 @@ cdef class DhtConfig(object):
        self._config.dht_config.node_config.node_id = id._infohash
    def setNetwork(self, unsigned netid):
        self._config.dht_config.node_config.network = netid
-    def setPort(self, cpp.in_port_t port):
-        self._config.dht_config.node_config.network_config.bind_port = port
-    def setBindAddr(self, string addr):
-        self._config.dht_config.node_config.network_config.bind_addr = addr
    def setMaintainStorage(self, bool maintain_storage):
        self._config.dht_config.node_config.maintain_storage = maintain_storage

@@ -466,7 +460,7 @@ cdef class DhtRunner(_WithID):
        if done_cb:
            cb_obj = {'done':done_cb}
            ref.Py_INCREF(cb_obj)
-            self.thisptr.get().ping(addr._addr, cpp.bindDoneCbSimple(done_callback_simple, <void*>cb_obj))
+            self.thisptr.get().bootstrap(addr._addr, cpp.bindDoneCbSimple(done_callback_simple, <void*>cb_obj))
        else:
            lock = threading.Condition()
            pending = 0
@@ -490,10 +484,12 @@ cdef class DhtRunner(_WithID):
    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)
-        config.setPort(port)
-        if ipv6:
-            config.setBindAddr(ipv6)
-        self.thisptr.get().run(config._config)
+        if ipv4 or ipv6:
+            bind4 = ipv4.encode() if ipv4 else b''
+            bind6 = ipv6.encode() if ipv6 else b''
+            self.thisptr.get().run(bind4, bind6, str(port).encode(), config._config)
+        else:
+            self.thisptr.get().run(port, config._config)
    def join(self):
        self.thisptr.get().join()
    def shutdown(self, shutdown_cb=None):
@@ -508,9 +504,9 @@ cdef class DhtRunner(_WithID):
        cpp.enableFileLogging(self.thisptr.get()[0], path.encode())
    def isRunning(self):
        return self.thisptr.get().isRunning()
-    def getBound(self):
+    def getBound(self, cpp.sa_family_t af = 0):
        s = SockAddr()
-        s._addr = self.thisptr.get().getBound()
+        s._addr = self.thisptr.get().getBound(af)
        return s
    def getStorageLog(self):
        return self.thisptr.get().getStorageLog().decode()
@@ -73,7 +73,7 @@ cdef extern from "opendht/sockaddr.h" namespace "dht":
        SockAddr() except +
        string toString() const
        in_port_t getPort() const
-        void setPort(in_port_t p) const
+        void setPort(in_port_t p)
        sa_family_t getFamily() const
        void setFamily(sa_family_t f)

@@ -198,15 +198,11 @@ cdef extern from "opendht/callbacks.h" namespace "dht":
    cdef DoneCallback bindDoneCb(DoneCallbackRaw cb, void *user_data)
    cdef DoneCallbackSimple bindDoneCbSimple(DoneCallbackSimpleRaw cb, void *user_data)

-    cppclass NetworkConfig:
-        string bind_addr
-        in_port_t bind_port
    cppclass Config:
        InfoHash node_id
        uint32_t network
        bool is_bootstrap
        bool maintain_storage
-        NetworkConfig network_config
    cppclass SecureDhtConfig:
        Config node_config
        Identity id
@@ -222,12 +218,13 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
        InfoHash getId() const
        InfoHash getNodeId() const
        void bootstrap(const_char*, const_char*)
-        void ping(const SockAddr&, DoneCallbackSimple done_cb)
-        void run(Config config)
+        void bootstrap(const SockAddr&, DoneCallbackSimple done_cb)
+        void run(in_port_t, Config config)
+        void run(const_char*, const_char*, const_char*, Config config)
        void join()
        void shutdown(ShutdownCallback)
        bool isRunning()
-        SockAddr getBound() const
+        SockAddr getBound(sa_family_t af) const
        string getStorageLog() const
        string getRoutingTablesLog(sa_family_t af) const
        string getSearchesLog(sa_family_t af) const
@@ -50,8 +50,8 @@ setup(name="opendht",
          ["@CURRENT_SOURCE_DIR@/opendht.pyx"],
          include_dirs = ['@PROJECT_SOURCE_DIR@/include'],
          language="c++",
-          extra_compile_args=["-std=c++14"],
-          extra_link_args=["-std=c++14"],
+          extra_compile_args=["-std=c++11"],
+          extra_link_args=["-std=c++11"],
          libraries=["opendht"],
          library_dirs = ['@CURRENT_BINARY_DIR@', '@PROJECT_BINARY_DIR@']
      ))
@@ -9,6 +9,7 @@ libopendht_la_SOURCES  = \
        listener.h \
        request.h \
        search.h \
+        net.h \
        parsed_message.h \
        node_cache.cpp \
        callbacks.cpp \
@@ -37,7 +38,6 @@ nobase_include_HEADERS = \
        ../include/opendht/node_cache.h \
        ../include/opendht/routing_table.h \
        ../include/opendht/network_engine.h \
-        ../include/opendht/net.h \
        ../include/opendht/scheduler.h \
        ../include/opendht/rate_limiter.h \
        ../include/opendht/utils.h \
@@ -116,7 +116,7 @@ bool aesKeySizeGood(size_t key_size)
 #define GCM_DIGEST_SIZE GCM_BLOCK_SIZE
 #endif

-Blob aesEncrypt(const Blob& data, const SecureBlob& key)
+Blob aesEncrypt(const Blob& data, const Blob& key)
 {
    if (not aesKeySizeGood(key.size()))
        throw DecryptError("Wrong key size");
@@ -127,7 +127,7 @@ Blob aesEncrypt(const Blob& data, const SecureBlob& key)
        std::generate_n(ret.begin(), GCM_IV_SIZE, std::bind(rand_byte, std::ref(rdev)));
    }
    struct gcm_aes_ctx aes;
-    gcm_aes_set_key(&aes, key.size(), key.makeInsecure().data());
+    gcm_aes_set_key(&aes, key.size(), key.data());
    gcm_aes_set_iv(&aes, GCM_IV_SIZE, ret.data());
    gcm_aes_update(&aes, data.size(), data.data());

@@ -139,13 +139,13 @@ Blob aesEncrypt(const Blob& data, const SecureBlob& key)
 Blob aesEncrypt(const Blob& data, const std::string& password)
 {
    Blob salt;
-    SecureBlob key = stretchKey(password, salt, 256 / 8);
+    Blob key = stretchKey(password, salt, 256 / 8);
    Blob encrypted = aesEncrypt(data, key);
    encrypted.insert(encrypted.begin(), salt.begin(), salt.end());
    return encrypted;
 }

-Blob aesDecrypt(const Blob& data, const SecureBlob& key)
+Blob aesDecrypt(const Blob& data, const Blob& key)
 {
    if (not aesKeySizeGood(key.size()))
        throw DecryptError("Wrong key size");
@@ -185,12 +185,12 @@ Blob aesDecrypt(const Blob& data, const std::string& password)
    if (data.size() <= PASSWORD_SALT_LENGTH)
        throw DecryptError("Wrong data size");
    Blob salt {data.begin(), data.begin()+PASSWORD_SALT_LENGTH};
-    SecureBlob key = stretchKey(password, salt, 256/8);
+    Blob key = stretchKey(password, salt, 256/8);
    Blob encrypted {data.begin()+PASSWORD_SALT_LENGTH, data.end()};
    return aesDecrypt(encrypted, key);
 }

-SecureBlob stretchKey(const std::string& password, Blob& salt, size_t key_length)
+Blob stretchKey(const std::string& password, Blob& salt, size_t key_length)
 {
    if (salt.empty()) {
        salt.resize(PASSWORD_SALT_LENGTH);
@@ -506,7 +506,11 @@ PublicKey::encrypt(const Blob& data) const
    unsigned aes_key_sz = aesKeySize(max_block_sz);
    if (aes_key_sz == 0)
        throw CryptoException("Key is not long enough for AES128");
-    auto key = SecureBlob::getRandom(aes_key_sz);
+    Blob key(aes_key_sz);
+    {
+        crypto::random_device rdev;
+        std::generate_n(key.begin(), key.size(), std::bind(rand_byte, std::ref(rdev)));
+    }
    auto data_encrypted = aesEncrypt(data, key);

    Blob ret;
@@ -525,7 +529,7 @@ PublicKey::getId() const
        return {};
    InfoHash id;
    size_t sz = id.size();
-    constexpr int flags = (id.size() == 32) ? GNUTLS_KEYID_USE_SHA256 : 0;
+    const int flags = (id.size() == 32) ? GNUTLS_KEYID_USE_SHA256 : 0;
    if (auto err = gnutls_pubkey_get_key_id(pk, flags, id.data(), &sz))
        throw CryptoException(std::string("Can't get public key ID: ") + gnutls_strerror(err));
    if (sz != id.size())
@@ -937,7 +941,7 @@ Certificate::generate(const PrivateKey& key, const std::string& name, Identity c

    if (ca.first && ca.second) {
        if (not ca.second->isCA()) {
-            std::cerr << "Error: signing certificate must be CA." << std::endl;
+            // Signing certificate must be CA.
            return {};
        }
        //if (gnutls_x509_crt_sign2(cert, ca.second->cert, ca.first->x509_key, get_dig(cert), 0) != GNUTLS_E_SUCCESS) {
@@ -62,16 +62,8 @@ Dht::getStatus(sa_family_t af) const
 void
 Dht::shutdown(ShutdownCallback cb)
 {
-    auto final_shutdown = [this,cb](){
-        scheduler.stop();
-        DHT_LOG.w("closing socket...");
-        network_engine.close([=](){
-            if (cb)
-                cb();
-        });
-    };
    if (not maintain_storage) {
-        final_shutdown();
+        if (cb) cb();
        return;
    }

@@ -81,7 +73,7 @@ Dht::shutdown(ShutdownCallback cb)
    auto str_donecb = [=](bool, const std::vector<Sp<Node>>&) {
        --*remaining;
        DHT_LOG.w("shuting down node: %u ops remaining", *remaining);
-        if (!*remaining) { final_shutdown(); }
+        if (!*remaining && cb) { cb(); }
    };

    for (auto& str : store)
@@ -89,7 +81,8 @@ Dht::shutdown(ShutdownCallback cb)

    if (!*remaining) {
        DHT_LOG.w("shuting down node: %u ops remaining", *remaining);
-        final_shutdown();
+        if (cb)
+            cb();
    }
 }

@@ -535,7 +528,7 @@ Dht::searchStep(Sp<Search> sr)
        }

        // true if this node is part of the target nodes cluter.
-        /*bool in = sr->id.xorCmp(getNodeId(), sr->nodes.back().node->id) < 0;
+        /*bool in = sr->id.xorCmp(myid, sr->nodes.back().node->id) < 0;

        DHT_LOG_DEBUG("[search %s IPv%c] synced%s",
                sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', in ? ", in" : "");*/
@@ -560,8 +553,8 @@ Dht::searchStep(Sp<Search> sr)
                        [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
                        { /* on done */
                            if (auto sr = ws.lock()) {
-                                onListenDone(req.node, answer, sr);
                                scheduler.edit(sr->nextSearchStep, scheduler.time());
+                                onListenDone(req.node, answer, sr);
                            }
                        },
                        [this,ws,query](const net::Request& req, bool over) mutable
@@ -576,8 +569,8 @@ Dht::searchStep(Sp<Search> sr)
                        [this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
                        { /* on new values */
                            if (auto sr = ws.lock()) {
-                                onGetValuesDone(node, answer, sr, query);
                                scheduler.edit(sr->nextSearchStep, scheduler.time());
+                                onGetValuesDone(node, answer, sr, query);
                            }
                        }
                    );
@@ -707,7 +700,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
    if (sr->nextSearchStep)
        scheduler.edit(sr->nextSearchStep, sr->getNextStepTime(scheduler.time()));
    else
-        sr->nextSearchStep = scheduler.run(std::bind(&Dht::searchStep, this, sr));
+        sr->nextSearchStep = scheduler.add(scheduler.time(), std::bind(&Dht::searchStep, this, sr));

    return sr;
 }
@@ -915,24 +908,24 @@ Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point cr
    DHT_LOG.d(id, "put: adding %s -> %s", id.toString().c_str(), val->toString().c_str());

    auto op = std::make_shared<OpStatus>();
-    auto donecb = [callback](std::vector<Sp<Node>>&& nodes, OpStatus& op) {
+    auto donecb = [callback](const std::vector<Sp<Node>>& nodes, OpStatus& op) {
        // Callback as soon as the value is announced on one of the available networks
        if (callback and not op.status.done and (op.status4.done && op.status6.done)) {
-            callback(op.status4.ok or op.status6.ok, std::move(nodes));
+            callback(op.status4.ok or op.status6.ok, nodes);
            op.status.done = true;
        }
    };
-    announce(id, AF_INET, val, [=](bool ok4, std::vector<Sp<Node>>&& nodes) {
+    announce(id, AF_INET, val, [=](bool ok4, const std::vector<Sp<Node>>& nodes) {
        DHT_LOG.d(id, "Announce done IPv4 %d", ok4);
        auto& o = *op;
        o.status4 = {true, ok4};
-        donecb(std::move(nodes), o);
+        donecb(nodes, o);
    }, created, permanent);
-    announce(id, AF_INET6, val, [=](bool ok6, std::vector<Sp<Node>>&& nodes) {
+    announce(id, AF_INET6, val, [=](bool ok6, const std::vector<Sp<Node>>& nodes) {
        DHT_LOG.d(id, "Announce done IPv6 %d", ok6);
        auto& o = *op;
        o.status6 = {true, ok6};
-        donecb(std::move(nodes), o);
+        donecb(nodes, o);
    }, created, permanent);
 }

@@ -945,7 +938,7 @@ void doneCallbackWrapper(DoneCallback dcb, const std::vector<Sp<Node>>& nodes, G
        bool ok = op.status.ok or op.status4.ok or op.status6.ok;
        op.status.done = true;
        if (dcb)
-            dcb(ok, std::move(op.nodes));
+            dcb(ok, op.nodes);
    }
 }

@@ -968,20 +961,6 @@ bool callbackWrapper(Cb get_cb,
    return !op.status.ok;
 }

-
-void
-Dht::bootstrap(const std::string& host, const std::string& service)
-{
-    for (const auto& bn : bootstrap_nodes)
-        if (bn.first == host and bn.second == service)
-            return;
-    bootstrap_nodes.emplace_back(host, service);
-    getAddrInfo(scheduler.getLoop(), host.c_str(), service.c_str(), [this](std::vector<SockAddr>&& res) {
-        for (const auto& addr : res)
-            pingNode(addr);
-    });
-}
-
 void
 Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filter&& filter, Where&& where)
 {
@@ -1009,15 +988,15 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
    /* Try to answer this search locally. */
    gcb(getLocal(id, f));

-    Dht::search(id, AF_INET, gcb, {}, [=](bool ok, std::vector<Sp<Node>>&& nodes) {
+    Dht::search(id, AF_INET, gcb, {}, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
        //DHT_LOG_WARN("DHT done IPv4");
        op->status4 = {true, ok};
-        doneCallbackWrapper(donecb, std::move(nodes), *op);
+        doneCallbackWrapper(donecb, nodes, *op);
    }, f, q);
-    Dht::search(id, AF_INET6, gcb, {}, [=](bool ok, std::vector<Sp<Node>>&& nodes) {
+    Dht::search(id, AF_INET6, gcb, {}, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
        //DHT_LOG_WARN("DHT done IPv6");
        op->status6 = {true, ok};
-        doneCallbackWrapper(donecb, std::move(nodes), *op);
+        doneCallbackWrapper(donecb, nodes, *op);
    }, f, q);
 }

@@ -1056,15 +1035,15 @@ void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Quer
    /* Try to answer this search locally. */
    qcb(local_fields);

-    Dht::search(id, AF_INET, {}, qcb, [=](bool ok, std::vector<Sp<Node>>&& nodes) {
+    Dht::search(id, AF_INET, {}, qcb, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
        //DHT_LOG_WARN("DHT done IPv4");
        op->status4 = {true, ok};
-        doneCallbackWrapper(done_cb, std::move(nodes), *op);
+        doneCallbackWrapper(done_cb, nodes, *op);
    }, f, q);
-    Dht::search(id, AF_INET6, {}, qcb, [=](bool ok, std::vector<Sp<Node>>&& nodes) {
+    Dht::search(id, AF_INET6, {}, qcb, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
        //DHT_LOG_WARN("DHT done IPv6");
        op->status6 = {true, ok};
-        doneCallbackWrapper(done_cb, std::move(nodes), *op);
+        doneCallbackWrapper(done_cb, nodes, *op);
    }, f, q);
 }

@@ -1321,7 +1300,6 @@ Dht::expireStore()
 void
 Dht::connectivityChanged(sa_family_t af)
 {
-    updateStatus();
    const auto& now = scheduler.time();
    scheduler.edit(nextNodesConfirmation, now);
    buckets(af).connectivityChanged(now);
@@ -1346,13 +1324,14 @@ Dht::rotateSecrets()
        secret = std::uniform_int_distribution<uint64_t>{}(rdev);
    }
    uniform_duration_distribution<> time_dist(std::chrono::minutes(15), std::chrono::minutes(45));
-    scheduler.add(time_dist(rd), std::bind(&Dht::rotateSecrets, this));
+    auto rotate_secrets_time = scheduler.time() + time_dist(rd);
+    scheduler.add(rotate_secrets_time, std::bind(&Dht::rotateSecrets, this));
 }

 Blob
 Dht::makeToken(const SockAddr& addr, bool old) const
 {
-    const void* ip;
+    const void *ip;
    size_t iplen;
    in_port_t port;

@@ -1361,12 +1340,12 @@ Dht::makeToken(const SockAddr& addr, bool old) const
        const auto& sin = addr.getIPv4();
        ip = &sin.sin_addr;
        iplen = 4;
-        port = ntohs(sin.sin_port);
+        port = sin.sin_port;
    } else if (family == AF_INET6) {
        const auto& sin6 = addr.getIPv6();
        ip = &sin6.sin6_addr;
        iplen = 16;
-        port = ntohs(sin6.sin6_port);
+        port = sin6.sin6_port;
    } else {
        return {};
    }
@@ -1410,7 +1389,7 @@ Dht::getNodesStats(sa_family_t af) const
        if (b.cached)
            stats.cached_nodes++;
    }
-    stats.table_depth = bcks.depth(bcks.findBucket(getNodeId()));
+    stats.table_depth = bcks.depth(bcks.findBucket(myid));
    return stats;
 }

@@ -1533,7 +1512,7 @@ void
 Dht::dumpTables() const
 {
    std::stringstream out;
-    out << "My id " << getNodeId() << std::endl;
+    out << "My id " << myid << std::endl;

    out << "Buckets IPv4 :" << std::endl;
    for (const auto& b : buckets4)
@@ -1674,13 +1653,14 @@ Dht::~Dht()
        s.second->clear();
 }

-Dht::Dht() : store(), scheduler(nullptr, DHT_LOG), network_engine(nullptr, DHT_LOG, scheduler) {}
+Dht::Dht() : store(), scheduler(DHT_LOG), network_engine(DHT_LOG, scheduler) {}

-Dht::Dht(uv_loop_t* loop, Config config)
-    : is_bootstrap(config.is_bootstrap), maintain_storage(config.maintain_storage),
-    buckets4{Bucket {AF_INET}},buckets6{Bucket {AF_INET6}}, store(), store_quota(),
-    scheduler(loop, DHT_LOG),
-    network_engine(loop, config.network_config, config.network, DHT_LOG, scheduler,
+Dht::Dht(int s, int s6, Config config)
+    : myid(config.node_id != zeroes ? config.node_id : InfoHash::getRandom()),
+    is_bootstrap(config.is_bootstrap),
+    maintain_storage(config.maintain_storage), store(), store_quota(),
+    scheduler(DHT_LOG),
+    network_engine(myid, config.network, s, s6, DHT_LOG, scheduler,
            std::bind(&Dht::onError, this, _1, _2),
            std::bind(&Dht::onNewNode, this, _1, _2),
            std::bind(&Dht::onReportedAddr, this, _1, _2),
@@ -1692,12 +1672,23 @@ Dht::Dht(uv_loop_t* loop, Config config)
            std::bind(&Dht::onRefresh, this, _1, _2, _3, _4))
 {
    scheduler.syncTime();
-    buckets4.is_client = config.is_bootstrap;
-    buckets6.is_client = config.is_bootstrap;
+    if (s < 0 && s6 < 0)
+        return;
+    if (s >= 0) {
+        buckets4 = {Bucket {AF_INET}};
+        buckets4.is_client = config.is_bootstrap;
+    }
+    if (s6 >= 0) {
+        buckets6 = {Bucket {AF_INET6}};
+        buckets6.is_client = config.is_bootstrap;
+    }
+
    search_id = std::uniform_int_distribution<decltype(search_id)>{}(rd);

    uniform_duration_distribution<> time_dis {std::chrono::seconds(3), std::chrono::seconds(5)};
-    nextNodesConfirmation = scheduler.add(time_dis(rd), std::bind(&Dht::confirmNodes, this));
+    auto confirm_nodes_time = scheduler.time() + time_dis(rd);
+    DHT_LOG.d(myid, "Scheduling %s", myid.toString().c_str());
+    nextNodesConfirmation = scheduler.add(confirm_nodes_time, std::bind(&Dht::confirmNodes, this));

    // Fill old secret
    {
@@ -1708,7 +1699,7 @@ Dht::Dht(uv_loop_t* loop, Config config)

    expire();

-    DHT_LOG.d("DHT initialised with node ID %s", getNodeId().toString().c_str());
+    DHT_LOG.d("DHT initialised with node ID %s", myid.toString().c_str());
 }


@@ -1716,11 +1707,11 @@ bool
 Dht::neighbourhoodMaintenance(RoutingTable& list)
 {
    //DHT_LOG_DEBUG("neighbourhoodMaintenance");
-    auto b = list.findBucket(getNodeId());
+    auto b = list.findBucket(myid);
    if (b == list.end())
        return false;

-    InfoHash id = getNodeId();
+    InfoHash id = myid;
 #ifdef _WIN32
    std::uniform_int_distribution<int> rand_byte{ 0, std::numeric_limits<uint8_t>::max() };
 #else
@@ -1744,13 +1735,7 @@ Dht::neighbourhoodMaintenance(RoutingTable& list)
                n->toString().c_str(), id.toString().c_str());
        /* Since our node-id is the same in both DHTs, it's probably
           profitable to query both families. */
-        network_engine.sendFindNode(n, id, network_engine.want(), nullptr, [this,n](const net::Request&, bool over) {
-            if (over) {
-                DHT_LOG.d(n->id, "[node %s] neighborhood maintenance op expired", n->toString().c_str());
-                scheduler.edit(nextNodesConfirmation, Node::MAX_RESPONSE_TIME);
-                updateStatus();
-            }
-        });
+        network_engine.sendFindNode(n, id, network_engine.want(), nullptr, nullptr);
    }

    return true;
@@ -1764,7 +1749,7 @@ Dht::bucketMaintenance(RoutingTable& list)

    bool sent {false};
    for (auto b = list.begin(); b != list.end(); ++b) {
-        if (b->time < scheduler.time() - std::chrono::minutes(10) || b->expired()) {
+        if (b->time < scheduler.time() - std::chrono::minutes(10) || b->nodes.empty()) {
            /* This bucket hasn't seen any positive confirmation for a long
               time. Pick a random id in this bucket's range, and send a request
               to a random node. */
@@ -1773,11 +1758,11 @@ Dht::bucketMaintenance(RoutingTable& list)
            /* If the bucket is empty, we try to fill it from a neighbour.
               We also sometimes do it gratuitiously to recover from
               buckets full of broken nodes. */
-            if (std::next(b) != list.end() && (q->expired() || rand_trial(rd)))
+            if (std::next(b) != list.end() && (q->nodes.empty() || rand_trial(rd)))
                q = std::next(b);
-            if (b != list.begin() && (q->expired() || rand_trial(rd))) {
+            if (b != list.begin() && (q->nodes.empty() || rand_trial(rd))) {
                auto r = std::prev(b);
-                if (!r->expired())
+                if (!r->nodes.empty())
                    q = r;
            }

@@ -1787,23 +1772,26 @@ Dht::bucketMaintenance(RoutingTable& list)

                if (network_engine.want() != want) {
                    auto otherbucket = findBucket(id, q->af == AF_INET ? AF_INET6 : AF_INET);
-                    if ((otherbucket and otherbucket->nodes.size() < TARGET_NODES) or rand_trial_38(rd))
+                    if (otherbucket && otherbucket->nodes.size() < TARGET_NODES)
                        /* The corresponding bucket in the other family
                           is emptyish -- querying both is useful. */
                        want = WANT4 | WANT6;
+                    else if (rand_trial_38(rd))
                        /* Most of the time, this just adds overhead.
                           However, it might help stitch back one of
                           the DHTs after a network collapse, so query
                           both, but only very occasionally. */
+                        want = WANT4 | WANT6;
                }

                DHT_LOG.d(id, n->id, "[node %s] sending find %s for bucket maintenance", n->toString().c_str(), id.toString().c_str());
                auto start = scheduler.time();
                network_engine.sendFindNode(n, id, want, nullptr, [this,start,n](const net::Request&, bool over) {
                    if (over) {
-                        DHT_LOG.d(n->id, "[node %s] bucket maintenance op expired", n->toString().c_str());
-                        scheduler.edit(nextNodesConfirmation, Node::MAX_RESPONSE_TIME);
-                        updateStatus();
+                        const auto& end = scheduler.time();
+                        using namespace std::chrono;
+                        DHT_LOG.d(n->id, "[node %s] bucket maintenance op expired after %llu ms", n->toString().c_str(), duration_cast<milliseconds>(end-start).count());
+                        scheduler.edit(nextNodesConfirmation, end + Node::MAX_RESPONSE_TIME);
                    }
                });
                sent = true;
@@ -1837,7 +1825,7 @@ Dht::maintainStorage(decltype(store)::value_type& storage, bool force, DoneCallb

    auto nodes = buckets4.findClosestNodes(storage.first, now);
    if (!nodes.empty()) {
-        if (force || storage.first.xorCmp(nodes.back()->id, getNodeId()) < 0) {
+        if (force || storage.first.xorCmp(nodes.back()->id, myid) < 0) {
            for (auto &value : storage.second.getValues()) {
                const auto& vt = getType(value.data->type);
                if (force || value.created + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
@@ -1852,7 +1840,7 @@ Dht::maintainStorage(decltype(store)::value_type& storage, bool force, DoneCallb

    auto nodes6 = buckets6.findClosestNodes(storage.first, now);
    if (!nodes6.empty()) {
-        if (force || storage.first.xorCmp(nodes6.back()->id, getNodeId()) < 0) {
+        if (force || storage.first.xorCmp(nodes6.back()->id, myid) < 0) {
            for (auto &value : storage.second.getValues()) {
                const auto& vt = getType(value.data->type);
                if (force || value.created + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
@@ -1875,7 +1863,7 @@ Dht::maintainStorage(decltype(store)::value_type& storage, bool force, DoneCallb
    return announce_per_af;
 }

-/*time_point
+time_point
 Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from)
 {
    scheduler.syncTime();
@@ -1887,37 +1875,19 @@ Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from)
        }
    }
    return scheduler.run();
-}*/
-
-void
-Dht::updateStatus()
-{
-    NodeStatus nstatus4 = getStatus(AF_INET);
-    NodeStatus nstatus6 = getStatus(AF_INET6);
-    if (nstatus4 != status4 || nstatus6 != status6) {
-        status4 = nstatus4;
-        status6 = nstatus6;
-        if (status4 == NodeStatus::Disconnected and status6 == NodeStatus::Disconnected) {
-            // We have lost connection with the DHT.  Try to recover using bootstrap nodes.
-            DHT_LOG.e(getNodeId(), "DHT disconnected", getNodeId().toString().c_str());
-        } else {
-            //bootstrap_nodes.clear();
-        }
-        if (statusCb)
-            statusCb(status4, status6);
-    }
 }

 void
 Dht::expire()
 {
    uniform_duration_distribution<> time_dis(std::chrono::minutes(2), std::chrono::minutes(6));
+    auto expire_stuff_time = scheduler.time() + duration(time_dis(rd));

    expireBuckets(buckets4);
    expireBuckets(buckets6);
    expireStore();
    expireSearches();
-    scheduler.add(time_dis(rd), std::bind(&Dht::expire, this));
+    scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this));
 }

 void
@@ -1927,15 +1897,13 @@ Dht::confirmNodes()
    bool soon = false;
    const auto& now = scheduler.time();

-    updateStatus();
-
-    if (searches4.empty() and status4 == NodeStatus::Connected) {
-        DHT_LOG.d(getNodeId(), "[confirm nodes] initial IPv4 'get' for my id (%s)", getNodeId().toString().c_str());
-        search(getNodeId(), AF_INET);
+    if (searches4.empty() and getStatus(AF_INET) == NodeStatus::Connected) {
+        DHT_LOG.d(myid, "[confirm nodes] initial IPv4 'get' for my id (%s)", myid.toString().c_str());
+        search(myid, AF_INET);
    }
-    if (searches6.empty() and status6 == NodeStatus::Connected) {
-        DHT_LOG.d(getNodeId(), "[confirm nodes] initial IPv6 'get' for my id (%s)", getNodeId().toString().c_str());
-        search(getNodeId(), AF_INET6);
+    if (searches6.empty() and getStatus(AF_INET6) == NodeStatus::Connected) {
+        DHT_LOG.d(myid, "[confirm nodes] initial IPv6 'get' for my id (%s)", myid.toString().c_str());
+        search(myid, AF_INET6);
    }

    soon |= bucketMaintenance(buckets4);
@@ -1955,7 +1923,9 @@ Dht::confirmNodes()
    auto time_dis = soon
        ? uniform_duration_distribution<> {seconds(5) , seconds(25)}
        : uniform_duration_distribution<> {seconds(60), seconds(180)};
-    scheduler.edit(nextNodesConfirmation, time_dis(rd));
+    auto confirm_nodes_time = now + time_dis(rd);
+
+    scheduler.edit(nextNodesConfirmation, confirm_nodes_time);
 }

 std::vector<ValuesExport>
@@ -2025,13 +1995,13 @@ Dht::exportNodes()
 {
    const auto& now = scheduler.time();
    std::vector<NodeExport> nodes;
-    const auto b4 = buckets4.findBucket(getNodeId());
+    const auto b4 = buckets4.findBucket(myid);
    if (b4 != buckets4.end()) {
        for (auto& n : b4->nodes)
            if (n->isGood(now))
                nodes.push_back(n->exportNode());
    }
-    const auto b6 = buckets6.findBucket(getNodeId());
+    const auto b6 = buckets6.findBucket(myid);
    if (b6 != buckets6.end()) {
        for (auto& n : b6->nodes)
            if (n->isGood(now))
@@ -2062,21 +2032,21 @@ Dht::insertNode(const InfoHash& id, const SockAddr& addr)
 }

 void
-Dht::pingNode(const SockAddr& sa, DoneCallback&& cb)
+Dht::pingNode(const sockaddr* sa, socklen_t salen, DoneCallbackSimple&& cb)
 {
    scheduler.syncTime();
-    DHT_LOG.d("Sending ping to %s", sa.toString().c_str());
-    auto& count = sa.getFamily() == AF_INET ? pending_pings4 : pending_pings6;
+    DHT_LOG.d("Sending ping to %s", print_addr(sa, salen).c_str());
+    auto& count = sa->sa_family == AF_INET ? pending_pings4 : pending_pings6;
    count++;
-    network_engine.sendPing(sa, [&count,cb](const net::Request& req, net::RequestAnswer&&) {
+    network_engine.sendPing(sa, salen, [&count,cb](const net::Request&, net::RequestAnswer&&) {
        count--;
        if (cb)
-            cb(true, {req.node});
+            cb(true);
    }, [&count,cb](const net::Request&, bool last){
        if (last) {
            count--;
            if (cb)
-                cb(false, {});
+                cb(false);
        }
    });
 }
@@ -2237,7 +2207,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
            net::DhtProtocolException::LISTEN_NO_INFOHASH
        };
    }
-    if (not tokenMatch(token, node->getAddress())) {
+    if (!tokenMatch(token, node->getAddr())) {
        DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str());
        throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
    }
@@ -2276,7 +2246,7 @@ Dht::onAnnounce(Sp<Node> n,
            net::DhtProtocolException::PUT_NO_INFOHASH
        };
    }
-    if (not tokenMatch(token, node.getAddr())) {
+    if (!tokenMatch(token, node.getAddr())) {
        DHT_LOG.w(hash, node.id, "[node %s] incorrect token %s for 'put'", node.toString().c_str(), hash.toString().c_str());
        throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::PUT_WRONG_TOKEN};
    }
@@ -2284,7 +2254,7 @@ Dht::onAnnounce(Sp<Node> n,
        // We store a value only if we think we're part of the
        // SEARCH_NODES nodes around the target id.
        auto closest_nodes = buckets(node.getFamily()).findClosestNodes(hash, scheduler.time(), SEARCH_NODES);
-        if (closest_nodes.size() >= TARGET_NODES and hash.xorCmp(closest_nodes.back()->id, getNodeId()) < 0) {
+        if (closest_nodes.size() >= TARGET_NODES and hash.xorCmp(closest_nodes.back()->id, myid) < 0) {
            DHT_LOG.w(hash, node.id, "[node %s] announce too far from the target. Dropping value.", node.toString().c_str());
            return {};
        }
@@ -20,32 +20,98 @@
 #include "dhtrunner.h"
 #include "securedht.h"

+#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)
+#endif
+
 namespace dht {

 constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;

 DhtRunner::DhtRunner() : dht_()
 {
-    uv_async_.data = this;
-    uv_async_init(uv_loop_.get(), &uv_async_, &DhtRunner::loop_callback);
+#ifdef _WIN32
+    WSADATA wsd;
+    if (WSAStartup(MAKEWORD(2,2), &wsd) != 0)
+        throw DhtException("Can't initialize Winsock2");
+#endif
 }

 DhtRunner::~DhtRunner()
 {
    join();
+#ifdef _WIN32
+    WSACleanup();
+#endif
 }

 void
-DhtRunner::run(DhtRunner::Config config)
+DhtRunner::run(in_port_t port, DhtRunner::Config config)
+{
+    sockaddr_in sin4;
+    std::fill_n((uint8_t*)&sin4, sizeof(sin4), 0);
+    sin4.sin_family = AF_INET;
+    sin4.sin_port = htons(port);
+    sockaddr_in6 sin6;
+    std::fill_n((uint8_t*)&sin6, sizeof(sin6), 0);
+    sin6.sin6_family = AF_INET6;
+    sin6.sin6_port = htons(port);
+    run(&sin4, &sin6, config);
+}
+
+void
+DhtRunner::run(const char* ip4, const char* ip6, const char* service, DhtRunner::Config config)
+{
+    auto res4 = getAddrInfo(ip4, service);
+    auto res6 = getAddrInfo(ip6, service);
+    run(res4.empty() ? nullptr : (sockaddr_in*) &res4.front().first,
+        res6.empty() ? nullptr : (sockaddr_in6*)&res6.front().first, config);
+}
+
+void
+DhtRunner::run(const sockaddr_in* local4, const sockaddr_in6* local6, DhtRunner::Config config)
 {
    if (running)
        return;
+    if (rcv_thread.joinable())
+        rcv_thread.join();
    running = true;
-    doRun(config.dht_config);
+    doRun(local4, local6, config.dht_config);
    if (not config.threaded)
        return;
    dht_thread = std::thread([this]() {
-        uv_loop_.run();
+        while (running) {
+            std::unique_lock<std::mutex> lk(dht_mtx);
+            auto wakeup = loop_();
+            cv.wait_until(lk, wakeup, [this]() {
+                if (not running)
+                    return true;
+                {
+                    std::lock_guard<std::mutex> lck(sock_mtx);
+                    if (not rcv.empty())
+                        return true;
+                }
+                {
+                    std::lock_guard<std::mutex> lck(storage_mtx);
+                    if (not pending_ops_prio.empty())
+                        return true;
+                    auto s = getStatus();
+                    if (not pending_ops.empty() and (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping)))
+                        return true;
+                }
+                return false;
+            });
+        }
    });
 }

@@ -55,19 +121,21 @@ DhtRunner::shutdown(ShutdownCallback cb) {
    pending_ops_prio.emplace([=](SecureDht& dht) mutable {
        dht.shutdown(cb);
    });
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
 DhtRunner::join()
 {
    running = false;
-    uv_async_send(&uv_async_);
-    //bootstrap_cv.notify_all();
+    cv.notify_all();
+    bootstrap_cv.notify_all();
    if (dht_thread.joinable())
        dht_thread.join();
-    /*if (bootstrap_thread.joinable())
-        bootstrap_thread.join();*/
+    if (rcv_thread.joinable())
+        rcv_thread.join();
+    if (bootstrap_thread.joinable())
+        bootstrap_thread.join();

    {
        std::lock_guard<std::mutex> lck(storage_mtx);
@@ -77,10 +145,10 @@ DhtRunner::join()
    {
        std::lock_guard<std::mutex> lck(dht_mtx);
        dht_.reset();
-        /*status4 = NodeStatus::Disconnected;
+        status4 = NodeStatus::Disconnected;
        status6 = NodeStatus::Disconnected;
        bound4 = {};
-        bound6 = {};*/
+        bound6 = {};
    }
 }

@@ -107,14 +175,6 @@ DhtRunner::getNodeId() const
    return dht_->getNodeId();
 }

-void
-DhtRunner::setOnStatusChanged(StatusCallback&& cb)
-{
-    std::lock_guard<std::mutex> lck(dht_mtx);
-    if (not dht_)
-        throw std::runtime_error("dht is not running");
-    dht_->setOnStatusChanged(std::move(cb));
-}

 std::pair<size_t, size_t>
 DhtRunner::getStoreSize() const {
@@ -247,11 +307,6 @@ DhtRunner::getPublicAddressStr(sa_family_t af)
    return ret;
 }

-SockAddr 
-DhtRunner::getBound() const {
-    return dht_->getBoundAddr();
-}
-
 void
 DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
    std::lock_guard<std::mutex> lck(dht_mtx);
@@ -263,12 +318,42 @@ DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
    dht_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
 }

-void
+time_point
 DhtRunner::loop_()
 {
    if (!dht_)
-        return;
-/*
+        return {};
+
+    decltype(pending_ops) ops {};
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
+        auto s = getStatus();
+        ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping))) ?
+               std::move(pending_ops) : std::move(pending_ops_prio);
+    }
+    while (not ops.empty()) {
+        ops.front()(*dht_);
+        ops.pop();
+    }
+
+    time_point wakeup {};
+    decltype(rcv) received {};
+    {
+        std::lock_guard<std::mutex> lck(sock_mtx);
+        // move to stack
+        received = std::move(rcv);
+    }
+    if (not received.empty()) {
+        for (const auto& pck : received) {
+            auto& buf = pck.first;
+            auto& from = pck.second;
+            wakeup = dht_->periodic(buf.data(), buf.size()-1, from);
+        }
+        received.clear();
+    } else {
+        wakeup = dht_->periodic(nullptr, 0, nullptr, 0);
+    }
+
    NodeStatus nstatus4 = dht_->getStatus(AF_INET);
    NodeStatus nstatus6 = dht_->getStatus(AF_INET6);
    if (nstatus4 != status4 || nstatus6 != status6) {
@@ -286,24 +371,106 @@ DhtRunner::loop_()
        if (statusCb)
            statusCb(status4, status6);
    }
-*/
-    decltype(pending_ops) ops {};
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        auto s = dht_->getStatus();
-        ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping))) ?
-               std::move(pending_ops) : std::move(pending_ops_prio);
-    }
-    while (not ops.empty()) {
-        ops.front()(*dht_);
-        ops.pop();
-    }
+
+    return wakeup;
 }

 void
-DhtRunner::doRun(SecureDht::Config config)
+DhtRunner::doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDht::Config config)
 {
-    dht_ = std::unique_ptr<SecureDht>(new SecureDht {uv_loop_.get(), config});
+    dht_.reset();
+
+    int s4 = -1,
+        s6 = -1;
+
+    bound4 = {};
+    if (sin4) {
+        s4 = socket(PF_INET, SOCK_DGRAM, 0);
+        if(s4 >= 0) {
+            int rc = bind(s4, (sockaddr*)sin4, sizeof(sockaddr_in));
+            if(rc < 0)
+                throw DhtException("Can't bind IPv4 socket on " + dht::print_addr((sockaddr*)sin4, sizeof(sockaddr_in)));
+            sockaddr_storage ss;
+            socklen_t ss_len = sizeof(ss);
+            getsockname(s4, (sockaddr*)&ss, &ss_len);
+            bound4 = {ss, ss_len};
+        }
+    }
+
+#if 1
+    bound6 = {};
+    if (sin6) {
+        s6 = socket(PF_INET6, SOCK_DGRAM, 0);
+        if(s6 >= 0) {
+            int val = 1;
+            int rc = setsockopt(s6, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&val, sizeof(val));
+            if(rc < 0)
+                throw DhtException("Can't set IPV6_V6ONLY");
+
+            rc = bind(s6, (sockaddr*)sin6, sizeof(sockaddr_in6));
+            if(rc < 0)
+                throw DhtException("Can't bind IPv6 socket on " + dht::print_addr((sockaddr*)sin6, sizeof(sockaddr_in6)));
+            sockaddr_storage ss;
+            socklen_t ss_len = sizeof(ss);
+            getsockname(s6, (sockaddr*)&ss, &ss_len);
+            bound6 = {ss, ss_len};
+        }
+    }
+#endif
+
+    dht_ = std::unique_ptr<SecureDht>(new SecureDht {s4, s6, config});
+
+    rcv_thread = std::thread([this,s4,s6]() {
+        try {
+            while (true) {
+                struct timeval tv {/*.tv_sec = */0, /*.tv_usec = */250000};
+                fd_set readfds;
+
+                FD_ZERO(&readfds);
+                if(s4 >= 0)
+                    FD_SET(s4, &readfds);
+                if(s6 >= 0)
+                    FD_SET(s6, &readfds);
+
+                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, &tv);
+                if(rc < 0) {
+                    if(errno != EINTR) {
+                        perror("select");
+                        std::this_thread::sleep_for( std::chrono::seconds(1) );
+                    }
+                }
+
+                if(!running)
+                    break;
+
+                if(rc > 0) {
+                    std::array<uint8_t, 1024 * 64> buf;
+                    sockaddr_storage from;
+                    socklen_t from_len = sizeof(from);
+
+                    if(s4 >= 0 && FD_ISSET(s4, &readfds))
+                        rc = recvfrom(s4, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
+                    else if(s6 >= 0 && FD_ISSET(s6, &readfds))
+                        rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
+                    else
+                        break;
+                    if (rc > 0) {
+                        {
+                            std::lock_guard<std::mutex> lck(sock_mtx);
+                            rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len));
+                        }
+                        cv.notify_all();
+                    }
+                }
+            }
+        } catch (const std::exception& e) {
+            std::cerr << "Error in DHT networking thread: " << e.what() << std::endl;
+        }
+        if (s4 >= 0)
+            close(s4);
+        if (s6 >= 0)
+            close(s6);
+    });
 }

 void
@@ -312,11 +479,10 @@ DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f
    {
        std::lock_guard<std::mutex> lck(storage_mtx);
        pending_ops.emplace([=](SecureDht& dht) mutable {
-            std::cout << "DhtRunner dht.get " << hash << std::endl;
            dht.get(hash, vcb, dcb, std::move(f), std::move(w));
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -331,7 +497,7 @@ void DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_
            dht.query(hash, cb, done_cb, std::move(q));
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 std::future<size_t>
@@ -344,7 +510,7 @@ DhtRunner::listen(InfoHash hash, GetCallback vcb, Value::Filter f, Where w)
            ret_token->set_value(dht.listen(hash, vcb, std::move(f), std::move(w)));
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
    return ret_token->get_future();
 }

@@ -363,7 +529,7 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
            dht.cancelListen(h, token);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -376,7 +542,7 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> token)
            dht.cancelListen(h, tk);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -389,7 +555,7 @@ DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created
            dht.put(hash, sv, cb, created, permanent);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -401,7 +567,7 @@ DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, tim
            dht.put(hash, value, cb, created, permanent);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -419,7 +585,7 @@ DhtRunner::cancelPut(const InfoHash& h , const Value::Id& id)
            dht.cancelPut(h, id);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -431,7 +597,7 @@ DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback c
            dht.putSigned(hash, value, cb);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -455,7 +621,7 @@ DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value
            dht.putEncrypted(hash, to, value, cb);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -470,7 +636,6 @@ DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, Done
    putEncrypted(InfoHash::get(key), to, std::forward<Value>(value), cb);
 }

-/*
 void
 DhtRunner::tryBootstrapContinuously()
 {
@@ -528,7 +693,7 @@ DhtRunner::tryBootstrapContinuously()
            }
        } while (bootstraping);
    });
-}*/
+}

 std::vector<std::pair<sockaddr_storage, socklen_t>>
 DhtRunner::getAddrInfo(const std::string& host, const std::string& service)
@@ -558,60 +723,45 @@ DhtRunner::getAddrInfo(const std::string& host, const std::string& service)
 void
 DhtRunner::bootstrap(const std::string& host, const std::string& service)
 {
-    /*std::lock_guard<std::mutex> lck(bootstrap_mtx);
+    std::lock_guard<std::mutex> lck(bootstrap_mtx);
    bootstrap_nodes_all.emplace_back(host, service);
    bootstrap_nodes.emplace_back(host, service);
-    tryBootstrapContinuously();*/
-    pending_ops_prio.emplace([=](SecureDht& dht) mutable {
-        dht.bootstrap(host, service);
-    });
-    uv_async_send(&uv_async_);
+    tryBootstrapContinuously();
 }

 void
 DhtRunner::clearBootstrap()
 {
-    //std::lock_guard<std::mutex> lck(bootstrap_mtx);
-    //bootstrap_nodes_all.clear();
-    pending_ops_prio.emplace([=](SecureDht& dht) mutable {
-        dht.clearBootstrap();
-    });
-    uv_async_send(&uv_async_);
+    std::lock_guard<std::mutex> lck(bootstrap_mtx);
+    bootstrap_nodes_all.clear();
 }

 void
-DhtRunner::ping(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes, DoneCallback&& cb)
+DhtRunner::bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes, DoneCallbackSimple&& cb)
 {
    std::lock_guard<std::mutex> lck(storage_mtx);
    pending_ops_prio.emplace([=](SecureDht& dht) mutable {
-        struct GroupPing {
-            size_t remaining {0};
-            bool ok {false};
-            std::vector<Sp<Node>> nodes;
-            GroupPing(size_t s) : remaining(s) {}
-        };
-        Sp<GroupPing> rem = cb ? std::make_shared<GroupPing>(nodes.size()) : nullptr;
+        auto rem = cb ? std::make_shared<std::pair<size_t, bool>>(nodes.size(), false) : nullptr;
        for (auto& node : nodes)
-            dht.pingNode(SockAddr((sockaddr*)&node.first, node.second), cb ? [rem,cb](bool ok, std::vector<Sp<Node>>&& nodes) {
+            dht.pingNode((sockaddr*)&node.first, node.second, cb ? [rem,cb](bool ok) {
                auto& r = *rem;
-                r.remaining--;
-                r.ok |= ok;
-                r.nodes.insert(r.nodes.end(), nodes.begin(), nodes.end());
-                if (r.remaining == 0)
-                    cb(r.ok, std::move(r.nodes));
-            } : DoneCallback{});
+                r.first--;
+                r.second |= ok;
+                if (not r.first)
+                    cb(r.second);
+            } : DoneCallbackSimple{});
    });
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
-DhtRunner::ping(const SockAddr& addr, DoneCallback&& cb)
+DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
 {
    std::lock_guard<std::mutex> lck(storage_mtx);
    pending_ops_prio.emplace([addr,cb](SecureDht& dht) mutable {
-        dht.pingNode(addr, std::move(cb));
+        dht.pingNode(addr.get(), addr.getLength(), std::move(cb));
    });
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -624,7 +774,7 @@ DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
                dht.insertNode(node);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -636,7 +786,7 @@ DhtRunner::connectivityChanged()
            dht.connectivityChanged();
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 void
@@ -647,7 +797,7 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
            dht.findCertificate(hash, cb);
        });
    }
-    uv_async_send(&uv_async_);
+    cv.notify_all();
 }

 }
@@ -0,0 +1,37 @@
+/*
+ *  Copyright (C) 2017 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
+
+namespace dht {
+namespace net {
+
+enum class MessageType {
+    Error = 0,
+    Reply,
+    Ping,
+    FindNode,
+    GetValues,
+    AnnounceValue,
+    Refresh,
+    Listen,
+    ValueData
+};
+
+} /* namespace net */
+} /* dht */
@@ -31,11 +31,11 @@ constexpr std::chrono::minutes Node::NODE_GOOD_TIME;
 constexpr std::chrono::seconds Node::MAX_RESPONSE_TIME;


-Node::Node(const InfoHash& id, const SockAddr& addr, const Sp<TcpSocket>& s, bool client)
-: id(id), addr(addr), sock(s), is_client(client), sockets_()
+Node::Node(const InfoHash& id, const SockAddr& addr, bool client)
+: id(id), addr(addr), is_client(client), sockets_()
 {
    crypto::random_device rd;
-    transaction_id = std::uniform_int_distribution<SocketId>{1}(rd);
+    transaction_id = std::uniform_int_distribution<Tid>{1}(rd);
 }

 /* This is our definition of a known-good node. */
@@ -69,11 +69,9 @@ Node::getPendingMessageCount() const
 }

 void
-Node::update(const SockAddr& new_addr, const Sp<TcpSocket>& s)
+Node::update(const SockAddr& new_addr)
 {
    addr = new_addr;
-    if (not sock or (s and sock->isClosed() and not s->isClosed()))
-        sock = s;
 }

 /** To be called when a message was sent to the node */
@@ -103,7 +101,7 @@ Node::received(time_point now, const Sp<net::Request>& req)
 }

 Sp<net::Request>
-Node::getRequest(const net::TransId& tid)
+Node::getRequest(Tid tid)
 {
    auto it = requests_.find(tid);
    return it != requests_.end() ? it->second : nullptr;
@@ -130,7 +128,7 @@ Node::setExpired()
    sockets_.clear();
 }

-SocketId
+Tid
 Node::openSocket(SocketCb&& cb)
 {
    if (++transaction_id == 0)
@@ -144,14 +142,14 @@ Node::openSocket(SocketCb&& cb)
 }

 Socket*
-Node::getSocket(SocketId id)
+Node::getSocket(Tid id)
 {
    auto it = sockets_.find(id);
    return it == sockets_.end() ? nullptr : &it->second;
 }

 void
-Node::closeSocket(SocketId id)
+Node::closeSocket(Tid id)
 {
    if (id) {
        sockets_.erase(id);
@@ -26,10 +26,10 @@ NodeCache::getNode(const InfoHash& id, sa_family_t family) {
 }

 Sp<Node>
-NodeCache::getNode(const InfoHash& id, const SockAddr& addr, const Sp<TcpSocket>& sock, time_point now, bool confirm, bool client) {
+NodeCache::getNode(const InfoHash& id, const SockAddr& addr, time_point now, bool confirm, bool client) {
    if (id == zeroes)
        return std::make_shared<Node>(id, addr);
-    return cache(addr.getFamily()).getNode(id, addr, sock, now, confirm, client);
+    return cache(addr.getFamily()).getNode(id, addr, now, confirm, client);
 }

 std::vector<Sp<Node>>
@@ -78,17 +78,6 @@ NodeCache::clearBadNodes(sa_family_t family)
    }
 }

-void
-NodeCache::closeAll(sa_family_t family)
-{
-    if (family == 0) {
-        closeAll(AF_INET);
-        closeAll(AF_INET6);
-    } else {
-        (family == AF_INET ? cache_4 : cache_6).closeAll();
-    }
-}
-
 Sp<Node>
 NodeCache::NodeMap::getNode(const InfoHash& id)
 {
@@ -102,15 +91,15 @@ NodeCache::NodeMap::getNode(const InfoHash& id)
 }

 Sp<Node>
-NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, const Sp<TcpSocket>& sock, time_point now, bool confirm, bool client)
+NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, time_point now, bool confirm, bool client)
 {
    auto it = emplace(id, std::weak_ptr<Node>{});
    auto node = it.first->second.lock();
    if (not node) {
-        node = std::make_shared<Node>(id, addr, sock, client);
+        node = std::make_shared<Node>(id, addr, client);
        it.first->second = node;
    } else if (confirm or node->isOld(now)) {
-        node->update(addr, sock);
+        node->update(addr);
    }
    return node;
 }
@@ -127,21 +116,4 @@ NodeCache::NodeMap::clearBadNodes() {
    }
 }

-void
-NodeCache::NodeMap::closeAll()
-{
-    for (auto it = cbegin(); it != cend();) {
-        if (auto n = it->second.lock()) {
-            if (n->sock) {
-                n->sock->close();
-                n->sock.reset();
-            }
-            ++it;
-        } else {
-            erase(it++);
-        }
-    }
-}
-
-
 }
@@ -26,17 +26,14 @@
 namespace dht {
 namespace net {

-enum class MessageType {
-    Error = 0,
-    Reply,
-    Ping,
-    FindNode,
-    GetValues,
-    AnnounceValue,
-    Refresh,
-    Listen,
-    ValueData
-};
+Tid unpackTid(const msgpack::object& o) {
+    switch (o.type) {
+    case msgpack::type::POSITIVE_INTEGER:
+        return o.as<Tid>();
+    default:
+        return ntohl(*reinterpret_cast<const uint32_t*>(o.as<std::array<char, 4>>().data()));
+    }
+}

 struct ParsedMessage {
    MessageType type;
@@ -51,9 +48,9 @@ struct ParsedMessage {
    /* target id around which to find nodes */
    InfoHash target;
    /* transaction id */
-    TransId tid;
+    Tid tid;
    /* tid for packets going through request socket */
-    TransId socket_id;
+    Tid socket_id;
    /* security token */
    Blob token;
    /* the value id (announce confirmation) */
@@ -59,18 +59,19 @@ struct Request {
    State getState() const { return state_; }

    Request(State state = State::PENDING) : state_(state) {}
-    Request(const TransId& tid,
+    Request(MessageType type, Tid tid,
            Sp<Node> node,
            Blob&& msg,
            std::function<void(const Request&, ParsedMessage&&)> on_done,
            std::function<void(const Request&, bool)> on_expired,
-            uint32_t socket = 0) :
-        node(node), on_done(on_done), on_expired(on_expired), tid(tid), msg(std::move(msg)), socket(socket) { }
+            Tid socket = 0) :
+        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) { }

-    TransId getTid() const { return tid; }
+    Tid getTid() const { return tid; }
+    MessageType getType() const { return type; }

-    uint32_t getSocket() { return socket; }
-    uint32_t closeSocket() { auto ret = socket; socket = 0; return ret; }
+    Tid getSocket() { return socket; }
+    Tid closeSocket() { auto ret = socket; socket = 0; return ret; }

    void setExpired() {
        if (pending()) {
@@ -107,6 +108,8 @@ private:
        msg.clear();
    }

+    const Tid tid {0}; /* the request id. */
+    const MessageType type {};
    State state_ {State::PENDING};

    unsigned attempt_count {0};                /* number of attempt to process the request. */
@@ -116,9 +119,8 @@ private:
    std::function<void(const Request&, ParsedMessage&&)> on_done {};
    std::function<void(const Request&, bool)> on_expired {};

-    const TransId tid; /* the request id. */
    Blob msg {};                      /* the serialized message. */
-    uint32_t socket;   /* the socket used for further reponses. */
+    Tid socket;   /* the socket used for further reponses. */
 };

 } /* namespace net  */
@@ -320,7 +320,7 @@ struct Dht::Search {
    uint16_t tid;
    time_point refill_time {time_point::min()};
    time_point step_time {time_point::min()};           /* the time of the last search step */
-    Sp<Job> nextSearchStep {};
+    Sp<Scheduler::Job> nextSearchStep {};

    bool expired {false};              /* no node, or all nodes expired */
    bool done {false};                 /* search is over, cached for later */
@@ -40,10 +40,10 @@ Config& getConfig(SecureDht::Config& conf)
    return c;
 }

-SecureDht::SecureDht(uv_loop_t* loop, SecureDht::Config conf)
-: Dht(loop, getConfig(conf)), key_(conf.id.first), certificate_(conf.id.second)
+SecureDht::SecureDht(int s, int s6, SecureDht::Config conf)
+: Dht(s, s6, getConfig(conf)), key_(conf.id.first), certificate_(conf.id.second)
 {
-    if (not loop)
+    if (s < 0 && s6 < 0)
        return;

    for (const auto& type : DEFAULT_TYPES)
@@ -19,16 +19,12 @@
 #include "utils.h"
 #include "sockaddr.h"
 #include "default_types.h"
-#include "net.h"
-#include "uv_utils.h"

 /* An IPv4 equivalent to IN6_IS_ADDR_UNSPECIFIED */
 #ifndef IN_IS_ADDR_UNSPECIFIED
 #define IN_IS_ADDR_UNSPECIFIED(a) (((long int) (a)->s_addr) == 0x00000000)
 #endif /* IN_IS_ADDR_UNSPECIFIED */

-static constexpr std::array<uint8_t, 12> MAPPED_IPV4_PREFIX {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}};
-
 namespace dht {

 std::string
@@ -115,57 +111,6 @@ SockAddr::isPrivate() const
    }
 }

-bool
-SockAddr::isMappedIPv4() const
-{
-    if (getFamily() != AF_INET6)
-        return false;
-    const uint8_t* addr6 = reinterpret_cast<const uint8_t*>(&getIPv6().sin6_addr);
-    return std::equal(MAPPED_IPV4_PREFIX.begin(), MAPPED_IPV4_PREFIX.end(), addr6);
-}
-
-SockAddr
-SockAddr::getMappedIPv4() const
-{
-    if (getFamily() != AF_INET6)
-        return *this;
-    SockAddr ret;
-    ret.setFamily(AF_INET);
-    ret.setPort(getPort());
-    auto addr6 = reinterpret_cast<const uint8_t*>(&getIPv6().sin6_addr);
-    auto addr4 = reinterpret_cast<uint8_t*>(&ret.getIPv4().sin_addr);
-    addr6 += MAPPED_IPV4_PREFIX.size();
-    std::copy_n(addr6, sizeof(in_addr), addr4);
-    return ret;
-}
-
-void get_addr_info_cb(uv_getaddrinfo_t* req, int status, struct addrinfo* res) {
-    if (auto cb = static_cast<GetAddrInfoCb*>(req->data)) {
-        size_t count {0};
-        for (auto addr = res; addr; addr = addr->ai_next)
-            count++;
-        std::vector<SockAddr> ret;
-        ret.reserve(count);
-        for (auto addr = res; addr; addr = addr->ai_next)
-            ret.emplace_back(addr->ai_addr, addr->ai_addrlen);
-        (*cb)(std::move(ret));
-        delete cb;
-    }
-    delete req;
-    uv_freeaddrinfo(res);
-}
-
-void getAddrInfo(uv_loop_t* loop, const char* node, const char* service, GetAddrInfoCb&& cb)
-{
-    auto req = new uv_getaddrinfo_t;
-    req->data = new GetAddrInfoCb(std::move(cb));
-    addrinfo hints {};
-    hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
-    hints.ai_family = AF_UNSPEC;
-    hints.ai_socktype = SOCK_DGRAM;
-    uv_getaddrinfo(loop, req, get_addr_info_cb, node, service, &hints);
-}
-
 bool operator==(const SockAddr& a, const SockAddr& b) {
    return a.equals(b);
 }
@@ -215,18 +160,4 @@ findMapValue(msgpack::object& map, const std::string& key) {
    return nullptr;
 }

-namespace net {
-
-TransId
-unpackTid(msgpack::object& o) {
-    switch (o.type) {
-    case msgpack::type::POSITIVE_INTEGER:
-        return o.as<uint32_t>();
-    default:
-        return o.as<std::array<char, 4>>();
-    }
-}
-
-}
-
 }
@@ -39,7 +39,7 @@ void print_id_req() {
 }

 void print_node_info(const std::shared_ptr<DhtRunner>& dht, const dht_params& params) {
-    std::cout << "OpenDht node " << dht->getNodeId() << " running on port " <<  dht->getBound().toString() << std::endl;
+    std::cout << "OpenDht node " << dht->getNodeId() << " running on port " <<  dht->getBoundPort() << std::endl;
    if (params.generate_identity)
        std::cout << "Public key ID " << dht->getId() << std::endl;
 }
@@ -370,9 +370,6 @@ main(int argc, char **argv)
        }

        dht->run(params.port, crt, true, params.network);
-        dht->setOnStatusChanged([](dht::NodeStatus status4, dht::NodeStatus status6){
-            std::cout << "Node status changed : IPv4:" << (unsigned)status4 << " IPv6:" << (unsigned)status6 << std::endl;
-        });

        if (params.log) {
            if (params.syslog or (params.daemonize and params.logfile.empty()))
Author	SHA1	Message	Date
Adrien Béraud	f4ebbc6651	update to version 1.4.1 * updated MSVC project * bug fixes	2017-10-21 01:30:12 +02:00
Adrien Béraud	464339822b	Merge pull request #215 from AmarOk1412/typoPeriod rate_limiter: typo fix, 'Perdiod'->'Period'	2017-10-20 23:56:02 +02:00
Sébastien Blin	fe09146609	rate_limiter: typo fix, 'Perdiod'->'Period'	2017-10-20 14:51:35 -04:00
Adrien Béraud	a789ffe7ae	dht: prevent use-after-free of socket callback	2017-10-20 17:46:27 +02:00
Adrien Béraud	476fbb1160	network: use full 32-bits of transaction IDs We used to use the first 16 bits for the message type.	2017-10-20 17:42:28 +02:00
Adrien Béraud	4501217c21	Merge pull request #214 from atraczyk/master remove vs2015 based constexpr error, update vs project files	2017-10-19 01:19:43 +02:00
atraczyk	207675a181	remove vs2015 based constexpr error, update vs project files	2017-10-18 17:50:19 -04:00
Adrien Béraud	cf6447dc80	NetworkEngine: use proper type for constant	2017-10-17 19:22:15 +02:00