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

Merge branch 'main' of https://gitee.com/yy-fighting/sskj

Browse Source
This commit is contained in:
yy-fighting 2026-07-09 15:24:08 +00:00
commit 90d1738605
8 changed files with 1136 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

162
experiments/dsv4_h200_sglang_tp_dp_matrix/compare.py Executable file
View File

@ -0,0 +1,162 @@
#!/usr/bin/env python3
"""Cross TP×DP configuration comparison for dsv4_h200_sglang_tp_dp_matrix.
Usage:
python3 compare.py --run-root results/<run_id> [--output comparison.md]
"""
import argparse
import json
import re
from collections import defaultdict
from pathlib import Path
def load_result(result_root: Path) -> dict:
path = result_root / "results.json"
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def slo_status(ttft_p95_ms: float, tpot_mean_ms: float,
ttft_limit_ms: float = 3000.0, tpot_limit_ms: float = 50.0) -> str:
ttft_ok = ttft_p95_ms < ttft_limit_ms
tpot_ok = tpot_mean_ms < tpot_limit_ms
if ttft_ok and tpot_ok:
return "PASS"
if ttft_ok or tpot_ok:
return "PARTIAL"
return "FAIL"
def gpu_memory_str(gpu: dict | None) -> str:
if not gpu:
return "-"
peak = gpu.get("peak_used_mb", 0)
total = gpu.get("memory_total_mb", 0)
if total:
return f"{peak:.0f}/{total:.0f} ({100*peak/total:.1f}%)"
return f"{peak:.0f}"
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--run-root", type=Path, required=True)
parser.add_argument("-o", "--output", type=Path, default=Path("comparison.md"))
parser.add_argument("--ttft-limit", type=float, default=3000.0)
parser.add_argument("--tpot-limit", type=float, default=50.0)
args = parser.parse_args()
# Discover configurations: tp*_dp* directories.
configs = []
for subdir in sorted(args.run_root.iterdir()):
if not subdir.is_dir():
continue
name = subdir.name
if not (name.startswith("tp") and "_dp" in name):
continue
results_json = subdir / "results.json"
if not results_json.exists():
continue
configs.append((name, load_result(subdir)))
if not configs:
print(f"No tp*_dp* results found under {args.run_root}")
return
model = configs[0][1].get("metadata", {}).get("model", "unknown")
hardware = configs[0][1].get("metadata", {}).get("hardware", "unknown")
# Group by scenario name.
by_scenario: dict[str, dict[str, dict]] = defaultdict(dict)
skipped: dict[str, dict[str, str]] = defaultdict(dict)
for label, data in configs:
for s in data.get("scenarios", []):
key = s["name"]
if s.get("status") == "skipped_oom":
skipped[key][label] = s.get("note", "skipped")
else:
by_scenario[key][label] = s
with open(args.output, "w", encoding="utf-8") as f:
f.write(f"# SGLang TP×DP matrix comparison ({hardware})\n\n")
f.write("## Summary\n\n")
f.write(f"- Model: `{model}`\n")
f.write(f"- Hardware: {hardware}\n")
f.write("- Backend: SGLang (Docker)\n")
f.write("- Benchmark client: `sglang.bench_serving`\n")
f.write(f"- SLO reference: TTFT P95 < {args.ttft_limit}ms, TPOT mean < {args.tpot_limit}ms\n\n")
# Configuration overview.
f.write("### Configurations\n\n")
f.write("| Config | TP | DP | GPUs/replica | Notes |\n")
f.write("|---|---:|---:|---:|---|\n")
for label, data in configs:
cfg = data.get("config", {})
tp = cfg.get("tp", "?")
dp = cfg.get("dp", "?")
f.write(f"| {label} | {tp} | {dp} | {tp} | server args recorded per ISL in results.json |\n")
f.write("\n")
# Side-by-side table.
f.write("## Side-by-side results\n\n")
headers = [
"Scenario", "ISL", "DSL", "Config", "Conc", "Req/s", "OutTok/s",
"TTFT P95(ms)", "TTFT P99(ms)", "TPOT Mean(ms)", "TPOT P95(ms)",
"TPOT P99(ms)", "E2E P99(ms)", "Peak GPU mem", "SLO"
]
f.write("| " + " | ".join(headers) + " |\n")
f.write("|" + "|".join(["---"] * len(headers)) + "|\n")
for scenario_name in sorted(by_scenario.keys(), key=lambda x: tuple(map(int, re.findall(r"\d+", x)))):
_, isl, dsl = re.findall(r"\d+", scenario_name)
for label, data in configs:
s = by_scenario[scenario_name].get(label)
if s is None:
if scenario_name in skipped and label in skipped[scenario_name]:
note = skipped[scenario_name][label]
f.write(f"| {scenario_name} | {isl} | {dsl} | {label} | - | - | - | - | - | - | - | - | - | - | {note} |\n")
continue
cfg = s["config"]
m = s["metrics"]
status = slo_status(m["ttft_ms"]["p95"], m["tpot_ms"]["mean"], args.ttft_limit, args.tpot_limit)
gpu = m.get("gpu_memory")
f.write(
f"| {scenario_name} | {isl} | {dsl} | {label} | {cfg['concurrency']} | "
f"{m['request_throughput']:.2f} | {m['output_token_throughput']:.2f} | "
f"{m['ttft_ms']['p95']:.2f} | {m['ttft_ms']['p99']:.2f} | "
f"{m['tpot_ms']['mean']:.2f} | {m['tpot_ms']['p95']:.2f} | {m['tpot_ms']['p99']:.2f} | "
f"{m['e2e_ms']['p99']:.2f} | {gpu_memory_str(gpu)} | {status} |\n"
)
# Best throughput per ISL/DSL.
f.write("\n## Best throughput per (ISL, DSL)\n\n")
f.write("| ISL | DSL | Best Config | Concurrency | OutTok/s | TTFT P95(ms) | TPOT Mean(ms) | SLO |\n")
f.write("|---:|---:|---|---:|---:|---:|---:|---:|\n")
best_by_shape: dict[tuple[int, int], tuple[float, str, dict]] = {}
for scenario_name, backends in by_scenario.items():
_, isl, dsl = re.findall(r"\d+", scenario_name)
isl_i, dsl_i = int(isl), int(dsl)
for label, s in backends.items():
m = s["metrics"]
out_tok = m["output_token_throughput"]
if (isl_i, dsl_i) not in best_by_shape or out_tok > best_by_shape[(isl_i, dsl_i)][0]:
best_by_shape[(isl_i, dsl_i)] = (out_tok, label, s)
for (isl_i, dsl_i), (out_tok, label, s) in sorted(best_by_shape.items()):
m = s["metrics"]
status = slo_status(m["ttft_ms"]["p95"], m["tpot_ms"]["mean"], args.ttft_limit, args.tpot_limit)
f.write(
f"| {isl_i} | {dsl_i} | {label} | {s['config']['concurrency']} | "
f"{out_tok:.2f} | {m['ttft_ms']['p95']:.2f} | {m['tpot_ms']['mean']:.2f} | {status} |\n"
)
f.write("\n## Notes\n\n")
f.write("- SLO check uses TTFT P95 and TPOT mean.\n")
f.write("- A PARTIAL indicates one of the two metrics is out of target; FAIL indicates both are out.\n")
f.write("- `Peak GPU mem` shows peak used / total MB and utilization percentage.\n")
f.write("- Optional (P) combinations that failed are marked as skipped/OOM and do not break the run.\n")
print(f"Wrote comparison to {args.output}")
if __name__ == "__main__":
main()

