Generalize the worker setup/runloop/teardown logic by adding a smol ot_worker helper

8 files changed, 128 insertions, 133 deletions

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 <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+/* System */
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+/* 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 <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+#ifndef OT_WORKER_H__
+#define OT_WORKER_H__
+
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdint.h>
+
+#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