diff --git a/README.md b/README.md index 35b5189..670e956 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,9 @@ # 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`。项目目的与工作流见下方。 > **项目目的**:当新显卡(GPU/NPU)到货时,用最短时间完成大模型在该平台上的推理性能评估与部署配置选型。 > 当前模型: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/pro6000/dsv4_pro6000_vllm_tp_dp_matrix/` | RTX 6000D + vLLM | | `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 老形态 & 专项) diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/.gitignore b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/.gitignore new file mode 100644 index 0000000..65e1c17 --- /dev/null +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/.gitignore @@ -0,0 +1,4 @@ +results/ +runtime/ +__pycache__/ +tests/__pycache__/ diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/README.md b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/README.md new file mode 100644 index 0000000..faeaf25 --- /dev/null +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/README.md @@ -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_manifest.json + run.log + summary.csv + summary.jsonl + aggregate.csv + report.md + cases//rep/ + 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. diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/config.env b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/config.env new file mode 100644 index 0000000..e252d4c --- /dev/null +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/config.env @@ -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}" diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_results.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_results.py new file mode 100755 index 0000000..ee08ac7 --- /dev/null +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_results.py @@ -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() diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_scenarios.tsv b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_scenarios.tsv new file mode 100644 index 0000000..b96ef69 --- /dev/null +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/quick_map_scenarios.tsv @@ -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 diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/run_quick_map.sh b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/run_quick_map.sh new file mode 100755 index 0000000..947f488 --- /dev/null +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/run_quick_map.sh @@ -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 "$@" diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/tests/test_quick_map_results.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/tests/test_quick_map_results.py new file mode 100644 index 0000000..1e31230 --- /dev/null +++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/tests/test_quick_map_results.py @@ -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()