From 0f95c7d1056cab5bd8c6334cf601349ac4631457 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Thu, 29 Oct 2026 00:53:50 +0100 Subject: Revert last wake-up experiment. Having each udp worker wake up once a second to check if the tracker stops is wasteful --- opentracker.c | 7 +++-- ot_livesync.c | 12 ++++---- ot_socket.c | 7 ----- ot_socket.h | 4 --- ot_udp.c | 58 ++++++++++++++++++++++-------------- tests/t_socket.c | 91 +++++++++++++++++++++++++++++++++----------------------- trackerlogic.h | 4 +++ 7 files changed, 105 insertions(+), 78 deletions(-) diff --git a/opentracker.c b/opentracker.c index ab4a934..e0f16d7 100644 --- a/opentracker.c +++ b/opentracker.c @@ -49,7 +49,9 @@ static void panic(const char *routine) { exit(111); } -/* Say we are stopping, wake the loop, and let main do the work on its own stack. */ +/* Say we are stopping, wake the loop, and let main do the work on its own + stack. Everything other than SIGINT falls through and does nothing, which + is what OT_SIGNAL_WAKE wants: returning from the read is all it is for. */ static void signal_handler(int s) { if (s == SIGINT) { const char byte = 0; @@ -85,7 +87,8 @@ static void install_signal_handlers(void) { sa.sa_handler = signal_handler; sigemptyset(&sa.sa_mask); sa.sa_flags = 0; - if ((sigaction(SIGINT, &sa, NULL) == -1) || (sigaction(SIGALRM, &sa, NULL) == -1)) + if ((sigaction(SIGINT, &sa, NULL) == -1) || (sigaction(SIGALRM, &sa, NULL) == -1) || + (sigaction(OT_SIGNAL_WAKE, &sa, NULL) == -1)) panic("install_signal_handlers"); sigaddset(&signal_mask, SIGINT); diff --git a/ot_livesync.c b/ot_livesync.c index e6db720..4a0ede1 100644 --- a/ot_livesync.c +++ b/ot_livesync.c @@ -32,9 +32,6 @@ ot_ip6 groupip_1 = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 224, 0, 23, 5}; #define LIVESYNC_MAXDELAY 15 /* seconds */ -/* How long the worker may sit in a read before it looks at the running flag */ -#define LIVESYNC_WORKER_PATIENCE 1 /* seconds */ - enum { OT_SYNC_PEER4, OT_SYNC_PEER6 }; /* Forward declaration */ @@ -58,6 +55,7 @@ static sync_buffer g_v4_buf; static pthread_t thread_id; static int thread_started; +static _Atomic int thread_left; void livesync_init(void) { uint32_t packet_type6 = htonl(OT_SYNC_PEER6); uint32_t packet_type4 = htonl(OT_SYNC_PEER4); @@ -78,13 +76,16 @@ void livesync_init(void) { g_v6_buf.next_packet_time = time_future_seconds(LIVESYNC_MAXDELAY); g_v4_buf.next_packet_time = time_future_seconds(LIVESYNC_MAXDELAY); - socket_set_timeout(g_socket_in, LIVESYNC_WORKER_PATIENCE); thread_started = !pthread_create(&thread_id, NULL, livesync_worker, NULL); } void livesync_deinit(void) { if (thread_started) { - shutdown(g_socket_in, SHUT_RD); + /* Repeats so it interrupts a read already under way */ + while (!atomic_load(&thread_left)) { + pthread_kill(thread_id, OT_SIGNAL_WAKE); + usleep(1000); + } pthread_join(thread_id, NULL); thread_started = 0; } @@ -235,6 +236,7 @@ static void *livesync_worker(void *args) { } free(ws.inbuf); + atomic_store(&thread_left, 1); return NULL; } diff --git a/ot_socket.c b/ot_socket.c index c73738c..158459f 100644 --- a/ot_socket.c +++ b/ot_socket.c @@ -9,7 +9,6 @@ #include #include #include -#include #include /* Opentracker */ @@ -49,12 +48,6 @@ 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 9340d0d..f73a5c9 100644 --- a/ot_socket.h +++ b/ot_socket.h @@ -39,10 +39,6 @@ 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 7aa0e61..1ec5148 100644 --- a/ot_udp.c +++ b/ot_udp.c @@ -9,7 +9,7 @@ #include #include #include -#include +#include /* Opentracker */ #include "ot_ip.h" @@ -20,16 +20,16 @@ #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 - +/* One of these per worker, each allocated on its own so that growing the + array below never moves the one a running worker holds. */ struct udp_worker { - pthread_t thread; - int64_t sock; + pthread_t thread; + int64_t sock; + _Atomic int left; }; -static struct udp_worker *g_workers; -static size_t g_worker_count; +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}; @@ -218,7 +218,7 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) { } static void *udp_worker(void *args) { - int64_t sock = (int64_t)args; + struct udp_worker *me = args; struct ot_workstruct ws; memset(&ws, 0, sizeof(ws)); @@ -230,7 +230,7 @@ static void *udp_worker(void *args) { #endif while (g_opentracker_running) - handle_udp6(sock, &ws); + handle_udp6(me->sock, &ws); stats_counters_retire(&ws); @@ -239,6 +239,8 @@ static void *udp_worker(void *args) { #ifdef _DEBUG_HTTPERROR free(ws.debugbuf); #endif + + atomic_store(&me->left, 1); return NULL; } @@ -248,34 +250,44 @@ void udp_init(void) { } void udp_worker_install(int64_t sock) { - struct udp_worker *grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers)); + struct udp_worker **grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers)); + struct udp_worker *worker; - if (!grown) { + if (!grown || !(worker = calloc(1, sizeof(*worker)))) { fprintf(stderr, "Warning: out of memory installing a udp worker, this socket stays with the workers already running.\n"); + g_workers = grown ? grown : g_workers; return; } - g_workers = grown; + g_workers = grown; + worker->sock = sock; #ifdef _DEBUG fprintf(stderr, "Installing a worker on udp socket %ld\n", (unsigned long)sock); #endif 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; + if (pthread_create(&worker->thread, NULL, udp_worker, worker)) + free(worker); + else + g_workers[g_worker_count++] = worker; } 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); + for (i = 0; i < g_worker_count; ++i) { + struct udp_worker *worker = g_workers[i]; + + /* Repeated because a signal only interrupts a read already under way: a + worker that was between two reads when the first one arrived is inside + the next one by now. */ + while (!atomic_load(&worker->left)) { + pthread_kill(worker->thread, OT_SIGNAL_WAKE); + usleep(1000); + } + pthread_join(worker->thread, NULL); + free(worker); + } free(g_workers); g_workers = NULL; diff --git a/tests/t_socket.c b/tests/t_socket.c index ede8ae0..ab4b785 100644 --- a/tests/t_socket.c +++ b/tests/t_socket.c @@ -8,14 +8,18 @@ #include #include #include +#include +#include +#include #include #include -#include +#include #include /* Opentracker */ #include "ot_ip.h" #include "ot_socket.h" +#include "trackerlogic.h" #include "test.h" /* These cases talk to the loopback interface rather than to a mock, because @@ -145,42 +149,55 @@ 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); +/* udp_workers_deinit and livesync_deinit end a read another thread is parked + in by signalling that thread. Which mechanisms wake a blocked reader + differs between kernels -- shutting the socket down does so on Linux and + FreeBSD but not on Darwin, closing it does so on Darwin but not on Linux -- + so the one the tracker relies on is checked rather than assumed. The + watchdog is the test's own: where a reader is not woken this fails with a + figure to look at instead of hanging the suite. */ +static _Atomic int reader_returned; +static int reader_socket; + +static void reader_wake(int s) { (void)s; } + +static void *blocked_reader(void *args) { + char buf[32], ip[16]; + uint16_t port = 0; + + (void)args; + socket_recv(reader_socket, buf, sizeof(buf), ip, &port); + atomic_store(&reader_returned, 1); + return NULL; +} - 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); +static void case_a_signal_ends_a_blocking_read(void) { + struct sigaction sa; + struct timeval watchdog = {2, 0}; + pthread_t thread; + int waited; + + test_case("a signal ends a read another thread is parked in"); + memset(&sa, 0, sizeof(sa)); + sa.sa_handler = reader_wake; + sa.sa_flags = 0; /* SA_RESTART would resume the read instead of ending it */ + CHECK_INT(sigaction(OT_SIGNAL_WAKE, &sa, NULL), 0); + + atomic_store(&reader_returned, 0); + reader_socket = socket_udp_bind(ip_loopback, 0, SOCKET_EXCLUSIVE); + CHECK(reader_socket != -1); + CHECK_INT(socket_set_blocking(reader_socket, 1), 0); + CHECK_INT(setsockopt(reader_socket, SOL_SOCKET, SO_RCVTIMEO, &watchdog, sizeof(watchdog)), 0); + + CHECK_INT(pthread_create(&thread, NULL, blocked_reader, NULL), 0); + usleep(100000); /* long enough to be sure it reached the read */ + CHECK_INT(pthread_kill(thread, OT_SIGNAL_WAKE), 0); + + for (waited = 0; waited < 1000 && !atomic_load(&reader_returned); ++waited) + usleep(1000); + CHECK(waited < 1000); + pthread_join(thread, NULL); + close(reader_socket); } /* An IPv4 peer must arrive in the v4 mapped form opentracker stores, however @@ -332,7 +349,7 @@ void test_socket(void) { case_sharing_is_opt_in(); #endif case_udp_roundtrip(); - case_blocking_read_comes_back(); + case_a_signal_ends_a_blocking_read(); case_udp_v4mapped(); case_tcp_roundtrip(); case_connect_refused(); diff --git a/trackerlogic.h b/trackerlogic.h index ad6c911..7d0693a 100644 --- a/trackerlogic.h +++ b/trackerlogic.h @@ -6,6 +6,7 @@ #ifndef OT_TRACKERLOGIC_H__ #define OT_TRACKERLOGIC_H__ +#include #include #include #include @@ -90,6 +91,9 @@ static inline ot_time time_future_seconds(int span) { return time_now_seconds() /* Cleared from the interrupt handler and read by every worker */ extern _Atomic int g_opentracker_running; +/* Sent to a thread parked in a blocking read */ +#define OT_SIGNAL_WAKE SIGUSR1 + extern uint32_t g_tracker_id; typedef enum { FLAG_TCP, FLAG_UDP, FLAG_MCA, FLAG_SELFPIPE } PROTO_FLAG; -- cgit v1.3