1 files changed, 35 insertions, 23 deletions

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 <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <sys/socket.h>
+#include <unistd.h>
 
 /* 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;