From 9ec918187ab65488ebcce427f5a14a6b38e87b78 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Wed, 28 Oct 2026 17:11:33 +0100 Subject: Add load tests to sanitize check the trackerlogic, based on udp requests --- tests/udp_load.c | 202 +++++++++++++++++++++++++++++++++++++++++ tests/udp_modes.c | 240 +++++++++++++++++++++++++++++++++++++++++++++++++ tests/udp_modes.py | 139 ++++++++++++++++++++++++++++ tests/udp_scaling.py | 135 ++++++++++++++++++++++++++++ tests/udp_wake_probe.c | 99 ++++++++++++++++++++ 5 files changed, 815 insertions(+) create mode 100644 tests/udp_load.c create mode 100644 tests/udp_modes.c create mode 100644 tests/udp_modes.py create mode 100644 tests/udp_scaling.py create mode 100644 tests/udp_wake_probe.c (limited to 'tests') 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 +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#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 +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#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 +#include +#include +#include +#include +#include +#include +#include +#include + +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; +} -- cgit v1.3