update to version 1.9.0

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 8.2)
+set (opendht_VERSION_MINOR 9.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
-m4_define([opendht_minor_version], 8)
-m4_define([opendht_patch_version], 2)
+m4_define([opendht_minor_version], 9)
+m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/peer_discovery.h

@@ -119,19 +119,6 @@ private:
      * Sender Parameters Setup
     */
     void sender_setup(const dht::InfoHash& nodeId, in_port_t port_to_send);
-
-    /**
-     * Binary Converters
-    */
-    static void inttolitend(uint32_t x, uint8_t *lit_int) {
-        lit_int[0] = (uint8_t)(x >>  0);
-        lit_int[1] = (uint8_t)(x >>  8);
-    }
-
-    static uint16_t litendtoint(uint8_t *lit_int) {
-        return (uint32_t)lit_int[0] <<  0
-            |  (uint32_t)lit_int[1] <<  8;
-    }
 };
 
 }

src/dhtrunner.cpp

@@ -33,10 +33,10 @@ constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
 static constexpr size_t RX_QUEUE_MAX_SIZE = 1024 * 16;
 
 struct DhtRunner::Listener {
-    size_t tokenClassicDht;
-    size_t tokenProxyDht;
+    size_t tokenClassicDht {0};
+    size_t tokenProxyDht {0};
     ValueCallback gcb;
-    InfoHash hash;
+    InfoHash hash {};
     Value::Filter f;
     Where w;
 };
