diff --git a/include/opendht/http.h b/include/opendht/http.h index 9adec063..6060da3b 100644 --- a/include/opendht/http.h +++ b/include/opendht/http.h @@ -162,11 +162,34 @@ public: const bool ssl = false, std::shared_ptr logger = {}); + /** + * Start from previously resolved endpoints while querying the url anyway. + * Callbacks are answered from the cached endpoints without waiting, and the + * query replaces them as soon as it returns, so an address that went stale + * costs a connection attempt rather than a resolution delay. + */ + Resolver(asio::io_context& ctx, + std::string url, + std::vector endpoints, + std::shared_ptr logger = {}); + ~Resolver(); inline const Url& get_url() const { return url_; } - void add_callback(ResolverCb cb, sa_family_t family = AF_UNSPEC); + /** Endpoints known so far, either cached or resolved. May be empty. */ + std::vector get_endpoints() const; + + /** Whether a query is still running, so a different answer may still come. */ + bool is_resolving() const; + + /** + * Register a callback for the resolved endpoints. + * When freshOnly is set, a cached answer is not enough: the callback waits + * for the pending query, so a caller that failed to reach a cached address + * can retry against whatever the system resolver returns. + */ + void add_callback(ResolverCb cb, sa_family_t family = AF_UNSPEC, bool freshOnly = false); std::shared_ptr getLogger() const { return logger_; } @@ -174,17 +197,27 @@ public: private: void resolve(std::string_view host, std::string_view service); + void complete(const asio::error_code& ec, std::vector&& endpoints); + + /** + * A stuck system resolver would otherwise hold every pending request for + * good: Request only arms its own timeout once resolution has answered. + */ + static constexpr std::chrono::seconds RESOLVE_TIMEOUT {15}; mutable std::mutex mutex_; Url url_; asio::error_code ec_; asio::ip::tcp::resolver resolver_; + std::unique_ptr timer_; std::shared_ptr destroyed_; std::vector endpoints_; bool completed_ {false}; + bool resolving_ {false}; std::queue cbs_; + std::queue freshCbs_; std::shared_ptr logger_; }; @@ -317,6 +350,12 @@ private: void connect(std::vector&& endpoints, HandlerCb cb = {}); + /** + * Resolve then connect. Retried once with freshOnly set when the endpoints + * that were handed out came from a cache and turned out to be unreachable. + */ + void resolve_and_connect(bool freshOnly); + void post(); void handle_request(const asio::error_code& ec); diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 96e4d414..c49702c0 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -652,7 +652,21 @@ DhtProxyClient::getProxyInfos() if (logger_) logger_->debug("[proxy:client] [status] sending request"); - auto resolver = std::make_shared(httpContext_, proxyUrl_, logger_); + // Reuse what the previous resolver found: the query still runs, but a + // connection can start at once instead of waiting for the system resolver, + // which is the slowest step when the device wakes up from sleep. + std::vector knownEndpoints; + { + std::lock_guard l(resolverLock_); + if (resolver_) + knownEndpoints = resolver_->get_endpoints(); + } + auto resolver = knownEndpoints.empty() + ? std::make_shared(httpContext_, proxyUrl_, logger_) + : std::make_shared(httpContext_, + proxyUrl_, + std::move(knownEndpoints), + logger_); queryProxyInfo(infoState, resolver, AF_INET); queryProxyInfo(infoState, resolver, AF_INET6); std::lock_guard l(resolverLock_); diff --git a/src/http.cpp b/src/http.cpp index e7639f97..aa897f73 100644 --- a/src/http.cpp +++ b/src/http.cpp @@ -867,12 +867,31 @@ Resolver::Resolver(asio::io_context& ctx, completed_ = true; } +Resolver::Resolver(asio::io_context& ctx, + std::string url, + std::vector endpoints, + std::shared_ptr logger) + : url_(std::move(url)) + , resolver_(ctx) + , destroyed_(std::make_shared(false)) + , endpoints_(std::move(endpoints)) + , logger_(logger) +{ + // Answer from the cached endpoints right away so a connection can start + // without waiting for the system resolver, and query anyway so a stale + // address is replaced as soon as the answer comes back. + completed_ = not endpoints_.empty(); + resolve(url_.host, url_.service.empty() ? url_.protocol : url_.service); +} + Resolver::~Resolver() { decltype(cbs_) cbs; + decltype(freshCbs_) freshCbs; { std::lock_guard lock(mutex_); cbs = std::move(cbs_); + freshCbs = std::move(freshCbs_); } while (not cbs.empty()) { auto cb = cbs.front(); @@ -880,9 +899,29 @@ Resolver::~Resolver() cb(asio::error::operation_aborted, {}); cbs.pop(); } + while (not freshCbs.empty()) { + auto cb = freshCbs.front(); + if (cb) + cb(asio::error::operation_aborted, {}); + freshCbs.pop(); + } *destroyed_ = true; } +std::vector +Resolver::get_endpoints() const +{ + std::lock_guard lock(mutex_); + return endpoints_; +} + +bool +Resolver::is_resolving() const +{ + std::lock_guard lock(mutex_); + return resolving_; +} + void Resolver::cancel() { @@ -905,27 +944,68 @@ filter(const std::vector& epts, sa_family_t family) } void -Resolver::add_callback(ResolverCb cb, sa_family_t family) +Resolver::add_callback(ResolverCb cb, sa_family_t family, bool freshOnly) { std::lock_guard lock(mutex_); + auto wrapped = family == AF_UNSPEC + ? std::move(cb) + : ResolverCb {[cb, family](const asio::error_code& ec, + const std::vector& endpoints) { + if (ec) + cb(ec, endpoints); + else + cb(ec, filter(endpoints, family)); + }}; if (!completed_) - cbs_.emplace(family == AF_UNSPEC ? std::move(cb) - : [cb, family](const asio::error_code& ec, - const std::vector& endpoints) { - if (ec) - cb(ec, endpoints); - else - cb(ec, filter(endpoints, family)); - }); + cbs_.emplace(std::move(wrapped)); + else if (freshOnly and resolving_) + // The cached endpoints already failed this caller, so hold the callback + // until the query that is still running has something else to offer. + freshCbs_.emplace(std::move(wrapped)); else { // Always dispatch the callback on the io_context thread. Invoking it // synchronously would run the caller's continuation (e.g. the whole // connect chain of Request::send) on the calling thread, mutating asio // objects concurrently with the io thread running the io_context. asio::post(resolver_.get_executor(), - [cb = std::move(cb), - ec = ec_, - endpoints = family == AF_UNSPEC ? endpoints_ : filter(endpoints_, family)] { cb(ec, endpoints); }); + [cb = std::move(wrapped), ec = ec_, endpoints = endpoints_] { cb(ec, endpoints); }); + } +} + +void +Resolver::complete(const asio::error_code& ec, std::vector&& endpoints) +{ + decltype(cbs_) cbs; + decltype(freshCbs_) freshCbs; + asio::error_code reportedEc; + std::vector reportedEndpoints; + { + std::lock_guard lock(mutex_); + resolving_ = false; + if (not ec) { + ec_ = ec; + endpoints_ = std::move(endpoints); + } else if (endpoints_.empty()) { + // Nothing cached to fall back on, so the failure has to be reported. + ec_ = ec; + } + completed_ = true; + cbs = std::move(cbs_); + freshCbs = std::move(freshCbs_); + reportedEc = ec_; + reportedEndpoints = endpoints_; + } + while (not cbs.empty()) { + auto cb = cbs.front(); + if (cb) + cb(reportedEc, reportedEndpoints); + cbs.pop(); + } + while (not freshCbs.empty()) { + auto cb = freshCbs.front(); + if (cb) + cb(reportedEc, reportedEndpoints); + freshCbs.pop(); } } @@ -943,30 +1023,45 @@ Resolver::resolve(std::string_view host, std::string_view service) } else if (service == "https"sv) { service = "443"sv; } + + { + std::lock_guard lock(mutex_); + resolving_ = true; + } + + if (not timer_) + timer_ = std::make_unique(resolver_.get_executor()); + timer_->expires_after(RESOLVE_TIMEOUT); + timer_->async_wait([this, destroyed = destroyed_](const asio::error_code& ec) { + if (ec == asio::error::operation_aborted or *destroyed) + return; + { + std::lock_guard lock(mutex_); + if (not resolving_) + return; + // Claim the completion under the lock so the query answering at the + // same moment cannot report a second time. + resolving_ = false; + } + if (logger_) + logger_->error("[http:client] [resolver] timed out resolving {:s}", url_.host); + resolver_.cancel(); + complete(asio::error::timed_out, {}); + }); + resolver_.async_resolve(host, service, [this, destroyed = destroyed_](const asio::error_code& ec, asio::ip::tcp::resolver::results_type endpoints) { if (ec == asio::error::operation_aborted or *destroyed) return; + if (timer_) + timer_->cancel(); if (logger_) { logger_->debug("[http:client] [resolver] got result: {:s}", ec.message()); } - decltype(cbs_) cbs; - { - std::lock_guard lock(mutex_); - ec_ = ec; - endpoints_ = std::vector {endpoints.begin(), - endpoints.end()}; - completed_ = true; - cbs = std::move(cbs_); - } - while (not cbs.empty()) { - auto cb = cbs.front(); - if (cb) - cb(ec, endpoints_); - cbs.pop(); - } + complete(ec, + std::vector {endpoints.begin(), endpoints.end()}); }); } @@ -1444,7 +1539,12 @@ void Request::send() { notify_state_change(State::CREATED); + resolve_and_connect(false); +} +void +Request::resolve_and_connect(bool freshOnly) +{ std::weak_ptr wthis = shared_from_this(); resolver_->add_callback( [wthis](const asio::error_code& ec, std::vector endpoints) { @@ -1458,17 +1558,28 @@ Request::send() } else if (!this_.conn_ or !this_.conn_->is_open()) { this_.connect(std::move(endpoints), [wthis](const asio::error_code& ec) { if (auto sthis = wthis.lock()) { - if (ec) - sthis->terminate(asio::error::not_connected); - else + if (not ec) { sthis->post(); + } else if (sthis->resolver_->is_resolving()) { + // These endpoints came from a cache that no longer + // points anywhere. The query is still running, so + // wait for it rather than fail on a stale address. + if (sthis->logger_) + sthis->logger_->debug( + "[http:request:{:d}] cached endpoints unreachable, awaiting resolution", + sthis->id_); + sthis->resolve_and_connect(true); + } else { + sthis->terminate(asio::error::not_connected); + } } }); } else this_.post(); } }, - family_); + family_, + freshOnly); } void