From 1e485b1b0ccbecf4f8e4507e88f689d950233ee2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Simon=20D=C3=A9saulniers?= Date: Thu, 10 Mar 2016 15:09:21 -0500 Subject: [PATCH] handling sending requests 3 times before expire --- include/opendht/network_engine.h | 60 +++++++++++++---- src/network_engine.cpp | 106 ++++++++++++++++--------------- 2 files changed, 104 insertions(+), 62 deletions(-) diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 3dbf7e39..8d4e457f 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -114,9 +114,12 @@ public: * information like the reply time for that request. */ struct RequestStatus { + static const constexpr size_t MAX_ATTEMPT_COUNT {3}; + std::shared_ptr node {}; /* the node to whom the request is destined. */ - bool canceled {false}; /* whether the request is canceled before done. */ + bool cancelled {false}; /* whether the request is canceled before done. */ bool completed {false}; /* whether the request is completed. */ + unsigned attempt_count {0}; /* number of attempt to process the request. */ 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. */ @@ -124,8 +127,10 @@ public: RequestStatus() {} RequestStatus(time_point start, time_point reply_time = time_point::min()) : start(start), last_try(start), reply_time(reply_time) {} + bool expired(time_point now) const { - return reply_time < last_try && now > last_try + Node::MAX_RESPONSE_TIME; + return now > last_try + Node::MAX_RESPONSE_TIME and attempt_count >= RequestStatus::MAX_ATTEMPT_COUNT + and not completed; } bool pending(time_point now) const { return reply_time < last_try && now - last_try <= Node::MAX_RESPONSE_TIME; @@ -422,28 +427,59 @@ private: * request is done. */ struct Request { - static const constexpr size_t MAX_ATTEMPT_COUNT {3}; - Request(uint16_t tid, std::shared_ptr node, Blob &&msg, - std::function req_status, uint16_t tid, ParsedMessage&&)> on_done, - std::function req_status, uint16_t tid, bool)> on_expired) : - on_done(on_done), on_expired(on_expired), tid(tid), msg(msg) { + std::function req_status, ParsedMessage&&)> on_done, + std::function req_status, bool)> on_expired) : + on_done(on_done), on_expired(on_expired), tid(tid), msg(msg), status(std::make_shared()) { status->node = node; } - bool expired() const { return attempt_count >= Request::MAX_ATTEMPT_COUNT; } - - std::function req_status, uint16_t tid, ParsedMessage&&)> on_done {}; - std::function req_status, uint16_t tid, bool)> on_expired {}; + std::function req_status, ParsedMessage&&)> on_done {}; + std::function req_status, bool)> on_expired {}; const uint16_t tid {0}; /* the request id. */ Blob msg {}; /* the serialized message. */ std::shared_ptr status {}; /* the request info for DHT layer. */ - unsigned attempt_count {0}; /* number of attempt to process the request. */ }; + void requestStep(std::shared_ptr req) { + if (req->status->completed or req->status->cancelled) + return; + auto now = scheduler.time(); + if (req->status->expired(now)) { + req->on_expired(req->status, false); + requests.erase(req->tid); + return; + } + + send((char*)req->msg.data(), req->msg.size(), + (req->status->node->reply_time >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM, + (sockaddr*)&req->status->node->ss, req->status->node->sslen); + ++req->status->attempt_count; + req->status->last_try = now; + std::weak_ptr wreq = req; + scheduler.add(req->status->last_try + Node::MAX_RESPONSE_TIME, [this,wreq]() { + if (auto req = wreq.lock()) { + DHT_LOG.DEBUG("Resending unreplied request %d", req->tid); + requestStep(req); + } + }); + } + + /** + * Sends a request to a node. RequestStatus::MAX_ATTEMPT_COUNT attempts will + * be made before the request expires. + */ + void sendRequest(Request&& request) { + auto sreq = std::make_shared(std::move(request)); + sreq->status->start = scheduler.time(); + DHT_LOG.DEBUG("New transaction with id %d", request.tid); + requests.emplace(request.tid, sreq); + requestStep(sreq); + } + /** * Generates a new request id, skipping the invalid id. * diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 665f35c9..78fff9c1 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -270,20 +270,22 @@ 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 {tid[2], n, std::move(b), - [=](std::shared_ptr req_status, uint16_t, ParsedMessage&&){ - on_done(req_status, {}); + Request req {tid[2], std::make_shared(InfoHash {}, sa, salen), std::move(b), + [=](std::shared_ptr req_status, ParsedMessage&&){ + if (on_done) { + on_done(req_status, {}); + } }, - [=](std::shared_ptr req_status, uint16_t tid, bool) { /* on expired */ - on_expired(req_status, {}); - requests.erase(tid); + [=](std::shared_ptr req_status, bool) { /* on expired */ + if (on_expired) { + on_expired(req_status, {}); + } } }; auto req_status = req.status; - requests.emplace(tid[2], std::make_shared(std::move(req))); - send(buffer.data(), buffer.size(), 0, (sockaddr*)&n->ss, n->sslen); - out_stats.ping++; - return std::move(req_status); + sendRequest(std::move(req)); + ++out_stats.ping; + return req_status; } void @@ -331,20 +333,21 @@ NetworkEngine::sendFindNode(std::shared_ptr n, const InfoHash& target, wan Blob b {buffer.data(), buffer.data() + buffer.size()}; Request req {tid[2], n, std::move(b), - [=](std::shared_ptr req_status, uint16_t, ParsedMessage&& msg) { /* on done */ - on_done(req_status, deserializeNodesValues(msg)); + [=](std::shared_ptr req_status, ParsedMessage&& msg) { /* on done */ + if (on_done) { + on_done(req_status, deserializeNodesValues(msg)); + } }, - [=](std::shared_ptr req_status, uint16_t tid, bool) { /* on expired */ - on_expired(req_status, {}); - requests.erase(tid); + [=](std::shared_ptr req_status, bool) { /* on expired */ + if (on_expired) { + on_expired(req_status, {}); + } } }; auto req_status = req.status; - requests.emplace(tid[2], std::make_shared(std::move(req))); - 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); + sendRequest(std::move(req)); + ++out_stats.find; + return req_status; } @@ -374,20 +377,21 @@ NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, wa Blob b {buffer.data(), buffer.data() + buffer.size()}; Request req {tid[2], n, std::move(b), - [=](std::shared_ptr req_status, uint16_t, ParsedMessage&& msg) { /* on done */ - on_done(req_status, deserializeNodesValues(msg)); + [=](std::shared_ptr req_status, ParsedMessage&& msg) { /* on done */ + if (on_done) { + on_done(req_status, deserializeNodesValues(msg)); + } }, - [=](std::shared_ptr req_status, uint16_t tid, bool) { /* on expired */ - on_expired(req_status, {}); - requests.erase(tid); + [=](std::shared_ptr req_status, bool) { /* on expired */ + if (on_expired) { + on_expired(req_status, {}); + } } }; auto req_status = req.status; - requests.emplace(tid[2], std::make_shared(std::move(req))); - 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); + sendRequest(std::move(req)); + ++out_stats.get; + return req_status; } NetworkEngine::RequestAnswer @@ -583,20 +587,21 @@ NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, con Blob b {buffer.data(), buffer.data() + buffer.size()}; Request req {tid[2], n, std::move(b), - [=](std::shared_ptr req_status, uint16_t, ParsedMessage&&) { /* on done */ - on_done(req_status, {}); + [=](std::shared_ptr req_status, ParsedMessage&&) { /* on done */ + if (on_done) { + on_done(req_status, {}); + } }, - [=](std::shared_ptr req_status, uint16_t tid, bool) { /* on expired */ - on_expired(req_status, {}); - requests.erase(tid); + [=](std::shared_ptr req_status, bool) { /* on expired */ + if (on_expired) { + on_expired(req_status, {}); + } } }; auto req_status = req.status; - requests.emplace(tid[2], std::make_shared(std::move(req))); - 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); + sendRequest(std::move(req)); + ++out_stats.listen; + return req_status; } void @@ -643,24 +648,25 @@ NetworkEngine::sendAnnounceValue(std::shared_ptr n, const InfoHash& infoha Blob b {buffer.data(), buffer.data() + buffer.size()}; Request req {tid[2], n, std::move(b), - [=](std::shared_ptr req_status, uint16_t, ParsedMessage&& msg) { /* on done */ + [=](std::shared_ptr req_status, ParsedMessage&& msg) { /* on done */ if (msg.value_id == Value::INVALID_ID) { DHT_LOG.DEBUG("Unknown search or announce!"); } else { - on_done(req_status, {}); + if (on_done) { + on_done(req_status, {}); + } } }, - [=](std::shared_ptr req_status, uint16_t tid, bool) { /* on expired */ - on_expired(req_status, {}); - requests.erase(tid); + [=](std::shared_ptr req_status, bool) { /* on expired */ + if (on_expired) { + on_expired(req_status, {}); + } } }; auto req_status = req.status; - requests.emplace(tid[2], std::make_shared(std::move(req))); - 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); + sendRequest(std::move(req)); + ++out_stats.put; + return req_status; } void