dns.cc

/*
 * This file is open source software, licensed to you under the terms
 * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
 * distributed with this work for additional information regarding copyright
 * ownership.  You may not use this file except in compliance with the License.
 *
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
/*
 * Copyright 2016 Cloudius Systems
 */

#include <chrono>
#include <experimental/string_view>

#include <c-ares/ares.h>

#include "ip.hh"
#include "api.hh"
#include "dns.hh"
#include "core/sstring.hh"
#include "core/timer.hh"
#include "core/reactor.hh"
#include "core/gate.hh"
#include "util/log.hh"

namespace seastar {

static logger dns_log("dns_resolver");

class ares_error_category : public std::error_category {
public:
    constexpr ares_error_category() noexcept : std::error_category{} {}
    const char * name() const noexcept {
        return "C-Ares";
    }
    std::string message(int error) const {
        switch (error) {
            /* Server error codes (ARES_ENODATA indicates no relevant answer) */
            case ARES_ENODATA: return "No data";
            case ARES_EFORMERR: return "Form error";
            case ARES_ESERVFAIL: return "Server failure";
            case ARES_ENOTFOUND: return "Not found";
            case ARES_ENOTIMP: return "Not implemented";
            case ARES_EREFUSED: return "Refused";

            /* Locally generated error codes */
            case ARES_EBADQUERY: return "Bad query";
            case ARES_EBADNAME: return "Bad name";
            case ARES_EBADFAMILY: return "Bad family";
            case ARES_EBADRESP: return "Bad response";
            case ARES_ECONNREFUSED :return "Connection refused";
            case ARES_ETIMEOUT: return "Timeout";
            case ARES_EOF: return "EOF";
            case ARES_EFILE: return "File error";
            case ARES_ENOMEM: return "No memory";
            case ARES_EDESTRUCTION: return "Destruction";
            case ARES_EBADSTR: return "Bad string";

            /* ares_getnameinfo error codes */
            case ARES_EBADFLAGS: return "Invalid flags";

            /* ares_getaddrinfo error codes */
            case ARES_ENONAME: return "No name";
            case ARES_EBADHINTS: return "Bad hints";

            /* Uninitialized library error code */
            case ARES_ENOTINITIALIZED: return "Not initialized";

            /* ares_library_init error codes */
            case ARES_ELOADIPHLPAPI: return "Load PHLPAPI";
            case ARES_EADDRGETNETWORKPARAMS: return "Get network parameters";

            /* More error codes */
            case ARES_ECANCELLED: return "Cancelled";
            default:
            return "Unknown error";
        }
    }
};

static const ares_error_category ares_errorc;

static void check_ares_error(int error) {
    if (error != ARES_SUCCESS) {
        throw std::system_error(error, ares_errorc);
    }
}

struct ares_initializer {
    ares_initializer() {
        check_ares_error(ares_library_init(0));
    }
    ~ares_initializer() {
        ares_library_cleanup();
    }
};

class net::dns_resolver::impl
    : public enable_shared_from_this<impl>
{
public:
    impl(network_stack& stack, const options& opts)
        : _stack(stack)
        , _timeout(opts.timeout ? *opts.timeout : std::chrono::milliseconds(5000) /* from ares private */)
        , _timer(std::bind(&impl::poll_sockets, this))
    {
        static const ares_initializer a_init;

        // this can "block" ever so slightly, because it will
        // look in resolv.conf etc for query setup. We could
        // do this ourselves, and instead set ares options
        // here, but it seems more error prone (me parsing
        // resolv.conf -> hah!)
        ares_options a_opts = { 0, };

        // For now, use the default "fb" query order
        // (set explicitly lest we forget).
        // We only do querying dns server really async.
        // Reading hosts files is doen by c-ares internally
        // and with normal fread calls. Thus they theorectically
        // block. This can potentially be an issue for some application
        // and if so, we need to revisit this. For now, assume
        // it won't block us in any measurable way.
        char buf[3] = "fb";
        a_opts.lookups = buf; // only net
        // Always set the timeout
        a_opts.timeout = _timeout.count();
        int flags = ARES_OPT_LOOKUPS|ARES_OPT_TIMEOUTMS;

        if (opts.use_tcp_query && *opts.use_tcp_query) {
            a_opts.flags = ARES_FLAG_USEVC | ARES_FLAG_PRIMARY;
            flags |= ARES_OPT_FLAGS;
        }
        std::vector<in_addr> addr_tmp;
        if (opts.servers) {
            std::transform(opts.servers->begin(), opts.servers->end(), std::back_inserter(addr_tmp), [](const inet_address& a) {
                if (a.in_family() != inet_address::family::INET) {
                    throw std::invalid_argument("Servers must be ipv4 addresses");
                }
                in_addr in = a;
                return in;
            });
            a_opts.servers = addr_tmp.data();
            a_opts.nservers = int(addr_tmp.size());
            flags |= ARES_OPT_SERVERS;
        }
        std::vector<const char *> dom_tmp;
        if (opts.domains) {
            std::transform(opts.domains->begin(), opts.domains->end(), std::back_inserter(dom_tmp), [](const sstring& s) {
                return s.data();
            });
            a_opts.domains = const_cast<char **>(dom_tmp.data());
            a_opts.ndomains = int(dom_tmp.size());
            flags |= ARES_OPT_DOMAINS;
        }
        if (opts.tcp_port) {
            a_opts.tcp_port = *opts.tcp_port;
            flags |= ARES_OPT_TCP_PORT;
        }
        if (opts.udp_port) {
            a_opts.udp_port = *opts.udp_port;
            flags |= ARES_OPT_UDP_PORT;
        }

        check_ares_error(ares_init_options(&_channel, &a_opts, flags));

        static auto get_impl = [](void * p) { return reinterpret_cast<impl *>(p); };
        static const ares_socket_functions callbacks = {
            [](int af, int type, int protocol, void * p) { return get_impl(p)->do_socket(af, type, protocol); },
            [](ares_socket_t s, void * p) { return get_impl(p)->do_close(s); },
            [](ares_socket_t s, const struct sockaddr * addr, socklen_t len, void * p) { return get_impl(p)->do_connect(s, addr, len); },
            [](ares_socket_t s, void * dst, size_t len, int flags, struct sockaddr * addr, socklen_t * alen, void * p) {
                return get_impl(p)->do_recvfrom(s, dst, len, flags, addr, alen);
            },
            [](ares_socket_t s, const struct iovec * vec, int len, void * p) {
                return get_impl(p)->do_sendv(s, vec, len);
            },
        };

        ares_set_socket_functions(_channel, &callbacks, this);

        // just in case you need printf-debug.
        // dns_log.set_level(log_level::trace);
    }
    ~impl() {
        _timer.cancel();
        if (_channel) {
            ares_destroy(_channel);
        }
    }

    future<inet_address> resolve_name(sstring name, inet_address::family family) {
        return get_host_by_name(std::move(name), family).then([](hostent h) {
            return make_ready_future<inet_address>(h.addr_list.front());
        });
    }

    future<hostent> get_host_by_name(sstring name, inet_address::family family)  {
        auto p = new promise<hostent>();
        auto f = p->get_future();

        dns_log.debug("Query name {} ({})", name, family);

        dns_call call(*this);

        ares_gethostbyname(_channel, name.c_str(), int(family), [](void* arg, int status, int timeouts, ::hostent* host) {
            auto p = reinterpret_cast<promise<hostent> *>(arg);

            switch (status) {
            default:
                dns_log.debug("Query failed: {}", status);
                p->set_exception(std::system_error(status, ares_errorc));
                break;
            case ARES_SUCCESS:
                p->set_value(make_hostent(*host));
                break;
            }

            delete p;
        }, reinterpret_cast<void *>(p));


        poll_sockets();

        return f.finally([this] {
            end_call();
        });
    }

    future<hostent> get_host_by_addr(inet_address addr) {
        auto p = new promise<hostent>();
        auto f = p->get_future();

        dns_log.debug("Query addr {}", addr);

        dns_call call(*this);

        ares_gethostbyaddr(_channel, addr.data(), addr.size(), int(addr.in_family()), [](void* arg, int status, int timeouts, ::hostent* host) {
            auto p = reinterpret_cast<promise<hostent> *>(arg);

            switch (status) {
            default:
                dns_log.debug("Query failed: {}", status);
                p->set_exception(std::system_error(status, ares_errorc));
                break;
            case ARES_SUCCESS:
                p->set_value(make_hostent(*host));
                break;
            }

            delete p;
        }, reinterpret_cast<void *>(p));


        poll_sockets();

        return f.finally([this] {
            end_call();
        });
    }

    future<sstring> resolve_addr(inet_address addr) {
        return get_host_by_addr(addr).then([](hostent h) {
            return make_ready_future<sstring>(h.names.front());
        });
    }

    future<> close() {
        _closed = true;
        ares_cancel(_channel);
        dns_log.trace("Shutting down {} sockets", _sockets.size());
        for (auto & p : _sockets) {
            do_close(p.first);
        }
        dns_log.trace("Closing gate");
        return _gate.close();
    }
private:
    enum class type {
        none, tcp, udp
    };
    struct dns_call {
        dns_call(impl & i)
            : _i(i)
            , _c(++i._calls)
        {}
        ~dns_call() {
            // If a query does not immediately complete
            // it might never do so, unless data actually
            // comes back to us and a waiting recv promise
            // is fulfilled.
            // We need to add a timer to do polling at ~timeout
            // ms later, so the ares logic can detect this and
            // tell us we're over.
            if (_c == 1 && _i._calls != 0) {
                _i._timer.arm_periodic(_i._timeout);
            }
        }
        impl& _i;
        uint64_t _c;
    };

    void end_call() {
        if (--_calls == 0) {
            _timer.cancel();
        }
    }
    void poll_sockets() {
        fd_set readers, writers;
        int n = 0;

        dns_log.trace("Poll sockets");

        do {
            // Retrieve the set of file descriptors that the library wants us to monitor.
            FD_ZERO(&readers);
            FD_ZERO(&writers);

            n = ares_fds(_channel, &readers, &writers);

            dns_log.trace("ares_fds: {}", n);

            if (n == 0) {
                break;
            }

            n = 0;

            for (auto & p : _sockets) {
                auto & e = p.second;
                auto fd = p.first;
                auto r = FD_ISSET(p.first, &readers);
                auto w = FD_ISSET(p.first, &writers);
                auto ra = e.avail & POLLIN;
                auto wa = e.avail & POLLOUT;

                dns_log.trace("fd {} {}{}/{}{}", fd, (r ? "r" : ""),
                                (w ? "w" : ""), (ra ? "r" : ""),
                                (wa ? "w" : ""));

                if (!wa) {
                    FD_CLR(fd, &writers);
                }
                if (!ra) {
                    FD_CLR(fd, &readers);
                }
                if (FD_ISSET(fd, &writers) || FD_ISSET(fd, &readers)) {
                    ++n;
                }
            }

            ares_process(_channel, &readers, &writers);
        } while (n != 0);
    }

    static hostent make_hostent(const ::hostent& host) {
        hostent e;
        e.names.emplace_back(host.h_name);
        auto np = host.h_aliases;
        while (*np != 0) {
            e.names.emplace_back(*np++);
        }
        auto p = host.h_addr_list;
        while (*p != nullptr) {
            switch (host.h_addrtype) {
            case AF_INET:
                assert(size_t(host.h_length) >= sizeof(in_addr));
                e.addr_list.emplace_back(*reinterpret_cast<const in_addr*>(*p));
                break;
            case AF_INET6:
                assert(size_t(host.h_length) >= sizeof(in6_addr));
                e.addr_list.emplace_back(*reinterpret_cast<const in6_addr*>(*p));
                break;
            default:
                break;
            }
            ++p;
        }

        dns_log.debug("Query success: {}/{}", e.names.front(), e.addr_list.front());

        return e;
    }
    // We need to partially ref-count our socket entries
    // when we have pending reads/writes, so we don't erase the
    // entry to early.
    void use(ares_socket_t fd) {
        _gate.enter();
        auto& e = _sockets.at(fd);
        ++e.pending;
    }
    void release(ares_socket_t fd) {
        auto& e = _sockets.at(fd);
        dns_log.trace("Release socket {} -> {}", fd, e.pending -  1);
        if (--e.pending < 0) {
            _sockets.erase(fd);
            dns_log.trace("Released socket {}", fd);
        }
        _gate.leave();
    }
    ares_socket_t do_socket(int af, int type, int protocol) {
        if (_closed) {
            return -1;
        }
        int fd = next_fd();
        switch (type) {
        case SOCK_STREAM:
            _sockets.emplace(fd, connected_socket());
            dns_log.trace("Created tcp socket {}", fd);
            break;
        case SOCK_DGRAM:
            _sockets.emplace(fd, _stack.make_udp_channel());
            dns_log.trace("Created udp socket {}", fd);
            break;
        default: return -1;
        }
        return fd;
    }
    int do_close(ares_socket_t fd) {
        dns_log.trace("Close socket {}", fd);
        auto& e = _sockets.at(fd);

        // Mark as closed.
        if (std::exchange(e.closed, true)) {
            return 0;
        }

        _gate.enter(); // "leave" is done in release(fd)

        switch (e.typ) {
        case type::tcp:
        {
            dns_log.trace("Close tcp socket {}, {} pending", fd, e.pending);
            future<> f = make_ready_future();
            if (e.tcp.in) {
                e.tcp.socket.shutdown_input();
                dns_log.trace("Closed tcp socket {} input", fd);
            }
            if (e.tcp.out) {
                f = f.then([&e] {
                    return e.tcp.out->close();
                }).then([fd] {
                    dns_log.trace("Closed tcp socket {} output", fd);
                });
            }
            f = f.finally([me = shared_from_this(), fd] {
                me->release(fd);
            });
            break;
        }
        case type::udp:
            e.udp.channel.close();
            release(fd);
            break;
        default:
            // should not happen
            _gate.leave();
            break;
        }
        return 0;
    }
    socket_address sock_addr(const sockaddr * addr, socklen_t len) {
        if (addr->sa_family != AF_INET) {
            throw std::invalid_argument("No ipv6 yet");
        }
        auto in = reinterpret_cast<const sockaddr_in *>(addr);
        return *in;
    }
    int do_connect(ares_socket_t fd, const sockaddr * addr, socklen_t len) {
        if (_closed) {
            return -1;
        }
        try {
            auto& e = get_socket_entry(fd);
            auto sa = sock_addr(addr, len);

            dns_log.trace("Connect {}({})->{}", fd, int(e.typ), sa);

            assert(e.avail == 0);

            e.avail = POLLOUT|POLLIN; // until we know otherwise

            switch (e.typ) {
            case type::tcp: {
                auto f = _stack.connect(sa);
                if (!f.available()) {
                    dns_log.trace("Connection pending: {}", fd);
                    e.avail = 0;
                    use(fd);
                    f.then_wrapped([me = shared_from_this(), &e, fd](future<connected_socket> f) {
                        try {
                            e.tcp.socket = f.get0();
                            dns_log.trace("Connection complete: {}", fd);
                        } catch (...) {
                            dns_log.debug("Connect {} failed: {}", fd, std::current_exception());
                        }
                        e.avail = POLLOUT|POLLIN;
                        me->poll_sockets();
                        me->release(fd);
                    });
                    errno = EWOULDBLOCK;
                    return -1;
                }
                e.tcp.socket = f.get0();
                break;
            }
            case type::udp:
                // we do not have udp connect, so just keep
                // track of the destination
                e.udp.dst = sa;
                break;
            default:
                return -1;
            }
            return 0;
        } catch (...) {
            return -1;
        }
    }
    ssize_t do_recvfrom(ares_socket_t fd, void * dst, size_t len, int flags, struct sockaddr * from, socklen_t * from_len) {
        if (_closed) {
            return -1;
        }
        try {
            auto& e = get_socket_entry(fd);
            dns_log.trace("Read {}({})", fd, int(e.typ));
            // check if we're already reading.
            if (!(e.avail & POLLIN)) {
                dns_log.trace("Read already pending {}", fd);
                errno = EWOULDBLOCK;
                return -1;
            }
            for (;;) {
                switch (e.typ) {
                case type::tcp: {
                    auto & tcp = e.tcp;
                    if (!tcp.indata.empty()) {
                        dns_log.trace("Read {}. {} bytes available", fd, tcp.indata.size());
                        len = std::min(len, tcp.indata.size());
                        std::copy(tcp.indata.begin(), tcp.indata.begin() + len, reinterpret_cast<char *>(dst));
                        tcp.indata.trim_front(len);
                        return len;
                    }
                    if (!tcp.in) {
                        tcp.in = tcp.socket.input();
                    }
                    auto f = tcp.in->read_up_to(len);
                    if (!f.available()) {
                        dns_log.trace("Read {}: data unavailable", fd);
                        e.avail &= ~POLLIN;
                        use(fd);
                        f.then_wrapped([me = shared_from_this(), &e, fd](future<temporary_buffer<char>> f) {
                            try {
                                auto buf = f.get0();
                                dns_log.trace("Read {} -> {} bytes", fd, buf.size());
                                e.tcp.indata = std::move(buf);
                            } catch (...) {
                                dns_log.debug("Read {} failed: {}", fd, std::current_exception());
                            }
                            e.avail |= POLLIN; // always reset state
                            me->poll_sockets();
                            me->release(fd);
                        });
                        errno = EWOULDBLOCK;
                        return -1;
                    }

                    try {
                        tcp.indata = f.get0();
                        continue; // loop will take care of data
                    } catch (std::system_error& e) {
                        errno = e.code().value();
                        return -1;
                    } catch (...) {
                    }
                    return -1;

                }
                case type::udp: {
                    auto & udp = e.udp;
                    if (udp.in) {
                        auto & p = udp.in->get_data();

                        dns_log.trace("Read {}. {} bytes available from {}", fd, p.len(), udp.in->get_src());

                        if (from != nullptr) {
                            *from = socket_address(udp.in->get_src()).as_posix_sockaddr();
                            if (from_len != nullptr) {
                                // TODO: ipvv6
                                *from_len = sizeof(sockaddr_in);
                            }
                        }

                        len = std::min(len, size_t(p.len()));
                        size_t rem = len;
                        auto * out = reinterpret_cast<char *>(dst);
                        for (auto & f : p.fragments()) {
                            auto n = std::min(rem, f.size);
                            out = std::copy_n(f.base, n, out);
                            rem = rem - n;
                        }
                        if (p.len() == len) {
                            udp.in = {};
                        } else {
                            p.trim_front(len);
                        }
                        return len;
                    }
                    auto f = udp.channel.receive();
                    if (!f.available()) {
                        e.avail &= ~POLLIN;
                        use(fd);
                        dns_log.trace("Read {}: data unavailable", fd);
                        f.then_wrapped([me = shared_from_this(), &e, fd](future<net::udp_datagram> f) {
                            try {
                                auto d = f.get0();
                                dns_log.trace("Read {} -> {} bytes", fd, d.get_data().len());
                                e.udp.in = std::move(d);
                                e.avail |= POLLIN;
                            } catch (...) {
                                dns_log.debug("Read {} failed: {}", fd, std::current_exception());
                            }
                            me->poll_sockets();
                            me->release(fd);
                        });
                        errno = EWOULDBLOCK;
                        return -1;
                    }

                    try {
                        udp.in = std::move(f.get0());
                        continue; // loop will take care of data
                    } catch (std::system_error& e) {
                        errno = e.code().value();
                        return -1;
                    } catch (...) {
                    }
                    return -1;
                }
                default:
                    return -1;
                }
            }
        } catch (...) {
        }
        return -1;
    }
    ssize_t do_sendv(ares_socket_t fd, const iovec * vec, int len) {
        if (_closed) {
            return -1;
        }
        try {
            auto& e = _sockets.at(fd);
            dns_log.trace("Send {}({})", fd, int(e.typ));

            // Assume we will be able to send data eventually very soon
            // and just assume that unless we get immediate
            // failures, we'll be ok. If we're not, the
            // timeout logic will have to handle the problem.
            //
            // This saves us on two accounts:
            // 1.) c-ares does not handle EWOULDBLOCK for
            //     udp sockets. Must pretend to finish
            //     immediately there anyway
            // 2.) Doing so for tcp writes saves us having to
            //     match iovec->packet fragments. Downside is we
            //     have to copy the data, but we pretty much
            //     have to anyway, since we could otherwise
            //     get a query time out while we're sending
            //     with zero-copy and suddenly have freed
            //     memory in packets. Bad.


            for (;;) {
                // check if we're already writing.
                if (e.typ == type::tcp && !(e.avail & POLLOUT)) {
                    dns_log.trace("Send already pending {}", fd);
                    errno = EWOULDBLOCK;
                    return -1;
                }

                net::packet p;
                p.reserve(len);
                for (int i = 0; i < len; ++i) {
                    p = net::packet(std::move(p), net::fragment{reinterpret_cast<char *>(vec[i].iov_base), vec[i].iov_len});
                }

                auto bytes = p.len();
                auto f = make_ready_future();

                switch (e.typ) {
                case type::tcp:
                    if (!e.tcp.out) {
                        e.tcp.out = e.tcp.socket.output();
                    }
                    f = e.tcp.out->write(std::move(p));
                    break;
                case type::udp:
                    f = e.udp.channel.send(e.udp.dst, std::move(p));
                    break;
                default:
                    return -1;
                }

                if (!f.available()) {
                    dns_log.trace("Send {} unavailable.", fd);
                    e.avail &= ~POLLOUT;
                    use(fd);
                    f.then_wrapped([me = shared_from_this(), &e, bytes, fd](future<> f) {
                        try {
                            f.get();
                            dns_log.trace("Send {}. {} bytes sent.", fd, bytes);
                        } catch (...) {
                            dns_log.debug("Send {} failed: {}", fd, std::current_exception());
                        }
                        e.avail |= POLLOUT;
                        me->poll_sockets();
                        me->release(fd);
                    });
                    // c-ares does _not_ use non-blocking retry for udp sockets. We just pretend
                    // all is fine even though we have no idea. Barring stack/adapter failure it
                    // is close to the same guarantee a "normal" message send would have anyway.
                    // For tcp we also pretend we're done, to make sure we don't have to deal with
                    // matching sent data
                }
                if (f.failed()) {
                    try {
                        f.get();
                    } catch (std::system_error& e) {
                        errno = e.code().value();
                    } catch (...) {
                    }
                    return -1;
                }

                return len;
            }
        } catch (...) {
        }
        return -1;
    }

    // Note: cannot use to much here, because fd_sets only handle
    // ~1024 fd:s. Set to something below that in case you need to
    // debug (maybe)
    static constexpr ares_socket_t socket_offset = 1;

    ares_socket_t next_fd() {
        ares_socket_t fd = ares_socket_t(_sockets.size() + socket_offset);
        while (_sockets.count(fd)) {
            ++fd;
        }
        return fd;
    }
    struct tcp_entry {
        tcp_entry(connected_socket s)
                        : socket(std::move(s)) {
        }
        ;
        connected_socket socket;
        std::experimental::optional<input_stream<char>> in;
        std::experimental::optional<output_stream<char>> out;
        temporary_buffer<char> indata;
    };
    struct udp_entry {
        udp_entry(net::udp_channel c)
                        : channel(std::move(c)) {
        }
        net::udp_channel channel;
        std::experimental::optional<net::udp_datagram> in;;
        socket_address dst;
    };
    struct sock_entry {
        union {
            tcp_entry tcp;
            udp_entry udp;
        };
        type typ;
        int avail = 0;
        int pending = 0;
        bool closed = false;

        sock_entry(sock_entry&& e)
            : typ(e.typ)
            , avail(e.avail)
        {
            e.typ = type::none;
            switch (typ) {
            case type::tcp:
                tcp = std::move(e.tcp);
                break;
            case type::udp:
                udp = std::move(e.udp);
                break;
            default:
                break;
            }
        }
        sock_entry(connected_socket s)
            : tcp(tcp_entry{std::move(s)})
            , typ(type::tcp)
        {}
        sock_entry(net::udp_channel c)
            : udp(udp_entry{std::move(c)})
            , typ(type::udp)
        {}
        ~sock_entry() {
            switch (typ) {
            case type::tcp: tcp.~tcp_entry(); break;
            case type::udp: udp.~udp_entry(); break;
            default: break;
            }
        }
    };

    sock_entry& get_socket_entry(ares_socket_t fd) {
        auto& e = _sockets.at(fd);
        if (e.closed) {
            throw std::runtime_error("Socket closed");
        }
        return e;
    }


    typedef std::unordered_map<ares_socket_t, sock_entry> socket_map;

    friend struct dns_call;

    socket_map _sockets;
    network_stack & _stack;

    ares_channel _channel = {};
    uint64_t _ops = 0, _calls = 0;
    std::chrono::milliseconds _timeout;
    timer<> _timer;
    gate _gate;
    bool _closed = false;
};

net::dns_resolver::dns_resolver()
    : dns_resolver(options())
{}

net::dns_resolver::dns_resolver(const options& opts)
    : dns_resolver(engine().net(), opts)
{}

net::dns_resolver::dns_resolver(network_stack& stack, const options& opts)
    : _impl(make_shared<impl>(stack, opts))
{}

net::dns_resolver::~dns_resolver()
{}

net::dns_resolver::dns_resolver(dns_resolver&&) noexcept = default;
net::dns_resolver& net::dns_resolver::operator=(dns_resolver&&) noexcept = default;

future<net::hostent> net::dns_resolver::get_host_by_name(const sstring& name, opt_family family) {
    return _impl->get_host_by_name(name, family.value_or(inet_address::family::INET));
}

future<net::hostent> net::dns_resolver::get_host_by_addr(const inet_address& addr) {
    return _impl->get_host_by_addr(addr);
}

future<net::inet_address> net::dns_resolver::resolve_name(const sstring& name, opt_family family) {
    return _impl->resolve_name(name, family.value_or(inet_address::family::INET));
}

future<sstring> net::dns_resolver::resolve_addr(const inet_address& addr) {
    return _impl->resolve_addr(addr);
}

future<> net::dns_resolver::close() {
    return _impl->close();
}

static net::dns_resolver& resolver() {
    static thread_local net::dns_resolver resolver;
    return resolver;
}


future<net::hostent> net::dns::get_host_by_name(const sstring& name, opt_family family) {
    return resolver().get_host_by_name(name, family.value_or(inet_address::family::INET));
}

future<net::hostent> net::dns::get_host_by_addr(const inet_address& addr) {
    return resolver().get_host_by_addr(addr);
}

future<net::inet_address> net::dns::resolve_name(const sstring& name, opt_family family) {
    return resolver().resolve_name(name, family.value_or(inet_address::family::INET));
}

future<sstring> net::dns::resolve_addr(const inet_address& addr) {
    return resolver().resolve_addr(addr);
}

future<sstring> net::inet_address::hostname() const {
    return dns::resolve_addr(*this);
}

future<std::vector<sstring>> net::inet_address::aliases() const {
    return dns::get_host_by_addr(*this).then([](hostent e) {
        return make_ready_future<std::vector<sstring>>(std::move(e.names));
    });
}

future<net::inet_address> net::inet_address::find(
                const sstring& name) {
    return dns::resolve_name(name);
}

future<net::inet_address> net::inet_address::find(
                const sstring& name, family f) {
    return dns::resolve_name(name, f);
}

future<std::vector<net::inet_address>> net::inet_address::find_all(
                const sstring& name) {
    return dns::get_host_by_name(name).then([](hostent e) {
        return make_ready_future<std::vector<net::inet_address>>(std::move(e.addr_list));
    });
}

future<std::vector<net::inet_address>> net::inet_address::find_all(
                const sstring& name, family f) {
    return dns::get_host_by_name(name, f).then([](hostent e) {
        return make_ready_future<std::vector<net::inet_address>>(std::move(e.addr_list));
    });
}

}