| """Extract numeric benchmark evidence without copying prompts, text or tensors. |
| |
| Usage: python -m offload.gb10_budget.evidence --results offload/results \ |
| --extra-results /path/to/coordinator/offload/results/K1 --output /campaign/evidence.json |
| Only allowlisted result directories are read. Every input is hashed; malformed JSON |
| fails with a source locator. Reported step p50s are per-request client SSE gaps, not |
| GPU timestamps. Acceptance belongs to a whole run, never an individual B4 request. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import math |
| from collections import defaultdict |
| from pathlib import Path |
| from statistics import mean |
| from typing import TypeAlias, cast |
|
|
| JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | str | int | float | bool | None |
| Object: TypeAlias = dict[str, JSON] |
| ALLOWED = ("A", "A-target-only", "B", "Bprod", "C", "D", "Dasync", |
| "Dasync-si0", "E", "E0-eager", "E1-si0", "W4-sync", "W4-p1", |
| "W4-E0", "W4-E1p", "W4-E1s", "W5", "W5-E1p", "W5-prod", "W5-drill", "K1") |
| REQUEST_FIELDS = ("target_ctx", "occupied_prompt_tokens", "concurrency", "mixed", |
| "rep", "prompt_tokens", "completion_tokens", "ttft_s", "total_s", |
| "decode_tps", "start_offset_s") |
|
|
|
|
| def obj(value: JSON) -> Object: |
| return value if isinstance(value, dict) else {} |
|
|
|
|
| def number(value: JSON) -> float | None: |
| if isinstance(value, (int, float)) and not isinstance(value, bool): |
| if not math.isfinite(value): |
| raise ValueError("Non-finite evidence number") |
| return float(value) |
| return None |
|
|
|
|
| def numeric_tree(value: JSON) -> JSON: |
| """Drop every string value, preserving numeric counter names and structure.""" |
| if isinstance(value, dict): |
| return {k: numeric_tree(v) for k, v in value.items() |
| if not isinstance(v, str) and v is not None} |
| if isinstance(value, list): |
| return [numeric_tree(v) for v in value if not isinstance(v, str)] |
| if isinstance(value, (int, float, bool)): |
| if not isinstance(value, bool): |
| number(value) |
| return value |
| return None |
|
|
|
|
| def extract(path: Path, raw_text: str) -> tuple[list[Object], list[Object], list[Object]]: |
| requests: list[Object] = [] |
| runs: list[Object] = [] |
| metrics: list[Object] = [] |
| for line_no, line in enumerate(raw_text.splitlines(), 1): |
| if not line.strip(): |
| continue |
| try: |
| row = obj(cast(JSON, json.loads(line))) |
| except ValueError as exc: |
| raise ValueError(f"{path}:{line_no}: malformed JSON") from exc |
| locator: Object = {"source": str(path), "line": line_no, "status": "MEASURED", |
| "run_id": row.get("run_id")} |
| if row.get("type") == "request": |
| out = locator | {k: row[k] for k in REQUEST_FIELDS |
| if k in row and (number(row[k]) is not None or isinstance(row[k], bool))} |
| out["cache_state"] = row.get("cache_state") if row.get("cache_state") in ("cold", "warm", "warm_repeat") else None |
| out["error_present"] = bool(row.get("error")) |
| out["step_gap_ms"] = numeric_tree(row.get("step_gap_ms")) |
| out["cached_tokens"] = obj(obj(row.get("usage")).get("prompt_tokens_details")).get("cached_tokens") |
| |
| p, t = number(row.get("prompt_tokens")), number(row.get("ttft_s")) |
| if row.get("concurrency") == 1 and p and t and not out.get("cached_tokens"): |
| out["prefill_proxy_tps"] = p / t |
| out["chunk4096_equivalent_ms"] = 4096 * t / p * 1000 |
| requests.append(out) |
| elif row.get("type") == "run": |
| spec = obj(row.get("spec")) |
| out = locator | {k: row.get(k) for k in ("target_ctx", "concurrency", "mixed", "rep")} |
| out["error_present"] = bool(row.get("error")) |
| out["accepted_per_round"] = spec.get("mean_accepted_per_round") |
| out["acceptance_per_position_per_round"] = numeric_tree(spec.get("acceptance_per_position_per_round")) |
| out["aggregate_including_prefill_tps"] = row.get("aggregate_tps") |
| out["rounds_request_sum"] = spec.get("rounds") |
| |
| runs.append(out) |
| |
| for name in ("metrics_before", "metrics_after"): |
| snapshot = obj(row.get(name)) |
| metrics.append(locator | {"phase": name, "values": { |
| k.split("{")[0]: v for k, v in snapshot.items() |
| if k.startswith(("vllm:time_per_output_token_seconds_sum", "vllm:time_per_output_token_seconds_count"))}}) |
| return requests, runs, metrics |
|
|
|
|
| def analyze(roots: list[Path]) -> Object: |
| files = sorted({p.resolve() for root in roots for p in root.rglob("*") |
| if p.is_file() and p.suffix in (".json", ".jsonl", ".txt", ".log")}) |
| manifest: list[JSON] = [] |
| requests: list[Object] = [] |
| runs: list[Object] = [] |
| stats: list[JSON] = [] |
| for path in files: |
| data = path.read_bytes() |
| manifest.append({"source": str(path), "sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data)}) |
| if path.suffix == ".jsonl": |
| req, run, _ = extract(path, data.decode()) |
| requests.extend(req) |
| runs.extend(run) |
| elif path.suffix == ".json" and "stats" in path.name: |
| stats.append({"source": str(path), "status": "MEASURED", "numeric": numeric_tree(cast(JSON, json.loads(data)))}) |
| groups: dict[tuple[str, int, int, str, bool], list[Object]] = defaultdict(list) |
| for row in requests: |
| key = (str(row["source"]), int(cast(int, row.get("target_ctx", 0))), |
| int(cast(int, row.get("concurrency", 0))), str(row.get("cache_state")), bool(row.get("mixed"))) |
| groups[key].append(row) |
| summaries: list[JSON] = [] |
| for (source, ctx, batch, cache, mixed), rows in sorted(groups.items()): |
| summary: Object = {"source": source, "context": ctx, "batch": batch, "cache": cache, |
| "mixed": mixed, "n_requests": len(rows), "status": "MEASURED"} |
| for name in ("ttft_s", "decode_tps", "total_s", "prefill_proxy_tps", "chunk4096_equivalent_ms"): |
| values = [v for row in rows if (v := number(row.get(name))) is not None] |
| summary[name] = {"mean": mean(values), "min": min(values), "max": max(values)} if values else None |
| values = [v for row in rows if (v := number(obj(row.get("step_gap_ms")).get("p50"))) is not None] |
| summary["request_step_p50_ms"] = {"mean": mean(values), "min": min(values), "max": max(values)} if values else None |
| |
| estimates = [] |
| if batch == 1 and not mixed: |
| for row in rows: |
| match = [r for r in runs if r["source"] == source and r.get("run_id") == row.get("run_id")] |
| if len(match) == 1: |
| a, tps = number(match[0].get("accepted_per_round")), number(row.get("decode_tps")) |
| if a is not None and tps: |
| estimates.append(1000 * (1 + a) / tps) |
| summary["acceptance_normalized_round_ms_ESTIMATED"] = mean(estimates) if estimates else None |
| summaries.append(summary) |
| return {"method": "Allowlisted aggregates only; SSE timings and wall-time proxies, not kernel profiles.", |
| "inputs": manifest, "requests": list[JSON](requests), "runs": list[JSON](runs), |
| "stats": stats, "summaries": summaries} |
|
|
|
|
| def markdown(result: Object) -> str: |
| lines = ["# SP1 complete numeric evidence reconciliation", "", |
| "All timing values: MEASURED saved-client observations; norm-ms: ESTIMATED acceptance-normalized B1 mean.", |
| "Each row is one source/context/batch/cache/mixed group. p50 column is the mean of request medians.", |
| "Do not pool medians as engine steps or count concurrent requests as independent repetitions.", "", |
| "| Source | ctx | B | cache | mixed | n requests | TTFT s | SSE p50 ms | norm ms | decode tok/s |", |
| "|---|---:|---:|---|---|---:|---:|---:|---:|---:|"] |
| summaries = result["summaries"] |
| assert isinstance(summaries, list) |
| for raw in summaries: |
| row = obj(raw) |
| def avg(key: str) -> str: |
| value = number(obj(row.get(key)).get("mean")) |
| return f"{value:.3f}" if value is not None else "—" |
| norm = number(row.get("acceptance_normalized_round_ms_ESTIMATED")) |
| norm_cell = f"{norm:.3f}" if norm is not None else "—" |
| lines.append(f"| {row['source']} | {row['context']} | {row['batch']} | {row['cache']} | " |
| f"{row['mixed']} | {row['n_requests']} | {avg('ttft_s')} | " |
| f"{avg('request_step_p50_ms')} | {norm_cell} | {avg('decode_tps')} |") |
| return "\n".join(lines) + "\n" |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--results", required=True, type=Path) |
| parser.add_argument("--extra-results", action="append", default=[], type=Path) |
| parser.add_argument("--output", required=True, type=Path) |
| args = parser.parse_args() |
| if not args.results.is_dir(): |
| parser.error(f"Missing result root: {args.results}") |
| roots = [args.results / name for name in ALLOWED if (args.results / name).is_dir()] |
| for root in args.extra_results: |
| if not root.is_dir() or root.name not in ALLOWED: |
| parser.error(f"Missing or non-allowlisted extra results: {root}") |
| roots.append(root) |
| result = analyze(roots) |
| args.output.parent.mkdir(parents=True, exist_ok=True) |
| args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n") |
| args.output.with_suffix(".md").write_text(markdown(result)) |
| print(f"Wrote {args.output}: {len(cast(list[JSON], result['inputs']))} hashed files, " |
| f"{len(cast(list[JSON], result['requests']))} request aggregates") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
