Quantong Qiu db00c1b1ac Add custom benchmark setup for vLLM TP=2 on dsv4_h200
- Create configuration file for dsv4_h200_vllm_tp2_custom_bench.
- Implement result parsing script to handle JSONL outputs from bench_client.py.
- Develop run_bench.sh script to orchestrate multi-service vLLM benchmarking.
- Add start_server.sh script to launch independent vLLM services on specified GPU pairs.
2026-07-10 09:06:10 +00:00

207 lines
7.8 KiB
Python
Executable File

#!/usr/bin/env python3
"""Parse custom bench_client.py JSONL outputs for vLLM TP=2 benchmark.
Reads JSONL files produced by bench_client.py (one request per line,
last line is the summary) and generates:
- results.json (appended scenarios, following the standard schema)
- report.md (human-readable summary)
Usage:
python3 parse_results.py <result_root>
"""
import json
import sys
from pathlib import Path
def parse_jsonl(path: Path) -> tuple[dict | None, list[dict]]:
"""Read the summary JSON (last line) and all per-request lines."""
requests: list[dict] = []
summary: dict | None = None
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
continue
# The summary line has "completed" and "duration" keys.
if "completed" in obj and "duration" in obj:
summary = obj
else:
requests.append(obj)
return summary, requests
def compute_metrics(summary: dict, requests: list[dict]) -> dict:
completed = summary.get("completed", 0)
total = len(requests)
failed = total - completed if total > 0 else 0
duration_s = summary.get("duration", 0.0)
# Extract per-request metrics for percentiles.
ttfts = [r["first_token_time_ms"] for r in requests if r.get("success")]
tpots = [r["tpot_ms"] for r in requests if r.get("success") and r.get("output_tokens", 0) > 1]
e2es = [r["e2e_latency_ms"] for r in requests if r.get("success")]
itls = [r["mean_itl_ms"] for r in requests if r.get("success") and r.get("inter_token_latencies")]
def percentile(values: list[float], p: float) -> float:
if not values:
return 0.0
s = sorted(values)
k = (len(s) - 1) * p / 100.0
f = int(k)
c = min(f + 1, len(s) - 1)
return s[f] + (k - f) * (s[c] - s[f])
return {
"success": completed,
"failed": failed,
"duration_s": duration_s,
"request_throughput": summary.get("request_throughput", 0.0),
"input_token_throughput": summary.get("input_throughput", 0.0),
"output_token_throughput": summary.get("output_throughput", 0.0),
"total_token_throughput": summary.get("total_throughput", 0.0),
"total_input_tokens": summary.get("total_input_tokens", 0),
"total_output_tokens": summary.get("total_output_tokens", 0),
"e2e_ms": {
"mean": summary.get("mean_e2e_latency_ms", 0.0),
"p50": percentile(e2es, 50),
"p90": percentile(e2es, 90),
"p95": percentile(e2es, 95),
"p99": percentile(e2es, 99),
},
"ttft_ms": {
"mean": summary.get("mean_ttft_ms", 0.0),
"p50": percentile(ttfts, 50),
"p90": percentile(ttfts, 90),
"p95": percentile(ttfts, 95),
"p99": percentile(ttfts, 99),
},
"tpot_ms": {
"mean": summary.get("mean_tpot_ms", 0.0),
"p50": percentile(tpots, 50),
"p90": percentile(tpots, 90),
"p95": percentile(tpots, 95),
"p99": percentile(tpots, 99),
},
"itl_ms": {
"mean": summary.get("mean_itl_ms", 0.0),
"p50": percentile(itls, 50),
"p90": percentile(itls, 90),
"p95": percentile(itls, 95),
"p99": percentile(itls, 99),
},
}
def scenario_name(concurrency: int, input_len: int, output_len: int) -> str:
return f"c{concurrency}_i{input_len}_o{output_len}"
def slo_status(metrics: dict, ttft_limit_ms: float = 3000.0, tpot_limit_ms: float = 50.0) -> dict:
ttft_ok = metrics["ttft_ms"]["p95"] < ttft_limit_ms
tpot_ok = metrics["tpot_ms"]["mean"] < tpot_limit_ms
if ttft_ok and tpot_ok:
mark = ""
elif ttft_ok or tpot_ok:
mark = "⚠️"
else:
mark = ""
return {
"ttft_p95_ok": ttft_ok,
"tpot_mean_ok": tpot_ok,
"overall": mark,
}
def append_scenario(results_json: Path, scenario: dict) -> None:
with open(results_json, "r", encoding="utf-8") as f:
data = json.load(f)
data["scenarios"].append(scenario)
with open(results_json, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def generate_report(result_root: Path, scenarios: list[dict]) -> None:
report_path = result_root / "report.md"
with open(report_path, "w", encoding="utf-8") as f:
f.write("# H200 vLLM TP=2 Custom Benchmark Report\n\n")
f.write("- **Client**: `bench_client.py` (async OpenAI API, per-request timing)\n")
f.write("- **Backend**: vLLM (TP=2, FP8 KV cache, no speculative decoding)\n")
f.write(f"- **Result root**: `{result_root}`\n\n")
f.write("## Results\n\n")
f.write("| Scenario | Concurrency | Input | Output | Duration(s) | Success | Failed | Req/s | In tok/s | Out tok/s | Total tok/s | Mean TTFT(ms) | P95 TTFT(ms) | P99 TTFT(ms) | Mean TPOT(ms) | P95 TPOT(ms) | P99 TPOT(ms) | Mean E2E(ms) | P95 E2E(ms) | P99 E2E(ms) | SLO |\n")
f.write("|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|\n")
for s in scenarios:
cfg = s["config"]
m = s["metrics"]
slo = s.get("slo_status", {}).get("overall", "")
f.write(
f"| {s['name']} | {cfg['concurrency']} | {cfg['input_len']} | {cfg['output_len']} | "
f"{m['duration_s']:.2f} | {m['success']} | {m['failed']} | {m['request_throughput']:.2f} | "
f"{m['input_token_throughput']:.2f} | {m['output_token_throughput']:.2f} | "
f"{m['total_token_throughput']:.2f} | "
f"{m['ttft_ms']['mean']:.2f} | {m['ttft_ms']['p95']:.2f} | {m['ttft_ms']['p99']:.2f} | "
f"{m['tpot_ms']['mean']:.2f} | {m['tpot_ms']['p95']:.2f} | {m['tpot_ms']['p99']:.2f} | "
f"{m['e2e_ms']['mean']:.2f} | {m['e2e_ms']['p95']:.2f} | {m['e2e_ms']['p99']:.2f} | {slo} |\n"
)
f.write("\n")
f.write("SLO: S2 tier — TTFT P95 < 3000ms, TPOT mean < 50ms. ✅ pass, ⚠️ partial, ❌ fail.\n\n")
def main() -> None:
result_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("results")
raw_dir = result_root / "raw_outputs"
results_json = result_root / "results.json"
if not raw_dir.exists():
raise SystemExit(f"raw_outputs directory not found: {raw_dir}")
scenarios = []
for jsonl_path in sorted(raw_dir.glob("vllm_*.jsonl")):
parts = jsonl_path.stem.split("_")
if len(parts) < 5:
continue
concurrency, input_len, output_len = int(parts[2]), int(parts[3]), int(parts[4])
summary, requests = parse_jsonl(jsonl_path)
if summary is None:
continue
metrics = compute_metrics(summary, requests)
scenario = {
"name": scenario_name(concurrency, input_len, output_len),
"config": {
"concurrency": concurrency,
"input_len": input_len,
"output_len": output_len,
"dataset": "random",
"num_prompts": metrics["success"] + metrics["failed"],
},
"metrics": metrics,
"slo_status": slo_status(metrics),
"raw_file": str(jsonl_path),
}
scenarios.append(scenario)
if not scenarios:
print("No benchmark outputs found to parse")
return
if results_json.exists():
for s in scenarios:
append_scenario(results_json, s)
generate_report(result_root, scenarios)
print(f"Parsed {len(scenarios)} scenarios into {result_root}/report.md")
if __name__ == "__main__":
main()