#pragma once
#include "Loop.h"
#include "Poll.h"
#include "Timer.h"
#include "Request.h"
#include "AddrInfo.h"
#include <map>
#include <ctime>
#include <vector>
#include <iosfwd>
#include <ares.h>
#include <cstdlib>
#include <unordered_map>
#include <panda/string_view.h>
namespace panda { namespace unievent {
struct Resolver : Refcnt, private backend::ITimerImplListener {
static constexpr uint64_t DEFAULT_RESOLVE_TIMEOUT = 5000; // [ms]
static constexpr uint32_t DEFAULT_CACHE_EXPIRATION_TIME = 10*60; // [s]
static constexpr size_t DEFAULT_CACHE_LIMIT = 10000; // [records]
static constexpr uint32_t DEFAULT_QUERY_TIMEOUT = 500; // [ms]
static constexpr uint32_t DEFAULT_WORKERS = 5;
static constexpr size_t MAX_WORKER_POLLS = 3;
struct Request;
using RequestSP = iptr<Request>;
using resolve_fptr = void(const AddrInfo&, const std::error_code&, const RequestSP&);
using resolve_fn = function<resolve_fptr>;
struct Config {
uint32_t cache_expiration_time;
size_t cache_limit;
uint32_t query_timeout;
uint32_t workers;
Config (uint32_t exptime = DEFAULT_CACHE_EXPIRATION_TIME, size_t limit = DEFAULT_CACHE_LIMIT,
uint32_t query_timeout = DEFAULT_QUERY_TIMEOUT, uint32_t workers = DEFAULT_WORKERS)
: cache_expiration_time(exptime), cache_limit(limit), query_timeout(query_timeout), workers(workers) {}
};
struct CachedAddress {
CachedAddress (const AddrInfo& ai, std::time_t update_time = std::time(0)) : address(ai), update_time(update_time) {}
bool expired (time_t now, time_t expiration_time) const { return update_time + expiration_time < now; }
AddrInfo address;
std::time_t update_time;
};
struct CacheKey : Refcnt {
CacheKey (const string& node, const string& service = {}, const AddrInfoHints& hints = {}) : node(node), service(service), hints(hints) {}
bool operator== (const CacheKey& other) const {
return node == other.node && service == other.service && hints == other.hints;
}
string node;
string service;
AddrInfoHints hints;
};
struct CacheHash {
template <class T> inline void hash_combine (std::size_t& seed, const T& v) const {
seed ^= std::hash<T>()(v) + 0x9e3779b9 + (seed << 6) + (seed >> 2);
}
std::size_t operator() (const CacheKey& p) const {
std::size_t seed = 0;
hash_combine(seed, p.node);
hash_combine(seed, p.service);
hash_combine(seed, p.hints.flags);
hash_combine(seed, p.hints.family);
hash_combine(seed, p.hints.socktype);
hash_combine(seed, p.hints.protocol);
return seed;
}
};
struct Cache : std::unordered_map<const CacheKey, CachedAddress, CacheHash> {
using Super = std::unordered_map<const CacheKey, CachedAddress, CacheHash>;
void mark_bad_address (const CacheKey&, const net::SockAddr&);
};
static ResolverSP create_loop_resolver (const LoopSP& loop);
static void disable_loop_resolver (Resolver*);
Resolver (const LoopSP& loop = Loop::default_loop(), uint32_t exptime = DEFAULT_CACHE_EXPIRATION_TIME, size_t limit = DEFAULT_CACHE_LIMIT)
: Resolver(loop, Config(exptime, limit)) {}
Resolver (const LoopSP& loop, const Config&);
Resolver (Resolver& other) = delete;
Resolver& operator= (Resolver& other) = delete;
LoopSP loop () const { return _loop; }
RequestSP resolve ();
RequestSP resolve (string node, resolve_fn callback, uint64_t timeout = DEFAULT_RESOLVE_TIMEOUT);
virtual void resolve (const RequestSP&);
virtual void reset ();
AddrInfo find (const string& node, const string& service = {}, const AddrInfoHints& hints = {});
uint32_t cache_expiration_time () const { return cfg.cache_expiration_time; }
size_t cache_limit () const { return cfg.cache_limit; }
size_t queue_size () const { return queue.size(); }
Cache& cache () { return _cache; }
void cache_expiration_time (uint32_t val) { cfg.cache_expiration_time = val; }
void cache_limit (size_t val) {
cfg.cache_limit = val;
if (_cache.size() > val) _cache.clear();
}
protected:
virtual void on_resolve (const AddrInfo&, const std::error_code&, const RequestSP&);
~Resolver ();
private:
using BTimer = backend::TimerImpl;
using BPoll = backend::PollImpl;
struct Worker : private backend::IPollImplListener {
Worker (Resolver*);
virtual ~Worker ();
void on_sockstate (sock_t sock, int read, int write);
void resolve (const RequestSP&);
void on_resolve (int status, int timeouts, ares_addrinfo* ai);
void finish_resolve (const AddrInfo&, const std::error_code& err);
void cancel ();
void handle_poll (int, const std::error_code&) override;
void rethrow_exception ();
using Polls = std::map<sock_t, BPoll*>;
Resolver* resolver;
ares_channel channel;
Polls polls;
RequestSP request;
bool ares_async;
std::exception_ptr exc;
};
using Requests = IntrusiveChain<RequestSP>;
using Workers = std::vector<std::unique_ptr<Worker>>;
Loop* _loop;
LoopSP _loop_hold;
Config cfg;
BTimer* dns_roll_timer;
Workers workers;
Requests queue;
Requests cache_delayed;
Cache _cache;
Resolver (const Config&, Loop*);
void add_worker ();
void resolve_localhost (const RequestSP&);
void finish_resolve (const RequestSP&, const AddrInfo&, const std::error_code&);
void handle_timer () override;
friend Request; friend Worker;
};
struct Resolver::Request : Refcnt, IntrusiveChainNode<Resolver::RequestSP>, AllocatedObject<Resolver::Request> {
CallbackDispatcher<resolve_fptr> event;
Request (const ResolverSP& r = {});
const ResolverSP& resolver () const { return _resolver; }
RequestSP node (string val) { _node = val; return this; }
RequestSP service (string val) { _service = val; return this; }
RequestSP port (uint16_t val) { _port = val; return this; }
RequestSP hints (const AddrInfoHints& val) { _hints = val; return this; }
RequestSP on_resolve (const resolve_fn& val) { event.add(val); return this; }
RequestSP use_cache (bool val) { _use_cache = val; return this; }
RequestSP timeout (uint64_t val) { _timeout = val; return this; }
RequestSP run () {
RequestSP self = this;
_resolver->resolve(self);
return self;
}
void cancel (const std::error_code& = make_error_code(std::errc::operation_canceled));
protected:
~Request ();
private:
friend Resolver;
LoopSP loop; // keep loop (for loop resolvers where resolver doesn't have strong ref to loop)
ResolverSP _resolver; // keep resolver
string _node;
string _service;
uint16_t _port;
AddrInfoHints _hints;
resolve_fn _callback;
bool _use_cache;
uint64_t _timeout;
Worker* worker;
TimerSP timer;
uint64_t delayed;
bool running;
bool queued;
};
inline Resolver::RequestSP Resolver::resolve () { return new Request(this); }
inline Resolver::RequestSP Resolver::resolve (string node, resolve_fn callback, uint64_t timeout) {
return resolve()->node(node)->on_resolve(callback)->timeout(timeout)->run();
}
std::ostream& operator<< (std::ostream&, const Resolver::CacheKey&);
}}