search: expire only after candidate expiration

CMakeLists.txt

@@ -4,7 +4,7 @@ include(CMakeDependentOption)
 
 project (opendht)
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 1.0)
+set (opendht_VERSION_MINOR 0.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

configure.ac

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

include/opendht/callbacks.h

@@ -82,8 +82,6 @@ struct OPENDHT_PUBLIC NodeInfo {
     NodeStats ipv4 {};
     NodeStats ipv6 {};
     size_t ongoing_ops {0};
-    in_port_t bound4 {0};
-    in_port_t bound6 {0};
 
 #ifdef OPENDHT_JSONCPP
     /**

include/opendht/dht.h

@@ -445,7 +445,7 @@ private:
     void reportedAddr(const SockAddr&);
 
     // Storage
-    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
+    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
     bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
     bool storageErase(const InfoHash& id, Value::Id vid);
     bool storageRefresh(const InfoHash& id, Value::Id vid);
@@ -600,8 +600,7 @@ private:
             const InfoHash& hash,
             const Blob& token,
             size_t socket_id,
-            const Query& query,
-            int version = 0);
+            const Query& query);
     void onListenDone(const Sp<Node>& status,
             net::RequestAnswer& a,
             Sp<Search>& sr);

include/opendht/network_engine.h

@@ -180,8 +180,7 @@ private:
             const InfoHash&,
             const Blob&,
             Tid,
-            const Query&,
-            int)> onListen {};
+            const Query&)> onListen {};
     /**
      * Called on announce request.
      *
@@ -238,10 +237,9 @@ public:
     void clear();
 
     /**
-     * Sends values (with closest nodes) to a listener.
+     * Sends values (with closest nodes) to a listenner.
      *
-     * @deprecated
-     * @param sa          The address of the listener.
+     * @param sa          The address of the listenner.
      * @param sslen       The length of the sockaddr structure.
      * @param socket_id  The tid to use to write to the request socket.
      * @param hash        The hash key of the value.
@@ -250,14 +248,13 @@ public:
      * @param nodes       The ipv4 closest nodes.
      * @param nodes6      The ipv6 closest nodes.
      * @param values      The values to send.
-     * @param version     If version = 1, a request will be used to answer to the listener
      */
     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, int version);
+            std::vector<Sp<Value>>&& values, const Query& q);
 
-    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
-    void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
+    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
+    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->hasIPv4() and dht_socket->hasIPv6() ? (WANT4 | WANT6) : -1; }
@@ -404,24 +401,6 @@ public:
                                  const Blob& token,
                                  RequestCb&& on_done,
                                  RequestExpiredCb&& on_expired);
