[WIP]net: add updateValues to replace Listen operations

Actually, packets are not retransmitted for listen operations. This
leads to some problems when they are already several values present
at one hash. Ideally, packets should be transmitted inside a request
to avoid packet loss and get all datas.

This patch breaks the compatibility with previous versions

include/opendht/network_engine.h

@@ -362,6 +362,37 @@ public:
                            RequestCb&& on_done,
                            RequestExpiredCb&& on_expired,
                            SocketCb&& socket_cb);
+    /**
+     * Send a "updateValues" request to a given node.
+     *
+     * @param n           The node.
+     * @param hash        The storage's hash.
+     * @param query       The query describing filters.
+     * @param token       A security token.
+     * @param previous    The previous request "updateValues" sent to this node.
+     * @param socket      **UNUSED** The socket for further response.
+     *
+     *                    For backward compatibility purpose, sendUpdateValues has to
+     *                    handle creation of the socket. Therefor, you cannot
+     *                    use openSocket yourself. TODO: Once we don't support
+     *                    the old "updateValues" negociation, sendUpdateValues shall not
+     *                    create the socket itself.
+     *
+     * @param on_done     Request callback when the request is completed.
+     * @param on_expired  Request callback when the request expires.
+     * @param socket_cb   Callback to execute each time new updates arrive on
+     *                    the socket.
+     *
+     * @return the request with information concerning its success.
+     */
+    Sp<Request> sendUpdateValues(Sp<Node> n,
+                                 const InfoHash& hash,
+                                 const Query& query,
+                                 const Blob& token,
+                                 Sp<Request> previous,
+                                 RequestCb&& on_done,
+                                 RequestExpiredCb&& on_expired,
+                                 SocketCb&& socket_cb);
     /**
      * Send a "announce" request to a given node.
      *
@@ -490,12 +521,13 @@ 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 ping          {0};
+        unsigned find          {0};
+        unsigned get           {0};
+        unsigned put           {0};
+        unsigned listen        {0};
+        unsigned updateValues  {0};
+        unsigned refresh       {0};
     };
 
 

src/dht.cpp

@@ -605,7 +605,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
             });
         }
         auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        auto new_req = network_engine.sendUpdateValues(n.node, sr->id, *query, n.token, prev_req,
             [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
             { /* on done */
                 if (auto sr = ws.lock()) {

src/net.h

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

src/network_engine.cpp

@@ -605,6 +605,10 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                 sendListenConfirmation(from, msg->tid);
                 break;
             }
+            case MessageType::UpdateValues: {
+                printf("@@@ Receive UpdateValues\n");
+                break;
+            }
             default:
                 break;
             }
@@ -1075,6 +1079,75 @@ NetworkEngine::sendListen(Sp<Node> n,
     return req;
 }
 
+Sp<Request>
+NetworkEngine::sendUpdateValues(Sp<Node> n,
+        const InfoHash& hash,
+        const Query& query,
+        const Blob& token,
+        Sp<Request> previous,
+        RequestCb&& on_done,
+        RequestExpiredCb&& on_expired,
+        SocketCb&& socket_cb)
+{
+    Tid socket;
+    TransId tid (n->getNewTid());
+    if (previous and previous->node == n) {
+        socket = previous->getSocket();
+    } else {
+        if (previous)
+            if (logger_)
+                logger_->e(hash, "[node %s] trying refresh listen contract with wrong node", previous->node->toString().c_str());
+        socket = n->openSocket(std::move(socket_cb));
+    }
+
+    if (not socket) {
+        if (logger_)
+            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(4 + has_query);
+      pk.pack(KEY_REQ_ID);    pk.pack(myid);
+      pk.pack(KEY_REQ_H);     pk.pack(hash);
+      pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
+      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_UPDATE_VALUES);
+    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::UpdateValues, tid.toInt(), n,
+        Blob(buffer.data(), buffer.data() + buffer.size()),
+        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
+            if (on_done)
+                on_done(req_status, {std::forward<ParsedMessage>(msg)});
+        },
+        [=](const Request& req_status, bool done) { /* on expired */
+            if (on_expired)
+                on_expired(req_status, done);
+        },
+        socket
+    );
+    sendRequest(req);
+    ++out_stats.updateValues;
+    return req;
+}
+
 void
 NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
     msgpack::sbuffer buffer;

src/parsed_message.h

@@ -58,6 +58,7 @@ static const std::string KEY_REQ_WANT {"w"};
 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_VALUES {"updateValues"};
 static const std::string QUERY_PUT {"put"};
 static const std::string QUERY_LISTEN {"listen"};
 static const std::string QUERY_REFRESH {"refresh"};
@@ -224,6 +225,8 @@ 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_VALUES)
+        type = MessageType::UpdateValues;
     else
         throw msgpack::type_error();
 

tests/dhtrunnertester.cpp

@@ -167,4 +167,54 @@ 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 {};
+    for (int i = 0; i < 10000; ++i) {
+        data += "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

@@ -57,6 +57,10 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test listen method
      */
     void testListen();
+    /**
+     * Test listen method with lot of datas
+     */
+    void testListenLotOfBytes();
 };
 
 }  // namespace test