12 files changed, 932 insertions, 18 deletions

diff --git a/opentracker.c b/opentracker.c
index 34b8fd5..f579b92 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -84,7 +84,7 @@ static void install_signal_handlers(void) {
 
   sa.sa_handler = signal_handler;
   sigemptyset(&sa.sa_mask);
-  sa.sa_flags = SA_RESTART;
+  sa.sa_flags = 0;
   if ((sigaction(SIGINT, &sa, NULL) == -1) || (sigaction(SIGALRM, &sa, NULL) == -1))
     panic("install_signal_handlers");
 
@@ -320,8 +320,17 @@ static void *server_mainloop(void *args) {
 }
 
 
+/* A group of sockets sharing one address only spreads datagrams if every
+   member asked for it before binding, the first one included, so whether
+   this entry will have workers has to be known here. */
+static socket_sharing listen_sharing_for(const ot_listen *entry) {
+  if (entry->proto == FLAG_UDP && entry->workers > 1 && socket_can_balance())
+    return SOCKET_BALANCED;
+  return entry->shared ? SOCKET_SHARED : SOCKET_EXCLUSIVE;
+}
+
 static void listen_bind_one(ot_listen *entry) {
-  socket_sharing sharing = entry->shared ? SOCKET_SHARED : SOCKET_EXCLUSIVE;
+  socket_sharing sharing = listen_sharing_for(entry);
 
 #ifdef _DEBUG
   {
@@ -370,6 +379,10 @@ static void listen_bind_all(void) {
     listen_bind_one(entries + i);
 }
 
+/* The socket the event loop was using becomes the first worker's. Where the
+   kernel spreads datagrams, every further worker binds one of its own, so
+   none of them share a receive queue.
+   All others read from the first */
 static void listen_start_udp_workers(void) {
   size_t     count, i;
   ot_listen *entries = config_listen_entries(&count);
@@ -377,14 +390,30 @@ static void listen_start_udp_workers(void) {
   udp_init();
 
   for (i = 0; i < count; ++i) {
-    ot_listen *entry = entries + i;
+    ot_listen *entry   = entries + i;
+    int        balance = listen_sharing_for(entry) == SOCKET_BALANCED;
+    int        worker;
 
     if (entry->proto != FLAG_UDP || entry->workers <= 0)
       continue;
 
     socket_set_blocking((int)entry->sock, 1);
     event_del((int)entry->sock);
-    udp_workers_install(entry->sock, (unsigned int)entry->workers);
+    udp_worker_install(entry->sock);
+
+    for (worker = 1; worker < entry->workers; ++worker) {
+      int64_t sock = entry->sock;
+
+      if (balance) {
+        sock = socket_udp_bind(entry->ip, entry->port, SOCKET_BALANCED);
+        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);
+    }
   }
 }
 
diff --git a/ot_event.c b/ot_event.c
index ce0feef..1542bf5 100644
--- a/ot_event.c
+++ b/ot_event.c
@@ -83,8 +83,10 @@ int event_wait(ot_event *out, int max, int timeout_ms) {
     g_ready_space = max;
   }
 
-  while ((count = epoll_wait(g_epoll, g_ready, max, timeout_ms)) == -1 && errno == EINTR)
-    ;
+  /* An interrupted wait is reported as no events rather than retried here,
+     so that the caller gets to look at whatever the signal was about */
+  if ((count = epoll_wait(g_epoll, g_ready, max, timeout_ms)) == -1)
+    return errno == EINTR ? 0 : -1;
   for (i = 0; i < count; ++i) {
     uint32_t events = g_ready[i].events;
 
@@ -189,8 +191,10 @@ int event_wait(ot_event *out, int max, int timeout_ms) {
     deadline        = &timeout;
   }
 
-  while ((count = kevent(g_kqueue, NULL, 0, g_ready, max, deadline)) == -1 && errno == EINTR)
-    ;
+  /* An interrupted wait is reported as no events rather than retried here,
+     so that the caller gets to look at whatever the signal was about */
+  if ((count = kevent(g_kqueue, NULL, 0, g_ready, max, deadline)) == -1)
+    return errno == EINTR ? 0 : -1;
   for (i = 0; i < count; ++i) {
     uint64_t udata = (uint64_t)(uintptr_t)g_ready[i].udata;
     int      eof   = (g_ready[i].flags & (EV_EOF | EV_ERROR)) != 0;
@@ -310,8 +314,10 @@ int event_del(int fd) {
 int event_wait(ot_event *out, int max, int timeout_ms) {
   int ready, found = 0, i;
 
-  while ((ready = poll(g_pfds, (nfds_t)g_count, timeout_ms)) == -1 && errno == EINTR)
-    ;
+  /* An interrupted wait is reported as no events rather than retried here,
+     so that the caller gets to look at whatever the signal was about */
+  if ((ready = poll(g_pfds, (nfds_t)g_count, timeout_ms)) == -1)
+    return errno == EINTR ? 0 : -1;
   if (ready <= 0)
     return ready;
 
diff --git a/ot_socket.c b/ot_socket.c
index 576087f..158459f 100644
--- a/ot_socket.c
+++ b/ot_socket.c
@@ -75,13 +75,37 @@ static int socket_v6(int type) {
   return s;
 }
 
+/* SO_REUSEPORT means three different things. Linux spreads datagrams over
+   the group. FreeBSD permits the bind and spreads nothing, which is what
+   SO_REUSEPORT_LB was added for. Darwin hands every datagram to whichever
+   socket bound last, so a group there starves all but one reader. */
+#if defined(SO_REUSEPORT_LB)
+#define OT_BALANCE_OPTION SO_REUSEPORT_LB
+#elif defined(SO_REUSEPORT) && !defined(__APPLE__)
+#define OT_BALANCE_OPTION SO_REUSEPORT
+#endif
+
+int socket_can_balance(void) {
+#ifdef OT_BALANCE_OPTION
+  return 1;
+#else
+  return 0;
+#endif
+}
+
 /* The only place in the tree that needs to know what a kernel calls this */
 static void socket_set_sharing(int s, socket_sharing sharing) {
   int one = 1;
 
   setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
+#ifdef OT_BALANCE_OPTION
+  if (sharing == SOCKET_BALANCED) {
+    setsockopt(s, SOL_SOCKET, OT_BALANCE_OPTION, &one, sizeof(one));
+    return;
+  }
+#endif
 #ifdef SO_REUSEPORT
-  if (sharing == SOCKET_SHARED)
+  if (sharing != SOCKET_EXCLUSIVE)
     setsockopt(s, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(one));
 #endif
 }
diff --git a/ot_socket.h b/ot_socket.h
index aa4e2d2..f73a5c9 100644
--- a/ot_socket.h
+++ b/ot_socket.h
@@ -16,8 +16,18 @@
    as the v4 mapped address */
 
 /* How an address may be shared with other sockets. Which socket option says
-   so differs between kernels, which is socket_set_sharing's business */
-typedef enum { SOCKET_EXCLUSIVE, SOCKET_SHARED } socket_sharing;
+   so differs between kernels, which is socket_set_sharing's business.
+
+   SHARED lets another socket bind the same address. BALANCED asks the kernel
+   to spread arriving datagrams over the sockets that did, which is a
+   different thing and not everywhere available: ask socket_can_balance
+   before binding a group that depends on it. */
+typedef enum { SOCKET_EXCLUSIVE, SOCKET_SHARED, SOCKET_BALANCED } socket_sharing;
+
+/* Whether this kernel spreads datagrams over a group of sockets bound to one
+   address. Where it does not, binding a group is worse than one socket with
+   several readers, so the caller wants to know first. */
+int     socket_can_balance(void);
 
 /* Create, bind and, for TCP, listen. The returned socket is non blocking
    Returns -1 on failure with errno set */
diff --git a/ot_udp.c b/ot_udp.c
index 14210c3..87741da 100644
--- a/ot_udp.c
+++ b/ot_udp.c
@@ -235,13 +235,12 @@ void udp_init(void) {
     udp_generate_rijndael_round_key();
 }
 
-void udp_workers_install(int64_t sock, unsigned int worker_count) {
+void udp_worker_install(int64_t sock) {
   pthread_t thread_id;
 
 #ifdef _DEBUG
-  fprintf(stderr, "Installing %d workers on udp socket %ld\n", worker_count, (unsigned long)sock);
+  fprintf(stderr, "Installing a worker on udp socket %ld\n", (unsigned long)sock);
 #endif
-  while (worker_count--)
-    pthread_create(&thread_id, NULL, udp_worker, (void *)sock);
+  pthread_create(&thread_id, NULL, udp_worker, (void *)sock);
 }
 
diff --git a/ot_udp.h b/ot_udp.h
index a8ab9fc..7d4432f 100644
--- a/ot_udp.h
+++ b/ot_udp.h
@@ -11,7 +11,8 @@
 #include "trackerlogic.h"
 
 void udp_init(void);
-void udp_workers_install(int64_t sock, unsigned int worker_count);
+/* One thread reading one socket. Called once per worker. */
+void udp_worker_install(int64_t sock);
 int  handle_udp6(int64_t serversocket, struct ot_workstruct *ws);
 
 #endif
diff --git a/tests/integration/harness.py b/tests/integration/harness.py
index 03a0561..122c36d 100644
--- a/tests/integration/harness.py
+++ b/tests/integration/harness.py
@@ -121,6 +121,7 @@ class Tracker:
         all: LeakSanitizer reports from an exit handler, so a killed tracker
         reports nothing and the suite passes over leaks it never looked for.
         """
+        refused_to_exit = False
         if self.process is not None and self.process.poll() is None:
             self.process.send_signal(signal.SIGINT)
             try:
@@ -128,11 +129,16 @@ class Tracker:
             except subprocess.TimeoutExpired:
                 self.process.kill()
                 self.process.wait()
+                refused_to_exit = True
         complaints = [
             line
             for line in self.output().splitlines()
             if "Sanitizer" in line or "runtime error" in line or line.startswith("ERROR")
         ]
+        if refused_to_exit:
+            # Killing it throws away whatever the sanitizer had to say on the
+            # way out, so a test that ends this way has checked nothing.
+            complaints.append("the tracker did not exit within 5 s of SIGINT and had to be killed")
         self._cleanup()
         return complaints
 
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..7691fcb
--- /dev/null
+++ b/tests/udp_scaling.py
@@ -0,0 +1,159 @@
+"""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.
+
+Three columns say which side the number came from. "cores" is the tracker's
+cpu, "load" the generator's. Whichever is near its machine's core count is
+the one being measured. "lost" counts datagrams that went unanswered: once
+it leaves zero the throughput figure is timeouts, not service, and the run
+needs a smaller WINDOW.
+"""
+import os
+import resource
+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 reaped_cpu_seconds():
+    """Cpu of every child already waited for. The generator is the only one
+       that starts and finishes inside a measured run."""
+    spent = resource.getrusage(resource.RUSAGE_CHILDREN)
+    return spent.ru_utime + spent.ru_stime
+
+
+def stop(tracker):
+    tracker.send_signal(signal.SIGINT)
+    try:
+        tracker.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        tracker.kill()
+        tracker.wait()
+
+
+def parse(line):
+    """replies/s and the count of announces that were never answered"""
+    words = line.split()
+    if not words or not words[0].replace(".", "").isdigit():
+        return None, None
+    lost = 0
+    if "lost," in words:
+        lost = int(words[words.index("lost,") - 1])
+    return float(words[0]), lost
+
+
+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 %7s %7s %7s %10s   %s"
+          % ("workers", "median/s", "spread", "cores", "load", "lost", "vs 0 workers"))
+    baseline = None
+    for index, workers in enumerate(WORKER_COUNTS):
+        results = []
+        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, offered, wall = cpu_seconds(tracker.pid), reaped_cpu_seconds(), 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, drove, elapsed = cpu_seconds(tracker.pid), reaped_cpu_seconds(), max(time.time() - wall, 1e-9)
+            finally:
+                stop(tracker)
+                log.close()
+            rate, lost = parse(result.stdout.strip())
+            if rate is None:
+                continue
+            cores = (used - spent) / elapsed if (used is not None and spent is not None) else None
+            results.append((rate, cores, (drove - offered) / elapsed, lost))
+
+        if not results:
+            print("  %-8s no result" % workers)
+            continue
+        results.sort()
+        rate, cores, load, lost = results[len(results) // 2]
+        if baseline is None:
+            baseline = rate
+        # A spread wider than the differences between rows means the rows are
+        # not telling you anything
+        spread = 100.0 * (results[-1][0] - results[0][0]) / rate
+        print("  %-8s %12.0f %6.0f%% %7s %7.2f %10d   %.2fx"
+              % (workers, rate, spread, ("%.2f" % cores) if cores else "?", load, lost, rate / baseline))
+        sys.stdout.flush()
+
+
+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;
+}