From 701785b0c23b12e7e9c03dbaae3a107f7de4f178 Mon Sep 17 00:00:00 2001 From: Shannon Atkinson Date: Fri, 4 Sep 2026 12:29:54 -0700 Subject: [PATCH] test(acceptance): measure what a destination costs, and whether N cost N times it #380's concurrency gap. README.md:59 and docs/COMPARISON.md:268 both publish "roughly 4% of one core per destination" and nothing tested it -- a reader sizes a box with that number. WHAT IT MEASURES, through the real server rather than raw ffmpeg: one destination, then six, sampling cumulative CPU per child out of the process table. `time` rather than `%cpu`, for the reason the ladder driver already records -- %cpu averages over a process's whole lifetime, so a child up for a minute barely moves it. THE FALSE RESULT IT REFUSES, which is why liveness is asserted before cost everywhere. Measuring this by hand against raw ffmpeg produced: 1 destination: 3.80% of a core 2 destinations: 2.23% each 4 destinations: 1.11% each 8 destinations: 0.55% each Per-destination cost halving with every doubling: exactly the result anyone measuring this wants, and an artefact. Every run had ONE surviving process -- several ffmpeg readers on one UDP unicast socket compete for packets and all but one die, so the total was one survivor's CPU divided by N. It was caught only because that harness happened to print a liveness count. The product does not have the problem: internal/relay.Hub gives each destination its own subscription port, which this run confirms with six alive out of six. AND THE FIGURE IT CORRECTS IS MY OWN. That standalone harness also reported 8.8% of a core for one destination, which I posted to #380 as "about double the documented 4%". Measured through the product instead: 2.69% for one, 3.44% each for six. The published ~4% is corroborated, not contradicted, and the 8.8% was an artefact of measuring something that was not a destination -- different args, file output, no relay. The correction is going on the issue too. WHAT IT ASSERTS IS THE SHAPE, NOT A PERCENTAGE. "4% of a core" is a property of a machine; a suite pinning a number would fail on hardware rather than on regressions. The three that hold anywhere: - every destination asked for is still running at the end - every one burned measurable CPU (alive and costing nothing is not delivering, which is why that is a separate check) - the Nth costs about what the first did, within 0.4x-2.5x The absolute figure is PRINTED every run, with the note that this machine is not the six-core VPS the README's number came from, so a difference is not by itself a defect in either. VALIDATED BY WATCHING IT FAIL, and the first attempt to do that did not work. SIGKILLing the destination children was not enough: supervisor.Spec carries AutoRestart, so they came back before the sample and the suite passed a run it should have failed. Stopping them through the API instead RETIRES them, and then it fails correctly -- naming the count, and printing the "below the band" branch that says work which appears to vanish as load grows usually has not been done. A guard nobody has watched fail is a guard nobody should trust, and this one needed two goes. Not wired into CI in this change. It spawns seven ffmpeg processes and measures for a minute; where it belongs in the matrix is a separate decision from whether it works. Refs #380. Claude-Session: https://claude.ai/code/session_01A8N3W5ct9SZtHK9sCDD9cL --- scripts/acceptance-concurrency.sh | 169 +++++++++++ scripts/acceptance_concurrency_driver.go | 339 +++++++++++++++++++++++ 2 files changed, 508 insertions(+) create mode 100755 scripts/acceptance-concurrency.sh create mode 100644 scripts/acceptance_concurrency_driver.go diff --git a/scripts/acceptance-concurrency.sh b/scripts/acceptance-concurrency.sh new file mode 100755 index 00000000..465d11a7 --- /dev/null +++ b/scripts/acceptance-concurrency.sh @@ -0,0 +1,169 @@ +#!/usr/bin/env bash +# +# What does a destination cost, and do N of them cost N times it? +# +# #380's concurrency gap. README.md and docs/COMPARISON.md both publish "roughly +# 4% of one core per destination" and nothing tests it -- a reader sizes a box +# with that number. +# +# THE FALSE RESULT THIS SUITE REFUSES. Measuring this by hand, against raw +# ffmpeg rather than through the product, produced: +# +# 1 destination: 3.80% of a core +# 2 destinations: 2.23% each +# 4 destinations: 1.11% each +# 8 destinations: 0.55% each +# +# Per-destination cost halving with every doubling, which is exactly the result +# anyone measuring this WANTS. It was an artefact: every run had one surviving +# process. Several ffmpeg readers on one UDP unicast socket compete for packets +# and all but one die, so the "total" was one survivor's CPU divided by N. +# +# It was caught only because the harness happened to count survivors. So this +# suite asserts LIVENESS BEFORE COST, everywhere, and the driver refuses to +# report a number without it. The product does not have the competing-reader +# problem -- internal/relay.Hub gives each destination its own subscription port +# -- which is why this measures through the real server. +# +# WHAT IT ASSERTS IS THE SHAPE, NOT THE PERCENTAGE. "4% of a core" is a property +# of a machine: the same destination measures 8.8% on Apple silicon against a +# published 4% on a six-core VPS, and neither is wrong. A suite pinning a number +# would fail on hardware rather than on regressions. The absolute figure is +# PRINTED every run, because it is what the README claims and somebody should be +# able to read it off a CI log. +# +# Usage: ./scripts/acceptance-concurrency.sh [workdir] +set -u + +WORK="${1:-/tmp/polyemesis-acceptance-concurrency}" +PORT=8098 +# How many destinations the N case runs. Six rather than sixteen because the +# smallest CI runner has two cores and this is a linearity check, not a load +# test: sixteen stream-copy destinations on two cores measures the runner's +# scheduler, not the product. +DESTS="${CONCURRENCY_DESTS:-6}" + +SCRIPTS="$(cd "$(dirname "$0")" && pwd)" +. "$SCRIPTS/lib-cleanup.sh" +. "$SCRIPTS/lib-watchdog.sh" +ROOT="$(cd "$SCRIPTS/.." && pwd)" +BIN="$ROOT/polyemesis" +. "$SCRIPTS/lib-preflight.sh" + +pass=0; fail=0 +ok() { printf " \033[32mPASS\033[0m %s\n" "$1"; pass=$((pass+1)); } +bad() { printf " \033[31mFAIL\033[0m %s\n" "$1"; fail=$((fail+1)); } +note() { printf " %s\n" "$1"; } +step() { printf "\n\033[1m%s\033[0m\n" "$1"; poly_step_record "$1"; } + +cleanup() { + pkill -f "acceptance-concurrency-source" 2>/dev/null + poly_cleanup_exit "${1:-0}" "$PORT" "$WORK" +} +trap 'poly_teardown_trap $? cleanup' EXIT + +poly_require_exec "$BIN" +poly_require_cmd go "needed to run the acceptance driver via 'go run'" +poly_require_cmd ffmpeg +poly_require_cmd ps "the measurement reads cumulative CPU out of the process table" + +rm -rf "$WORK"; mkdir -p "$WORK"; cd "$WORK" || exit 1 +poly_watchdog_arm + +step "1. Start the binary" +"$BIN" -addr ":$PORT" -data ./data -log warn > server.log 2>&1 & +for _ in $(seq 1 40); do + sleep 0.3 + if grep -q "web ui" server.log 2>/dev/null; then break; fi +done +sleep 1 +grep -q "polyemesis" server.log && ok "server started" || bad "server did not start" + +SRVPID=$(pgrep -f "polyemesis -addr :$PORT" | head -1) +RELAY=$(lsof -nP -iUDP -a -p "$SRVPID" 2>/dev/null | awk '/UDP 127.0.0.1/{split($NF,a,":"); print a[2]; exit}') + +step "2. Measure one destination, then $DESTS (via the API the UI uses)" +FACTS="$WORK/facts.env" +# RUN FROM $ROOT, IN A SUBSHELL. `go run` resolves a module import against the +# CURRENT directory's go.mod rather than the source file's location, and this +# suite has already cd'd into $WORK, which is under /tmp and inside no module. +# The same trap is recorded in driverlib's package comment. +( cd "$ROOT" && go run "$SCRIPTS/acceptance_concurrency_driver.go" \ + "$PORT" "$RELAY" "$FACTS" "$DESTS" 2>&1 ) | sed 's/^/ /' + +[[ -s "$FACTS" ]] || { bad "driver wrote no facts"; step "Summary"; printf " %d passed, %d failed\n\n" "$pass" "$fail"; exit 1; } +# shellcheck disable=SC1090 +source "$FACTS" + +if [[ -n "${DRIVER_FAILED:-}" ]]; then bad "driver aborted: $DRIVER_FAILED"; fi + +# ------------------------------------------------------------------ 3. alive +# +# BEFORE ANY COST CHECK. This is the whole lesson of the false result above: a +# per-destination figure computed over processes that are no longer running is +# not a small error, it is a number pointing the wrong way. +step "3. Every destination that was asked for is still running" +if [[ "${CONC_ALIVE_1:-0}" == "1" ]]; then + ok "the single destination survived its measurement window" +else + bad "the single destination did not survive: alive=${CONC_ALIVE_1:-0}" +fi +if [[ "${CONC_ALIVE_N:-0}" == "$DESTS" ]]; then + ok "all $DESTS destinations survived their measurement window" +else + bad "asked for $DESTS destinations, ${CONC_ALIVE_N:-0} were alive at the end" + note "This is the failure the suite exists to catch. Every per-destination" + note "number below is meaningless while it holds -- dividing one survivor's" + note "CPU by $DESTS is what produced a perfect halving and looked like news." +fi + +# ------------------------------------------------------------------ 4. work +step "4. Every destination did measurable work" +if [[ "${CONC_ZERO_CPU_N:-1}" == "0" ]]; then + ok "no destination burned zero CPU across the window" +else + bad "${CONC_ZERO_CPU_N} destination(s) burned no measurable CPU" + note "A process that is alive and costs nothing is not delivering. Alive is" + note "necessary and not sufficient, which is why this check is separate." +fi + +# ------------------------------------------------------------- 5. linearity +# +# THE PROPERTY, rather than a percentage. What must hold on every machine is +# that the Nth destination costs about what the first did. The bounds are wide +# on purpose: a two-core runner with six destinations contends, and contention +# is a fact about the runner. +step "5. The Nth destination costs about what the first did" +RATIO="${CONC_RATIO:-0}" +LOW=0.4 +HIGH=2.5 +if awk -v r="$RATIO" -v lo="$LOW" -v hi="$HIGH" 'BEGIN{exit !(r>=lo && r<=hi)}'; then + ok "per-destination cost at $DESTS is ${RATIO}x the cost at 1 (within ${LOW}-${HIGH})" +else + bad "per-destination cost at $DESTS is ${RATIO}x the cost at 1, outside ${LOW}-${HIGH}" + if awk -v r="$RATIO" -v lo="$LOW" 'BEGIN{exit !(r ") + } + port, relay := os.Args[1], os.Args[2] + factsFile = os.Args[3] + n, err := strconv.Atoi(os.Args[4]) + if err != nil || n < 2 { + die("destination count %q must be an integer >= 2", os.Args[4]) + } + + ffmpegBin = toolPath("ffmpeg") + psBin = toolPath("ps") + + driverlib.Init("http://127.0.0.1:" + port) + defer writeFacts() + driverlib.WaitUp() + driverlib.Setup("admin", "acceptance-pw") + sourceID = driverlib.EnsureSource("Main") + + // Recording and metering off, for the same reason the ladder suite turns + // them off: both spawn FFmpeg of their own, and this suite's whole number + // is what the DESTINATIONS cost. A recorder competing for the same cores + // moves it for a reason that has nothing to do with concurrency. + settings := driverlib.LoadSettings() + if rec, ok := settings["recording"].(map[string]any); ok { + rec["enabled"] = false + } + if m, ok := settings["meters"].(map[string]any); ok { + m["enabled"] = false + } + driverlib.SaveSettings(settings, "recording and meters off") + + relayPort, err := driverlib.ResolveRelayPort(relay, func(p string) map[string]any { + var doc map[string]any + driverlib.GetJSON(p, "status", &doc) + return doc + }) + if err != nil { + die("resolve relay port: %v", err) + } + + // -t bounds the publisher rather than trusting the teardown: a source that + // outlives a failed run keeps a port bound and the next run cannot start. + src := exec.Command(ffmpegBin, "-hide_banner", "-loglevel", "error", "-re", + "-f", "lavfi", "-i", "testsrc2=size=1280x720:rate=30", + "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000", + "-map", "0:v", "-map", "1:a", + "-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency", + "-g", "60", "-b:v", "3000k", "-c:a", "aac", "-b:a", "128k", + "-metadata", "comment=acceptance-concurrency-source", "-t", "300", + "-f", "mpegts", "-flush_packets", "1", + fmt.Sprintf("udp://127.0.0.1:%d?pkt_size=1316", relayPort)) + if err := src.Start(); err != nil { + die("start source: %v", err) + } + defer func() { + _ = src.Process.Kill() + _ = src.Wait() + }() + fmt.Printf("source publishing to udp/%d\n", relayPort) + + // ---------------------------------------------------------------- one + // + // The baseline every later number is read against. Measured FIRST and on + // its own, because a per-destination cost is only meaningful next to the + // cost of one. + newDest("conc-1", "conc-1.mkv") + one := measure("1 destination", 1) + + // ---------------------------------------------------------------- N + for i := 2; i <= n; i++ { + newDest(fmt.Sprintf("conc-%d", i), fmt.Sprintf("conc-%d.mkv", i)) + } + many := measure(fmt.Sprintf("%d destinations", n), n) + + // ------------------------------------------------------------- verdict + perOne := one.cores + perMany := many.cores / float64(n) + ratio := 0.0 + if perOne > 0 { + ratio = perMany / perOne + } + + fmt.Printf("\n cost of one destination: %.4f cores (%.2f%% of a core)\n", perOne, perOne*100) + fmt.Printf(" cost of %d, per destination: %.4f cores (%.2f%% of a core)\n", n, perMany, perMany*100) + fmt.Printf(" linearity ratio (per-N / per-1): %.2f\n", ratio) + + facts["CONC_N"] = strconv.Itoa(n) + facts["CONC_ALIVE_1"] = strconv.Itoa(one.alive) + facts["CONC_ALIVE_N"] = strconv.Itoa(many.alive) + facts["CONC_CORES_1"] = fmt.Sprintf("%.5f", one.cores) + facts["CONC_CORES_N"] = fmt.Sprintf("%.5f", many.cores) + facts["CONC_PER_1_PCT"] = fmt.Sprintf("%.2f", perOne*100) + facts["CONC_PER_N_PCT"] = fmt.Sprintf("%.2f", perMany*100) + facts["CONC_RATIO"] = fmt.Sprintf("%.3f", ratio) + facts["CONC_ZERO_CPU_N"] = strconv.Itoa(many.zero) +} + +type reading struct { + // alive is how many destination children were still running at the END of + // the window. It is reported before cost everywhere, because a cost + // computed over dead processes is the false result this suite exists to + // refuse. + alive int + // zero counts survivors that burned no measurable CPU across the window. + zero int + // cores is the total, summed across every destination child. + cores float64 +} + +// measure samples every destination child over a window and returns what they +// cost, having first established that they are all still there. +// +// `time` rather than `%cpu`, which is the difference between a measurement and +// a guess -- %cpu on both platforms averages over the process's whole LIFETIME, +// so a child that has been up a minute barely moves it. The ladder driver +// records the same reasoning; this is the same technique against a different +// population. +func measure(label string, want int) reading { + // Settle before sampling. A destination's first seconds are spent probing + // its relay subscription, which costs more than steady state and would + // flatter or penalise whichever count is measured first. + time.Sleep(8 * time.Second) + + before := destCPU() + if len(before) < want { + fmt.Printf(" %-16s FAILED to start: %d of %d destination children exist\n", + label, len(before), want) + } + const window = 20 * time.Second + t0 := time.Now() + time.Sleep(window) + after := destCPU() + elapsed := time.Since(t0).Seconds() + + r := reading{alive: len(after)} + // SURVIVORS ONLY, and pids present in BOTH samples. A child that died + // mid-window contributed real CPU to `before` and nothing to `after`; + // counting it would credit the survivors with its work and understate the + // per-destination cost -- which is the exact direction the false result + // went. + var pids []string + for pid := range after { + if _, ok := before[pid]; ok { + pids = append(pids, pid) + } + } + sort.Strings(pids) + for _, pid := range pids { + d := after[pid] - before[pid] + if d <= 0 { + r.zero++ + continue + } + r.cores += d / elapsed + } + + fmt.Printf(" %-16s asked for %d, alive %d, measured %d, %.4f cores total\n", + label, want, r.alive, len(pids), r.cores) + return r +} + +// destCPU returns cumulative CPU seconds for every destination child, by pid. +// +// Matched on the output path this suite gives its destinations, which is what +// separates them from the ingest, the source publisher and anything else on the +// machine. Matching on "ffmpeg" alone would sweep in the publisher started +// above and roughly double every number here. +func destCPU() map[string]float64 { + out := map[string]float64{} + for _, r := range psProcs() { + if strings.Contains(r.args, "conc-") && strings.Contains(r.args, ".mkv") && + strings.Contains(r.args, "ffmpeg") { + out[r.pid] = r.cpu + } + } + return out +} + +type psRow struct { + pid string + cpu float64 + args string +} + +func psProcs() []psRow { + out, err := exec.Command(psBin, "-Ao", "pid=,time=,args=").Output() + if err != nil { + die("ps: %v", err) + } + var rows []psRow + for _, line := range strings.Split(string(out), "\n") { + f := strings.Fields(line) + if len(f) < 3 { + continue + } + rows = append(rows, psRow{pid: f[0], cpu: parseCPUTime(f[1]), args: strings.Join(f[2:], " ")}) + } + return rows +} + +// parseCPUTime reads ps's TIME column on both platforms this runs on: darwin +// prints MM:SS.CC and Linux prints [DD-]HH:MM:SS, so the fractional part exists +// on one and not the other and the number of colons differs. Parsed from the +// RIGHT, which is the one reading correct for both. +func parseCPUTime(s string) float64 { + if i := strings.Index(s, "-"); i >= 0 { + days, _ := strconv.ParseFloat(s[:i], 64) + return days*86400 + parseCPUTime(s[i+1:]) + } + parts := strings.Split(s, ":") + mult, total := 1.0, 0.0 + for i := len(parts) - 1; i >= 0; i-- { + v, err := strconv.ParseFloat(parts[i], 64) + if err != nil { + return 0 + } + total += v * mult + mult *= 60 + } + return total +} + +// newDest creates one file destination through the API the UI uses. +// +// Through driverlib.CreateDest rather than a hand-rolled POST, so this suite +// inherits the sourceId handling every other driver has: the server refuses a +// create that does not name its programme, and filling that in each caller is +// how the other suites drifted. +func newDest(name, file string) { + driverlib.CreateDest(name, map[string]any{ + "sourceId": sourceID, + "name": name, "kind": "file", "platform": "custom", "url": file, + "enabled": true, "audioBitrate": 160, + "profile": map[string]any{ + "mode": "simple", "tracks": driverlib.Sel(0), + "normalize": "auto", "sampleRate": 48000, + }, + }) +} + +func toolPath(name string) string { + p, err := exec.LookPath(name) + if err != nil { + die("%s is required: %v", name, err) + } + return p +} + +func writeFacts() { + var b strings.Builder + keys := make([]string, 0, len(facts)) + for k := range facts { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + fmt.Fprintf(&b, "%s=%s\n", k, facts[k]) + } + if err := os.WriteFile(factsFile, []byte(b.String()), 0o644); err != nil { + fmt.Fprintf(os.Stderr, "write facts: %v\n", err) + } +} + +func die(format string, a ...any) { + facts["DRIVER_FAILED"] = fmt.Sprintf(format, a...) + writeFacts() + fmt.Fprintf(os.Stderr, "driver: "+format+"\n", a...) + os.Exit(1) +}