-    /**
-     * Send a "update" request to a given node. Used for Listen operations
-     *
-     * @param n           The node.
-     * @param hash        The target hash.
-     * @param values      The values.
-     * @param created     Time id.
-     * @param token       A security token.
-     * @param sid         The socket id.
-     *
-     * @return the request with information concerning its success.
-     */
-    Sp<Request> sendUpdateValues(Sp<Node> n,
-                                 const InfoHash& infohash,
-                                 const std::vector<Sp<Value>>& values,
-                                 time_point created,
-                                 const Blob& token,
-                                 const size_t& sid);
 
     /**
      * Parses a message and calls appropriate callbacks.
@@ -511,20 +490,19 @@ private:
     void sendRequest(const Sp<Request>& request);
 
     struct MessageStats {
-        unsigned ping         {0};
-        unsigned find         {0};
-        unsigned get          {0};
-        unsigned put          {0};
-        unsigned listen       {0};
-        unsigned refresh      {0};
-        unsigned updateValue  {0};
+        unsigned ping    {0};
+        unsigned find    {0};
+        unsigned get     {0};
+        unsigned put     {0};
+        unsigned listen  {0};
+        unsigned refresh {0};
     };
 
 
     // basic wrapper for socket sendto function
     int send(const SockAddr& addr, const char *buf, size_t len, bool confirmed = false);
 
-    void sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr);
+    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);
 

src/dht.cpp

@@ -1201,7 +1201,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
                 vals.push_back(v.data);
                 Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
                 network_engine.tellListener(node_listeners.first, l.first, id, 0, ntoken, {}, {},
-                        std::move(vals), l.second.query, l.second.version);
+                        std::move(vals), l.second.query);
             }
         }
     }
@@ -1259,7 +1259,7 @@ Dht::storageErase(const InfoHash& id, Value::Id vid)
 }
 
 void
-Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
+Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
 {
     const auto& now = scheduler.time();
     auto st = store.find(id);
@@ -1275,9 +1275,9 @@ Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_
         if (not vals.empty()) {
             network_engine.tellListener(node, socket_id, id, WANT4 | WANT6, makeToken(node->getAddr(), false),
                     dht4.buckets.findClosestNodes(id, now, TARGET_NODES), dht6.buckets.findClosestNodes(id, now, TARGET_NODES),
-                    std::move(vals), query, version);
+                    std::move(vals), query);
         }
-        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query), version});
+        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query)});
     }
     else
         l->second.refresh(now, std::forward<Query>(query));
@@ -1312,7 +1312,7 @@ Dht::expireStore(decltype(store)::iterator i)
                             id.toString().c_str(),
                             node_listeners.first->toString().c_str());
                     Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
+                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids);
                 }
             }
         }
@@ -1755,7 +1755,7 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
             std::bind(&Dht::onPing, this, _1),
             std::bind(&Dht::onFindNode, this, _1, _2, _3),
             std::bind(&Dht::onGetValues, this, _1, _2, _3, _4),
-            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5, _6),
+            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5),
             std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5),
             std::bind(&Dht::onRefresh, this, _1, _2, _3, _4)),
     persistPath(config.persist_path),
@@ -1982,19 +1982,14 @@ Dht::onDisconnected()
         return;
     if (logger_)
         logger_->d(myid, "Bootstraping");
-    for (const auto& boootstrap : bootstrap_nodes) {
+    for (const auto& boootstrap : bootstrap_nodes)
         try {
-            auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
-            for (auto& ip : ips) {
-                if (ip.getPort() == 0)
-                    ip.setPort(net::DHT_DEFAULT_PORT);
+            for (const auto& ip : network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second))
                 pingNode(ip);
-            }
         } catch (const std::exception& e) {
             if (logger_)
                 logger_->e(myid, "Can't resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
         }
-    }
     if (bootstrapJob)
         bootstrapJob->cancel();
     bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::onDisconnected, this));
@@ -2325,7 +2320,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
 }
 
 net::RequestAnswer
-Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query, int version)
+Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query)
 {
     if (not hash) {
         if (logger_)
@@ -2341,7 +2336,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
         throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
     }
     Query q = query;
-    storageAddListener(hash, node, socket_id, std::move(q), version);
+    storageAddListener(hash, node, socket_id, std::move(q));
     return {};
 }
 
@@ -2480,7 +2475,7 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
                             id.toString().c_str(),
                             node_listeners.first->toString().c_str());
                     Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
+                    network_engine.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids);
                 }
             }
         }

src/dhtrunner.cpp

@@ -475,10 +475,6 @@ DhtRunner::getNodeInfo() const {
         info.node_id = dht->getNodeId();
         info.ipv4 = dht->getNodesStats(AF_INET);
         info.ipv6 = dht->getNodesStats(AF_INET6);
-        if (auto sock = dht->getSocket()) {
-            info.bound4 = sock->getBoundRef(AF_INET).getPort();
-            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
-        }
     }
     info.ongoing_ops = ongoing_ops;
     return info;
@@ -496,10 +492,6 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
         info.node_id = dht.getNodeId();
         info.ipv4 = dht.getNodesStats(AF_INET);
         info.ipv6 = dht.getNodesStats(AF_INET6);
-        if (auto sock = dht.getSocket()) {
-            info.bound4 = sock->getBoundRef(AF_INET).getPort();
-            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
-        }
         info.ongoing_ops = ongoing_ops;
         cb(std::move(sinfo));
         opEnded();

src/listener.h

@@ -30,9 +30,8 @@ namespace dht {
 struct Listener {
     time_point time;
     Query query;
-    int version;
 
-    Listener(time_point t, Query&& q, int version = 0) : time(t), query(std::move(q)), version(version) {}
+    Listener(time_point t, Query&& q) : time(t), query(std::move(q)) {}
 
     void refresh(time_point t, Query&& q) {
         time = t;

src/net.h

@@ -31,8 +31,7 @@ enum class MessageType {
     Refresh,
     Listen,
     ValueData,
-    ValueUpdate,
-    UpdateValue
+    ValueUpdate
 };
 
 } /* namespace net */