@@ -138,13 +138,13 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const DhtRunner::
     if (config.peer_discovery or config.peer_publish) {
         try {
             peerDiscovery4_.reset(new PeerDiscovery(AF_INET, port_multicast));
-        } catch(std::exception &exception){
-            perror(exception.what());
+        } catch(const std::exception& e){
+            std::cerr << "Can't start peer discovery (IPv4): " << e.what() << std::endl;
         }
-        try{
+        try {
             peerDiscovery6_.reset(new PeerDiscovery(AF_INET6, port_multicast));
-        } catch(std::exception &exception){
-            perror(exception.what());
+        } catch(const std::exception& e) {
+            std::cerr << "Can't start peer discovery (IPv6): " << e.what() << std::endl;
         }
     }
     if (config.peer_discovery) {
@@ -661,22 +661,21 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
         pending_ops.emplace([=](SecureDht& dht) mutable {
 #ifdef OPENDHT_PROXY_CLIENT
             auto tokenbGlobal = listener_token_++;
-            Listener listener {};
+            auto& listener = listeners_[tokenbGlobal];
             listener.hash = hash;
             listener.f = std::move(f);
             listener.w = std::move(w);
-            listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired){
+            listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired) {
                 if (not vcb(vals, expired)) {
                     cancelListen(hash, tokenbGlobal);
                     return false;
                 }
                 return true;
             };
-            if (use_proxy)
-                listener.tokenProxyDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
-            else
-                listener.tokenClassicDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
-            listeners_.emplace(tokenbGlobal, std::move(listener));
+            if (auto token = dht.listen(hash, listener.gcb, listener.f, listener.w)) {
+                if (use_proxy)  listener.tokenProxyDht = token;
+                else            listener.tokenClassicDht = token;
+            }
             ret_token->set_value(tokenbGlobal);
 #else
             ret_token->set_value(dht.listen(hash, std::move(vcb), std::move(f), std::move(w)));

src/network_utils.h

@@ -23,6 +23,10 @@
 #define write(s, b, f) send(s, b, (int)strlen(b), 0)
 #endif
 
+#ifndef IPV6_JOIN_GROUP
+#define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
+#endif
+
 namespace dht {
 namespace net {
 

src/peer_discovery.cpp

@@ -75,13 +75,16 @@ PeerDiscovery::listener_setup()
     sockAddrListen_.setAny();
 
     unsigned int opt = 1;
-    if (setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR|SO_REUSEPORT, (char*) &opt, sizeof(opt)) < 0){
-       throw std::runtime_error(std::string("Reusing ADDR failed: ") + strerror(errno));
+    if (setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
+       std::cerr << "setsockopt SO_REUSEADDR failed: " << strerror(errno) << std::endl;
+    }
+    if (setsockopt(sockfd_, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
+       std::cerr << "setsockopt SO_REUSEPORT failed: " << strerror(errno) << std::endl;
     }
 
     // bind to receive address
     if (bind(sockfd_, sockAddrListen_.get(), sockAddrListen_.getLength()) < 0){
-        throw std::runtime_error(std::string("Bind Socket For Listener Error: ") + strerror(errno));
+        throw std::runtime_error(std::string("Error binding socket: ") + strerror(errno));
     }
 }
 
@@ -97,31 +100,31 @@ PeerDiscovery::socketJoinMulticast(int sockfd, sa_family_t family)
         //multicast datagrams from the sockets application.
         config_ipv4.imr_interface.s_addr = htonl(INADDR_ANY);
         if( setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_IF, &config_ipv4.imr_interface, sizeof( struct in_addr )) < 0 ) {
-            throw std::runtime_error(std::string("Bound Network Interface IPV4 Error: ") + strerror(errno));
+            throw std::runtime_error(std::string("Bound Network Interface IPv4 Error: ") + strerror(errno));
         }
 
         //The IP_MULTICAST_TTL socket option allows the application to primarily
         //limit the lifetime of the packet in the Internet and prevent it from circulating indefinitely
         unsigned char ttl4 = 20;
         if( setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl4, sizeof( ttl4 )) < 0 ) {
-            throw std::runtime_error(std::string(" TTL Sockopt Error: ") + strerror(errno));
+            throw std::runtime_error(std::string("TTL Sockopt Error: ") + strerror(errno));
         }
 
         // config the listener to be interested in joining in the multicast group
         config_ipv4.imr_multiaddr.s_addr = inet_addr(MULTICAST_ADDRESS_IPV4);
         config_ipv4.imr_interface.s_addr = htonl(INADDR_ANY);
         if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char*)&config_ipv4, sizeof(config_ipv4)) < 0){
-            throw std::runtime_error(std::string(" Member Addition IPV4 Error: ") + strerror(errno));
+            throw std::runtime_error(std::string(" Member Addition IPv4 Error: ") + strerror(errno));
         }
         break;
     }
     case AF_INET6: {
         ipv6_mreq config_ipv6;
 
-        unsigned int outif = 0;
-        if( setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_IF, &outif, sizeof( outif )) < 0 ) {
-            throw std::runtime_error(std::string("Bound Network Interface IPV6 Error: ") + strerror(errno));
-        }
+        /* unsigned int outif = 0;
+        if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_IF, &outif, sizeof(outif)) < 0) {
+            std::cerr << "Can't assign multicast interface: " << strerror(errno) << std::endl;
+        } */
 
         unsigned int ttl6 = 20;
         if( setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &ttl6, sizeof( ttl6 )) < 0 ) {
@@ -130,8 +133,8 @@ PeerDiscovery::socketJoinMulticast(int sockfd, sa_family_t family)
 
         config_ipv6.ipv6mr_interface = 0;
         inet_pton(AF_INET6, MULTICAST_ADDRESS_IPV6, &config_ipv6.ipv6mr_multiaddr);
-        if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &config_ipv6, sizeof(config_ipv6)) < 0){
-            throw std::runtime_error(std::string("Member Addition IPV6 Error: ") + strerror(errno));
+        if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &config_ipv6, sizeof(config_ipv6)) < 0){
+            throw std::runtime_error(std::string("Member Addition IPv6 Error: ") + strerror(errno));
         }
         break;
     }
