Remove ss_event and the C resolver shim

Each program's loop, signals, plugin watcher and stats timer now go
through ss::event::Program (src/event/program.hpp), held on main()'s
stack; ss-manager uses it too. ss-server resolves the manager address
with the relay resolver. With no C caller left, ss_event.cpp/.h,
resolv.cpp/.h and their C tests go; test_resolver_cpp.cpp replaces the
resolver tests against the same delayed local DNS server, and ss_ev is
now an interface target over libuv.

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)
 

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()

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)

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

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})

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 <array>
+#include <csignal>
+#include <cstdint>
+
+#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<SignalWatch, &SignalWatch::on_signal> signal;
+    };
+    struct SocketWatch {
+        Callback fn = nullptr;
+        void *ctx   = nullptr;
+        Flow on_ready(int) noexcept
+        {
+            fn(ctx);
+            return Flow::proceed;
+        }
+        Poll<SocketWatch, &SocketWatch::on_ready> poll;
+    };
+    struct Wake {
+        Program *program = nullptr;
+        void on_wake() noexcept { program->stop(0); }
+        Async<Wake, &Wake::on_wake> async;
+    };
+    struct Tick {
+        Callback fn = nullptr;
+        void *ctx   = nullptr;
+        void on_fire() noexcept { fn(ctx); }
+        Timer<Tick, &Tick::on_fire> 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<SignalWatch, 4> signals_{};
+    std::size_t signal_count_ = 0;
+    std::array<SocketWatch, 4> 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

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<ss::event::Program *>(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();

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);
 

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;

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)

src/resolv.cpp

@@ -1,123 +0,0 @@
-/*
- * Copyright (c) 2014, Dustin Lundquist <dustin@null-ptr.net>
- * 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 <cstdint>
-#include <cstring>
-#include <memory>
-#include <new>
-
-#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<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);
-    }
-}
-
-} // 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<ss::event::Loop *>(loop_storage));
-    if (!loop->adopt(ss_native_loop(ss_loop))) {
-        FATAL("failed to initialize c-ares");
-    }
-    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");
-    }
-    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<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);
-    }
-}
-
-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;
-}

src/resolv.h

@@ -1,55 +0,0 @@
-/*
- * Copyright (c) 2014, Dustin Lundquist <dustin@null-ptr.net>
- * 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 <stdint.h>
-#ifndef __MINGW32__
-#include <sys/socket.h>
-#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

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 <libnetfilter_conntrack/libnetfilter_conntrack.h>
@@ -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<ss::event::Program *>(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

src/ss_event.cpp

@@ -1,278 +0,0 @@
-#include "ss_event.h"
-#include <uv.h>
-#include "uthash.h"
-#include <assert.h>
-#include <math.h>
-#include <signal.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-
-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);
-}

src/ss_event.h

@@ -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 <stdint.h>
-
-#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

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<ss::event::Program *>(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

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,

tests/test_event.c

@@ -1,88 +0,0 @@
-#include "test_helpers.h"
-#include <string.h>
-#include "ss_event.h"
-#include <uv.h>
-#ifdef __APPLE__
-#include <sys/event.h>
-#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;
-}

tests/test_resolv.c

@@ -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;
-}

tests/test_resolv_async.c

@@ -1,140 +0,0 @@
-/* Local delayed DNS server proves A/AAAA/NXDOMAIN completion without blocking. */
-#include "test_helpers.h"
-#include <string.h>
-#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;
-}

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 <cstring>
+#include <vector>
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#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<Answered *>(cb.ctx);
+    ++r->called;
+    if (address != nullptr) {
+        r->family = address->sa_family;
+        r->ok     = address->sa_family == AF_INET
+                    ? reinterpret_cast<const sockaddr_in *>(address)->sin_port == htons(443)
+                    : reinterpret_cast<const sockaddr_in6 *>(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<sockaddr *>(&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<sockaddr *>(&p.peer),
+                              &p.peer_len);
+        if (n < 17) {
+            return event::Flow::proceed;
+        }
+        std::size_t end = 12;
+        while (end < static_cast<std::size_t>(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<unsigned char>(type), 0, 1, 0, 0, 0, 1, 0,
+                                       static_cast<unsigned char>(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<Packet> pending_;
+    event::Poll<FakeDns, &FakeDns::on_question> poll_;
+    event::Timer<FakeDns, &FakeDns::on_tick> 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<sockaddr *>(&a), sizeof(a)) == 0);
+        socklen_t len = sizeof(a);
+        SS_CHECK(getsockname(fd, reinterpret_cast<sockaddr *>(&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<bool>(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 = &late;
+        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;
+}