src/network_engine.cpp

@@ -48,6 +48,23 @@ static constexpr uint8_t v4prefix[16] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xF
 constexpr unsigned SEND_NODES {8};
 
 
+/* 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> {
+    TransId() { std::fill(begin(), end(), 0); }
+    TransId(const std::array<char, 4>& o) { std::copy(o.begin(), o.end(), begin()); }
+    TransId(uint32_t id) {
+        *reinterpret_cast<uint32_t*>(data()) = htonl(id);
+    }
+
+    uint32_t toInt() const {
+        return ntohl(*reinterpret_cast<const uint32_t*>(&(*this)[0]));
+    }
+};
+
+
 struct NetworkEngine::PartialMessage {
     SockAddr from;
     time_point start;
@@ -123,15 +140,11 @@ void
 NetworkEngine::tellListener(Sp<Node> node, 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& query, int version)
+        const Query& query)
 {
     auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
     try {
-        if (version >= 1) {
-            sendUpdateValues(node, hash, values, scheduler.time(), ntoken, socket_id);
-        } else {
-            sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
-        }
+        sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
     } catch (const std::overflow_error& e) {
         if (logger_)
             logger_->e("Can't send value: buffer not large enough !");
@@ -139,12 +152,11 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
 }
 
 void
-NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
+NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
-
-    pk.pack_map(4 + (version >= 1 ? 2 : 0) + (config.network?1:0));
+    pk.pack_map(4+(config.network?1:0));
 
     pk.pack(KEY_U);
     pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
@@ -166,34 +178,16 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&,
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    if (version >= 1) {
-        Tid tid (n->getNewTid());
-        Tid sid (socket_id);
-
-        pk.pack(KEY_REQ_SID);   pk.pack(sid);
-        pk.pack(KEY_TID); pk.pack(tid);
-
-        auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
-            Blob(buffer.data(), buffer.data() + buffer.size()),
-            [=](const Request&, ParsedMessage&&) { /* on done */ },
-            [=](const Request&, bool) { /* on expired */ }
-        );
-        sendRequest(req);
-        ++out_stats.updateValue;
-        return;
-    }
-
     // send response
     send(n->getAddr(), buffer.data(), buffer.size());
 }
 
 void
-NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
+NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
-
-    pk.pack_map(4 + (version >= 1 ? 2 : 0) + (config.network?1:0));
+    pk.pack_map(4+(config.network?1:0));
 
     pk.pack(KEY_U);
     pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
@@ -215,23 +209,6 @@ NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, c
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    if (version >= 1) {
-        Tid tid (n->getNewTid());
-        Tid sid (socket_id);
-
-        pk.pack(KEY_REQ_SID);   pk.pack(sid);
-        pk.pack(KEY_TID); pk.pack(tid);
-
-        auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
-            Blob(buffer.data(), buffer.data() + buffer.size()),
-            [=](const Request&, ParsedMessage&&) { /* on done */ },
-            [=](const Request&, bool) { /* on expired */ }
-        );
-        sendRequest(req);
-        ++out_stats.updateValue;
-        return;
-    }
-
     // send response
     send(n->getAddr(), buffer.data(), buffer.size());
 }
@@ -640,28 +617,18 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 if (logIncoming_ and logger_)
                     logger_->d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                 ++in_stats.listen;
-                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query), msg->version);
+                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
                 auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
                 sendListenConfirmation(from, msg->tid);
                 break;
             }
