From 74b265641dab405779afcae7a33a64252eb44e03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Simon=20D=C3=A9saulniers?= Date: Tue, 1 Mar 2016 12:11:36 -0500 Subject: [PATCH] checking UDP replytime in network engine --- include/opendht/dht.h | 4 ++-- include/opendht/network_engine.h | 28 ++++++++++++++++++---------- include/opendht/scheduler.h | 16 ++++++++++++---- src/dht.cpp | 1 - src/network_engine.cpp | 21 +++++++++++++-------- 5 files changed, 45 insertions(+), 25 deletions(-) diff --git a/include/opendht/dht.h b/include/opendht/dht.h index 61443232..cb9f0544 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -31,6 +31,7 @@ THE SOFTWARE. #include "value.h" #include "utils.h" #include "network_engine.h" +#include "scheduler.h" #include #include @@ -346,8 +347,6 @@ private: static constexpr std::chrono::seconds REANNOUNCE_MARGIN {5}; - static constexpr std::chrono::seconds UDP_REPLY_TIME {15}; - /* The maximum number of nodes that we snub. There is probably little reason to increase this value. */ static constexpr unsigned BLACKLISTED_MAX {10}; @@ -766,6 +765,7 @@ private: std::unique_ptr network_engine {}; // timing + std::shared_ptr scheduler; time_point now; time_point mybucket_grow_time {time_point::min()}, mybucket6_grow_time {time_point::min()}; time_point expire_stuff_time {time_point::min()}; diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 5ad003fa..119ee0b2 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -29,6 +29,7 @@ THE SOFTWARE. #include "value.h" #include "infohash.h" #include "node.h" +#include "scheduler.h" #include "utils.h" #include "rng.h" @@ -119,6 +120,13 @@ public: time_point start {time_point::min()}; /* time when the request is created. */ time_point last_try {time_point::min()}; /* time of the last attempt to process the request. */ time_point reply_time {time_point::min()}; /* time when we received the response from the node. */ + + bool expired(time_point now) const { + return reply_time < last_try && now > last_try + Node::MAX_RESPONSE_TIME; + } + bool pending(time_point now) const { + return reply_time < last_try && now - last_try <= Node::MAX_RESPONSE_TIME; + } }; private: @@ -220,7 +228,7 @@ public: const Logger& DHT_LOG; }; - NetworkEngine(DhtInfo info, + NetworkEngine(DhtInfo info, std::shared_ptr scheduler, decltype(NetworkEngine::onError) onError, decltype(NetworkEngine::onNewNode) onNewNode, decltype(NetworkEngine::onReportedAddr) onReportedAddr, @@ -229,9 +237,9 @@ public: decltype(NetworkEngine::onGetValues) onGetValues, decltype(NetworkEngine::onListen) onListen, decltype(NetworkEngine::onAnnounce) onAnnounce) : - onError(onError), onNewNode(onNewNode), onReportedAddr(onReportedAddr), onPing(onPing), onFindNode(onFindNode), - onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), myid(info.myid), - dht_socket(info.dht_socket), dht_socket6(info.dht_socket6), DHT_LOG(info.DHT_LOG) + myid(info.myid), dht_socket(info.dht_socket), dht_socket6(info.dht_socket6),DHT_LOG(info.DHT_LOG), + scheduler(scheduler), onError(onError), onNewNode(onNewNode), onReportedAddr(onReportedAddr), onPing(onPing), + onFindNode(onFindNode), onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce) { transaction_id = std::uniform_int_distribution{1}(rd_device); } @@ -265,21 +273,18 @@ public: sendFindNode(std::shared_ptr n, const InfoHash& target, want_t want, - int confirm, RequestCb on_done, RequestCb on_expired); std::shared_ptr sendGetValues(std::shared_ptr n, const InfoHash& target, want_t want, - int confirm, RequestCb on_done, RequestCb on_expired); std::shared_ptr sendListen(std::shared_ptr n, const InfoHash& infohash, const Blob& token, - int confirm, RequestCb on_done, RequestCb on_expired); std::shared_ptr @@ -288,7 +293,6 @@ public: const Value& v, time_point created, const Blob& token, - int confirm, RequestCb on_done, RequestCb on_expired); @@ -321,7 +325,9 @@ private: static const constexpr size_t NODE4_INFO_BUF_LEN {26}; /* the length of a node info buffer in ipv6 format */ static const constexpr size_t NODE6_INFO_BUF_LEN {38}; - /* TODO: ???? */ + /* TODO */ + static constexpr std::chrono::seconds UDP_REPLY_TIME {15}; + /* TODO */ static const std::string my_v; /* DHT info */ @@ -430,7 +436,7 @@ private: std::function req_status, uint16_t tid, ParsedMessage&&)> on_done {}; std::function req_status, uint16_t tid, bool)> on_expired {}; - const uint16_t tid {0}; /* the request id. */ + const uint16_t tid {0}; /* the request id. */ std::shared_ptr status {}; /* the request info for DHT layer. */ unsigned attempt_count {0}; /* number of attempt to process the request. */ Blob msg {}; /* the serialized message. */ @@ -499,6 +505,8 @@ private: uint16_t transaction_id {1}; std::map> requests; MessageStats in_stats {}, out_stats {}; + + std::shared_ptr scheduler; }; } diff --git a/include/opendht/scheduler.h b/include/opendht/scheduler.h index 49b4e658..45ba5c24 100644 --- a/include/opendht/scheduler.h +++ b/include/opendht/scheduler.h @@ -62,21 +62,29 @@ public: /** * Runs the jobs to do up to now. * - * @param now The time reference to "now". + * @return The time reference to "now". */ - void run(time_point now) { + time_point run() { + now = clock::now(); for (auto t = timers.begin(); t != timers.end(); ) { if (t->first > now) - return; + return now; t->second->do_(); t = timers.erase(t); } + return now; } - time_point getNextJobTime() { return timers.begin()->first; } + inline time_point getNextJobTime() { return timers.begin()->first; } + /** + * Accessors for the common time reference used for synchronizing + * operations. + */ + inline time_point time() const { return now; } private: + time_point now {time_point::min()}; std::multimap> timers {}; /* the jobs ordered by time */ }; diff --git a/src/dht.cpp b/src/dht.cpp index 10b2f957..4e13aaa0 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -104,7 +104,6 @@ constexpr std::chrono::minutes Dht::MAX_STORAGE_MAINTENANCE_EXPIRE_TIME; constexpr std::chrono::minutes Dht::SEARCH_EXPIRE_TIME; constexpr std::chrono::seconds Dht::LISTEN_EXPIRE_TIME; constexpr std::chrono::seconds Dht::REANNOUNCE_MARGIN; -constexpr std::chrono::seconds Dht::UDP_REPLY_TIME; void Dht::setLoggers(LogMethod&& error, LogMethod&& warn, LogMethod&& debug) diff --git a/src/network_engine.cpp b/src/network_engine.cpp index dd7dc747..9f2c8170 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -37,6 +37,7 @@ const std::string DhtProtocolException::PUT_NO_INFOHASH {"Put with no info_hash" const std::string DhtProtocolException::PUT_WRONG_TOKEN {"Put with wrong token"}; const std::string DhtProtocolException::PUT_INVALID_ID {"Put with invalid id"}; +constexpr std::chrono::seconds NetworkEngine::UDP_REPLY_TIME; const std::string NetworkEngine::my_v {"RNG1"}; const constexpr uint16_t NetworkEngine::TransId::INVALID; static std::mt19937 rd_device {dht::crypto::random_device{}()}; @@ -281,7 +282,7 @@ NetworkEngine::sendPong(const sockaddr* sa, socklen_t salen, TransId tid) { } std::shared_ptr -NetworkEngine::sendFindNode(std::shared_ptr n, const InfoHash& target, want_t want, int confirm, +NetworkEngine::sendFindNode(std::shared_ptr n, const InfoHash& target, want_t want, RequestCb on_done, RequestCb on_expired) { auto tid = TransId {TransPrefix::FIND_NODE, getNewTid()}; msgpack::sbuffer buffer; @@ -317,14 +318,15 @@ NetworkEngine::sendFindNode(std::shared_ptr n, const InfoHash& target, wan }; auto req_status = std::make_shared(req.status); requests.emplace(std::move(req)); - send(buffer.data(), buffer.size(), confirm ? 0 : MSG_CONFIRM, (sockaddr*)&n->ss, n->sslen); + send(buffer.data(), buffer.size(), (n->reply_time >= scheduler->time() - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM, + (sockaddr*)&n->ss, n->sslen); out_stats.find++; return std::move(req_status); } std::shared_ptr -NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, want_t want, int confirm, +NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, want_t want, RequestCb on_done, RequestCb on_expired) { auto tid = TransId {TransPrefix::GET_VALUES, getNewTid()}; msgpack::sbuffer buffer; @@ -359,7 +361,8 @@ NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, wa }; auto req_status = std::make_shared(req.status); requests.emplace(std::move(req)); - send(buffer.data(), buffer.size(), confirm ? 0 : MSG_CONFIRM, (sockaddr*)&n->ss, n->sslen); + send(buffer.data(), buffer.size(), (n->reply_time >= scheduler->time() - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM, + (sockaddr*)&n->ss, n->sslen); out_stats.get++; return std::move(req_status); } @@ -538,7 +541,7 @@ NetworkEngine::bufferNodes(sa_family_t af, const InfoHash& id, want_t want, } std::shared_ptr -NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, const Blob& token, int confirm, +NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, const Blob& token, RequestCb on_done, RequestCb on_expired) { auto tid = TransId {TransPrefix::LISTEN, getNewTid()}; msgpack::sbuffer buffer; @@ -569,7 +572,8 @@ NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, con }; auto req_status = std::make_shared(req.status); requests.emplace(std::move(req)); - send(buffer.data(), buffer.size(), confirm ? 0 : MSG_CONFIRM, (sockaddr*)&n->ss, n->sslen); + send(buffer.data(), buffer.size(), (n->reply_time >= scheduler->time() - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM, + (sockaddr*)&n->ss, n->sslen); out_stats.listen++; return std::move(req_status); } @@ -594,7 +598,7 @@ NetworkEngine::sendListenConfirmation(const sockaddr* sa, socklen_t salen, Trans std::shared_ptr NetworkEngine::sendAnnounceValue(std::shared_ptr n, const InfoHash& infohash, const Value& value, time_point created, - const Blob& token, int confirm, RequestCb on_done, RequestCb on_expired) { + const Blob& token, RequestCb on_done, RequestCb on_expired) { auto tid = TransId {TransPrefix::ANNOUNCE_VALUES, getNewTid()}; msgpack::sbuffer buffer; msgpack::packer pk(&buffer); @@ -628,7 +632,8 @@ NetworkEngine::sendAnnounceValue(std::shared_ptr n, const InfoHash& infoha }; auto req_status = std::make_shared(req.status); requests.emplace(std::move(req)); - send(buffer.data(), buffer.size(), confirm ? 0 : MSG_CONFIRM, (sockaddr*)&n->ss, n->sslen); + send(buffer.data(), buffer.size(), (n->reply_time >= scheduler->time() - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM, + (sockaddr*)&n->ss, n->sslen); out_stats.put++; return std::move(req_status); }