net: add logger support for UdpSocket

CMakeLists.txt

@@ -151,7 +151,6 @@ list (APPEND opendht_SOURCES
     src/dhtrunner.cpp
     src/log.cpp
     src/peer_discovery.cpp
-    src/network_utils.h
     src/network_utils.cpp
     src/thread_pool.cpp
 )
@@ -179,6 +178,7 @@ list (APPEND opendht_HEADERS
     include/opendht/log_enable.h
     include/opendht/peer_discovery.h
     include/opendht/thread_pool.h
+    include/opendht/network_utils.h
     include/opendht.h
 )
 

include/opendht/dht.h

@@ -68,7 +68,7 @@ public:
      * Initialise the Dht with two open sockets (for IPv4 and IP6)
      * and an ID for the node.
      */
-    Dht(const int& s, const int& s6, const Config& config, const Logger& l = {});
+    Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, const Logger& l = {});
     virtual ~Dht();
 
     /**
@@ -85,6 +85,8 @@ public:
         return std::max(getStatus(AF_INET), getStatus(AF_INET6));
     }
 
+    net::DatagramSocket* getSocket() const override { return network_engine.getSocket(); };
+
     /**
      * Performs final operations before quitting.
      */

include/opendht/dht_interface.h

@@ -23,6 +23,10 @@
 
 namespace dht {
 
+namespace net {
+    class DatagramSocket;
+}
+
 class OPENDHT_PUBLIC DhtInterface {
 public:
     DhtInterface() = default;
@@ -40,6 +44,8 @@ public:
     virtual NodeStatus getStatus(sa_family_t af) const = 0;
     virtual NodeStatus getStatus() const = 0;
 
+    virtual net::DatagramSocket* getSocket() const { return {}; };
+
     /**
      * Get the ID of the DHT node.
      */

include/opendht/dhtrunner.h

@@ -20,12 +20,13 @@
 
 #pragma once
 
+#include "def.h"
 #include "infohash.h"
 #include "value.h"
 #include "callbacks.h"
 #include "sockaddr.h"
 #include "log_enable.h"
-#include "def.h"
+#include "network_utils.h"
 
 #include <thread>
 #include <mutex>
@@ -65,6 +66,7 @@ public:
 
     struct Context {
         std::unique_ptr<Logger> logger {};
+        std::unique_ptr<net::DatagramSocket> sock;
         std::shared_ptr<PeerDiscovery> peerDiscovery {};
         StatusCallback statusChangedCallback {};
         CertificateStoreQuery certificateStore {};
@@ -304,9 +306,7 @@ public:
      * Returns the currently bound address.
      * @param f: address family of the bound address to retreive.
      */
-    const SockAddr& getBound(sa_family_t f = AF_INET) const {
-        return (f == AF_INET) ? bound4 : bound6;
-    }
+    SockAddr getBound(sa_family_t f = AF_INET) const;
 
     /**
      * Returns the currently bound port, in host byte order.
@@ -401,6 +401,8 @@ public:
      */
     void run(const char* ip4, const char* ip6, const char* service, const Config& config, Context&& context = {});
 
+    void run(const Config& config, Context&& context);
+
     void setOnStatusChanged(StatusCallback&& cb) {
         statusCb = std::move(cb);
     }
@@ -412,13 +414,7 @@ public:
      */
     time_point loop() {
         std::lock_guard<std::mutex> lck(dht_mtx);
-        time_point wakeup = time_point::min();
-        try {
-            wakeup = loop_();
-        } catch (const dht::SocketException& e) {
-            startNetwork(bound4, bound6);
-        }
-        return wakeup;
+        return loop_();
     }
 
     /**
@@ -470,8 +466,6 @@ private:
      */
     void tryBootstrapContinuously();
 
-    void stopNetwork();
-    void startNetwork(const SockAddr sin4, const SockAddr sin6);
     time_point loop_();
 
     NodeStatus getStatus() const {
@@ -509,16 +503,8 @@ private:
     mutable std::mutex dht_mtx {};
     std::thread dht_thread {};
     std::condition_variable cv {};
-
-    std::thread rcv_thread {};
     std::mutex sock_mtx {};
-
-    struct ReceivedPacket {
-        Blob data;
-        SockAddr from;
-        time_point received;
-    };
-    std::queue<ReceivedPacket> rcv {};
+    std::queue<std::unique_ptr<net::ReceivedPacket>> rcv {};
 
     /** true if currently actively boostraping */
     std::atomic_bool bootstraping {false};
@@ -526,7 +512,6 @@ private:
     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 {};
-    /** protects bootstrap_nodes, bootstrap_thread */
     std::mutex bootstrap_mtx {};
     std::condition_variable bootstrap_cv {};
 
@@ -535,17 +520,11 @@ private:
     std::mutex storage_mtx {};
 
     std::atomic_bool running {false};
-    std::atomic_bool running_network {false};
 
     NodeStatus status4 {NodeStatus::Disconnected},
                status6 {NodeStatus::Disconnected};
     StatusCallback statusCb {nullptr};
 
-    int stop_writefd {-1};
-    int s4 {-1}, s6 {-1};
-    SockAddr bound4 {};
-    SockAddr bound6 {};
-
     /** Push notification token */
     std::string pushToken_;
 

include/opendht/network_engine.h

@@ -28,6 +28,7 @@
 #include "rng.h"
 #include "rate_limiter.h"
 #include "log_enable.h"
+#include "network_utils.h"
 
 #include <vector>
 #include <string>
@@ -205,8 +206,8 @@ public:
     using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
     using RequestExpiredCb = std::function<void(const Request&, bool)>;
 
-    NetworkEngine(Logger& log, Scheduler& scheduler, const int& s = -1, const int& s6 = -1);
-    NetworkEngine(InfoHash& myid, NetId net, const int& s, const int& s6, Logger& log, Scheduler& scheduler,
+    NetworkEngine(Logger& log, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock);
+    NetworkEngine(InfoHash& myid, NetId net, std::unique_ptr<DatagramSocket>&& sock, Logger& log, Scheduler& scheduler,
             decltype(NetworkEngine::onError)&& onError,
             decltype(NetworkEngine::onNewNode)&& onNewNode,
             decltype(NetworkEngine::onReportedAddr)&& onReportedAddr,
@@ -217,7 +218,9 @@ public:
             decltype(NetworkEngine::onAnnounce)&& onAnnounce,
             decltype(NetworkEngine::onRefresh)&& onRefresh);
 
-    virtual ~NetworkEngine();
+    ~NetworkEngine();
+
+    net::DatagramSocket* getSocket() const { return dht_socket.get(); };
 
     void clear();
 
@@ -242,7 +245,7 @@ public:
     void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
 
     bool isRunning(sa_family_t af) const;
-    inline want_t want () const { return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1; }
+    inline want_t want () const { return dht_socket->hasIPv4() and dht_socket->hasIPv6() ? (WANT4 | WANT6) : -1; }
 
     void connectivityChanged(sa_family_t);
 
@@ -469,7 +472,7 @@ private:
 
 
     // basic wrapper for socket sendto function
-    int send(const char *buf, size_t len, int flags, const SockAddr& addr);
+    int send(const SockAddr& addr, const char *buf, size_t len, bool confirmed = false);
 
     void sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr);
     std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
@@ -511,8 +514,7 @@ private:
     /* DHT info */
     const InfoHash& myid;
     const NetId network {0};
-    const int& dht_socket;
-    const int& dht_socket6;
+    const std::unique_ptr<DatagramSocket> dht_socket;
     const Logger& DHT_LOG;
 
     NodeCache cache {};

include/opendht/network_utils.h

@@ -0,0 +1,115 @@
+/*
+ *  Copyright (C) 2019 Savoir-faire Linux Inc.
+ *  Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+#pragma once
+
+#include "def.h"
+
+#include "sockaddr.h"
+#include "utils.h"
+#include "log_enable.h"
+
+#ifdef _WIN32
+#include <ws2tcpip.h>
+#include <winsock2.h>
+#else
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <unistd.h>
+#endif
+
+#include <functional>
+#include <thread>
+#include <atomic>
+#include <iostream>
+
+namespace dht {
+namespace net {
+
+int bindSocket(const SockAddr& addr, SockAddr& bound);
+
+bool setNonblocking(int fd, bool nonblocking = true);
+
+#ifdef _WIN32
+void udpPipe(int fds[2]);
+#endif
+struct ReceivedPacket {
+    Blob data;
+    SockAddr from;
+    time_point received;
+};
+
+class OPENDHT_PUBLIC DatagramSocket {
+public:
+    using OnReceive = std::function<void(std::unique_ptr<ReceivedPacket>&& packet)>;
+    virtual ~DatagramSocket() {};
+
+    virtual int sendTo(const SockAddr& dest, const uint8_t* data, size_t size, bool replied) = 0;
+
+    inline void setOnReceive(OnReceive&& cb) {
+        rx_callback = std::move(cb);
+    }
+
+    virtual const SockAddr& getBound(sa_family_t family = AF_UNSPEC) const = 0;
+    virtual bool hasIPv4() const = 0;
+    virtual bool hasIPv6() const = 0;
+
+    in_port_t getPort(sa_family_t family = AF_UNSPEC) const {
+        return getBound(family).getPort();
+    }
+
+    virtual void stop() = 0;
+protected:
+
+    inline void onReceived(std::unique_ptr<ReceivedPacket>&& packet) {
+        if (rx_callback)
+            rx_callback(std::move(packet));
+    }
+private:
+    OnReceive rx_callback;
+};
+
+class OPENDHT_PUBLIC UdpSocket : public DatagramSocket {
+public:
+    UdpSocket(in_port_t port, const Logger& l = {});
+    UdpSocket(const SockAddr& bind4, const SockAddr& bind6, const Logger& l = {});
+    ~UdpSocket();
+
+    int sendTo(const SockAddr& dest, const uint8_t* data, size_t size, bool replied) override;
+
+    const SockAddr& getBound(sa_family_t family = AF_UNSPEC) const override {
+        return (family == AF_INET6) ? bound6 : bound4;
+    }
+
+    bool hasIPv4() const override { return s4 != -1; }
+    bool hasIPv6() const override { return s6 != -1; }
+
+    void stop() override;
+private:
+    Logger logger;
+    int s4 {-1};
+    int s6 {-1};
+    int stopfd {-1};
+    SockAddr bound4, bound6;
+    std::thread rcv_thread {};
+    std::atomic_bool running {false};
+
+    void openSockets(const SockAddr& bind4, const SockAddr& bind6);
+};
+
+}
+}

include/opendht/securedht.h

@@ -207,6 +207,9 @@ public:
     NodeStatus getStatus() const override {
         return dht_->getStatus();
     }
+    net::DatagramSocket* getSocket() const override {
+        return dht_->getSocket();
+    };
     bool isRunning(sa_family_t af = 0) const override {
         return dht_->isRunning(af);
     }

include/opendht/sockaddr.h

@@ -119,7 +119,7 @@ public:
     /**
      * Returns the address family or AF_UNSPEC if the address is not set.
      */
-    sa_family_t getFamily() const { return len > sizeof(sa_family_t) ? addr->sa_family : AF_UNSPEC; }
+    sa_family_t getFamily() const { return len ? addr->sa_family : AF_UNSPEC; }
 
     /**
      * Resize the managed structure to the appropriate size (if needed),

src/dht.cpp

@@ -1689,11 +1689,11 @@ Dht::~Dht()
         s.second->clear();
 }
 
-Dht::Dht() : store(), network_engine(DHT_LOG, scheduler) {}
+Dht::Dht() : store(), network_engine(DHT_LOG, scheduler, {}) {}
 
-Dht::Dht(const int& s, const int& s6, const Config& config, const Logger& l)
-    : DhtInterface(l), myid(config.node_id ? config.node_id : InfoHash::getRandom()), store(), store_quota(),
-    network_engine(myid, config.network, s, s6, DHT_LOG, scheduler,
+Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, const Logger& l)
+    : myid(config.node_id ? config.node_id : InfoHash::getRandom()), store(), store_quota(),
+    network_engine(myid, config.network, std::move(sock), DHT_LOG, scheduler,
             std::bind(&Dht::onError, this, _1, _2),
             std::bind(&Dht::onNewNode, this, _1, _2),
             std::bind(&Dht::onReportedAddr, this, _1, _2),
@@ -1708,13 +1708,14 @@ Dht::Dht(const int& s, const int& s6, const Config& config, const Logger& l)
     maintain_storage(config.maintain_storage)
 {
     scheduler.syncTime();
-    if (s < 0 && s6 < 0)
-        return;
-    if (s >= 0) {
+    auto s = network_engine.getSocket();
+    if (not s or (not s->hasIPv4() and not s->hasIPv6()))
+        throw DhtException("Opened socket required");
+    if (s->hasIPv4()) {
         buckets4 = {Bucket {AF_INET}};
         buckets4.is_client = config.is_bootstrap;
     }
-    if (s6 >= 0) {
+    if (s->hasIPv6()) {
         buckets6 = {Bucket {AF_INET6}};
         buckets6.is_client = config.is_bootstrap;
     }

src/dhtrunner.cpp

@@ -77,7 +77,7 @@ DhtRunner::~DhtRunner()
 }
 
 void
-DhtRunner::run(in_port_t port, const DhtRunner::Config& config, Context&& context)
+DhtRunner::run(in_port_t port, const Config& config, Context&& context)
 {
     SockAddr sin4;
     sin4.setFamily(AF_INET);
@@ -89,7 +89,7 @@ DhtRunner::run(in_port_t port, const DhtRunner::Config& config, Context&& contex
 }
 
 void
-DhtRunner::run(const char* ip4, const char* ip6, const char* service, const DhtRunner::Config& config, Context&& context)
+DhtRunner::run(const char* ip4, const char* ip6, const char* service, const Config& config, Context&& context)
 {
     auto res4 = SockAddr::resolve(ip4, service);
     auto res6 = SockAddr::resolve(ip6, service);
@@ -98,13 +98,34 @@ DhtRunner::run(const char* ip4, const char* ip6, const char* service, const DhtR
 }
 
 void
-DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const DhtRunner::Config& config, Context&& context)
+DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const Config& config, Context&& context)
+{
+    if (not running) {
+        if (not context.sock)
+            context.sock.reset(new net::UdpSocket(local4, local6, context.logger ? *context.logger : Logger{}));
+        run(config, std::move(context));
+    }
+}
+
+void
+DhtRunner::run(const Config& config, Context&& context)
 {
     if (running)
         return;
-    startNetwork(local4, local6);
 
-    auto dht = std::unique_ptr<DhtInterface>(new Dht(s4, s6, SecureDht::getConfig(config.dht_config), context.logger ? *context.logger : Logger{}));
+    context.sock->setOnReceive([&] (std::unique_ptr<net::ReceivedPacket>&& pkt) {
+        {
+            std::lock_guard<std::mutex> lck(sock_mtx);
+            if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
+                std::cerr << "Dropping packet: queue is full!" << std::endl;
+                rcv.pop();
+            }
+            rcv.emplace(std::move(pkt));
+        }
+        cv.notify_all();
+    });
+
+    auto dht = std::unique_ptr<DhtInterface>(new Dht(std::move(context.sock), SecureDht::getConfig(config.dht_config)));
     dht_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht), config.dht_config));
 
 #ifdef OPENDHT_PROXY_CLIENT
@@ -126,15 +147,10 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const DhtRunner::
     running = true;
     if (not config.threaded)
         return;
-    dht_thread = std::thread([this, local4, local6]() {
+    dht_thread = std::thread([this]() {
         while (running) {
             std::unique_lock<std::mutex> lk(dht_mtx);
-            time_point wakeup;
-            try {
-                wakeup = loop_();
-            } catch (const dht::SocketException& e) {
-                startNetwork(local4, local6);
-            }
+            time_point wakeup = loop_();
 
             auto hasJobToDo = [this]() {
                 if (not running)
@@ -205,7 +221,8 @@ DhtRunner::shutdown(ShutdownCallback cb) {
 #endif
     std::lock_guard<std::mutex> lck(storage_mtx);
     pending_ops_prio.emplace([=](SecureDht& dht) mutable {
-        dht.shutdown(cb);
+        if (dht_)
+            dht_->shutdown(cb);
     });
     cv.notify_all();
 }
@@ -213,18 +230,21 @@ DhtRunner::shutdown(ShutdownCallback cb) {
 void
 DhtRunner::join()
 {
-    stopNetwork();
     running = false;
     cv.notify_all();
     bootstrap_cv.notify_all();
     if (peerDiscovery_) peerDiscovery_->stop();
+    {
+        std::lock_guard<std::mutex> lck(dht_mtx);
+        if (dht_)
+            if (auto sock = dht_->getSocket())
+                sock->stop();
+    }
 
     if (dht_thread.joinable())
         dht_thread.join();
     if (bootstrap_thread.joinable())
         bootstrap_thread.join();
-    if (rcv_thread.joinable())
-        rcv_thread.join();
 
     if (peerDiscovery_) {
         peerDiscovery_->join();
@@ -243,6 +263,15 @@ DhtRunner::join()
     }
 }
 
+SockAddr
+DhtRunner::getBound(sa_family_t af) const {
+    std::lock_guard<std::mutex> lck(dht_mtx);
+    if (dht_)
+        if (auto sock = dht_->getSocket())
+            return sock->getBound(af);
+    return SockAddr{};
+}
+
 void
 DhtRunner::dumpTables() const
 {
@@ -475,7 +504,7 @@ DhtRunner::loop_()
     size_t dropped {0};
     if (not received.empty()) {
         auto now = clock::now();
-        while (not received.empty() and now - received.front().received > RX_QUEUE_MAX_DELAY) {
+        while (not received.empty() and now - received.front()->received > RX_QUEUE_MAX_DELAY) {
             received.pop();
             dropped++;
         }
@@ -485,10 +514,10 @@ DhtRunner::loop_()
     if (not received.empty()) {
         while (not received.empty()) {
             auto& pck = received.front();
-            if (clock::now() - pck.received > RX_QUEUE_MAX_DELAY)
+            if (clock::now() - pck->received > RX_QUEUE_MAX_DELAY)
                 dropped++;
             else
-                wakeup = dht->periodic(pck.data.data(), pck.data.size(), pck.from);
+                wakeup = dht->periodic(pck->data.data(), pck->data.size(), pck->from);
             received.pop();
         }
     } else {
@@ -497,7 +526,7 @@ DhtRunner::loop_()
     }
 
     if (dropped)
-        std::cerr << "Dropped %zu packets with high delay" << dropped << std::endl;
+        std::cerr << "Dropped " << dropped << " packets with high delay" << std::endl;
 
     NodeStatus nstatus4 = dht->getStatus(AF_INET);
     NodeStatus nstatus6 = dht->getStatus(AF_INET6);
@@ -520,167 +549,6 @@ DhtRunner::loop_()
     return wakeup;
 }
 
-
-int bindSocket(const SockAddr& addr, SockAddr& bound)
-{
-    bool is_ipv6 = addr.getFamily() == AF_INET6;
-    int sock = socket(is_ipv6 ? PF_INET6 : PF_INET, SOCK_DGRAM, 0);
-    if (sock < 0)
-        throw DhtException(std::string("Can't open socket: ") + strerror(sock));
-    int set = 1;
-#ifdef SO_NOSIGPIPE
-    setsockopt(sock, SOL_SOCKET, SO_NOSIGPIPE, (const char*)&set, sizeof(set));
-#endif
-    if (is_ipv6)
-        setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&set, sizeof(set));
-    net::set_nonblocking(sock);
-    int rc = bind(sock, addr.get(), addr.getLength());
-    if(rc < 0) {
-        rc = errno;
-        close(sock);
-        throw DhtException("Can't bind socket on " + addr.toString() + " " + strerror(rc));
-    }
-    sockaddr_storage ss;
-    socklen_t ss_len = sizeof(ss);
-    getsockname(sock, (sockaddr*)&ss, &ss_len);
-    bound = {ss, ss_len};
-    return sock;
-}
-
-void
-DhtRunner::stopNetwork()
-{
-    running_network = false;
-    if (stop_writefd != -1) {
-        if (write(stop_writefd, "\0", 1) == -1) {
-            perror("write");
-        }
-    }
-}
-
-void
-DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
-{
-    stopNetwork();
-    if (rcv_thread.joinable())
-        rcv_thread.join();
-
-    int stopfds[2];
-#ifndef _WIN32
-    auto status = pipe(stopfds);
-    if (status == -1) {
-        throw DhtException(std::string("Can't open pipe: ") + strerror(errno));
-    }
-#else
-    net::udpPipe(stopfds);
-#endif
-    int stop_readfd = stopfds[0];
-    stop_writefd = stopfds[1];
-
-    s4 = -1;
-    s6 = -1;
-
-    bound4 = {};
-    if (sin4) {
-        try {
-            s4 = bindSocket(sin4, bound4);
-        } catch (const DhtException& e) {
-            std::cerr << "Can't bind inet socket: " << e.what() << std::endl;
-        }
-    }
-
-#if 1
-    bound6 = {};
-    if (sin6) {
-        try {
-            s6 = bindSocket(sin6, bound6);
-        } catch (const DhtException& e) {
-            std::cerr << "Can't bind inet6 socket: " << e.what() << std::endl;
-        }
-    }
-#endif
-
-    if (s4 == -1 && s6 == -1) {
-        throw DhtException("Can't bind socket");
-    }
-
-    running_network = true;
-    rcv_thread = std::thread([this, stop_readfd]() {
-        try {
-            while (running_network) {
-                fd_set readfds;
-
-                FD_ZERO(&readfds);
-                FD_SET(stop_readfd, &readfds);
-                if(s4 >= 0)
-                    FD_SET(s4, &readfds);
-                if(s6 >= 0)
-                    FD_SET(s6, &readfds);
-
-                int selectFd = std::max({s4, s6, stop_readfd}) + 1;
-                int rc = select(selectFd, &readfds, nullptr, nullptr, nullptr);
-                if(rc < 0) {
-                    if(errno != EINTR) {
-                        perror("select");
-                        std::this_thread::sleep_for( std::chrono::seconds(1) );
-                    }
-                }
-
-                if (not running_network)
-                    break;
-
-                if (rc > 0) {
-                    std::array<uint8_t, 1024 * 64> buf;
-                    sockaddr_storage from;
-                    socklen_t from_len = sizeof(from);
-
-                    if (FD_ISSET(stop_readfd, &readfds)) {
-                        if (recv(stop_readfd, (char*)buf.data(), buf.size(), 0) < 0) {
-                            std::cerr << "Got stop packet error: " << strerror(errno) << std::endl;
-                            break;
-                        }
-                    }
-                    else 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
-                        continue;
-
-                    if (rc > 0) {
-                        {
-                            std::lock_guard<std::mutex> lck(sock_mtx);
-                            if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
-                                std::cerr << "Dropping packet: queue is full!" << std::endl;
-                                rcv.pop();
-                            }
-                            rcv.emplace(ReceivedPacket {Blob {buf.begin(), buf.begin()+rc}, SockAddr(from, from_len), clock::now()});
-                        }
-                        cv.notify_all();
-                    } else if (rc == -1) {
-                        std::cerr << "Error receiving packet: " << strerror(errno) << std::endl;
-                    }
-                }
-            }
-        } 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);
-        s4 = -1;
-        s6 = -1;
-        bound4 = {};
-        bound6 = {};
-        if (stop_readfd != -1)
-            close(stop_readfd);
-        if (stop_writefd != -1)
-            close(stop_writefd);
-        stop_writefd = -1;
-    });
-}
-
 void
 DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f, Where w)
 {

src/network_engine.cpp

@@ -99,11 +99,11 @@ RequestAnswer::RequestAnswer(ParsedMessage&& msg)
    nodes6(std::move(msg.nodes6))
 {}
 
-NetworkEngine::NetworkEngine(Logger& log, Scheduler& scheduler, const int& s, const int& s6)
-    : myid(zeroes), dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
+NetworkEngine::NetworkEngine(Logger& log, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock)
+    : myid(zeroes), dht_socket(std::move(sock)), DHT_LOG(log), scheduler(scheduler)
 {}
 
-NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, const int& s, const int& s6, Logger& log, Scheduler& scheduler,
+NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, std::unique_ptr<DatagramSocket>&& sock, Logger& log, Scheduler& scheduler,
         decltype(NetworkEngine::onError)&& onError,
         decltype(NetworkEngine::onNewNode)&& onNewNode,
         decltype(NetworkEngine::onReportedAddr)&& onReportedAddr,
@@ -122,7 +122,7 @@ NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, const int& s, const int&
     onListen(std::move(onListen)),
     onAnnounce(std::move(onAnnounce)),
     onRefresh(std::move(onRefresh)),
-    myid(myid), network(net), dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
+    myid(myid), network(net), dht_socket(std::move(sock)), DHT_LOG(log), scheduler(scheduler)
 {}
 
 NetworkEngine::~NetworkEngine() {
@@ -170,7 +170,7 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&,
     }
 
     // send response
-    send(buffer.data(), buffer.size(), 0, n->getAddr());
+    send(n->getAddr(), buffer.data(), buffer.size());
 }
 
 void
@@ -200,7 +200,7 @@ NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, c
     }
 
     // send response
-    send(buffer.data(), buffer.size(), 0, n->getAddr());
+    send(n->getAddr(), buffer.data(), buffer.size());
 }
 
 
@@ -209,11 +209,11 @@ NetworkEngine::isRunning(sa_family_t af) const
 {
     switch (af) {
     case 0:
-        return dht_socket  >= 0 ||  dht_socket6 >= 0;
+        return dht_socket->hasIPv4() or dht_socket->hasIPv6();
     case AF_INET:
-        return dht_socket  >= 0;
+        return dht_socket->hasIPv4();
     case AF_INET6:
-        return dht_socket6 >= 0;
+        return dht_socket->hasIPv6();
     default:
         return false;
     }
@@ -254,9 +254,7 @@ NetworkEngine::requestStep(Sp<Request> sreq)
         req.on_expired(req, false);
     }
 
-    auto err = send((char*)req.msg.data(), req.msg.size(),
-            (node.getReplyTime() >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM,
-            node.getAddr());
+    auto err = send(node.getAddr(), (char*)req.msg.data(), req.msg.size(), node.getReplyTime() < now - UDP_REPLY_TIME);
     if (err == ENETUNREACH  ||
         err == EHOSTUNREACH ||
         err == EAFNOSUPPORT ||
@@ -610,33 +608,9 @@ insertAddr(msgpack::packer<msgpack::sbuffer>& pk, const SockAddr& addr)
 }
 
 int
-NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr)
+NetworkEngine::send(const SockAddr& addr, const char *buf, size_t len, bool confirmed)
 {
-    if (not addr)
-        return EFAULT;
-
-    int s;
-    if (addr.getFamily() == AF_INET)
-        s = dht_socket;
-    else if (addr.getFamily() == AF_INET6)
-        s = dht_socket6;
-    else
-        s = -1;
-
-    if (s < 0)
-        return EAFNOSUPPORT;
-#ifdef MSG_NOSIGNAL
-    flags |= MSG_NOSIGNAL;
-#endif
-    if (sendto(s, buf, len, flags, addr.get(), addr.getLength()) == -1) {
-        int err = errno;
-        DHT_LOG.e("Can't send message to %s: %s", addr.toString().c_str(), strerror(err));
-        if (err == EPIPE) {
-            throw SocketException(EPIPE);
-        }
-        return err;
-    }
-    return 0;
+    return dht_socket ? dht_socket->sendTo(addr, (const uint8_t*)buf, len, confirmed) : ENOTCONN;
 }
 
 Sp<Request>
@@ -696,7 +670,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
         pk.pack(KEY_NETID); pk.pack(network);
     }
 
-    send(buffer.data(), buffer.size(), 0, addr);
+    send(addr, buffer.data(), buffer.size());
 }
 
 Sp<Request>
@@ -898,7 +872,7 @@ NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals
                     pk.pack(std::string("o")); pk.pack(start);
                     pk.pack(std::string("d")); pk.pack_bin(end-start);
                                                pk.pack_bin_body((const char*)v.data()+start, end-start);
-            send(buffer.data(), buffer.size(), 0, addr);
+            send(addr, buffer.data(), buffer.size());
             start = end;
         } while (start != v.size());
         i++;
@@ -957,7 +931,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
     }
 
     // send response
-    send(buffer.data(), buffer.size(), 0, addr);
+    send(addr, buffer.data(), buffer.size());
 
     // send parts
     if (not svals.empty())
@@ -1100,7 +1074,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
         pk.pack(KEY_NETID); pk.pack(network);
     }
 
-    send(buffer.data(), buffer.size(), 0, addr);
+    send(addr, buffer.data(), buffer.size());
 }
 
 Sp<Request>
@@ -1234,7 +1208,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid)
         pk.pack(KEY_NETID); pk.pack(network);
     }
 
-    send(buffer.data(), buffer.size(), 0, addr);
+    send(addr, buffer.data(), buffer.size());
 }
 
 void
@@ -1266,7 +1240,7 @@ NetworkEngine::sendError(const SockAddr& addr,
         pk.pack(KEY_NETID); pk.pack(network);
     }
 
-    send(buffer.data(), buffer.size(), 0, addr);
+    send(addr, buffer.data(), buffer.size());
 }
 
 void

src/network_utils.cpp

@@ -20,7 +20,6 @@
 
 #ifdef _WIN32
 #include "utils.h"
-#include "sockaddr.h"
 #include <io.h>
 #include <string>
 #include <cstring>
@@ -30,11 +29,40 @@
 #include <fcntl.h>
 #endif
 
+#include <iostream>
+
 namespace dht {
 namespace net {
 
+int
+bindSocket(const SockAddr& addr, SockAddr& bound)
+{
+    bool is_ipv6 = addr.getFamily() == AF_INET6;
+    int sock = socket(is_ipv6 ? PF_INET6 : PF_INET, SOCK_DGRAM, 0);
+    if (sock < 0)
+        throw DhtException(std::string("Can't open socket: ") + strerror(sock));
+    int set = 1;
+#ifdef SO_NOSIGPIPE
+    setsockopt(sock, SOL_SOCKET, SO_NOSIGPIPE, (const char*)&set, sizeof(set));
+#endif
+    if (is_ipv6)
+        setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&set, sizeof(set));
+    net::setNonblocking(sock);
+    int rc = bind(sock, addr.get(), addr.getLength());
+    if (rc < 0) {
+        rc = errno;
+        close(sock);
+        throw DhtException("Can't bind socket on " + addr.toString() + " " + strerror(rc));
+    }
+    sockaddr_storage ss;
+    socklen_t ss_len = sizeof(ss);
+    getsockname(sock, (sockaddr*)&ss, &ss_len);
+    bound = {ss, ss_len};
+    return sock;
+}
+
 bool
-set_nonblocking(int fd, bool nonblocking)
+setNonblocking(int fd, bool nonblocking)
 {
 #ifdef _WIN32
     unsigned long mode = !!nonblocking;
@@ -77,5 +105,214 @@ void udpPipe(int fds[2])
 }
 #endif
 
+UdpSocket::UdpSocket(in_port_t port, const Logger& l) : logger(l) {
+    SockAddr bind4;
+    bind4.setFamily(AF_INET);
+    bind4.setPort(port);
+    SockAddr bind6;
+    bind6.setFamily(AF_INET6);
+    bind6.setPort(port);
+    openSockets(bind4, bind6);
+}
+
+UdpSocket::UdpSocket(const SockAddr& bind4, const SockAddr& bind6, const Logger& l) : logger(l)
+{
+    openSockets(bind4, bind6);
+}
+
+UdpSocket::~UdpSocket() {
+    stop();
+    if (rcv_thread.joinable())
+        rcv_thread.join();
+}
+
+int
+UdpSocket::sendTo(const SockAddr& dest, const uint8_t* data, size_t size, bool replied) {
+    if (not dest)
+        return EFAULT;
+
+    int s;
+    switch (dest.getFamily()) {
+    case AF_INET:  s = s4; break;
+    case AF_INET6: s = s6; break;
+    default:       s = -1; break;
+    }
+
+    if (s < 0)
+        return EAFNOSUPPORT;
+
+    int flags = 0;
+#ifdef MSG_CONFIRM
+    if (replied)
+        flags |= MSG_CONFIRM;
+#endif
+#ifdef MSG_NOSIGNAL
+    flags |= MSG_NOSIGNAL;
+#endif
+
+    if (sendto(s, data, size, flags, dest.get(), dest.getLength()) == -1) {
+        int err = errno;
+        logger.d("Can't send message to %s: %s", dest.toString().c_str(), strerror(err));
+        if (err == EPIPE || err == ENOTCONN || err == ECONNRESET) {
+            auto bind4 = std::move(bound4), bind6 = std::move(bound6);
+            openSockets(bind4, bind6);
+            return sendTo(dest, data, size, false);
+        }
+        return err;
+    }
+    return 0;
+}
+
+void
+UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6)
+{
+    stop();
+    if (rcv_thread.joinable())
+        rcv_thread.join();
+
+    int stopfds[2];
+#ifndef _WIN32
+    auto status = pipe(stopfds);
+    if (status == -1) {
+        throw DhtException(std::string("Can't open pipe: ") + strerror(errno));
+    }
+#else
+    udpPipe(stopfds);
+#endif
+    int stop_readfd = stopfds[0];
+    stopfd = stopfds[1];
+
+    s4 = -1;
+    s6 = -1;
+
+    bound4 = {};
+    if (bind4) {
+        try {
+            s4 = bindSocket(bind4, bound4);
+        } catch (const DhtException& e) {
+            logger.e("Can't bind inet socket: %s", e.what());
+        }
+    }
+
+#if 1
+    bound6 = {};
+    if (bind6) {
+        try {
+            s6 = bindSocket(bind6, bound6);
+        } catch (const DhtException& e) {
+            logger.e("Can't bind inet6 socket: %s", e.what());
+        }
+    }
+#endif
+
+    if (s4 == -1 && s6 == -1) {
+        throw DhtException("Can't bind socket");
+    }
+
+    running = true;
+    rcv_thread = std::thread([this, stop_readfd]() {
+        try {
+            while (running) {
+                fd_set readfds;
+
+                FD_ZERO(&readfds);
+                FD_SET(stop_readfd, &readfds);
+                if(s4 >= 0)
+                    FD_SET(s4, &readfds);
+                if(s6 >= 0)
+                    FD_SET(s6, &readfds);
+
+                int selectFd = std::max({s4, s6, stop_readfd}) + 1;
+                int rc = select(selectFd, &readfds, nullptr, nullptr, nullptr);
+                if (rc < 0) {
+                    if (errno != EINTR) {
+                        logger.e("Select error: %s", strerror(errno));
+                        std::this_thread::sleep_for(std::chrono::seconds(1));
+                    }
+                }
+
+                if (not running)
+                    break;
+
+                if (rc > 0) {
+                    std::array<uint8_t, 1024 * 64> buf;
+                    sockaddr_storage from;
+                    socklen_t from_len = sizeof(from);
+
+                    if (FD_ISSET(stop_readfd, &readfds)) {
+                        if (recv(stop_readfd, (char*)buf.data(), buf.size(), 0) < 0) {
+                            logger.e("Got stop packet error: %s", strerror(errno));
+                            break;
+                        }
+                    }
+                    else 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
+                        continue;
+
+                    if (rc > 0) {
+                        auto pkt = std::unique_ptr<ReceivedPacket>(new ReceivedPacket);
+                        pkt->data = {buf.begin(), buf.begin()+rc};
+                        pkt->from = {from, from_len};
+                        pkt->received = clock::now();
+                        onReceived(std::move(pkt));
+                    } else if (rc == -1) {
+                        logger.e("Error receiving packet: %s", strerror(errno));
+                        int err = errno;
+                        if (err == EPIPE || err == ENOTCONN || err == ECONNRESET) {
+                            if (s4 >= 0) {
+                                close(s4);
+                                try {
+                                    s4 = bindSocket(bound4, bound4);
+                                } catch (const DhtException& e) {
+                                    logger.e("Can't bind inet socket: %s", e.what());
+                                }
+                            }
+                            if (s6 >= 0) {
+                                close(s6);
+                                try {
+                                    s6 = bindSocket(bound6, bound6);
+                                } catch (const DhtException& e) {
+                                    logger.e("Can't bind inet6 socket: %s", e.what());
+                                }
+                            }
+                            if (s4 < 0 && s6 < 0)
+                                break;
+                        }
+                    }
+                }
+            }
+        } catch (const std::exception& e) {
+            logger.e("Error in UdpSocket rx thread: %s", e.what());
+        }
+        if (s4 >= 0)
+            close(s4);
+        if (s6 >= 0)
+            close(s6);
+        s4 = -1;
+        s6 = -1;
+        bound4 = {};
+        bound6 = {};
+        if (stop_readfd != -1)
+            close(stop_readfd);
+        if (stopfd != -1)
+            close(stopfd);
+        stopfd = -1;
+    });
+}
+
+void
+UdpSocket::stop()
+{
+    if (running.exchange(false)) {
+        auto sfd = stopfd;
+        if (sfd != -1 && write(sfd, "\0", 1) == -1) {
+            logger.e("Can't write to stop fd");
+        }
+    }
+}
+
 }
 }

src/network_utils.h

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

src/peer_discovery.cpp

@@ -35,6 +35,10 @@ typedef SSIZE_T ssize_t;
 #endif
 #include <fcntl.h>
 
+#ifndef IPV6_JOIN_GROUP
+#define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
+#endif
+
 namespace dht {
 
 // Organization-local Scope multicast
@@ -175,7 +179,7 @@ PeerDiscovery::DomainPeerDiscovery::initialize_socket(sa_family_t domain)
     if (sockfd < 0) {
         throw std::runtime_error(std::string("Socket Creation Error: ") + strerror(errno));
     }
-    net::set_nonblocking(sockfd);
+    net::setNonblocking(sockfd);
     return sockfd;
 }
 
@@ -502,7 +506,6 @@ PeerDiscovery::PeerDiscovery(in_port_t port)
     } catch(const std::exception& e) {
         std::cerr << "Can't start peer discovery (IPv6): " << e.what() << std::endl;
     }
-
 }
 
 PeerDiscovery::~PeerDiscovery(){}

src/utils.cpp

@@ -97,9 +97,11 @@ SockAddr::setAddress(const char* address)
     case AF_INET6:
         addr = &getIPv6().sin6_addr;
         break;
+    default:
+        throw std::runtime_error("Unknown address family");
     }
-    if (not addr or inet_pton(family, address, addr) <= 0)
-        throw std::runtime_error("Can't parse IP address");
+    if (inet_pton(family, address, addr) <= 0)
+        throw std::runtime_error(std::string("Can't parse IP address: ") + strerror(errno));
 }
 
 std::string