-            case MessageType::UpdateValue: {
-                if (logIncoming_ and logger_)
-                    logger_->d(msg->info_hash, node->id, "[node %s] got 'update' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
-                ++in_stats.updateValue;
-                if (auto rsocket = node->getSocket(msg->socket_id))
-                    rsocket->on_receive(node, std::move(*msg));
-                else if (logger_)
-                    logger_->e(msg->info_hash, node->id, "[node %s] 'update' request without socket for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
-                break;
-            }
             default:
                 break;
             }
         } catch (const std::overflow_error& e) {
             if (logger_)
                 logger_->e("Can't send value: buffer not large enough !");
-        } catch (const DhtProtocolException& e) {
+        } catch (DhtProtocolException& e) {
             sendError(from, msg->tid, e.getCode(), e.getMsg().c_str(), true);
         }
     }
@@ -687,7 +654,7 @@ NetworkEngine::send(const SockAddr& addr, const char *buf, size_t len, bool conf
 
 Sp<Request>
 NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    Tid tid (node->getNewTid());
+    TransId tid (node->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -696,14 +663,15 @@ NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& o
      pk.pack(KEY_REQ_ID); pk.pack(myid);
 
     pk.pack(KEY_Q); pk.pack(QUERY_PING);
-    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                              pk.pack_bin_body((const char*)tid.data(), tid.size());
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Ping, tid, node,
+    auto req = std::make_shared<Request>(MessageType::Ping, tid.toInt(), node,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&&) {
             if (logger_)
@@ -733,7 +701,9 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
       pk.pack(KEY_REQ_ID); pk.pack(myid);
       insertAddr(pk, addr);
 
-    pk.pack(KEY_TID); pk.pack(tid);
+    TransId t (tid);
+    pk.pack(KEY_TID); pk.pack_bin(t.size());
+                               pk.pack_bin_body((const char*)t.data(), t.size());
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -746,7 +716,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
 Sp<Request>
 NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
         RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -762,14 +732,15 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
     }
 
     pk.pack(KEY_Q); pk.pack(QUERY_FIND);
-    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                               pk.pack_bin_body((const char*)tid.data(), tid.size());
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::FindNode, tid, n,
+    auto req = std::make_shared<Request>(MessageType::FindNode, tid.toInt(), n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done) {
@@ -791,7 +762,7 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
 Sp<Request>
 NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query& query, want_t want,
         RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -815,14 +786,15 @@ NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query&
       }
 
     pk.pack(KEY_Q); pk.pack(QUERY_GET);
-    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                      pk.pack_bin_body((const char*)tid.data(), tid.size());
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::GetValues, tid, n,
+    auto req = std::make_shared<Request>(MessageType::GetValues, tid.toInt(), n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done) {
@@ -924,7 +896,7 @@ NetworkEngine::packValueHeader(msgpack::sbuffer& buffer, const std::vector<Sp<Va
 }
 
 void
-NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr)
+NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr)
 {
     msgpack::sbuffer buffer;
     unsigned i=0;
@@ -939,7 +911,8 @@ NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const Soc
                 pk.pack(KEY_NETID); pk.pack(config.network);
             }
             pk.pack(KEY_Y); pk.pack(KEY_V);
-            pk.pack(KEY_TID); pk.pack(tid);
+            pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                                       pk.pack_bin_body((const char*)tid.data(), tid.size());
             pk.pack(KEY_V); pk.pack_map(1);
                 pk.pack(i); pk.pack_map(2);
                     pk.pack(std::string("o")); pk.pack(start);
@@ -994,7 +967,9 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
         }
     }
 
