MirrorHub/construct
0
0
Fork 0
mirror of https://github.com/matrix-construct/construct synced 2024-12-26 15:33:54 +01:00
Code Issues Releases Wiki Activity

ircd::net: Move resolver service out to modules/s_resolver.

Browse source
This commit is contained in:
Jason Volk 2018-09-30 20:10:34 -07:00
parent b43b094f2c
commit 05cc6ddf83
7 changed files with 749 additions and 713 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
include/ircd/net/asio.h
View file

@ -39,7 +39,6 @@ namespace ircd::net
#include <ircd/net/socket.h>
#include <ircd/net/acceptor.h>
#include <ircd/net/resolver.h>
namespace ircd
{

5
include/ircd/net/dns.h
View file

@ -28,12 +28,11 @@ namespace ircd::net
struct ircd::net::dns
{
struct opts;
struct resolver;
struct cache;
struct resolver;
struct cache static cache;
struct resolver static *resolver;
struct opts static const opts_default;
struct cache static cache;
using callback = std::function<void (std::exception_ptr, const hostport &, const vector_view<const rfc1035::record *> &)>;
using callback_A_one = std::function<void (std::exception_ptr, const hostport &, const rfc1035::record::A &)>;

2
include/ircd/net/net.h
View file

@ -71,8 +71,6 @@ namespace ircd
struct ircd::net::init
{
std::unique_ptr<struct dns::resolver> resolver;
init();
~init() noexcept;
};

722
ircd/net.cc
View file

@ -35,14 +35,8 @@ ircd::net::wait_close_sockets()
/// Network subsystem initialization
ircd::net::init::init()
:resolver
{
std::make_unique<struct dns::resolver>()
}
{
assert(ircd::ios);
assert(!net::dns::resolver);
dns::resolver = resolver.get();
sslv23_client.set_verify_mode(asio::ssl::verify_peer);
sslv23_client.set_default_verify_paths();
@ -53,8 +47,6 @@ ircd::net::init::~init()
noexcept
{
wait_close_sockets();
assert(net::dns::resolver == resolver.get());
net::dns::resolver = nullptr;
}
///////////////////////////////////////////////////////////////////////////////
@ -2896,11 +2888,6 @@ decltype(ircd::net::dns::cache)
ircd::net::dns::cache
{};
/// Singleton instance of the internal boost resolver wrapper.
decltype(ircd::net::dns::resolver)
ircd::net::dns::resolver
{};
/// Linkage for default opts
decltype(ircd::net::dns::opts_default)
ircd::net::dns::opts_default
@ -2997,7 +2984,6 @@ ircd::net::dns::operator()(const hostport &hp,
const opts &opts,
callback_SRV_one callback)
{
assert(bool(ircd::net::dns::resolver));
operator()(hp, opts, [callback(std::move(callback))]
(std::exception_ptr eptr, const hostport &hp, const vector_view<const rfc1035::record *> rrs)
{
@ -3028,7 +3014,6 @@ ircd::net::dns::operator()(const hostport &hp,
const opts &opts,
callback_A_one callback)
{
assert(bool(ircd::net::dns::resolver));
operator()(hp, opts, [callback(std::move(callback))]
(std::exception_ptr eptr, const hostport &hp, const vector_view<const rfc1035::record *> &rrs)
{
@ -3054,16 +3039,35 @@ ircd::net::dns::operator()(const hostport &hp,
/// Fundamental callback with a vector of abstract resource records.
void
ircd::net::dns::operator()(const hostport &hostport,
const opts &opts,
ircd::net::dns::operator()(const hostport &hp,
const opts &op,
callback cb)
try
{
if(opts.cache_check)
if(cache.get(hostport, opts, cb))
using prototype = void (const hostport &, const opts &, callback &&);
static mods::import<prototype> resolve
{
"s_resolver", "resolve"
};
if(op.cache_check)
if(cache.get(hp, op, cb))
return;
assert(bool(ircd::net::dns::resolver));
(*resolver)(hostport, opts, std::move(cb));
resolve(hp, op, std::move(cb));
}
catch(const mods::unavailable &e)
{
thread_local char buf[128];
log::error
{
log, "Unable to resolve '%s' :%s",
string(buf, hp),
e.what()
};
throw;
}
/// Really assumptional and hacky right now. We're just assuming the SRV
@ -3227,682 +3231,6 @@ ircd::net::dns::cache::for_each(const uint16_t &type,
return function(type, c);
}
///////////////////////////////////////////////////////////////////////////////
//
// net/resolver.h
//
decltype(ircd::net::dns::resolver::servers)
ircd::net::dns::resolver::servers
{
{
{ "name", "ircd.net.dns.resolver.servers" },
{ "default", "4.2.2.1;4.2.2.2;4.2.2.3;4.2.2.4;4.2.2.5;4.2.2.6" },
}, []
{
if(ircd::net::dns::resolver)
ircd::net::dns::resolver->set_servers();
}
};
decltype(ircd::net::dns::resolver::timeout)
ircd::net::dns::resolver::timeout
{
{ "name", "ircd.net.dns.resolver.timeout" },
{ "default", 10000L },
};
decltype(ircd::net::dns::resolver::send_rate)
ircd::net::dns::resolver::send_rate
{
{ "name", "ircd.net.dns.resolver.send_rate" },
{ "default", 60L },
};
decltype(ircd::net::dns::resolver::send_burst)
ircd::net::dns::resolver::send_burst
{
{ "name", "ircd.net.dns.resolver.send_burst" },
{ "default", 8L },
};
decltype(ircd::net::dns::resolver::retry_max)
ircd::net::dns::resolver::retry_max
{
{ "name", "ircd.net.dns.resolver.retry_max" },
{ "default", 4L },
};
ircd::net::dns::resolver::resolver()
:ns{*ircd::ios}
,reply
{
64_KiB // worst-case UDP datagram size
}
,timeout_context
{
"dnsres T", 64_KiB, std::bind(&resolver::timeout_worker, this), context::POST
}
,sendq_context
{
"dnsres S", 64_KiB, std::bind(&resolver::sendq_worker, this), context::POST
}
{
ns.open(ip::udp::v4());
ns.non_blocking(true);
set_servers();
set_handle();
}
ircd::net::dns::resolver::~resolver()
noexcept
{
ns.close();
while(!tags.empty())
{
log::warning
{
log, "Waiting for %zu unfinished DNS resolutions",
tags.size()
};
ctx::sleep(3);
}
sendq_context.interrupt();
timeout_context.interrupt();
assert(tags.empty());
}
__attribute__((noreturn))
void
ircd::net::dns::resolver::sendq_worker()
{
while(1)
{
assert(sendq.empty() || !tags.empty());
dock.wait([this]
{
return !sendq.empty();
});
assert(sendq.size() < 65535);
assert(sendq.size() <= tags.size());
if(tags.size() > size_t(send_burst))
ctx::sleep(milliseconds(send_rate));
const unwind::nominal::assertion na;
assert(!sendq.empty());
const uint16_t next(sendq.front());
sendq.pop_front();
flush(next);
}
}
void
ircd::net::dns::resolver::flush(const uint16_t &next)
try
{
auto &tag
{
tags.at(next)
};
send_query(tag);
}
catch(const std::out_of_range &e)
{
log::error
{
"Queued tag id[%u] is no longer mapped", next
};
}
__attribute__((noreturn))
void
ircd::net::dns::resolver::timeout_worker()
{
while(1)
{
dock.wait([this]
{
return !tags.empty();
});
ctx::sleep(milliseconds(timeout));
check_timeouts(milliseconds(timeout));
}
}
void
ircd::net::dns::resolver::check_timeouts(const milliseconds &timeout)
{
const auto cutoff
{
now<steady_point>() - timeout
};
auto it(begin(tags));
while(it != end(tags))
{
const auto &id(it->first);
auto &tag(it->second);
if(check_timeout(id, tag, cutoff))
it = tags.erase(it);
else
++it;
}
}
bool
ircd::net::dns::resolver::check_timeout(const uint16_t &id,
tag &tag,
const steady_point &cutoff)
{
if(tag.last == steady_point{})
return false;
if(tag.last > cutoff)
return false;
log::warning
{
log, "DNS timeout id:%u on attempt %u", id, tag.tries
};
tag.last = steady_point{};
if(ns.is_open() && tag.tries < size_t(retry_max))
{
submit(tag);
return false;
}
if(!tag.cb)
return true;
// Callback gets a fresh stack off this timeout worker ctx's stack.
ircd::post([this, id, &tag]
{
using boost::system::system_error;
static const error_code ec
{
boost::system::errc::timed_out, boost::system::system_category()
};
// Have to check if the tag is still mapped at this point. It may
// have been removed if a belated reply came in while this closure
// was posting. If so, that's good news and we bail on the timeout.
if(!tags.count(id))
return;
log::error
{
log, "DNS timeout id:%u", id
};
tag.cb(std::make_exception_ptr(system_error{ec}), tag.hp, {});
const auto erased(tags.erase(tag.id));
assert(erased == 1);
});
return false;
}
/// Internal resolver entry interface.
void
ircd::net::dns::resolver::operator()(const hostport &hp,
const opts &opts,
callback &&callback)
{
auto &tag
{
set_tag(hp, opts, std::move(callback))
};
// Escape trunk
const unwind::exceptional untag{[this, &tag]
{
tags.erase(tag.id);
}};
tag.question = make_query(tag.qbuf, tag);
submit(tag);
}
ircd::const_buffer
ircd::net::dns::resolver::make_query(const mutable_buffer &buf,
const tag &tag)
const
{
//TODO: Better deduction
if(tag.hp.service || tag.opts.srv)
{
thread_local char srvbuf[512];
const string_view srvhost
{
make_SRV_key(srvbuf, host(tag.hp), tag.opts)
};
const rfc1035::question question{srvhost, "SRV"};
return rfc1035::make_query(buf, tag.id, question);
}
const rfc1035::question question{host(tag.hp), "A"};
return rfc1035::make_query(buf, tag.id, question);
}
template<class... A>
ircd::net::dns::resolver::tag &
ircd::net::dns::resolver::set_tag(A&&... args)
{
while(tags.size() < 65535)
{
auto id(ircd::rand::integer(1, 65535));
auto it{tags.lower_bound(id)};
if(it != end(tags) && it->first == id)
continue;
it = tags.emplace_hint(it,
std::piecewise_construct,
std::forward_as_tuple(id),
std::forward_as_tuple(std::forward<A>(args)...));
it->second.id = id;
dock.notify_one();
return it->second;
}
throw assertive
{
"Too many DNS queries"
};
}
void
ircd::net::dns::resolver::queue_query(tag &tag)
{
assert(sendq.size() <= tags.size());
sendq.emplace_back(tag.id);
dock.notify_one();
}
void
ircd::net::dns::resolver::submit(tag &tag)
{
const auto rate(milliseconds(send_rate) / server.size());
const auto elapsed(now<steady_point>() - send_last);
if(elapsed >= rate || tags.size() < size_t(send_burst))
send_query(tag);
else
queue_query(tag);
}
void
ircd::net::dns::resolver::send_query(tag &tag)
try
{
assert(!server.empty());
++server_next %= server.size();
const auto &ep
{
server.at(server_next)
};
send_query(ep, tag);
}
catch(const std::out_of_range &)
{
throw error
{
"No DNS servers available for query"
};
}
void
ircd::net::dns::resolver::send_query(const ip::udp::endpoint &ep,
tag &tag)
{
assert(ns.non_blocking());
assert(!empty(tag.question));
const const_buffer &buf{tag.question};
ns.send_to(asio::const_buffers_1(buf), ep);
send_last = now<steady_point>();
tag.last = send_last;
tag.tries++;
}
void
ircd::net::dns::resolver::set_handle()
{
auto handler
{
std::bind(&resolver::handle, this, ph::_1, ph::_2)
};
const asio::mutable_buffers_1 bufs{reply};
ns.async_receive_from(bufs, reply_from, std::move(handler));
}
void
ircd::net::dns::resolver::handle(const error_code &ec,
const size_t &bytes)
noexcept try
{
if(!handle_error(ec))
return;
const unwind reset{[this]
{
set_handle();
}};
if(unlikely(bytes < sizeof(rfc1035::header)))
throw rfc1035::error
{
"Got back %zu bytes < rfc1035 %zu byte header",
bytes,
sizeof(rfc1035::header)
};
char *const reply
{
data(this->reply)
};
rfc1035::header &header
{
*reinterpret_cast<rfc1035::header *>(reply)
};
bswap(&header.qdcount);
bswap(&header.ancount);
bswap(&header.nscount);
bswap(&header.arcount);
const const_buffer body
{
reply + sizeof(header), bytes - sizeof(header)
};
handle_reply(header, body);
}
catch(const std::exception &e)
{
throw assertive
{
"resolver::handle_reply(): %s", e.what()
};
}
void
ircd::net::dns::resolver::handle_reply(const header &header,
const const_buffer &body)
try
{
const auto &id{header.id};
const auto it{tags.find(id)};
if(it == end(tags))
throw error
{
"DNS reply from %s for unrecognized tag id:%u",
string(reply_from),
id
};
auto &tag{it->second};
const unwind untag{[this, &it]
{
tags.erase(it);
}};
assert(tag.tries > 0);
tag.last = steady_point{};
handle_reply(header, body, tag);
}
catch(const std::exception &e)
{
log::error
{
log, "%s", e.what()
};
return;
}
void
ircd::net::dns::resolver::handle_reply(const header &header,
const const_buffer &body,
tag &tag)
try
{
if(unlikely(header.qr != 1))
throw rfc1035::error
{
"Response header is marked as 'Query' and not 'Response'"
};
if(header.qdcount > MAX_COUNT || header.ancount > MAX_COUNT)
throw error
{
"Response contains too many sections..."
};
const_buffer buffer
{
body
};
// Questions are regurgitated back to us so they must be parsed first
thread_local std::array<rfc1035::question, MAX_COUNT> qd;
for(size_t i(0); i < header.qdcount; ++i)
consume(buffer, size(qd.at(i).parse(buffer)));
if(!handle_error(header, qd.at(0), tag))
throw rfc1035::error
{
"protocol error #%u :%s", header.rcode, rfc1035::rcode.at(header.rcode)
};
// Answers are parsed into this buffer
thread_local std::array<rfc1035::answer, MAX_COUNT> an;
for(size_t i(0); i < header.ancount; ++i)
consume(buffer, size(an[i].parse(buffer)));
if(tag.opts.cache_result)
{
// We convert all TTL values in the answers to absolute epoch time
// indicating when they expire. This makes more sense for our caches.
const auto &now{ircd::time()};
for(size_t i(0); i < header.ancount; ++i)
{
const uint &min_ttl(seconds(cache.min_ttl).count());
an[i].ttl = now + std::max(an[i].ttl, min_ttl);
}
}
// The callback to the user will be passed a vector_view of pointers
// to this array. The actual record instances will either be located
// in the cache map or placement-newed to the buffer below.
thread_local const rfc1035::record *record[MAX_COUNT];
// This will be where we place the record instances which are dynamically
// laid out and sized types. 512 bytes is assumed as a soft maximum for
// each RR instance.
thread_local uint8_t recbuf[MAX_COUNT * 512];
size_t i(0);
uint8_t *pos{recbuf};
for(; i < header.ancount; ++i) switch(an[i].qtype)
{
case 1: // A records are inserted into cache
{
if(!tag.opts.cache_result)
{
record[i] = new (pos) rfc1035::record::A(an[i]);
pos += sizeof(rfc1035::record::A);
continue;
}
record[i] = cache.put(qd.at(0), an[i]);
continue;
}
case 5:
{
record[i] = new (pos) rfc1035::record::CNAME(an[i]);
pos += sizeof(rfc1035::record::CNAME);
continue;
}
case 33:
{
if(!tag.opts.cache_result)
{
record[i] = new (pos) rfc1035::record::SRV(an[i]);
pos += sizeof(rfc1035::record::SRV);
continue;
}
record[i] = cache.put(qd.at(0), an[i]);
continue;
}
default:
{
record[i] = new (pos) rfc1035::record(an[i]);
pos += sizeof(rfc1035::record);
continue;
}
}
// Cache no answers here.
if(!header.ancount && tag.opts.cache_result)
cache.put_error(qd.at(0), header.rcode);
if(tag.cb)
{
const vector_view<const rfc1035::record *> records(record, i);
tag.cb(std::exception_ptr{}, tag.hp, records);
}
}
catch(const std::exception &e)
{
// There's no need to flash red to the log for NXDOMAIN which is
// common in this system when probing SRV.
if(unlikely(header.rcode != 3))
log::error
{
log, "resolver tag:%u: %s",
tag.id,
e.what()
};
if(tag.cb)
{
assert(header.rcode != 3 || tag.opts.nxdomain_exceptions);
tag.cb(std::current_exception(), tag.hp, {});
}
}
bool
ircd::net::dns::resolver::handle_error(const header &header,
const rfc1035::question &question,
tag &tag)
{
switch(header.rcode)
{
case 0: // NoError; continue
return true;
case 3: // NXDomain; exception
{
if(!tag.opts.cache_result)
return false;
const auto *record
{
cache.put_error(question, header.rcode)
};
// When the user doesn't want an eptr for nxdomain we just make
// their callback here and then null the cb pointer so it's not
// called again. It is done here because we have a reference to
// the cached error record readily accessible.
if(!tag.opts.nxdomain_exceptions && tag.cb)
{
assert(record);
tag.cb({}, tag.hp, vector_view<const rfc1035::record *>(&record, 1));
tag.cb = {};
}
return false;
}
default: // Unhandled error; exception
return false;
}
}
bool
ircd::net::dns::resolver::handle_error(const error_code &ec)
const
{
using namespace boost::system::errc;
switch(ec.value())
{
case operation_canceled:
return false;
case success:
return true;
default:
throw boost::system::system_error(ec);
}
}
void
ircd::net::dns::resolver::set_servers()
{
const std::string &list(resolver::servers);
set_servers(list);
}
void
ircd::net::dns::resolver::set_servers(const string_view &list)
{
server.clear();
server_next = 0;
tokens(list, ';', [this]
(const hostport &hp)
{
const auto &port
{
net::port(hp) != canon_port? net::port(hp) : uint16_t(53)
};
const ipport ipp
{
host(hp), port
};
add_server(ipp);
});
}
void
ircd::net::dns::resolver::add_server(const ipport &ipp)
{
server.emplace_back(make_endpoint_udp(ipp));
log::debug
{
log, "Adding [%s] as DNS server #%zu",
string(ipp),
server.size()
};
}
///////////////////////////////////////////////////////////////////////////////
//
// net/ipport.h

2
modules/Makefile.am
View file

@ -69,6 +69,7 @@ s_dns_la_SOURCES = s_dns.cc
s_node_la_SOURCES = s_node.cc
s_listen_la_SOURCES = s_listen.cc
s_keys_la_SOURCES = s_keys.cc
s_resolver_la_SOURCES = s_resolver.cc
s_module_LTLIBRARIES = \
s_conf.la \
@ -77,6 +78,7 @@ s_module_LTLIBRARIES = \
s_node.la \
s_listen.la \
s_keys.la \
s_resolver.la \
###
###############################################################################

720
modules/s_resolver.cc Normal file
View file

@ -0,0 +1,720 @@
// Matrix Construct
//
// Copyright (C) Matrix Construct Developers, Authors & Contributors
// Copyright (C) 2016-2018 Jason Volk <jason@zemos.net>
//
// Permission to use, copy, modify, and/or distribute this software for any
// purpose with or without fee is hereby granted, provided that the above
// copyright notice and this permission notice is present in all copies. The
// full license for this software is available in the LICENSE file.
#include <ircd/asio.h>
#include "s_resolver.h"
std::unique_ptr<ircd::net::dns::resolver>
resolver_instance;
ircd::mapi::header
IRCD_MODULE
{
"Server Domain Name Resolver", []
{
resolver_instance = std::make_unique<ircd::net::dns::resolver>();
}, []
{
resolver_instance.reset(nullptr);
}
};
extern "C" void
resolve(const ircd::net::hostport &hp,
const ircd::net::dns::opts &opts,
ircd::net::dns::callback &&callback)
{
if(unlikely(!resolver_instance))
throw ircd::mods::unavailable
{
"Resolver module loaded but the service is unavailable."
};
(*resolver_instance)(hp, opts, std::move(callback));
}
//
// resolver
//
decltype(ircd::net::dns::resolver::servers)
ircd::net::dns::resolver::servers
{
{
{ "name", "ircd.net.dns.resolver.servers" },
{ "default", "4.2.2.1;4.2.2.2;4.2.2.3;4.2.2.4;4.2.2.5;4.2.2.6" },
}, []
{
if(bool(resolver_instance))
resolver_instance->set_servers();
}
};
decltype(ircd::net::dns::resolver::timeout)
ircd::net::dns::resolver::timeout
{
{ "name", "ircd.net.dns.resolver.timeout" },
{ "default", 10000L },
};
decltype(ircd::net::dns::resolver::send_rate)
ircd::net::dns::resolver::send_rate
{
{ "name", "ircd.net.dns.resolver.send_rate" },
{ "default", 60L },
};
decltype(ircd::net::dns::resolver::send_burst)
ircd::net::dns::resolver::send_burst
{
{ "name", "ircd.net.dns.resolver.send_burst" },
{ "default", 8L },
};
decltype(ircd::net::dns::resolver::retry_max)
ircd::net::dns::resolver::retry_max
{
{ "name", "ircd.net.dns.resolver.retry_max" },
{ "default", 4L },
};
//
// resolver::resolver
//
ircd::net::dns::resolver::resolver()
:ns{*ircd::ios}
,reply
{
64_KiB // worst-case UDP datagram size
}
,timeout_context
{
"dnsres T", 64_KiB, std::bind(&resolver::timeout_worker, this), context::POST
}
,sendq_context
{
"dnsres S", 64_KiB, std::bind(&resolver::sendq_worker, this), context::POST
}
{
ns.open(ip::udp::v4());
ns.non_blocking(true);
set_servers();
set_handle();
}
ircd::net::dns::resolver::~resolver()
noexcept
{
ns.close();
while(!tags.empty())
{
log::warning
{
log, "Waiting for %zu unfinished DNS resolutions",
tags.size()
};
ctx::sleep(3);
}
sendq_context.interrupt();
timeout_context.interrupt();
assert(tags.empty());
}
__attribute__((noreturn))
void
ircd::net::dns::resolver::sendq_worker()
{
while(1)
{
assert(sendq.empty() || !tags.empty());
dock.wait([this]
{
return !sendq.empty();
});
assert(sendq.size() < 65535);
assert(sendq.size() <= tags.size());
if(tags.size() > size_t(send_burst))
ctx::sleep(milliseconds(send_rate));
const unwind::nominal::assertion na;
assert(!sendq.empty());
const uint16_t next(sendq.front());
sendq.pop_front();
flush(next);
}
}
void
ircd::net::dns::resolver::flush(const uint16_t &next)
try
{
auto &tag
{
tags.at(next)
};
send_query(tag);
}
catch(const std::out_of_range &e)
{
log::error
{
"Queued tag id[%u] is no longer mapped", next
};
}
__attribute__((noreturn))
void
ircd::net::dns::resolver::timeout_worker()
{
while(1)
{
dock.wait([this]
{
return !tags.empty();
});
ctx::sleep(milliseconds(timeout));
check_timeouts(milliseconds(timeout));
}
}
void
ircd::net::dns::resolver::check_timeouts(const milliseconds &timeout)
{
const auto cutoff
{
now<steady_point>() - timeout
};
auto it(begin(tags));
while(it != end(tags))
{
const auto &id(it->first);
auto &tag(it->second);
if(check_timeout(id, tag, cutoff))
it = tags.erase(it);
else
++it;
}
}
bool
ircd::net::dns::resolver::check_timeout(const uint16_t &id,
tag &tag,
const steady_point &cutoff)
{
if(tag.last == steady_point{})
return false;
if(tag.last > cutoff)
return false;
log::warning
{
log, "DNS timeout id:%u on attempt %u", id, tag.tries
};
tag.last = steady_point{};
if(ns.is_open() && tag.tries < size_t(retry_max))
{
submit(tag);
return false;
}
if(!tag.cb)
return true;
// Callback gets a fresh stack off this timeout worker ctx's stack.
ircd::post([this, id, &tag]
{
using boost::system::system_error;
static const error_code ec
{
boost::system::errc::timed_out, boost::system::system_category()
};
// Have to check if the tag is still mapped at this point. It may
// have been removed if a belated reply came in while this closure
// was posting. If so, that's good news and we bail on the timeout.
if(!tags.count(id))
return;
log::error
{
log, "DNS timeout id:%u", id
};
tag.cb(std::make_exception_ptr(system_error{ec}), tag.hp, {});
const auto erased(tags.erase(tag.id));
assert(erased == 1);
});
return false;
}
/// Internal resolver entry interface.
void
ircd::net::dns::resolver::operator()(const hostport &hp,
const opts &opts,
callback &&callback)
{
auto &tag
{
set_tag(hp, opts, std::move(callback))
};
// Escape trunk
const unwind::exceptional untag{[this, &tag]
{
tags.erase(tag.id);
}};
tag.question = make_query(tag.qbuf, tag);
submit(tag);
}
ircd::const_buffer
ircd::net::dns::resolver::make_query(const mutable_buffer &buf,
const tag &tag)
const
{
//TODO: Better deduction
if(tag.hp.service || tag.opts.srv)
{
thread_local char srvbuf[512];
const string_view srvhost
{
make_SRV_key(srvbuf, host(tag.hp), tag.opts)
};
const rfc1035::question question{srvhost, "SRV"};
return rfc1035::make_query(buf, tag.id, question);
}
const rfc1035::question question{host(tag.hp), "A"};
return rfc1035::make_query(buf, tag.id, question);
}
template<class... A>
ircd::net::dns::resolver::tag &
ircd::net::dns::resolver::set_tag(A&&... args)
{
while(tags.size() < 65535)
{
auto id(ircd::rand::integer(1, 65535));
auto it{tags.lower_bound(id)};
if(it != end(tags) && it->first == id)
continue;
it = tags.emplace_hint(it,
std::piecewise_construct,
std::forward_as_tuple(id),
std::forward_as_tuple(std::forward<A>(args)...));
it->second.id = id;
dock.notify_one();
return it->second;
}
throw assertive
{
"Too many DNS queries"
};
}
void
ircd::net::dns::resolver::queue_query(tag &tag)
{
assert(sendq.size() <= tags.size());
sendq.emplace_back(tag.id);
dock.notify_one();
}
void
ircd::net::dns::resolver::submit(tag &tag)
{
const auto rate(milliseconds(send_rate) / server.size());
const auto elapsed(now<steady_point>() - send_last);
if(elapsed >= rate || tags.size() < size_t(send_burst))
send_query(tag);
else
queue_query(tag);
}
void
ircd::net::dns::resolver::send_query(tag &tag)
try
{
assert(!server.empty());
++server_next %= server.size();
const auto &ep
{
server.at(server_next)
};
send_query(ep, tag);
}
catch(const std::out_of_range &)
{
throw error
{
"No DNS servers available for query"
};
}
void
ircd::net::dns::resolver::send_query(const ip::udp::endpoint &ep,
tag &tag)
{
assert(ns.non_blocking());
assert(!empty(tag.question));
const const_buffer &buf{tag.question};
ns.send_to(asio::const_buffers_1(buf), ep);
send_last = now<steady_point>();
tag.last = send_last;
tag.tries++;
}
void
ircd::net::dns::resolver::set_handle()
{
auto handler
{
std::bind(&resolver::handle, this, ph::_1, ph::_2)
};
const asio::mutable_buffers_1 bufs{reply};
ns.async_receive_from(bufs, reply_from, std::move(handler));
}
void
ircd::net::dns::resolver::handle(const error_code &ec,
const size_t &bytes)
noexcept try
{
if(!handle_error(ec))
return;
const unwind reset{[this]
{
set_handle();
}};
if(unlikely(bytes < sizeof(rfc1035::header)))
throw rfc1035::error
{
"Got back %zu bytes < rfc1035 %zu byte header",
bytes,
sizeof(rfc1035::header)
};
char *const reply
{
data(this->reply)
};
rfc1035::header &header
{
*reinterpret_cast<rfc1035::header *>(reply)
};
bswap(&header.qdcount);
bswap(&header.ancount);
bswap(&header.nscount);
bswap(&header.arcount);
const const_buffer body
{
reply + sizeof(header), bytes - sizeof(header)
};
handle_reply(header, body);
}
catch(const std::exception &e)
{
throw assertive
{
"resolver::handle_reply(): %s", e.what()
};
}
void
ircd::net::dns::resolver::handle_reply(const header &header,
const const_buffer &body)
try
{
const auto &id{header.id};
const auto it{tags.find(id)};
if(it == end(tags))
throw error
{
"DNS reply from %s for unrecognized tag id:%u",
string(reply_from),
id
};
auto &tag{it->second};
const unwind untag{[this, &it]
{
tags.erase(it);
}};
assert(tag.tries > 0);
tag.last = steady_point{};
handle_reply(header, body, tag);
}
catch(const std::exception &e)
{
log::error
{
log, "%s", e.what()
};
return;
}
void
ircd::net::dns::resolver::handle_reply(const header &header,
const const_buffer &body,
tag &tag)
try
{
if(unlikely(header.qr != 1))
throw rfc1035::error
{
"Response header is marked as 'Query' and not 'Response'"
};
if(header.qdcount > MAX_COUNT || header.ancount > MAX_COUNT)
throw error
{
"Response contains too many sections..."
};
const_buffer buffer
{
body
};
// Questions are regurgitated back to us so they must be parsed first
thread_local std::array<rfc1035::question, MAX_COUNT> qd;
for(size_t i(0); i < header.qdcount; ++i)
consume(buffer, size(qd.at(i).parse(buffer)));
if(!handle_error(header, qd.at(0), tag))
throw rfc1035::error
{
"protocol error #%u :%s", header.rcode, rfc1035::rcode.at(header.rcode)
};
// Answers are parsed into this buffer
thread_local std::array<rfc1035::answer, MAX_COUNT> an;
for(size_t i(0); i < header.ancount; ++i)
consume(buffer, size(an[i].parse(buffer)));
if(tag.opts.cache_result)
{
// We convert all TTL values in the answers to absolute epoch time
// indicating when they expire. This makes more sense for our caches.
const auto &now{ircd::time()};
for(size_t i(0); i < header.ancount; ++i)
{
const uint &min_ttl(seconds(cache.min_ttl).count());
an[i].ttl = now + std::max(an[i].ttl, min_ttl);
}
}
// The callback to the user will be passed a vector_view of pointers
// to this array. The actual record instances will either be located
// in the cache map or placement-newed to the buffer below.
thread_local const rfc1035::record *record[MAX_COUNT];
// This will be where we place the record instances which are dynamically
// laid out and sized types. 512 bytes is assumed as a soft maximum for
// each RR instance.
thread_local uint8_t recbuf[MAX_COUNT * 512];
size_t i(0);
uint8_t *pos{recbuf};
for(; i < header.ancount; ++i) switch(an[i].qtype)
{
case 1: // A records are inserted into cache
{
if(!tag.opts.cache_result)
{
record[i] = new (pos) rfc1035::record::A(an[i]);
pos += sizeof(rfc1035::record::A);
continue;
}
record[i] = cache.put(qd.at(0), an[i]);
continue;
}
case 5:
{
record[i] = new (pos) rfc1035::record::CNAME(an[i]);
pos += sizeof(rfc1035::record::CNAME);
continue;
}
case 33:
{
if(!tag.opts.cache_result)
{
record[i] = new (pos) rfc1035::record::SRV(an[i]);
pos += sizeof(rfc1035::record::SRV);
continue;
}
record[i] = cache.put(qd.at(0), an[i]);
continue;
}
default:
{
record[i] = new (pos) rfc1035::record(an[i]);
pos += sizeof(rfc1035::record);
continue;
}
}
// Cache no answers here.
if(!header.ancount && tag.opts.cache_result)
cache.put_error(qd.at(0), header.rcode);
if(tag.cb)
{
const vector_view<const rfc1035::record *> records(record, i);
tag.cb(std::exception_ptr{}, tag.hp, records);
}
}
catch(const std::exception &e)
{
// There's no need to flash red to the log for NXDOMAIN which is
// common in this system when probing SRV.
if(unlikely(header.rcode != 3))
log::error
{
log, "resolver tag:%u: %s",
tag.id,
e.what()
};
if(tag.cb)
{
assert(header.rcode != 3 || tag.opts.nxdomain_exceptions);
tag.cb(std::current_exception(), tag.hp, {});
}
}
bool
ircd::net::dns::resolver::handle_error(const header &header,
const rfc1035::question &question,
tag &tag)
{
switch(header.rcode)
{
case 0: // NoError; continue
return true;
case 3: // NXDomain; exception
{
if(!tag.opts.cache_result)
return false;
const auto *record
{
cache.put_error(question, header.rcode)
};
// When the user doesn't want an eptr for nxdomain we just make
// their callback here and then null the cb pointer so it's not
// called again. It is done here because we have a reference to
// the cached error record readily accessible.
if(!tag.opts.nxdomain_exceptions && tag.cb)
{
assert(record);
tag.cb({}, tag.hp, vector_view<const rfc1035::record *>(&record, 1));
tag.cb = {};
}
return false;
}
default: // Unhandled error; exception
return false;
}
}
bool
ircd::net::dns::resolver::handle_error(const error_code &ec)
const
{
using namespace boost::system::errc;
switch(ec.value())
{
case operation_canceled:
return false;
case success:
return true;
default:
throw boost::system::system_error(ec);
}
}
void
ircd::net::dns::resolver::set_servers()
{
const std::string &list(resolver::servers);
set_servers(list);
}
void
ircd::net::dns::resolver::set_servers(const string_view &list)
{
server.clear();
server_next = 0;
tokens(list, ';', [this]
(const hostport &hp)
{
const auto &port
{
net::port(hp) != canon_port? net::port(hp) : uint16_t(53)
};
const ipport ipp
{
host(hp), port
};
add_server(ipp);
});
}
void
ircd::net::dns::resolver::add_server(const ipport &ipp)
{
server.emplace_back(make_endpoint_udp(ipp));
log::debug
{
log, "Adding [%s] as DNS server #%zu",
string(ipp),
server.size()
};
}

10
include/ircd/net/resolver.h → modules/s_resolver.h
View file

@ -8,21 +8,11 @@
// copyright notice and this permission notice is present in all copies. The
// full license for this software is available in the LICENSE file.
#pragma once
#define HAVE_IRCD_NET_RESOLVER_H
// This file is not included with the IRCd standard include stack because
// it requires symbols we can't forward declare without boost headers. It
// is part of the <ircd/asio.h> stack which can be included in your
// definition file if you need low level access to this resolver API.
/// Internal resolver service
struct ircd::net::dns::resolver
{
struct tag;
using header = rfc1035::header;
static constexpr const size_t &MAX_COUNT{64};
static conf::item<std::string> servers;
static conf::item<milliseconds> timeout;
static conf::item<milliseconds> send_rate;