Add load tests to sanitize check the trackerlogic, based on udp requests

5 files changed, 815 insertions, 0 deletions

diff --git a/tests/udp_load.c b/tests/udp_load.c
new file mode 100644
index 0000000..d86c157
--- /dev/null
+++ b/tests/udp_load.c
@@ -0,0 +1,202 @@
+/* A UDP load generator speaking the tracker's own protocol.
+
+   Each thread keeps a socket of its own, does the BEP 15 connect handshake
+   once and then announces in a loop, counting replies. Separate sockets
+   matter: the kernel spreads datagrams across a SO_REUSEPORT group by a hash
+   of the four tuple, so a generator sending everything from one source port
+   would pile every packet onto one of the tracker's workers and measure the
+   wrong thing.
+
+     cc -O2 -pthread -o udp_load tests/udp_load.c
+     ./udp_load 127.0.0.1 6969 4 5 10000 64
+
+   The last argument is how many announces each thread keeps in flight. With
+   one, a thread sends and waits, so the offered load is threads divided by
+   the round trip and the figure says more about scheduling than about the
+   tracker: measured here, a tracker answering 60k/s that way was using
+   three quarters of one core. Raising the window offers load without
+   needing a thread per outstanding request, which is what lets the tracker
+   be pushed to the point where it is actually the limit.
+
+   The last argument is how many distinct torrents to announce. It is what
+   keeps the figure honest: a fresh info hash every time measures how fast
+   the tracker can insert a million torrents into sorted vectors, which is
+   not what a tracker spends its life doing, and which punishes whichever
+   configuration is quickest because it reaches the expensive end of the run
+   soonest. Cycling a fixed set reaches a steady state where almost every
+   announce renews a peer that is already there. */
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <time.h>
+#include <unistd.h>
+
+#define CONNECT_MAGIC 0x41727101980ULL
+
+static struct sockaddr_in g_tracker;
+static long               g_torrents = 10000;
+static long               g_window   = 64;
+static _Atomic long long  g_sent     = 0;
+static _Atomic int        g_running  = 1;
+static _Atomic long long  g_replies  = 0;
+static _Atomic long long  g_timeouts = 0;
+
+static double             seconds(void) {
+  struct timespec now;
+  clock_gettime(CLOCK_MONOTONIC, &now);
+  return now.tv_sec + now.tv_nsec / 1e9;
+}
+
+static void put64(unsigned char *at, uint64_t value) {
+  int i;
+  for (i = 0; i < 8; ++i)
+    at[i] = (unsigned char)(value >> (56 - 8 * i));
+}
+static void put32(unsigned char *at, uint32_t value) {
+  at[0] = (unsigned char)(value >> 24);
+  at[1] = (unsigned char)(value >> 16);
+  at[2] = (unsigned char)(value >> 8);
+  at[3] = (unsigned char)value;
+}
+static uint64_t get64(const unsigned char *at) {
+  uint64_t value = 0;
+  int      i;
+  for (i = 0; i < 8; ++i)
+    value = (value << 8) | at[i];
+  return value;
+}
+
+static void *loader(void *args) {
+  long          index = (long)args;
+  unsigned char request[98], reply[1500];
+  uint64_t      connection_id = 0;
+  int           sock          = socket(AF_INET, SOCK_DGRAM, 0);
+  struct timeval patience     = {0, 50000}; /* 50 ms */
+  unsigned      seed          = (unsigned)index * 2654435761u + 1;
+
+  if (sock < 0)
+    return NULL;
+  setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &patience, sizeof(patience));
+
+  /* connect handshake */
+  memset(request, 0, sizeof(request));
+  put64(request, CONNECT_MAGIC);
+  put32(request + 8, 0);
+  put32(request + 12, (uint32_t)index);
+  if (sendto(sock, request, 16, 0, (struct sockaddr *)&g_tracker, sizeof(g_tracker)) != 16) {
+    close(sock);
+    return NULL;
+  }
+  if (recvfrom(sock, reply, sizeof(reply), 0, NULL, NULL) < 16) {
+    fprintf(stderr, "  thread %ld: no answer to connect\n", index);
+    close(sock);
+    return NULL;
+  }
+  connection_id = get64(reply + 8);
+
+  /* Keep g_window announces outstanding: send that many, then drain what has
+     come back, rather than waiting for each reply in turn */
+  while (atomic_load_explicit(&g_running, memory_order_relaxed)) {
+    ssize_t got;
+    int     i;
+    long    outstanding;
+
+    for (outstanding = 0; outstanding < g_window; ++outstanding) {
+    memset(request, 0, sizeof(request));
+    put64(request, connection_id);
+    put32(request + 8, 1); /* announce */
+    put32(request + 12, (uint32_t)index);
+    /* One of a fixed set, so that after the first pass almost every announce
+       renews a peer rather than creating a torrent. The index is hashed into
+       all twenty bytes so the buckets, which key off the first four, are
+       spread rather than walked in order. */
+    {
+      unsigned which = (unsigned)((seed = seed * 1103515245u + 12345u) >> 8) % (unsigned)g_torrents;
+      unsigned spread;
+      for (i = 0; i < 20; ++i) {
+        spread             = which * 2654435761u + (unsigned)i * 40503u;
+        spread            ^= spread >> 13;
+        request[16 + i]    = (unsigned char)(spread >> 11);
+      }
+    }
+    for (i = 0; i < 20; ++i)
+      request[36 + i] = (unsigned char)('a' + (i % 26));
+    put64(request + 56, 0);                             /* downloaded */
+    put64(request + 64, 0);                             /* left: a seeder */
+    put64(request + 72, 0);                             /* uploaded */
+    put32(request + 80, 0);                             /* event */
+    put32(request + 84, 0);                             /* ip */
+    put32(request + 88, (uint32_t)index);               /* key */
+    put32(request + 92, (uint32_t)-1);                  /* num_want */
+    /* A port of this thread's own, so each thread is a distinct peer and the
+       swarms hold more than one */
+    request[96] = (unsigned char)((6881 + index) >> 8);
+    request[97] = (unsigned char)((6881 + index) & 0xff);
+
+    if (sendto(sock, request, 98, 0, (struct sockaddr *)&g_tracker, sizeof(g_tracker)) != 98)
+      break;
+    atomic_fetch_add_explicit(&g_sent, 1, memory_order_relaxed);
+    }
+
+    for (outstanding = 0; outstanding < g_window; ++outstanding) {
+      got = recvfrom(sock, reply, sizeof(reply), 0, NULL, NULL);
+      if (got >= 20)
+        atomic_fetch_add_explicit(&g_replies, 1, memory_order_relaxed);
+      else {
+        atomic_fetch_add_explicit(&g_timeouts, 1, memory_order_relaxed);
+        break; /* nothing more is coming; go round and send again */
+      }
+    }
+  }
+
+  close(sock);
+  return NULL;
+}
+
+int main(int argc, char **argv) {
+  const char *host    = argc > 1 ? argv[1] : "127.0.0.1";
+  int         port    = argc > 2 ? atoi(argv[2]) : 6969;
+  long        threads = argc > 3 ? atol(argv[3]) : 16;
+  double      run_for = argc > 4 ? atof(argv[4]) : 5.0;
+  long        torrents = argc > 5 ? atol(argv[5]) : 10000;
+  long        window   = argc > 6 ? atol(argv[6]) : 64;
+  pthread_t  *ids     = calloc((size_t)threads, sizeof(pthread_t));
+  long        i;
+  double      started, elapsed;
+
+  g_torrents = torrents > 0 ? torrents : 1;
+  g_window   = window > 0 ? window : 1;
+
+  memset(&g_tracker, 0, sizeof(g_tracker));
+  g_tracker.sin_family = AF_INET;
+  g_tracker.sin_port   = htons((uint16_t)port);
+  if (inet_pton(AF_INET, host, &g_tracker.sin_addr) != 1) {
+    fprintf(stderr, "cannot parse address %s\n", host);
+    return 1;
+  }
+
+  for (i = 0; i < threads; ++i)
+    pthread_create(ids + i, NULL, loader, (void *)i);
+
+  started = seconds();
+  usleep((useconds_t)(run_for * 1e6));
+  atomic_store(&g_running, 0);
+  elapsed = seconds() - started;
+
+  for (i = 0; i < threads; ++i)
+    pthread_join(ids[i], NULL);
+
+  printf("%.0f replies/s  (%.0f sent/s, %lld replies, %lld lost, %ld threads x %ld in flight, %.1f s)\n",
+         (double)atomic_load(&g_replies) / elapsed, (double)atomic_load(&g_sent) / elapsed,
+         (long long)atomic_load(&g_replies), (long long)atomic_load(&g_timeouts), threads, g_window, elapsed);
+  free(ids);
+  return 0;
+}
diff --git a/tests/udp_modes.c b/tests/udp_modes.c
new file mode 100644
index 0000000..654cfd8
--- /dev/null
+++ b/tests/udp_modes.c
@@ -0,0 +1,240 @@
+/* A receiver with the udp arrangements we are choosing between, so they can
+   be compared without any of them being built into the tracker first.
+
+   It echoes what it is sent, so tests/udp_load.c drives it unchanged. There
+   is no tracker work here on purpose: with nothing else happening, whatever
+   the kernel's receive path costs is the whole figure, which is the clearest
+   view of it. Pass a work time to dilute that back towards realism.
+
+     cc -O2 -pthread -o udp_modes tests/udp_modes.c
+     ./udp_modes 6969 4 shared 1
+     ./udp_modes 6969 4 shared 32
+     ./udp_modes 6969 4 per-thread 1
+     ./udp_modes 6969 4 per-thread 32
+
+   Arguments: port, threads, shared|per-thread, datagrams per syscall,
+   and optionally nanoseconds of pretend work per datagram. */
+
+#define _GNU_SOURCE /* struct mmsghdr on glibc */
+
+#include <errno.h>
+#include <netinet/in.h>
+#include <pthread.h>
+#include <signal.h>
+#include <stdatomic.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <time.h>
+#include <unistd.h>
+
+#if defined(__linux__) || defined(__FreeBSD__)
+#define HAVE_MMSG 1
+#endif
+
+#define MAX_BATCH   256
+#define BUFFER_SIZE 2048
+
+static int         g_port;
+static long        g_threads;
+static long        g_batch;
+static long        g_work_ns;
+static int         g_shared_socket = -1; /* -1 means one socket per thread */
+static _Atomic int g_running       = 1;
+
+static void        stop(int sig) {
+  (void)sig;
+  atomic_store(&g_running, 0);
+}
+
+/* Which option the platform offers for spreading datagrams over a group of
+   sockets bound to one address. They are not the same thing: FreeBSD's plain
+   SO_REUSEPORT permits the bind without spreading anything, and Darwin's
+   hands every datagram to whichever socket bound last. */
+static const char *share_the_port(int sock) {
+  int one = 1;
+
+  setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
+#ifdef SO_REUSEPORT_LB
+  if (!setsockopt(sock, SOL_SOCKET, SO_REUSEPORT_LB, &one, sizeof(one)))
+    return "SO_REUSEPORT_LB";
+#endif
+#ifdef SO_REUSEPORT
+  if (!setsockopt(sock, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(one)))
+    return "SO_REUSEPORT";
+#endif
+  return "neither";
+}
+
+static int bind_one(int share, const char **option) {
+  struct sockaddr_in address;
+  int                sock = socket(AF_INET, SOCK_DGRAM, 0);
+
+  if (sock < 0)
+    return -1;
+  *option = share ? share_the_port(sock) : "exclusive";
+
+  memset(&address, 0, sizeof(address));
+  address.sin_family      = AF_INET;
+  address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+  address.sin_port        = htons((uint16_t)g_port);
+  if (bind(sock, (struct sockaddr *)&address, sizeof(address)) < 0) {
+    close(sock);
+    return -1;
+  }
+  return sock;
+}
+
+static void pretend_to_work(void) {
+  struct timespec started, now;
+  long            waited = 0;
+
+  if (!g_work_ns)
+    return;
+  clock_gettime(CLOCK_MONOTONIC, &started);
+  while (waited < g_work_ns) {
+    clock_gettime(CLOCK_MONOTONIC, &now);
+    waited = (now.tv_sec - started.tv_sec) * 1000000000L + (now.tv_nsec - started.tv_nsec);
+  }
+}
+
+static void one_at_a_time(int sock) {
+  char               buffer[BUFFER_SIZE];
+  struct sockaddr_in from;
+  socklen_t          length;
+  ssize_t            got;
+
+  while (atomic_load_explicit(&g_running, memory_order_relaxed)) {
+    length = sizeof(from);
+    got    = recvfrom(sock, buffer, sizeof(buffer), 0, (struct sockaddr *)&from, &length);
+    if (got < 0) {
+      if (errno == EINTR || errno == EAGAIN)
+        continue;
+      return;
+    }
+    pretend_to_work();
+    sendto(sock, buffer, (size_t)got, 0, (struct sockaddr *)&from, length);
+  }
+}
+
+#ifdef HAVE_MMSG
+static void in_batches(int sock) {
+  static _Thread_local char       buffers[MAX_BATCH][BUFFER_SIZE];
+  static _Thread_local struct     sockaddr_in froms[MAX_BATCH];
+  static _Thread_local struct     iovec iovecs[MAX_BATCH];
+  static _Thread_local struct     mmsghdr messages[MAX_BATCH];
+  int                             got, i;
+
+  while (atomic_load_explicit(&g_running, memory_order_relaxed)) {
+    for (i = 0; i < g_batch; ++i) {
+      iovecs[i].iov_base             = buffers[i];
+      iovecs[i].iov_len              = BUFFER_SIZE;
+      memset(messages + i, 0, sizeof(messages[i]));
+      messages[i].msg_hdr.msg_name    = froms + i;
+      messages[i].msg_hdr.msg_namelen = sizeof(froms[i]);
+      messages[i].msg_hdr.msg_iov     = iovecs + i;
+      messages[i].msg_hdr.msg_iovlen  = 1;
+    }
+
+    /* WAITFORONE so a quiet receiver behaves like the unbatched one rather
+       than holding the first datagram until the batch fills */
+    got = recvmmsg(sock, messages, (unsigned)g_batch, MSG_WAITFORONE, NULL);
+    if (got < 0) {
+      if (errno == EINTR || errno == EAGAIN)
+        continue;
+      return;
+    }
+
+    for (i = 0; i < got; ++i) {
+      pretend_to_work();
+      iovecs[i].iov_len              = messages[i].msg_len;
+      messages[i].msg_hdr.msg_namelen = sizeof(froms[i]);
+    }
+    sendmmsg(sock, messages, (unsigned)got, 0);
+  }
+}
+#endif
+
+static void *worker(void *args) {
+  const char *option = "";
+  int         sock   = g_shared_socket;
+
+  (void)args;
+  if (sock < 0) {
+    sock = bind_one(1, &option);
+    if (sock < 0) {
+      fprintf(stderr, "  a worker could not bind: %s\n", strerror(errno));
+      return NULL;
+    }
+  }
+
+#ifdef HAVE_MMSG
+  if (g_batch > 1)
+    in_batches(sock);
+  else
+#endif
+    one_at_a_time(sock);
+
+  if (g_shared_socket < 0)
+    close(sock);
+  return NULL;
+}
+
+int main(int argc, char **argv) {
+  const char *how     = argc > 3 ? argv[3] : "shared";
+  const char *option  = "";
+  pthread_t  *threads;
+  long        i;
+
+  g_port    = argc > 1 ? atoi(argv[1]) : 6969;
+  g_threads = argc > 2 ? atol(argv[2]) : 1;
+  g_batch   = argc > 4 ? atol(argv[4]) : 1;
+  g_work_ns = argc > 5 ? atol(argv[5]) : 0;
+  if (g_batch < 1)
+    g_batch = 1;
+  if (g_batch > MAX_BATCH)
+    g_batch = MAX_BATCH;
+
+#ifndef HAVE_MMSG
+  if (g_batch > 1) {
+    fprintf(stderr, "  no recvmmsg on this platform; batching unavailable\n");
+    return 2;
+  }
+#endif
+
+  signal(SIGINT, stop);
+  signal(SIGTERM, stop);
+
+  if (!strcmp(how, "shared")) {
+    g_shared_socket = bind_one(0, &option);
+    if (g_shared_socket < 0) {
+      fprintf(stderr, "  bind failed: %s\n", strerror(errno));
+      return 1;
+    }
+  } else {
+    /* Probe once so the mode line can say what the platform gave us */
+    int probe = bind_one(1, &option);
+    if (probe < 0) {
+      fprintf(stderr, "  bind failed: %s\n", strerror(errno));
+      return 1;
+    }
+    close(probe);
+  }
+
+  printf("  port %d, %ld threads, %s socket%s (%s), %ld per syscall, %ld ns work\n",
+         g_port, g_threads, strcmp(how, "shared") ? "one per thread" : "one shared",
+         strcmp(how, "shared") ? "s" : "", option, g_batch, g_work_ns);
+  fflush(stdout);
+
+  threads = calloc((size_t)g_threads, sizeof(pthread_t));
+  for (i = 0; i < g_threads; ++i)
+    pthread_create(threads + i, NULL, worker, NULL);
+  for (i = 0; i < g_threads; ++i)
+    pthread_join(threads[i], NULL);
+
+  if (g_shared_socket >= 0)
+    close(g_shared_socket);
+  free(threads);
+  return 0;
+}
diff --git a/tests/udp_modes.py b/tests/udp_modes.py
new file mode 100644
index 0000000..19cd264
--- /dev/null
+++ b/tests/udp_modes.py
@@ -0,0 +1,139 @@
+"""The four udp arrangements, side by side.
+
+    cc -O2 -pthread -o /tmp/udp_modes tests/udp_modes.c
+    cc -O2 -pthread -o /tmp/udp_load  tests/udp_load.c
+    python3 tests/udp_modes.py
+
+Two questions at once. Down the table: does a shared socket stop scaling as
+threads are added? Across it: does taking several datagrams per syscall make
+that better, or only cheaper?
+
+The receiver does no tracker work by default, so the figure is the kernel's
+receive path and nothing else. WORK_NS=1500 dilutes it back towards what the
+tracker actually costs per announce.
+
+The number reported is cpu microseconds the receiver spent per datagram
+answered, not throughput. Throughput here says more about the load generator
+than the receiver, because both run on the same machine and an echo costs so
+little that the generator runs out of cores first. Cpu per datagram does not
+care: it is what the receive path costs, whoever is the bottleneck.
+"""
+import os
+import signal
+import socket
+import subprocess
+import sys
+import time
+
+MODES = os.environ.get("MODES", "shared per-thread").split()
+BATCHES = [int(n) for n in os.environ.get("BATCHES", "1 8 32").split()]
+THREAD_COUNTS = [int(n) for n in os.environ.get("THREADS", "1 2 4 8").split()]
+LOAD_THREADS = int(os.environ.get("LOAD_THREADS", "4"))
+WINDOW = int(os.environ.get("WINDOW", "32"))
+SECONDS = float(os.environ.get("SECONDS", "3"))
+REPEATS = int(os.environ.get("REPEATS", "2"))
+WORK_NS = int(os.environ.get("WORK_NS", "0"))
+RECEIVER = os.environ.get("UDP_MODES", "/tmp/udp_modes")
+LOADER = os.environ.get("UDP_LOAD", "/tmp/udp_load")
+PORT_BASE = int(os.environ.get("PORT_BASE", "20100"))
+
+
+def cpu_seconds(pid):
+    try:
+        out = subprocess.run(["ps", "-o", "time=", "-p", str(pid)], capture_output=True, text=True).stdout.strip()
+    except OSError:
+        return None
+    if not out:
+        return None
+    total = 0.0
+    for part in out.replace("-", ":").split(":"):
+        total = total * 60 + float(part)
+    return total
+
+
+def one_run(port, threads, mode, batch):
+    receiver = subprocess.Popen(
+        [RECEIVER, str(port), str(threads), mode, str(batch), str(WORK_NS)],
+        stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True,
+    )
+    for _ in range(200):
+        if receiver.poll() is not None:
+            return None, None
+        probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        probe.settimeout(0.1)
+        try:
+            probe.sendto(b"\x00" * 16, ("127.0.0.1", port))
+            probe.recv(64)
+            probe.close()
+            break
+        except OSError:
+            probe.close()
+            time.sleep(0.02)
+    else:
+        receiver.send_signal(signal.SIGINT)
+        receiver.wait()
+        return None, None
+
+    try:
+        subprocess.run([LOADER, "127.0.0.1", str(port), str(LOAD_THREADS), "1", "10000", str(WINDOW)],
+                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        spent, wall = cpu_seconds(receiver.pid), time.time()
+        out = subprocess.run(
+            [LOADER, "127.0.0.1", str(port), str(LOAD_THREADS), str(SECONDS), "10000", str(WINDOW)],
+            capture_output=True, text=True,
+        ).stdout.strip()
+        used = cpu_seconds(receiver.pid)
+    finally:
+        receiver.send_signal(signal.SIGINT)
+        try:
+            receiver.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            receiver.kill()
+            receiver.wait()
+
+    head = out.split()
+    if not head or not head[0].replace(".", "").isdigit():
+        return None, None
+    cores = (used - spent) / max(time.time() - wall, 1e-9) if (used is not None and spent is not None) else None
+    return float(head[0]), cores
+
+
+def main():
+    for binary in (RECEIVER, LOADER):
+        if not os.path.exists(binary):
+            raise SystemExit("missing %s; see the comment at the top of tests/udp_modes.c" % binary)
+
+    print("  %d load threads x %d in flight, %.0f s x %d runs, %d ns work per datagram"
+          % (LOAD_THREADS, WINDOW, SECONDS, REPEATS, WORK_NS))
+    print("  cpu microseconds per datagram, and the throughput it was measured at")
+    header = "  %-14s %-8s" % ("sockets", "threads")
+    for batch in BATCHES:
+        header += "%20s" % ("%d per syscall" % batch)
+    print(header)
+
+    port = PORT_BASE
+    for mode in MODES:
+        for threads in THREAD_COUNTS:
+            row = "  %-14s %-8d" % (mode, threads)
+            for batch in BATCHES:
+                results = []
+                for _ in range(REPEATS):
+                    port += 1
+                    rate, cores = one_run(port, threads, mode, batch)
+                    if rate:
+                        results.append((rate, cores))
+                if not results:
+                    row += "%18s" % "-"
+                    continue
+                results.sort()
+                rate, cores = results[len(results) // 2]
+                if cores:
+                    row += "%9.2f us %6.0fk/s" % (cores * 1e6 / rate, rate / 1000.0)
+                else:
+                    row += "%20s" % ("%.0fk/s" % (rate / 1000.0))
+            print(row)
+            sys.stdout.flush()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/udp_scaling.py b/tests/udp_scaling.py
new file mode 100644
index 0000000..b5048cd
--- /dev/null
+++ b/tests/udp_scaling.py
@@ -0,0 +1,135 @@
+"""Does listen.udp.workers help or hurt, on this platform, with this tracker?
+
+Starts a tracker once per worker count and drives it with tests/udp_load.c,
+which speaks the real protocol, so what comes out is the tracker's own
+throughput rather than an echo server's.
+
+    cc -O2 -pthread -o /tmp/udp_load tests/udp_load.c
+    make opentracker
+    python3 tests/udp_scaling.py
+
+Worker count 0 means no dedicated workers: udp is served from the main event
+loop, which is what a default build does.
+
+LOAD_THREADS and WINDOW decide how much load is offered. Both matter. Too
+little and the tracker idles while the figure measures round trips; too much
+and the kernel drops datagrams, every lost reply costs the generator a
+timeout, and the figure collapses. The generator prints how many were lost:
+tune until that is near zero and the throughput has stopped rising.
+"""
+import os
+import signal
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+
+TRACKER = os.environ.get("OPENTRACKER", "./opentracker")
+LOADER = os.environ.get("UDP_LOAD", "/tmp/udp_load")
+WORKER_COUNTS = [int(n) for n in os.environ.get("WORKERS", "0 1 2 4 8").split()]
+LOAD_THREADS = int(os.environ.get("LOAD_THREADS", "4"))
+SECONDS = float(os.environ.get("SECONDS", "5"))
+TORRENTS = int(os.environ.get("TORRENTS", "10000"))
+WINDOW = int(os.environ.get("WINDOW", "32"))
+REPEATS = int(os.environ.get("REPEATS", "3"))
+PORT_BASE = int(os.environ.get("PORT_BASE", "19960"))
+
+
+def start(port, workers):
+    directory = tempfile.mkdtemp(prefix="ot-udpscale-")
+    config = "listen.tcp 127.0.0.1:%d\nlisten.udp 127.0.0.1:%d\n" % (port, port)
+    if workers:
+        config += "listen.udp.workers %d\n" % workers
+    with open(os.path.join(directory, "conf"), "w") as handle:
+        handle.write(config)
+    log = open(os.path.join(directory, "log"), "w+b")
+    tracker = subprocess.Popen([TRACKER, "-f", os.path.join(directory, "conf")], stdout=log, stderr=log)
+    for _ in range(300):
+        if tracker.poll() is not None:
+            log.seek(0)
+            raise SystemExit("tracker exited at once on port %d with %d workers:\n%s"
+                             % (port, workers, log.read().decode("utf-8", "replace")))
+        try:
+            socket.create_connection(("127.0.0.1", port), timeout=0.2).close()
+            return tracker, log
+        except OSError:
+            time.sleep(0.02)
+    tracker.kill()
+    raise SystemExit("tracker never listened on %d" % port)
+
+
+def cpu_seconds(pid):
+    """Cumulative cpu time of a process, the one spelling ps has everywhere."""
+    try:
+        out = subprocess.run(["ps", "-o", "time=", "-p", str(pid)], capture_output=True, text=True).stdout.strip()
+    except OSError:
+        return None
+    if not out:
+        return None
+    total, parts = 0.0, out.replace("-", ":").split(":")
+    for part in parts:
+        total = total * 60 + float(part)
+    return total
+
+
+def stop(tracker):
+    tracker.send_signal(signal.SIGINT)
+    try:
+        tracker.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        tracker.kill()
+        tracker.wait()
+
+
+def main():
+    if not os.path.exists(LOADER):
+        raise SystemExit("build the generator first: cc -O2 -pthread -o %s tests/udp_load.c" % LOADER)
+
+    print("  tracker %s, %d load threads x %d in flight, %d torrents, %.0f s x %d runs"
+          % (TRACKER, LOAD_THREADS, WINDOW, TORRENTS, SECONDS, REPEATS))
+    print("  %-8s %12s %12s %12s %7s   %s"
+          % ("workers", "median/s", "lowest", "highest", "cores", "vs 0 workers"))
+    baseline = None
+    for index, workers in enumerate(WORKER_COUNTS):
+        rates, cores = [], []
+        for repeat in range(REPEATS):
+            port = PORT_BASE + index * 16 + repeat
+            tracker, log = start(port, workers)
+            try:
+                # Warm up until the torrent set exists and the peers are in
+                # place, so the measured run is renewals rather than insertions
+                subprocess.run([LOADER, "127.0.0.1", str(port), str(LOAD_THREADS), "2", str(TORRENTS), str(WINDOW)],
+                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+                spent, wall = cpu_seconds(tracker.pid), time.time()
+                result = subprocess.run(
+                    [LOADER, "127.0.0.1", str(port), str(LOAD_THREADS), str(SECONDS), str(TORRENTS), str(WINDOW)],
+                    capture_output=True, text=True,
+                )
+                used = cpu_seconds(tracker.pid)
+            finally:
+                stop(tracker)
+                log.close()
+            head = result.stdout.strip().split()
+            if head and head[0].replace(".", "").isdigit():
+                rates.append(float(head[0]))
+                if used is not None and spent is not None:
+                    cores.append((used - spent) / max(time.time() - wall, 1e-9))
+
+        if not rates:
+            print("  %-8s no result" % workers)
+            continue
+        rates.sort()
+        median = rates[len(rates) // 2]
+        if baseline is None:
+            baseline = median
+        # A spread wider than the differences between rows means the rows are
+        # not telling you anything
+        busy = ("%7.2f" % (sorted(cores)[len(cores) // 2])) if cores else "      ?"
+        print("  %-8s %12.0f %12.0f %12.0f %s   %.2fx  (spread %.0f%%)"
+              % (workers, median, rates[0], rates[-1], busy, median / baseline,
+                 100.0 * (rates[-1] - rates[0]) / median))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/udp_wake_probe.c b/tests/udp_wake_probe.c
new file mode 100644
index 0000000..8a0f7a8
--- /dev/null
+++ b/tests/udp_wake_probe.c
@@ -0,0 +1,99 @@
+/* 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.
+
+   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.
+
+     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. */
+#include <errno.h>
+#include <netinet/in.h>
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdio.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <unistd.h>
+
+static _Atomic int returned;
+static _Atomic int saved_errno;
+static int         sock;
+
+static void       *reader(void *args) {
+  char buffer[64];
+  (void)args;
+  if (recvfrom(sock, buffer, sizeof(buffer), 0, NULL, NULL) < 0)
+    atomic_store(&saved_errno, errno);
+  atomic_store(&returned, 1);
+  return NULL;
+}
+
+static int bind_udp(int timeout_ms) {
+  struct sockaddr_in address;
+  int                s = socket(AF_INET, SOCK_DGRAM, 0);
+
+  memset(&address, 0, sizeof(address));
+  address.sin_family      = AF_INET;
+  address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+  address.sin_port        = 0;
+  bind(s, (struct sockaddr *)&address, sizeof(address));
+
+  if (timeout_ms) {
+    struct timeval how_long;
+    how_long.tv_sec  = timeout_ms / 1000;
+    how_long.tv_usec = (timeout_ms % 1000) * 1000;
+    setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, &how_long, sizeof(how_long));
+  }
+  return s;
+}
+
+static void try_it(const char *name, int timeout_ms, int do_close, int do_shutdown) {
+  pthread_t thread;
+  int       waited;
+
+  atomic_store(&returned, 0);
+  atomic_store(&saved_errno, 0);
+  sock = bind_udp(timeout_ms);
+  pthread_create(&thread, NULL, reader, NULL);
+  usleep(300000); /* let it reach recvfrom */
+
+  if (do_shutdown)
+    shutdown(sock, SHUT_RD);
+  if (do_close)
+    close(sock);
+
+  for (waited = 0; waited < 2000 && !atomic_load(&returned); ++waited)
+    usleep(1000);
+
+  if (atomic_load(&returned)) {
+    printf("  %-24s returned after %4d ms, errno %s\n", name, waited,
+           atomic_load(&saved_errno) ? strerror(atomic_load(&saved_errno)) : "none");
+    pthread_join(thread, NULL);
+  } else {
+    printf("  %-24s STILL BLOCKED after 2000 ms\n", name);
+    /* unblock it so the probe can finish */
+    struct sockaddr_in to;
+    socklen_t          length = sizeof(to);
+    int                poke   = socket(AF_INET, SOCK_DGRAM, 0);
+    if (!getsockname(sock, (struct sockaddr *)&to, &length))
+      sendto(poke, "x", 1, 0, (struct sockaddr *)&to, length);
+    close(poke);
+    usleep(200000);
+    if (atomic_load(&returned))
+      pthread_join(thread, NULL);
+    else
+      pthread_detach(thread);
+  }
+  if (!do_close)
+    close(sock);
+}
+
+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);
+  return 0;
+}