-    pk.pack(KEY_TID); pk.pack(tid);
+    TransId t (tid);
+    pk.pack(KEY_TID); pk.pack_bin(t.size());
+                      pk.pack_bin_body((const char*)t.data(), t.size());
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1070,7 +1045,7 @@ NetworkEngine::sendListen(Sp<Node> n,
         SocketCb&& socket_cb)
 {
     Tid socket;
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
     if (previous and previous->node == n) {
         socket = previous->getSocket();
     } else {
@@ -1085,30 +1060,33 @@ NetworkEngine::sendListen(Sp<Node> n,
             logger_->e(hash, "[node %s] unable to get a valid socket for listen. Aborting listen", n->toString().c_str());
         return {};
     }
+    TransId sid(socket);
+
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
 
     auto has_query = not query.where.empty() or not query.select.empty();
-    pk.pack(KEY_A); pk.pack_map(5 + has_query);
+    pk.pack(KEY_A); pk.pack_map(4 + has_query);
       pk.pack(KEY_REQ_ID);    pk.pack(myid);
-      pk.pack(KEY_VERSION);   pk.pack(1);
       pk.pack(KEY_REQ_H);     pk.pack(hash);
       pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-      pk.pack(KEY_REQ_SID);   pk.pack(socket);
+      pk.pack(KEY_REQ_SID);   pk.pack_bin(sid.size());
+                              pk.pack_bin_body((const char*)sid.data(), sid.size());
       if (has_query) {
           pk.pack(KEY_REQ_QUERY); pk.pack(query);
       }
 
     pk.pack(KEY_Q); pk.pack(QUERY_LISTEN);
-    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                               pk.pack_bin_body((const char*)tid.data(), tid.size());
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Listen, tid, n,
+    auto req = std::make_shared<Request>(MessageType::Listen, tid.toInt(), n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (on_done)
@@ -1135,7 +1113,9 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
       pk.pack(KEY_REQ_ID); pk.pack(myid);
       insertAddr(pk, addr);
 
-    pk.pack(KEY_TID); pk.pack(tid);
+    TransId t (tid);
+    pk.pack(KEY_TID); pk.pack_bin(t.size());
+                               pk.pack_bin_body((const char*)t.data(), t.size());
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1154,7 +1134,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
         RequestCb&& on_done,
         RequestExpiredCb&& on_expired)
 {
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -1170,14 +1150,15 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
       pk.pack(KEY_REQ_TOKEN);  pk.pack(token);
 
     pk.pack(KEY_Q);   pk.pack(QUERY_PUT);
-    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                      pk.pack_bin_body((const char*)tid.data(), tid.size());
     pk.pack(KEY_Y);   pk.pack(KEY_Q);
     pk.pack(KEY_UA);  pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid, n,
+    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid.toInt(), n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (msg.value_id == Value::INVALID_ID) {
@@ -1203,52 +1184,6 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
     return req;
 }
 
-Sp<Request>
-NetworkEngine::sendUpdateValues(Sp<Node> n,
-                                const InfoHash& infohash,
-                                const std::vector<Sp<Value>>& values,
-                                time_point created,
-                                const Blob& token,
-                                const size_t& socket_id)
-{
-    Tid tid (n->getNewTid());
-    Tid sid (socket_id);
-
-    msgpack::sbuffer buffer;
-    msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(5+(config.network?1:0));
-
-    pk.pack(KEY_A); pk.pack_map((created < scheduler.time() ? 7 : 6));
-      pk.pack(KEY_REQ_ID);     pk.pack(myid);
-      pk.pack(KEY_VERSION);    pk.pack(1);
-      pk.pack(KEY_REQ_H);      pk.pack(infohash);
-      pk.pack(KEY_REQ_SID);   pk.pack(sid);
-      auto v = packValueHeader(buffer, values);
-      if (created < scheduler.time()) {
-          pk.pack(KEY_REQ_CREATION);
-          pk.pack(to_time_t(created));
-      }
-      pk.pack(KEY_REQ_TOKEN);  pk.pack(token);
-
-    pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
-    pk.pack(KEY_TID); pk.pack(tid);
-    pk.pack(KEY_Y);   pk.pack(KEY_Q);
-    pk.pack(KEY_UA);  pk.pack(my_v);
-    if (config.network) {
-        pk.pack(KEY_NETID); pk.pack(config.network);
-    }
-
-    auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
-        Blob(buffer.data(), buffer.data() + buffer.size()),
-        [=](const Request&, ParsedMessage&&) { /* on done */ },
-        [=](const Request&, bool) { /* on expired */ }
-    );
-    req->parts = std::move(v);
-    sendRequest(req);
-    ++out_stats.updateValue;
-    return req;
-}
-
 Sp<Request>
 NetworkEngine::sendRefreshValue(Sp<Node> n,
                 const InfoHash& infohash,
@@ -1257,7 +1192,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                 RequestCb&& on_done,
                 RequestExpiredCb&& on_expired)
 {
-    Tid tid (n->getNewTid());
+    TransId tid (n->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -1269,14 +1204,15 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
       pk.pack(KEY_REQ_TOKEN);    pk.pack(token);
 
     pk.pack(KEY_Q); pk.pack(QUERY_REFRESH);
-    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_TID); pk.pack_bin(tid.size());
+                               pk.pack_bin_body((const char*)tid.data(), tid.size());
     pk.pack(KEY_Y); pk.pack(KEY_Q);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
     }
 
-    auto req = std::make_shared<Request>(MessageType::Refresh, tid, n,
+    auto req = std::make_shared<Request>(MessageType::Refresh, tid.toInt(), n,
         Blob(buffer.data(), buffer.data() + buffer.size()),
         [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
             if (msg.value_id == Value::INVALID_ID) {
@@ -1312,7 +1248,9 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid)
       pk.pack(KEY_REQ_VALUE_ID); pk.pack(vid);
       insertAddr(pk, addr);
 
-    pk.pack(KEY_TID); pk.pack(tid);
+    TransId t(tid);
+    pk.pack(KEY_TID); pk.pack_bin(t.size());
+                               pk.pack_bin_body((const char*)t.data(), t.size());
     pk.pack(KEY_Y); pk.pack(KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
@@ -1342,7 +1280,9 @@ NetworkEngine::sendError(const SockAddr& addr,
           pk.pack(KEY_REQ_ID); pk.pack(myid);
     }
 
-    pk.pack(KEY_TID); pk.pack(tid);
+    TransId t(tid);
+    pk.pack(KEY_TID); pk.pack_bin(t.size());
+                               pk.pack_bin_body((const char*)t.data(), t.size());
     pk.pack(KEY_Y); pk.pack(KEY_E);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {

src/parsed_message.h

@@ -54,12 +54,10 @@ static const std::string KEY_REQ_EXPIRED {"exp"};
 static const std::string KEY_REQ_REFRESHED {"re"};
 static const std::string KEY_REQ_FIELDS {"fileds"};
 static const std::string KEY_REQ_WANT {"w"};
-static const std::string KEY_VERSION {"ve"};
 
 static const std::string QUERY_PING {"ping"};
 static const std::string QUERY_FIND {"find"};
 static const std::string QUERY_GET {"get"};
-static const std::string QUERY_UPDATE {"update"};
 static const std::string QUERY_PUT {"put"};
 static const std::string QUERY_LISTEN {"listen"};
 static const std::string QUERY_REFRESH {"refresh"};
@@ -115,7 +113,6 @@ struct ParsedMessage {
     uint16_t error_code;
     /* reported address by the distant node */
     std::string ua;
-    int version {0};
     SockAddr addr;
     void msgpack_unpack(const msgpack::object& o);
 
@@ -227,8 +224,6 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
         type = MessageType::AnnounceValue;
     else if (parsed.q == QUERY_REFRESH)
         type = MessageType::Refresh;
-    else if (parsed.q == QUERY_UPDATE)
-        type = MessageType::UpdateValue;
     else
         throw msgpack::type_error();
 
@@ -300,8 +295,6 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
             parsedReq.fields = &o.val;
         else if (key == KEY_REQ_WANT)
             parsedReq.want = &o.val;
-        else if (key == KEY_VERSION)
-            version = o.val.as<int>();
     }
 
     if (parsedReq.sa) {

tests/dhtrunnertester.cpp

@@ -167,51 +167,4 @@ DhtRunnerTester::testListen() {
     node1.cancelListen(d, tokend);
 }
 
-void
-DhtRunnerTester::testListenLotOfBytes() {
-    std::mutex mutex;
-    std::condition_variable cv;
-    std::atomic_uint valueCount(0);
-    unsigned putCount(0);
-    unsigned putOkCount(0);
-
-    std::string data(10000, 'a');
-
-    auto foo = dht::InfoHash::get("foo");
-    constexpr unsigned N = 50;
-
-    for (unsigned i=0; i<N; i++) {
-        node2.put(foo, data, [&](bool ok) {
-            std::lock_guard<std::mutex> lock(mutex);
-            putCount++;
-            if (ok) putOkCount++;
-            cv.notify_all();
-        });
-    }
-    {
-        std::unique_lock<std::mutex> lk(mutex);
-        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N; }));
-    }
-
-    dht::DhtRunner node3 {};
-    dht::DhtRunner::Config config;
-    config.dht_config.node_config.max_peer_req_per_sec = -1;
-    config.dht_config.node_config.max_req_per_sec = -1;
-    node3.run(42242, config);
-    node3.bootstrap(node1.getBound());
-
-    auto ftokenfoo = node3.listen(foo, [&](const std::shared_ptr<dht::Value>&) {
-        valueCount++;
-        cv.notify_all();
-        return true;
-    });
-
-    {
-        std::unique_lock<std::mutex> lk(mutex);
-        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return valueCount == N; }));
-    }
-
-    node3.cancelListen(foo, ftokenfoo.get());
-}
-
 }  // namespace test

tests/dhtrunnertester.h

@@ -32,7 +32,6 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     CPPUNIT_TEST(testConstructors);
     CPPUNIT_TEST(testGetPut);
     CPPUNIT_TEST(testListen);
-    CPPUNIT_TEST(testListenLotOfBytes);
     CPPUNIT_TEST_SUITE_END();
 
     dht::DhtRunner node1 {};
@@ -58,10 +57,6 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test listen method
      */
     void testListen();
-    /**
-     * Test listen method with lot of datas
-     */
-    void testListenLotOfBytes();
 };
 
 }  // namespace test

