diff --git a/CMakeLists.txt b/CMakeLists.txt index ab5b6b5d..7de10c5b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -105,7 +105,7 @@ endif() include(${PROJECT_SOURCE_DIR}/cmake/configure.cmake) configure_file(${PROJECT_SOURCE_DIR}/cmake/config.h.cmake ${PROJECT_BINARY_DIR}/src/config.h) include(cmake/ProjectOptions.cmake) -ss_project_options(ss_ev) + include_directories(${PROJECT_BINARY_DIR}/src) add_compile_definitions(HAVE_CONFIG_H) diff --git a/cmake/EventLoop.cmake b/cmake/EventLoop.cmake index 9055c296..f1d4d171 100644 --- a/cmake/EventLoop.cmake +++ b/cmake/EventLoop.cmake @@ -1,6 +1,7 @@ -add_library(ss_ev STATIC ${PROJECT_SOURCE_DIR}/src/ss_event.cpp) -target_include_directories(ss_ev PUBLIC ${PROJECT_SOURCE_DIR}/src) -target_link_libraries(ss_ev PUBLIC ss_uv Threads::Threads ${SS_MATH_LIBRARY} ${CMAKE_DL_LIBS}) +# The event loop is libuv, used through the C++ layer in src/event. +add_library(ss_ev INTERFACE) +target_include_directories(ss_ev INTERFACE ${PROJECT_SOURCE_DIR}/src) +target_link_libraries(ss_ev INTERFACE ss_uv Threads::Threads ${SS_MATH_LIBRARY} ${CMAKE_DL_LIBS}) if(CMAKE_SYSTEM_NAME STREQUAL "Linux") - target_link_libraries(ss_ev PUBLIC rt) + target_link_libraries(ss_ev INTERFACE rt) endif() diff --git a/cmake/InstallPackage.cmake b/cmake/InstallPackage.cmake index 16cc97ef..88f18617 100644 --- a/cmake/InstallPackage.cmake +++ b/cmake/InstallPackage.cmake @@ -5,7 +5,7 @@ set(SS_PC_PRIVATE "") set(private_links) set(private_content "# Generated private linkage for the installed static library.\n") if(SS_BUILD_STATIC_LIBRARY) - set(private_targets bloom blake3 ss_ev) + set(private_targets bloom blake3) if(SS_DEPENDENCY_MODE STREQUAL "bundled") list(APPEND private_targets ss_sodium ss_mbedcrypto ss_uv ss_cares) if(SS_ENABLE_REGEX) diff --git a/scripts/no_heap_baseline.txt b/scripts/no_heap_baseline.txt index 276ecdf0..7bd77ec5 100644 --- a/scripts/no_heap_baseline.txt +++ b/scripts/no_heap_baseline.txt @@ -15,8 +15,7 @@ src/plugin.cpp 3 src/process.cpp 26 src/redir.cpp 1 src/rule.cpp 2 -src/server.cpp 4 -src/ss_event.cpp 5 +src/server.cpp 3 src/ssurl.cpp 3 src/stream.cpp 3 src/tunnel.cpp 1 diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index c60676b1..63f80f52 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -78,7 +78,6 @@ set(LIBSHADOWSOCKS_LIBEV_SOURCE ${SS_SHARED_SOURCES} cache.cpp local.cpp - resolv.cpp ssurl.cpp ${SS_RELAY_SOURCE} ${SS_CRYPTO_SOURCE} @@ -103,7 +102,6 @@ set(SS_TUNNEL_SOURCE set(SS_SERVER_SOURCE ${SS_SHARED_SOURCES} cache.cpp - resolv.cpp server.cpp ${SS_RELAY_SOURCE} ${SS_CRYPTO_SOURCE} @@ -139,7 +137,7 @@ if(ENABLE_NFTABLES) endif() # Modules rewritten in C++ compile with the stricter warning set. -ss_strict_warnings(cache.cpp resolv.cpp ${SS_RELAY_SOURCE}) +ss_strict_warnings(cache.cpp ${SS_RELAY_SOURCE}) function(ss_program name module sources) add_executable(${name} ${sources}) diff --git a/src/event/program.hpp b/src/event/program.hpp new file mode 100644 index 00000000..1701b2e2 --- /dev/null +++ b/src/event/program.hpp @@ -0,0 +1,194 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +#ifndef SS_EVENT_PROGRAM_HPP +#define SS_EVENT_PROGRAM_HPP + +// The event loop of one program run: the loop itself, the signals that end +// it, and a few program-level watchers (a periodic callback, sockets). +// Replaces each main()'s ss_signal/ss_timer/ss_io plumbing. Everything is +// held by value in the Program object, which main() keeps on its stack. + +#include +#include +#include + +#include "core/result.hpp" +#include "event/loop.hpp" +#include "utils.h" + +namespace ss::event { + +class Program { +public: + using Callback = void (*)(void *ctx) noexcept; + + Program() noexcept = default; + Program(const Program &) = delete; + Program &operator=(const Program &) = delete; + ~Program() { close(); } + + Status init() noexcept { return loop_.init(); } + Loop &loop() noexcept { return loop_; } + + // SIGINT and SIGTERM end the run (and SIGUSR1 with `usr1`). With + // `plugin_alive`, SIGCHLD ends it with exit code -1 once that reports + // the plugin gone. Signals libuv cannot watch on this platform are + // skipped. + Status handle_signals(bool usr1, bool (*plugin_alive)() = nullptr) noexcept + { + plugin_alive_ = plugin_alive; + int signals[] = { SIGINT, SIGTERM, +#ifndef _WIN32 + usr1 ? SIGUSR1 : 0, plugin_alive != nullptr ? SIGCHLD : 0 +#endif + }; + for (int signum : signals) { + if (signum == 0) { + continue; + } + if (signal_count_ == signals_.size()) { + return fail(Errc::exhausted); + } + SignalWatch &w = signals_[signal_count_++]; + w.program = this; + (void)w.signal.open(loop_, w, signum); + } + (void)usr1; + return {}; + } + + // Call `fn` every `interval_ms`. + Status every(std::uint64_t interval_ms, Callback fn, void *ctx) noexcept + { + tick_.fn = fn; + tick_.ctx = ctx; + SS_TRY(tick_.timer.open(loop_, tick_)); + tick_.timer.start(interval_ms, interval_ms); + return {}; + } + + // Call `fn` whenever `fd` is readable. The Program owns the socket. + Status watch(ss_socket_t fd, Callback fn, void *ctx) noexcept + { + if (watch_count_ == watches_.size()) { + return fail(Errc::exhausted); + } + SocketWatch &w = watches_[watch_count_]; + w.fn = fn; + w.ctx = ctx; + SS_TRY(w.poll.open(loop_, w, fd)); + ++watch_count_; + return w.poll.want(readable); + } + + // Let other threads end the run with request_stop(). + Status allow_remote_stop() noexcept + { + wake_.program = this; + return wake_.async.open(loop_, wake_); + } + // Thread-safe; needs allow_remote_stop() first. + void request_stop() noexcept { wake_.async.send(); } + + // Run until stop() or a signal; returns the exit code. + int run() noexcept + { + loop_.run(); + return exit_code_; + } + + void stop(int exit_code = 0) noexcept + { + if (exit_code != 0) { + exit_code_ = exit_code; + } + loop_.stop(); + } + + int exit_code() const noexcept { return exit_code_; } + + // Close the program's own handles and let their closes finish. Other + // owners (relays, resolvers) must be closed and drained first. + void close() noexcept + { + if (closed_) { + return; + } + closed_ = true; + for (std::size_t i = 0; i < signal_count_; ++i) { + signals_[i].signal.close(closer_); + } + for (std::size_t i = 0; i < watch_count_; ++i) { + watches_[i].poll.close(closer_); + } + tick_.timer.close(closer_); + wake_.async.close(closer_); + static bool done; + done = false; + closer_.arm(CloseCompletion([](void *, std::uint32_t) { done = true; }, nullptr, 0)); + closer_.seal(); + while (!done) { // NOLINT(bugprone-infinite-loop): the close callback sets it + loop_.run_nowait(); + } + (void)loop_.shutdown(); + } + +private: + struct SignalWatch { + Program *program = nullptr; + void on_signal(int signum) noexcept { program->on_signal(signum); } + Signal signal; + }; + struct SocketWatch { + Callback fn = nullptr; + void *ctx = nullptr; + Flow on_ready(int) noexcept + { + fn(ctx); + return Flow::proceed; + } + Poll poll; + }; + struct Wake { + Program *program = nullptr; + void on_wake() noexcept { program->stop(0); } + Async async; + }; + struct Tick { + Callback fn = nullptr; + void *ctx = nullptr; + void on_fire() noexcept { fn(ctx); } + Timer timer; + }; + + void on_signal(int signum) noexcept + { +#ifndef _WIN32 + if (signum == SIGCHLD) { + if (plugin_alive_ == nullptr || plugin_alive_()) { + return; + } + LOGE("plugin service exit unexpectedly"); + stop(-1); + return; + } +#endif + (void)signum; + stop(0); + } + + Loop loop_; + std::array signals_{}; + std::size_t signal_count_ = 0; + std::array watches_{}; + std::size_t watch_count_ = 0; + Tick tick_; + Wake wake_; + CloseGroup closer_; + bool (*plugin_alive_)() = nullptr; + int exit_code_ = 0; + bool closed_ = false; +}; + +} // namespace ss::event + +#endif diff --git a/src/local.cpp b/src/local.cpp index 06021d09..712e6510 100644 --- a/src/local.cpp +++ b/src/local.cpp @@ -53,19 +53,18 @@ #include "core.h" #include "netutils.h" -#include "resolv.h" #include "ssurl.h" #include "utils.h" #include "socks5.h" #include "acl.h" #include "plugin.h" -#include "ss_event.h" #include "crypto.h" #include "jconf.h" #include "common.h" #include "ss_windows.h" #include "core/heap_guard.hpp" +#include "event/program.hpp" #include "relay/tcp.hpp" #include "relay/udp.hpp" @@ -118,15 +117,10 @@ static int printable_salt = 0; static int udp_fd = 0; static int ret_val = 0; -static struct ss_signal sigint_watcher; -static struct ss_signal sigterm_watcher; #ifndef __MINGW32__ -static struct ss_signal sigchld_watcher; -static struct ss_signal sigusr1_watcher; #else #ifndef LIB_ONLY static struct plugin_watcher_t { - ss_io io; SOCKET fd; uint16_t port; int valid; @@ -140,9 +134,7 @@ static int nofile = 0; #endif #endif -static void signal_cb(SS_P_ ss_signal *w, int revents); #if defined(__MINGW32__) && !defined(LIB_ONLY) -static void plugin_watcher_cb(SS_P_ ss_io *w, int revents); #endif static int create_and_bind(const char *addr, const char *port); @@ -278,7 +270,7 @@ stat_update_cb() namespace { struct Relays { - ss::event::Loop loop; + ss::event::Loop *loop = NULL; ss::relay::Method method; ss::relay::Resolver resolver; ss::relay::TcpRelay tcp; @@ -287,11 +279,9 @@ struct Relays { bool tcp_on = false; bool udp_on = false; - bool init(struct ss_loop *ss_loop, const char *cipher, const char *pass, const char *key) + bool init(ss::event::Loop &event_loop, const char *cipher, const char *pass, const char *key) { - if (!loop.adopt(ss_native_loop(ss_loop))) { - return false; - } + loop = &event_loop; LOGI("initializing ciphers... %s", cipher); if (!method.init(cipher, pass, key, ss::crypto::Role::client)) { LOGE("failed to initialize ciphers"); @@ -306,12 +296,12 @@ struct Relays { config.acl_hooks.match_host = &acl_match_host; config.acl_hooks.mode = &get_acl_mode; config.acl_hooks.outbound_blocked = &outbound_block_match_host; - if (!resolver.init(loop, NULL, ipv6first != 0, config.max_connections)) { + if (!resolver.init(*loop, NULL, ipv6first != 0, config.max_connections)) { return false; } resolving = true; } - if (!tcp.init(loop, method, config, resolving ? &resolver : NULL) || !tcp.listen(listenfd)) { + if (!tcp.init(*loop, method, config, resolving ? &resolver : NULL) || !tcp.listen(listenfd)) { return false; } tcp_on = true; @@ -337,7 +327,7 @@ struct Relays { config.on_traffic = &stat_update_cb; #endif auto fd = ss::relay::open_udp_listener(host, port, false, reuse_port != 0); - if (!fd || !udp.init(loop, method, config, NULL)) { + if (!fd || !udp.init(*loop, method, config, NULL)) { return -1; } if (!udp.listen(*fd)) { @@ -362,7 +352,7 @@ struct Relays { resolver.shutdown(ss::CloseCompletion([](void *, uint32_t) { resolver_closed = true; }, NULL, 0)); } while ((tcp_on && !tcp.closed()) || (udp_on && !udp.closed()) || !resolver_closed) { - loop.run_nowait(); + loop->run_nowait(); } tcp_on = udp_on = resolving = false; } @@ -395,40 +385,10 @@ fill_tcp_config(ss::relay::TcpConfig &config, const struct sockaddr_storage *rem } // namespace -static void -signal_cb(SS_P_ ss_signal *w, int revents) -{ - if (revents & SS_SIGNAL) { - switch (w->signum) { -#ifndef __MINGW32__ - case SIGCHLD: - if (!is_plugin_running()) { - LOGE("plugin service exit unexpectedly"); - ret_val = -1; - } else - return; - case SIGUSR1: -#endif - case SIGINT: - case SIGTERM: - ss_signal_stop(SS_DEFAULT, &sigint_watcher); - ss_signal_stop(SS_DEFAULT, &sigterm_watcher); -#ifndef __MINGW32__ - ss_signal_stop(SS_DEFAULT, &sigchld_watcher); - ss_signal_stop(SS_DEFAULT, &sigusr1_watcher); -#else -#ifndef LIB_ONLY - ss_io_stop(SS_DEFAULT, &plugin_watcher.io); -#endif -#endif - ss_unloop(SS_A_ SS_UNLOOP_ALL); - } - } -} #if defined(__MINGW32__) && !defined(LIB_ONLY) static void -plugin_watcher_cb(SS_P_ ss_io *w, int revents) +plugin_watcher_cb(void *ctx) noexcept { char buf[1]; SOCKET fd = ss_accept(plugin_watcher.fd, NULL, NULL); @@ -438,22 +398,27 @@ plugin_watcher_cb(SS_P_ ss_io *w, int revents) recv(fd, buf, 1, 0); closesocket(fd); LOGE("plugin service exit unexpectedly"); - ret_val = -1; - ss_signal_stop(SS_DEFAULT, &sigint_watcher); - ss_signal_stop(SS_DEFAULT, &sigterm_watcher); - ss_io_stop(SS_DEFAULT, &plugin_watcher.io); - ss_unloop(SS_A_ SS_UNLOOP_ALL); + static_cast(ctx)->stop(-1); } #endif #ifndef LIB_ONLY +static bool +plugin_alive() +{ + return is_plugin_running() != 0; +} + int main(int argc, char **argv) { // Bounded heaps for libuv, c-ares and Mbed TLS before any of them runs. if (!ss::install_library_heaps()) FATAL("failed to reserve the library heaps"); + ss::event::Program program; + if (!program.init()) + FATAL("failed to set up the event loop"); int i, c; int pid_flags = 0; @@ -1163,8 +1128,8 @@ Same behavior as `-S`; see that option for details. Short alias: `-S`. break; } plugin_watcher.fd = fd; - ss_io_init(&plugin_watcher.io, plugin_watcher_cb, fd, SS_READ); - ss_io_start(SS_DEFAULT, &plugin_watcher.io); + if (!program.watch(fd, plugin_watcher_cb, &program)) + break; plugin_watcher.valid = 1; } while (0); if (!plugin_watcher.valid) { @@ -1206,7 +1171,7 @@ Same behavior as `-S`; see that option for details. Short alias: `-S`. // Setup keys Relays relays; - if (!relays.init(SS_DEFAULT, method, password, key)) + if (!relays.init(program.loop(), method, password, key)) FATAL("failed to initialize ciphers"); // Setup proxy context @@ -1233,21 +1198,14 @@ Same behavior as `-S`; see that option for details. Short alias: `-S`. fill_tcp_config(tcp_config, remote_storage, remote_count, timeout_secs, mptcp, iface); // Setup signal handler - ss_signal_init(&sigint_watcher, signal_cb, SIGINT); - ss_signal_init(&sigterm_watcher, signal_cb, SIGTERM); - ss_signal_start(SS_DEFAULT, &sigint_watcher); - ss_signal_start(SS_DEFAULT, &sigterm_watcher); -#ifndef __MINGW32__ - ss_signal_init(&sigchld_watcher, signal_cb, SIGCHLD); - ss_signal_start(SS_DEFAULT, &sigchld_watcher); -#endif + if (!program.handle_signals(false, plugin != NULL ? &plugin_alive : NULL)) + FATAL("failed to watch signals"); if (ss_is_ipv6addr(local_addr)) LOGI("listening at [%s]:%s", local_addr, local_port); else LOGI("listening at %s:%s", local_addr, local_port); - struct ss_loop *loop = SS_DEFAULT; if (mode != UDP_ONLY) { // Setup socket @@ -1303,7 +1261,11 @@ Same behavior as `-S`; see that option for details. Short alias: `-S`. // Enter the loop ss::heap_seal(); - ss_run(loop, 0); + { + int code = program.run(); + if (code != 0) + ret_val = code; + } if (verbose) { LOGI("closed gracefully"); @@ -1315,12 +1277,9 @@ Same behavior as `-S`; see that option for details. Short alias: `-S`. } relays.stop(); - ss_loop_destroy(loop); + program.close(); #ifdef __MINGW32__ - if (plugin_watcher.valid) { - closesocket(plugin_watcher.fd); - } winsock_cleanup(); #endif @@ -1349,18 +1308,16 @@ struct ss_local_handle { namespace { -// The running server's stop signal. ss_local_stop() and the server thread -// take the lock around it, so a stop never wakes a loop that is gone. +// The running server's loop. ss_local_stop() and the server thread take +// the lock around it, so a stop never wakes a loop that is gone. uv_mutex_t control_lock; uv_once_t control_once = UV_ONCE_INIT; -uv_async_t *stop_signal = NULL; +ss::event::Program *running_program = NULL; ss_local_handle *running_handle = NULL; bool server_busy = false; void control_init(void) { uv_mutex_init(&control_lock); } -void on_stop_signal(uv_async_t *) { ss_unloop(SS_DEFAULT, SS_UNLOOP_ALL); } - char *copy_string(const char *s) { return s == NULL ? NULL : strdup(s); // heap-ok: startup, library only @@ -1440,8 +1397,8 @@ ss_local_stop(ss_local_handle *handle) uv_once(&control_once, control_init); uv_mutex_lock(&control_lock); handle->stop_requested = true; - if (running_handle == handle && stop_signal != NULL) { - uv_async_send(stop_signal); + if (running_handle == handle && running_program != NULL) { + running_program->request_stop(); } uv_mutex_unlock(&control_lock); } @@ -1517,18 +1474,13 @@ _start_ss_local_server(profile_t profile, ss_local_callback callback, void *udat signal(SIGABRT, SIG_IGN); #endif - ss_signal_init(&sigint_watcher, signal_cb, SIGINT); - ss_signal_init(&sigterm_watcher, signal_cb, SIGTERM); - ss_signal_start(SS_DEFAULT, &sigint_watcher); - ss_signal_start(SS_DEFAULT, &sigterm_watcher); -#ifndef __MINGW32__ - ss_signal_init(&sigusr1_watcher, signal_cb, SIGUSR1); - ss_signal_start(SS_DEFAULT, &sigusr1_watcher); -#endif + ss::event::Program program; + if (!program.init() || !program.handle_signals(true) || !program.allow_remote_stop()) + FATAL("failed to set up the event loop"); // Setup keys Relays relays; - if (!relays.init(SS_DEFAULT, method, password, NULL)) + if (!relays.init(program.loop(), method, password, NULL)) FATAL("failed to init ciphers"); struct sockaddr_storage storage; @@ -1538,7 +1490,6 @@ _start_ss_local_server(profile_t profile, ss_local_callback callback, void *udat } // Setup proxy context - struct ss_loop *loop = SS_DEFAULT; ss::relay::TcpConfig tcp_config; fill_tcp_config(tcp_config, &storage, 1, timeout, mptcp, NULL); int listen_fd = -1; // stays -1 in UDP_ONLY mode but is still passed to the callback below @@ -1580,23 +1531,23 @@ _start_ss_local_server(profile_t profile, ss_local_callback callback, void *udat callback(listen_fd, udp_fd, udata); } - // ss_local_stop() wakes the loop through this handle. - uv_async_t stop_async; - uv_async_init(ss_native_loop(loop), &stop_async, on_stop_signal); + // ss_local_stop() wakes this loop from another thread. uv_once(&control_once, control_init); uv_mutex_lock(&control_lock); - bool stop_now = running_handle != NULL && running_handle->stop_requested; - stop_signal = &stop_async; + bool stop_now = running_handle != NULL && running_handle->stop_requested; + running_program = &program; uv_mutex_unlock(&control_lock); // Enter the loop - if (!stop_now) - ss_run(loop, 0); + if (!stop_now) { + int code = program.run(); + if (code != 0) + ret_val = code; + } uv_mutex_lock(&control_lock); - stop_signal = NULL; + running_program = NULL; uv_mutex_unlock(&control_lock); - uv_close((uv_handle_t *)&stop_async, NULL); if (verbose) { LOGI("closed gracefully"); @@ -1604,7 +1555,7 @@ _start_ss_local_server(profile_t profile, ss_local_callback callback, void *udat // Clean up relays.stop(); - ss_loop_destroy(loop); + program.close(); #ifdef __MINGW32__ winsock_cleanup(); diff --git a/src/manager.cpp b/src/manager.cpp index 62475048..c806fa80 100644 --- a/src/manager.cpp +++ b/src/manager.cpp @@ -64,6 +64,8 @@ #include "netutils.h" #include "manager.h" +#include "event/program.hpp" + #ifndef BUF_SIZE #define BUF_SIZE 65535 #endif @@ -953,9 +955,9 @@ update_stat(char *port, uint64_t traffic) } static void -manager_recv_cb(SS_P_ ss_io *w, int revents) +manager_recv_cb(void *ctx) noexcept { - struct manager_ctx *manager = (struct manager_ctx *)w; + struct manager_ctx *manager = (struct manager_ctx *)ctx; socklen_t len; ssize_t r; struct sockaddr_un claddr; @@ -1146,17 +1148,6 @@ ERROR_MSG: } } -static void -signal_cb(SS_P_ ss_signal *w, int revents) -{ - if (revents & SS_SIGNAL) { - switch (w->signum) { - case SIGINT: - case SIGTERM: - ss_unloop(SS_A_ SS_UNLOOP_ALL); - } - } -} int create_server_socket(const char *host, const char *port) @@ -1714,12 +1705,9 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. signal(SIGCHLD, SIG_IGN); signal(SIGABRT, SIG_IGN); - struct ss_signal sigint_watcher; - struct ss_signal sigterm_watcher; - ss_signal_init(&sigint_watcher, signal_cb, SIGINT); - ss_signal_init(&sigterm_watcher, signal_cb, SIGTERM); - ss_signal_start(SS_DEFAULT, &sigint_watcher); - ss_signal_start(SS_DEFAULT, &sigterm_watcher); + ss::event::Program program; + if (!program.init() || !program.handle_signals(false)) + FATAL("failed to set up the event loop"); struct manager_ctx manager; memset(&manager, 0, sizeof(struct manager_ctx)); @@ -1748,9 +1736,6 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. manager.nofile = nofile; #endif - // initialize ev loop - struct ss_loop *loop = SS_DEFAULT; - // Clean up all existed processes DIR *dp; struct dirent *ep; @@ -1836,11 +1821,11 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. } manager.fd = sfd; - ss_io_init(&manager.io, manager_recv_cb, manager.fd, SS_READ); - ss_io_start(loop, &manager.io); + if (!program.watch(manager.fd, manager_recv_cb, &manager)) + FATAL("failed to watch the control socket"); // start ev loop - ss_run(loop, 0); + (void)program.run(); if (verbose) { LOGI("closed gracefully"); @@ -1855,8 +1840,7 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. ss_free(server); } - ss_signal_stop(SS_DEFAULT, &sigint_watcher); - ss_signal_stop(SS_DEFAULT, &sigterm_watcher); + program.close(); ss_free(working_dir); free_addr(&ip_addr); diff --git a/src/manager.h b/src/manager.h index 9c5867be..33c799b8 100644 --- a/src/manager.h +++ b/src/manager.h @@ -27,8 +27,6 @@ #include "core.h" #include "uthash.h" -#include "ss_event.h" - #include "jconf.h" #include "common.h" @@ -38,7 +36,6 @@ extern "C" { #endif struct manager_ctx { - ss_io io; int fd; int fast_open; int no_delay; diff --git a/src/redir.cpp b/src/redir.cpp index c688dfcf..af519117 100644 --- a/src/redir.cpp +++ b/src/redir.cpp @@ -51,12 +51,12 @@ #include "netutils.h" #include "utils.h" #include "common.h" -#include "ss_event.h" #include "crypto.h" #include "jconf.h" #include "common.h" #include "core/heap_guard.hpp" +#include "event/program.hpp" #include "relay/tcp.hpp" #include "relay/udp.hpp" @@ -102,9 +102,6 @@ static int max_connections = 0; static int max_udp_sessions = 0; static int ret_val = 0; -static struct ss_signal sigint_watcher; -static struct ss_signal sigterm_watcher; -static struct ss_signal sigchld_watcher; static int tcp_tproxy = 0; /* use tproxy instead of redirect (for tcp) */ @@ -202,26 +199,11 @@ create_and_bind(const char *addr, const char *port) return listen_sock; } -static void -signal_cb(SS_P_ ss_signal *w, int revents) -{ - if (revents & SS_SIGNAL) { - switch (w->signum) { - case SIGCHLD: - if (!is_plugin_running()) { - LOGE("plugin service exit unexpectedly"); - ret_val = -1; - } else - return; - case SIGINT: - case SIGTERM: - ss_signal_stop(SS_DEFAULT, &sigint_watcher); - ss_signal_stop(SS_DEFAULT, &sigterm_watcher); - ss_signal_stop(SS_DEFAULT, &sigchld_watcher); - ss_unloop(SS_A_ SS_UNLOOP_ALL); - } - } +static bool +plugin_alive() +{ + return is_plugin_running() != 0; } int @@ -230,6 +212,9 @@ main(int argc, char **argv) // Bounded heaps for libuv, c-ares and Mbed TLS before any of them runs. if (!ss::install_library_heaps()) FATAL("failed to reserve the library heaps"); + ss::event::Program program; + if (!program.init()) + FATAL("failed to set up the event loop"); int i, c; int pid_flags = 0; @@ -842,12 +827,8 @@ Same behavior as `-T`; see that option for details. Short alias: `-T`. signal(SIGPIPE, SIG_IGN); signal(SIGABRT, SIG_IGN); - ss_signal_init(&sigint_watcher, signal_cb, SIGINT); - ss_signal_init(&sigterm_watcher, signal_cb, SIGTERM); - ss_signal_init(&sigchld_watcher, signal_cb, SIGCHLD); - ss_signal_start(SS_DEFAULT, &sigint_watcher); - ss_signal_start(SS_DEFAULT, &sigterm_watcher); - ss_signal_start(SS_DEFAULT, &sigchld_watcher); + if (!program.handle_signals(false, plugin != NULL ? &plugin_alive : NULL)) + FATAL("failed to watch signals"); // Setup keys LOGI("initializing ciphers... %s", method); @@ -876,14 +857,10 @@ Same behavior as `-T`; see that option for details. Short alias: `-T`. break; } - struct ss_loop *loop = SS_DEFAULT; - ss::event::Loop tcp_loop; + ss::event::Loop &tcp_loop = program.loop(); ss::relay::TcpRelay tcp_relay; ss::relay::UdpRelay udp_relay; bool tcp_running = false; - if (!tcp_loop.adopt(ss_native_loop(loop))) { - FATAL("failed to set up the event loop"); - } if (mode != TCP_ONLY) { ss::relay::UdpConfig udp_config; udp_config.kind = ss::relay::Kind::redir; @@ -972,7 +949,11 @@ Same behavior as `-T`; see that option for details. Short alias: `-T`. } ss::heap_seal(); - ss_run(loop, 0); + { + int code = program.run(); + if (code != 0) + ret_val = code; + } if (plugin != NULL) { stop_plugin(); @@ -983,6 +964,7 @@ Same behavior as `-T`; see that option for details. Short alias: `-T`. while (!tcp_relay.closed() || !udp_relay.closed()) { tcp_loop.run_nowait(); } + program.close(); (void)tcp_running; if (verbose) diff --git a/src/resolv.cpp b/src/resolv.cpp deleted file mode 100644 index b3c521f0..00000000 --- a/src/resolv.cpp +++ /dev/null @@ -1,123 +0,0 @@ -/* - * Copyright (c) 2014, Dustin Lundquist - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright - * notice, this list of conditions and the following disclaimer in the - * documentation and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#ifdef HAVE_CONFIG_H -#include "config.h" -#endif - -// 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 -#include -#include -#include - -#include "ss_event.h" - -#include "relay/resolver.hpp" -#include "resolv.h" -#include "utils.h" - -namespace { - -constexpr std::uint32_t kMaxQueries = 4096; - -// 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; - -using client_cb_t = void (*)(struct sockaddr *, void *); -using free_cb_t = void (*)(void *); - -void deliver(const ss::relay::ResolveCallback &cb, const sockaddr *address) noexcept -{ - auto client = reinterpret_cast(cb.ctx); - auto release = reinterpret_cast(cb.aux); - void *data = reinterpret_cast(static_cast(cb.token)); - client(const_cast(address), data); - if (release != nullptr) { - release(data); - } else { - ss_free(data); - } -} - -} // namespace - -extern "C" int -resolv_init(struct ss_loop *ss_loop, char *nameservers, int ipv6first) -{ - if (resolver != nullptr) { - return 0; - } - loop = std::construct_at(reinterpret_cast(loop_storage)); - if (!loop->adopt(ss_native_loop(ss_loop))) { - FATAL("failed to initialize c-ares"); - } - resolver = std::construct_at(reinterpret_cast(resolver_storage)); - if (!resolver->init(*loop, nameservers, ipv6first != 0, kMaxQueries)) { - FATAL("failed to initialize c-ares"); - } - return 0; -} - -extern "C" void -resolv_start(const char *hostname, uint16_t port, - void (*client_cb)(struct sockaddr *, void *), - void (*free_cb)(void *), void *data) -{ - ss::relay::ResolveCallback cb; - cb.fn = &deliver; - cb.ctx = reinterpret_cast(client_cb); - cb.aux = reinterpret_cast(free_cb); - cb.token = static_cast(reinterpret_cast(data)); - if (resolver == nullptr || !resolver->resolve(hostname, port, cb)) { - LOGE("resolver: too many queries in flight"); - deliver(cb, nullptr); - } -} - -extern "C" void -resolv_shutdown(struct ss_loop *) -{ - if (resolver == nullptr) { - return; - } - static bool closed; - closed = false; - resolver->shutdown(ss::CloseCompletion([](void *, std::uint32_t) { closed = true; }, nullptr, 0)); - while (!closed) { // NOLINT(bugprone-infinite-loop): the completion callback sets it - loop->run_nowait(); - } - std::destroy_at(resolver); - std::destroy_at(loop); - resolver = nullptr; - loop = nullptr; -} diff --git a/src/resolv.h b/src/resolv.h deleted file mode 100644 index ef35fb44..00000000 --- a/src/resolv.h +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Copyright (c) 2014, Dustin Lundquist - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright - * notice, this list of conditions and the following disclaimer in the - * documentation and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ -#ifndef RESOLV_H -#define RESOLV_H - -#ifdef HAVE_CONFIG_H -#include "config.h" -#endif - -#include -#ifndef __MINGW32__ -#include -#endif - -#ifdef __cplusplus -extern "C" { -#endif - -struct ss_loop; -struct resolv_query; - -int resolv_init(struct ss_loop *, char *, int); -void resolv_start(const char *hostname, uint16_t port, - void (*client_cb)(struct sockaddr *, void *), - void (*free_cb)(void *), void *data); -void resolv_shutdown(struct ss_loop *); - -#ifdef __cplusplus -} -#endif - -#endif diff --git a/src/server.cpp b/src/server.cpp index 86f7cf73..9b82ea79 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -70,12 +70,10 @@ enum datatypes { #include "utils.h" #include "acl.h" #include "plugin.h" -#include "ss_event.h" #include "crypto.h" #include "jconf.h" #include "common.h" #include "ss_windows.h" -#include "resolv.h" #ifdef USE_NFCONNTRACK_TOS #include @@ -83,6 +81,7 @@ enum datatypes { #endif #include "core/heap_guard.hpp" +#include "event/program.hpp" #include "relay/tcp.hpp" #include "relay/udp.hpp" @@ -113,7 +112,6 @@ enum datatypes { #endif -static void signal_cb(SS_P_ ss_signal *w, int revents); int verbose = 0; int reuse_port = 0; @@ -147,16 +145,11 @@ uint64_t tx = 0; uint64_t rx = 0; #ifndef __MINGW32__ -ss_timer stat_update_watcher; #endif -static struct ss_signal sigint_watcher; -static struct ss_signal sigterm_watcher; #ifndef __MINGW32__ -static struct ss_signal sigchld_watcher; #else static struct plugin_watcher_t { - ss_io io; SOCKET fd; uint16_t port; int valid; @@ -165,21 +158,27 @@ static struct plugin_watcher_t { #ifndef __MINGW32__ static int setnonblocking(int fd); -static void stat_resolved(struct sockaddr *address, void *data) +// Resolves the manager's address when it is given by name. +static ss::relay::Resolver *stat_resolver = NULL; + +static void stat_resolved(const ss::relay::ResolveCallback &cb, const sockaddr *address) noexcept { - if (!address) return; - int fd = ss_socket(address->sa_family, SOCK_DGRAM, 0); - if (fd == -1) return; - setnonblocking(fd); - const char *message = (const char *)data; - size_t length = strlen(message) + 1; - if (sendto(fd, message, length, 0, address, get_sockaddr_len(address)) != (ssize_t)length) - ERROR("stat_sendto"); - ss_socket_close(fd); + char *message = (char *)cb.ctx; + if (address) { + int fd = ss_socket(address->sa_family, SOCK_DGRAM, 0); + if (fd != -1) { + setnonblocking(fd); + size_t length = strlen(message) + 1; + if (sendto(fd, message, length, 0, address, get_sockaddr_len((struct sockaddr *)address)) != (ssize_t)length) + ERROR("stat_sendto"); + ss_socket_close(fd); + } + } + free(message); // heap-ok: one stats datagram every few seconds } static void -stat_update_cb(SS_P_ ss_timer *watcher, int revents) +stat_update_cb(void *) noexcept { struct sockaddr_un svaddr, claddr; int sfd = -1; @@ -238,9 +237,13 @@ stat_update_cb(SS_P_ ss_timer *watcher, int revents) } else { uint16_t port; if (parse_numeric_port(ip_addr.port, &port) == 0) { - char *message = strdup(resp); + char *message = strdup(resp); // heap-ok: one stats datagram every few seconds if (!message) FATAL("out of memory"); - resolv_start(ip_addr.host, htons(port), stat_resolved, NULL, message); + ss::relay::ResolveCallback cb; + cb.fn = &stat_resolved; + cb.ctx = message; + if (stat_resolver == NULL || !stat_resolver->resolve(ip_addr.host, htons(port), cb)) + free(message); // heap-ok: see above } ss_free(ip_addr.host); ss_free(ip_addr.port); @@ -773,36 +776,10 @@ report_failure(ss_socket_t fd, const char *reason) report_addr((int)fd, reason); } -static void -signal_cb(SS_P_ ss_signal *w, int revents) -{ - if (revents & SS_SIGNAL) { - switch (w->signum) { -#ifndef __MINGW32__ - case SIGCHLD: - if (!is_plugin_running()) { - LOGE("plugin service exit unexpectedly"); - ret_val = -1; - } else - return; -#endif - case SIGINT: - case SIGTERM: - ss_signal_stop(SS_DEFAULT, &sigint_watcher); - ss_signal_stop(SS_DEFAULT, &sigterm_watcher); -#ifndef __MINGW32__ - ss_signal_stop(SS_DEFAULT, &sigchld_watcher); -#else - ss_io_stop(SS_DEFAULT, &plugin_watcher.io); -#endif - ss_unloop(SS_A_ SS_UNLOOP_ALL); - } - } -} #ifdef __MINGW32__ static void -plugin_watcher_cb(SS_P_ ss_io *w, int revents) +plugin_watcher_cb(void *ctx) noexcept { char buf[1]; SOCKET fd = ss_accept(plugin_watcher.fd, NULL, NULL); @@ -812,21 +789,26 @@ plugin_watcher_cb(SS_P_ ss_io *w, int revents) recv(fd, buf, 1, 0); closesocket(fd); LOGE("plugin service exit unexpectedly"); - ret_val = -1; - ss_signal_stop(SS_DEFAULT, &sigint_watcher); - ss_signal_stop(SS_DEFAULT, &sigterm_watcher); - ss_io_stop(SS_DEFAULT, &plugin_watcher.io); - ss_unloop(SS_A_ SS_UNLOOP_ALL); + static_cast(ctx)->stop(-1); } #endif +static bool +plugin_alive() +{ + return is_plugin_running() != 0; +} + int main(int argc, char **argv) { // Bounded heaps for libuv, c-ares and Mbed TLS before any of them runs. if (!ss::install_library_heaps()) FATAL("failed to reserve the library heaps"); + ss::event::Program program; + if (!program.init()) + FATAL("failed to set up the event loop"); int i, c; int pid_flags = 0; @@ -1465,14 +1447,8 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. signal(SIGABRT, SIG_IGN); #endif - ss_signal_init(&sigint_watcher, signal_cb, SIGINT); - ss_signal_init(&sigterm_watcher, signal_cb, SIGTERM); - ss_signal_start(SS_DEFAULT, &sigint_watcher); - ss_signal_start(SS_DEFAULT, &sigterm_watcher); -#ifndef __MINGW32__ - ss_signal_init(&sigchld_watcher, signal_cb, SIGCHLD); - ss_signal_start(SS_DEFAULT, &sigchld_watcher); -#endif + if (!program.handle_signals(false, plugin != NULL ? &plugin_alive : NULL)) + FATAL("failed to watch signals"); // setup keys LOGI("initializing ciphers... %s", method); @@ -1483,11 +1459,8 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. FATAL("failed to set up the AEAD-2022 salt pool"); // initialize ev loop - struct ss_loop *loop = SS_DEFAULT; // setup dns - resolv_init(loop, nameservers, ipv6first); - if (nameservers != NULL) LOGI("using nameserver: %s", nameservers); @@ -1513,8 +1486,8 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. break; } plugin_watcher.fd = fd; - ss_io_init(&plugin_watcher.io, plugin_watcher_cb, fd, SS_READ); - ss_io_start(SS_DEFAULT, &plugin_watcher.io); + if (!program.watch(fd, plugin_watcher_cb, &program)) + break; plugin_watcher.valid = 1; } while (0); if (!plugin_watcher.valid) { @@ -1550,18 +1523,17 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. } } - // The relays run on the ss_event loop next to the C modules, and share - // one resolver for targets given by name. - ss::event::Loop tcp_loop; + // The relays share one resolver for targets given by name. + ss::event::Loop &tcp_loop = program.loop(); ss::relay::Resolver tcp_resolver; ss::relay::TcpRelay tcp_relay; ss::relay::UdpRelay udp_relay; bool tcp_running = false; - if (!tcp_loop.adopt(ss_native_loop(loop)) - || !tcp_resolver.init(tcp_loop, nameservers, ipv6first != 0, + if (!tcp_resolver.init(tcp_loop, nameservers, ipv6first != 0, (max_connections > 0 ? max_connections : DEFAULT_MAX_CONNECTIONS) + 64)) { FATAL("failed to set up the resolver"); } + stat_resolver = &tcp_resolver; if (mode != UDP_ONLY) { ss::relay::TcpConfig tcp_config; tcp_config.kind = ss::relay::Kind::server; @@ -1689,8 +1661,8 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. #ifndef __MINGW32__ if (manager_addr != NULL) { - ss_timer_init(&stat_update_watcher, stat_update_cb, UPDATE_INTERVAL, UPDATE_INTERVAL); - ss_timer_start(SS_DEFAULT, &stat_update_watcher); + if (!program.every(UPDATE_INTERVAL * 1000, stat_update_cb, NULL)) + FATAL("failed to start the traffic statistics timer"); } #endif @@ -1707,26 +1679,22 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. // start ev loop ss::heap_seal(); - ss_run(loop, 0); + { + int code = program.run(); + if (code != 0) + ret_val = code; + } if (verbose) { LOGI("closed gracefully"); } -#ifndef __MINGW32__ - if (manager_addr != NULL) { - ss_timer_stop(SS_DEFAULT, &stat_update_watcher); - } -#endif - if (plugin != NULL) { stop_plugin(); } // Clean up - resolv_shutdown(loop); - { static bool resolver_closed; resolver_closed = false; @@ -1738,12 +1706,10 @@ Same behavior as `-d`; see that option for details. Short alias: `-d`. } } (void)tcp_running; - ss_loop_destroy(loop); + stat_resolver = NULL; + program.close(); #ifdef __MINGW32__ - if (plugin_watcher.valid) { - closesocket(plugin_watcher.fd); - } winsock_cleanup(); #endif diff --git a/src/ss_event.cpp b/src/ss_event.cpp deleted file mode 100644 index 2d33c1a5..00000000 --- a/src/ss_event.cpp +++ /dev/null @@ -1,278 +0,0 @@ -#include "ss_event.h" -#include -#include "uthash.h" -#include -#include -#include -#include -#include -#include - -struct poll_state { - uv_poll_t handle; - struct ss_loop *loop; - ss_io *watchers; - int fd, events; - UT_hash_handle hh; -}; -struct timer_state { - uv_timer_t handle; - struct ss_loop *loop; - ss_timer *watcher; -}; -struct signal_state { - uv_signal_t handle; - struct ss_loop *loop; - ss_signal *watcher; -}; -struct ss_loop { - uv_loop_t native; - struct poll_state *polls; - uint64_t generation; -}; -static struct ss_loop *default_loop; -static void checked(int status) -{ - if (status < 0) { - fprintf(stderr, "event loop: %s\n", uv_strerror(status)); - abort(); - } -} -static void *allocate(size_t size) -{ - void *p = calloc(1, size); - if (!p) abort(); - return p; -} -static void closed(uv_handle_t *handle) { free(handle); } -uv_loop_t *ss_native_loop(struct ss_loop *loop) { return &loop->native; } -const char *ss_backend_name(void) -{ -#ifdef _WIN32 - return "iocp"; -#elif defined(__APPLE__) || defined(__FreeBSD__) - return "kqueue"; -#elif defined(__linux__) - return "epoll"; -#else - return "libuv"; -#endif -} -struct ss_loop *ss_loop_new(unsigned flags) -{ - (void)flags; - struct ss_loop *loop = (struct ss_loop *)allocate(sizeof(*loop)); - if (uv_loop_init(&loop->native)) { free(loop); return NULL; } - return loop; -} -struct ss_loop *ss_default_loop(void) -{ - if (!default_loop) { - default_loop = ss_loop_new(0); - if (!default_loop) abort(); - } - return default_loop; -} -static void close_walk(uv_handle_t *handle, void *arg) -{ - (void)arg; - if (!uv_is_closing(handle)) uv_close(handle, closed); -} -void ss_loop_destroy(struct ss_loop *loop) -{ - HASH_CLEAR(hh, loop->polls); - uv_walk(&loop->native, close_walk, NULL); - uv_run(&loop->native, UV_RUN_DEFAULT); - checked(uv_loop_close(&loop->native)); - if (loop == default_loop) default_loop = NULL; - free(loop); -} -int ss_run(struct ss_loop *loop, int flags) -{ - return uv_run(&loop->native, flags & SS_RUN_NOWAIT ? UV_RUN_NOWAIT : UV_RUN_DEFAULT); -} -void ss_unloop(struct ss_loop *loop, int how) -{ - (void)how; - uv_stop(&loop->native); -} -double ss_now(struct ss_loop *loop) { return uv_now(&loop->native) / 1000.0; } -double ss_time(void) -{ - uv_timeval64_t now; - checked(uv_gettimeofday(&now)); - return (double)now.tv_sec + now.tv_usec / 1000000.0; -} -struct pending { ss_io *watcher; uint64_t generation; }; -static void polled(uv_poll_t *handle, int status, int events) -{ - struct poll_state *state = (struct poll_state *)handle; - int ready = (events & UV_READABLE ? SS_READ : 0) | - (events & UV_WRITABLE ? SS_WRITE : 0); - if (status < 0) ready = SS_READ | SS_WRITE | SS_ERROR; - size_t count = 0; - for (ss_io *w = state->watchers; w; w = w->next) count++; - struct pending local[8]; - struct pending *pending = count > 8 ? (struct pending *)allocate(count * sizeof(*pending)) : local; - size_t n = 0; - for (ss_io *w = state->watchers; w; w = w->next) { - pending[n].watcher = w; - pending[n].generation = w->generation; - n++; - } - /* A callback can stop/free itself or another watcher, or reuse its memory. - * Revalidate both membership and generation before dereferencing it. */ - for (size_t i = 0; i < count; i++) { - for (ss_io *w = state->watchers; w; w = w->next) { - if (w == pending[i].watcher && w->generation == pending[i].generation) { - int deliver = ready & (w->events | SS_ERROR); - if (deliver) w->cb(state->loop, w, deliver); - break; - } - } - } - if (pending != local) free(pending); -} -static void update_poll(struct poll_state *state) -{ - int events = 0; - for (ss_io *w = state->watchers; w; w = w->next) events |= w->events; - if (events == state->events) return; - state->events = events; - checked(uv_poll_start(&state->handle, - (events & SS_READ ? UV_READABLE : 0) | (events & SS_WRITE ? UV_WRITABLE : 0), polled)); -} -void ss_io_init(ss_io *w, void (*cb)(struct ss_loop *, ss_io *, int), int fd, int events) -{ - memset(w, 0, sizeof(*w)); - w->fd = fd; - w->events = events; - w->cb = cb; -} -void ss_io_set(ss_io *w, int fd, int events) -{ - assert(!w->backend); - w->fd = fd; w->events = events; -} -void ss_io_start(struct ss_loop *loop, ss_io *w) -{ - if (w->backend) return; - struct poll_state *state; - HASH_FIND_INT(loop->polls, &w->fd, state); - if (!state) { - state = (struct poll_state *)allocate(sizeof(*state)); - state->fd = w->fd; state->loop = loop; -#ifdef _WIN32 - checked(uv_poll_init_socket(&loop->native, &state->handle, (SOCKET)(unsigned int)w->fd)); -#else - checked(uv_poll_init(&loop->native, &state->handle, w->fd)); -#endif - // NOLINTNEXTLINE(clang-analyzer-core.DivideZero): uthash expands a nonempty table, so ceil(num_items / num_buckets) is positive. - HASH_ADD_INT(loop->polls, fd, state); - } - w->backend = state; - w->generation = ++loop->generation; - w->next = state->watchers; - state->watchers = w; - update_poll(state); -} -void ss_io_stop(struct ss_loop *loop, ss_io *w) -{ - struct poll_state *state = (struct poll_state *)w->backend; - if (!state) return; - assert(state->loop == loop); - ss_io **item = &state->watchers; - while (*item != w) item = &(*item)->next; - *item = w->next; - w->backend = NULL; w->next = NULL; - if (!state->watchers) { - checked(uv_poll_stop(&state->handle)); - HASH_DEL(loop->polls, state); - uv_close((uv_handle_t *)&state->handle, closed); - } else update_poll(state); -} -static uint64_t milliseconds(double seconds) -{ - return seconds > 0 ? (uint64_t)ceil(seconds * 1000.0) : 0; -} -static void timed(uv_timer_t *handle) -{ - struct timer_state *state = (struct timer_state *)handle; - ss_timer *w = state->watcher; - struct ss_loop *loop = state->loop; - if (!w->repeat) ss_timer_stop(loop, w); - w->cb(loop, w, SS_TIMER); -} -void ss_timer_init(ss_timer *w, void (*cb)(struct ss_loop *, ss_timer *, int), double after, double repeat) -{ - memset(w, 0, sizeof(*w)); - w->after = after; - w->repeat = repeat; - w->cb = cb; -} -void ss_timer_set(ss_timer *w, double after, double repeat) -{ - w->after = after; w->repeat = repeat; -} -void ss_timer_start(struct ss_loop *loop, ss_timer *w) -{ - if (w->backend) return; - struct timer_state *state = (struct timer_state *)allocate(sizeof(*state)); - state->loop = loop; state->watcher = w; - checked(uv_timer_init(&loop->native, &state->handle)); - w->backend = state; - uv_update_time(&loop->native); - checked(uv_timer_start(&state->handle, timed, milliseconds(w->after), milliseconds(w->repeat))); -} -void ss_timer_stop(struct ss_loop *loop, ss_timer *w) -{ - (void)loop; - struct timer_state *state = (struct timer_state *)w->backend; - if (!state) return; - w->backend = NULL; - checked(uv_timer_stop(&state->handle)); - uv_close((uv_handle_t *)&state->handle, closed); -} -void ss_timer_again(struct ss_loop *loop, ss_timer *w) -{ - ss_timer_stop(loop, w); - if (w->repeat > 0) { - w->after = w->repeat; - ss_timer_start(loop, w); - } -} -static void signaled(uv_signal_t *handle, int signum) -{ - (void)signum; - struct signal_state *state = (struct signal_state *)handle; - ss_signal *w = state->watcher; - w->cb(state->loop, w, SS_SIGNAL); -} -void ss_signal_init(ss_signal *w, void (*cb)(struct ss_loop *, ss_signal *, int), int signum) -{ - memset(w, 0, sizeof(*w)); - w->signum = signum; - w->cb = cb; -} -void ss_signal_start(struct ss_loop *loop, ss_signal *w) -{ - if (w->backend) return; -#ifdef _WIN32 - /* Windows has no SIGTERM delivery; process termination is an OS operation. */ - if (w->signum == SIGTERM) return; -#endif - struct signal_state *state = (struct signal_state *)allocate(sizeof(*state)); - state->loop = loop; state->watcher = w; - checked(uv_signal_init(&loop->native, &state->handle)); - checked(uv_signal_start(&state->handle, signaled, w->signum)); - w->backend = state; -} -void ss_signal_stop(struct ss_loop *loop, ss_signal *w) -{ - (void)loop; - struct signal_state *state = (struct signal_state *)w->backend; - if (!state) return; - w->backend = NULL; - checked(uv_signal_stop(&state->handle)); - uv_close((uv_handle_t *)&state->handle, closed); -} diff --git a/src/ss_event.h b/src/ss_event.h deleted file mode 100644 index b05ed1d1..00000000 --- a/src/ss_event.h +++ /dev/null @@ -1,66 +0,0 @@ -/* Project event API backed by bundled/system libuv. Watchers may be freed - * immediately after stop; libuv handle storage has a separate lifetime. */ -#ifndef SS_EVENT_H -#define SS_EVENT_H -#include - -#ifdef __cplusplus -extern "C" { -#endif -/* Keep platform headers private: Linux TCP headers differ from libc headers. */ -struct uv_loop_s; -struct ss_loop; -typedef double ss_tstamp; -#define SS_READ 1 -#define SS_WRITE 2 -#define SS_TIMER 4 -#define SS_SIGNAL 8 -#define SS_ERROR 16 -#define SS_RUN_NOWAIT 1 -#define SS_UNLOOP_ALL 2 -#define SS_P_ struct ss_loop *loop, -#define SS_A_ loop, -#define SS_DEFAULT ss_default_loop() -typedef struct ss_io { - void *backend; - struct ss_io *next; - uint64_t generation; - int fd, events; - void (*cb)(struct ss_loop *, struct ss_io *, int); -} ss_io; -typedef struct ss_timer { - void *backend; - double after, repeat; - void (*cb)(struct ss_loop *, struct ss_timer *, int); -} ss_timer; -typedef struct ss_signal { - void *backend; - int signum; - void (*cb)(struct ss_loop *, struct ss_signal *, int); -} ss_signal; -struct ss_loop *ss_default_loop(void); -struct ss_loop *ss_loop_new(unsigned flags); -void ss_loop_destroy(struct ss_loop *loop); -struct uv_loop_s *ss_native_loop(struct ss_loop *loop); -const char *ss_backend_name(void); -int ss_run(struct ss_loop *loop, int flags); -void ss_unloop(struct ss_loop *loop, int how); -double ss_now(struct ss_loop *loop); -double ss_time(void); -void ss_io_init(ss_io *, void (*)(struct ss_loop *, ss_io *, int), int, int); -void ss_io_set(ss_io *, int, int); -void ss_io_start(struct ss_loop *, ss_io *); -void ss_io_stop(struct ss_loop *, ss_io *); -void ss_timer_init(ss_timer *, void (*)(struct ss_loop *, ss_timer *, int), double, double); -void ss_timer_set(ss_timer *, double, double); -void ss_timer_start(struct ss_loop *, ss_timer *); -void ss_timer_stop(struct ss_loop *, ss_timer *); -void ss_timer_again(struct ss_loop *, ss_timer *); -void ss_signal_init(ss_signal *, void (*)(struct ss_loop *, ss_signal *, int), int); -void ss_signal_start(struct ss_loop *, ss_signal *); -void ss_signal_stop(struct ss_loop *, ss_signal *); -#ifdef __cplusplus -} -#endif - -#endif diff --git a/src/tunnel.cpp b/src/tunnel.cpp index 3409861f..abcd14f1 100644 --- a/src/tunnel.cpp +++ b/src/tunnel.cpp @@ -52,13 +52,13 @@ #include "netutils.h" #include "utils.h" #include "plugin.h" -#include "ss_event.h" #include "crypto.h" #include "jconf.h" #include "common.h" #include "ss_windows.h" #include "core/heap_guard.hpp" +#include "event/program.hpp" #include "relay/tcp.hpp" #include "relay/udp.hpp" @@ -118,13 +118,9 @@ static int max_udp_sessions = 0; int fast_open = 0; static int ret_val = 0; -static struct ss_signal sigint_watcher; -static struct ss_signal sigterm_watcher; #ifndef __MINGW32__ -static struct ss_signal sigchld_watcher; #else static struct plugin_watcher_t { - ss_io io; SOCKET fd; uint16_t port; int valid; @@ -198,36 +194,10 @@ create_and_bind(const char *addr, const char *port) return listen_sock; } -static void -signal_cb(SS_P_ ss_signal *w, int revents) -{ - if (revents & SS_SIGNAL) { - switch (w->signum) { -#ifndef __MINGW32__ - case SIGCHLD: - if (!is_plugin_running()) { - LOGE("plugin service exit unexpectedly"); - ret_val = -1; - } else - return; -#endif - case SIGINT: - case SIGTERM: - ss_signal_stop(SS_DEFAULT, &sigint_watcher); - ss_signal_stop(SS_DEFAULT, &sigterm_watcher); -#ifndef __MINGW32__ - ss_signal_stop(SS_DEFAULT, &sigchld_watcher); -#else - ss_io_stop(SS_DEFAULT, &plugin_watcher.io); -#endif - ss_unloop(SS_A_ SS_UNLOOP_ALL); - } - } -} #ifdef __MINGW32__ static void -plugin_watcher_cb(SS_P_ ss_io *w, int revents) +plugin_watcher_cb(void *ctx) noexcept { char buf[1]; SOCKET fd = ss_accept(plugin_watcher.fd, NULL, NULL); @@ -237,21 +207,26 @@ plugin_watcher_cb(SS_P_ ss_io *w, int revents) recv(fd, buf, 1, 0); closesocket(fd); LOGE("plugin service exit unexpectedly"); - ret_val = -1; - ss_signal_stop(SS_DEFAULT, &sigint_watcher); - ss_signal_stop(SS_DEFAULT, &sigterm_watcher); - ss_io_stop(SS_DEFAULT, &plugin_watcher.io); - ss_unloop(SS_A_ SS_UNLOOP_ALL); + static_cast(ctx)->stop(-1); } #endif +static bool +plugin_alive() +{ + return is_plugin_running() != 0; +} + int main(int argc, char **argv) { // Bounded heaps for libuv, c-ares and Mbed TLS before any of them runs. if (!ss::install_library_heaps()) FATAL("failed to reserve the library heaps"); + ss::event::Program program; + if (!program.init()) + FATAL("failed to set up the event loop"); int i, c; int pid_flags = 0; @@ -907,8 +882,8 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. break; } plugin_watcher.fd = fd; - ss_io_init(&plugin_watcher.io, plugin_watcher_cb, fd, SS_READ); - ss_io_start(SS_DEFAULT, &plugin_watcher.io); + if (!program.watch(fd, plugin_watcher_cb, &program)) + break; plugin_watcher.valid = 1; } while (0); if (!plugin_watcher.valid) { @@ -948,14 +923,8 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. signal(SIGABRT, SIG_IGN); #endif - ss_signal_init(&sigint_watcher, signal_cb, SIGINT); - ss_signal_init(&sigterm_watcher, signal_cb, SIGTERM); - ss_signal_start(SS_DEFAULT, &sigint_watcher); - ss_signal_start(SS_DEFAULT, &sigterm_watcher); -#ifndef __MINGW32__ - ss_signal_init(&sigchld_watcher, signal_cb, SIGCHLD); - ss_signal_start(SS_DEFAULT, &sigchld_watcher); -#endif + if (!program.handle_signals(false, plugin != NULL ? &plugin_alive : NULL)) + FATAL("failed to watch signals"); // Setup keys LOGI("initializing ciphers... %s", method); @@ -1011,13 +980,9 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. LOGI("listening at %s:%s", local_addr, local_port); - struct ss_loop *loop = SS_DEFAULT; - ss::event::Loop tcp_loop; + ss::event::Loop &tcp_loop = program.loop(); ss::relay::TcpRelay tcp_relay; ss::relay::UdpRelay udp_relay; - if (!tcp_loop.adopt(ss_native_loop(loop))) { - FATAL("failed to set up the event loop"); - } if (mode != UDP_ONLY) { if (!tcp_relay.init(tcp_loop, relay_method, tcp_config, NULL)) { @@ -1079,7 +1044,11 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. #endif ss::heap_seal(); - ss_run(loop, 0); + { + int code = program.run(); + if (code != 0) + ret_val = code; + } if (plugin != NULL) { stop_plugin(); @@ -1090,11 +1059,9 @@ Same behavior as `-V`; see that option for details. Short alias: `-V`. while (!tcp_relay.closed() || !udp_relay.closed()) { tcp_loop.run_nowait(); } + program.close(); #ifdef __MINGW32__ - if (plugin_watcher.valid) { - closesocket(plugin_watcher.fd); - } winsock_cleanup(); #endif diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 6a75a1de..701e4725 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -12,7 +12,7 @@ endif() # Some hosts (notably macOS, whose default soft limit is 256) start test # processes with too few file descriptors for descriptor-heavy tests such as -# test_event. Raise the soft limit before exec'ing each test; a failure to +# test_event_cpp. Raise the soft limit before exec'ing each test; a failure to # raise it (hard limit too low) is ignored so the test still runs. if(NOT WIN32) set(SS_TEST_LAUNCHER sh -c "ulimit -n 1024 2>/dev/null\; exec \"$0\" \"$@\"") @@ -274,10 +274,14 @@ ss_add_test(test_process "ss_sodium") 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/relay/resolver.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp" - "ss_ev;ss_cares;ss_sodium") -set_tests_properties(test_resolv PROPERTIES TIMEOUT 20) +if(UNIX) + # The resolver against a delayed local DNS server. + ss_add_test(test_resolver_cpp + "test_resolver_cpp.cpp;${PROJECT_SOURCE_DIR}/src/relay/resolver.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp" + "ss_ev;ss_cares;ss_sodium") + ss_strict_warnings(test_resolver_cpp.cpp) + set_tests_properties(test_resolver_cpp PROPERTIES TIMEOUT 30) +endif() set(TEST_LITERAL_LIBS ss_sodium) if(SS_ENABLE_REGEX) @@ -287,12 +291,6 @@ ss_add_test(test_acl_literal "test_acl_literal.c;${PROJECT_SOURCE_DIR}/src/acl.cpp;${PROJECT_SOURCE_DIR}/src/ipset.cpp;${PROJECT_SOURCE_DIR}/src/rule.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp" "${TEST_LITERAL_LIBS}") -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/relay/resolver.cpp;${PROJECT_SOURCE_DIR}/src/utils.cpp" - "ss_ev;ss_cares;ss_sodium") -set_tests_properties(test_resolv_async PROPERTIES TIMEOUT 30) # Preloadable malloc counter for the allocation audit. It interposes the # system allocator, so it is only built where that is supported (glibc, diff --git a/tests/test_event.c b/tests/test_event.c deleted file mode 100644 index 31181084..00000000 --- a/tests/test_event.c +++ /dev/null @@ -1,88 +0,0 @@ -#include "test_helpers.h" -#include -#include "ss_event.h" -#include -#ifdef __APPLE__ -#include -#endif -int verbose = 0; -struct receiver { ss_io io; int count; }; -static void receive(struct ss_loop *loop, ss_io *io, int events) -{ - struct receiver *r = (struct receiver *)io; - char byte; - assert(events & SS_READ); - assert(recv(io->fd, &byte, 1, 0) == 1 && byte == 'x'); - r->count++; - ss_io_stop(loop, io); -} -struct shared { ss_io io; ss_io *other; int *calls; }; -static void destroy_peer(struct ss_loop *loop, ss_io *io, int events) -{ - (void)events; - struct shared *s = (struct shared *)io; - (*s->calls)++; - ss_io_stop(loop, s->other); - free(s->other); - ss_io_stop(loop, io); - free(s); -} -static void must_not_run(struct ss_loop *loop, ss_io *io, int events) -{ - (void)loop; (void)io; (void)events; - assert(!"stopped watcher was invoked"); -} -int main(void) -{ - test_network_init(); - /* 256 receivers + sender + loop internals must fit under RLIMIT_NOFILE. */ - test_raise_fd_limit(512); - struct ss_loop *loop = ss_loop_new(0); - assert(loop); -#ifdef _WIN32 - assert(strcmp(ss_backend_name(), "iocp") == 0); - assert(ss_native_loop(loop)->iocp != NULL); -#elif defined(__APPLE__) - assert(strcmp(ss_backend_name(), "kqueue") == 0); - struct timespec zero = {0}; - assert(kevent(uv_backend_fd(ss_native_loop(loop)), NULL, 0, NULL, 0, &zero) == 0); -#endif - /* Exceed Winsock select's default 64 descriptors. */ - struct receiver receivers[256] = {0}; - ss_socket_t sender = socket(AF_INET, SOCK_DGRAM, 0); - assert((int)sender >= 0); - for (unsigned i = 0; i < 256; i++) { - int fd = (int)socket(AF_INET, SOCK_DGRAM, 0); - assert(fd >= 0); - struct sockaddr_in address = {0}; - address.sin_family = AF_INET; - address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); - assert(bind(fd, (struct sockaddr *)&address, sizeof(address)) == 0); - socklen_t length = sizeof(address); - assert(getsockname(fd, (struct sockaddr *)&address, &length) == 0); - ss_io_init(&receivers[i].io, receive, fd, SS_READ); - ss_io_start(loop, &receivers[i].io); - assert(sendto(sender, "x", 1, 0, (struct sockaddr *)&address, length) == 1); - } - ss_run(loop, 0); - for (unsigned i = 0; i < 256; i++) { - assert(receivers[i].count == 1); - ss_socket_close(receivers[i].io.fd); - } - /* Both directions share one poll handle; freeing a pending peer is safe. */ - int calls = 0; - ss_io *other = calloc(1, sizeof(*other)); - struct shared *owner = calloc(1, sizeof(*owner)); - assert(other && owner); - ss_io_init(other, must_not_run, (int)sender, SS_WRITE); - ss_io_init(&owner->io, destroy_peer, (int)sender, SS_WRITE); - owner->other = other; owner->calls = &calls; - ss_io_start(loop, other); - ss_io_start(loop, &owner->io); - ss_run(loop, 0); - assert(calls == 1); - ss_socket_close(sender); - assert(ss_run(loop, SS_RUN_NOWAIT) == 0); - ss_loop_destroy(loop); - return 0; -} diff --git a/tests/test_resolv.c b/tests/test_resolv.c deleted file mode 100644 index 743811e5..00000000 --- a/tests/test_resolv.c +++ /dev/null @@ -1,56 +0,0 @@ -/* Exercise real c-ares cancellation with outstanding A and AAAA requests. */ -#include "test_helpers.h" -#include "ss_event.h" -#include "resolv.h" - -int verbose = 0; -struct result { unsigned called, freed; }; -static void resolved(struct sockaddr *address, void *data) -{ - struct result *result = data; - assert(address == NULL); - assert(result->freed == 0); - result->called++; -} -static void released(void *data) -{ - struct result *result = data; - assert(result->called == 1); - result->freed++; -} -int main(void) -{ - test_network_init(); - /* A bound but unread UDP socket prevents DNS responses or ICMP rejection; - * the test needs no external DNS server or connectivity. */ - ss_socket_t sink = socket(AF_INET, SOCK_DGRAM, 0); - struct sockaddr_in address = {0}; - address.sin_family = AF_INET; - address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); - assert(bind(sink, (struct sockaddr *)&address, sizeof(address)) == 0); - socklen_t size = sizeof(address); - assert(getsockname(sink, (struct sockaddr *)&address, &size) == 0); - char nameserver[64]; - snprintf(nameserver, sizeof(nameserver), "127.0.0.1:%u", ntohs(address.sin_port)); - struct ss_loop *loop = ss_loop_new(0); - assert(loop != NULL); - for (unsigned round = 0; round < 3; round++) { - struct result results[64] = {{0}}; - assert(resolv_init(loop, nameserver, 0) == 0); - for (unsigned i = 0; i < 64; i++) { - char hostname[64]; - snprintf(hostname, sizeof(hostname), "pending-%u.invalid", i); - resolv_start(hostname, htons(443), resolved, released, &results[i]); - } - ss_run(loop, SS_RUN_NOWAIT); - resolv_shutdown(loop); - for (unsigned i = 0; i < 64; i++) { - assert(results[i].called == 1); - assert(results[i].freed == 1); - } - assert(ss_run(loop, SS_RUN_NOWAIT) == 0); - } - ss_loop_destroy(loop); - ss_socket_close(sink); - return 0; -} diff --git a/tests/test_resolv_async.c b/tests/test_resolv_async.c deleted file mode 100644 index 164af217..00000000 --- a/tests/test_resolv_async.c +++ /dev/null @@ -1,140 +0,0 @@ -/* Local delayed DNS server proves A/AAAA/NXDOMAIN completion without blocking. */ -#include "test_helpers.h" -#include -#include "ss_event.h" -#include "resolv.h" -int verbose = 0; -struct packet { - struct packet *next; - struct sockaddr_storage peer; - socklen_t peer_len; - unsigned char bytes[512]; - size_t length; -}; -static struct packet *packets; -static ss_io dns_io; -static ss_timer reply_timer, ticker, deadline; -static unsigned ticks, completed; -static int family; -struct result { unsigned called, freed; int missing; }; -static void reply(struct ss_loop *loop, ss_timer *timer, int events) -{ - (void)loop; (void)timer; (void)events; - while (packets) { - struct packet *p = packets; - packets = p->next; - assert(sendto(dns_io.fd, (const char *)p->bytes, (int)p->length, 0, - (struct sockaddr *)&p->peer, p->peer_len) == (int)p->length); - free(p); - } -} -static void question(struct ss_loop *loop, ss_io *io, int events) -{ - (void)events; - struct packet *p = calloc(1, sizeof(*p)); - assert(p); - p->peer_len = sizeof(p->peer); - int count = (int)recvfrom(io->fd, (char *)p->bytes, sizeof(p->bytes), 0, - (struct sockaddr *)&p->peer, &p->peer_len); - assert(count > 16 && p->bytes[4] == 0 && p->bytes[5] == 1); - size_t end = 12; - while (end < (size_t)count && p->bytes[end]) end += p->bytes[end] + 1; - end += 5; /* terminating zero, QTYPE, QCLASS */ - assert(end <= (size_t)count && end + 28 <= sizeof(p->bytes)); - unsigned type = p->bytes[end - 3]; - int missing = p->bytes[13] == 'm'; - p->bytes[2] = 0x81; p->bytes[3] = missing ? 0x83 : 0x80; - p->bytes[6] = 0; p->bytes[7] = missing ? 0 : 1; - memset(p->bytes + 8, 0, 4); - p->length = end; - if (!missing) { - unsigned char answer[] = {0xc0,0x0c,0,0,0,1,0,0,0,1,0,0}; - assert(type == 1 || type == 28); - answer[3] = (unsigned char)type; - answer[11] = type == 1 ? 4 : 16; - memcpy(p->bytes + end, answer, sizeof(answer)); - p->length += sizeof(answer); - memset(p->bytes + p->length, 0, 16); - if (type == 1) { p->bytes[p->length] = 127; p->bytes[p->length+3] = 1; } - else p->bytes[p->length+15] = 1; - p->length += answer[11]; - } - p->next = packets; packets = p; - ss_timer_start(loop, &reply_timer); -} -static void tick(struct ss_loop *loop, ss_timer *timer, int events) -{ - (void)loop; (void)timer; (void)events; ticks++; -} -static void timed_out(struct ss_loop *loop, ss_timer *timer, int events) -{ - (void)loop; (void)timer; (void)events; - assert(!"asynchronous DNS did not complete"); -} -static struct ss_loop *test_loop; -static void resolved(struct sockaddr *address, void *data) -{ - struct result *r = data; - assert(!r->called && !r->freed && ticks > 2); - r->called++; - if (r->missing) assert(address == NULL); - else { - assert(address && address->sa_family == family); - if (family == AF_INET) assert(((struct sockaddr_in *)address)->sin_port == htons(443)); - else assert(((struct sockaddr_in6 *)address)->sin6_port == htons(443)); - } - if (++completed == 33) ss_unloop(test_loop, SS_UNLOOP_ALL); -} -static void released(void *data) -{ - struct result *r = data; - assert(r->called == 1 && !r->freed); - r->freed++; -} -int main(void) -{ - test_network_init(); - test_loop = ss_loop_new(0); - int fd = (int)socket(AF_INET, SOCK_DGRAM, 0); - struct sockaddr_in address = {0}; - address.sin_family = AF_INET; - address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); - assert(bind(fd, (struct sockaddr *)&address, sizeof(address)) == 0); - socklen_t length = sizeof(address); - assert(getsockname(fd, (struct sockaddr *)&address, &length) == 0); - char nameserver[64]; - snprintf(nameserver, sizeof(nameserver), "127.0.0.1:%u", ntohs(address.sin_port)); - ss_io_init(&dns_io, question, fd, SS_READ); - ss_timer_init(&reply_timer, reply, .05, 0); - ss_timer_init(&ticker, tick, .001, .001); - ss_timer_init(&deadline, timed_out, 10, 0); - for (int ipv6 = 0; ipv6 < 2; ipv6++) { - struct result results[33] = {{0}}; - results[32].missing = 1; - family = ipv6 ? AF_INET6 : AF_INET; - ticks = completed = 0; - assert(resolv_init(test_loop, nameserver, ipv6) == 0); - ss_io_start(test_loop, &dns_io); - ss_timer_start(test_loop, &ticker); - ss_timer_start(test_loop, &deadline); - for (unsigned i = 0; i < 33; i++) { - char host[64]; - snprintf(host, sizeof(host), "%s-%u.invalid", i == 32 ? "missing" : "resolved", i); - resolv_start(host, htons(443), resolved, released, &results[i]); - } - assert(completed == 0); - ss_run(test_loop, 0); - assert(completed == 33); - for (unsigned i = 0; i < 33; i++) assert(results[i].called == 1 && results[i].freed == 1); - resolv_shutdown(test_loop); - ss_io_stop(test_loop, &dns_io); - ss_timer_stop(test_loop, &ticker); - ss_timer_stop(test_loop, &deadline); - ss_timer_stop(test_loop, &reply_timer); - assert(!packets); - assert(ss_run(test_loop, SS_RUN_NOWAIT) == 0); - } - ss_socket_close(fd); - ss_loop_destroy(test_loop); - return 0; -} diff --git a/tests/test_resolver_cpp.cpp b/tests/test_resolver_cpp.cpp new file mode 100644 index 00000000..348c45d6 --- /dev/null +++ b/tests/test_resolver_cpp.cpp @@ -0,0 +1,211 @@ +/* SPDX-License-Identifier: GPL-3.0-or-later */ +// ss::relay::Resolver against a local DNS server that answers after a delay: +// A/AAAA/NXDOMAIN complete without blocking the loop, the query pool bounds +// the number in flight, and shutdown reports pending queries as failed. +#include +#include + +#include +#include +#include +#include + +#include "event/loop.hpp" +#include "relay/resolver.hpp" +#include "test_helpers.hpp" + +extern "C" { +int verbose = 0; +} + +using namespace ss; + +namespace { + +struct Answered { + unsigned called = 0; + bool missing = false; + int family = 0; + bool ok = false; +}; + +int expected_family = AF_INET; +unsigned completed = 0; +event::Loop *test_loop = nullptr; +unsigned stop_after = 0; + +void on_answer(const relay::ResolveCallback &cb, const sockaddr *address) noexcept +{ + auto *r = static_cast(cb.ctx); + ++r->called; + if (address != nullptr) { + r->family = address->sa_family; + r->ok = address->sa_family == AF_INET + ? reinterpret_cast(address)->sin_port == htons(443) + : reinterpret_cast(address)->sin6_port == htons(443); + } + if (++completed == stop_after) { + test_loop->stop(); + } +} + +// Answers every question 50 ms later: names starting with 'm' do not exist. +class FakeDns { +public: + void on_tick() noexcept + { + for (auto &p : pending_) { + (void)sendto(poll_.fd(), p.bytes, p.length, 0, reinterpret_cast(&p.peer), p.peer_len); + } + pending_.clear(); + } + event::Flow on_question(int) noexcept + { + Packet p{}; + p.peer_len = sizeof(p.peer); + ssize_t n = recvfrom(poll_.fd(), p.bytes, sizeof(p.bytes), 0, reinterpret_cast(&p.peer), + &p.peer_len); + if (n < 17) { + return event::Flow::proceed; + } + std::size_t end = 12; + while (end < static_cast(n) && p.bytes[end] != 0) { + end += p.bytes[end] + 1; + } + end += 5; + unsigned type = p.bytes[end - 3]; + bool missing = p.bytes[13] == 'm'; + p.bytes[2] = 0x81; + p.bytes[3] = missing ? 0x83 : 0x80; + p.bytes[6] = 0; + p.bytes[7] = missing ? 0 : 1; + std::memset(p.bytes + 8, 0, 4); + p.length = end; + if (!missing) { + unsigned char answer[] = { 0xc0, 0x0c, 0, static_cast(type), 0, 1, 0, 0, 0, 1, 0, + static_cast(type == 1 ? 4 : 16) }; + std::memcpy(p.bytes + end, answer, sizeof(answer)); + p.length += sizeof(answer); + std::memset(p.bytes + p.length, 0, 16); + if (type == 1) { + p.bytes[p.length] = 127; + p.bytes[p.length + 3] = 1; + } else { + p.bytes[p.length + 15] = 1; + } + p.length += type == 1 ? 4 : 16; + } + pending_.push_back(p); + timer_.start(50); + return event::Flow::proceed; + } + + void open(event::Loop &loop, int fd) + { + SS_CHECK(poll_.open(loop, *this, fd)); + SS_CHECK(poll_.want(event::readable)); + SS_CHECK(timer_.open(loop, *this)); + } + void close(event::CloseGroup &g) + { + poll_.close(g); + timer_.close(g); + } + +private: + struct Packet { + sockaddr_storage peer; + socklen_t peer_len; + unsigned char bytes[512]; + std::size_t length; + }; + std::vector pending_; + event::Poll poll_; + event::Timer timer_; +}; + +void drain(event::Loop &loop, bool &flag) +{ + while (!flag) { + loop.run_nowait(); + } +} + +void test_answers_and_limits() +{ + for (int ipv6 = 0; ipv6 < 2; ++ipv6) { + event::Loop loop; + SS_CHECK(loop.init()); + test_loop = &loop; + int fd = socket(AF_INET, SOCK_DGRAM, 0); + sockaddr_in a{}; + a.sin_family = AF_INET; + a.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + SS_CHECK(bind(fd, reinterpret_cast(&a), sizeof(a)) == 0); + socklen_t len = sizeof(a); + SS_CHECK(getsockname(fd, reinterpret_cast(&a), &len) == 0); + char nameserver[64]; + std::snprintf(nameserver, sizeof(nameserver), "127.0.0.1:%u", ntohs(a.sin_port)); + FakeDns dns; + dns.open(loop, fd); + + relay::Resolver resolver; + SS_CHECK(resolver.init(loop, nameserver, ipv6 != 0, 33)); + expected_family = ipv6 ? AF_INET6 : AF_INET; + completed = 0; + stop_after = 33; + Answered results[34]; + results[32].missing = true; + for (unsigned i = 0; i < 34; ++i) { + char host[64]; + std::snprintf(host, sizeof(host), "%s-%u.invalid", i == 32 ? "missing" : "resolved", i); + relay::ResolveCallback cb; + cb.fn = &on_answer; + cb.ctx = &results[i]; + auto st = resolver.resolve(host, htons(443), cb); + // The pool holds 33 queries; the 34th is refused up front. + SS_CHECK(i < 33 ? static_cast(st) : (!st && st.error().code == Errc::exhausted)); + } + SS_CHECK(completed == 0 && resolver.in_flight() == 33); + loop.run(); + SS_CHECK(completed == 33 && resolver.in_flight() == 0); + for (unsigned i = 0; i < 33; ++i) { + SS_CHECK(results[i].called == 1); + if (results[i].missing) { + SS_CHECK(!results[i].ok && results[i].family == 0); + } else { + SS_CHECK(results[i].ok && results[i].family == expected_family); + } + } + SS_CHECK(results[33].called == 0); + + // Shutting down with queries in flight reports them as failed. + Answered late; + relay::ResolveCallback cb; + cb.fn = &on_answer; + cb.ctx = ⪭ + SS_CHECK(resolver.resolve("resolved-late.invalid", htons(443), cb)); + static bool closed; + closed = false; + resolver.shutdown(CloseCompletion([](void *, std::uint32_t) { closed = true; }, nullptr, 0)); + drain(loop, closed); + SS_CHECK(late.called == 1 && !late.ok); + + event::CloseGroup g; + static bool dns_closed; + dns_closed = false; + g.arm(CloseCompletion([](void *, std::uint32_t) { dns_closed = true; }, nullptr, 0)); + dns.close(g); + g.seal(); + drain(loop, dns_closed); + SS_CHECK(loop.shutdown()); + } +} + +} // namespace + +int main() +{ + SS_RUN(test_answers_and_limits); + return 0; +}