qqtang/sskj
1
3
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Feat] add DSV4-Pro two-node SGLang quick map

Browse Source
This commit is contained in:
Zhiyi Hong 2026-07-30 14:40:16 +08:00
parent e885fd0dc2
commit d5d96bd7e4
8 changed files with 1835 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
README.md
View File

@ -1,5 +1,9 @@
# sskj — 多平台大模型推理性能基准测试项目 # sskj — 多平台大模型推理性能基准测试项目
> **更新（2026-07-30 14:33:52 CST）**
>
> 新增独立的 `dsv4pro_pro6000d_2node_sglang_tp16_quick_map` 快速性能地图与混合干扰 A/B。实验只保留一个 Shell 入口；旧 TP16 全量脚本保持不变。首轮真机验证已确认双机 TP16 服务可用，并据实测耗时将快速矩阵缩为一波请求，同时修正 Warm-up 污染 Prefix Cache 和混合负载注入时序。
>
> 历史更新见 `git log`。项目目的与工作流见下方。 > 历史更新见 `git log`。项目目的与工作流见下方。
> **项目目的**：当新显卡（GPU/NPU）到货时，用最短时间完成大模型在该平台上的推理性能评估与部署配置选型。 > **项目目的**：当新显卡（GPU/NPU）到货时，用最短时间完成大模型在该平台上的推理性能评估与部署配置选型。
> 当前模型：DeepSeek-V4-Flash（FP8 / INT8）；后续接入 GLM5.2，**完全复用**本项目的实验与报告流程。 > 当前模型：DeepSeek-V4-Flash（FP8 / INT8）；后续接入 GLM5.2，**完全复用**本项目的实验与报告流程。
@ -44,8 +48,9 @@
| `experiments/p800/dsv4_p800_sglang_tp_dp_matrix/` | P800 + SGLang（INT8；TP2/DP4 启动 OOM 无数据，见 config.env 注释） | | `experiments/p800/dsv4_p800_sglang_tp_dp_matrix/` | P800 + SGLang（INT8；TP2/DP4 启动 OOM 无数据，见 config.env 注释） |
| `experiments/pro6000/dsv4_pro6000_vllm_tp_dp_matrix/` | RTX 6000D + vLLM | | `experiments/pro6000/dsv4_pro6000_vllm_tp_dp_matrix/` | RTX 6000D + vLLM |
| `experiments/pro6000/dsv4_pro6000_sglang_tp_dp_matrix/` | RTX 6000D + SGLang | | `experiments/pro6000/dsv4_pro6000_sglang_tp_dp_matrix/` | RTX 6000D + SGLang |
| `experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/` | 双机 RTX 6000D + DeepSeek-V4-Pro，SGLang TP16 快速性能地图与混合干扰 A/B |
每个目录内：`run_bench.sh` 跑固定并发矩阵；`run_adaptive_concurrency.sh` 从 C=1 指数倍增搜饱和点；`run_adaptive_concurrency_add16.sh` 从 C=16 线性 +16 步进、带 TTFT SLO 停止与回退（当前主力用法，见 `experiments/ADAPTIVE_CONCURRENCY_USAGE.md`）。 TP/DP matrix 目录内：`run_bench.sh` 跑固定并发矩阵；`run_adaptive_concurrency.sh` 从 C=1 指数倍增搜饱和点；`run_adaptive_concurrency_add16.sh` 从 C=16 线性 +16 步进、带 TTFT SLO 停止与回退（当前主力用法，见 `experiments/ADAPTIVE_CONCURRENCY_USAGE.md`）。专项 quick-map 的入口以对应目录 README 为准。
### 其他实验（H200 老形态 & 专项） ### 其他实验（H200 老形态 & 专项）

4
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/.gitignore vendored Normal file
View File

@ -0,0 +1,4 @@
results/
runtime/
__pycache__/
tests/__pycache__/

118
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/README.md Normal file
View File

@ -0,0 +1,118 @@
# DeepSeek-V4-Pro / Pro6000D / Two-Node / SGLang TP16 Quick Map
This directory contains the short, repeatable performance-map suite for
DeepSeek-V4-Pro on two RTX PRO 6000 Blackwell nodes. It does not modify or call the existing
`dsv4_pro6000_sglang_tp16/run_batch.sh`.
## Scope
The fixed suite covers nine points:
| Group | ISL | OSL | Concurrency |
|---|---:|---:|---:|
| Prefill latency | 1K / 32K / 128K | 1 | 1 |
| Prefill throughput | 32K | 1 | 16 |
| Decode latency | 1K | 1K | 1 |
| Decode throughput | 1K | 1K | 16 / 32 / 64 |
| Balanced pressure | 32K | 1K | 8 |
The same main entry also runs a mixed-interference A/B:
1. Run a finite 64-request `1K -> 1K, C=32` decode control.
2. Run the same decode workload again.
3. Wait for the background benchmark to enter its measured run, then inject one
`128K -> 1, C=1` prefill request.
4. Report the change in output TPS and P95 TPOT/TTFT.
This aggregate mixed test is sufficient for the first performance-map pass. Per-request time-series
attribution belongs to the later scheduler/interference phase.
## Files
| File | Purpose |
|---|---|
| `run_quick_map.sh` | The only Shell entry: service lifecycle, fixed map, mixed A/B, cleanup |
| `config.env` | Machine, model, SGLang and benchmark configuration |
| `quick_map_scenarios.tsv` | Nine fixed workload points |
| `quick_map_results.py` | Validate output and generate CSV/JSONL/Markdown summaries |
| `tests/test_quick_map_results.py` | Result parser regression tests |
## Before using GPUs
The scripts default to ports `20002` and `30002`, and use unique container
names. They never stop containers belonging to another experiment.
```bash
cd /data/hzy/sskj/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map
bash -n run_quick_map.sh
python3 -m unittest discover -s tests -v
DRY_RUN=1 bash run_quick_map.sh all
```
The launcher refuses to start while either node has an active GPU compute
process. `ALLOW_BUSY_GPU=1` exists only for deliberate operator override.
## GPU run
Run the complete quick map in `tmux` after both nodes are free:
```bash
tmux new-session -d -s dsv4pro-pro6000d-2node-sglang-quick-map \
-c /data/hzy/sskj/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map
tmux send-keys -t dsv4pro-pro6000d-2node-sglang-quick-map \
'bash run_quick_map.sh all' Enter
```
The default `all` action starts the two-node service, runs the fixed and mixed
suites, writes summaries, and then removes both service containers. The same
entry supports targeted operations when debugging:
```bash
bash run_quick_map.sh start
bash run_quick_map.sh fixed
bash run_quick_map.sh mixed
bash run_quick_map.sh stop
```
## Outputs
```text
results/<RUN_ID>/
run_manifest.json
run.log
summary.csv
summary.jsonl
aggregate.csv
report.md
cases/<case_id>/rep<N>/
bench_cmd.txt
bench.jsonl
bench.log
meta.json
server/
```
`summary.csv` is one row per repetition. `aggregate.csv` groups repetitions and
reports mean throughput plus coefficient of variation. Raw request arrays stay
inside ignored result directories and are not committed.
## Runtime policy
- Quick default: one repetition and one measured request wave
(`num_prompts=C`, including one request when `C=1`).
- Milestone: three repetitions.
- Short Prefill and Decode cases use one warm-up request. Expensive 32K/128K
Prefill cases do not add a same-shape warm-up.
- Every isolated point calls SGLang benchmark with `--flush-cache`; the benchmark
performs its warm-up first and then clears the server Prefix Cache before
measured requests.
- No SLO-based early stop in the fixed map.
- A case failure is recorded; an unhealthy service aborts the run to avoid
producing a cascade of meaningless failures.
- Benchmark and profiler results must not be mixed. This directory does not
enable a profiler.
Hardware validation is recorded in the implementation report linked from the
repository's optimization plan.

67
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/config.env Normal file
View File

