Convert UDP stress test from echo to one-way

.github/workflows/udp-stress-tests.yml

@@ -2,7 +2,7 @@ name: UDP stress tests
 
 on:
   schedule:
-    - cron: "30 10 * * *"
+    - cron: "0 0,8,16 * * *"
   workflow_dispatch:
 
 permissions:

stress-tests/udp-flood/README.md

@@ -1,9 +1,10 @@
 # UDP flood stress engine
 
-A count-driven UDP workload for stressing lori's UDP stack. A fixed number of
-clients send stamped datagrams to a server through a single echo socket; the
-server echoes each datagram back to its sender, and the client verifies the
-echo byte-for-byte against a per-client keystream.
+A count-driven, one-way UDP workload for stressing lori's UDP stack. A fixed
+number of clients send stamped datagrams to a server; the server verifies the
+payload of each received datagram against a per-client keystream and reports
+results via actor messaging. There is no UDP echo: the only UDP traffic is
+client-to-server.
 
 This stress test was written to empirically verify that the ASIO backend
 delivers persistent edge-triggered notifications correctly for UDP sockets
@@ -18,10 +19,8 @@ edge-triggered notifications for UDP sockets across many read-loop re-entries.
 Each flag is tied to a distinct code path in `udp_socket.pony`:
 
 - `--datagrams` / `--payload-size` -- volume and per-datagram size.
-- `--batch-size` -- how many datagrams a client sends before waiting for echoes.
-  With batch 1 the client waits for each echo before sending the next (one
-  event delivery per round-trip). Larger batches burst datagrams and exercise
-  the read loop's datagram-count and byte budgets.
+- `--batch-size` -- how many datagrams a client sends per scheduling turn
+  before yielding. The client sends continuously until all datagrams are sent.
 - `--clients` -- concurrent client sockets sending to the same server.
 - `--read-buffer-size` -- the per-socket read buffer, which sets the byte
   budget in `_pending_reads`.
@@ -30,18 +29,18 @@ Each flag is tied to a distinct code path in `udp_socket.pony`:
 
 ## Oracles
 
-- **Echo integrity** -- each client sends a per-client pseudo-random byte
-  stream (the byte at position `p` is the low 8 bits of a splitmix64 hash of
-  the client id and `p`) and verifies every echoed byte against it.
-  Position-based: the Nth echo must match the Nth datagram sent (UDP preserves
-  order on loopback).
-- **Conservation** -- every client must send and verify all its datagrams; the
-  `RESULT` line reports the tally.
+- **Conservation** -- two counters tracked independently: client-side sent and
+  server-side received. At the end of a run, client_sent must equal
+  server_received. At the volumes this engine runs, the server's receive buffer
+  is large enough to hold every datagram; a conservation failure indicates a
+  lori event-delivery bug, not expected UDP loss.
+- **Payload integrity** -- the server reads a 4-byte header (client id +
+  sequence number), regenerates the expected keystream for that position, and
+  compares. A mismatch is corruption.
 - **Crash / assert** -- debug build, asserts on.
 
-On success (every client verified) the engine prints `RESULT ...` then `PASS`
-and returns. Anything short of full verification prints `FAIL` and exits
-non-zero.
+On success (every invariant holds) the engine prints `RESULT ...` then `PASS`
+and returns. Anything short of that prints `FAIL` and exits non-zero.
 
 ## Building and running
 
@@ -55,7 +54,7 @@ make stress-tests config=debug ssl=3.0.x     # -> build/debug/udp-flood
 Run the engine directly for a single workload:
 
 ```bash
-build/debug/udp-flood --datagrams 1000 --clients 8 --payload-size 256 \
+build/debug/udp-flood --datagrams 100 --clients 4 --payload-size 256 \
   --batch-size 10
 ```
 

stress-tests/udp-flood/orchestrate_udp.py

@@ -5,7 +5,8 @@
 batch sizes, client counts, and read-loop tuning -- and runs the prebuilt engine
 binary once. Omission is the swarm mechanism: each lever is drawn independently,
 so different seeds push lori's UDP stack down different code paths. A failure
-(echo mismatch, crash, or hang) writes a bundle recording the seed.
+(payload corruption, conservation violation, crash, or hang) writes a bundle
+recording the seed.
 
 Build the engine with `make stress-tests config=debug ssl=<version>` and point
 `--binary` at the result (`build/debug/udp-flood`); this orchestrator only runs
@@ -43,18 +44,18 @@
 
 DEFAULT_PROFILE = {
     "payload_sizes": [1, 8, 64, 256, 1024, 4096, 8192],
-    "batch_sizes": [1, 5, 10, 50, 100],
-    "clients": [1, 2, 4, 8, 16, 32],
+    "batch_sizes": [1, 3, 5, 10, 15],
+    "clients": [1, 2, 4],
     "read_buffer_sizes": [1024, 4096, 16384, 65536],
     "max_datagrams_per_turn": [1, 4, 16, 64, 256],
-    "datagram_buckets": {"small": (10, 100), "medium": (101, 1000),
-                         "large": (1001, 5000)},
+    "datagram_buckets": {"small": (10, 50), "medium": (51, 200),
+                         "large": (201, 500)},
 }
 
 WORKLOAD_PROFILES = {"default": DEFAULT_PROFILE}
 