tools/dhtchat.cpp

@@ -64,10 +64,10 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         dht.run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.empty())
-            dht.bootstrap(params.bootstrap);
+        if (not params.bootstrap.first.empty())
+            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
 
-        print_node_info(dht.getNodeInfo());
+        print_node_info(dht, params);
         std::cout << "  type 'c {hash}' to join a channel" << std::endl << std::endl;
 
         bool connected {false};

tools/dhtnode.cpp

@@ -114,7 +114,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
 #endif
 )
 {
-    print_node_info(node->getNodeInfo());
+    print_node_info(*node, params);
     std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl << std::endl;
 
 #ifndef WIN32_NATIVE
@@ -143,23 +143,22 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
             print_help();
             continue;
         } else if (op == "ll") {
-            node->getNodeInfo([&](const std::shared_ptr<dht::NodeInfo>& nodeInfo) {
-                print_node_info(*nodeInfo);
-                std::cout << nodeInfo->ongoing_ops << " ongoing operations" << std::endl;
-                std::cout << "IPv4 stats:" << std::endl;
-                std::cout << nodeInfo->ipv4.toString() << std::endl;
-                std::cout << "IPv6 stats:" << std::endl;
-                std::cout << nodeInfo->ipv6.toString() << std::endl;
+            auto nodeInfo = node->getNodeInfo();
+            print_node_info(*node, params);
+            std::cout << nodeInfo.ongoing_ops << " ongoing operations" << std::endl;
+            std::cout << "IPv4 stats:" << std::endl;
+            std::cout << nodeInfo.ipv4.toString() << std::endl;
+            std::cout << "IPv6 stats:" << std::endl;
+            std::cout << nodeInfo.ipv6.toString() << std::endl;
 #ifdef OPENDHT_PROXY_SERVER
-                for (const auto& proxy : proxies) {
-                    std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
-                    if (auto stats = proxy.second->stats())
-                        std::cout << "  " << stats->toString() << std::endl;
-                    else
-                        std::cout << "  (stats not available yet)" << std::endl;                
-                }
+            for (const auto& proxy : proxies) {
+                std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
+                if (auto stats = proxy.second->stats())
+                    std::cout << "  " << stats->toString() << std::endl;
+                else
+                    std::cout << "  (stats not available yet)" << std::endl;                
+            }
 #endif
-            });
             continue;
         } else if (op == "lr") {
             std::cout << "IPv4 routing table:" << std::endl;
@@ -191,10 +190,14 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
             continue;
         } else if (op == "b") {
             iss >> idstr;
-            if (not idstr.empty())
-                node->bootstrap(idstr);
-            else
-                std::cerr << "No service provided. Syntax: b hostname:port" << std::endl;
+            try {
+                auto addr = splitPort(idstr);
+                if (not addr.first.empty() and addr.second.empty())
+                    addr.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
+                node->bootstrap(addr.first.c_str(), addr.second.c_str());
+            } catch (const std::exception& e) {
+                std::cerr << e.what() << std::endl;
+            }
             continue;
         } else if (op == "log") {
             iss >> idstr;
@@ -534,9 +537,9 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         node->run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.empty()) {
-            std::cout << "Bootstrap: " << params.bootstrap << std::endl;
-            node->bootstrap(params.bootstrap);
+        if (not params.bootstrap.first.empty()) {
+            std::cout << "Bootstrap: " << params.bootstrap.first << ":" << params.bootstrap.second << std::endl;
+            node->bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
         }
 
 #ifdef OPENDHT_PROXY_SERVER

tools/dhtproxy_stats.py

@@ -48,10 +48,10 @@ for i in range(80,101):
         
         stats['timestamp'] = str(ts)
 
-        #with open("stats_proxy_" + str(i), "a") as stat_file:
-        #    stat_file.write(str(stats))
-        #    stat_file.write('\n')
+        with open("stats_proxy_" + str(i), "a") as stat_file:
+            stat_file.write(str(stats))
+            stat_file.write('\n')
 
-with open("stats_proxy_total", "w") as stat_file:
+with open("stats_proxy_total", "a") as stat_file:
     stat_file.write(str(stats_total))
     stat_file.write('\n')

tools/dhtscanner.cpp

@@ -91,10 +91,10 @@ main(int argc, char **argv)
         auto dhtConf = getDhtConfig(params);
         dht.run(params.port, dhtConf.first, std::move(dhtConf.second));
 
-        if (not params.bootstrap.empty())
-            dht.bootstrap(params.bootstrap);
+        if (not params.bootstrap.first.empty())
+            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
 
-        print_node_info(dht.getNodeInfo());
+        print_node_info(dht, params);
         std::cout << "Scanning network..." << std::endl;
         auto all_nodes = std::make_shared<NodeSet>();
 

tools/tools_common.h

@@ -124,7 +124,7 @@ struct dht_params {
     bool log {false};
     bool syslog {false};
     std::string logfile {};
-    std::string bootstrap {};
+    std::pair<std::string, std::string> bootstrap {};
     dht::NetId network {0};
     in_port_t port {0};
     in_port_t proxyserver {0};
@@ -191,14 +191,16 @@ getDhtConfig(dht_params& params)
     return {std::move(config), std::move(context)};
 }
 
-void print_node_info(const dht::NodeInfo& info) {
-    std::cout << "OpenDHT node " << info.node_id << " running on ";
-    if (info.bound4 == info.bound6)
-        std::cout << "port " << info.bound4 << std::endl;
+void print_node_info(const dht::DhtRunner& node, const dht_params& params) {
+    std::cout << "OpenDHT node " << node.getNodeId() << " running on ";
+    auto port4 = node.getBoundPort(AF_INET);
+    auto port6 = node.getBoundPort(AF_INET6);
+    if (port4 == port6)
+        std::cout << "port " << port4 << std::endl;
     else
-        std::cout << "IPv4 port " << info.bound4 << ", IPv6 port " << info.bound6 << std::endl;
-    if (info.id)
-        std::cout << "Public key ID " << info.id << std::endl;
+        std::cout << "IPv4 port " << port4 << ", IPv6 port " << port6 << std::endl;
+    if (params.id.first)
+        std::cout << "Public key ID " << node.getId() << std::endl;
 }
 
 static const constexpr struct option long_options[] = {
@@ -289,7 +291,10 @@ parseArgs(int argc, char **argv) {
             params.public_stable = true;
             break;
         case 'b':
-            params.bootstrap = (optarg[0] == '=') ? optarg+1 : optarg;
+            params.bootstrap = dht::splitPort((optarg[0] == '=') ? optarg+1 : optarg);
+            if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
+                params.bootstrap.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
+            }
             break;
         case 'V':
             params.version = true;