@ -0,0 +1,67 @@
# DeepSeek-V4-Pro SGLang two-node quick performance map.
# This experiment is independent from dsv4_pro6000_sglang_tp16.
EXPERIMENT="dsv4pro_pro6000d_2node_sglang_tp16_quick_map"
MODEL_NAME="DeepSeek-V4-Pro"
MODEL_PATH="${MODEL_PATH:-/data/hf_models/DeepSeek-V4-Pro}"
# Two RTX PRO 6000 Blackwell nodes.
HEAD_NODE="${HEAD_NODE:-10.101.0.11}"
WORKER_NODE="${WORKER_NODE:-10.101.0.13}"
HEAD_IP="${HEAD_IP:-10.101.0.11}"
DIST_INIT_PORT="${DIST_INIT_PORT:-20002}"
SGLANG_PORT="${SGLANG_PORT:-30002}"
TP_SIZE="${TP_SIZE:-16}"
EP_SIZE="${EP_SIZE:-2}"
NNODES="${NNODES:-2}"
CUDA_VISIBLE_DEVICES="${CUDA_VISIBLE_DEVICES:-0,1,2,3,4,5,6,7}"
DOCKER_IMAGE="${DOCKER_IMAGE:-lmsysorg/sglang:nightly-dev-cu13-20260720-b3570a45}"
DOCKER_CLIENT_IMAGE="${DOCKER_CLIENT_IMAGE:-$DOCKER_IMAGE}"
SGLANG_CACHE_DIR="${SGLANG_CACHE_DIR:-/data/hzy/sglang_cache/dsv4_pro_tp16}"
# eth1 is the TCP bootstrap interface. mlx5_0/mlx5_3 are the two RoCE rails.
NCCL_SOCKET_IFNAME="${NCCL_SOCKET_IFNAME:-eth1}"
NCCL_IB_HCA="${NCCL_IB_HCA:-mlx5_0,mlx5_3}"
NCCL_CROSS_NIC="${NCCL_CROSS_NIC:-1}"
NCCL_DEBUG="${NCCL_DEBUG:-WARN}"
# Keep the known working TP16 baseline. The quick map changes workload, not serving knobs.
MEM_FRACTION_STATIC="${MEM_FRACTION_STATIC:-0.9}"
CUDA_GRAPH_MAX_BS_DECODE="${CUDA_GRAPH_MAX_BS_DECODE:-64}"
MAX_RUNNING_REQUESTS="${MAX_RUNNING_REQUESTS:-256}"
SGLANG_EXTRA_ARGS="${SGLANG_EXTRA_ARGS:-}"
BENCH_DATASET_NAME="${BENCH_DATASET_NAME:-random}"
DATASET_PATH="${DATASET_PATH:-/data/yy/sskj/dataset/ShareGPT_V3_unfiltered_cleaned_split.json}"
SGLANG_BENCH_MODULE="${SGLANG_BENCH_MODULE:-sglang.benchmark.serving}"
RANDOM_RANGE_RATIO="${RANDOM_RANGE_RATIO:-1.0}"
REQUEST_RATE="${REQUEST_RATE:-10000}"
BENCH_EXTRA_ARGS="${BENCH_EXTRA_ARGS:-}"
# Quick runs use one request wave. Milestone repetitions are selected separately.
# Set NUM_REPETITIONS=3 for a milestone baseline after the quick run is stable.
NUM_REPETITIONS="${NUM_REPETITIONS:-1}"
PROMPT_MULTIPLIER_OVERRIDE="${PROMPT_MULTIPLIER_OVERRIDE:-}"
MIN_NUM_PROMPTS_OVERRIDE="${MIN_NUM_PROMPTS_OVERRIDE:-}"
WARMUP_REQUESTS_OVERRIDE="${WARMUP_REQUESTS_OVERRIDE:-}"
SCENARIO_TIMEOUT_S="${SCENARIO_TIMEOUT_S:-7200}"
MIXED_TIMEOUT_S="${MIXED_TIMEOUT_S:-7200}"
CASE_COOLDOWN_S="${CASE_COOLDOWN_S:-5}"
MIXED_INJECT_DELAY_S="${MIXED_INJECT_DELAY_S:-10}"
MIXED_BACKGROUND_PROMPTS="${MIXED_BACKGROUND_PROMPTS:-64}"
HEALTH_CHECK_RETRIES="${HEALTH_CHECK_RETRIES:-600}"
HEALTH_CHECK_INTERVAL_S="${HEALTH_CHECK_INTERVAL_S:-5}"
ALLOW_BUSY_GPU="${ALLOW_BUSY_GPU:-0}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "${SCRIPT_DIR}/../../.." && pwd)"
SCENARIO_FILE="${SCENARIO_FILE:-${SCRIPT_DIR}/quick_map_scenarios.tsv}"
RESULT_BASE="${RESULT_BASE:-${SCRIPT_DIR}/results}"
RUNTIME_BASE="${RUNTIME_BASE:-${SCRIPT_DIR}/runtime}"
DRY_RUN="${DRY_RUN:-0}"
RESUME="${RESUME:-1}"

708
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_results.py Executable file
View File

