fix: replace 5 buggy parse_results.py (raw_requests bloat) with shared parse_backend.py wrapper

This commit is contained in:
yy-fighting 2026-07-22 03:47:22 +00:00
parent 489c5a5e61
commit 3b0297516e
5 changed files with 86 additions and 1225 deletions

View File

@ -1,312 +1,22 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Parse P800 SGLang benchmark outputs and produce results.json + report.md. """Thin wrapper: delegate to the shared parser at scripts/common/parse_backend.py.
Usage: Replaces per-experiment parse_results.py copies that embedded raw_requests
python3 parse_results.py <result_root> (the results.json bloat bug). Forwards all CLI args to the shared parser.
""" """
import json
import os import os
import re import subprocess
import sys import sys
from pathlib import Path
from collections import OrderedDict
METRIC_PATTERNS = OrderedDict( _HERE = os.path.dirname(os.path.abspath(__file__))
[ _d = _HERE
("successful_requests", [r"Successful requests:\s+(\d+)"]), while True:
("benchmark_duration_s", [r"Benchmark duration \(s\):\s+([\d.]+)"]), cand = os.path.join(_d, "scripts", "common", "parse_backend.py")
("request_throughput", [r"Request throughput \(req/s\):\s+([\d.]+)"]), if os.path.isfile(cand):
("input_token_throughput", [r"Input token throughput \(tok/s\):\s+([\d.]+)"]), sys.exit(subprocess.call([sys.executable, cand, *sys.argv[1:]]))
("output_token_throughput", [r"Output token throughput \(tok/s\):\s+([\d.]+)"]), parent = os.path.dirname(_d)
("total_token_throughput", [r"Total token throughput \(tok/s\):\s+([\d.]+)"]), if parent == _d:
("ttft_mean", [r"Mean TTFT \(ms\):\s+([\d.]+)"]), break
("ttft_p50", [r"Median TTFT \(ms\):\s+([\d.]+)", r"P50 TTFT \(ms\):\s+([\d.]+)"]), _d = parent
("ttft_p90", [r"P90 TTFT \(ms\):\s+([\d.]+)"]), print("ERROR: scripts/common/parse_backend.py not found upwards from " + _HERE, file=sys.stderr)
("ttft_p99", [r"P99 TTFT \(ms\):\s+([\d.]+)"]), sys.exit(1)
("tpot_mean", [r"Mean TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p50", [r"Median TPOT \(ms\):\s+([\d.]+)", r"P50 TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p90", [r"P90 TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p99", [r"P99 TPOT \(ms\):\s+([\d.]+)"]),
("e2e_mean", [r"Mean E2E \(ms\):\s+([\d.]+)"]),
("e2e_p50", [r"Median E2E \(ms\):\s+([\d.]+)", r"P50 E2E \(ms\):\s+([\d.]+)"]),
("e2e_p90", [r"P90 E2E \(ms\):\s+([\d.]+)"]),
("e2e_p99", [r"P99 E2E \(ms\):\s+([\d.]+)"]),
]
)
def find_float(text: str, patterns: list[str]) -> float | None:
for pat in patterns:
m = re.search(pat, text)
if m:
try:
return float(m.group(1))
except ValueError:
return None
return None
def parse_summary_log(log_text: str) -> dict:
result = {}
for key, patterns in METRIC_PATTERNS.items():
result[key] = find_float(log_text, patterns)
return result
def percentile(values: list[float], p: float) -> float:
if not values:
return 0.0
sorted_values = sorted(values)
k = (len(sorted_values) - 1) * p / 100.0
f = int(k)
c = min(f + 1, len(sorted_values) - 1)
if f == c:
return sorted_values[f]
return sorted_values[f] * (c - k) + sorted_values[c] * (k - f)
def parse_jsonl(jsonl_path: Path) -> tuple[list[dict], dict]:
"""Parse sglang.bench_serving --output-file output.
Newer sglang writes one aggregate JSON object. Older versions write one
JSON object per request. Returns (raw_requests, aggregates).
"""
text = jsonl_path.read_text(encoding="utf-8", errors="replace").strip()
if not text:
return [], {}
# Try single aggregate JSON object first.
try:
data = json.loads(text)
if isinstance(data, dict) and "mean_e2e_latency_ms" in data:
aggregates = {
"success": data.get("total_output_tokens", 0) > 0 and 1 or 0,
"failed": 0,
"input_tokens": data.get("total_input_tokens", 0),
"output_tokens": data.get("total_output_tokens", 0),
"latencies": {
"e2e_ms": {
"mean": data.get("mean_e2e_latency_ms"),
"p50": data.get("median_e2e_latency_ms"),
"p90": data.get("p90_e2e_latency_ms"),
"p95": None,
"p99": data.get("p99_e2e_latency_ms"),
},
"ttft_ms": {
"mean": data.get("mean_ttft_ms"),
"p50": data.get("median_ttft_ms"),
"p90": None,
"p95": None,
"p99": data.get("p99_ttft_ms"),
},
"tpot_ms": {
"mean": data.get("mean_tpot_ms"),
"p50": data.get("median_tpot_ms"),
"p90": None,
"p95": None,
"p99": data.get("p99_tpot_ms"),
},
"itl_ms": {
"mean": data.get("mean_itl_ms"),
"p50": data.get("median_itl_ms"),
"p90": None,
"p95": data.get("p95_itl_ms"),
"p99": data.get("p99_itl_ms"),
},
},
}
return [data], aggregates
except json.JSONDecodeError:
pass
# Fall back to JSONL per-request parsing.
raw_requests = []
ttfts = []
tpots = []
itls = []
e2es = []
input_tokens = []
output_tokens = []
success = 0
failed = 0
for line in text.splitlines():
line = line.strip()
if not line:
continue
try:
req = json.loads(line)
except json.JSONDecodeError:
continue
raw_requests.append(req)
ttft = req.get("ttft") or req.get("ttft_ms") or 0
tpot = req.get("tpot") or req.get("tpot_ms") or 0
itl = req.get("inter_token_latency") or req.get("itl") or req.get("itl_ms") or 0
e2e = req.get("e2e_latency") or req.get("e2e") or req.get("e2e_ms") or 0
in_tok = req.get("input_tokens") or req.get("prompt_tokens") or 0
out_tok = req.get("output_tokens") or req.get("completion_tokens") or 0
if ttft:
ttfts.append(float(ttft))
if tpot:
tpots.append(float(tpot))
if itl:
itls.append(float(itl))
if e2e:
e2es.append(float(e2e))
if in_tok:
input_tokens.append(int(in_tok))
if out_tok:
output_tokens.append(int(out_tok))
if req.get("success", True):
success += 1
else:
failed += 1
def latency_bucket(values: list[float]) -> dict:
if not values:
return {"mean": None, "p50": None, "p90": None, "p95": None, "p99": None}
return {
"mean": round(sum(values) / len(values), 2),
"p50": round(percentile(values, 50), 2),
"p90": round(percentile(values, 90), 2),
"p95": round(percentile(values, 95), 2),
"p99": round(percentile(values, 99), 2),
}
aggregates = {
"success": success,
"failed": failed,
"input_tokens": sum(input_tokens),
"output_tokens": sum(output_tokens),
"latencies": {
"e2e_ms": latency_bucket(e2es),
"ttft_ms": latency_bucket(ttfts),
"tpot_ms": latency_bucket(tpots),
"itl_ms": latency_bucket(itls),
},
}
return raw_requests, aggregates
def parse_scenario(result_root: Path, raw_file: Path) -> dict | None:
"""Parse one raw output file into a scenario dict."""
# Filename: {chip}_{engine}_{MMDD}_{concurrency}_{input_len}_{output_len}.jsonl
parts = raw_file.stem.split("_")
if len(parts) < 6:
return None
try:
concurrency = int(parts[-3])
input_len = int(parts[-2])
output_len = int(parts[-1])
except ValueError:
return None
log_file = result_root / "logs" / f"sglang_c{concurrency}_i{input_len}_o{output_len}.log"
summary = {}
if log_file.exists():
summary = parse_summary_log(log_file.read_text(encoding="utf-8", errors="replace"))
_, aggregates = parse_jsonl(raw_file)
# For newer sglang aggregate JSON, success/failed are not present in the
# raw output file. Override them from the human-readable summary log when
# it is available.
if summary.get("successful_requests") is not None:
aggregates["success"] = int(summary["successful_requests"])
# The summary log only reports successes; assume failures are zero
# unless the aggregate JSON already provided a non-zero failed count.
if not aggregates.get("failed"):
aggregates["failed"] = 0
scenario = {
"name": f"c{concurrency}_i{input_len}_o{output_len}",
"concurrency": concurrency,
"input_len": input_len,
"output_len": output_len,
"success": aggregates["success"],
"failed": aggregates["failed"],
"duration_s": summary.get("benchmark_duration_s"),
"request_throughput": summary.get("request_throughput"),
"input_token_throughput": summary.get("input_token_throughput"),
"output_token_throughput": summary.get("output_token_throughput"),
"total_token_throughput": summary.get("total_token_throughput"),
"accept_length": None,
"latencies": aggregates["latencies"],
}
return scenario
def main() -> None:
result_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("results")
raw_dir = result_root / "raw_outputs"
json_path = result_root / "results.json"
report_path = result_root / "report.md"
if not json_path.exists():
raise SystemExit(f"metadata results.json not found: {json_path}")
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
scenarios = []
if raw_dir.exists():
for raw_file in sorted(raw_dir.glob("*.jsonl")):
scenario = parse_scenario(result_root, raw_file)
if scenario:
scenarios.append(scenario)
data["scenarios"] = scenarios
with open(json_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
# Generate report.md
with open(report_path, "w", encoding="utf-8") as f:
meta = data["metadata"]
f.write(f"# Benchmark Report: {meta['experiment']}\n\n")
f.write("## Metadata\n\n")
f.write(f"- **Run ID**: {meta['run_id']}\n")
f.write(f"- **Timestamp**: {meta['timestamp']}\n")
f.write(f"- **Chip/Accelerator**: {meta.get('accelerator', '')} / {meta.get('chip', '')}\n")
f.write(f"- **Engine/Backend**: {meta.get('engine', '')} / {meta.get('backend', '')}\n")
f.write(f"- **Hardware**: {meta.get('hardware', '')}\n")
f.write(f"- **Model**: {meta['model']}\n")
f.write(f"- **Git Commit**: {meta.get('git_commit', 'unknown')}\n")
f.write("\n## Results\n\n")
f.write(
"| Scenario | Conc | In/Out | Success | Failed | Req/s | OutTok/s | "
"TTFT p50 | TTFT p99 | TPOT p50 | TPOT p99 | E2E p99 |\n"
)
f.write(
"|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|\n"
)
for s in scenarios:
lat = s["latencies"]
f.write(
f"| {s['name']} "
f"| {s['concurrency']} "
f"| {s['input_len']}/{s['output_len']} "
f"| {s['success']} "
f"| {s['failed']} "
f"| {s.get('request_throughput') or ''} "
f"| {s.get('output_token_throughput') or ''} "
f"| {lat['ttft_ms']['p50'] or ''} "
f"| {lat['ttft_ms']['p99'] or ''} "
f"| {lat['tpot_ms']['p50'] or ''} "
f"| {lat['tpot_ms']['p99'] or ''} "
f"| {lat['e2e_ms']['p99'] or ''} |\n"
)
print(f"Updated {json_path}")
print(f"Wrote {report_path}")
if __name__ == "__main__":
main()

View File

@ -1,312 +1,22 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Parse P800 SGLang benchmark outputs and produce results.json + report.md. """Thin wrapper: delegate to the shared parser at scripts/common/parse_backend.py.
Usage: Replaces per-experiment parse_results.py copies that embedded raw_requests
python3 parse_results.py <result_root> (the results.json bloat bug). Forwards all CLI args to the shared parser.
""" """
import json
import os import os
import re import subprocess
import sys import sys
from pathlib import Path
from collections import OrderedDict
METRIC_PATTERNS = OrderedDict( _HERE = os.path.dirname(os.path.abspath(__file__))
[ _d = _HERE
("successful_requests", [r"Successful requests:\s+(\d+)"]), while True:
("benchmark_duration_s", [r"Benchmark duration \(s\):\s+([\d.]+)"]), cand = os.path.join(_d, "scripts", "common", "parse_backend.py")
("request_throughput", [r"Request throughput \(req/s\):\s+([\d.]+)"]), if os.path.isfile(cand):
("input_token_throughput", [r"Input token throughput \(tok/s\):\s+([\d.]+)"]), sys.exit(subprocess.call([sys.executable, cand, *sys.argv[1:]]))
("output_token_throughput", [r"Output token throughput \(tok/s\):\s+([\d.]+)"]), parent = os.path.dirname(_d)
("total_token_throughput", [r"Total token throughput \(tok/s\):\s+([\d.]+)"]), if parent == _d:
("ttft_mean", [r"Mean TTFT \(ms\):\s+([\d.]+)"]), break
("ttft_p50", [r"Median TTFT \(ms\):\s+([\d.]+)", r"P50 TTFT \(ms\):\s+([\d.]+)"]), _d = parent
("ttft_p90", [r"P90 TTFT \(ms\):\s+([\d.]+)"]), print("ERROR: scripts/common/parse_backend.py not found upwards from " + _HERE, file=sys.stderr)
("ttft_p99", [r"P99 TTFT \(ms\):\s+([\d.]+)"]), sys.exit(1)
("tpot_mean", [r"Mean TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p50", [r"Median TPOT \(ms\):\s+([\d.]+)", r"P50 TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p90", [r"P90 TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p99", [r"P99 TPOT \(ms\):\s+([\d.]+)"]),
("e2e_mean", [r"Mean E2E \(ms\):\s+([\d.]+)"]),
("e2e_p50", [r"Median E2E \(ms\):\s+([\d.]+)", r"P50 E2E \(ms\):\s+([\d.]+)"]),
("e2e_p90", [r"P90 E2E \(ms\):\s+([\d.]+)"]),
("e2e_p99", [r"P99 E2E \(ms\):\s+([\d.]+)"]),
]
)
def find_float(text: str, patterns: list[str]) -> float | None:
for pat in patterns:
m = re.search(pat, text)
if m:
try:
return float(m.group(1))
except ValueError:
return None
return None
def parse_summary_log(log_text: str) -> dict:
result = {}
for key, patterns in METRIC_PATTERNS.items():
result[key] = find_float(log_text, patterns)
return result
def percentile(values: list[float], p: float) -> float:
if not values:
return 0.0
sorted_values = sorted(values)
k = (len(sorted_values) - 1) * p / 100.0
f = int(k)
c = min(f + 1, len(sorted_values) - 1)
if f == c:
return sorted_values[f]
return sorted_values[f] * (c - k) + sorted_values[c] * (k - f)
def parse_jsonl(jsonl_path: Path) -> tuple[list[dict], dict]:
"""Parse sglang.bench_serving --output-file output.
Newer sglang writes one aggregate JSON object. Older versions write one
JSON object per request. Returns (raw_requests, aggregates).
"""
text = jsonl_path.read_text(encoding="utf-8", errors="replace").strip()
if not text:
return [], {}
# Try single aggregate JSON object first.
try:
data = json.loads(text)
if isinstance(data, dict) and "mean_e2e_latency_ms" in data:
aggregates = {
"success": data.get("total_output_tokens", 0) > 0 and 1 or 0,
"failed": 0,
"input_tokens": data.get("total_input_tokens", 0),
"output_tokens": data.get("total_output_tokens", 0),
"latencies": {
"e2e_ms": {
"mean": data.get("mean_e2e_latency_ms"),
"p50": data.get("median_e2e_latency_ms"),
"p90": data.get("p90_e2e_latency_ms"),
"p95": None,
"p99": data.get("p99_e2e_latency_ms"),
},
"ttft_ms": {
"mean": data.get("mean_ttft_ms"),
"p50": data.get("median_ttft_ms"),
"p90": None,
"p95": None,
"p99": data.get("p99_ttft_ms"),
},
"tpot_ms": {
"mean": data.get("mean_tpot_ms"),
"p50": data.get("median_tpot_ms"),
"p90": None,
"p95": None,
"p99": data.get("p99_tpot_ms"),
},
"itl_ms": {
"mean": data.get("mean_itl_ms"),
"p50": data.get("median_itl_ms"),
"p90": None,
"p95": data.get("p95_itl_ms"),
"p99": data.get("p99_itl_ms"),
},
},
}
return [data], aggregates
except json.JSONDecodeError:
pass
# Fall back to JSONL per-request parsing.
raw_requests = []
ttfts = []
tpots = []
itls = []
e2es = []
input_tokens = []
output_tokens = []
success = 0
failed = 0
for line in text.splitlines():
line = line.strip()
if not line:
continue
try:
req = json.loads(line)
except json.JSONDecodeError:
continue
raw_requests.append(req)
ttft = req.get("ttft") or req.get("ttft_ms") or 0
tpot = req.get("tpot") or req.get("tpot_ms") or 0
itl = req.get("inter_token_latency") or req.get("itl") or req.get("itl_ms") or 0
e2e = req.get("e2e_latency") or req.get("e2e") or req.get("e2e_ms") or 0
in_tok = req.get("input_tokens") or req.get("prompt_tokens") or 0
out_tok = req.get("output_tokens") or req.get("completion_tokens") or 0
if ttft:
ttfts.append(float(ttft))
if tpot:
tpots.append(float(tpot))
if itl:
itls.append(float(itl))
if e2e:
e2es.append(float(e2e))
if in_tok:
input_tokens.append(int(in_tok))
if out_tok:
output_tokens.append(int(out_tok))
if req.get("success", True):
success += 1
else:
failed += 1
def latency_bucket(values: list[float]) -> dict:
if not values:
return {"mean": None, "p50": None, "p90": None, "p95": None, "p99": None}
return {
"mean": round(sum(values) / len(values), 2),
"p50": round(percentile(values, 50), 2),
"p90": round(percentile(values, 90), 2),
"p95": round(percentile(values, 95), 2),
"p99": round(percentile(values, 99), 2),
}
aggregates = {
"success": success,
"failed": failed,
"input_tokens": sum(input_tokens),
"output_tokens": sum(output_tokens),
"latencies": {
"e2e_ms": latency_bucket(e2es),
"ttft_ms": latency_bucket(ttfts),
"tpot_ms": latency_bucket(tpots),
"itl_ms": latency_bucket(itls),
},
}
return raw_requests, aggregates
def parse_scenario(result_root: Path, raw_file: Path) -> dict | None:
"""Parse one raw output file into a scenario dict."""
# Filename: {chip}_{engine}_{MMDD}_{concurrency}_{input_len}_{output_len}.jsonl
parts = raw_file.stem.split("_")
if len(parts) < 6:
return None
try:
concurrency = int(parts[-3])
input_len = int(parts[-2])
output_len = int(parts[-1])
except ValueError:
return None
log_file = result_root / "logs" / f"sglang_c{concurrency}_i{input_len}_o{output_len}.log"
summary = {}
if log_file.exists():
summary = parse_summary_log(log_file.read_text(encoding="utf-8", errors="replace"))
_, aggregates = parse_jsonl(raw_file)
# For newer sglang aggregate JSON, success/failed are not present in the
# raw output file. Override them from the human-readable summary log when
# it is available.
if summary.get("successful_requests") is not None:
aggregates["success"] = int(summary["successful_requests"])
# The summary log only reports successes; assume failures are zero
# unless the aggregate JSON already provided a non-zero failed count.
if not aggregates.get("failed"):
aggregates["failed"] = 0
scenario = {
"name": f"c{concurrency}_i{input_len}_o{output_len}",
"concurrency": concurrency,
"input_len": input_len,
"output_len": output_len,
"success": aggregates["success"],
"failed": aggregates["failed"],
"duration_s": summary.get("benchmark_duration_s"),
"request_throughput": summary.get("request_throughput"),
"input_token_throughput": summary.get("input_token_throughput"),
"output_token_throughput": summary.get("output_token_throughput"),
"total_token_throughput": summary.get("total_token_throughput"),
"accept_length": None,
"latencies": aggregates["latencies"],
}
return scenario
def main() -> None:
result_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("results")
raw_dir = result_root / "raw_outputs"
json_path = result_root / "results.json"
report_path = result_root / "report.md"
if not json_path.exists():
raise SystemExit(f"metadata results.json not found: {json_path}")
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
scenarios = []
if raw_dir.exists():
for raw_file in sorted(raw_dir.glob("*.jsonl")):
scenario = parse_scenario(result_root, raw_file)
if scenario:
scenarios.append(scenario)
data["scenarios"] = scenarios
with open(json_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
# Generate report.md
with open(report_path, "w", encoding="utf-8") as f:
meta = data["metadata"]
f.write(f"# Benchmark Report: {meta['experiment']}\n\n")
f.write("## Metadata\n\n")
f.write(f"- **Run ID**: {meta['run_id']}\n")
f.write(f"- **Timestamp**: {meta['timestamp']}\n")
f.write(f"- **Chip/Accelerator**: {meta.get('accelerator', '')} / {meta.get('chip', '')}\n")
f.write(f"- **Engine/Backend**: {meta.get('engine', '')} / {meta.get('backend', '')}\n")
f.write(f"- **Hardware**: {meta.get('hardware', '')}\n")
f.write(f"- **Model**: {meta['model']}\n")
f.write(f"- **Git Commit**: {meta.get('git_commit', 'unknown')}\n")
f.write("\n## Results\n\n")
f.write(
"| Scenario | Conc | In/Out | Success | Failed | Req/s | OutTok/s | "
"TTFT p50 | TTFT p99 | TPOT p50 | TPOT p99 | E2E p99 |\n"
)
f.write(
"|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|\n"
)
for s in scenarios:
lat = s["latencies"]
f.write(
f"| {s['name']} "
f"| {s['concurrency']} "
f"| {s['input_len']}/{s['output_len']} "
f"| {s['success']} "
f"| {s['failed']} "
f"| {s.get('request_throughput') or ''} "
f"| {s.get('output_token_throughput') or ''} "
f"| {lat['ttft_ms']['p50'] or ''} "
f"| {lat['ttft_ms']['p99'] or ''} "
f"| {lat['tpot_ms']['p50'] or ''} "
f"| {lat['tpot_ms']['p99'] or ''} "
f"| {lat['e2e_ms']['p99'] or ''} |\n"
)
print(f"Updated {json_path}")
print(f"Wrote {report_path}")
if __name__ == "__main__":
main()

View File

@ -1,312 +1,22 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Parse P800 SGLang benchmark outputs and produce results.json + report.md. """Thin wrapper: delegate to the shared parser at scripts/common/parse_backend.py.
Usage: Replaces per-experiment parse_results.py copies that embedded raw_requests
python3 parse_results.py <result_root> (the results.json bloat bug). Forwards all CLI args to the shared parser.
""" """
import json
import os import os
import re import subprocess
import sys import sys
from pathlib import Path
from collections import OrderedDict
METRIC_PATTERNS = OrderedDict( _HERE = os.path.dirname(os.path.abspath(__file__))
[ _d = _HERE
("successful_requests", [r"Successful requests:\s+(\d+)"]), while True:
("benchmark_duration_s", [r"Benchmark duration \(s\):\s+([\d.]+)"]), cand = os.path.join(_d, "scripts", "common", "parse_backend.py")
("request_throughput", [r"Request throughput \(req/s\):\s+([\d.]+)"]), if os.path.isfile(cand):
("input_token_throughput", [r"Input token throughput \(tok/s\):\s+([\d.]+)"]), sys.exit(subprocess.call([sys.executable, cand, *sys.argv[1:]]))
("output_token_throughput", [r"Output token throughput \(tok/s\):\s+([\d.]+)"]), parent = os.path.dirname(_d)
("total_token_throughput", [r"Total token throughput \(tok/s\):\s+([\d.]+)"]), if parent == _d:
("ttft_mean", [r"Mean TTFT \(ms\):\s+([\d.]+)"]), break
("ttft_p50", [r"Median TTFT \(ms\):\s+([\d.]+)", r"P50 TTFT \(ms\):\s+([\d.]+)"]), _d = parent
("ttft_p90", [r"P90 TTFT \(ms\):\s+([\d.]+)"]), print("ERROR: scripts/common/parse_backend.py not found upwards from " + _HERE, file=sys.stderr)
("ttft_p99", [r"P99 TTFT \(ms\):\s+([\d.]+)"]), sys.exit(1)
("tpot_mean", [r"Mean TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p50", [r"Median TPOT \(ms\):\s+([\d.]+)", r"P50 TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p90", [r"P90 TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p99", [r"P99 TPOT \(ms\):\s+([\d.]+)"]),
("e2e_mean", [r"Mean E2E \(ms\):\s+([\d.]+)"]),
("e2e_p50", [r"Median E2E \(ms\):\s+([\d.]+)", r"P50 E2E \(ms\):\s+([\d.]+)"]),
("e2e_p90", [r"P90 E2E \(ms\):\s+([\d.]+)"]),
("e2e_p99", [r"P99 E2E \(ms\):\s+([\d.]+)"]),
]
)
def find_float(text: str, patterns: list[str]) -> float | None:
for pat in patterns:
m = re.search(pat, text)
if m:
try:
return float(m.group(1))
except ValueError:
return None
return None
def parse_summary_log(log_text: str) -> dict:
result = {}
for key, patterns in METRIC_PATTERNS.items():
result[key] = find_float(log_text, patterns)
return result
def percentile(values: list[float], p: float) -> float:
if not values:
return 0.0
sorted_values = sorted(values)
k = (len(sorted_values) - 1) * p / 100.0
f = int(k)
c = min(f + 1, len(sorted_values) - 1)
if f == c:
return sorted_values[f]
return sorted_values[f] * (c - k) + sorted_values[c] * (k - f)
def parse_jsonl(jsonl_path: Path) -> tuple[list[dict], dict]:
"""Parse sglang.bench_serving --output-file output.
Newer sglang writes one aggregate JSON object. Older versions write one
JSON object per request. Returns (raw_requests, aggregates).
"""
text = jsonl_path.read_text(encoding="utf-8", errors="replace").strip()
if not text:
return [], {}
# Try single aggregate JSON object first.
try:
data = json.loads(text)
if isinstance(data, dict) and "mean_e2e_latency_ms" in data:
aggregates = {
"success": data.get("total_output_tokens", 0) > 0 and 1 or 0,
"failed": 0,
"input_tokens": data.get("total_input_tokens", 0),
"output_tokens": data.get("total_output_tokens", 0),
"latencies": {
"e2e_ms": {
"mean": data.get("mean_e2e_latency_ms"),
"p50": data.get("median_e2e_latency_ms"),
"p90": data.get("p90_e2e_latency_ms"),
"p95": None,
"p99": data.get("p99_e2e_latency_ms"),
},
"ttft_ms": {
"mean": data.get("mean_ttft_ms"),
"p50": data.get("median_ttft_ms"),
"p90": None,
"p95": None,
"p99": data.get("p99_ttft_ms"),
},
"tpot_ms": {
"mean": data.get("mean_tpot_ms"),
"p50": data.get("median_tpot_ms"),
"p90": None,
"p95": None,
"p99": data.get("p99_tpot_ms"),
},
"itl_ms": {
"mean": data.get("mean_itl_ms"),
"p50": data.get("median_itl_ms"),
"p90": None,
"p95": data.get("p95_itl_ms"),
"p99": data.get("p99_itl_ms"),
},
},
}
return [data], aggregates
except json.JSONDecodeError:
pass
# Fall back to JSONL per-request parsing.
raw_requests = []
ttfts = []
tpots = []
itls = []
e2es = []
input_tokens = []
output_tokens = []
success = 0
failed = 0
for line in text.splitlines():
line = line.strip()
if not line:
continue
try:
req = json.loads(line)
except json.JSONDecodeError:
continue
raw_requests.append(req)
ttft = req.get("ttft") or req.get("ttft_ms") or 0
tpot = req.get("tpot") or req.get("tpot_ms") or 0
itl = req.get("inter_token_latency") or req.get("itl") or req.get("itl_ms") or 0
e2e = req.get("e2e_latency") or req.get("e2e") or req.get("e2e_ms") or 0
in_tok = req.get("input_tokens") or req.get("prompt_tokens") or 0
out_tok = req.get("output_tokens") or req.get("completion_tokens") or 0
if ttft:
ttfts.append(float(ttft))
if tpot:
tpots.append(float(tpot))
if itl:
itls.append(float(itl))
if e2e:
e2es.append(float(e2e))
if in_tok:
input_tokens.append(int(in_tok))
if out_tok:
output_tokens.append(int(out_tok))
if req.get("success", True):
success += 1
else:
failed += 1
def latency_bucket(values: list[float]) -> dict:
if not values:
return {"mean": None, "p50": None, "p90": None, "p95": None, "p99": None}
return {
"mean": round(sum(values) / len(values), 2),
"p50": round(percentile(values, 50), 2),
"p90": round(percentile(values, 90), 2),
"p95": round(percentile(values, 95), 2),
"p99": round(percentile(values, 99), 2),
}
aggregates = {
"success": success,
"failed": failed,
"input_tokens": sum(input_tokens),
"output_tokens": sum(output_tokens),
"latencies": {
"e2e_ms": latency_bucket(e2es),
"ttft_ms": latency_bucket(ttfts),
"tpot_ms": latency_bucket(tpots),
"itl_ms": latency_bucket(itls),
},
}
return raw_requests, aggregates
def parse_scenario(result_root: Path, raw_file: Path) -> dict | None:
"""Parse one raw output file into a scenario dict."""
# Filename: {chip}_{engine}_{MMDD}_{concurrency}_{input_len}_{output_len}.jsonl
parts = raw_file.stem.split("_")
if len(parts) < 6:
return None
try:
concurrency = int(parts[-3])
input_len = int(parts[-2])
output_len = int(parts[-1])
except ValueError:
return None
log_file = result_root / "logs" / f"sglang_c{concurrency}_i{input_len}_o{output_len}.log"
summary = {}
if log_file.exists():
summary = parse_summary_log(log_file.read_text(encoding="utf-8", errors="replace"))
_, aggregates = parse_jsonl(raw_file)
# For newer sglang aggregate JSON, success/failed are not present in the
# raw output file. Override them from the human-readable summary log when
# it is available.
if summary.get("successful_requests") is not None:
aggregates["success"] = int(summary["successful_requests"])
# The summary log only reports successes; assume failures are zero
# unless the aggregate JSON already provided a non-zero failed count.
if not aggregates.get("failed"):
aggregates["failed"] = 0
scenario = {
"name": f"c{concurrency}_i{input_len}_o{output_len}",
"concurrency": concurrency,
"input_len": input_len,
"output_len": output_len,
"success": aggregates["success"],
"failed": aggregates["failed"],
"duration_s": summary.get("benchmark_duration_s"),
"request_throughput": summary.get("request_throughput"),
"input_token_throughput": summary.get("input_token_throughput"),
"output_token_throughput": summary.get("output_token_throughput"),
"total_token_throughput": summary.get("total_token_throughput"),
"accept_length": None,
"latencies": aggregates["latencies"],
}
return scenario
def main() -> None:
result_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("results")
raw_dir = result_root / "raw_outputs"
json_path = result_root / "results.json"
report_path = result_root / "report.md"
if not json_path.exists():
raise SystemExit(f"metadata results.json not found: {json_path}")
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
scenarios = []
if raw_dir.exists():
for raw_file in sorted(raw_dir.glob("*.jsonl")):
scenario = parse_scenario(result_root, raw_file)
if scenario:
scenarios.append(scenario)
data["scenarios"] = scenarios
with open(json_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
# Generate report.md
with open(report_path, "w", encoding="utf-8") as f:
meta = data["metadata"]
f.write(f"# Benchmark Report: {meta['experiment']}\n\n")
f.write("## Metadata\n\n")
f.write(f"- **Run ID**: {meta['run_id']}\n")
f.write(f"- **Timestamp**: {meta['timestamp']}\n")
f.write(f"- **Chip/Accelerator**: {meta.get('accelerator', '')} / {meta.get('chip', '')}\n")
f.write(f"- **Engine/Backend**: {meta.get('engine', '')} / {meta.get('backend', '')}\n")
f.write(f"- **Hardware**: {meta.get('hardware', '')}\n")
f.write(f"- **Model**: {meta['model']}\n")
f.write(f"- **Git Commit**: {meta.get('git_commit', 'unknown')}\n")
f.write("\n## Results\n\n")
f.write(
"| Scenario | Conc | In/Out | Success | Failed | Req/s | OutTok/s | "
"TTFT p50 | TTFT p99 | TPOT p50 | TPOT p99 | E2E p99 |\n"
)
f.write(
"|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|\n"
)
for s in scenarios:
lat = s["latencies"]
f.write(
f"| {s['name']} "
f"| {s['concurrency']} "
f"| {s['input_len']}/{s['output_len']} "
f"| {s['success']} "
f"| {s['failed']} "
f"| {s.get('request_throughput') or ''} "
f"| {s.get('output_token_throughput') or ''} "
f"| {lat['ttft_ms']['p50'] or ''} "
f"| {lat['ttft_ms']['p99'] or ''} "
f"| {lat['tpot_ms']['p50'] or ''} "
f"| {lat['tpot_ms']['p99'] or ''} "
f"| {lat['e2e_ms']['p99'] or ''} |\n"
)
print(f"Updated {json_path}")
print(f"Wrote {report_path}")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,22 @@
#!/usr/bin/env python3
"""Thin wrapper: delegate to the shared parser at scripts/common/parse_backend.py.
Replaces per-experiment parse_results.py copies that embedded raw_requests
(the results.json bloat bug). Forwards all CLI args to the shared parser.
"""
import os
import subprocess
import sys
_HERE = os.path.dirname(os.path.abspath(__file__))
_d = _HERE
while True:
cand = os.path.join(_d, "scripts", "common", "parse_backend.py")
if os.path.isfile(cand):
sys.exit(subprocess.call([sys.executable, cand, *sys.argv[1:]]))
parent = os.path.dirname(_d)
if parent == _d:
break
_d = parent
print("ERROR: scripts/common/parse_backend.py not found upwards from " + _HERE, file=sys.stderr)
sys.exit(1)

View File

@ -1,313 +1,22 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Parse P800 SGLang benchmark outputs and produce results.json + report.md. """Thin wrapper: delegate to the shared parser at scripts/common/parse_backend.py.
Usage: Replaces per-experiment parse_results.py copies that embedded raw_requests
python3 parse_results.py <result_root> (the results.json bloat bug). Forwards all CLI args to the shared parser.
""" """
import json
import os import os
import re import subprocess
import sys import sys
from pathlib import Path
from collections import OrderedDict
METRIC_PATTERNS = OrderedDict( _HERE = os.path.dirname(os.path.abspath(__file__))
[ _d = _HERE
("successful_requests", [r"Successful requests:\s+(\d+)"]), while True:
("benchmark_duration_s", [r"Benchmark duration \(s\):\s+([\d.]+)"]), cand = os.path.join(_d, "scripts", "common", "parse_backend.py")
("request_throughput", [r"Request throughput \(req/s\):\s+([\d.]+)"]), if os.path.isfile(cand):
("input_token_throughput", [r"Input token throughput \(tok/s\):\s+([\d.]+)"]), sys.exit(subprocess.call([sys.executable, cand, *sys.argv[1:]]))
("output_token_throughput", [r"Output token throughput \(tok/s\):\s+([\d.]+)"]), parent = os.path.dirname(_d)
("total_token_throughput", [r"Total token throughput \(tok/s\):\s+([\d.]+)"]), if parent == _d:
("ttft_mean", [r"Mean TTFT \(ms\):\s+([\d.]+)"]), break
("ttft_p50", [r"Median TTFT \(ms\):\s+([\d.]+)", r"P50 TTFT \(ms\):\s+([\d.]+)"]), _d = parent
("ttft_p90", [r"P90 TTFT \(ms\):\s+([\d.]+)"]), print("ERROR: scripts/common/parse_backend.py not found upwards from " + _HERE, file=sys.stderr)
("ttft_p99", [r"P99 TTFT \(ms\):\s+([\d.]+)"]), sys.exit(1)
("tpot_mean", [r"Mean TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p50", [r"Median TPOT \(ms\):\s+([\d.]+)", r"P50 TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p90", [r"P90 TPOT \(ms\):\s+([\d.]+)"]),
("tpot_p99", [r"P99 TPOT \(ms\):\s+([\d.]+)"]),
("e2e_mean", [r"Mean E2E \(ms\):\s+([\d.]+)"]),
("e2e_p50", [r"Median E2E \(ms\):\s+([\d.]+)", r"P50 E2E \(ms\):\s+([\d.]+)"]),
("e2e_p90", [r"P90 E2E \(ms\):\s+([\d.]+)"]),
("e2e_p99", [r"P99 E2E \(ms\):\s+([\d.]+)"]),
]
)
def find_float(text: str, patterns: list[str]) -> float | None:
for pat in patterns:
m = re.search(pat, text)
if m:
try:
return float(m.group(1))
except ValueError:
return None
return None
def parse_summary_log(log_text: str) -> dict:
result = {}
for key, patterns in METRIC_PATTERNS.items():
result[key] = find_float(log_text, patterns)
return result
def percentile(values: list[float], p: float) -> float:
if not values:
return 0.0
sorted_values = sorted(values)
k = (len(sorted_values) - 1) * p / 100.0
f = int(k)
c = min(f + 1, len(sorted_values) - 1)
if f == c:
return sorted_values[f]
return sorted_values[f] * (c - k) + sorted_values[c] * (k - f)
def parse_jsonl(jsonl_path: Path) -> tuple[list[dict], dict]:
"""Parse sglang.bench_serving --output-file output.
Newer sglang writes one aggregate JSON object. Older versions write one
JSON object per request. Returns (raw_requests, aggregates).
"""
text = jsonl_path.read_text(encoding="utf-8", errors="replace").strip()
if not text:
return [], {}
# Try single aggregate JSON object first.
try:
data = json.loads(text)
if isinstance(data, dict) and "mean_e2e_latency_ms" in data:
aggregates = {
"success": data.get("total_output_tokens", 0) > 0 and 1 or 0,
"failed": 0,
"input_tokens": data.get("total_input_tokens", 0),
"output_tokens": data.get("total_output_tokens", 0),
"latencies": {
"e2e_ms": {
"mean": data.get("mean_e2e_latency_ms"),
"p50": data.get("median_e2e_latency_ms"),
"p90": data.get("p90_e2e_latency_ms"),
"p95": None,
"p99": data.get("p99_e2e_latency_ms"),
},
"ttft_ms": {
"mean": data.get("mean_ttft_ms"),
"p50": data.get("median_ttft_ms"),
"p90": None,
"p95": None,
"p99": data.get("p99_ttft_ms"),
},
"tpot_ms": {
"mean": data.get("mean_tpot_ms"),
"p50": data.get("median_tpot_ms"),
"p90": None,
"p95": None,
"p99": data.get("p99_tpot_ms"),
},
"itl_ms": {
"mean": data.get("mean_itl_ms"),
"p50": data.get("median_itl_ms"),
"p90": None,
"p95": data.get("p95_itl_ms"),
"p99": data.get("p99_itl_ms"),
},
},
}
return [data], aggregates
except json.JSONDecodeError:
pass
# Fall back to JSONL per-request parsing.
raw_requests = []
ttfts = []
tpots = []
itls = []
e2es = []
input_tokens = []
output_tokens = []
success = 0
failed = 0
for line in text.splitlines():
line = line.strip()
if not line:
continue
try:
req = json.loads(line)
except json.JSONDecodeError:
continue
raw_requests.append(req)
ttft = req.get("ttft") or req.get("ttft_ms") or 0
tpot = req.get("tpot") or req.get("tpot_ms") or 0
itl = req.get("inter_token_latency") or req.get("itl") or req.get("itl_ms") or 0
e2e = req.get("e2e_latency") or req.get("e2e") or req.get("e2e_ms") or 0
in_tok = req.get("input_tokens") or req.get("prompt_tokens") or 0
out_tok = req.get("output_tokens") or req.get("completion_tokens") or 0
if ttft:
ttfts.append(float(ttft))
if tpot:
tpots.append(float(tpot))
if itl:
itls.append(float(itl))
if e2e:
e2es.append(float(e2e))
if in_tok:
input_tokens.append(int(in_tok))
if out_tok:
output_tokens.append(int(out_tok))
if req.get("success", True):
success += 1
else:
failed += 1
def latency_bucket(values: list[float]) -> dict:
if not values:
return {"mean": None, "p50": None, "p90": None, "p95": None, "p99": None}
return {
"mean": round(sum(values) / len(values), 2),
"p50": round(percentile(values, 50), 2),
"p90": round(percentile(values, 90), 2),
"p95": round(percentile(values, 95), 2),
"p99": round(percentile(values, 99), 2),
}
aggregates = {
"success": success,
"failed": failed,
"input_tokens": sum(input_tokens),
"output_tokens": sum(output_tokens),
"latencies": {
"e2e_ms": latency_bucket(e2es),
"ttft_ms": latency_bucket(ttfts),
"tpot_ms": latency_bucket(tpots),
"itl_ms": latency_bucket(itls),
},
}
return raw_requests, aggregates
def parse_scenario(result_root: Path, raw_file: Path) -> dict | None:
"""Parse one raw output file into a scenario dict."""
# Filename: {chip}_{engine}_{MMDD}_{concurrency}_{input_len}_{output_len}.jsonl
parts = raw_file.stem.split("_")
if len(parts) < 6:
return None
try:
concurrency = int(parts[-3])
input_len = int(parts[-2])
output_len = int(parts[-1])
except ValueError:
return None
log_file = result_root / "logs" / f"sglang_c{concurrency}_i{input_len}_o{output_len}.log"
summary = {}
if log_file.exists():
summary = parse_summary_log(log_file.read_text(encoding="utf-8", errors="replace"))
raw_requests, aggregates = parse_jsonl(raw_file)
# For newer sglang aggregate JSON, success/failed are not present in the
# raw output file. Override them from the human-readable summary log when
# it is available.
if summary.get("successful_requests") is not None:
aggregates["success"] = int(summary["successful_requests"])
# The summary log only reports successes; assume failures are zero
# unless the aggregate JSON already provided a non-zero failed count.
if not aggregates.get("failed"):
aggregates["failed"] = 0
scenario = {
"name": f"c{concurrency}_i{input_len}_o{output_len}",
"concurrency": concurrency,
"input_len": input_len,
"output_len": output_len,
"success": aggregates["success"],
"failed": aggregates["failed"],
"duration_s": summary.get("benchmark_duration_s"),
"request_throughput": summary.get("request_throughput"),
"input_token_throughput": summary.get("input_token_throughput"),
"output_token_throughput": summary.get("output_token_throughput"),
"total_token_throughput": summary.get("total_token_throughput"),
"accept_length": None,
"latencies": aggregates["latencies"],
"raw_requests": raw_requests[:100] if len(raw_requests) <= 100 else None,
}
return scenario
def main() -> None:
result_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("results")
raw_dir = result_root / "raw_outputs"
json_path = result_root / "results.json"
report_path = result_root / "report.md"
if not json_path.exists():
raise SystemExit(f"metadata results.json not found: {json_path}")
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
scenarios = []
if raw_dir.exists():
for raw_file in sorted(raw_dir.glob("*.jsonl")):
scenario = parse_scenario(result_root, raw_file)
if scenario:
scenarios.append(scenario)
data["scenarios"] = scenarios
with open(json_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
# Generate report.md
with open(report_path, "w", encoding="utf-8") as f:
meta = data["metadata"]
f.write(f"# Benchmark Report: {meta['experiment']}\n\n")
f.write("## Metadata\n\n")
f.write(f"- **Run ID**: {meta['run_id']}\n")
f.write(f"- **Timestamp**: {meta['timestamp']}\n")
f.write(f"- **Chip/Accelerator**: {meta.get('accelerator', '')} / {meta.get('chip', '')}\n")
f.write(f"- **Engine/Backend**: {meta.get('engine', '')} / {meta.get('backend', '')}\n")
f.write(f"- **Hardware**: {meta.get('hardware', '')}\n")
f.write(f"- **Model**: {meta['model']}\n")
f.write(f"- **Git Commit**: {meta.get('git_commit', 'unknown')}\n")
f.write("\n## Results\n\n")
f.write(
"| Scenario | Conc | In/Out | Success | Failed | Req/s | OutTok/s | "
"TTFT p50 | TTFT p99 | TPOT p50 | TPOT p99 | E2E p99 |\n"
)
f.write(
"|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|\n"
)
for s in scenarios:
lat = s["latencies"]
f.write(
f"| {s['name']} "
f"| {s['concurrency']} "
f"| {s['input_len']}/{s['output_len']} "
f"| {s['success']} "
f"| {s['failed']} "
f"| {s.get('request_throughput') or ''} "
f"| {s.get('output_token_throughput') or ''} "
f"| {lat['ttft_ms']['p50'] or ''} "
f"| {lat['ttft_ms']['p99'] or ''} "
f"| {lat['tpot_ms']['p50'] or ''} "
f"| {lat['tpot_ms']['p99'] or ''} "
f"| {lat['e2e_ms']['p99'] or ''} |\n"
)
print(f"Updated {json_path}")
print(f"Wrote {report_path}")
if __name__ == "__main__":
main()