From 31da5b91fa1a03dabe490a27005cd475875b3a75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Blin?= Date: Thu, 23 Apr 2020 12:16:46 -0400 Subject: [PATCH] [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 | 44 ++++++++++++++++--- src/dht.cpp | 2 +- src/net.h | 3 +- src/network_engine.cpp | 73 ++++++++++++++++++++++++++++++++ src/parsed_message.h | 3 ++ tests/dhtrunnertester.cpp | 50 ++++++++++++++++++++++ tests/dhtrunnertester.h | 4 ++ 7 files changed, 171 insertions(+), 8 deletions(-) diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 49b0a039..b42926ef 100644 --- a/include/opendht/network_engine.h +++ b/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 sendUpdateValues(Sp n, + const InfoHash& hash, + const Query& query, + const Blob& token, + Sp 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); 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}; }; diff --git a/src/dht.cpp b/src/dht.cpp index ebc8b104..3219b3f0 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -605,7 +605,7 @@ Dht::searchSynchedNodeListen(const Sp& 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()) { diff --git a/src/net.h b/src/net.h index 15b05b83..17f4805c 100644 --- a/src/net.h +++ b/src/net.h @@ -31,7 +31,8 @@ enum class MessageType { Refresh, Listen, ValueData, - ValueUpdate + ValueUpdate, + UpdateValues }; } /* namespace net */ diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 62de69e3..b0234f18 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -605,6 +605,10 @@ NetworkEngine::process(std::unique_ptr&& 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 n, return req; } +Sp +NetworkEngine::sendUpdateValues(Sp n, + const InfoHash& hash, + const Query& query, + const Blob& token, + Sp 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 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(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(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; diff --git a/src/parsed_message.h b/src/parsed_message.h index edf5af2d..f5ed2cb7 100644 --- a/src/parsed_message.h +++ b/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(); diff --git a/tests/dhtrunnertester.cpp b/tests/dhtrunnertester.cpp index 06d3a321..4ab1c150 100644 --- a/tests/dhtrunnertester.cpp +++ b/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 lock(mutex); + putCount++; + if (ok) putOkCount++; + cv.notify_all(); + }); + } + { + std::unique_lock 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&) { + valueCount++; + cv.notify_all(); + return true; + }); + + { + std::unique_lock lk(mutex); + CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return valueCount == N; })); + } + + node3.cancelListen(foo, ftokenfoo.get()); +} + } // namespace test diff --git a/tests/dhtrunnertester.h b/tests/dhtrunnertester.h index b358d396..0069a5ed 100644 --- a/tests/dhtrunnertester.h +++ b/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