mirror of
https://github.com/shadowsocks/shadowsocks-libev.git
synced 2026-10-28 11:43:50 +00:00
Rewrite the resolver in C++ on a fixed query pool
Queries live in pool slots with their first A and AAAA answer inline, c-ares sockets are watched from a fixed table, and answers reach the waiter through a callback plus opaque words instead of a back-pointer. resolv.cpp keeps the C API as a shim. The event Loop can now adopt the ss_event loop, and a Poll can stop watching a socket it does not own.
This commit is contained in:
Max Lv
@@ -14,7 +14,6 @@ src/manager.cpp 14
|
||||
src/plugin.cpp 3
|
||||
src/process.cpp 26
|
||||
src/redir.cpp 15
|
||||
src/resolv.cpp 6
|
||||
src/rule.cpp 2
|
||||
src/server.cpp 17
|
||||
src/ss_event.cpp 5
|
||||
|
||||
+3
-1
@@ -65,6 +65,7 @@ set(LIBSHADOWSOCKS_LIBEV_SOURCE
|
||||
cache.cpp
|
||||
local.cpp
|
||||
resolv.cpp
|
||||
relay/resolver.cpp
|
||||
ssurl.cpp
|
||||
${SS_CRYPTO_SOURCE}
|
||||
${SS_PLUGIN_SOURCE}
|
||||
@@ -90,6 +91,7 @@ set(SS_SERVER_SOURCE
|
||||
udprelay.cpp
|
||||
cache.cpp
|
||||
resolv.cpp
|
||||
relay/resolver.cpp
|
||||
server.cpp
|
||||
${SS_CRYPTO_SOURCE}
|
||||
${SS_PLUGIN_SOURCE}
|
||||
@@ -124,7 +126,7 @@ if(ENABLE_NFTABLES)
|
||||
endif()
|
||||
|
||||
# Modules rewritten in C++ compile with the stricter warning set.
|
||||
ss_strict_warnings(cache.cpp)
|
||||
ss_strict_warnings(cache.cpp resolv.cpp relay/resolver.cpp)
|
||||
|
||||
function(ss_program name module sources)
|
||||
add_executable(${name} ${sources})
|
||||
|
||||
@@ -161,6 +161,16 @@ public:
|
||||
return PoolPtr<T>(object, PoolDeleter<T>{ this });
|
||||
}
|
||||
|
||||
// Re-own an object whose PoolPtr was release()d, e.g. to hand a raw
|
||||
// pointer to a C library callback and take it back afterwards.
|
||||
PoolPtr<T> adopt(T *object) noexcept
|
||||
{
|
||||
if (meta_at(index_of(object))->state != State::live) {
|
||||
std::abort();
|
||||
}
|
||||
return PoolPtr<T>(object, PoolDeleter<T>{ this });
|
||||
}
|
||||
|
||||
// Handle for a live object owned by this pool.
|
||||
Handle<T> handle_of(const T *object) const noexcept
|
||||
{
|
||||
|
||||
+33
-15
@@ -62,7 +62,7 @@ public:
|
||||
|
||||
Status init() noexcept
|
||||
{
|
||||
if (initialised_) {
|
||||
if (initialised_ || native_ != &loop_) {
|
||||
return fail(Errc::invalid);
|
||||
}
|
||||
int rc = uv_loop_init(&loop_);
|
||||
@@ -73,19 +73,30 @@ public:
|
||||
return {};
|
||||
}
|
||||
|
||||
// Use a loop owned by someone else (the ss_event default loop while C
|
||||
// modules still run on it). The loop is neither run nor closed here.
|
||||
Status adopt(uv_loop_t *loop) noexcept
|
||||
{
|
||||
if (initialised_ || loop == nullptr) {
|
||||
return fail(Errc::invalid);
|
||||
}
|
||||
native_ = loop;
|
||||
return {};
|
||||
}
|
||||
|
||||
// Run until stop() or until nothing is left to do.
|
||||
void run() noexcept { uv_run(&loop_, UV_RUN_DEFAULT); }
|
||||
void run() noexcept { uv_run(native_, UV_RUN_DEFAULT); }
|
||||
// Poll once without blocking.
|
||||
void run_nowait() noexcept { uv_run(&loop_, UV_RUN_NOWAIT); }
|
||||
void run_nowait() noexcept { uv_run(native_, UV_RUN_NOWAIT); }
|
||||
// Block for one round of events.
|
||||
void run_once() noexcept { uv_run(&loop_, UV_RUN_ONCE); }
|
||||
void stop() noexcept { uv_stop(&loop_); }
|
||||
void run_once() noexcept { uv_run(native_, UV_RUN_ONCE); }
|
||||
void stop() noexcept { uv_stop(native_); }
|
||||
|
||||
// Cached loop time in milliseconds.
|
||||
std::uint64_t now() const noexcept { return uv_now(&loop_); }
|
||||
void update_time() noexcept { uv_update_time(&loop_); }
|
||||
std::uint64_t now() const noexcept { return uv_now(native_); }
|
||||
void update_time() noexcept { uv_update_time(native_); }
|
||||
|
||||
uv_loop_t *native() noexcept { return &loop_; }
|
||||
uv_loop_t *native() noexcept { return native_; }
|
||||
|
||||
// Let pending close callbacks run and release the loop. Fails, leaving
|
||||
// the loop usable, while handles are still open.
|
||||
@@ -104,7 +115,8 @@ public:
|
||||
|
||||
private:
|
||||
uv_loop_t loop_{};
|
||||
bool initialised_ = false;
|
||||
uv_loop_t *native_ = &loop_;
|
||||
bool initialised_ = false;
|
||||
};
|
||||
|
||||
// Collects the asynchronous closes of one owner's handles and invokes a
|
||||
@@ -241,13 +253,16 @@ public:
|
||||
int interest() const noexcept { return interest_; }
|
||||
ss_socket_t fd() const noexcept { return fd_; }
|
||||
bool is_open() const noexcept { return live; }
|
||||
bool is_closing() const noexcept { return group != nullptr; }
|
||||
|
||||
// Stop polling and close; the socket is closed once libuv lets go.
|
||||
void close(CloseGroup &closer) noexcept
|
||||
// Stop polling and close; the socket is closed once libuv lets go,
|
||||
// unless someone else (c-ares) owns it.
|
||||
void close(CloseGroup &closer, bool close_socket = true) noexcept
|
||||
{
|
||||
if (!live || group != nullptr) {
|
||||
return;
|
||||
}
|
||||
close_socket_ = close_socket;
|
||||
group = &closer;
|
||||
closer.enter();
|
||||
uv_close(as_handle(), &Poll::on_closed);
|
||||
@@ -267,16 +282,19 @@ private:
|
||||
static void on_closed(uv_handle_t *h) noexcept
|
||||
{
|
||||
auto *self = static_cast<Poll *>(h->data);
|
||||
ss_socket_close(self->fd_);
|
||||
if (self->close_socket_) {
|
||||
ss_socket_close(self->fd_);
|
||||
}
|
||||
self->live = false;
|
||||
CloseGroup *closer = self->group;
|
||||
self->group = nullptr;
|
||||
closer->leave(); // may destroy self
|
||||
}
|
||||
|
||||
Owner *owner_ = nullptr;
|
||||
ss_socket_t fd_ = 0;
|
||||
int interest_ = 0;
|
||||
Owner *owner_ = nullptr;
|
||||
ss_socket_t fd_ = 0;
|
||||
int interest_ = 0;
|
||||
bool close_socket_ = true;
|
||||
};
|
||||
|
||||
template <class Owner, void (Owner::*OnFire)()>
|
||||
|
||||
@@ -0,0 +1,214 @@
|
||||
/* SPDX-License-Identifier: GPL-3.0-or-later */
|
||||
#include "relay/resolver.hpp"
|
||||
|
||||
#include <cstring>
|
||||
|
||||
#include "utils.h"
|
||||
|
||||
extern "C" int verbose;
|
||||
|
||||
namespace ss::relay {
|
||||
|
||||
Resolver::~Resolver()
|
||||
{
|
||||
if (channel_ != nullptr) {
|
||||
std::abort(); // shutdown() first
|
||||
}
|
||||
}
|
||||
|
||||
Status Resolver::init(event::Loop &loop, const char *nameservers, bool ipv6_first, std::uint32_t max_queries) noexcept
|
||||
{
|
||||
if (channel_ != nullptr) {
|
||||
return fail(Errc::invalid);
|
||||
}
|
||||
SS_TRY(queries_.reserve(max_queries));
|
||||
int status = ares_library_init(ARES_LIB_INIT_ALL);
|
||||
if (status != ARES_SUCCESS) {
|
||||
LOGE("c-ares error: %s", ares_strerror(status));
|
||||
return fail(Errc::config);
|
||||
}
|
||||
ares_options options{};
|
||||
options.sock_state_cb = &Resolver::sock_state;
|
||||
options.sock_state_cb_data = this;
|
||||
options.timeout = 3000;
|
||||
options.tries = 2;
|
||||
status = ares_init_options(&channel_, &options,
|
||||
ARES_OPT_NOROTATE | ARES_OPT_TIMEOUTMS | ARES_OPT_TRIES | ARES_OPT_SOCK_STATE_CB);
|
||||
if (status != ARES_SUCCESS) {
|
||||
LOGE("c-ares error: %s", ares_strerror(status));
|
||||
channel_ = nullptr;
|
||||
return fail(Errc::config);
|
||||
}
|
||||
if (nameservers != nullptr && ares_set_servers_ports_csv(channel_, nameservers) != ARES_SUCCESS) {
|
||||
LOGE("failed to set nameservers %s", nameservers);
|
||||
ares_destroy(channel_);
|
||||
channel_ = nullptr;
|
||||
return fail(Errc::config);
|
||||
}
|
||||
loop_ = &loop;
|
||||
ipv6_first_ = ipv6_first;
|
||||
for (Watch &w : watches_) {
|
||||
w.owner = this;
|
||||
}
|
||||
SS_TRY(timer_.open(loop, *this));
|
||||
return {};
|
||||
}
|
||||
|
||||
Status Resolver::resolve(const char *host, std::uint16_t port, const ResolveCallback &cb) noexcept
|
||||
{
|
||||
if (channel_ == nullptr) {
|
||||
return fail(Errc::closed);
|
||||
}
|
||||
SS_TRY_ASSIGN(query, queries_.acquire());
|
||||
query->owner = this;
|
||||
query->cb = cb;
|
||||
query->port = port;
|
||||
// c-ares holds the query until both lookups have reported back.
|
||||
Query *raw = query.release();
|
||||
ares_gethostbyname(channel_, host, AF_INET, &Resolver::on_host, raw);
|
||||
ares_gethostbyname(channel_, host, AF_INET6, &Resolver::on_host, raw);
|
||||
rearm();
|
||||
return {};
|
||||
}
|
||||
|
||||
void Resolver::on_host(void *arg, int status, int, hostent *he) noexcept
|
||||
{
|
||||
auto *q = static_cast<Query *>(arg);
|
||||
if (status == ARES_SUCCESS && he != nullptr && he->h_addr_list[0] != nullptr) {
|
||||
if (he->h_addrtype == AF_INET && !q->has4 && he->h_length == sizeof(in_addr)) {
|
||||
q->v4.sin_family = AF_INET;
|
||||
q->v4.sin_port = q->port;
|
||||
std::memcpy(&q->v4.sin_addr, he->h_addr_list[0], sizeof(in_addr));
|
||||
q->has4 = true;
|
||||
} else if (he->h_addrtype == AF_INET6 && !q->has6 && he->h_length == sizeof(in6_addr)) {
|
||||
q->v6.sin6_family = AF_INET6;
|
||||
q->v6.sin6_port = q->port;
|
||||
std::memcpy(&q->v6.sin6_addr, he->h_addr_list[0], sizeof(in6_addr));
|
||||
q->has6 = true;
|
||||
}
|
||||
} else if (verbose && status != ARES_EDESTRUCTION) {
|
||||
LOGI("failed to lookup address: %s", ares_strerror(status));
|
||||
}
|
||||
q->owner->finish(q, status);
|
||||
}
|
||||
|
||||
void Resolver::finish(Query *q, int status) noexcept
|
||||
{
|
||||
if (--q->pending != 0) {
|
||||
return;
|
||||
}
|
||||
PoolPtr<Query> owned = queries_.adopt(q);
|
||||
const sockaddr *best = nullptr;
|
||||
if (status != ARES_EDESTRUCTION) {
|
||||
if (ipv6_first_) {
|
||||
best = q->has6 ? reinterpret_cast<const sockaddr *>(&q->v6)
|
||||
: (q->has4 ? reinterpret_cast<const sockaddr *>(&q->v4) : nullptr);
|
||||
} else {
|
||||
best = q->has4 ? reinterpret_cast<const sockaddr *>(&q->v4)
|
||||
: (q->has6 ? reinterpret_cast<const sockaddr *>(&q->v6) : nullptr);
|
||||
}
|
||||
}
|
||||
ResolveCallback cb = q->cb;
|
||||
sockaddr_storage copy{};
|
||||
if (best != nullptr) {
|
||||
std::memcpy(©, best, best->sa_family == AF_INET ? sizeof(sockaddr_in) : sizeof(sockaddr_in6));
|
||||
}
|
||||
owned.reset(); // the slot is free before the callback can start another query
|
||||
cb.fn(cb, best != nullptr ? reinterpret_cast<const sockaddr *>(©) : nullptr);
|
||||
}
|
||||
|
||||
event::Flow Resolver::Watch::on_ready(int events) noexcept
|
||||
{
|
||||
ares_socket_t rfd = (events & (event::readable | event::failed)) ? socket : ARES_SOCKET_BAD;
|
||||
ares_socket_t wfd = (events & (event::writable | event::failed)) ? socket : ARES_SOCKET_BAD;
|
||||
ares_process_fd(owner->channel_, rfd, wfd);
|
||||
owner->rearm();
|
||||
return event::Flow::proceed;
|
||||
}
|
||||
|
||||
void Resolver::sock_state(void *data, ares_socket_t s, int readable, int writable) noexcept
|
||||
{
|
||||
auto *self = static_cast<Resolver *>(data);
|
||||
Watch *slot = nullptr, *free_slot = nullptr;
|
||||
for (Watch &w : self->watches_) {
|
||||
if (w.poll.is_open() && !w.poll.is_closing() && w.socket == s) {
|
||||
slot = &w;
|
||||
break;
|
||||
}
|
||||
if (free_slot == nullptr && !w.poll.is_open()) {
|
||||
free_slot = &w;
|
||||
}
|
||||
}
|
||||
int events = (readable ? event::readable : 0) | (writable ? event::writable : 0);
|
||||
if (events == 0) {
|
||||
if (slot != nullptr) {
|
||||
// c-ares closes the socket itself right after this.
|
||||
slot->poll.close(self->closer_, false);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (slot == nullptr) {
|
||||
if (free_slot == nullptr) {
|
||||
LOGE("resolver: too many DNS sockets");
|
||||
return;
|
||||
}
|
||||
slot = free_slot;
|
||||
if (!slot->poll.open(*self->loop_, *slot, static_cast<ss_socket_t>(s))) {
|
||||
LOGE("resolver: cannot watch DNS socket");
|
||||
return;
|
||||
}
|
||||
slot->socket = s;
|
||||
}
|
||||
(void)slot->poll.want(events);
|
||||
}
|
||||
|
||||
void Resolver::rearm() noexcept
|
||||
{
|
||||
if (channel_ == nullptr) {
|
||||
return;
|
||||
}
|
||||
timeval tv{};
|
||||
if (ares_timeout(channel_, nullptr, &tv) == nullptr) {
|
||||
timer_.stop();
|
||||
return;
|
||||
}
|
||||
std::uint64_t ms = static_cast<std::uint64_t>(tv.tv_sec) * 1000 + static_cast<std::uint64_t>(tv.tv_usec) / 1000;
|
||||
timer_.start(ms == 0 ? 1 : ms);
|
||||
}
|
||||
|
||||
void Resolver::on_timer() noexcept
|
||||
{
|
||||
ares_process_fd(channel_, ARES_SOCKET_BAD, ARES_SOCKET_BAD);
|
||||
rearm();
|
||||
}
|
||||
|
||||
void Resolver::shutdown(CloseCompletion done) noexcept
|
||||
{
|
||||
if (channel_ == nullptr) {
|
||||
done();
|
||||
return;
|
||||
}
|
||||
done_ = done;
|
||||
closer_.arm(CloseCompletion(
|
||||
[](void *self, std::uint32_t) { static_cast<Resolver *>(self)->on_closed_all(); }, this, 0));
|
||||
// Cancelled queries report nullptr; destroying the channel then reports
|
||||
// ARES_EDESTRUCTION for anything left and closes the sockets.
|
||||
ares_cancel(channel_);
|
||||
ares_channel channel = channel_;
|
||||
channel_ = nullptr;
|
||||
ares_destroy(channel);
|
||||
ares_library_cleanup();
|
||||
for (Watch &w : watches_) {
|
||||
w.poll.close(closer_, false);
|
||||
}
|
||||
timer_.close(closer_);
|
||||
closer_.seal();
|
||||
}
|
||||
|
||||
void Resolver::on_closed_all() noexcept
|
||||
{
|
||||
CloseCompletion done = done_;
|
||||
done();
|
||||
}
|
||||
|
||||
} // namespace ss::relay
|
||||
@@ -0,0 +1,100 @@
|
||||
/* SPDX-License-Identifier: GPL-3.0-or-later */
|
||||
#ifndef SS_RELAY_RESOLVER_HPP
|
||||
#define SS_RELAY_RESOLVER_HPP
|
||||
|
||||
// Asynchronous DNS over c-ares with a fixed number of queries in flight.
|
||||
//
|
||||
// Replaces resolv.cpp's per-query state, per-answer sockaddr allocations
|
||||
// and per-socket watcher nodes. A query lives in a pool slot; its answers are
|
||||
// kept inline (the first A and the first AAAA record are all that is ever
|
||||
// used); c-ares sockets are watched from a fixed table. A query never points
|
||||
// at the object waiting for it: it carries a callback plus opaque words, and
|
||||
// the relay resolves its own generation-checked handle when the answer comes
|
||||
// back, so a connection that closed in the meantime is simply not found.
|
||||
|
||||
#include <array>
|
||||
#include <cstdint>
|
||||
|
||||
#include <ares.h>
|
||||
|
||||
#include "core/pool.hpp"
|
||||
#include "core/result.hpp"
|
||||
#include "event/loop.hpp"
|
||||
#include "netutils.h"
|
||||
|
||||
namespace ss::relay {
|
||||
|
||||
class Resolver;
|
||||
|
||||
// Called once per query with the chosen address, or nullptr when the name
|
||||
// did not resolve (or the resolver is shutting down).
|
||||
struct ResolveCallback {
|
||||
void (*fn)(const ResolveCallback &self, const sockaddr *address) noexcept = nullptr;
|
||||
void *ctx = nullptr;
|
||||
void *aux = nullptr;
|
||||
std::uint64_t token = 0;
|
||||
};
|
||||
|
||||
class Resolver {
|
||||
public:
|
||||
static constexpr std::uint32_t kMaxSockets = 32;
|
||||
|
||||
Resolver() noexcept = default;
|
||||
Resolver(const Resolver &) = delete;
|
||||
Resolver &operator=(const Resolver &) = delete;
|
||||
~Resolver();
|
||||
|
||||
// `nameservers` is a c-ares CSV list (host[:port],...) or nullptr for the
|
||||
// system configuration.
|
||||
Status init(event::Loop &loop, const char *nameservers, bool ipv6_first, std::uint32_t max_queries) noexcept;
|
||||
|
||||
// `port` is in network byte order and copied into the answer. Fails with
|
||||
// Errc::exhausted when max_queries are already in flight; the callback
|
||||
// is then never called.
|
||||
Status resolve(const char *host, std::uint16_t port, const ResolveCallback &cb) noexcept;
|
||||
|
||||
std::uint32_t in_flight() const noexcept { return queries_.live(); }
|
||||
bool ready() const noexcept { return channel_ != nullptr; }
|
||||
|
||||
// Cancel everything (callbacks get nullptr) and close the watchers; the
|
||||
// closes finish on the next loop iterations, then `done` runs.
|
||||
void shutdown(CloseCompletion done) noexcept;
|
||||
|
||||
private:
|
||||
struct Query {
|
||||
Resolver *owner = nullptr;
|
||||
ResolveCallback cb;
|
||||
std::uint16_t port = 0;
|
||||
std::uint8_t pending = 2; // A and AAAA
|
||||
bool has4 = false, has6 = false;
|
||||
sockaddr_in v4{};
|
||||
sockaddr_in6 v6{};
|
||||
};
|
||||
|
||||
struct Watch {
|
||||
Resolver *owner = nullptr;
|
||||
ares_socket_t socket = ARES_SOCKET_BAD;
|
||||
event::Flow on_ready(int events) noexcept;
|
||||
event::Poll<Watch, &Watch::on_ready> poll;
|
||||
};
|
||||
|
||||
static void sock_state(void *data, ares_socket_t s, int readable, int writable) noexcept;
|
||||
static void on_host(void *arg, int status, int timeouts, hostent *he) noexcept;
|
||||
void on_timer() noexcept;
|
||||
void rearm() noexcept;
|
||||
void finish(Query *q, int status) noexcept;
|
||||
void on_closed_all() noexcept;
|
||||
|
||||
event::Loop *loop_ = nullptr;
|
||||
ares_channel channel_ = nullptr;
|
||||
bool ipv6_first_ = false;
|
||||
ObjectPool<Query> queries_;
|
||||
std::array<Watch, kMaxSockets> watches_{};
|
||||
event::Timer<Resolver, &Resolver::on_timer> timer_;
|
||||
event::CloseGroup closer_;
|
||||
CloseCompletion done_;
|
||||
};
|
||||
|
||||
} // namespace ss::relay
|
||||
|
||||
#endif
|
||||
+61
-419
@@ -28,454 +28,96 @@
|
||||
#include "config.h"
|
||||
#endif
|
||||
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <fcntl.h>
|
||||
#ifndef __MINGW32__
|
||||
#include <netdb.h>
|
||||
#include <sys/socket.h>
|
||||
#include <netinet/in.h>
|
||||
#include <errno.h>
|
||||
#include <unistd.h>
|
||||
#else
|
||||
#include "ss_windows.h" // Should be before <ares.h>
|
||||
#endif
|
||||
#include <ares.h>
|
||||
// The C resolver API, now a thin shim over ss::relay::Resolver (see
|
||||
// src/relay/resolver.hpp), kept for the modules that still call it.
|
||||
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <memory>
|
||||
#include <new>
|
||||
|
||||
#include "ss_event.h"
|
||||
|
||||
#include "core.h"
|
||||
|
||||
#include "relay/resolver.hpp"
|
||||
#include "resolv.h"
|
||||
#include "utils.h"
|
||||
#include "netutils.h"
|
||||
|
||||
/*
|
||||
* Implement DNS resolution interface using libc-ares
|
||||
*/
|
||||
namespace {
|
||||
|
||||
#define SS_TIMER_AFTER 1.0
|
||||
constexpr std::uint32_t kMaxQueries = 4096;
|
||||
|
||||
struct resolv_io {
|
||||
ss_io io;
|
||||
struct resolv_io *next;
|
||||
};
|
||||
// Constructed in resolv_init and destroyed in resolv_shutdown, so there is
|
||||
// no static destructor to run at exit.
|
||||
alignas(ss::relay::Resolver) std::byte resolver_storage[sizeof(ss::relay::Resolver)];
|
||||
alignas(ss::event::Loop) std::byte loop_storage[sizeof(ss::event::Loop)];
|
||||
ss::relay::Resolver *resolver = nullptr;
|
||||
ss::event::Loop *loop = nullptr;
|
||||
|
||||
struct resolv_ctx {
|
||||
struct resolv_io *ios;
|
||||
struct ss_timer timer;
|
||||
ss_tstamp last_tick;
|
||||
using client_cb_t = void (*)(struct sockaddr *, void *);
|
||||
using free_cb_t = void (*)(void *);
|
||||
|
||||
ares_channel channel;
|
||||
struct ares_options options;
|
||||
};
|
||||
|
||||
struct resolv_query {
|
||||
int requests[2];
|
||||
size_t response_count;
|
||||
struct sockaddr **responses;
|
||||
|
||||
void (*client_cb)(struct sockaddr *, void *);
|
||||
void (*free_cb)(void *);
|
||||
|
||||
uint16_t port;
|
||||
|
||||
void *data;
|
||||
|
||||
int is_closed;
|
||||
};
|
||||
|
||||
extern int verbose;
|
||||
|
||||
static struct resolv_ctx default_ctx;
|
||||
static struct ss_loop *default_loop;
|
||||
|
||||
enum {
|
||||
MODE_IPV4_FIRST = 0,
|
||||
MODE_IPV6_FIRST = 1
|
||||
};
|
||||
|
||||
static int resolv_mode = MODE_IPV4_FIRST;
|
||||
|
||||
static void resolv_sock_cb(struct ss_loop *, struct ss_io *, int);
|
||||
static void resolv_timer_cb(struct ss_loop *, struct ss_timer *, int);
|
||||
static void resolv_sock_state_cb(void *, ares_socket_t, int, int);
|
||||
|
||||
static void dns_query_v4_cb(void *, int, int, struct hostent *);
|
||||
static void dns_query_v6_cb(void *, int, int, struct hostent *);
|
||||
|
||||
static void process_client_callback(struct resolv_query *);
|
||||
static inline int all_requests_are_null(struct resolv_query *);
|
||||
static struct sockaddr *choose_ipv4_first(struct resolv_query *);
|
||||
static struct sockaddr *choose_ipv6_first(struct resolv_query *);
|
||||
static struct sockaddr *choose_any(struct resolv_query *);
|
||||
|
||||
/*
|
||||
* DNS UDP socket activity callback
|
||||
*/
|
||||
static void
|
||||
resolv_sock_cb(SS_P_ ss_io *w, int revents)
|
||||
void deliver(const ss::relay::ResolveCallback &cb, const sockaddr *address) noexcept
|
||||
{
|
||||
ares_socket_t rfd = ARES_SOCKET_BAD, wfd = ARES_SOCKET_BAD;
|
||||
|
||||
if (revents & SS_READ)
|
||||
rfd = w->fd;
|
||||
if (revents & SS_WRITE)
|
||||
wfd = w->fd;
|
||||
|
||||
default_ctx.last_tick = ss_now(default_loop);
|
||||
|
||||
ares_process_fd(default_ctx.channel, rfd, wfd);
|
||||
auto client = reinterpret_cast<client_cb_t>(cb.ctx);
|
||||
auto release = reinterpret_cast<free_cb_t>(cb.aux);
|
||||
void *data = reinterpret_cast<void *>(static_cast<std::uintptr_t>(cb.token));
|
||||
client(const_cast<sockaddr *>(address), data);
|
||||
if (release != nullptr) {
|
||||
release(data);
|
||||
} else {
|
||||
ss_free(data);
|
||||
}
|
||||
}
|
||||
|
||||
int
|
||||
resolv_init(struct ss_loop *loop, char *nameservers, int ipv6first)
|
||||
} // namespace
|
||||
|
||||
extern "C" int
|
||||
resolv_init(struct ss_loop *ss_loop, char *nameservers, int ipv6first)
|
||||
{
|
||||
int status;
|
||||
|
||||
if (ipv6first)
|
||||
resolv_mode = MODE_IPV6_FIRST;
|
||||
else
|
||||
resolv_mode = MODE_IPV4_FIRST;
|
||||
|
||||
default_loop = loop;
|
||||
|
||||
if ((status = ares_library_init(ARES_LIB_INIT_ALL)) != ARES_SUCCESS) {
|
||||
LOGE("c-ares error: %s", ares_strerror(status));
|
||||
if (resolver != nullptr) {
|
||||
return 0;
|
||||
}
|
||||
loop = std::construct_at(reinterpret_cast<ss::event::Loop *>(loop_storage));
|
||||
if (!loop->adopt(ss_native_loop(ss_loop))) {
|
||||
FATAL("failed to initialize c-ares");
|
||||
}
|
||||
|
||||
memset(&default_ctx, 0, sizeof(struct resolv_ctx));
|
||||
|
||||
default_ctx.options.sock_state_cb_data = &default_ctx;
|
||||
default_ctx.options.sock_state_cb = resolv_sock_state_cb;
|
||||
default_ctx.options.timeout = 3000;
|
||||
default_ctx.options.tries = 2;
|
||||
|
||||
status = ares_init_options(&default_ctx.channel, &default_ctx.options,
|
||||
#if ARES_VERSION_MAJOR >= 1 && ARES_VERSION_MINOR >= 12
|
||||
ARES_OPT_NOROTATE |
|
||||
#endif
|
||||
ARES_OPT_TIMEOUTMS | ARES_OPT_TRIES | ARES_OPT_SOCK_STATE_CB);
|
||||
|
||||
if (status != ARES_SUCCESS) {
|
||||
resolver = std::construct_at(reinterpret_cast<ss::relay::Resolver *>(resolver_storage));
|
||||
if (!resolver->init(*loop, nameservers, ipv6first != 0, kMaxQueries)) {
|
||||
FATAL("failed to initialize c-ares");
|
||||
}
|
||||
|
||||
if (nameservers != NULL) {
|
||||
#if ARES_VERSION_MAJOR >= 1 && ARES_VERSION_MINOR >= 11
|
||||
status = ares_set_servers_ports_csv(default_ctx.channel, nameservers);
|
||||
#else
|
||||
status = ares_set_servers_csv(default_ctx.channel, nameservers);
|
||||
#endif
|
||||
}
|
||||
|
||||
if (status != ARES_SUCCESS) {
|
||||
FATAL("failed to set nameservers");
|
||||
}
|
||||
|
||||
default_ctx.last_tick = ss_now(default_loop);
|
||||
ss_timer_init(&default_ctx.timer, resolv_timer_cb, 0, 0);
|
||||
resolv_timer_cb(default_loop, &default_ctx.timer, 0);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
void
|
||||
resolv_shutdown(struct ss_loop *loop)
|
||||
{
|
||||
ss_timer_stop(default_loop, &default_ctx.timer);
|
||||
while (default_ctx.ios) {
|
||||
struct resolv_io *node = default_ctx.ios;
|
||||
default_ctx.ios = node->next;
|
||||
ss_io_stop(default_loop, &node->io);
|
||||
ss_free(node);
|
||||
}
|
||||
|
||||
ares_cancel(default_ctx.channel);
|
||||
ares_destroy(default_ctx.channel);
|
||||
|
||||
ares_library_cleanup();
|
||||
}
|
||||
|
||||
void
|
||||
extern "C" void
|
||||
resolv_start(const char *hostname, uint16_t port,
|
||||
void (*client_cb)(struct sockaddr *, void *),
|
||||
void (*free_cb)(void *), void *data)
|
||||
{
|
||||
/*
|
||||
* Wrap c-ares's call back in our own
|
||||
*/
|
||||
struct resolv_query *query = (struct resolv_query *)ss_malloc(sizeof(struct resolv_query));
|
||||
|
||||
memset(query, 0, sizeof(struct resolv_query));
|
||||
|
||||
query->port = port;
|
||||
query->client_cb = client_cb;
|
||||
query->response_count = 0;
|
||||
query->responses = NULL;
|
||||
query->data = data;
|
||||
query->free_cb = free_cb;
|
||||
|
||||
query->requests[0] = AF_INET;
|
||||
query->requests[1] = AF_INET6;
|
||||
|
||||
ares_gethostbyname(default_ctx.channel, hostname, AF_INET, dns_query_v4_cb, query);
|
||||
ares_gethostbyname(default_ctx.channel, hostname, AF_INET6, dns_query_v6_cb, query);
|
||||
ss::relay::ResolveCallback cb;
|
||||
cb.fn = &deliver;
|
||||
cb.ctx = reinterpret_cast<void *>(client_cb);
|
||||
cb.aux = reinterpret_cast<void *>(free_cb);
|
||||
cb.token = static_cast<std::uint64_t>(reinterpret_cast<std::uintptr_t>(data));
|
||||
if (resolver == nullptr || !resolver->resolve(hostname, port, cb)) {
|
||||
LOGE("resolver: too many queries in flight");
|
||||
deliver(cb, nullptr);
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* Wrapper for client callback we provide to c-ares
|
||||
*/
|
||||
static void
|
||||
dns_query_v4_cb(void *arg, int status, int timeouts, struct hostent *he)
|
||||
extern "C" void
|
||||
resolv_shutdown(struct ss_loop *)
|
||||
{
|
||||
int i, n;
|
||||
struct resolv_query *query = (struct resolv_query *)arg;
|
||||
|
||||
if (status == ARES_EDESTRUCTION) {
|
||||
if (resolver == nullptr) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!he || status != ARES_SUCCESS) {
|
||||
if (verbose) {
|
||||
LOGI("failed to lookup v4 address %s", ares_strerror(status));
|
||||
}
|
||||
goto CLEANUP;
|
||||
}
|
||||
|
||||
if (verbose) {
|
||||
LOGI("found address name v4 address %s", he->h_name);
|
||||
}
|
||||
|
||||
n = 0;
|
||||
while (he->h_addr_list[n])
|
||||
n++;
|
||||
|
||||
if (n > 0) {
|
||||
struct sockaddr **new_responses = (struct sockaddr **)ss_realloc(query->responses,
|
||||
(query->response_count + n)
|
||||
* sizeof(struct sockaddr *));
|
||||
|
||||
if (new_responses == NULL) {
|
||||
LOGE("failed to allocate memory for additional DNS responses");
|
||||
} else {
|
||||
query->responses = new_responses;
|
||||
|
||||
for (i = 0; i < n; i++) {
|
||||
struct sockaddr_in *sa = (struct sockaddr_in *)ss_malloc(sizeof(struct sockaddr_in));
|
||||
memset(sa, 0, sizeof(struct sockaddr_in));
|
||||
sa->sin_family = AF_INET;
|
||||
sa->sin_port = query->port;
|
||||
memcpy(&sa->sin_addr, he->h_addr_list[i], he->h_length);
|
||||
|
||||
query->responses[query->response_count] = (struct sockaddr *)sa;
|
||||
if (query->responses[query->response_count] == NULL) {
|
||||
LOGE("failed to allocate memory for DNS query result address");
|
||||
} else {
|
||||
query->response_count++;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
CLEANUP:
|
||||
|
||||
query->requests[0] = 0; /* mark A query as being completed */
|
||||
|
||||
/* Once all requests have completed, call client callback */
|
||||
if (all_requests_are_null(query)) {
|
||||
return process_client_callback(query);
|
||||
static bool closed;
|
||||
closed = false;
|
||||
resolver->shutdown(ss::CloseCompletion([](void *, std::uint32_t) { closed = true; }, nullptr, 0));
|
||||
while (!closed) {
|
||||
loop->run_nowait();
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
dns_query_v6_cb(void *arg, int status, int timeouts, struct hostent *he)
|
||||
{
|
||||
int i, n;
|
||||
struct resolv_query *query = (struct resolv_query *)arg;
|
||||
|
||||
if (status == ARES_EDESTRUCTION) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!he || status != ARES_SUCCESS) {
|
||||
if (verbose) {
|
||||
LOGI("failed to lookup v6 address %s", ares_strerror(status));
|
||||
}
|
||||
goto CLEANUP;
|
||||
}
|
||||
|
||||
if (verbose) {
|
||||
LOGI("found address name v6 address %s", he->h_name);
|
||||
}
|
||||
|
||||
n = 0;
|
||||
while (he->h_addr_list[n])
|
||||
n++;
|
||||
|
||||
if (n > 0) {
|
||||
struct sockaddr **new_responses = (struct sockaddr **)ss_realloc(query->responses,
|
||||
(query->response_count + n)
|
||||
* sizeof(struct sockaddr *));
|
||||
|
||||
if (new_responses == NULL) {
|
||||
LOGE("failed to allocate memory for additional DNS responses");
|
||||
} else {
|
||||
query->responses = new_responses;
|
||||
|
||||
for (i = 0; i < n; i++) {
|
||||
struct sockaddr_in6 *sa = (struct sockaddr_in6 *)ss_malloc(sizeof(struct sockaddr_in6));
|
||||
memset(sa, 0, sizeof(struct sockaddr_in6));
|
||||
sa->sin6_family = AF_INET6;
|
||||
sa->sin6_port = query->port;
|
||||
memcpy(&sa->sin6_addr, he->h_addr_list[i], he->h_length);
|
||||
|
||||
query->responses[query->response_count] = (struct sockaddr *)sa;
|
||||
if (query->responses[query->response_count] == NULL) {
|
||||
LOGE("failed to allocate memory for DNS query result address");
|
||||
} else {
|
||||
query->response_count++;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
CLEANUP:
|
||||
|
||||
query->requests[1] = 0; /* mark A query as being completed */
|
||||
|
||||
/* Once all requests have completed, call client callback */
|
||||
if (all_requests_are_null(query)) {
|
||||
return process_client_callback(query);
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* Called once all requests have been completed
|
||||
*/
|
||||
static void
|
||||
process_client_callback(struct resolv_query *query)
|
||||
{
|
||||
struct sockaddr *best_address = NULL;
|
||||
|
||||
if (resolv_mode == MODE_IPV4_FIRST) {
|
||||
best_address = choose_ipv4_first(query);
|
||||
} else if (resolv_mode == MODE_IPV6_FIRST) {
|
||||
best_address = choose_ipv6_first(query);
|
||||
} else {
|
||||
best_address = choose_any(query);
|
||||
}
|
||||
|
||||
query->client_cb(best_address, query->data);
|
||||
|
||||
for (int i = 0; i < query->response_count; i++)
|
||||
ss_free(query->responses[i]);
|
||||
|
||||
ss_free(query->responses);
|
||||
|
||||
if (query->free_cb != NULL)
|
||||
query->free_cb(query->data);
|
||||
else
|
||||
ss_free(query->data);
|
||||
|
||||
ss_free(query);
|
||||
}
|
||||
|
||||
static struct sockaddr *
|
||||
choose_ipv4_first(struct resolv_query *query)
|
||||
{
|
||||
for (int i = 0; i < query->response_count; i++)
|
||||
if (query->responses[i]->sa_family == AF_INET) {
|
||||
return query->responses[i];
|
||||
}
|
||||
|
||||
return choose_any(query);
|
||||
}
|
||||
|
||||
static struct sockaddr *
|
||||
choose_ipv6_first(struct resolv_query *query)
|
||||
{
|
||||
for (int i = 0; i < query->response_count; i++)
|
||||
if (query->responses[i]->sa_family == AF_INET6) {
|
||||
return query->responses[i];
|
||||
}
|
||||
|
||||
return choose_any(query);
|
||||
}
|
||||
|
||||
static struct sockaddr *
|
||||
choose_any(struct resolv_query *query)
|
||||
{
|
||||
if (query->response_count >= 1) {
|
||||
return query->responses[0];
|
||||
}
|
||||
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static inline int
|
||||
all_requests_are_null(struct resolv_query *query)
|
||||
{
|
||||
int result = 1;
|
||||
|
||||
for (int i = 0; i < sizeof(query->requests) / sizeof(query->requests[0]);
|
||||
i++)
|
||||
result = result && query->requests[i] == 0;
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/*
|
||||
* Timer callback
|
||||
*/
|
||||
static void
|
||||
resolv_timer_cb(struct ss_loop *loop, struct ss_timer *w, int revents)
|
||||
{
|
||||
struct resolv_ctx *ctx = &default_ctx;
|
||||
|
||||
ss_tstamp now = ss_now(default_loop);
|
||||
ss_tstamp after = ctx->last_tick - now + SS_TIMER_AFTER;
|
||||
|
||||
if (after < 0.0) {
|
||||
ctx->last_tick = now;
|
||||
ares_process_fd(ctx->channel, ARES_SOCKET_BAD, ARES_SOCKET_BAD);
|
||||
|
||||
ss_timer_set(w, SS_TIMER_AFTER, 0.0);
|
||||
} else {
|
||||
ss_timer_set(w, after, 0.0);
|
||||
}
|
||||
|
||||
ss_timer_start(loop, w);
|
||||
}
|
||||
|
||||
/*
|
||||
* Handle c-ares events
|
||||
*/
|
||||
static void
|
||||
resolv_sock_state_cb(void *data, ares_socket_t s, int read, int write)
|
||||
{
|
||||
struct resolv_ctx *ctx = (struct resolv_ctx *)data;
|
||||
int events = (read ? SS_READ : 0) | (write ? SS_WRITE : 0);
|
||||
|
||||
struct resolv_io **entry = &ctx->ios;
|
||||
while (*entry && (ares_socket_t)(unsigned int)(*entry)->io.fd != s)
|
||||
entry = &(*entry)->next;
|
||||
struct resolv_io *node = *entry;
|
||||
if (node) ss_io_stop(default_loop, &node->io);
|
||||
if (!events) {
|
||||
if (node) {
|
||||
*entry = node->next;
|
||||
ss_free(node);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (!node) {
|
||||
node = (struct resolv_io *)ss_malloc(sizeof(*node));
|
||||
node->next = NULL;
|
||||
*entry = node;
|
||||
ss_io_init(&node->io, resolv_sock_cb, (int)s, events);
|
||||
} else ss_io_set(&node->io, (int)s, events);
|
||||
ss_io_start(default_loop, &node->io);
|
||||
std::destroy_at(resolver);
|
||||
std::destroy_at(loop);
|
||||
resolver = nullptr;
|
||||
loop = nullptr;
|
||||
}
|
||||
|
||||
@@ -263,7 +263,7 @@ ss_add_test(test_process
|
||||
set_tests_properties(test_process PROPERTIES TIMEOUT 20)
|
||||
|
||||
ss_add_test(test_resolv
|
||||
"test_resolv.c;${PROJECT_SOURCE_DIR}/src/resolv.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp"
|
||||
"test_resolv.c;${PROJECT_SOURCE_DIR}/src/resolv.cpp;${PROJECT_SOURCE_DIR}/src/relay/resolver.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp"
|
||||
"ss_ev;ss_cares;ss_sodium")
|
||||
set_tests_properties(test_resolv PROPERTIES TIMEOUT 20)
|
||||
|
||||
@@ -278,7 +278,7 @@ ss_add_test(test_acl_literal
|
||||
ss_add_test(test_event "test_event.c" "ss_ev")
|
||||
set_tests_properties(test_event PROPERTIES TIMEOUT 30)
|
||||
ss_add_test(test_resolv_async
|
||||
"test_resolv_async.c;${PROJECT_SOURCE_DIR}/src/resolv.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp"
|
||||
"test_resolv_async.c;${PROJECT_SOURCE_DIR}/src/resolv.cpp;${PROJECT_SOURCE_DIR}/src/relay/resolver.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp"
|
||||
"ss_ev;ss_cares;ss_sodium")
|
||||
set_tests_properties(test_resolv_async PROPERTIES TIMEOUT 30)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user