62
experiments/dsv4_h200_sglang_tp_dp_matrix/config.env Normal file
View File

@ -0,0 +1,62 @@
# TP×DP matrix experiment for DeepSeek-V4-Flash on H200 (8 GPUs) using SGLang Docker.
# Tests SGLang with three parallel configurations:
# TP=2, DP=4 -> 2 GPUs per replica, 4 replicas
# TP=4, DP=2 -> 4 GPUs per replica, 2 replicas
# TP=8, DP=1 -> 8 GPUs, no data parallelism
EXPERIMENT="dsv4_h200_sglang_tp_dp_matrix"
MODEL_NAME="DeepSeek-V4-Flash"
MODEL_PATH="/data3/hf_models/DeepSeek-V4-Flash"
SERVED_MODEL_NAME="deepseek-v4-flash"
SGLANG_PORT="${SGLANG_PORT:-30031}"
# Python interpreter for orchestration scripts (parse_backend.py, compare.py, etc.)
# and the benchmark client. Defaults to the system python3 if the sglang venv
# does not exist on the host.
VENV_CLIENT="${VENV_CLIENT:-/data/user1/yy/envs/sglang}"
# When USE_DOCKER_CLIENT=1, the benchmark client (sglang.bench_serving) is run
# inside the same SGLang Docker image used for the server. This avoids requiring
# a native sglang installation on the host.
USE_DOCKER_CLIENT="${USE_DOCKER_CLIENT:-1}"
export CUDA_VISIBLE_DEVICES="${CUDA_VISIBLE_DEVICES:-0,1,2,3,4,5,6,7}"
# Parallel configurations to test. Format: "TP DP"
declare -a PARALLEL_CONFIGS=(
"2 4"
"4 2"
"8 1"
)
# SGLang server settings
CONTEXT_LENGTH="${CONTEXT_LENGTH:-65536}"
MAX_RUNNING_REQUESTS="${MAX_RUNNING_REQUESTS:-64}"
MEM_FRACTION_STATIC="${MEM_FRACTION_STATIC:-0.88}"
MOE_RUNNER_BACKEND="${MOE_RUNNER_BACKEND:-marlin}"
# Docker deployment switch. SGLang is launched via Docker using the image below.
USE_DOCKER="${USE_DOCKER:-1}"
DOCKER_IMAGE="${DOCKER_IMAGE:-lmsysorg/sglang:latest}"
# Matrix and concurrency rules are defined in matrix.json by default.
MATRIX_FILE="${MATRIX_FILE:-${SCRIPT_DIR:-.}/matrix.json}"
MATRIX_MODE="${MATRIX_MODE:-Y}"
# Sampling density for concurrency. 0 means use the default heuristic in
# generate_scenarios.py (6-8 points, or all integers when range is small).
CONCURRENCY_SAMPLES="${CONCURRENCY_SAMPLES:-0}"
# Per-scenario timeout to avoid hangs (seconds).
SCENARIO_TIMEOUT_S="${SCENARIO_TIMEOUT_S:-1800}"
# GPU memory sampling interval (seconds).
GPU_MEM_SAMPLE_INTERVAL_S="${GPU_MEM_SAMPLE_INTERVAL_S:-1}"
# Dry-run mode: if 1, only log the server args and scenario plan without starting
# any server or sending requests.
DRY_RUN="${DRY_RUN:-0}"
# Per-config scenario limit for quick smoke tests. 0 = run all generated scenarios.
GRID_LIMIT="${GRID_LIMIT:-0}"