-RUN_MAX_ROUND_TRIPS = 500_000        # clients * datagrams
-RUN_MAX_BYTES = 2_000_000_000        # ~2 GB moved per run
+RUN_MAX_ROUND_TRIPS = 10_000         # clients * datagrams
+RUN_MAX_BYTES = 50_000_000           # ~50 MB moved per run
 MAX_BURST_BYTES = 64_000             # initial burst (clients * batch * payload);
                                      # must fit in the server's UDP receive buffer
                                      # (Windows default ~64 KB is the floor)
@@ -123,8 +124,8 @@ def resolve_config(master_seed, max_threads, profile="default"):
         workload["clients"], workload["datagrams"], payload)
 
     # UDP drops on localhost when actors can't drain receive buffers fast
-    # enough: the kernel drops datagrams and the echo oracle hangs.  Keep
-    # clients ≤ 1.5× the available threads; reduce clients when needed.
+    # enough: the kernel drops datagrams and the conservation oracle fires.
+    # Keep clients ≤ 1.5× the available threads; reduce clients when needed.
     max_safe_clients = max(1, max_threads * 3 // 2)
     if workload["clients"] > max_safe_clients:
         workload["clients"] = max_safe_clients
@@ -175,7 +176,8 @@ def build_argv(binary, config):
 def parse_result(stdout):
     """Extract the engine's RESULT tally, or {}."""
     result = {}
-    for key in ("clients", "completed", "verified", "mismatched", "bind_failed"):
+    for key in ("clients", "client_sent", "server_received",
+                "server_corrupted", "client_send_errors", "bind_failed"):
         match = re.search(r"\b" + key + r"=(\d+)", stdout)
         if match is not None:
             result[key] = int(match.group(1))
@@ -207,23 +209,23 @@ def _watchdog_kill_reason(now, start, last_progress, timeout,
     return None
 
 
-_DONE_RE = re.compile(rb"HEARTBEAT done=(\d+)")
+_REPORTED_RE = re.compile(rb"HEARTBEAT reported=(\d+)")
 
 
-def _parse_done(line):
-    match = _DONE_RE.search(line)
+def _parse_reported(line):
+    match = _REPORTED_RE.search(line)
     return int(match.group(1)) if match is not None else None
 
 
-def _is_progress(done, max_done):
-    return (done is not None) and (done > max_done)
+def _is_progress(reported, max_reported):
+    return (reported is not None) and (reported > max_reported)
 
 
 def _watch_for_progress(proc, timeout, no_progress_seconds,
                         poll=time.monotonic, sleep=time.sleep):
     start = poll()
     last_progress = [start]
-    max_done = [-1]
+    max_reported = [-1]
     lock = threading.Lock()
     chunks = {"out": [], "err": []}
 
@@ -232,10 +234,10 @@ def drain(stream, key, track_progress):
             for line in iter(stream.readline, b""):
                 chunks[key].append(line)
                 if track_progress:
-                    done = _parse_done(line)
+                    reported = _parse_reported(line)
                     with lock:
-                        if _is_progress(done, max_done[0]):
-                            max_done[0] = done
+                        if _is_progress(reported, max_reported[0]):
+                            max_reported[0] = reported
                             last_progress[0] = poll()
         finally:
             stream.close()
@@ -381,9 +383,11 @@ def probe_max_threads(binary):
 def summary_line(config, result):
     parsed = parse_result(result.stdout)
     shape = config["workload"]
-    detail = ("clients=%s completed=%s verified=%s mismatched=%s bind_failed=%s"
-              % (parsed.get("clients", "?"), parsed.get("completed", "?"),
-                 parsed.get("verified", "?"), parsed.get("mismatched", "?"),
+    detail = ("clients=%s client_sent=%s server_received=%s "
+              "server_corrupted=%s bind_failed=%s"
+              % (parsed.get("clients", "?"), parsed.get("client_sent", "?"),
+                 parsed.get("server_received", "?"),
+                 parsed.get("server_corrupted", "?"),
                  parsed.get("bind_failed", "?")))
     return ("[seed %d] %s (payload=%s datagrams=%s batch=%s clients=%s) %s"
             % (config["master_seed"], result.outcome.upper(),
@@ -493,7 +497,7 @@ def main():
                                "pass")
     parser.add_argument("--no-progress-seconds", type=int,
                         default=DEFAULT_NO_PROGRESS_SECONDS,
-                        help="hang threshold: fail a run whose completed count has "
+                        help="hang threshold: fail a run whose reported count has "
                              "not advanced for this long")
     parser.add_argument("--timeout-seconds", type=int,
                         default=DEFAULT_TIMEOUT_SECONDS,

stress-tests/udp-flood/orchestrate_udp_test.py

@@ -33,7 +33,7 @@ def test_clamp_run():
           (c * d * 8192) <= o.RUN_MAX_BYTES)
     check("clamp: leaves at least MIN_DATAGRAMS datagrams", d >= o.MIN_DATAGRAMS)
     # A small config is left untouched.
-    check("clamp: small config unchanged", o.clamp_run(4, 100, 64) == (4, 100))
+    check("clamp: small config unchanged", o.clamp_run(2, 50, 64) == (2, 50))
     # Pathological input exercises the last-resort client-trimming path.
     c3, d3 = o.clamp_run(30000, 100000, 8192)
     check("clamp: pathological input stays under ceilings",
@@ -232,12 +232,13 @@ def test_build_argv():
 
 
 def test_parse_result():
-    out = ("RESULT clients=4 completed=4 verified=4 "
-           "mismatched=0 bind_failed=0\nPASS")
+    out = ("RESULT clients=4 client_sent=400 server_received=400 "
+           "server_corrupted=0 client_send_errors=0 bind_failed=0\nPASS")
     parsed = o.parse_result(out)
     check("parse_result: reads the tally",
-          parsed == {"clients": 4, "completed": 4, "verified": 4,
-                     "mismatched": 0, "bind_failed": 0})
+          parsed == {"clients": 4, "client_sent": 400, "server_received": 400,
+                     "server_corrupted": 0, "client_send_errors": 0,
+                     "bind_failed": 0})
     check("parse_result: empty on garbage", o.parse_result("nothing here") == {})
 
 
@@ -295,13 +296,13 @@ def test_watchdog_kill_reason():
           == "backstop")
 
 
-def test_parse_done():
-    check("parse_done: reads the count from a heartbeat",
-          o._parse_done(b"HEARTBEAT done=1234 of 44688\n") == 1234)
-    check("parse_done: None on the RESULT line",
-          o._parse_done(b"RESULT clients=4 completed=4\n") is None)
-    check("parse_done: None on a bare done= without the HEARTBEAT prefix",
-          o._parse_done(b"done=5 but not a heartbeat\n") is None)
+def test_parse_reported():
+    check("parse_reported: reads the count from a heartbeat",
+          o._parse_reported(b"HEARTBEAT reported=3 of 4\n") == 3)
+    check("parse_reported: None on the RESULT line",
+          o._parse_reported(b"RESULT clients=4 client_sent=400\n") is None)
+    check("parse_reported: None on a bare reported= without the HEARTBEAT prefix",
+          o._parse_reported(b"reported=5 but not a heartbeat\n") is None)
 
 
 def test_is_progress():
@@ -368,8 +369,9 @@ def test_watch_for_progress():
     killed = []
     o._kill_process_tree = lambda p: killed.append(p)
     try:
-        proc = _FakeProc([b"HEARTBEAT done=1 of 4\n", b"RESULT clients=4\n"],
-                         [b"debug output\n"], alive_polls=0, returncode=0)
+        proc = _FakeProc(
+            [b"HEARTBEAT reported=1 of 4\n", b"RESULT clients=4\n"],
+            [b"debug output\n"], alive_polls=0, returncode=0)
         reason, rc, out, err = o._watch_for_progress(
             proc, 3000, 300, poll=lambda: 0.0, sleep=lambda _s: None)
         check("watch: a clean run has no kill reason", reason is None)
@@ -383,11 +385,11 @@ def test_watch_for_progress():
         def advance(_s):
             clock[0] += 1000.0
 
-        frozen = [b"HEARTBEAT done=2 of 4\n"] * 4
+        frozen = [b"HEARTBEAT reported=2 of 4\n"] * 4
         proc2 = _FakeProc(frozen, [], alive_polls=1000, returncode=0)
         reason2, rc2, _o, _e = o._watch_for_progress(
             proc2, 3000, 300, poll=lambda: clock[0], sleep=advance)
-        check("watch: heartbeats with a frozen done count is a hang",
+        check("watch: heartbeats with a frozen reported count is a hang",
               reason2 == "no_progress")
         check("watch: a hung run is killed", len(killed) == 1)
         check("watch: a hung run has no returncode", rc2 is None)
@@ -397,7 +399,7 @@ def advance(_s):
         def small_advance(_s):
             clock3[0] += 100.0
 
-        proc3 = _FakeProc([b"HEARTBEAT done=1 of 4\n"], [], alive_polls=2,
+        proc3 = _FakeProc([b"HEARTBEAT reported=1 of 4\n"], [], alive_polls=2,
                           returncode=0)
         reason3, rc3, _o3, _e3 = o._watch_for_progress(
             proc3, 3000, 300, poll=lambda: clock3[0], sleep=small_advance)
@@ -498,7 +500,7 @@ def main():
                test_resolve_seeds, test_validate_args,
                test_build_argv, test_parse_result, test_lldb_argv,
                test_lldb_exit_code, test_watchdog_kill_reason,
-               test_parse_done, test_is_progress, test_classify_outcome,
+               test_parse_reported, test_is_progress, test_classify_outcome,
                test_is_failure, test_watch_for_progress, test_run_once,
                test_run_under_lldb, test_rlimit_as_supported):
         fn()