@ -0,0 +1,708 @@
#!/usr/bin/env python3
"""Structured metadata and summaries for the two-node SGLang quick map."""
from __future__ import annotations
import argparse
import csv
import json
import math
import re
import statistics
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any
SUMMARY_FIELDS = [
"run_id",
"suite",
"case_id",
"role",
"stage",
"repetition",
"isl",
"osl",
"concurrency",
"num_prompts",
"warmup_requests",
"status",
"error_type",
"exit_code",
"started_at",
"ended_at",
"elapsed_s",
"completed",
"failed",
"duration_s",
"actual_concurrency",
"peak_concurrent_requests",
"total_input_tokens",
"total_output_tokens",
"request_throughput",
"input_token_throughput",
"output_token_throughput",
"total_token_throughput",
"peak_output_token_throughput",
"e2e_mean_ms",
"e2e_p50_ms",
"e2e_p95_ms",
"e2e_p99_ms",
"ttft_mean_ms",
"ttft_p50_ms",
"ttft_p95_ms",
"ttft_p99_ms",
"tpot_mean_ms",
"tpot_p50_ms",
"tpot_p95_ms",
"tpot_p99_ms",
"itl_mean_ms",
"itl_p50_ms",
"itl_p95_ms",
"itl_p99_ms",
"bench_file",
"bench_log",
]
AGGREGATE_FIELDS = [
"suite",
"case_id",
"role",
"stage",
"isl",
"osl",
"concurrency",
"planned_repetitions",
"completed_repetitions",
"statuses",
"total_tps_mean",
"total_tps_cv_pct",
"output_tps_mean",
"output_tps_cv_pct",
"ttft_p95_mean_ms",
"tpot_p95_mean_ms",
"itl_p95_mean_ms",
"e2e_p95_mean_ms",
]
def utcish_now() -> str:
return datetime.now().astimezone().isoformat(timespec="seconds")
def read_json(path: Path) -> dict[str, Any]:
with path.open("r", encoding="utf-8") as handle:
return json.load(handle)
def write_json(path: Path, value: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as handle:
json.dump(value, handle, indent=2, ensure_ascii=False)
handle.write("\n")
def read_bench_output(path: Path) -> dict[str, Any] | None:
if not path.exists() or path.stat().st_size == 0:
return None
text = path.read_text(encoding="utf-8").strip()
if not text:
return None
try:
value = json.loads(text)
if isinstance(value, dict):
return value
except json.JSONDecodeError:
pass
for line in text.splitlines():
try:
value = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(value, dict):
return value
return None
def percentile_ms(values_s: list[float], percentile: float) -> float | None:
if not values_s:
return None
values = sorted(values_s)
if len(values) == 1:
return values[0] * 1000.0
rank = (percentile / 100.0) * (len(values) - 1)
lower = math.floor(rank)
upper = math.ceil(rank)
if lower == upper:
return values[lower] * 1000.0
fraction = rank - lower
return (values[lower] * (1.0 - fraction) + values[upper] * fraction) * 1000.0
def as_float(value: Any) -> float | None:
if value is None or value == "":
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def first_float(data: dict[str, Any], *keys: str) -> float | None:
for key in keys:
value = as_float(data.get(key))
if value is not None:
return value
return None
def finite_floats(values: Any) -> list[float]:
if not isinstance(values, list):
return []
result = []
for value in values:
parsed = as_float(value)
if parsed is not None and math.isfinite(parsed):
result.append(parsed)
return result
def flattened_itls(data: dict[str, Any]) -> list[float]:
result = []
for request_itls in data.get("itls", []):
if isinstance(request_itls, list):
result.extend(finite_floats(request_itls))
else:
value = as_float(request_itls)
if value is not None:
result.append(value)
return result
def per_request_tpots(data: dict[str, Any]) -> list[float]:
direct = finite_floats(data.get("tpots"))
if direct:
return direct
result = []
for request_itls in data.get("itls", []):
values = finite_floats(request_itls)
if values:
result.append(sum(values) / len(values))
return result
def per_request_e2e(data: dict[str, Any]) -> list[float]:
direct = finite_floats(data.get("request_latencies"))
if direct:
return direct
ttfts = finite_floats(data.get("ttfts"))
itls = data.get("itls", [])
result = []
for index, ttft in enumerate(ttfts):
request_itls = finite_floats(itls[index]) if index < len(itls) else []
result.append(ttft + sum(request_itls))
return result
def latency_stats(data: dict[str, Any], name: str) -> dict[str, float | None]:
if name == "e2e":
prefix = "e2e_latency"
fallback = per_request_e2e(data)
elif name == "ttft":
prefix = "ttft"
fallback = finite_floats(data.get("ttfts"))
elif name == "tpot":
prefix = "tpot"
fallback = per_request_tpots(data)
elif name == "itl":
prefix = "itl"
fallback = flattened_itls(data)
else:
raise ValueError(f"unsupported latency metric: {name}")
mean = first_float(data, f"mean_{prefix}_ms")
p50 = first_float(data, f"median_{prefix}_ms", f"p50_{prefix}_ms")
p95 = first_float(data, f"p95_{prefix}_ms")
p99 = first_float(data, f"p99_{prefix}_ms")
return {
"mean": mean if mean is not None else (statistics.fmean(fallback) * 1000.0 if fallback else None),
"p50": p50 if p50 is not None else percentile_ms(fallback, 50),
"p95": p95 if p95 is not None else percentile_ms(fallback, 95),
"p99": p99 if p99 is not None else percentile_ms(fallback, 99),
}
def compute_metrics(data: dict[str, Any]) -> dict[str, Any]:
completed = int(data.get("completed", data.get("successful_requests", 0)) or 0)
if data.get("failed") is not None:
failed = int(data["failed"] or 0)
else:
attempted = len(data.get("input_lens", []))
failed = max(0, attempted - completed)
input_tps = first_float(data, "input_throughput", "input_token_throughput") or 0.0
output_tps = first_float(data, "output_throughput", "output_token_throughput") or 0.0
total_tps = first_float(data, "total_throughput", "total_token_throughput")
if total_tps is None:
total_tps = input_tps + output_tps
metrics: dict[str, Any] = {
"completed": completed,
"failed": failed,
"duration_s": first_float(data, "duration", "benchmark_duration") or 0.0,
"actual_concurrency": first_float(data, "concurrency"),
"peak_concurrent_requests": first_float(data, "peak_concurrent_requests"),
"total_input_tokens": int(data.get("total_input_tokens", 0) or 0),
"total_output_tokens": int(data.get("total_output_tokens", data.get("total_generated_tokens", 0)) or 0),
"request_throughput": first_float(data, "request_throughput") or 0.0,
"input_token_throughput": input_tps,
"output_token_throughput": output_tps,
"total_token_throughput": total_tps,
"peak_output_token_throughput": first_float(
data,
"peak_output_throughput",
"peak_output_token_throughput",
),
}
for name in ("e2e", "ttft", "tpot", "itl"):
stats = latency_stats(data, name)
for statistic, value in stats.items():
metrics[f"{name}_{statistic}_ms"] = value
return metrics
def parse_scenarios(path: Path) -> list[dict[str, Any]]:
scenarios = []
seen = set()
with path.open("r", encoding="utf-8", newline="") as handle:
reader = csv.reader((line for line in handle if line.strip() and not line.lstrip().startswith("#")), delimiter="\t")
for line_number, fields in enumerate(reader, start=1):
if len(fields) != 9:
raise ValueError(f"{path}: data row {line_number} has {len(fields)} fields, expected 9")
case_id, stage, isl, osl, concurrency, multiplier, minimum, warmup, note = fields
if not re.fullmatch(r"[a-z0-9][a-z0-9_]*", case_id):
raise ValueError(f"{path}: invalid case_id {case_id!r}")
if case_id in seen:
raise ValueError(f"{path}: duplicate case_id {case_id!r}")
seen.add(case_id)
numeric = {
"isl": int(isl),
"osl": int(osl),
"concurrency": int(concurrency),
"prompt_multiplier": int(multiplier),
"min_prompts": int(minimum),
"warmup_requests": int(warmup),
}
if any(value <= 0 for key, value in numeric.items() if key != "warmup_requests"):
raise ValueError(f"{path}: {case_id} contains a non-positive value")
if numeric["warmup_requests"] < 0:
raise ValueError(f"{path}: {case_id} has negative warmup_requests")
scenarios.append({"case_id": case_id, "stage": stage, "note": note, **numeric})
if not scenarios:
raise ValueError(f"{path}: no scenarios")
return scenarios
def write_case(args: argparse.Namespace) -> None:
value = {
"run_id": args.run_id,
"suite": args.suite,
"case_id": args.case_id,
"role": args.role,
"stage": args.stage,
"repetition": args.repetition,
"isl": args.isl,
"osl": args.osl,
"concurrency": args.concurrency,
"num_prompts": args.num_prompts,
"warmup_requests": args.warmup_requests,
"status": args.status,
"error_type": args.error_type,
"exit_code": args.exit_code,
"started_at": args.started_at,
"ended_at": args.ended_at,
"elapsed_s": args.elapsed_s,
"bench_file": args.bench_file,
"bench_log": args.bench_log,
"note": args.note,
}
write_json(args.path, value)
def write_manifest(args: argparse.Namespace) -> None:
value: dict[str, Any] = {}
if args.path.exists():
value = read_json(args.path)
suites = set(value.get("suites", []))
suites.add(args.suite)
value.update(
{
"schema_version": 1,
"workflow_stage": "quick_performance_map",
"run_id": args.run_id,
"status": "RUNNING",
"started_at": value.get("started_at", utcish_now()),
"updated_at": utcish_now(),
"suites": sorted(suites),
"engine": "sglang",
"model_name": args.model_name,
"model_path": args.model_path,
"docker_image": args.docker_image,
"head_node": args.head_node,
"worker_node": args.worker_node,
"head_ip": args.head_ip,
"sglang_port": args.sglang_port,
"dist_init_port": args.dist_init_port,
"tp_size": args.tp_size,
"ep_size": args.ep_size,
"nnodes": args.nnodes,
"mem_fraction_static": args.mem_fraction_static,
"cuda_graph_max_bs_decode": args.cuda_graph_max_bs_decode,
"max_running_requests": args.max_running_requests,
"nccl_socket_ifname": args.nccl_socket_ifname,
"nccl_ib_hca": args.nccl_ib_hca,
"nccl_cross_nic": args.nccl_cross_nic,
"git_commit": args.git_commit,
"git_dirty": bool(args.git_dirty),
"scenario_file": args.scenario_file,
"notes": [
"The fixed quick map does not stop on SLO.",
"Profiler is disabled; these results are eligible for performance comparison.",
"Speculative decoding is not enabled.",
],
}
)
write_json(args.path, value)
def complete_manifest(path: Path, status: str) -> None:
if not path.exists():
return
value = read_json(path)
value["status"] = status
value["ended_at"] = utcish_now()
value["updated_at"] = value["ended_at"]
write_json(path, value)
def mark_case_failed(path: Path, error_type: str, note: str) -> None:
if not path.exists():
raise SystemExit(f"case metadata does not exist: {path}")
value = read_json(path)
value["status"] = "FAILED"
value["error_type"] = error_type
value["exit_code"] = 66
value["ended_at"] = utcish_now()
value["note"] = note
write_json(path, value)
def row_from_meta(meta_path: Path) -> dict[str, Any]:
meta = read_json(meta_path)
row = {field: meta.get(field) for field in SUMMARY_FIELDS}
bench_file = Path(meta.get("bench_file", ""))
if bench_file and not bench_file.is_absolute():
bench_file = (meta_path.parent / bench_file).resolve()
data = read_bench_output(bench_file) if bench_file else None
if data is not None:
row.update(compute_metrics(data))
return row
def cv_pct(values: list[float]) -> float | None:
if len(values) < 2:
return None
mean = statistics.fmean(values)
if mean == 0:
return None
return statistics.stdev(values) / mean * 100.0
def mean_or_none(values: list[float]) -> float | None:
return statistics.fmean(values) if values else None
def numeric_values(rows: list[dict[str, Any]], field: str) -> list[float]:
result = []
for row in rows:
value = as_float(row.get(field))
if value is not None:
result.append(value)
return result
def aggregate_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
groups: dict[tuple[Any, ...], list[dict[str, Any]]] = defaultdict(list)
for row in rows:
key = (
row.get("suite"),
row.get("case_id"),
row.get("role"),
row.get("stage"),
row.get("isl"),
row.get("osl"),
row.get("concurrency"),
)
groups[key].append(row)
aggregates = []
for key, group in sorted(groups.items(), key=lambda item: tuple(str(part) for part in item[0])):
completed = [row for row in group if row.get("status") == "COMPLETED" and row.get("completed") is not None]
total_tps = numeric_values(completed, "total_token_throughput")
output_tps = numeric_values(completed, "output_token_throughput")
aggregate = {
"suite": key[0],
"case_id": key[1],
"role": key[2],
"stage": key[3],
"isl": key[4],
"osl": key[5],
"concurrency": key[6],
"planned_repetitions": len(group),
"completed_repetitions": len(completed),
"statuses": ",".join(sorted({str(row.get("status", "")) for row in group})),
"total_tps_mean": mean_or_none(total_tps),
"total_tps_cv_pct": cv_pct(total_tps),
"output_tps_mean": mean_or_none(output_tps),
"output_tps_cv_pct": cv_pct(output_tps),
"ttft_p95_mean_ms": mean_or_none(numeric_values(completed, "ttft_p95_ms")),
"tpot_p95_mean_ms": mean_or_none(numeric_values(completed, "tpot_p95_ms")),
"itl_p95_mean_ms": mean_or_none(numeric_values(completed, "itl_p95_ms")),
"e2e_p95_mean_ms": mean_or_none(numeric_values(completed, "e2e_p95_ms")),
}
aggregates.append(aggregate)
return aggregates
def percentage_change(baseline: float | None, candidate: float | None) -> float | None:
if baseline is None or candidate is None or baseline == 0:
return None
return (candidate - baseline) / baseline * 100.0
def format_number(value: Any, digits: int = 2) -> str:
parsed = as_float(value)
return "-" if parsed is None else f"{parsed:.{digits}f}"
def write_report(path: Path, rows: list[dict[str, Any]], aggregates: list[dict[str, Any]]) -> None:
lines = [
"# DeepSeek-V4-Pro Pro6000D Two-Node SGLang Quick Map",
"",
"Profiler: disabled. Speculative decoding: disabled.",
"",
"## Aggregate results",
"",
"| Case | Suite / role | Stage | ISL | OSL | C | Reps | Total TPS | CV | Output TPS | TTFT P95 | TPOT P95 | E2E P95 | Status |",
"|---|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---|",
]
for item in aggregates:
role = item.get("role") or "-"
lines.append(
"| {case} | {suite} / {role} | {stage} | {isl} | {osl} | {concurrency} | "
"{done}/{planned} | {total_tps} | {total_cv}% | {output_tps} | {ttft} ms | "
"{tpot} ms | {e2e} ms | {statuses} |".format(
case=item["case_id"],
suite=item["suite"],
role=role,
stage=item["stage"],
isl=item["isl"],
osl=item["osl"],
concurrency=item["concurrency"],
done=item["completed_repetitions"],
planned=item["planned_repetitions"],
total_tps=format_number(item["total_tps_mean"]),
total_cv=format_number(item["total_tps_cv_pct"]),
output_tps=format_number(item["output_tps_mean"]),
ttft=format_number(item["ttft_p95_mean_ms"]),
tpot=format_number(item["tpot_p95_mean_ms"]),
e2e=format_number(item["e2e_p95_mean_ms"]),
statuses=item["statuses"],
)
)
by_case = {item["case_id"]: item for item in aggregates}
control = by_case.get("decode_control_1k_to_1k_c32")
mixed = by_case.get("decode_with_128k_prefill_1k_to_1k_c32")
if control and mixed:
lines.extend(
[
"",
"## Mixed-interference A/B",
"",
"| Metric | Control | With 128K prefill | Change |",
"|---|---:|---:|---:|",
]
)
for label, field in (
("Output TPS", "output_tps_mean"),
("TTFT P95 (ms)", "ttft_p95_mean_ms"),
("TPOT P95 (ms)", "tpot_p95_mean_ms"),
("E2E P95 (ms)", "e2e_p95_mean_ms"),
):
change = percentage_change(as_float(control.get(field)), as_float(mixed.get(field)))
lines.append(
f"| {label} | {format_number(control.get(field))} | {format_number(mixed.get(field))} | "
f"{format_number(change)}% |"
)
failures = [row for row in rows if row.get("status") != "COMPLETED"]
if failures:
lines.extend(
[
"",
"## Failed or incomplete cases",
"",
"| Case | Repetition | Status | Error | Exit code |",
"|---|---:|---|---|---:|",
]
)
for row in failures:
lines.append(
f"| {row.get('case_id')} | {row.get('repetition')} | {row.get('status')} | "
f"{row.get('error_type') or '-'} | {row.get('exit_code')} |"
)
lines.extend(
[
"",
"The fixed map is descriptive and never stops on SLO. With one repetition, CV is intentionally unavailable.",
"",
]
)
path.write_text("\n".join(lines), encoding="utf-8")
def write_csv(path: Path, rows: list[dict[str, Any]], fields: list[str]) -> None:
with path.open("w", encoding="utf-8", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
writer.writeheader()
writer.writerows(rows)
def summarize(result_dir: Path) -> None:
meta_paths = sorted(result_dir.glob("**/meta.json"))
rows = [row_from_meta(path) for path in meta_paths]
aggregates = aggregate_rows(rows)
result_dir.mkdir(parents=True, exist_ok=True)
write_csv(result_dir / "summary.csv", rows, SUMMARY_FIELDS)
with (result_dir / "summary.jsonl").open("w", encoding="utf-8") as handle:
for row in rows:
handle.write(json.dumps(row, ensure_ascii=False) + "\n")
write_csv(result_dir / "aggregate.csv", aggregates, AGGREGATE_FIELDS)
write_report(result_dir / "report.md", rows, aggregates)
print(f"summarized {len(rows)} case runs into {result_dir}")
def add_case_arguments(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--path", type=Path, required=True)
parser.add_argument("--run-id", required=True)
parser.add_argument("--suite", required=True)
parser.add_argument("--case-id", required=True)
parser.add_argument("--role", default="")
parser.add_argument("--stage", required=True)
parser.add_argument("--repetition", type=int, required=True)
parser.add_argument("--isl", type=int, required=True)
parser.add_argument("--osl", type=int, required=True)
parser.add_argument("--concurrency", type=int, required=True)
parser.add_argument("--num-prompts", type=int, required=True)
parser.add_argument("--warmup-requests", type=int, required=True)
parser.add_argument("--status", required=True)
parser.add_argument("--error-type", default="")
parser.add_argument("--exit-code", type=int, required=True)
parser.add_argument("--started-at", required=True)
parser.add_argument("--ended-at", required=True)
parser.add_argument("--elapsed-s", type=float, required=True)
parser.add_argument("--bench-file", required=True)
parser.add_argument("--bench-log", required=True)
parser.add_argument("--note", default="")
def add_manifest_arguments(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--path", type=Path, required=True)
parser.add_argument("--run-id", required=True)
parser.add_argument("--suite", required=True)
parser.add_argument("--model-name", required=True)
parser.add_argument("--model-path", required=True)
parser.add_argument("--docker-image", required=True)
parser.add_argument("--head-node", required=True)
parser.add_argument("--worker-node", required=True)
parser.add_argument("--head-ip", required=True)
parser.add_argument("--sglang-port", type=int, required=True)
parser.add_argument("--dist-init-port", type=int, required=True)
parser.add_argument("--tp-size", type=int, required=True)
parser.add_argument("--ep-size", type=int, required=True)
parser.add_argument("--nnodes", type=int, required=True)
parser.add_argument("--mem-fraction-static", type=float, required=True)
parser.add_argument("--cuda-graph-max-bs-decode", type=int, required=True)
parser.add_argument("--max-running-requests", type=int, required=True)
parser.add_argument("--nccl-socket-ifname", required=True)
parser.add_argument("--nccl-ib-hca", required=True)
parser.add_argument("--nccl-cross-nic", required=True)
parser.add_argument("--git-commit", required=True)
parser.add_argument("--git-dirty", type=int, required=True)
parser.add_argument("--scenario-file", required=True)
def main() -> None:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="command", required=True)
validate_parser = subparsers.add_parser("validate-scenarios")
validate_parser.add_argument("path", type=Path)
validate_bench_parser = subparsers.add_parser("validate-bench")
validate_bench_parser.add_argument("path", type=Path)
case_parser = subparsers.add_parser("write-case")
add_case_arguments(case_parser)
manifest_parser = subparsers.add_parser("write-manifest")
add_manifest_arguments(manifest_parser)
complete_parser = subparsers.add_parser("complete-manifest")
complete_parser.add_argument("path", type=Path)
complete_parser.add_argument("--status", required=True)
mark_failed_parser = subparsers.add_parser("mark-case-failed")
mark_failed_parser.add_argument("path", type=Path)
mark_failed_parser.add_argument("--error-type", required=True)
mark_failed_parser.add_argument("--note", required=True)
summarize_parser = subparsers.add_parser("summarize")
summarize_parser.add_argument("result_dir", type=Path)
args = parser.parse_args()
if args.command == "validate-scenarios":
scenarios = parse_scenarios(args.path)
for scenario in scenarios:
prompts = max(
scenario["min_prompts"],
scenario["concurrency"] * scenario["prompt_multiplier"],
)
print(
f"{scenario['case_id']}: stage={scenario['stage']} "
f"isl={scenario['isl']} osl={scenario['osl']} "
f"c={scenario['concurrency']} prompts={prompts} "
f"warmup={scenario['warmup_requests']}"
)
elif args.command == "validate-bench":
value = read_bench_output(args.path)
if value is None:
raise SystemExit(f"no JSON object found in {args.path}")
print(json.dumps(compute_metrics(value), ensure_ascii=False))
elif args.command == "write-case":
write_case(args)
elif args.command == "write-manifest":
write_manifest(args)
elif args.command == "complete-manifest":
complete_manifest(args.path, args.status)
elif args.command == "mark-case-failed":
mark_case_failed(args.path, args.error_type, args.note)
elif args.command == "summarize":
summarize(args.result_dir)
if __name__ == "__main__":
main()

10
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_scenarios.tsv Normal file
View File

@ -0,0 +1,10 @@
# case_id stage isl osl concurrency prompt_multiplier min_prompts warmup_requests note
short_prefill_latency_1k_c1 prefill_latency 1024 1 1 1 1 1 Short-prefill latency floor
mid_prefill_latency_32k_c1 prefill_latency 32768 1 1 1 1 0 Mid-context prefill latency
long_prefill_latency_128k_c1 prefill_latency 131072 1 1 1 1 0 Long-context prefill latency
mid_prefill_throughput_32k_c16 prefill_throughput 32768 1 16 1 1 0 Mid-context prefill throughput
decode_latency_1k_to_1k_c1 decode_latency 1024 1024 1 1 1 1 Single-request decode latency
decode_throughput_1k_to_1k_c16 decode_throughput 1024 1024 16 1 1 1 Decode throughput at C16
decode_throughput_1k_to_1k_c32 decode_throughput 1024 1024 32 1 1 1 Decode throughput at C32
decode_throughput_1k_to_1k_c64 decode_throughput 1024 1024 64 1 1 1 Decode throughput at C64
balanced_32k_to_1k_c8 balanced 32768 1024 8 1 1 0 Balanced prefill and decode pressure
1 # case_id stage isl osl concurrency prompt_multiplier min_prompts warmup_requests note
2 short_prefill_latency_1k_c1 prefill_latency 1024 1 1 1 1 1 Short-prefill latency floor
3 mid_prefill_latency_32k_c1 prefill_latency 32768 1 1 1 1 0 Mid-context prefill latency
4 long_prefill_latency_128k_c1 prefill_latency 131072 1 1 1 1 0 Long-context prefill latency
5 mid_prefill_throughput_32k_c16 prefill_throughput 32768 1 16 1 1 0 Mid-context prefill throughput
6 decode_latency_1k_to_1k_c1 decode_latency 1024 1024 1 1 1 1 Single-request decode latency
7 decode_throughput_1k_to_1k_c16 decode_throughput 1024 1024 16 1 1 1 Decode throughput at C16
8 decode_throughput_1k_to_1k_c32 decode_throughput 1024 1024 32 1 1 1 Decode throughput at C32
9 decode_throughput_1k_to_1k_c64 decode_throughput 1024 1024 64 1 1 1 Decode throughput at C64
10 balanced_32k_to_1k_c8 balanced 32768 1024 8 1 1 0 Balanced prefill and decode pressure

779
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/run_quick_map.sh Executable file
View File

@ -0,0 +1,779 @@
#!/usr/bin/env bash
# Run the two-node DeepSeek-V4-Pro SGLang TP16 quick performance map.
set -Eeuo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/config.env"
ACTION="${1:-all}"
RUN_ID="${RUN_ID:-dsv4pro-pro6000d-2node-sglang-quick-$(date +%Y%m%d-%H%M%S)}"
RESULT_DIR="${RESULT_BASE}/${RUN_ID}"
RESULT_TOOL="${SCRIPT_DIR}/quick_map_results.py"
HEAD_CONTAINER="${EXPERIMENT}_head"
WORKER_CONTAINER="${EXPERIMENT}_worker"
SERVER_ARTIFACT_DIR="${SERVER_ARTIFACT_DIR:-${RUNTIME_BASE}/server}"
LAST_CASE_STATUS=""
LAST_CASE_ERROR=""
FIXED_FAILURES=0
MIXED_FAILURES=0
CLEANUP_ON_EXIT=0
RUN_LOG_ACTIVE=0
declare -a BENCH_CMD=()
declare -a DOCKER_CMD=()
log() {
printf '[%s] %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$*"
}
iso_now() {
date '+%Y-%m-%dT%H:%M:%S%z'
}
print_command() {
printf '%q ' "$@"
printf '\n'
}
node_is_local() {
local node="$1"
local local_ips
[[ "${node}" == "localhost" || "${node}" == "127.0.0.1" ]] && return 0
local_ips="$(hostname -I 2>/dev/null || true)"
[[ " ${local_ips} " == *" ${node} "* ]]
}
run_on_node() {
local node="$1"
local command="$2"
if node_is_local "${node}"; then
bash -lc "${command}"
else
ssh -o BatchMode=yes -o StrictHostKeyChecking=no "${node}" "${command}"
fi
}
service_is_healthy() {
curl --fail --silent --show-error --max-time 5 \
"http://${HEAD_IP}:${SGLANG_PORT}/health" >/dev/null 2>&1
}
require_healthy_service() {
if ! service_is_healthy; then
log "ERROR: SGLang is not healthy at ${HEAD_IP}:${SGLANG_PORT}"
log "Start it with: bash ${SCRIPT_DIR}/run_quick_map.sh start"
return 1
fi
}
remote_has_gpu_processes() {
local node="$1"
local output
output="$(run_on_node "${node}" \
"nvidia-smi --query-compute-apps=pid --format=csv,noheader 2>/dev/null" || true)"
[[ -n "${output//[[:space:]]/}" ]]
}
preflight_service_node() {
local node="$1"
run_on_node "${node}" \
"test -d '${MODEL_PATH}' && command -v docker >/dev/null && command -v nvidia-smi >/dev/null && docker image inspect '${DOCKER_IMAGE}' >/dev/null"
if [[ "${ALLOW_BUSY_GPU}" != "1" ]] && remote_has_gpu_processes "${node}"; then
log "ERROR: ${node} has active GPU compute processes"
log "This experiment will not evict another workload."
return 1
fi
}
preflight_bench_client() {
command -v docker >/dev/null || {
log "ERROR: docker is not available on the benchmark host"
return 1
}
[[ -d "${MODEL_PATH}" ]] || {
log "ERROR: model directory does not exist: ${MODEL_PATH}"
return 1
}
[[ -f "${DATASET_PATH}" ]] || {
log "ERROR: benchmark dataset does not exist: ${DATASET_PATH}"
return 1
}
docker image inspect "${DOCKER_CLIENT_IMAGE}" >/dev/null 2>&1 || {
log "ERROR: benchmark image is not available locally: ${DOCKER_CLIENT_IMAGE}"
return 1
}
}
build_server_command() {
local node_rank="$1"
local container_name="$2"
DOCKER_CMD=(
docker run -d
--name "${container_name}"
--gpus all
--network host
--ipc host
--shm-size 20g
--ulimit memlock=-1
--ulimit stack=67108864
-v "${MODEL_PATH}:${MODEL_PATH}:ro"
-v "${SGLANG_CACHE_DIR}:/root/.cache"
-e "CUDA_VISIBLE_DEVICES=${CUDA_VISIBLE_DEVICES}"
-e PYTHONUNBUFFERED=1
-e HF_HUB_OFFLINE=1
-e TRANSFORMERS_OFFLINE=1
-e PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True
-e "NCCL_SOCKET_IFNAME=${NCCL_SOCKET_IFNAME}"
-e "NCCL_IB_HCA=${NCCL_IB_HCA}"
-e "NCCL_CROSS_NIC=${NCCL_CROSS_NIC}"
-e "NCCL_DEBUG=${NCCL_DEBUG}"
-e SGLANG_SHARED_EXPERT_TP1=1
--entrypoint python3
"${DOCKER_IMAGE}"
-m sglang.launch_server
--model-path "${MODEL_PATH}"
--tp-size "${TP_SIZE}"
--ep-size "${EP_SIZE}"
--nnodes "${NNODES}"
--node-rank "${node_rank}"
--dist-init-addr "${HEAD_IP}:${DIST_INIT_PORT}"
--trust-remote-code
--host 0.0.0.0
--port "${SGLANG_PORT}"
--mem-fraction-static "${MEM_FRACTION_STATIC}"
--cuda-graph-max-bs-decode "${CUDA_GRAPH_MAX_BS_DECODE}"
--max-running-requests "${MAX_RUNNING_REQUESTS}"
)
if [[ -n "${SGLANG_EXTRA_ARGS}" ]]; then
local -a extra_args=()
read -r -a extra_args <<< "${SGLANG_EXTRA_ARGS}"
DOCKER_CMD+=("${extra_args[@]}")
fi
}
start_service_node() {
local node="$1"
local node_rank="$2"
local container_name="$3"
local role="$4"
build_server_command "${node_rank}" "${container_name}"
print_command "${DOCKER_CMD[@]}" > "${SERVER_ARTIFACT_DIR}/${role}_server_cmd.txt"
log "Starting ${role} node=${node} rank=${node_rank} container=${container_name}"
run_on_node "${node}" "mkdir -p '${SGLANG_CACHE_DIR}'"
run_on_node "${node}" "docker rm -f '${container_name}' >/dev/null 2>&1 || true"
local command
command="$(print_command "${DOCKER_CMD[@]}")"
run_on_node "${node}" "${command}" \
> "${SERVER_ARTIFACT_DIR}/${role}_container_id.txt"
}
start_service() {
build_server_command 1 "${WORKER_CONTAINER}"
if [[ "${DRY_RUN}" == "1" ]]; then
printf '[DRY] worker (%s): ' "${WORKER_NODE}"
print_command "${DOCKER_CMD[@]}"
build_server_command 0 "${HEAD_CONTAINER}"
printf '[DRY] head (%s): ' "${HEAD_NODE}"
print_command "${DOCKER_CMD[@]}"
return 0
fi
mkdir -p "${SERVER_ARTIFACT_DIR}"
preflight_service_node "${HEAD_NODE}"
preflight_service_node "${WORKER_NODE}"
start_service_node "${WORKER_NODE}" 1 "${WORKER_CONTAINER}" "worker"
sleep 5
start_service_node "${HEAD_NODE}" 0 "${HEAD_CONTAINER}" "head"
log "Waiting for SGLang health at ${HEAD_IP}:${SGLANG_PORT}"
local attempt
for (( attempt=1; attempt<=HEALTH_CHECK_RETRIES; attempt++ )); do
if service_is_healthy; then
log "SGLang is healthy after ${attempt} checks"
return 0
fi
if ! run_on_node "${HEAD_NODE}" \
"docker inspect -f '{{.State.Running}}' '${HEAD_CONTAINER}' 2>/dev/null" \
| grep -qx true; then
log "ERROR: head container exited"
run_on_node "${HEAD_NODE}" \
"docker logs --tail 100 '${HEAD_CONTAINER}'" || true
return 1
fi
if ! run_on_node "${WORKER_NODE}" \
"docker inspect -f '{{.State.Running}}' '${WORKER_CONTAINER}' 2>/dev/null" \
| grep -qx true; then
log "ERROR: worker container exited"
run_on_node "${WORKER_NODE}" \
"docker logs --tail 100 '${WORKER_CONTAINER}'" || true
return 1
fi
if (( attempt % 12 == 0 )); then
log "Still starting: checks=${attempt}/${HEALTH_CHECK_RETRIES}"
fi
sleep "${HEALTH_CHECK_INTERVAL_S}"
done
log "ERROR: SGLang health timeout"
return 1
}
stop_service_node() {
local node="$1"
local container_name="$2"
local role="$3"
if [[ "${DRY_RUN}" == "1" ]]; then
log "[DRY] ${node}: docker rm -f ${container_name}"
return 0
fi
mkdir -p "${SERVER_ARTIFACT_DIR}"
run_on_node "${node}" \
"docker inspect '${container_name}' 2>/dev/null" \
> "${SERVER_ARTIFACT_DIR}/${role}_inspect.json" 2>/dev/null || true
run_on_node "${node}" \
"docker logs '${container_name}' 2>&1" \
> "${SERVER_ARTIFACT_DIR}/${role}_server.log" 2>&1 || true
run_on_node "${node}" \
"docker rm -f '${container_name}' >/dev/null 2>&1" || true
log "Stopped ${role} container=${container_name} on ${node}"
}
stop_service() {
stop_service_node "${HEAD_NODE}" "${HEAD_CONTAINER}" "head"
stop_service_node "${WORKER_NODE}" "${WORKER_CONTAINER}" "worker"
}
prepare_bench_command() {
local output_file="$1"
local isl="$2"
local osl="$3"
local concurrency="$4"
local num_prompts="$5"
local warmup_requests="$6"
local seed="${7:-42}"
local flush_cache="${8:-1}"
BENCH_CMD=(
docker run --rm
--network host
-v "${MODEL_PATH}:${MODEL_PATH}:ro"
-v "${DATASET_PATH}:${DATASET_PATH}:ro"
-v "$(dirname "$output_file"):$(dirname "$output_file")"
-e PYTHONUNBUFFERED=1
-e HF_HUB_OFFLINE=1
-e TRANSFORMERS_OFFLINE=1
--entrypoint python3
"${DOCKER_CLIENT_IMAGE}"
-m "${SGLANG_BENCH_MODULE}"
--backend sglang
--host "${HEAD_IP}"
--port "${SGLANG_PORT}"
--dataset-name "${BENCH_DATASET_NAME}"
--dataset-path "${DATASET_PATH}"
--random-input-len "${isl}"
--random-output-len "${osl}"
--random-range-ratio "${RANDOM_RANGE_RATIO}"
--num-prompts "${num_prompts}"
--max-concurrency "${concurrency}"
--request-rate "${REQUEST_RATE}"
--output-file "${output_file}"
--output-details
--disable-tqdm
--warmup-requests "${warmup_requests}"
--seed "${seed}"
)
if [[ "${flush_cache}" == "1" ]]; then
BENCH_CMD+=(--flush-cache)
fi
if [[ -n "${BENCH_EXTRA_ARGS}" ]]; then
local -a extra_args=()
read -r -a extra_args <<< "${BENCH_EXTRA_ARGS}"
BENCH_CMD+=("${extra_args[@]}")
fi
}
write_case_meta() {
local meta_path="$1"
local suite="$2"
local case_id="$3"
local role="$4"
local stage="$5"
local repetition="$6"
local isl="$7"
local osl="$8"
local concurrency="$9"
local num_prompts="${10}"
local warmup_requests="${11}"
local status="${12}"
local error_type="${13}"
local exit_code="${14}"
local started_at="${15}"
local ended_at="${16}"
local elapsed_s="${17}"
local bench_file="${18}"
local bench_log="${19}"
local note="${20}"
python3 "${RESULT_TOOL}" write-case \
--path "${meta_path}" \
--run-id "${RUN_ID}" \
--suite "${suite}" \
--case-id "${case_id}" \
--role "${role}" \
--stage "${stage}" \
--repetition "${repetition}" \
--isl "${isl}" \
--osl "${osl}" \
--concurrency "${concurrency}" \
--num-prompts "${num_prompts}" \
--warmup-requests "${warmup_requests}" \
--status "${status}" \
--error-type "${error_type}" \
--exit-code "${exit_code}" \
--started-at "${started_at}" \
--ended-at "${ended_at}" \
--elapsed-s "${elapsed_s}" \
--bench-file "${bench_file}" \
--bench-log "${bench_log}" \
--note "${note}"
}
case_already_completed() {
local meta_path="$1"
local bench_file="$2"
[[ "${RESUME}" == "1" ]] || return 1
[[ -s "${meta_path}" && -s "${bench_file}" ]] || return 1
python3 -c '
import json
import sys
with open(sys.argv[1], encoding="utf-8") as handle:
meta = json.load(handle)
raise SystemExit(0 if meta.get("status") == "COMPLETED" else 1)
' "${meta_path}" || return 1
python3 "${RESULT_TOOL}" validate-bench "${bench_file}" >/dev/null 2>&1
}
detect_error_type() {
local rc="$1"
local log_file="$2"
if (( rc == 124 )); then
printf 'TIMEOUT\n'
elif grep -Eqi \
'CUDA out of memory|torch\.OutOfMemoryError|OutOfMemory|out of memory|RESOURCE_EXHAUSTED|Failed to allocate memory' \
"${log_file}" 2>/dev/null; then
printf 'OOM\n'
elif ! service_is_healthy; then
printf 'ENGINE_UNHEALTHY\n'
elif grep -Eqi 'ClientPayloadError|TransferEncodingError|connection refused|Server disconnected' \
"${log_file}" 2>/dev/null; then
printf 'TRANSPORT_ERROR\n'
else
printf 'BENCH_FAILED\n'
fi
}
run_bench_case() {
local suite="$1"
local case_id="$2"
local role="$3"
local stage="$4"
local repetition="$5"
local isl="$6"
local osl="$7"
local concurrency="$8"
local num_prompts="$9"
local warmup_requests="${10}"
local timeout_s="${11}"
local note="${12}"
local seed="${13:-42}"
local flush_cache="${14:-1}"
local case_dir="${RESULT_DIR}/cases/${case_id}/rep${repetition}"
local bench_file="${case_dir}/bench.jsonl"
local bench_log="${case_dir}/bench.log"
local meta_path="${case_dir}/meta.json"
local command_file="${case_dir}/bench_cmd.txt"
if case_already_completed "${meta_path}" "${bench_file}"; then
log "SKIP completed case=${case_id} rep=${repetition}"
LAST_CASE_STATUS="COMPLETED"
LAST_CASE_ERROR=""
return 0
fi
prepare_bench_command \
"${bench_file}" "${isl}" "${osl}" "${concurrency}" \
"${num_prompts}" "${warmup_requests}" "${seed}" "${flush_cache}"
if [[ "${DRY_RUN}" == "1" ]]; then
printf '[DRY] case=%s rep=%s stage=%s isl=%s osl=%s c=%s prompts=%s warmup=%s\n' \
"${case_id}" "${repetition}" "${stage}" "${isl}" "${osl}" \
"${concurrency}" "${num_prompts}" "${warmup_requests}"
printf '[DRY] '
print_command timeout --signal=TERM --kill-after=30s "${timeout_s}s" "${BENCH_CMD[@]}"
LAST_CASE_STATUS="DRY_RUN"
LAST_CASE_ERROR=""
return 0
fi
mkdir -p "${case_dir}"
print_command timeout --signal=TERM --kill-after=30s "${timeout_s}s" \
"${BENCH_CMD[@]}" > "${command_file}"
local started_at start_epoch ended_at elapsed_s rc status error_type
started_at="$(iso_now)"
start_epoch="$(date +%s)"
log "START case=${case_id} rep=${repetition} isl=${isl} osl=${osl} c=${concurrency}"
set +e
timeout --signal=TERM --kill-after=30s "${timeout_s}s" \
"${BENCH_CMD[@]}" > "${bench_log}" 2>&1
rc=$?
set -e
ended_at="$(iso_now)"
elapsed_s="$(( $(date +%s) - start_epoch ))"
status="COMPLETED"
error_type=""
if (( rc != 0 )); then
status="FAILED"
error_type="$(detect_error_type "${rc}" "${bench_log}")"
elif ! python3 "${RESULT_TOOL}" validate-bench "${bench_file}" >/dev/null 2>&1; then
status="FAILED"
error_type="INVALID_BENCH_OUTPUT"
rc=65
fi
write_case_meta \
"${meta_path}" "${suite}" "${case_id}" "${role}" "${stage}" \
"${repetition}" "${isl}" "${osl}" "${concurrency}" "${num_prompts}" \
"${warmup_requests}" "${status}" "${error_type}" "${rc}" "${started_at}" \
"${ended_at}" "${elapsed_s}" "${bench_file}" "${bench_log}" "${note}"
LAST_CASE_STATUS="${status}"
LAST_CASE_ERROR="${error_type}"
if [[ "${status}" == "COMPLETED" ]]; then
log "DONE case=${case_id} rep=${repetition} elapsed=${elapsed_s}s"
return 0
fi
log "FAIL case=${case_id} rep=${repetition} error=${error_type} rc=${rc}"
return 1
}
mark_case_failed() {
local meta_path="$1"
local error_type="$2"
local note="$3"
python3 "${RESULT_TOOL}" mark-case-failed \
"${meta_path}" --error-type "${error_type}" --note "${note}"
}
write_run_manifest() {
local suite="$1"
local git_commit git_dirty
git_commit="$(git -C "${REPO_ROOT}" rev-parse HEAD 2>/dev/null || printf 'unknown')"
git_dirty=0
[[ -n "$(git -C "${REPO_ROOT}" status --porcelain 2>/dev/null || true)" ]] && git_dirty=1
python3 "${RESULT_TOOL}" write-manifest \
--path "${RESULT_DIR}/run_manifest.json" \
--run-id "${RUN_ID}" \
--suite "${suite}" \
--model-name "${MODEL_NAME}" \
--model-path "${MODEL_PATH}" \
--docker-image "${DOCKER_IMAGE}" \
--head-node "${HEAD_NODE}" \
--worker-node "${WORKER_NODE}" \
--head-ip "${HEAD_IP}" \
--sglang-port "${SGLANG_PORT}" \
--dist-init-port "${DIST_INIT_PORT}" \
--tp-size "${TP_SIZE}" \
--ep-size "${EP_SIZE}" \
--nnodes "${NNODES}" \
--mem-fraction-static "${MEM_FRACTION_STATIC}" \
--cuda-graph-max-bs-decode "${CUDA_GRAPH_MAX_BS_DECODE}" \
--max-running-requests "${MAX_RUNNING_REQUESTS}" \
--nccl-socket-ifname "${NCCL_SOCKET_IFNAME}" \
--nccl-ib-hca "${NCCL_IB_HCA}" \
--nccl-cross-nic "${NCCL_CROSS_NIC}" \
--git-commit "${git_commit}" \
--git-dirty "${git_dirty}" \
--scenario-file "${SCENARIO_FILE}"
}
summarize_results() {
python3 "${RESULT_TOOL}" summarize "${RESULT_DIR}"
}
complete_manifest() {
local status="$1"
python3 "${RESULT_TOOL}" complete-manifest \
"${RESULT_DIR}/run_manifest.json" --status "${status}"
}
enable_result_logging() {
[[ "${DRY_RUN}" == "1" || "${RUN_LOG_ACTIVE}" == "1" ]] && return 0
mkdir -p "${RESULT_DIR}"
exec > >(tee -a "${RESULT_DIR}/run.log") 2>&1
RUN_LOG_ACTIVE=1
}
run_fixed_suite() {
log "Validating scenario file: ${SCENARIO_FILE}"
python3 "${RESULT_TOOL}" validate-scenarios "${SCENARIO_FILE}"
local completed=0 total=0
local rep case_id stage isl osl concurrency multiplier minimum warmup note
for (( rep=1; rep<=NUM_REPETITIONS; rep++ )); do
while IFS=$'\t' read -r case_id stage isl osl concurrency multiplier minimum warmup note; do
[[ -z "${case_id}" || "${case_id}" == \#* ]] && continue
[[ -n "${PROMPT_MULTIPLIER_OVERRIDE}" ]] && multiplier="${PROMPT_MULTIPLIER_OVERRIDE}"
[[ -n "${MIN_NUM_PROMPTS_OVERRIDE}" ]] && minimum="${MIN_NUM_PROMPTS_OVERRIDE}"
[[ -n "${WARMUP_REQUESTS_OVERRIDE}" ]] && warmup="${WARMUP_REQUESTS_OVERRIDE}"
local num_prompts=$(( concurrency * multiplier ))
(( num_prompts < minimum )) && num_prompts="${minimum}"
((total+=1))
if run_bench_case \
"fixed" "${case_id}" "" "${stage}" "${rep}" "${isl}" "${osl}" \
"${concurrency}" "${num_prompts}" "${warmup}" \
"${SCENARIO_TIMEOUT_S}" "${note}"; then
((completed+=1))
else
((FIXED_FAILURES+=1))
if [[ "${LAST_CASE_ERROR}" == "ENGINE_UNHEALTHY" ]]; then
log "FATAL: service became unhealthy; stopping fixed suite"
return 2
fi
fi
[[ "${DRY_RUN}" == "1" ]] || sleep "${CASE_COOLDOWN_S}"
done < "${SCENARIO_FILE}"
done
log "Fixed quick map complete: completed=${completed}/${total} failed=${FIXED_FAILURES}"
}
wait_for_bench_main() {
local bench_log="$1"
local bench_pid="$2"
local timeout_s="$3"
local waited=0
while (( waited < timeout_s )); do
if grep -Fq "Starting main benchmark run" "${bench_log}" 2>/dev/null; then
return 0
fi
kill -0 "${bench_pid}" 2>/dev/null || return 1
sleep 1
((waited+=1))
done
return 1
}
run_mixed_repetition() {
local rep="$1"
local control_id="decode_control_1k_to_1k_c32"
local mixed_id="decode_with_128k_prefill_1k_to_1k_c32"
local injection_id="long_prefill_injection_128k_to_1_c1"
run_bench_case \
"mixed" "${control_id}" "control" "mixed_interference" "${rep}" \
1024 1024 32 "${MIXED_BACKGROUND_PROMPTS}" 1 "${MIXED_TIMEOUT_S}" \
"Decode-only control with the same request count as the mixed background" \
4201 1 \
|| return 1
if [[ "${DRY_RUN}" == "1" ]]; then
run_bench_case \
"mixed" "${mixed_id}" "decode_background" "mixed_interference" \
"${rep}" 1024 1024 32 "${MIXED_BACKGROUND_PROMPTS}" 0 \
"${MIXED_TIMEOUT_S}" \
"Decode background while one 128K prefill request is injected" \
4201 1
run_bench_case \
"mixed" "${injection_id}" "prefill_injection" "mixed_interference" \
"${rep}" 131072 1 1 1 0 "${MIXED_TIMEOUT_S}" \
"Single long prefill injected into the active decode background" \
4202 0
return 0
fi
log "START mixed background rep=${rep}; injection delay=${MIXED_INJECT_DELAY_S}s"
(
run_bench_case \
"mixed" "${mixed_id}" "decode_background" "mixed_interference" \
"${rep}" 1024 1024 32 "${MIXED_BACKGROUND_PROMPTS}" 0 \
"${MIXED_TIMEOUT_S}" \
"Decode background while one 128K prefill request is injected" \
4201 1
) &
local background_pid=$!
local background_log="${RESULT_DIR}/cases/${mixed_id}/rep${rep}/bench.log"
if ! wait_for_bench_main "${background_log}" "${background_pid}" 180; then
log "ERROR: decode background did not enter its measured run within 180s"
kill "${background_pid}" 2>/dev/null || true
wait "${background_pid}" 2>/dev/null || true
return 1
fi
sleep "${MIXED_INJECT_DELAY_S}"
if ! kill -0 "${background_pid}" 2>/dev/null; then
local early_background_rc=0
wait "${background_pid}" || early_background_rc=$?
if (( early_background_rc == 0 )); then
mark_case_failed \
"${RESULT_DIR}/cases/${mixed_id}/rep${rep}/meta.json" \
"BACKGROUND_FINISHED_BEFORE_INJECTION" \
"Decode background completed before the configured injection delay"
log "ERROR: decode background finished before long-prefill injection"
else
log "ERROR: decode background failed before long-prefill injection"
fi
return 1
fi
local injection_rc=0
run_bench_case \
"mixed" "${injection_id}" "prefill_injection" "mixed_interference" \
"${rep}" 131072 1 1 1 0 "${MIXED_TIMEOUT_S}" \
"Single long prefill injected into the active decode background" \
4202 0 \
|| injection_rc=$?
local background_rc=0
wait "${background_pid}" || background_rc=$?
if (( injection_rc != 0 || background_rc != 0 )); then
log "Mixed repetition failed: background_rc=${background_rc} injection_rc=${injection_rc}"
return 1
fi
}
run_mixed_suite() {
local rep
for (( rep=1; rep<=NUM_REPETITIONS; rep++ )); do
if ! run_mixed_repetition "${rep}"; then
((MIXED_FAILURES+=1))
if [[ "${DRY_RUN}" != "1" ]] && ! service_is_healthy; then
log "FATAL: service became unhealthy during mixed test"
return 2
fi
fi
[[ "${DRY_RUN}" == "1" ]] || sleep "${CASE_COOLDOWN_S}"
done
log "Mixed-interference A/B complete: failed_repetitions=${MIXED_FAILURES}"
}
cleanup() {
if [[ "${CLEANUP_ON_EXIT}" == "1" ]]; then
stop_service || true
fi
}
run_standalone_suite() {
local suite="$1"
if [[ "${DRY_RUN}" != "1" ]]; then
preflight_bench_client
require_healthy_service
enable_result_logging
write_run_manifest "${suite}"
fi
local suite_rc=0
if [[ "${suite}" == "fixed" ]]; then
run_fixed_suite || suite_rc=$?
else
run_mixed_suite || suite_rc=$?
fi
summarize_results
local status="COMPLETED"
if (( suite_rc == 2 )); then
status="ABORTED_ENGINE_UNHEALTHY"
elif (( FIXED_FAILURES + MIXED_FAILURES > 0 )); then
status="COMPLETED_WITH_FAILURES"
fi
complete_manifest "${status}"
log "Results: ${RESULT_DIR}"
return "${suite_rc}"
}
run_all() {
SERVER_ARTIFACT_DIR="${RESULT_DIR}/server"
if [[ "${DRY_RUN}" != "1" ]]; then
preflight_bench_client
enable_result_logging
write_run_manifest "fixed"
write_run_manifest "mixed"
fi
CLEANUP_ON_EXIT=1
trap cleanup EXIT INT TERM
if ! start_service; then
[[ "${DRY_RUN}" == "1" ]] || complete_manifest "FAILED_SERVICE_START"
return 1
fi
local suite_rc=0
run_fixed_suite || suite_rc=$?
if (( suite_rc != 2 )); then
run_mixed_suite || suite_rc=$?
fi
if [[ "${DRY_RUN}" == "1" ]]; then
stop_service
CLEANUP_ON_EXIT=0
log "Dry run complete; no result directory was created"
return 0
fi
stop_service
CLEANUP_ON_EXIT=0
summarize_results
local status="COMPLETED"
if (( suite_rc == 2 )); then
status="ABORTED_ENGINE_UNHEALTHY"
elif (( FIXED_FAILURES + MIXED_FAILURES > 0 )); then
status="COMPLETED_WITH_FAILURES"
fi
complete_manifest "${status}"
log "Quick performance map complete: status=${status}"
log "Results: ${RESULT_DIR}"
return "${suite_rc}"
}
main() {
case "${ACTION}" in
all)
run_all
;;
start)
start_service
;;
fixed)
run_standalone_suite "fixed"
;;
mixed)
run_standalone_suite "mixed"
;;
stop)
stop_service
;;
*)
printf 'Usage: %s {all|start|fixed|mixed|stop}\n' "$0" >&2
return 2
;;
esac
}
main "$@"

