From 772873104658182a5b76f5c0eed76047c62c6651 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Wed, 28 Oct 2026 23:31:07 +0100 Subject: With more workers blockingly waiting on udp sockets, we need to keep track of all sockets to shut them down on exit --- opentracker.c | 9 +++--- ot_socket.c | 7 +++++ ot_socket.h | 4 +++ ot_udp.c | 42 +++++++++++++++++++++++++-- ot_udp.h | 3 ++ tests/integration/test_shutdown.py | 58 ++++++++++++++++++++++++++++++++++++++ tests/t_socket.c | 40 ++++++++++++++++++++++++++ 7 files changed, 156 insertions(+), 7 deletions(-) create mode 100644 tests/integration/test_shutdown.py diff --git a/opentracker.c b/opentracker.c index f579b92..ab4a934 100644 --- a/opentracker.c +++ b/opentracker.c @@ -397,7 +397,6 @@ static void listen_start_udp_workers(void) { if (entry->proto != FLAG_UDP || entry->workers <= 0) continue; - socket_set_blocking((int)entry->sock, 1); event_del((int)entry->sock); udp_worker_install(entry->sock); @@ -409,8 +408,7 @@ static void listen_start_udp_workers(void) { if (sock == -1) { fprintf(stderr, "Warning: could not bind another socket for udp worker %d: %s. Sharing the first instead.\n", worker, strerror(errno)); sock = entry->sock; - } else - socket_set_blocking((int)sock, 1); + } } udp_worker_install(sock); } @@ -548,9 +546,10 @@ int main(int argc, char **argv) { listen_start_udp_workers(); server_mainloop(0); + udp_workers_deinit(); - /* The loop has left, so no thread is serving any more and the teardown - can run where a debugger and the sanitizers can follow it */ + /* No thread is serving any more, so the teardown can run where a debugger + and the sanitizers can follow it */ trackerlogic_deinit(); #ifdef WANT_SYSLOGS closelog(); diff --git a/ot_socket.c b/ot_socket.c index 158459f..c73738c 100644 --- a/ot_socket.c +++ b/ot_socket.c @@ -9,6 +9,7 @@ #include #include #include +#include #include /* Opentracker */ @@ -48,6 +49,12 @@ int socket_set_blocking(int s, int blocking) { return fcntl(s, F_SETFL, blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK)); } +int socket_set_timeout(int s, int seconds) { + struct timeval patience = {seconds, 0}; + + return setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, &patience, sizeof(patience)); +} + static int set_nonblocking(int s) { return socket_set_blocking(s, 0); } /* Close without letting the failure we are about to report be overwritten */ diff --git a/ot_socket.h b/ot_socket.h index f73a5c9..9340d0d 100644 --- a/ot_socket.h +++ b/ot_socket.h @@ -39,6 +39,10 @@ int socket_is_connected(int s); int socket_accept(int s, ot_ip6 ip, uint16_t *port); int socket_set_blocking(int s, int blocking); +/* Bound how long a blocking read waits, so the thread sitting in one gets to + look at the running flag now and then */ +int socket_set_timeout(int s, int seconds); + ssize_t socket_recv(int s, void *buf, size_t len, ot_ip6 ip, uint16_t *port); ssize_t socket_send(int s, const void *buf, size_t len, const ot_ip6 ip, uint16_t port); diff --git a/ot_udp.c b/ot_udp.c index 87741da..7aa0e61 100644 --- a/ot_udp.c +++ b/ot_udp.c @@ -9,6 +9,7 @@ #include #include #include +#include /* Opentracker */ #include "ot_ip.h" @@ -19,6 +20,17 @@ #include "ot_udp.h" #include "trackerlogic.h" +/* How long a worker may sit in a read before it looks at the running flag */ +#define OT_UDP_WORKER_PATIENCE 1 + +struct udp_worker { + pthread_t thread; + int64_t sock; +}; + +static struct udp_worker *g_workers; +static size_t g_worker_count; + static uint32_t g_rijndael_round_key[44] = {0}; static volatile uint32_t g_key_of_the_hour[2] = {0}; static volatile ot_time g_hour_of_the_key; @@ -236,11 +248,37 @@ void udp_init(void) { } void udp_worker_install(int64_t sock) { - pthread_t thread_id; + struct udp_worker *grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers)); + + if (!grown) { + fprintf(stderr, "Warning: out of memory installing a udp worker, this socket stays with the workers already running.\n"); + return; + } + g_workers = grown; #ifdef _DEBUG fprintf(stderr, "Installing a worker on udp socket %ld\n", (unsigned long)sock); #endif - pthread_create(&thread_id, NULL, udp_worker, (void *)sock); + socket_set_blocking((int)sock, 1); + socket_set_timeout((int)sock, OT_UDP_WORKER_PATIENCE); + + g_workers[g_worker_count].sock = sock; + if (!pthread_create(&g_workers[g_worker_count].thread, NULL, udp_worker, (void *)sock)) + ++g_worker_count; +} + +void udp_workers_deinit(void) { + size_t i; + + /* The receive timeout bounds this on its own, shutting the socket down + cuts it short wherever the platform wakes a blocked reader */ + for (i = 0; i < g_worker_count; ++i) + shutdown((int)g_workers[i].sock, SHUT_RD); + for (i = 0; i < g_worker_count; ++i) + pthread_join(g_workers[i].thread, NULL); + + free(g_workers); + g_workers = NULL; + g_worker_count = 0; } diff --git a/ot_udp.h b/ot_udp.h index 7d4432f..b731aca 100644 --- a/ot_udp.h +++ b/ot_udp.h @@ -13,6 +13,9 @@ void udp_init(void); /* One thread reading one socket. Called once per worker. */ void udp_worker_install(int64_t sock); + +/* Stop every worker installed above and wait for it to be gone */ +void udp_workers_deinit(void); int handle_udp6(int64_t serversocket, struct ot_workstruct *ws); #endif diff --git a/tests/integration/test_shutdown.py b/tests/integration/test_shutdown.py new file mode 100644 index 0000000..063d779 --- /dev/null +++ b/tests/integration/test_shutdown.py @@ -0,0 +1,58 @@ +"""That every thread the tracker starts also ends when it is asked to stop. + +The rest of the suite stops a tracker too, so a thread left behind already +shows up in teardown as the complaint that the tracker had to be killed. +These ask for the arrangements that park a thread in a blocking read in the +first place, and they time the exit, because a reader woken only by its own +receive timeout passes a correctness check while taking a second over it. +""" + +import time +import unittest + +from harness import Tracker, TrackerTestCase + +# Generous enough for a platform where shutting the socket down does not wake +# the reader and the receive timeout has to expire, tight enough that a reader +# which is never woken at all fails instead of hanging the suite +PATIENCE = 3.0 + + +def time_the_exit(extra_config): + tracker = Tracker(extra_config) + try: + # Serve one request of each kind first, so the threads that are about + # to be joined have actually run + tracker.announce(b"shutdown-probe\0\0\0\0\0\0") + sock, connection_id = tracker.udp_connect() + tracker.udp_announce(sock, connection_id, b"shutdown-probe\0\0\0\0\0\0") + sock.close() + started = time.time() + complaints = tracker.stop() + return time.time() - started, complaints + finally: + tracker._cleanup() + + +class UdpWorkers(TrackerTestCase): + config = "listen.udp.workers 8\n" + + def test_announce_reaches_one_of_eight_workers(self): + """Keeps the workers from being merely started and never used.""" + sock, connection_id = self.tracker.udp_connect() + answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("eight-workers")) + sock.close() + self.assertEqual(answer["seeders"], 1) + + +class ShutdownEndsEveryThread(unittest.TestCase): + def test_eight_udp_workers_are_joined(self): + took, complaints = time_the_exit("listen.udp.workers 8\n") + self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints)) + self.assertLess(took, PATIENCE, "stopping eight udp workers took %.1f s" % took) + + def test_udp_in_the_event_loop_still_stops(self): + """Worker count zero leaves no thread to join, which must also work.""" + took, complaints = time_the_exit("listen.udp.workers 0\n") + self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints)) + self.assertLess(took, PATIENCE, "stopping without workers took %.1f s" % took) diff --git a/tests/t_socket.c b/tests/t_socket.c index 8eef7fd..ede8ae0 100644 --- a/tests/t_socket.c +++ b/tests/t_socket.c @@ -10,6 +10,7 @@ #include #include #include +#include #include /* Opentracker */ @@ -144,6 +145,44 @@ static void case_udp_roundtrip(void) { close(s); } +/* udp_workers_deinit relies on a worker parked in a read coming back. The + receive timeout is what guarantees that; shutting the socket down is what + makes it prompt where the platform allows it. Neither is uniform across the + systems opentracker runs on, so both are checked rather than assumed. */ +static void case_blocking_read_comes_back(void) { + char buf[32], ip[16], own_ip[16]; + uint16_t port = 0, own_port = 0; + time_t started; + int s; + + test_case("a receive timeout does not delay a datagram that is already there"); + s = socket_udp_bind(ip_loopback, 0, SOCKET_EXCLUSIVE); + CHECK(s != -1); + CHECK_INT(socket_set_blocking(s, 1), 0); + CHECK_INT(socket_set_timeout(s, 1), 0); + bound_address(s, own_ip, &own_port); + CHECK_INT(socket_send(s, "opentracker", 11, own_ip, own_port), 11); + CHECK_INT(socket_recv(s, buf, sizeof(buf), ip, &port), 11); + + test_case("and ends a read that would otherwise never return"); + started = time(NULL); + CHECK_INT(socket_recv(s, buf, sizeof(buf), ip, &port), -1); + CHECK(errno == EAGAIN || errno == EWOULDBLOCK); + CHECK(time(NULL) - started <= 3); + close(s); + + test_case("shutting the socket down ends one too"); + s = socket_udp_bind(ip_loopback, 0, SOCKET_EXCLUSIVE); + CHECK(s != -1); + CHECK_INT(socket_set_blocking(s, 1), 0); + CHECK_INT(socket_set_timeout(s, 1), 0); + shutdown(s, SHUT_RD); + started = time(NULL); + CHECK(socket_recv(s, buf, sizeof(buf), ip, &port) <= 0); + CHECK(time(NULL) - started <= 3); + close(s); +} + /* An IPv4 peer must arrive in the v4 mapped form opentracker stores, however the kernel chose to report it. Linux fills a v4 mapped sockaddr_in6 here and the BSDs fill a plain sockaddr_in, so this is the case that tells the @@ -293,6 +332,7 @@ void test_socket(void) { case_sharing_is_opt_in(); #endif case_udp_roundtrip(); + case_blocking_read_comes_back(); case_udp_v4mapped(); case_tcp_roundtrip(); case_connect_refused(); -- cgit v1.3