From af0f2cb1be5aa10bb6a4bdd9f7ef4aa3950ecb31 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Sun, 1 Nov 2026 00:53:53 +0100 Subject: Generalize the worker setup/runloop/teardown logic by adding a smol ot_worker helper --- Makefile | 4 ++-- opentracker.c | 4 ++-- ot_tcp.c | 66 ++++------------------------------------------------ ot_tcp.h | 2 -- ot_udp.c | 69 ++++-------------------------------------------------- ot_udp.h | 2 -- ot_worker.c | 75 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ ot_worker.h | 39 +++++++++++++++++++++++++++++++ 8 files changed, 128 insertions(+), 133 deletions(-) create mode 100644 ot_worker.c create mode 100644 ot_worker.h diff --git a/Makefile b/Makefile index dbc6166..c9aeafd 100644 --- a/Makefile +++ b/Makefile @@ -69,8 +69,8 @@ CFLAGS_libowfat=-I$(LIBOWFAT_HEADERS) LDFLAGS_libowfat=-L$(LIBOWFAT_LIBRARY) -lowfat BINARY =opentracker -HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.h ot_workqueue.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h -SOURCES=opentracker.c trackerlogic.c ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.c ot_workqueue.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c +HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.h ot_worker.h ot_workqueue.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h +SOURCES=opentracker.c trackerlogic.c ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.c ot_worker.c ot_workqueue.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c SOURCES_proxy=proxy.c ot_batch.c ot_fdstate.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c all: $(BINARY) $(BINARY).debug diff --git a/opentracker.c b/opentracker.c index 2f2746d..644e99c 100644 --- a/opentracker.c +++ b/opentracker.c @@ -38,6 +38,7 @@ #include "ot_stats.h" #include "ot_tcp.h" #include "ot_udp.h" +#include "ot_worker.h" #include "trackerlogic.h" /* Globals */ @@ -429,8 +430,7 @@ int main(int argc, char **argv) { listen_start_udp_workers(); server_mainloop(0); - tcp_workers_deinit(); - udp_workers_deinit(); + worker_stop_all(); /* No thread is serving any more, so the teardown can run where a debugger and the sanitizers can follow it */ diff --git a/ot_tcp.c b/ot_tcp.c index 47697ca..4a5008b 100644 --- a/ot_tcp.c +++ b/ot_tcp.c @@ -20,6 +20,7 @@ #include "ot_http.h" #include "ot_mutex.h" #include "ot_socket.h" +#include "ot_worker.h" #include "ot_stats.h" #include "ot_tcp.h" #include "trackerlogic.h" @@ -164,75 +165,18 @@ void tcp_accept(const int64_t serversocket, struct ot_workstruct *ws) { } } -/* 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 tcp_worker { - pthread_t thread; - int64_t sock; - _Atomic int left; -}; - -static struct tcp_worker **g_workers; -static size_t g_worker_count; - /* A worker blocks in accept and nowhere else: the connection it is handed is non blocking, so a client that stops talking is given to the event loop rather than keeping this thread. */ -static void *tcp_worker(void *args) { - struct tcp_worker *me = args; - struct ot_workstruct ws; - - if (workstruct_init(&ws) == -1) { - fprintf(stderr, "Warning: a tcp worker could not get its buffers and will not run.\n"); - atomic_store(&me->left, 1); - return NULL; - } - +static void tcp_run(struct ot_worker *me) { while (g_opentracker_running) - tcp_accept(me->sock, &ws); - - workstruct_deinit(&ws); - atomic_store(&me->left, 1); - return NULL; + tcp_accept(me->sock, &me->ws); } void tcp_worker_install(int64_t sock) { - struct tcp_worker **grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers)); - struct tcp_worker *worker; - - if (!grown || !(worker = calloc(1, sizeof(*worker)))) { - fprintf(stderr, "Warning: out of memory installing a tcp worker, this socket stays with the workers already running.\n"); - g_workers = grown ? grown : g_workers; - return; - } - g_workers = grown; - worker->sock = sock; - socket_set_blocking((int)sock, 1); socket_defer_accept((int)sock); - if (pthread_create(&worker->thread, NULL, tcp_worker, worker)) - free(worker); - else - g_workers[g_worker_count++] = worker; -} - -void tcp_workers_deinit(void) { - size_t i; - - for (i = 0; i < g_worker_count; ++i) { - struct tcp_worker *worker = g_workers[i]; - - /* Repeated because a signal only interrupts a call already under way */ - 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; - g_worker_count = 0; + if (worker_start(tcp_run, sock) == -1) + fprintf(stderr, "Warning: could not install a tcp worker, this socket stays with the workers already running.\n"); } diff --git a/ot_tcp.h b/ot_tcp.h index 96b6ec1..55a172c 100644 --- a/ot_tcp.h +++ b/ot_tcp.h @@ -22,7 +22,5 @@ void tcp_write(const int64_t sock); /* One thread accepting on one socket. Called once per worker. */ void tcp_worker_install(int64_t sock); -/* Stop every worker installed above and wait for it to be gone */ -void tcp_workers_deinit(void); #endif diff --git a/ot_udp.c b/ot_udp.c index 00ec566..309cbdb 100644 --- a/ot_udp.c +++ b/ot_udp.c @@ -16,21 +16,11 @@ #include "ot_random.h" #include "ot_rijndael.h" #include "ot_socket.h" +#include "ot_worker.h" #include "ot_stats.h" #include "ot_udp.h" #include "trackerlogic.h" -/* 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; - _Atomic int left; -}; - -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; @@ -217,22 +207,9 @@ int handle_udp6(int64_t serversocket, struct ot_workstruct *ws) { return 1; } -static void *udp_worker(void *args) { - struct udp_worker *me = args; - struct ot_workstruct ws; - - if (workstruct_init(&ws) == -1) { - fprintf(stderr, "Warning: a udp worker could not get its buffers and will not run.\n"); - atomic_store(&me->left, 1); - return NULL; - } - +static void udp_run(struct ot_worker *me) { while (g_opentracker_running) - handle_udp6(me->sock, &ws); - - workstruct_deinit(&ws); - atomic_store(&me->left, 1); - return NULL; + handle_udp6(me->sock, &me->ws); } void udp_init(void) { @@ -241,47 +218,11 @@ 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 *worker; - - 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; - 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); - if (pthread_create(&worker->thread, NULL, udp_worker, worker)) - free(worker); - else - g_workers[g_worker_count++] = worker; + if (worker_start(udp_run, sock) == -1) + fprintf(stderr, "Warning: could not install a udp worker, this socket stays with the workers already running.\n"); } - -void udp_workers_deinit(void) { - size_t i; - - 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; - g_worker_count = 0; -} - diff --git a/ot_udp.h b/ot_udp.h index b731aca..1cd363a 100644 --- a/ot_udp.h +++ b/ot_udp.h @@ -14,8 +14,6 @@ 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/ot_worker.c b/ot_worker.c new file mode 100644 index 0000000..97eaeb2 --- /dev/null +++ b/ot_worker.c @@ -0,0 +1,75 @@ +/* This software was written by Dirk Engling + It is considered beerware. Prost. Skol. Cheers or whatever. + + $id$ */ + +/* System */ +#include +#include +#include +#include +#include + +/* Opentracker */ +#include "ot_worker.h" +#include "trackerlogic.h" + +static struct ot_worker **g_workers; +static size_t g_worker_count; + +/* Everything a worker needs doing around its own loop, in one place, so that + a new kind of worker cannot miss a piece of it. */ +static void *worker_trampoline(void *args) { + struct ot_worker *me = args; + + if (workstruct_init(&me->ws) == -1) + fprintf(stderr, "Warning: a worker could not get its buffers and will not run.\n"); + else { + me->run(me); + workstruct_deinit(&me->ws); + } + + atomic_store(&me->left, 1); + return NULL; +} + +int worker_start(ot_worker_run run, int64_t sock) { + struct ot_worker **grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers)); + struct ot_worker *worker; + + if (!grown) + return -1; + g_workers = grown; + + if (!(worker = calloc(1, sizeof(*worker)))) + return -1; + worker->sock = sock; + worker->run = run; + + if (pthread_create(&worker->thread, NULL, worker_trampoline, worker)) { + free(worker); + return -1; + } + g_workers[g_worker_count++] = worker; + return 0; +} + +void worker_stop_all(void) { + size_t i; + + for (i = 0; i < g_worker_count; ++i) { + struct ot_worker *worker = g_workers[i]; + + /* Repeated because a signal only interrupts a call already under way */ + 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; + g_worker_count = 0; +} diff --git a/ot_worker.h b/ot_worker.h new file mode 100644 index 0000000..ea4f379 --- /dev/null +++ b/ot_worker.h @@ -0,0 +1,39 @@ +/* This software was written by Dirk Engling + It is considered beerware. Prost. Skol. Cheers or whatever. + + $id$ */ + +#ifndef OT_WORKER_H__ +#define OT_WORKER_H__ + +#include +#include +#include + +#include "trackerlogic.h" + +struct ot_worker; + +/* What a worker does between being started and being told to stop. It reads + worker->sock and works out of worker->ws; everything around it, from the + buffers to the counters to saying goodbye, is handled here. */ +typedef void (*ot_worker_run)(struct ot_worker *worker); + +/* A thread that serves requests on a socket of its own, with the buffers and + counters it needs. Each is allocated on its own, so taking on another one + never moves the record a running worker holds. */ +struct ot_worker { + pthread_t thread; + _Atomic int left; + int64_t sock; + ot_worker_run run; + struct ot_workstruct ws; +}; + +/* Start one. Returns -1 if it could not be started. */ +int worker_start(ot_worker_run run, int64_t sock); + +/* Wake every worker until it has left, then join it */ +void worker_stop_all(void); + +#endif -- cgit v1.3