#!/usr/bin/env bash
# E1: does the ka low-concurrency latency floor track idle-scheduler count?
#
# Mechanism under test (smarm): one shared level-triggered wake pipe; every
# completion byte wakes ALL idle schedulers; one drain-lock winner; losers
# stampede the timers/io/queue mutexes and re-sleep; `enqueue` is silent.
# Prediction if right: at c <= 8, ka latency tail shrinks and throughput
# rises as scheduler count drops (fewer idle pollers -> smaller herd);
# close mode (control) stays flat or worsens as threads drop.
#
# Sweeps URUS_SCHED_THREADS x CONNS over the plain ka/close matrix,
# PLAIN_ONLY — causal cells are irrelevant to E1. wrk side is byte-identical
# to the 96b40ad5 conns sweep (THREADS=4, --latency) for comparability.
#
# Knobs: THREADS_SET CONNS_SET DUR REPS OUT (+ everything the matrix takes)
set -euo pipefail
cd "$(dirname "$0")/.."

THREADS_SET="${THREADS_SET:-1 2 4 8}"
CONNS_SET="${CONNS_SET:-4 8}"
DUR="${DUR:-15}"
REPS="${REPS:-2}"
OUT="${OUT:-/workspace/results}"

mkdir -p "$OUT"
say() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$OUT/e1.log"; }

for t in $THREADS_SET; do
  for c in $CONNS_SET; do
    cell="$OUT/t${t}-c${c}"
    say "=== E1 cell: sched_threads=$t conns=$c -> $cell ==="
    URUS_SCHED_THREADS="$t" PLAIN_ONLY=1 \
      DUR="$DUR" REPS="$REPS" CONNS="$c" OUT="$cell" \
      bash scripts/ka-close-matrix.sh
    # Cell audit: the knob must have actually reached the server. A cell
    # whose server silently ran at default threads poisons the sweep — the
    # exact failure class per-cell verification exists to catch.
    for m in ka close; do
      if ! grep -q "scheduler_threads=Some($t)" "$cell/plain-$m/server.log"; then
        say "FATAL: t=$t c=$c mode=$m server.log lacks scheduler_threads=Some($t)"
        exit 1
      fi
    done
    say "cell t=$t c=$c audit OK"
  done
done

say "=== E1 SUMMARY ==="
python3 - "$OUT" <<'PY' | tee -a "$OUT/e1.log"
import re, sys, pathlib, statistics

out = pathlib.Path(sys.argv[1])

def ms(tok):
    m = re.match(r"([\d.]+)(us|ms|s)$", tok)
    if not m:
        return None
    v = float(m.group(1))
    return {"us": v / 1000, "ms": v, "s": v * 1000}[m.group(2)]

def cell_stats(d):
    reps, pcts = [], {}
    for rep in sorted(d.glob("rep*.txt")):
        t = rep.read_text()
        m = re.search(r"Requests/sec:\s+([\d.]+)", t)
        if m:
            reps.append(float(m.group(1)))
        for p, tok in re.findall(r"^\s+(50|75|90|99)%\s+(\S+)$", t, re.M):
            pcts.setdefault(p, []).append(ms(tok))
    if not reps:
        return None
    return (statistics.mean(reps),
            {p: statistics.mean([v for v in vs if v is not None])
             for p, vs in pcts.items()})

cells = sorted(out.glob("t*-c*"))
hdr = f"{'cell':>10} {'mode':>6} {'req/s':>9} {'p50ms':>7} {'p75ms':>7} {'p90ms':>7} {'p99ms':>7}"
print(hdr); print("-" * len(hdr))
for cell in cells:
    for mode in ("ka", "close"):
        s = cell_stats(cell / f"plain-{mode}")
        if s is None:
            continue
        rps, p = s
        print(f"{cell.name:>10} {mode:>6} {rps:>9.0f} "
              f"{p.get('50', float('nan')):>7.3f} {p.get('75', float('nan')):>7.3f} "
              f"{p.get('90', float('nan')):>7.3f} {p.get('99', float('nan')):>7.3f}")
PY
say "=== E1 DONE ==="