@@ -181,20 +184,15 @@ void
 PeerDiscovery::sender_setup(const dht::InfoHash& nodeId, in_port_t port_to_send)
 {
     nodeId_ = nodeId;
-    //Set up for Sender
+    // Setup sender address
     sockAddrSend_.setFamily(domain_);
     sockAddrSend_.setAddress(domain_ == AF_INET ? MULTICAST_ADDRESS_IPV4 : MULTICAST_ADDRESS_IPV6);
     sockAddrSend_.setPort(port_);
 
-    //Setup for send data
-    int port_node = port_to_send;
-    uint8_t port_node_binary[2];
-    PeerDiscovery::inttolitend(port_node,port_node_binary);
-
-    //Copy Node id and node port
-    memcpy (data_send_.data(), nodeId.data(), nodeId.size());
-    data_send_[InfoHash::size()] = port_node_binary[0];
-    data_send_[InfoHash::size() + 1] = port_node_binary[1];
+    // Setup sent data
+    std::copy_n(nodeId.cbegin(), nodeId.size(), data_send_.begin());
+    auto portAddr = reinterpret_cast<in_port_t*>(data_send_.data() + dht::InfoHash::size());
+    *portAddr = htons(port_to_send);
 }
 
 void
@@ -252,29 +250,22 @@ PeerDiscovery::listener_thread(PeerDiscoveredCallback callback)
         }
 
         if (data_coming > 0) {
-
-            if(FD_ISSET(stop_readfd, &readfds)){ break; }
+            if (FD_ISSET(stop_readfd, &readfds)) { break; }
 
             std::array<uint8_t,dht::InfoHash::size() + sizeof(in_port_t)> data_receive;
             size_t data_receive_size = data_receive.size();
             auto from = recvFrom(data_receive.data(), data_receive_size);
 
-            //Data_receive_size as a value-result member will hlep to filter packs
+            // Data_receive_size as a value-result member will hlep to filter packs
             if(data_receive_size != data_receive.size()){
-                perror("Data Received Unmatch");
+                // std::cerr << "Received invalid peer discovery packet" << std::endl;
                 continue;
             }
 
-            std::array<uint8_t,dht::InfoHash::size()> data_infohash;
-            uint8_t data_port[2];
-
-            memcpy (data_infohash.data(), data_receive.data(), dht::InfoHash::size());
-            data_port[0] = data_receive[dht::InfoHash::size()];
-            data_port[1] = data_receive[dht::InfoHash::size() + 1];
-
-            auto port = PeerDiscovery::litendtoint(data_port);
-            auto nodeId = dht::InfoHash(data_infohash.data(), dht::InfoHash::size());
-
+            dht::InfoHash nodeId;
+            std::copy_n(data_receive.begin(), dht::InfoHash::size(), nodeId.begin());
+            auto portAddr = reinterpret_cast<in_port_t*>(data_receive.data() + dht::InfoHash::size());
+            auto port = ntohs(*portAddr);
             if (nodeId != nodeId_){
                 from.setPort(port);
                 callback(nodeId, from);

tests/dhtrunnertester.cpp

@@ -71,7 +71,7 @@ DhtRunnerTester::testListen() {
     auto a = dht::InfoHash::get("234");
     auto b = dht::InfoHash::get("2345");
     auto c = dht::InfoHash::get("23456");
-    constexpr unsigned N = 64;
+    constexpr unsigned N = 32;
 
     auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&) {
         valueCount++;
@@ -114,18 +114,21 @@ DhtRunnerTester::testListen() {
         CPPUNIT_ASSERT_EQUAL(N * 2u, putOkCount);
     }
 
+    CPPUNIT_ASSERT(ftokena.valid());
+    CPPUNIT_ASSERT(ftokenb.valid());
+    CPPUNIT_ASSERT(ftokenc.valid());
+
     auto tokena = ftokena.get();
-    auto tokenb = ftokenb.get();
     auto tokenc = ftokenc.get();
+    // tokenb might be 0 since the callback returns false.
 
     CPPUNIT_ASSERT(tokena);
-    CPPUNIT_ASSERT(tokenb);
     CPPUNIT_ASSERT(tokenc);
     CPPUNIT_ASSERT_EQUAL(N + 1u, valueCount.load());
 
     node1.cancelListen(a, tokena);
-    node1.cancelListen(b, tokena);
-    node1.cancelListen(c, tokena);
+    node1.cancelListen(b, std::move(ftokenb));
+    node1.cancelListen(c, tokenc);
 }
 
 }  // namespace test