With more workers blockingly waiting on udp sockets, we need to keep track of all sockets to shut them down on exit

7 files changed, 156 insertions, 7 deletions

diff --git a/opentracker.c b/opentracker.c
index f579b92..ab4a934 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -397,7 +397,6 @@ static void listen_start_udp_workers(void) {
     if (entry->proto != FLAG_UDP || entry->workers <= 0)
       continue;
 
-    socket_set_blocking((int)entry->sock, 1);
     event_del((int)entry->sock);
     udp_worker_install(entry->sock);
 
@@ -409,8 +408,7 @@ static void listen_start_udp_workers(void) {
         if (sock == -1) {
           fprintf(stderr, "Warning: could not bind another socket for udp worker %d: %s. Sharing the first instead.\n", worker, strerror(errno));
           sock = entry->sock;
-        } else
-          socket_set_blocking((int)sock, 1);
+        }
       }
       udp_worker_install(sock);
     }
@@ -548,9 +546,10 @@ int main(int argc, char **argv) {
   listen_start_udp_workers();
 
   server_mainloop(0);
+  udp_workers_deinit();
 
-  /* The loop has left, so no thread is serving any more and the teardown
-     can run where a debugger and the sanitizers can follow it */
+  /* No thread is serving any more, so the teardown can run where a debugger
+     and the sanitizers can follow it */
   trackerlogic_deinit();
 #ifdef WANT_SYSLOGS
   closelog();
diff --git a/ot_socket.c b/ot_socket.c
index 158459f..c73738c 100644
--- a/ot_socket.c
+++ b/ot_socket.c
@@ -9,6 +9,7 @@
 #include <netinet/in.h>
 #include <string.h>
 #include <sys/socket.h>
+#include <sys/time.h>
 #include <unistd.h>
 
 /* Opentracker */
@@ -48,6 +49,12 @@ 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 f73a5c9..9340d0d 100644
--- a/ot_socket.h
+++ b/ot_socket.h
@@ -39,6 +39,10 @@ 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 87741da..7aa0e61 100644
--- a/ot_udp.c
+++ b/ot_udp.c
@@ -9,6 +9,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/socket.h>
 
 /* Opentracker */
 #include "ot_ip.h"
@@ -19,6 +20,17 @@
 #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
+
+struct udp_worker {
+  pthread_t thread;
+  int64_t   sock;
+};
+
+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;
@@ -236,11 +248,37 @@ void udp_init(void) {
 }
 
 void udp_worker_install(int64_t sock) {
-  pthread_t thread_id;
+  struct udp_worker *grown = realloc(g_workers, (g_worker_count + 1) * sizeof(*g_workers));
+
+  if (!grown) {
+    fprintf(stderr, "Warning: out of memory installing a udp worker, this socket stays with the workers already running.\n");
+    return;
+  }
+  g_workers = grown;
 
 #ifdef _DEBUG
   fprintf(stderr, "Installing a worker on udp socket %ld\n", (unsigned long)sock);
 #endif
-  pthread_create(&thread_id, NULL, udp_worker, (void *)sock);
+  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;
+}
+
+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);
+
+  free(g_workers);
+  g_workers      = NULL;
+  g_worker_count = 0;
 }
 
diff --git a/ot_udp.h b/ot_udp.h
index 7d4432f..b731aca 100644
--- a/ot_udp.h
+++ b/ot_udp.h
@@ -13,6 +13,9 @@
 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/tests/integration/test_shutdown.py b/tests/integration/test_shutdown.py
new file mode 100644
index 0000000..063d779
--- /dev/null
+++ b/tests/integration/test_shutdown.py
@@ -0,0 +1,58 @@
+"""That every thread the tracker starts also ends when it is asked to stop.
+
+The rest of the suite stops a tracker too, so a thread left behind already
+shows up in teardown as the complaint that the tracker had to be killed.
+These ask for the arrangements that park a thread in a blocking read in the
+first place, and they time the exit, because a reader woken only by its own
+receive timeout passes a correctness check while taking a second over it.
+"""
+
+import time
+import unittest
+
+from harness import Tracker, TrackerTestCase
+
+# Generous enough for a platform where shutting the socket down does not wake
+# the reader and the receive timeout has to expire, tight enough that a reader
+# which is never woken at all fails instead of hanging the suite
+PATIENCE = 3.0
+
+
+def time_the_exit(extra_config):
+    tracker = Tracker(extra_config)
+    try:
+        # Serve one request of each kind first, so the threads that are about
+        # to be joined have actually run
+        tracker.announce(b"shutdown-probe\0\0\0\0\0\0")
+        sock, connection_id = tracker.udp_connect()
+        tracker.udp_announce(sock, connection_id, b"shutdown-probe\0\0\0\0\0\0")
+        sock.close()
+        started = time.time()
+        complaints = tracker.stop()
+        return time.time() - started, complaints
+    finally:
+        tracker._cleanup()
+
+
+class UdpWorkers(TrackerTestCase):
+    config = "listen.udp.workers 8\n"
+
+    def test_announce_reaches_one_of_eight_workers(self):
+        """Keeps the workers from being merely started and never used."""
+        sock, connection_id = self.tracker.udp_connect()
+        answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("eight-workers"))
+        sock.close()
+        self.assertEqual(answer["seeders"], 1)
+
+
+class ShutdownEndsEveryThread(unittest.TestCase):
+    def test_eight_udp_workers_are_joined(self):
+        took, complaints = time_the_exit("listen.udp.workers 8\n")
+        self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints))
+        self.assertLess(took, PATIENCE, "stopping eight udp workers took %.1f s" % took)
+
+    def test_udp_in_the_event_loop_still_stops(self):
+        """Worker count zero leaves no thread to join, which must also work."""
+        took, complaints = time_the_exit("listen.udp.workers 0\n")
+        self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints))
+        self.assertLess(took, PATIENCE, "stopping without workers took %.1f s" % took)
diff --git a/tests/t_socket.c b/tests/t_socket.c
index 8eef7fd..ede8ae0 100644
--- a/tests/t_socket.c
+++ b/tests/t_socket.c
@@ -10,6 +10,7 @@
 #include <poll.h>
 #include <string.h>
 #include <sys/socket.h>
+#include <time.h>
 #include <unistd.h>
 
 /* Opentracker */
@@ -144,6 +145,44 @@ 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;
+
+  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);
+
+  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);
+
+  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);
+}
+
 /* An IPv4 peer must arrive in the v4 mapped form opentracker stores, however
    the kernel chose to report it. Linux fills a v4 mapped sockaddr_in6 here
    and the BSDs fill a plain sockaddr_in, so this is the case that tells the
@@ -293,6 +332,7 @@ void test_socket(void) {
   case_sharing_is_opt_in();
 #endif
   case_udp_roundtrip();
+  case_blocking_read_comes_back();
   case_udp_v4mapped();
   case_tcp_roundtrip();
   case_connect_refused();