File size: 10,380 Bytes
e1acc15 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | """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")
# Only B1 has a meaningful occupied/TTFT prefill proxy. It still includes startup/first-step costs.
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")
# Counters at B>1 count request-rounds, NOT GPU engine iterations.
runs.append(out)
# Full per-server snapshots only numeric values and fixed prefixes; no model label strings.
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
# Match B1 run counters by repetition; never attribute global acceptance to one concurrent stream.
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()
|