1 files changed, 5 insertions, 73 deletions

diff --git a/ot_udp.c b/ot_udp.c
index 1ec5148..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,31 +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;
-  memset(&ws, 0, sizeof(ws));
-
-  stats_counters_register(&ws);
-  ws.inbuf  = malloc(G_INBUF_SIZE);
-  ws.outbuf = malloc(G_OUTBUF_SIZE);
-#ifdef _DEBUG_HTTPERROR
-  ws.debugbuf = malloc(G_DEBUGBUF_SIZE);
-#endif
-
+static void udp_run(struct ot_worker *me) {
   while (g_opentracker_running)
-    handle_udp6(me->sock, &ws);
-
-  stats_counters_retire(&ws);
-
-  free(ws.inbuf);
-  free(ws.outbuf);
-#ifdef _DEBUG_HTTPERROR
-  free(ws.debugbuf);
-#endif
-
-  atomic_store(&me->left, 1);
-  return NULL;
+    handle_udp6(me->sock, &me->ws);
 }
 
 void udp_init(void) {
@@ -250,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;
-}
-
-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;
+  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");
 }
-