From 9faef0427f9c034fa12636c72b5d87e3f5d354d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Simon=20D=C3=A9saulniers?= Date: Fri, 26 Feb 2016 18:25:29 -0500 Subject: [PATCH] send* methods return NetworkEngine::RequestStatus Previously, the request id was returned. This was needed for cancelling requests. Returning std::shared_ptr enables cancelling the requests, checking if a request is done and access to timings concerning this request. This will replace the SearchNode's RequestStatus. --- include/opendht/dht.h | 2 +- include/opendht/network_engine.h | 168 +++++++++++++++++-------------- src/dht.cpp | 11 +- src/network_engine.cpp | 146 +++++++++++---------------- 4 files changed, 157 insertions(+), 170 deletions(-) diff --git a/include/opendht/dht.h b/include/opendht/dht.h index 672e186a..d75e2771 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -868,7 +868,7 @@ private: void processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen); - void onError(std::shared_ptr node, DhtProtocolException e); + void onError(std::shared_ptr node, DhtProtocolException e); /* when our address is reported by a distant peer. */ void onReportedAddr(const InfoHash& id, sockaddr* sa , socklen_t salen); /* when we receive a ping request */ diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 2ada7363..5ad003fa 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -79,11 +79,14 @@ private: * interface for handling sending and receiving packets. The following * parameters specify callbacks for DHT work: * - * @param onPing callback for ping request. - * @param onFindNode callback for "find node" request. - * @param onGetValues callback for "get values" request. - * @param onListen callback for "listen" request. - * @param onAnnounce callback for "announce" request. + * @param onError callback for handling error messages. + * @param onNewNode callback for handling new nodes. + * @param onReportedAddr callback for reporting an our address as seen from the other peer. + * @param onPing callback for ping request. + * @param onFindNode callback for "find node" request. + * @param onGetValues callback for "get values" request. + * @param onListen callback for "listen" request. + * @param onAnnounce callback for "announce" request. */ class NetworkEngine final { public: @@ -101,14 +104,32 @@ public: std::vector> nodes6; }; + /*! + * @class RequestStatus + * @brief Request info for DHT layer. + * @details + * Request info associated to a NetworkEngine request. This enables + * interracting with the NetworkEngine by cancelling a request or extracting + * information like the reply time for that request. + */ + struct RequestStatus { + std::shared_ptr node {}; /* the node to whom the request is destined. */ + bool canceled {false}; /* whether the request is canceled before done. */ + bool completed {false}; /* whether the request is completed. */ + 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. */ + }; + private: /** * @brief when we receive an error message. * - * @param node (type: std::shared_ptr) the node we received the error from. + * @param node (type: std::shared_ptr) the associated request for + * which we got an error; */ - std::function, DhtProtocolException e)> onError; + std::function, DhtProtocolException)> onError; /** * @brief when a new node happens. * @@ -117,12 +138,15 @@ private: * @param id (type: InfoHash) id of the node. * @param saddr (type: sockaddr*) sockaddr* pointer containing address ip information. * @param saddr_len (type: socklen_t) lenght of the sockaddr struct. + * @param confirm (type: int) 1 if the node sent a message, 2 if it sent us a reply. */ std::function(const InfoHash&, const sockaddr*, socklen_t, int)> onNewNode; /** * @brief when an addres is reported from a distant node. * - * @param arg1 Arg description + * @param id (type: InfoHash) id of the node. + * @param saddr (type: sockaddr*) sockaddr* pointer containing address ip information. + * @param saddr_len (type: socklen_t) lenght of the sockaddr struct. */ std::function onReportedAddr; /** @@ -159,12 +183,12 @@ private: * @param node (type: std::shared_ptr) the requesting node. * @param vhash (type: InfoHash) hash of the value of interest. * @param token (type: Blob) security token. - * @param rid (type: size_t) request id. + * @param rid (type: uint16_t) request id. */ std::function, InfoHash&, Blob&, - size_t)> onListen {}; + uint16_t)> onListen {}; /** * @brief on announce request callback. * @@ -186,7 +210,7 @@ public: * * @param success (type: bool) true if no error occured, else false. */ - using RequestCb = std::function, size_t rid, RequestAnswer&&)>; + using RequestCb = std::function, RequestAnswer&&)>; // Config from Dht. struct DhtInfo { @@ -209,7 +233,7 @@ public: onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), myid(info.myid), dht_socket(info.dht_socket), dht_socket6(info.dht_socket6), DHT_LOG(info.DHT_LOG) { - req_ids = std::uniform_int_distribution{1}(rd_device); + transaction_id = std::uniform_int_distribution{1}(rd_device); } virtual ~NetworkEngine() {}; @@ -226,50 +250,47 @@ public: * @param nodes6 The ipv6 closest nodes. * @param values The values to send. */ - void tellListener(const sockaddr* sa, socklen_t salen, size_t rid, InfoHash hash, want_t want, Blob ntoken, + void tellListener(const sockaddr* sa, socklen_t salen, uint16_t rid, InfoHash hash, want_t want, Blob ntoken, std::vector> nodes, std::vector> nodes6, std::vector> values); - /** - * Cancel an ongoing request before it's processed. - * - * @param rid The request id. - * - * @return true if it can be canceled, false otherwise. - */ - bool cancelRequest(size_t rid); - /************** * Requests * **************/ - size_t sendPing(std::shared_ptr n, RequestCb on_done, RequestCb on_expired); - size_t sendPing(sockaddr* n, socklen_t salen, RequestCb on_done, RequestCb on_expired); - size_t sendFindNode(std::shared_ptr n, - const InfoHash& target, - want_t want, - int confirm, - RequestCb on_done, - RequestCb on_expired); - size_t sendGetValues(std::shared_ptr n, - const InfoHash& target, - want_t want, - int confirm, - RequestCb on_done, - RequestCb on_expired); - size_t sendListen(std::shared_ptr n, - const InfoHash& infohash, - const Blob& token, - int confirm, - RequestCb on_done, - RequestCb on_expired); - size_t sendAnnounceValue(std::shared_ptr n, - const InfoHash& infohash, - const Value& v, - time_point created, - const Blob& token, - int confirm, - RequestCb on_done, - RequestCb on_expired); + std::shared_ptr + sendPing(std::shared_ptr n, RequestCb on_done, RequestCb on_expired); + std::shared_ptr + sendPing(sockaddr* n, socklen_t salen, RequestCb on_done, RequestCb on_expired); + std::shared_ptr + 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 + sendAnnounceValue(std::shared_ptr n, + const InfoHash& infohash, + const Value& v, + time_point created, + const Blob& token, + int confirm, + RequestCb on_done, + RequestCb on_expired); /** * Parses a message and calls appropriate callbacks. @@ -309,7 +330,6 @@ private: const int& dht_socket6; const Logger& DHT_LOG; - bool rateLimit(); struct TransPrefix : public std::array { @@ -326,6 +346,8 @@ private: * host order. */ struct TransId final : public std::array { + static const constexpr uint16_t INVALID {0}; + TransId() {} TransId(const std::array& o) { std::copy(o.begin(), o.end(), begin()); } TransId(const TransPrefix prefix, uint16_t seqno = 0) { @@ -383,15 +405,6 @@ private: void msgpack_unpack(msgpack::object o); }; - /** - * When a request has expired, i.e when Request::MAX_ATTEMPT_COUNT attempts - * of processing the request have been made, we consider a node expired. - * Therefor, the requests for this node can be cancelled. - * - * @param rid The request id. - */ - void clearExpiredRequests(size_t rid); - /*! * @class Request * @brief An atomic request destined to a node. @@ -402,26 +415,25 @@ private: */ struct Request { static const constexpr size_t MAX_ATTEMPT_COUNT {3}; - static const constexpr uint32_t INVALID_ID {0}; - Request(uint32_t id, + Request(uint16_t tid, std::shared_ptr node, Blob &&msg, - std::function on_done, - std::function on_expired) : - id(id), node(node), msg(msg), on_done(on_done), on_expired(on_expired) {} + std::function req_status, uint16_t tid, ParsedMessage&&)> on_done, + std::function req_status, uint16_t tid, bool)> on_expired) : + tid(tid), msg(msg), on_done(on_done), on_expired(on_expired) { + status->node = node; + } bool expired() const { return attempt_count >= Request::MAX_ATTEMPT_COUNT; } - std::function on_done {}; - std::function on_expired {}; + std::function req_status, uint16_t tid, ParsedMessage&&)> on_done {}; + std::function req_status, uint16_t tid, bool)> on_expired {}; - uint32_t id {INVALID_ID}; - unsigned attempt_count {0}; /* number of attempt to process the request. */ - time_point start {clock::now()}; /* time when the request is created. */ - time_point last_try {time_point::min()}; /* time of the last attempt to process the request. */ - std::shared_ptr node {}; /* the node to whom the request is destined. */ - Blob msg {}; /* the serialized message. */ + 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. */ }; /** @@ -429,9 +441,9 @@ private: * * @return the new id. */ - uint32_t getNewRequestId() { - ++req_ids; - return req_ids == Request::INVALID_ID ? ++req_ids : req_ids; + uint16_t getNewTid() { + ++transaction_id; + return transaction_id == TransId::INVALID ? ++transaction_id : transaction_id; } struct MessageStats { @@ -484,8 +496,8 @@ private: time_point now; // requests handling - uint32_t req_ids {1}; - std::map> requests; + uint16_t transaction_id {1}; + std::map> requests; MessageStats in_stats {}, out_stats {}; }; diff --git a/src/dht.cpp b/src/dht.cpp index 36c82160..b8701c9c 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -2423,15 +2423,16 @@ Dht::pingNode(const sockaddr *sa, socklen_t salen) } void -Dht::onError(std::shared_ptr node, DhtProtocolException err) { - if (err.getCode() == DhtProtocolException::UNAUTHORIZED) { +Dht::onError(std::shared_ptr status, DhtProtocolException e) { + if (e.getCode() == DhtProtocolException::UNAUTHORIZED) { //TODO: utiliser map de search /*auto esr = findSearch(ttid, from->sa_family);*/ //if (!esr) return; unsigned cleared = 0; - for (auto& sr : searches) { + for (auto& srp : status->node->ss.ss_family == AF_INET ? searches4 : searches6) { + auto sr = srp.second; for (auto& n : sr.nodes) { - if (n.node != node) continue; + if (n.node != status->node) continue; cleared++; n.getStatus.request_time = TIME_INVALID; n.getStatus.reply_time = TIME_INVALID; @@ -2441,7 +2442,7 @@ Dht::onError(std::shared_ptr node, DhtProtocolException err) { } } DHT_LOG.WARN("[node %s %s] token flush (%d searches affected)", - node->id.toString().c_str(), print_addr((sockaddr*)&node->ss, node->sslen).c_str(), cleared); + status->node->id.toString().c_str(), print_addr((sockaddr*)&status->node->ss, status->node->sslen).c_str(), cleared); } } diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 66327c91..dd7dc747 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -38,7 +38,7 @@ const std::string DhtProtocolException::PUT_WRONG_TOKEN {"Put with wrong token"} const std::string DhtProtocolException::PUT_INVALID_ID {"Put with invalid id"}; const std::string NetworkEngine::my_v {"RNG1"}; -const constexpr uint32_t NetworkEngine::Request::INVALID_ID; +const constexpr uint16_t NetworkEngine::TransId::INVALID; static std::mt19937 rd_device {dht::crypto::random_device{}()}; const NetworkEngine::TransPrefix NetworkEngine::TransPrefix::PING = {"pn"}; @@ -49,7 +49,7 @@ const NetworkEngine::TransPrefix NetworkEngine::TransPrefix::LISTEN = {"lt"}; constexpr long unsigned NetworkEngine::MAX_REQUESTS_PER_SEC; void -NetworkEngine::tellListener(const sockaddr *sa, socklen_t salen, size_t rid, InfoHash hash, want_t want, +NetworkEngine::tellListener(const sockaddr *sa, socklen_t salen, uint16_t rid, InfoHash hash, want_t want, Blob ntoken, std::vector> nodes, std::vector> nodes6, std::vector> values) { @@ -114,13 +114,21 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const sockaddr uint16_t ttid = 0; if (msg.type == MessageType::Error or msg.type == MessageType::Reply) { auto node = onNewNode(msg.id, from, fromlen, 2); + Request* req = nullptr; + const auto& reqp = requests.find(msg.tid[2]); + if (reqp != requests.end()) + req = &(*reqp->second); + + if (not req) + throw DhtProtocolException {DhtProtocolException::INVALID_REPLY_TID, "", msg.id}; + switch (msg.type) { case MessageType::Error: { if (msg.error_code == DhtProtocolException::UNAUTHORIZED && msg.id != zeroes && (msg.tid.matches(TransPrefix::ANNOUNCE_VALUES, &ttid) || msg.tid.matches(TransPrefix::LISTEN, &ttid))) { - onError(node, DhtProtocolException {DhtProtocolException::UNAUTHORIZED}); + onError(req->status, DhtProtocolException {DhtProtocolException::UNAUTHORIZED}); } else { DHT_LOG.WARN("[node %s %s] received unknown error message %u", msg.id.toString().c_str(), print_addr(from, fromlen).c_str(), msg.error_code); @@ -129,15 +137,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const sockaddr break; } case MessageType::Reply: - const auto& reqp = requests.find(msg.tid[2]); - Request* req = nullptr; - if (reqp != requests.end()) - req = &(*reqp->second); - - if (not req) - throw DhtProtocolException {DhtProtocolException::INVALID_REPLY_TID, "", msg.id}; - - req->on_done(req->id, std::move(msg)); + req->on_done(req->status, msg.tid[2], std::move(msg)); break; } } else { @@ -232,40 +232,9 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const sockaddr *sa, return sendto(s, buf, len, flags, sa, salen); } -bool -NetworkEngine::cancelRequest(size_t rid) { - // ids are at most 32bit. - if ((rid >> 32) != 0) { - //TODO: utiliser exception? - return false; - } else { - auto rit = requests.find(rid); - if (rit == requests.end()) { - return false; - } else { - requests.erase(rit); - return true; - } - } -} - -void -NetworkEngine::clearExpiredRequests(size_t rid) { - auto reqp = requests.find(rid); - if (reqp != requests.end() and reqp->second->expired()) { - auto& req = reqp->second; - erase_if(requests, [=,&req](std::pair>& rp) { - auto& r = rp.second; - if (r->node->id == req->node->id && r->expired()) { - return true; - } else { return false; } - }); - } -} - -size_t +std::shared_ptr NetworkEngine::sendPing(std::shared_ptr n, RequestCb on_done, RequestCb on_expired) { - auto tid = TransId {TransPrefix::PING}; + auto tid = TransId {TransPrefix::PING, getNewTid()}; msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5); @@ -280,16 +249,17 @@ NetworkEngine::sendPing(std::shared_ptr n, RequestCb on_done, RequestCb on pk.pack(std::string("v")); pk.pack(my_v); Blob b {buffer.data(), buffer.data() + buffer.size()}; - Request req {getNewRequestId(), n, std::move(b), nullptr, - [=](size_t rid, bool s) { /* on expired */ - on_expired(n, rid, {}); - clearExpiredRequests(rid); + Request req {tid[2], n, std::move(b), nullptr, + [=](std::shared_ptr req_status, uint16_t tid, bool s) { /* on expired */ + on_expired(req_status, {}); + requests.erase(tid); } }; + auto req_status = std::make_shared(req.status); requests.emplace(std::move(req)); send(buffer.data(), buffer.size(), 0, (sockaddr*)&n->ss, n->sslen); out_stats.ping++; - return tid[2]; + return std::move(req_status); } void @@ -310,10 +280,10 @@ NetworkEngine::sendPong(const sockaddr* sa, socklen_t salen, TransId tid) { send(buffer.data(), buffer.size(), 0, sa, salen); } -size_t +std::shared_ptr NetworkEngine::sendFindNode(std::shared_ptr n, const InfoHash& target, want_t want, int confirm, RequestCb on_done, RequestCb on_expired) { - auto tid = TransId {TransPrefix::FIND_NODE}; + auto tid = TransId {TransPrefix::FIND_NODE, getNewTid()}; msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5); @@ -336,26 +306,27 @@ NetworkEngine::sendFindNode(std::shared_ptr n, const InfoHash& target, wan Blob b {buffer.data(), buffer.data() + buffer.size()}; - Request req {getNewRequestId(), n, std::move(b), - [=](size_t rid, ParsedMessage&& msg) { /* on done */ - on_done(n, rid, deserializeNodesValues(msg)); + Request req {tid[2], n, std::move(b), + [=](std::shared_ptr req_status, uint16_t tid, ParsedMessage&& msg) { /* on done */ + on_done(req_status, deserializeNodesValues(msg)); }, - [=](size_t rid, bool s) { /* on expired */ - on_expired(n, rid, {}); - clearExpiredRequests(rid); + [=](std::shared_ptr req_status, uint16_t tid, bool s) { /* on expired */ + on_expired(req_status, {}); + requests.erase(tid); } }; + 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); out_stats.find++; - return tid[2]; + return std::move(req_status); } -size_t +std::shared_ptr NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, want_t want, int confirm, RequestCb on_done, RequestCb on_expired) { - auto tid = TransId {TransPrefix::GET_VALUES}; + auto tid = TransId {TransPrefix::GET_VALUES, getNewTid()}; msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5); @@ -377,19 +348,20 @@ NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, wa pk.pack(std::string("v")); pk.pack(my_v); Blob b {buffer.data(), buffer.data() + buffer.size()}; - Request req {getNewRequestId(), n, std::move(b), - [=](size_t rid, ParsedMessage&& msg) { /* on done */ - on_done(n, rid, deserializeNodesValues(msg)); + Request req {tid[2], n, std::move(b), + [=](std::shared_ptr req_status, uint16_t tid, ParsedMessage&& msg) { /* on done */ + on_done(req_status, deserializeNodesValues(msg)); }, - [=](size_t rid, bool s) { /* on expired */ - on_expired(n, rid, {}); - clearExpiredRequests(rid); + [=](std::shared_ptr req_status, uint16_t tid, bool s) { /* on expired */ + on_expired(req_status, {}); + requests.erase(tid); } }; + 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); out_stats.get++; - return tid[2]; + return std::move(req_status); } NetworkEngine::RequestAnswer @@ -565,10 +537,10 @@ NetworkEngine::bufferNodes(sa_family_t af, const InfoHash& id, want_t want, return {std::move(bn4), std::move(bn6)}; } -size_t +std::shared_ptr NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, const Blob& token, int confirm, RequestCb on_done, RequestCb on_expired) { - auto tid = TransId {TransPrefix::LISTEN}; + auto tid = TransId {TransPrefix::LISTEN, getNewTid()}; msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5); @@ -586,19 +558,20 @@ NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, con Blob b {buffer.data(), buffer.data() + buffer.size()}; - Request req {getNewRequestId(), n, std::move(b), - [=](size_t rid, ParsedMessage&& msg) { /* on done */ - on_done(n, rid, {}); + Request req {tid[2], n, std::move(b), + [=](std::shared_ptr req_status, uint16_t tid, ParsedMessage&& msg) { /* on done */ + on_done(req_status, {}); }, - [=](size_t rid, bool s) { /* on expired */ - on_expired(n, rid, {}); - clearExpiredRequests(rid); + [=](std::shared_ptr req_status, uint16_t tid, bool s) { /* on expired */ + on_expired(req_status, {}); + requests.erase(tid); } }; + 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); out_stats.listen++; - return tid[2]; + return std::move(req_status); } void @@ -619,10 +592,10 @@ NetworkEngine::sendListenConfirmation(const sockaddr* sa, socklen_t salen, Trans send(buffer.data(), buffer.size(), 0, sa, salen); } -size_t +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) { - auto tid = TransId {TransPrefix::ANNOUNCE_VALUES}; + auto tid = TransId {TransPrefix::ANNOUNCE_VALUES, getNewTid()}; msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5); @@ -644,19 +617,20 @@ NetworkEngine::sendAnnounceValue(std::shared_ptr n, const InfoHash& infoha pk.pack(std::string("v")); pk.pack(my_v); Blob b {buffer.data(), buffer.data() + buffer.size()}; - Request req {getNewRequestId(), n, std::move(b), - [=](size_t rid, ParsedMessage&& msg) { /* on done */ - on_done(n, rid, {}); + Request req {tid[2], n, std::move(b), + [=](std::shared_ptr req_status, uint16_t tid, ParsedMessage&& msg) { /* on done */ + on_done(req_status, {}); }, - [=](size_t rid, bool s) { /* on expired */ - on_expired(n, rid, {}); - clearExpiredRequests(rid); + [=](std::shared_ptr req_status, uint16_t tid, bool s) { /* on expired */ + on_expired(req_status, {}); + requests.erase(tid); } }; + 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); out_stats.put++; - return tid[2]; + return std::move(req_status); } void