4 files changed, 157 insertions, 12 deletions

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_fdstate.c b/tests/t_fdstate.c
index bf8724a..421f076 100644
--- a/tests/t_fdstate.c
+++ b/tests/t_fdstate.c
@@ -101,6 +101,8 @@ static void case_out_of_range(void) {
 #define RECYCLE_AFTER  32
 #define STRESS_CLAIMS  200000
 #define STRESS_GIVEUP  (STRESS_CLAIMS * 100)
+/* What the run has to reach for its result to mean anything */
+#define STRESS_CYCLES  (STRESS_CLAIMS / (RECYCLE_AFTER * 4))
 
 static _Atomic long     stress_claims, stress_recycles, stress_violations, stress_attempts;
 static _Atomic int      stress_lives[STRESS_SLOTS];
@@ -116,8 +118,14 @@ static _Atomic uint64_t stress_next_cookie;
 #define STRESS_CONN_COOKIE(w)    ((void *)(uintptr_t)((w) >> 32))
 #define STRESS_CONN_GEN(w)       ((uint32_t)(w))
 
+/* The workers reach their claim target on their own, but the recycles that
+   make the run worth anything are the keeper's, and on a loaded machine the
+   keeper is the thread that loses. So the run goes on until both targets are
+   met rather than until the faster of the two is. */
 static int stress_running(void) {
-  return atomic_load(&stress_claims) < STRESS_CLAIMS && atomic_load(&stress_attempts) < STRESS_GIVEUP;
+  if (atomic_load(&stress_attempts) >= STRESS_GIVEUP)
+    return 0;
+  return atomic_load(&stress_claims) < STRESS_CLAIMS || atomic_load(&stress_recycles) <= STRESS_CYCLES;
 }
 
 static void *stress_worker(void *arg) {
@@ -203,7 +211,7 @@ static void case_recycling_under_load(void) {
 
   test_case("and the run actually recycled descriptors, or it proved nothing");
   CHECK(atomic_load(&stress_claims) >= STRESS_CLAIMS);
-  CHECK(atomic_load(&stress_recycles) > STRESS_CLAIMS / (RECYCLE_AFTER * 4));
+  CHECK(atomic_load(&stress_recycles) > STRESS_CYCLES);
 
   for (i = 0; i < STRESS_SLOTS; ++i)
     fdstate_close((int)i);
diff --git a/tests/t_socket.c b/tests/t_socket.c
index 8eef7fd..ab4b785 100644
--- a/tests/t_socket.c
+++ b/tests/t_socket.c
@@ -8,13 +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 <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
@@ -144,6 +149,57 @@ static void case_udp_roundtrip(void) {
   close(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;
+
+static void        reader_wake(int s) { (void)s; }
+
+static void       *blocked_reader(void *args) {
+  char     buf[32], ip[16];
+  uint16_t port = 0;
+
+  (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
    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 +349,7 @@ void test_socket(void) {
   case_sharing_is_opt_in();
 #endif
   case_udp_roundtrip();
+  case_a_signal_ends_a_blocking_read();
   case_udp_v4mapped();
   case_tcp_roundtrip();
   case_connect_refused();
diff --git a/tests/udp_wake_probe.c b/tests/udp_wake_probe.c
index 8a0f7a8..e948c34 100644
--- a/tests/udp_wake_probe.c
+++ b/tests/udp_wake_probe.c
@@ -1,13 +1,17 @@
-/* Three ways to get a thread out of a blocking recvfrom, tested rather than
-   assumed: close the socket, shut it down, or give it a receive timeout.
+/* Four ways to get a thread out of a blocking recvfrom, tested rather than
+   assumed: close the socket, shut it down, give it a receive timeout, or
+   send it a signal.
 
-   This decides how udp workers can be stopped. They sit in recvfrom and are
-   never joined today, and the sockets of a SO_REUSEPORT group all share one
-   address, so a worker cannot be singled out by sending it a datagram.
+   This decides how udp workers are stopped. The sockets of a SO_REUSEPORT
+   group all share one address, so a worker cannot be singled out by sending
+   it a datagram, and the two mechanisms differ in what they cost while the
+   tracker is running: a receive timeout wakes every worker on every tick
+   forever, a signal and a shutdown cost nothing until the one moment they
+   are used.
 
      cc -O2 -pthread -o /tmp/wake tests/udp_wake_probe.c && /tmp/wake
 
-   On Linux 6.1: close does not wake it, shutdown does, SO_RCVTIMEO does. */
+   On Linux 6.1: close does not wake it, the other three do. */
 #include <errno.h>
 #include <netinet/in.h>
 #include <pthread.h>
@@ -15,9 +19,12 @@
 #include <stdio.h>
 #include <string.h>
 #include <sys/socket.h>
+#include <signal.h>
 #include <sys/time.h>
 #include <unistd.h>
 
+#define WAKE_SIGNAL SIGUSR1
+
 static _Atomic int returned;
 static _Atomic int saved_errno;
 static int         sock;
@@ -31,6 +38,11 @@ static void       *reader(void *args) {
   return NULL;
 }
 
+/* A handler that does nothing is all it takes: the point is the return from
+   recvfrom with EINTR, not anything the handler does. SA_RESTART must stay
+   off or the kernel resumes the call instead of reporting it. */
+static void noop(int s) { (void)s; }
+
 static int bind_udp(int timeout_ms) {
   struct sockaddr_in address;
   int                s = socket(AF_INET, SOCK_DGRAM, 0);
@@ -50,7 +62,7 @@ static int bind_udp(int timeout_ms) {
   return s;
 }
 
-static void try_it(const char *name, int timeout_ms, int do_close, int do_shutdown) {
+static void try_it(const char *name, int timeout_ms, int do_close, int do_shutdown, int do_signal) {
   pthread_t thread;
   int       waited;
 
@@ -62,6 +74,8 @@ static void try_it(const char *name, int timeout_ms, int do_close, int do_shutdo
 
   if (do_shutdown)
     shutdown(sock, SHUT_RD);
+  if (do_signal)
+    pthread_kill(thread, WAKE_SIGNAL);
   if (do_close)
     close(sock);
 
@@ -92,8 +106,16 @@ static void try_it(const char *name, int timeout_ms, int do_close, int do_shutdo
 }
 
 int main(void) {
-  try_it("close() the socket", 0, 1, 0);
-  try_it("shutdown(SHUT_RD)", 0, 0, 1);
-  try_it("SO_RCVTIMEO 500 ms", 500, 0, 0);
+  struct sigaction sa;
+
+  memset(&sa, 0, sizeof(sa));
+  sa.sa_handler = noop;
+  sa.sa_flags   = 0;
+  sigaction(WAKE_SIGNAL, &sa, NULL);
+
+  try_it("close() the socket", 0, 1, 0, 0);
+  try_it("shutdown(SHUT_RD)", 0, 0, 1, 0);
+  try_it("SO_RCVTIMEO 500 ms", 500, 0, 0, 0);
+  try_it("pthread_kill(SIGUSR1)", 0, 0, 0, 1);
   return 0;
 }