sskj/experiments/dsv4_h200_vllm/parse_results.py

#!/usr/bin/env python3
"""Parse H200 vLLM baseline benchmark JSONL outputs.

Reads raw JSONL files produced by `sglang.bench_serving --output-file --output-details`
and generates:
    - results.json  (appended scenarios)
    - report.md     (human-readable summary)

Usage:
    python3 parse_results.py <result_root>
"""

import json
import math
import os
import sys
from pathlib import Path


def percentile(values: list[float], p: float) -> float:
    if not values:
        return 0.0
    values = sorted(values)
    if len(values) == 1:
        return values[0]
    k = (len(values) - 1) * (p / 100.0)
    f = math.floor(k)
    c = math.ceil(k)
    if f == c:
        return values[int(k)]
    return values[f] * (c - k) + values[c] * (k - f)


def parse_jsonl(path: Path) -> list[dict]:
    requests = []
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                requests.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return requests


def compute_metrics(requests: list[dict]) -> dict:
    success_reqs = [r for r in requests if r.get("success", True)]
    failed = len(requests) - len(success_reqs)

    if not success_reqs:
        return {"success": 0, "failed": failed}

    # Time boundaries.
    start_times = [r["tstamp_start"] for r in success_reqs]
    end_times = [r["tstamp_finished"] for r in success_reqs]
    duration_s = max(end_times) - min(start_times)

    # Token counts.
    input_tokens = [r.get("prompt_tokens", 0) for r in success_reqs]
    output_tokens = [r.get("completion_tokens", 0) for r in success_reqs]
    total_input = sum(input_tokens)
    total_output = sum(output_tokens)

    # Latencies (ms).
    e2e = [r.get("e2e_latency", 0) * 1000 for r in success_reqs]
    ttft = [r.get("ttft", 0) * 1000 for r in success_reqs]
    itls = []
    for r in success_reqs:
        itls.extend(r.get("itl", []))
    # TPOT from itl averages per request.
    tpots = []
    for r in success_reqs:
        req_itls = r.get("itl", [])
        if req_itls:
            tpots.append(sum(req_itls) / len(req_itls) * 1000)

    def latency_stats(values: list[float]) -> dict:
        return {
            "mean": sum(values) / len(values),
            "p50": percentile(values, 50),
            "p90": percentile(values, 90),
            "p95": percentile(values, 95),
            "p99": percentile(values, 99),
        }

    return {
        "success": len(success_reqs),
        "failed": failed,
        "duration_s": duration_s,
        "request_throughput": len(success_reqs) / duration_s if duration_s > 0 else 0.0,
        "input_token_throughput": total_input / duration_s if duration_s > 0 else 0.0,
        "output_token_throughput": total_output / duration_s if duration_s > 0 else 0.0,
        "total_token_throughput": (total_input + total_output) / duration_s if duration_s > 0 else 0.0,
        "total_input_tokens": total_input,
        "total_output_tokens": total_output,
        "e2e_ms": latency_stats(e2e),
        "ttft_ms": latency_stats(ttft),
        "tpot_ms": latency_stats(tpots),
        "itl_ms": latency_stats([v * 1000 for v in itls]),
    }


def scenario_name(concurrency: int, input_len: int, output_len: int) -> str:
    return f"c{concurrency}_i{input_len}_o{output_len}"


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 Baseline Benchmark Report\n\n")
        f.write(f"- Result root: `{result_root}`\n")
        f.write("- Model: `/data/models/DeepSeek-V4-Flash`\n")
        f.write("- Backend: vLLM (TP=8, FP8 KV cache, no speculative decoding)\n")
        f.write("- Benchmark client: `sglang.bench_serving --backend vllm`\n\n")

        f.write("## Results\n\n")
        f.write("| Scenario | Concurrency | Input | Output | Duration(s) | Success | 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) |\n")
        f.write("|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|\n")

        for s in scenarios:
            cfg = s["config"]
            m = s["metrics"]
            f.write(
                f"| {cfg['name']} | {cfg['concurrency']} | {cfg['input_len']} | {cfg['output_len']} | "
                f"{m['duration_s']:.2f} | {m['success']} | {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} |\n"
            )
        f.write("\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")):
        # Filename: vllm_MMDD_concurrency_inputlen_outputlen.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])

        requests = parse_jsonl(jsonl_path)
        if not requests:
            continue

        metrics = compute_metrics(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,
            "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()