diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/compare.py b/experiments/dsv4_h200_sglang_tp_dp_matrix/compare.py new file mode 100755 index 0000000..2029e7b --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/compare.py @@ -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/ [--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() diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/config.env b/experiments/dsv4_h200_sglang_tp_dp_matrix/config.env new file mode 100644 index 0000000..3f2bf08 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/config.env @@ -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}" diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/generate_scenarios.py b/experiments/dsv4_h200_sglang_tp_dp_matrix/generate_scenarios.py new file mode 100755 index 0000000..ef46e01 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/generate_scenarios.py @@ -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() diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/matrix.json b/experiments/dsv4_h200_sglang_tp_dp_matrix/matrix.json new file mode 100644 index 0000000..55dcf58 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/matrix.json @@ -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 } + } +} diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/run_bench.sh b/experiments/dsv4_h200_sglang_tp_dp_matrix/run_bench.sh new file mode 100755 index 0000000..13cf241 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/run_bench.sh @@ -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}" diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/smoke_matrix.json b/experiments/dsv4_h200_sglang_tp_dp_matrix/smoke_matrix.json new file mode 100644 index 0000000..5715c28 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/smoke_matrix.json @@ -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 } + } +} diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/start_sglang_docker.sh b/experiments/dsv4_h200_sglang_tp_dp_matrix/start_sglang_docker.sh new file mode 100755 index 0000000..4e1ca53 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/start_sglang_docker.sh @@ -0,0 +1,96 @@ +#!/usr/bin/env bash +# Start SGLang server in Docker for a given TPxDP configuration. +# Usage: start_sglang_docker.sh +# +# 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 diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/start_sglang_dp.sh b/experiments/dsv4_h200_sglang_tp_dp_matrix/start_sglang_dp.sh new file mode 100755 index 0000000..9a47e07 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/start_sglang_dp.sh @@ -0,0 +1,85 @@ +#!/usr/bin/env bash +# Start SGLang server for a given TP×DP configuration. +# Usage: start_sglang_dp.sh +# +# 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