105
experiments/dsv4_h200_sglang_tp_dp_matrix/generate_scenarios.py Executable file
View File

@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""Generate the scenario list for the TP×DP matrix experiment.
Reads matrix.json and prints TSV lines:
mark input_len output_len concurrency num_prompts
mark is one of Y/P/N. The caller (run_bench.sh) decides how to treat each.
"""
import argparse
import json
import math
import os
from pathlib import Path
def sample_concurrency(low: int, high: int, target: int) -> list[int]:
"""Return a sorted list of concurrency values in [low, high].
If the range is small, return every integer. Otherwise sample roughly
`target` points linearly between low and high (inclusive).
"""
assert 1 <= low <= high, f"invalid concurrency range: {low}-{high}"
if high - low + 1 <= target:
return list(range(low, high + 1))
points = set()
points.add(low)
points.add(high)
step = (high - low) / (target - 1)
for i in range(1, target - 1):
v = low + round(step * i)
points.add(max(low, min(high, v)))
return sorted(points)
def generate_scenarios(matrix_path: Path, mode: str, target_samples: int) -> list[dict]:
with open(matrix_path, "r", encoding="utf-8") as f:
data = json.load(f)
matrix = data["matrix"]
concurrency_cfg = data["concurrency"]
scenarios = []
for isl_str in sorted(matrix.keys(), key=int):
dsl_map = matrix[isl_str]
low = concurrency_cfg[isl_str]["low"]
high = concurrency_cfg[isl_str]["high"]
concurrencies = sample_concurrency(low, high, target_samples)
for dsl_str in sorted(dsl_map.keys(), key=int):
mark = dsl_map[dsl_str]
if mode == "Y" and mark != "Y":
continue
if mode == "Y+P" and mark not in ("Y", "P"):
continue
# mode == "all" keeps everything, including N.
for conc in concurrencies:
scenarios.append(
{
"mark": mark,
"input_len": int(isl_str),
"output_len": int(dsl_str),
"concurrency": conc,
"num_prompts": conc * 5,
}
)
return scenarios
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--matrix", type=Path, default=Path("matrix.json"))
parser.add_argument("--mode", choices=["Y", "Y+P", "all"], default=None,
help="Scenario selection mode. Defaults to matrix.mode.")
parser.add_argument("--target-samples", type=int, default=0,
help="Target number of concurrency samples. 0 = heuristic (6-8).")
args = parser.parse_args()
with open(args.matrix, "r", encoding="utf-8") as f:
data = json.load(f)
mode = args.mode if args.mode else data.get("mode", "Y+P")
target_samples = args.target_samples
if target_samples <= 0:
env_samples = os.getenv("CONCURRENCY_SAMPLES", "0")
try:
target_samples = int(env_samples)
except ValueError:
target_samples = 0
if target_samples <= 0:
target_samples = 7
scenarios = generate_scenarios(args.matrix, mode, target_samples)
print("mark\tinput_len\toutput_len\tconcurrency\tnum_prompts")
for s in scenarios:
print(f"{s['mark']}\t{s['input_len']}\t{s['output_len']}\t{s['concurrency']}\t{s['num_prompts']}")
if __name__ == "__main__":
main()

81
experiments/dsv4_h200_sglang_tp_dp_matrix/matrix.json Normal file
View File

@ -0,0 +1,81 @@
{
"comment": "ISL/DSL matrix for dsv4_h200_sglang_tp_dp_matrix. Y=must test, P=optional (record N on failure), N=skip.",
"mode": "Y",
"comment": "Only mandatory (Y) combinations are tested; 1M ISL is excluded per user request.",
"matrix": {
"1024": {
"128": "Y",
"256": "Y",
"512": "Y",
"1024": "Y",
"2048": "Y",
"4096": "Y"
},
"4096": {
"128": "Y",
"256": "Y",
"512": "Y",
"1024": "Y",
"2048": "Y",
"4096": "Y"
},
"16384": {
"128": "Y",
"256": "Y",
"512": "Y",
"1024": "Y",
"2048": "Y",
"4096": "P"
},
"65536": {
"128": "Y",
"256": "Y",
"512": "Y",
"1024": "Y",
"2048": "P",
"4096": "N"
},
"131072": {
"128": "Y",
"256": "Y",
"512": "Y",
"1024": "P",
"2048": "N",
"4096": "N"
},
"262144": {
"128": "Y",
"256": "Y",
"512": "P",
"1024": "N",
"2048": "N",
"4096": "N"
},
"524288": {
"128": "Y",
"256": "P",
"512": "N",
"1024": "N",
"2048": "N",
"4096": "N"
},
"1048576": {
"128": "Y",
"256": "P",
"512": "N",
"1024": "N",
"2048": "N",
"4096": "N"
}
},
"concurrency": {
"1024": { "low": 1, "high": 128 },
"4096": { "low": 1, "high": 64 },
"16384": { "low": 1, "high": 32 },
"65536": { "low": 1, "high": 8 },
"131072": { "low": 1, "high": 4 },
"262144": { "low": 1, "high": 2 },
"524288": { "low": 1, "high": 2 },
"1048576": { "low": 1, "high": 2 }
}
}

527
experiments/dsv4_h200_sglang_tp_dp_matrix/run_bench.sh Executable file
View File

@ -0,0 +1,527 @@
#!/usr/bin/env bash
# TP×DP matrix benchmark for DeepSeek-V4-Flash on SGLang (Docker).
set -Eeuo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
EXPERIMENT_NAME="$(basename "$SCRIPT_DIR")"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/../../scripts/common/lib.sh"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/../../scripts/common/platform.sh"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/config.env"
RUN_ID="${RUN_ID:-$(date '+%Y%m%d-%H%M%S')}"
RESULT_BASE="${SCRIPT_DIR}/results"
MATRIX_FILE="${MATRIX_FILE:-${SCRIPT_DIR}/matrix.json}"
MATRIX_MODE="${MATRIX_MODE:-Y}"
SCENARIO_TIMEOUT_S="${SCENARIO_TIMEOUT_S:-1800}"
GPU_MEM_SAMPLE_INTERVAL_S="${GPU_MEM_SAMPLE_INTERVAL_S:-1}"
DRY_RUN="${DRY_RUN:-0}"
GRID_LIMIT="${GRID_LIMIT:-0}"
if [[ -x "${VENV_CLIENT}/bin/python" ]]; then
PYTHON="${VENV_CLIENT}/bin/python"
else
PYTHON="$(command -v python3)"
fi
DOCKER_IMAGE="${DOCKER_IMAGE:-lmsysorg/sglang:latest}"
log_dir_global="${RESULT_BASE}/${RUN_ID}/logs"
mkdir -p "$log_dir_global"
log_init "${log_dir_global}/orchestrator.log"
log "experiment=${EXPERIMENT_NAME} run_id=${RUN_ID} platform=${PLATFORM} hardware=${HARDWARE}"
log "matrix_mode=${MATRIX_MODE} matrix_file=${MATRIX_FILE} dry_run=${DRY_RUN} grid_limit=${GRID_LIMIT}"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
is_server_healthy() {
curl --fail --silent --show-error --max-time 5 "http://127.0.0.1:${SGLANG_PORT}/health" >/dev/null 2>&1
}
stop_server() {
local tp="$1"
local dp="$2"
local pid_file="/data/user1/yy/${EXPERIMENT}_sglang_tp${tp}_dp${dp}.pid"
if [[ -f "$pid_file" ]]; then
local pid
pid="$(cat "$pid_file")"
if kill -0 "$pid" 2>/dev/null; then
log "stopping sglang server pid=${pid} (tp=${tp}, dp=${dp})"
kill "$pid" 2>/dev/null || true
sleep 5
kill -9 "$pid" 2>/dev/null || true
fi
rm -f "$pid_file"
fi
# Fallback: remove any Docker container started by this experiment.
docker rm -f "${EXPERIMENT}_sglang_tp${tp}_dp${dp}" >/dev/null 2>&1 || true
# Fallback: kill any sglang serve processes for this model.
pkill -9 -f "sglang serve.*${MODEL_NAME}" 2>/dev/null || true
pkill -9 -f "sglang serve.*${MODEL_PATH}" 2>/dev/null || true
sleep 2
}
build_server_args() {
local tp="$1"
local dp="$2"
local args=(
"sglang serve" "$MODEL_PATH"
--trust-remote-code
--tp-size "$tp"
--moe-runner-backend "$MOE_RUNNER_BACKEND"
--context-length "$CONTEXT_LENGTH"
--max-running-requests "$MAX_RUNNING_REQUESTS"
--mem-fraction-static "$MEM_FRACTION_STATIC"
--host 0.0.0.0
--port "$SGLANG_PORT"
)
if [[ "$dp" -gt 1 ]]; then
args+=(
--dp-size "$dp"
)
fi
printf '%s ' "${args[@]}"
}
start_server() {
local tp="$1"
local dp="$2"
log "starting sglang server tp=${tp} dp=${dp}"
bash "${SCRIPT_DIR}/start_sglang_dp.sh" "$tp" "$dp" \
>> "${log_dir_global}/sglang_tp${tp}_dp${dp}.server.outer.log" 2>&1
if ! is_server_healthy; then
log "error: sglang server tp=${tp} dp=${dp} failed health check on port ${SGLANG_PORT}"
return 1
fi
log "sglang server tp=${tp} dp=${dp} is healthy on port ${SGLANG_PORT}"
}
restart_server() {
local tp="$1"
local dp="$2"
log "restarting sglang server tp=${tp} dp=${dp} after non-OOM failure"
stop_server "$tp" "$dp"
sleep 10
start_server "$tp" "$dp"
}
run_bench_serving() {
# Run sglang.bench_serving either natively or inside the SGLang Docker image.
if [[ "${USE_DOCKER_CLIENT:-1}" == "1" ]]; then
docker run --rm \
--network host \
-v "${MODEL_PATH}:${MODEL_PATH}:ro" \
-v "${RESULT_BASE}:${RESULT_BASE}" \
-e PYTHONUNBUFFERED=1 \
"${DOCKER_IMAGE}" \
python -m sglang.bench_serving "$@"
else
"$PYTHON" -m sglang.bench_serving "$@"
fi
}
run_warmup() {
local input_len="$1"
local output_len="$2"
log "warming up (input=${input_len}, output=${output_len}, num=1)"
run_bench_serving \
--backend sglang \
--host 127.0.0.1 \
--port "$SGLANG_PORT" \
--dataset-name random \
--random-input-len "$input_len" \
--random-output-len "$output_len" \
--num-prompts 1 \
--max-concurrency 1 \
--request-rate 10000 \
--output-file /dev/null \
--output-details \
>> "${log_dir_global}/warmup.log" 2>&1
log "warmup completed"
}
scenario_already_completed() {
local output_file="$1"
local expected="$2"
[[ -s "$output_file" ]] || return 1
local completed
completed="$("$PYTHON" -c "
import json, sys
path = sys.argv[1]
try:
with open(path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line:
data = json.loads(line)
print(data.get('completed', 0))
break
except Exception:
print(0)
" "$output_file")"
[[ "${completed:-0}" -ge "$expected" ]]
}
scenario_already_processed() {
local result_root="$1"
local scenario_name="$2"
local json_path="${result_root}/results.json"
[[ -f "$json_path" ]] || return 1
"$PYTHON" -c "
import json, sys
path, name = sys.argv[1], sys.argv[2]
try:
with open(path, 'r', encoding='utf-8') as f:
data = json.load(f)
for s in data.get('scenarios', []):
if s.get('name') == name:
if s.get('status') or s.get('metrics', {}).get('success', 0) > 0:
sys.exit(0)
except Exception:
pass
sys.exit(1)
" "$json_path" "$scenario_name"
}
detect_oom() {
local detail_log="$1"
local server_outer_log="$2"
local pattern='CUDA out of memory|torch\.OutOfMemoryError|OutOfMemory|out of memory|OOM|RESOURCE_EXHAUSTED|Failed to allocate memory'
if grep -Eiq "$pattern" "$detail_log" "$server_outer_log" 2>/dev/null; then
return 0
fi
return 1
}
start_gpu_monitor() {
local csv_path="$1"
mkdir -p "$(dirname "$csv_path")"
nvidia-smi \
--query-gpu=timestamp,index,memory.used,memory.total,utilization.gpu \
--format=csv \
-l "$GPU_MEM_SAMPLE_INTERVAL_S" \
> "$csv_path" 2>/dev/null &
echo $!
}
stop_gpu_monitor() {
local pid="$1"
if kill -0 "$pid" 2>/dev/null; then
kill "$pid" 2>/dev/null || true
sleep 1
kill -9 "$pid" 2>/dev/null || true
fi
}
append_scenario_record() {
local result_root="$1"
local json_path="$result_root/results.json"
shift
local scenario_json
scenario_json="$("$PYTHON" -c "
import json, sys
pairs = [a.split('=', 1) for a in sys.argv[1:]]
d = {}
for k, v in pairs:
try:
d[k] = json.loads(v)
except json.JSONDecodeError:
d[k] = v
print(json.dumps(d, ensure_ascii=False))
" "$@")"
PYTHON="$PYTHON" append_scenario_to_json "$json_path" "$scenario_json"
}
record_skipped_csv() {
local csv_path="$1"
shift
# Args: key=value
local row
row="$("$PYTHON" -c "
import csv, json, sys, io
pairs = [a.split('=', 1) for a in sys.argv[1:]]
d = {}
for k, v in pairs:
try:
d[k] = json.loads(v)
except json.JSONDecodeError:
d[k] = v
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=['engine','tp','dp','mark','isl','dsl','concurrency','status','reason','detail_log'], extrasaction='ignore')
writer.writerow(d)
print(buf.getvalue().strip())
" "$@")"
echo "$row" >> "$csv_path"
}
skip_remaining_scenarios() {
local result_root="$1"
local scenario_tsv="$2"
local start_index="$3"
local status="$4"
local reason="$5"
local tp="$6"
local dp="$7"
local skipped_csv="${RESULT_BASE}/${RUN_ID}/skipped_after_oom.csv"
local i=0
tail -n +2 "$scenario_tsv" | while IFS=$'\t' read -r mark isl dsl conc num; do
if (( i < start_index )); then
i=$((i + 1))
continue
fi
i=$((i + 1))
local sname="c${conc}_i${isl}_o${dsl}"
if scenario_already_processed "$result_root" "$sname"; then
continue
fi
append_scenario_record "$result_root" \
"name=${sname}" \
"config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset random --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
"status=\"${status}\"" \
"note=\"${reason}\""
record_skipped_csv "$skipped_csv" \
"engine=sglang" "tp=${tp}" "dp=${dp}" "mark=${mark}" "isl=${isl}" "dsl=${dsl}" "concurrency=${conc}" "status=${status}" "reason=${reason}"
done
}
# ---------------------------------------------------------------------------
# Per-configuration runner
# ---------------------------------------------------------------------------
run_parallel_config() {
local tp="$1"
local dp="$2"
local config_label="tp${tp}_dp${dp}"
local result_root="${RESULT_BASE}/${RUN_ID}/${config_label}"
local raw_dir="${result_root}/raw_outputs"
local gpu_log_dir="${result_root}/gpu_logs"
local phase_log_dir="${result_root}/logs"
mkdir -p "$raw_dir" "$gpu_log_dir" "$phase_log_dir"
log "===== ${config_label} START ====="
# Generate scenario list for this config.
local scenario_tsv="${result_root}/scenarios.tsv"
"$PYTHON" "${SCRIPT_DIR}/generate_scenarios.py" \
--matrix "$MATRIX_FILE" \
--mode "$MATRIX_MODE" \
> "$scenario_tsv"
local total_scenarios
total_scenarios="$(tail -n +2 "$scenario_tsv" | wc -l)"
log "generated ${total_scenarios} scenarios for ${config_label}"
# Write metadata.
ensure_result_root "$result_root"
write_metadata_json \
"${result_root}/results.json" \
"${EXPERIMENT_NAME}_${config_label}" \
"$RUN_ID" \
"$MODEL_PATH" \
"sglang" \
"sglang" \
"$HARDWARE" \
"$ACCELERATOR" \
"$CHIP" \
"experiments/${EXPERIMENT_NAME}/run_bench.sh" \
"$DOCKER_IMAGE" \
"H200 SGLang TP×DP matrix for DeepSeek-V4-Flash"
local server_args_str
server_args_str="$(build_server_args "$tp" "$dp")"
jq --arg tp "$tp" --arg dp "$dp" --arg cuda "$CUDA_VISIBLE_DEVICES" --arg args "$server_args_str" \
'.config = {
"tp": ($tp | tonumber),
"dp": ($dp | tonumber),
"cuda_visible_devices": $cuda,
"backend": "sglang",
"server_start_script": "experiments/'${EXPERIMENT_NAME}'/start_sglang_dp.sh",
"server_args": $args
}' "${result_root}/results.json" > "${result_root}/results.json.tmp" && \
mv "${result_root}/results.json.tmp" "${result_root}/results.json"
if [[ "$DRY_RUN" == "1" ]]; then
log "DRY_RUN: would start server with args: ${server_args_str}"
local line
tail -n +2 "$scenario_tsv" | while IFS=$'\t' read -r mark isl dsl conc num; do
log "DRY_RUN: ${config_label} scenario mark=${mark} c=${conc} i=${isl} o=${dsl} n=${num}"
done
log "===== ${config_label} DONE (dry run) ====="
return 0
fi
# Initialize skipped_after_oom.csv for this run.
local skipped_csv="${RESULT_BASE}/${RUN_ID}/skipped_after_oom.csv"
if [[ ! -f "$skipped_csv" ]]; then
echo "engine,tp,dp,mark,isl,dsl,concurrency,status,reason,detail_log" > "$skipped_csv"
fi
# Start server once for this TP×DP config.
if ! start_server "$tp" "$dp"; then
log "ERROR: ${config_label} failed to start; skipping all scenarios"
skip_remaining_scenarios "$result_root" "$scenario_tsv" 0 "SKIPPED_SERVICE_START_FAILED" "service failed to start" "$tp" "$dp"
log "===== ${config_label} DONE ====="
return 0
fi
# Warmup with a small prompt before the first scenario.
run_warmup 1024 128 || true
# Read scenarios into an array so we can skip remaining entries on failure.
local -a scenarios=()
while IFS= read -r line; do
scenarios+=("$line")
done < <(tail -n +2 "$scenario_tsv")
local i mark isl dsl conc num
local output_file detail_log gpu_csv sname bench_rc
for (( i = 0; i < ${#scenarios[@]}; i++ )); do
IFS=$'\t' read -r mark isl dsl conc num <<< "${scenarios[$i]}"
if [[ "$GRID_LIMIT" -gt 0 && "$i" -ge "$GRID_LIMIT" ]]; then
log "GRID_LIMIT=${GRID_LIMIT} reached; skipping remaining scenarios"
skip_remaining_scenarios "$result_root" "$scenario_tsv" "$i" "SKIPPED_GRID_LIMIT" "GRID_LIMIT reached" "$tp" "$dp"
break
fi
sname="c${conc}_i${isl}_o${dsl}"
output_file="${raw_dir}/sglang_main_${conc}_${isl}_${dsl}.jsonl"
detail_log="${phase_log_dir}/sglang_${config_label}_${sname}.log"
gpu_csv="${gpu_log_dir}/gpu_mem_${conc}_${isl}_${dsl}.csv"
if scenario_already_completed "$output_file" "$num" || scenario_already_processed "$result_root" "$sname"; then
log "skipping already-processed ${config_label} scenario: ${sname}"
continue
fi
log "running ${config_label} scenario: mark=${mark} c=${conc} i=${isl} o=${dsl} n=${num}"
local gpu_pid
gpu_pid="$(start_gpu_monitor "$gpu_csv")"
bench_rc=0
timeout "$SCENARIO_TIMEOUT_S" \
run_bench_serving \
--backend sglang \
--host 127.0.0.1 \
--port "$SGLANG_PORT" \
--dataset-name random \
--random-input-len "$isl" \
--random-output-len "$dsl" \
--num-prompts "$num" \
--max-concurrency "$conc" \
--request-rate 10000 \
--output-file "$output_file" \
--output-details \
> "$detail_log" 2>&1 || bench_rc=$?
stop_gpu_monitor "$gpu_pid"
if [[ "$bench_rc" -eq 0 ]]; then
log "finished ${config_label} scenario: output=${output_file}"
append_scenario_record "$result_root" \
"name=${sname}" \
"config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset random --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
"status=\"completed\"" \
"note=\"benchmark finished successfully\""
continue
fi
# Failure handling.
if detect_oom "$detail_log" "${log_dir_global}/sglang_tp${tp}_dp${dp}.server.outer.log"; then
log "ERROR: ${config_label} scenario ${sname} triggered OOM; stopping config"
append_scenario_record "$result_root" \
"name=${sname}" \
"config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset random --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
"status=\"OOM\"" \
"note=\"detected CUDA out-of-memory\""
record_skipped_csv "$skipped_csv" \
"engine=sglang" "tp=${tp}" "dp=${dp}" "mark=${mark}" "isl=${isl}" "dsl=${dsl}" "concurrency=${conc}" "status=OOM" "reason=detected CUDA out-of-memory" "detail_log=${detail_log}"
stop_server "$tp" "$dp"
skip_remaining_scenarios "$result_root" "$scenario_tsv" "$((i + 1))" "SKIPPED_AFTER_OOM" "previous case OOM" "$tp" "$dp"
break
fi
log "ERROR: ${config_label} scenario ${sname} failed (rc=${bench_rc}); see ${detail_log}"
if [[ "$mark" == "P" ]]; then
log "optional (P) scenario failed; recording as skipped and continuing"
append_scenario_record "$result_root" \
"name=${sname}" \
"config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset random --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
"status=\"skipped_optional\"" \
"note=\"optional scenario failed (rc=${bench_rc})\""
record_skipped_csv "$skipped_csv" \
"engine=sglang" "tp=${tp}" "dp=${dp}" "mark=${mark}" "isl=${isl}" "dsl=${dsl}" "concurrency=${conc}" "status=skipped_optional" "reason=optional scenario failed (rc=${bench_rc})" "detail_log=${detail_log}"
continue
fi
# Mandatory scenario failed but not OOM: try to restart the server.
if restart_server "$tp" "$dp"; then
run_warmup 1024 128 || true
log "resuming ${config_label} after server restart"
continue
fi
log "ERROR: ${config_label} server restart failed; skipping remaining scenarios"
append_scenario_record "$result_root" \
"name=${sname}" \
"config=$(jq -n --arg phase main --argjson c "$conc" --argjson i "$isl" --argjson o "$dsl" --arg dataset random --argjson n "$num" '{phase: $phase, concurrency: $c, input_len: $i, output_len: $o, dataset: $dataset, num_prompts: $n}')" \
"status=\"FAILED\"" \
"note=\"scenario failed and server restart failed (rc=${bench_rc})\""
record_skipped_csv "$skipped_csv" \
"engine=sglang" "tp=${tp}" "dp=${dp}" "mark=${mark}" "isl=${isl}" "dsl=${dsl}" "concurrency=${conc}" "status=FAILED" "reason=scenario failed and server restart failed" "detail_log=${detail_log}"
skip_remaining_scenarios "$result_root" "$scenario_tsv" "$((i + 1))" "SKIPPED_RESTART_FAILED" "server restart failed" "$tp" "$dp"
break
done
stop_server "$tp" "$dp"
# Parse results.
log "parsing ${config_label} results"
"$PYTHON" "${SCRIPT_DIR}/../../scripts/common/parse_backend.py" "$result_root" --backend sglang \
>> "${phase_log_dir}/parse.log" 2>&1 || {
log "WARNING: parser failed for ${config_label}; see ${phase_log_dir}/parse.log"
}
log "===== ${config_label} DONE ====="
}
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
# Cleanup any leftovers.
for cfg in "${PARALLEL_CONFIGS[@]}"; do
read -r tp dp <<< "$cfg"
stop_server "$tp" "$dp"
done
# Run each parallel configuration.
for cfg in "${PARALLEL_CONFIGS[@]}"; do
read -r tp dp <<< "$cfg"
run_parallel_config "$tp" "$dp"
done
# Generate cross-configuration comparison.
log "generating comparison report"
"$PYTHON" "${SCRIPT_DIR}/compare.py" \
--run-root "${RESULT_BASE}/${RUN_ID}" \
--output "${RESULT_BASE}/${RUN_ID}/comparison.md" \
>> "${log_dir_global}/compare.log" 2>&1 || {
log "WARNING: comparison script failed; see ${log_dir_global}/compare.log"
}
log "all results saved to ${RESULT_BASE}/${RUN_ID}"

18
experiments/dsv4_h200_sglang_tp_dp_matrix/smoke_matrix.json Normal file
View File

@ -0,0 +1,18 @@
{
"comment": "Small smoke matrix for quickly verifying TP×DP server startup and bench path.",
"mode": "Y",
"matrix": {
"1024": {
"128": "Y",
"256": "Y"
},
"4096": {
"128": "Y",
"256": "Y"
}
},
"concurrency": {
"1024": { "low": 1, "high": 8 },
"4096": { "low": 1, "high": 4 }
}
}

96
experiments/dsv4_h200_sglang_tp_dp_matrix/start_sglang_docker.sh Executable file
View File

@ -0,0 +1,96 @@
#!/usr/bin/env bash
# Start SGLang server in Docker for a given TPxDP configuration.
# Usage: start_sglang_docker.sh <TP> <DP>
#
# Uses the lmsysorg/sglang image and the same argument set as the
# bare-metal start script. The container is removed automatically on stop.
set -e
TP="${1}"
DP="${2}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/config.env"
cd /data/user1/yy
mkdir -p logs
IMAGE="${DOCKER_IMAGE:-lmsysorg/sglang:latest}"
PORT="${SGLANG_PORT:-30031}"
NAME="${EXPERIMENT}_sglang_tp${TP}_dp${DP}"
PID_FILE="/data/user1/yy/${EXPERIMENT}_sglang_tp${TP}_dp${DP}.pid"
LOG="/data/user1/yy/logs/${EXPERIMENT}_sglang_docker_tp${TP}_dp${DP}_$(date +%Y%m%d_%H%M%S).log"
rm -f "$PID_FILE"
# Clean up any stale container with the same name.
docker rm -f "$NAME" >/dev/null 2>&1 || true
SERVER_ARGS=(
serve "$MODEL_PATH"
--trust-remote-code
--tp-size "$TP"
--moe-runner-backend "$MOE_RUNNER_BACKEND"
--context-length "$CONTEXT_LENGTH"
--max-running-requests "$MAX_RUNNING_REQUESTS"
--mem-fraction-static "$MEM_FRACTION_STATIC"
--host 0.0.0.0
--port "$PORT"
)
if [[ "$DP" -gt 1 ]]; then
SERVER_ARGS+=(
--dp-size "$DP"
)
fi
SERVER_ARGS_STR="${SERVER_ARGS[*]}"
echo "=== Starting SGLang server in Docker (TP=${TP}, DP=${DP}) ==="
echo "Image: $IMAGE"
echo "Model: $MODEL_PATH"
echo "Container name: $NAME"
echo "Host port: $PORT"
echo "Command: sglang ${SERVER_ARGS_STR}"
echo "Log: $LOG"
# Run docker in the foreground so that killing the host process stops the
# container (the --rm flag ensures cleanup). nohup lets us background it and
# capture the host PID in the same way as the bare-metal start script.
nohup docker run --rm \
--name "$NAME" \
--gpus all \
--entrypoint sglang \
-p "${PORT}:${PORT}" \
-v "${MODEL_PATH}:${MODEL_PATH}:ro" \
-v /data/user1/yy/tmp:/tmp \
-e CUDA_VISIBLE_DEVICES="${CUDA_VISIBLE_DEVICES}" \
-e PYTHONUNBUFFERED=1 \
"$IMAGE" \
"${SERVER_ARGS[@]}" \
> "$LOG" 2>&1 &
PID=$!
echo $PID > "$PID_FILE"
echo "PID: $PID"
echo "Waiting for health on port ${PORT}..."
for i in $(seq 1 240); do
if curl --fail --silent --show-error --max-time 5 "http://127.0.0.1:${PORT}/health" >/dev/null 2>&1; then
echo "SGLang server is ready at http://127.0.0.1:${PORT}"
echo "Log: $LOG"
exit 0
fi
if ! kill -0 $PID 2>/dev/null; then
echo "ERROR: Docker SGLang server exited early"
tail -200 "$LOG"
exit 1
fi
echo "Waiting... ($i/240)"
sleep 5
done
echo "ERROR: Docker SGLang server not healthy after 240 retries"
tail -200 "$LOG"
exit 1

85
experiments/dsv4_h200_sglang_tp_dp_matrix/start_sglang_dp.sh Executable file
View File

@ -0,0 +1,85 @@
#!/usr/bin/env bash
# Start SGLang server for a given TP×DP configuration.
# Usage: start_sglang_dp.sh <TP> <DP>
#
# By default this delegates to the Docker start script because the experiment
# is intended to run SGLang inside a container. Set USE_DOCKER=0 to use the
# local VENV_CLIENT environment instead.
set -e
TP="${1}"
DP="${2}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/config.env"
if [[ "${USE_DOCKER:-1}" == "1" ]]; then
exec "${SCRIPT_DIR}/start_sglang_docker.sh" "$@"
fi
cd /data/user1/yy
mkdir -p logs
VENV="${VENV_CLIENT}"
export PATH="$VENV/bin:$PATH"
export PYTHONUNBUFFERED=1
export TMPDIR=/data/user1/yy/tmp
export CUDA_VISIBLE_DEVICES="${CUDA_VISIBLE_DEVICES}"
LOG="/data/user1/yy/logs/${EXPERIMENT}_sglang_tp${TP}_dp${DP}_$(date +%Y%m%d_%H%M%S).log"
PID_FILE="/data/user1/yy/${EXPERIMENT}_sglang_tp${TP}_dp${DP}.pid"
rm -f "$PID_FILE"
SERVER_ARGS=(
sglang serve "$MODEL_PATH"
--trust-remote-code
--tp-size "$TP"
--moe-runner-backend "$MOE_RUNNER_BACKEND"
--context-length "$CONTEXT_LENGTH"
--max-running-requests "$MAX_RUNNING_REQUESTS"
--mem-fraction-static "$MEM_FRACTION_STATIC"
--host 0.0.0.0
--port "$SGLANG_PORT"
)
if [[ "$DP" -gt 1 ]]; then
SERVER_ARGS+=(
--dp-size "$DP"
)
fi
SERVER_ARGS_STR="${SERVER_ARGS[*]}"
echo "=== Starting SGLang server (TP=${TP}, DP=${DP}) ==="
echo "Model: $MODEL_PATH"
echo "Port: $SGLANG_PORT"
echo "Command: $SERVER_ARGS_STR"
echo "Log: $LOG"
nohup "${SERVER_ARGS[@]}" > "$LOG" 2>&1 &
PID=$!
echo $PID > "$PID_FILE"
echo "PID: $PID"
echo "Waiting for health on port ${SGLANG_PORT}..."
for i in $(seq 1 240); do
if curl --fail --silent --show-error --max-time 5 "http://127.0.0.1:${SGLANG_PORT}/health" >/dev/null 2>&1; then
echo "SGLang server is ready at http://127.0.0.1:${SGLANG_PORT}"
echo "Log: $LOG"
exit 0
fi
if ! kill -0 $PID 2>/dev/null; then
echo "ERROR: SGLang server exited early"
tail -200 "$LOG"
exit 1
fi
echo "Waiting... ($i/240)"
sleep 5
done
echo "ERROR: SGLang server not healthy after 240 retries"
tail -200 "$LOG"
exit 1