143
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/tests/test_quick_map_results.py Normal file
View File

@ -0,0 +1,143 @@
from __future__ import annotations
import json
import sys
import tempfile
import unittest
from pathlib import Path
EXPERIMENT_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(EXPERIMENT_DIR))
import quick_map_results # noqa: E402
class QuickMapResultsTest(unittest.TestCase):
def test_scenario_file_is_valid_and_unique(self) -> None:
scenarios = quick_map_results.parse_scenarios(EXPERIMENT_DIR / "quick_map_scenarios.tsv")
self.assertEqual(len(scenarios), 9)
self.assertEqual(len({item["case_id"] for item in scenarios}), 9)
self.assertEqual(scenarios[2]["isl"], 131072)
self.assertEqual(scenarios[2]["case_id"], "long_prefill_latency_128k_c1")
self.assertEqual(scenarios[2]["prompt_multiplier"], 1)
self.assertEqual(scenarios[2]["min_prompts"], 1)
self.assertEqual(scenarios[2]["warmup_requests"], 0)
def test_latency_percentiles_fall_back_to_request_arrays(self) -> None:
data = {
"completed": 2,
"input_lens": [1024, 1024],
"ttfts": [0.1, 0.3],
"itls": [[0.01, 0.02], [0.03, 0.04]],
"request_latencies": [0.13, 0.37],
"input_throughput": 100.0,
"output_throughput": 20.0,
}
metrics = quick_map_results.compute_metrics(data)
self.assertAlmostEqual(metrics["ttft_p95_ms"], 290.0)
self.assertAlmostEqual(metrics["tpot_p50_ms"], 25.0)
self.assertAlmostEqual(metrics["e2e_p95_ms"], 358.0)
self.assertEqual(metrics["total_token_throughput"], 120.0)
def test_summary_keeps_failed_case_and_aggregates_completed_case(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
result_dir = Path(temporary)
completed_dir = result_dir / "cases" / "decode_throughput_1k_to_1k_c32" / "rep1"
completed_dir.mkdir(parents=True)
bench_path = completed_dir / "bench.jsonl"
bench_path.write_text(
json.dumps(
{
"completed": 2,
"failed": 0,
"duration": 1.0,
"total_input_tokens": 2048,
"total_output_tokens": 2048,
"request_throughput": 2.0,
"input_throughput": 2048.0,
"output_throughput": 2048.0,
"total_throughput": 4096.0,
"mean_ttft_ms": 100.0,
"median_ttft_ms": 100.0,
"p95_ttft_ms": 110.0,
"p99_ttft_ms": 120.0,
"mean_tpot_ms": 10.0,
"median_tpot_ms": 10.0,
"p95_tpot_ms": 11.0,
"p99_tpot_ms": 12.0,
"mean_itl_ms": 10.0,
"median_itl_ms": 10.0,
"p95_itl_ms": 11.0,
"p99_itl_ms": 12.0,
"mean_e2e_latency_ms": 1000.0,
"median_e2e_latency_ms": 1000.0,
"p95_e2e_latency_ms": 1100.0,
"p99_e2e_latency_ms": 1200.0,
}
)
+ "\n",
encoding="utf-8",
)
self._write_meta(
completed_dir / "meta.json",
case_id="decode_throughput_1k_to_1k_c32",
status="COMPLETED",
bench_path=bench_path,
)
failed_dir = result_dir / "cases" / "long_prefill_latency_128k_c1" / "rep1"
failed_dir.mkdir(parents=True)
self._write_meta(
failed_dir / "meta.json",
case_id="long_prefill_latency_128k_c1",
status="FAILED",
bench_path=failed_dir / "bench.jsonl",
error_type="OOM",
)
quick_map_results.summarize(result_dir)
summary_lines = (result_dir / "summary.jsonl").read_text(encoding="utf-8").splitlines()
self.assertEqual(len(summary_lines), 2)
aggregate_text = (result_dir / "aggregate.csv").read_text(encoding="utf-8")
self.assertIn("4096.0", aggregate_text)
report_text = (result_dir / "report.md").read_text(encoding="utf-8")
self.assertIn("long_prefill_latency_128k_c1", report_text)
self.assertIn("OOM", report_text)
@staticmethod
def _write_meta(
path: Path,
*,
case_id: str,
status: str,
bench_path: Path,
error_type: str = "",
) -> None:
value = {
"run_id": "test",
"suite": "fixed",
"case_id": case_id,
"role": "",
"stage": "test",
"repetition": 1,
"isl": 1024,
"osl": 1024,
"concurrency": 32,
"num_prompts": 64,
"warmup_requests": 1,
"status": status,
"error_type": error_type,
"exit_code": 0 if status == "COMPLETED" else 1,
"started_at": "2026-07-30T00:00:00+0800",
"ended_at": "2026-07-30T00:00:01+0800",
"elapsed_s": 1,
"bench_file": str(bench_path),
"bench_log": str(path.parent / "bench.log"),
"note": "",
}
path.write_text(json.dumps(value), encoding="utf-8")
if __name__ == "__main__":
unittest.main()