7 files changed, 103 insertions, 76 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 <netinet/in.h>
 #include <string.h>
 #include <sys/socket.h>
-#include <sys/time.h>
 #include <unistd.h>
 
 /* 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 <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;
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 <fcntl.h>
 #include <netinet/in.h>
 #include <poll.h>
+#include <pthread.h>
+#include <signal.h>
+#include <stdatomic.h>
 #include <string.h>
 #include <sys/socket.h>
-#include <time.h>
+#include <sys/time.h>
 #include <unistd.h>
 
 /* 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;
+/* 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;
 
-  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);
+static void        reader_wake(int s) { (void)s; }
 
-  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);
+static void       *blocked_reader(void *args) {
+  char     buf[32], ip[16];
+  uint16_t port = 0;
 
-  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);
+  (void)args;
+  socket_recv(reader_socket, buf, sizeof(buf), ip, &port);
+  atomic_store(&reader_returned, 1);
+  return NULL;
+}
+
+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 <signal.h>
 #include <stdatomic.h>
 #include <stdint.h>
 #include <stdlib.h>
@@ -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;