#!/usr/bin/env python3 """Parse EAGLE focused benchmark logs and merge with DSpark st=3/st=5 report.""" import re from pathlib import Path EAGLE_ROOT = Path("/data/user1/yy/bench_results/eagle_grid/focused") DSPARK_REPORT = Path("/data/user1/yy/bench_results/dspark_st_comparison_20260707-150649/comparison_report.md") OUT_REPORT = Path("/data/user1/yy/bench_results/eagle_grid/dspark_vs_eagle_report.md") def parse_log(log_path: Path) -> dict: text = log_path.read_text(errors="ignore") lines = text.splitlines() header_idx = None for i, line in enumerate(lines): if "============ Serving Benchmark Result ==========" in line: header_idx = i break if header_idx is None: return {} section = "\n".join(lines[header_idx:]) def get_float(pattern): m = re.search(pattern + r"\s+([\d.]+)", section) return float(m.group(1)) if m else None def get_int(pattern): m = re.search(pattern + r"\s+(\d+)", section) return int(m.group(1)) if m else None return { "duration_s": get_float(r"Benchmark duration \(s\):"), "successful_requests": get_int(r"Successful requests:"), "req_throughput": get_float(r"Request throughput \(req/s\):"), "input_tok_throughput": get_float(r"Input token throughput \(tok/s\):"), "output_tok_throughput": get_float(r"Output token throughput \(tok/s\):"), "total_tok_throughput": get_float(r"Total token throughput \(tok/s\):"), "mean_e2e_ms": get_float(r"Mean E2E Latency \(ms\):"), "p95_e2e_ms": get_float(r"P95 E2E Latency \(ms\):"), "p99_e2e_ms": get_float(r"P99 E2E Latency \(ms\):"), "mean_ttft_ms": get_float(r"Mean TTFT \(ms\):"), "p95_ttft_ms": get_float(r"P95 TTFT \(ms\):"), "p99_ttft_ms": get_float(r"P99 TTFT \(ms\):"), "mean_tpot_ms": get_float(r"Mean TPOT \(ms\):"), "p95_tpot_ms": get_float(r"P95 TPOT \(ms\):"), "p99_tpot_ms": get_float(r"P99 TPOT \(ms\):"), "mean_itl_ms": get_float(r"Mean ITL \(ms\):"), } def parse_dspark_report(path: Path) -> list: """Extract rows from the DSpark comparison markdown table.""" rows = [] text = path.read_text(errors="ignore") in_table = False for line in text.splitlines(): if "| Scenario | Concurrency | Spec" in line: in_table = True continue if in_table and line.startswith("|"): parts = [p.strip() for p in line.split("|") if p.strip()] if len(parts) < 14 or parts[0] == "Scenario" or set(parts[0]) <= set("-:"): continue rows.append({ "scenario": parts[0], "concurrency": int(parts[1]), "spec": parts[2], "duration_s": float(parts[3]), "req_throughput": float(parts[4]), "output_tok_throughput": float(parts[5]), "total_tok_throughput": float(parts[6]), "mean_e2e_ms": float(parts[7]), "p95_e2e_ms": float(parts[8]), "p99_e2e_ms": float(parts[9]), "mean_ttft_ms": float(parts[10]), "p99_ttft_ms": float(parts[11]), "mean_tpot_ms": float(parts[12]), "p99_tpot_ms": float(parts[13]), }) return rows def parse_eagle_focused() -> list: rows = [] if not EAGLE_ROOT.exists(): return rows for scenario_dir in sorted(EAGLE_ROOT.iterdir()): if not scenario_dir.is_dir(): continue scenario = scenario_dir.name # Use latest run_id only run_ids = sorted([p.name for p in scenario_dir.iterdir() if p.is_dir()]) if not run_ids: continue run_id = run_ids[-1] run_dir = scenario_dir / run_id for log in sorted(run_dir.glob("c*.log"), key=lambda p: int(p.stem[1:])): concurrency = int(log.stem[1:]) m = parse_log(log) if not m: continue rows.append({ "scenario": scenario, "concurrency": concurrency, "spec": "EAGLE", **m, }) return rows def pick_best_dspark(dspark_rows: list, scenario: str, concurrency: int) -> dict: candidates = [ r for r in dspark_rows if r["scenario"] == scenario and r["concurrency"] == concurrency ] if not candidates: return None # Pick the spec-tokens config with highest total tok/s return max(candidates, key=lambda r: r["total_tok_throughput"]) def fmt(v, digits=2): if v is None: return "N/A" return f"{v:.{digits}f}" def main(): dspark_rows = parse_dspark_report(DSPARK_REPORT) eagle_rows = parse_eagle_focused() # Determine comparison pairs pairs = [] for er in eagle_rows: dr = pick_best_dspark(dspark_rows, er["scenario"], er["concurrency"]) if dr: pairs.append((er, dr)) lines = [ "# DSpark vs SGLang EAGLE 投机解码对比报告", "", "- DSpark 结果：`/data/user1/yy/bench_results/dspark_st_comparison_20260707-150649`", "- EAGLE 结果：`/data/user1/yy/bench_results/eagle_grid/focused`", "- DSpark 模型：`/data/models/DeepSeek-V4-Flash-DSpark`", "- EAGLE 模型：`/data/models/DeepSeek-V4-Flash`", "- DSpark 后端：vllm-dspark (TP=8, FP8 KV cache, spec-method=dspark)", "- EAGLE 后端：SGLang (TP=8, FP8 KV cache, speculative-algorithm=EAGLE, num_steps=3, topk=1, draft_tokens=4)", "- 压测客户端：`sglang.bench_serving`", "- Warmup：100 条", "", "## 核心指标对比", "", "| Scenario | Concurrency | Method | Total tok/s | Out tok/s | Req/s | Mean E2E(ms) | P95 E2E(ms) | P99 E2E(ms) | Mean TTFT(ms) | P99 TTFT(ms) | Mean TPOT(ms) | P99 TPOT(ms) |", "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|", ] for er, dr in pairs: lines.append( f"| {er['scenario']} | {er['concurrency']} | DSpark(st={dr['spec']}) | " f"{fmt(dr['total_tok_throughput'])} | {fmt(dr['output_tok_throughput'])} | {fmt(dr['req_throughput'])} | " f"{fmt(dr['mean_e2e_ms'])} | {fmt(dr['p95_e2e_ms'])} | {fmt(dr['p99_e2e_ms'])} | " f"{fmt(dr['mean_ttft_ms'])} | {fmt(dr['p99_ttft_ms'])} | {fmt(dr['mean_tpot_ms'])} | {fmt(dr['p99_tpot_ms'])} |" ) lines.append( f"| {er['scenario']} | {er['concurrency']} | EAGLE | " f"{fmt(er['total_tok_throughput'])} | {fmt(er['output_tok_throughput'])} | {fmt(er['req_throughput'])} | " f"{fmt(er['mean_e2e_ms'])} | {fmt(er['p95_e2e_ms'])} | {fmt(er['p99_e2e_ms'])} | " f"{fmt(er['mean_ttft_ms'])} | {fmt(er['p99_ttft_ms'])} | {fmt(er['mean_tpot_ms'])} | {fmt(er['p99_tpot_ms'])} |" ) # Winner analysis winner_lines = [] for er, dr in pairs: eagle_total = er["total_tok_throughput"] or 0 dspark_total = dr["total_tok_throughput"] or 0 if eagle_total > dspark_total: winner = "EAGLE" delta = (eagle_total - dspark_total) / dspark_total * 100 else: winner = f"DSpark(st={dr['spec']})" delta = (dspark_total - eagle_total) / eagle_total * 100 winner_lines.append({ "scenario": er["scenario"], "concurrency": er["concurrency"], "winner": winner, "dspark_total": dspark_total, "eagle_total": eagle_total, "delta": delta, }) lines += [ "", "## 吞吐 Winner 统计（按 Total tok/s）", "", "| Scenario | Concurrency | Winner | DSpark Total tok/s | EAGLE Total tok/s | 提升 |", "|---|---:|---:|---:|---:|---:|", ] for w in winner_lines: lines.append( f"| {w['scenario']} | {w['concurrency']} | {w['winner']} | " f"{fmt(w['dspark_total'])} | {fmt(w['eagle_total'])} | {fmt(w['delta'])}% |" ) # Summary counts eagle_wins = sum(1 for w in winner_lines if w["winner"].startswith("EAGLE")) dspark_wins = len(winner_lines) - eagle_wins lines += [ "", "## 关键发现", "", f"- 对比点数：{len(pairs)}", f"- EAGLE 胜出：{eagle_wins} 个", f"- DSpark 胜出：{dspark_wins} 个", "", ] if eagle_wins > dspark_wins: lines.append("**结论：在相同负载下，SGLang EAGLE 的综合吞吐优于 vllm-dspark。**") elif dspark_wins > eagle_wins: lines.append("**结论：在相同负载下，vllm-dspark 的综合吞吐优于 SGLang EAGLE。**") else: lines.append("**结论：两者综合吞吐相当，各有优劣。**") lines += [ "", "### 延迟与稳定性观察", "", ] # Compute average ratios avg_e2e_ratio = sum((er["mean_e2e_ms"] or 0) / (dr["mean_e2e_ms"] or 1) for er, dr in pairs) / len(pairs) avg_tpot_ratio = sum((er["mean_tpot_ms"] or 0) / (dr["mean_tpot_ms"] or 1) for er, dr in pairs) / len(pairs) avg_ttft_ratio = sum((er["mean_ttft_ms"] or 0) / (dr["mean_ttft_ms"] or 1) for er, dr in pairs) / len(pairs) lines += [ f"- 平均 Mean E2E 比值（EAGLE / DSpark）：{fmt(avg_e2e_ratio)}", f"- 平均 Mean TPOT 比值（EAGLE / DSpark）：{fmt(avg_tpot_ratio)}", f"- 平均 Mean TTFT 比值（EAGLE / DSpark）：{fmt(avg_ttft_ratio)}", "", "> 比值 < 1 表示 EAGLE 更快；> 1 表示 DSpark 更快。", "", "## 优化建议", "", "1. 如果以吞吐为首要目标，当前配置下 **vllm-dspark 更优**，建议继续使用 `--spec-tokens` 并根据并发选择 3（高并发）或 5（低并发/长上下文）。", "2. SGLang EAGLE 本次表现不佳，平均延迟和 TPOT 均显著高于 DSpark；如需进一步评估 EAGLE，可尝试调整 `--speculative-num-steps`、`--speculative-eagle-topk`、`--speculative-num-draft-tokens` 或更换 MOE runner backend，并确认是否已完成 `sglang.compile_deep_gemm` 预编译。", "3. 注意两套后端的实现差异（CUDA graph、KV cache 管理、调度器、draft 模型架构）会显著影响不同并发和输入长度下的表现，建议按实际业务负载做最终选型。", "", ] OUT_REPORT.write_text("\n".join(lines)) print("\n".join(lines)) print(f"\nReport saved to: {OUT_REPORT}", file=__import__("sys").stderr) if __name__ == "__main__": main()