From 5e95cee1a0d9a7007263c856d63387eea4269d56 Mon Sep 17 00:00:00 2001 From: Root User Date: Sat, 11 Jul 2026 06:11:19 +0000 Subject: [PATCH] bench: add adaptive concurrency search --- experiments/ADAPTIVE_CONCURRENCY_USAGE.md | 128 ++++ .../README_adaptive_concurrency.md | 109 +++ .../adaptive_config.env | 49 ++ .../dsv4_h200_sglang_tp_dp_matrix/config.env | 4 +- .../run_adaptive_concurrency.sh | 181 +++++ .../README_adaptive_concurrency.md | 115 +++ .../adaptive_config.env | 49 ++ .../dsv4_h200_vllm_tp_dp_matrix/config.env | 2 +- .../run_adaptive_concurrency.sh | 164 +++++ scripts/common/adaptive_bench_lib.sh | 682 ++++++++++++++++++ scripts/common/adaptive_concurrency.py | 320 ++++++++ 11 files changed, 1800 insertions(+), 3 deletions(-) create mode 100644 experiments/ADAPTIVE_CONCURRENCY_USAGE.md create mode 100644 experiments/dsv4_h200_sglang_tp_dp_matrix/README_adaptive_concurrency.md create mode 100644 experiments/dsv4_h200_sglang_tp_dp_matrix/adaptive_config.env create mode 100755 experiments/dsv4_h200_sglang_tp_dp_matrix/run_adaptive_concurrency.sh create mode 100644 experiments/dsv4_h200_vllm_tp_dp_matrix/README_adaptive_concurrency.md create mode 100644 experiments/dsv4_h200_vllm_tp_dp_matrix/adaptive_config.env create mode 100755 experiments/dsv4_h200_vllm_tp_dp_matrix/run_adaptive_concurrency.sh create mode 100755 scripts/common/adaptive_bench_lib.sh create mode 100755 scripts/common/adaptive_concurrency.py diff --git a/experiments/ADAPTIVE_CONCURRENCY_USAGE.md b/experiments/ADAPTIVE_CONCURRENCY_USAGE.md new file mode 100644 index 0000000..4e28428 --- /dev/null +++ b/experiments/ADAPTIVE_CONCURRENCY_USAGE.md @@ -0,0 +1,128 @@ +# 自适应并发 Benchmark 使用说明 + +这套脚本对每个固定的 `(TP, DP, ISL, DSL)` 依次测试 +`C=1,2,4,8,...`，直到 Total TPS 连续两次增长不足 2%，或达到并发上限、 +发生 OOM。每个正式测点发送 `C * 5` 个请求，并先用相同并发 C warm-up。 + +## 1. 进入目录 + +vLLM： + +```bash +cd /data3/yy/sskj/experiments/dsv4_h200_vllm_tp_dp_matrix +``` + +SGLang： + +```bash +cd /data3/yy/sskj/experiments/dsv4_h200_sglang_tp_dp_matrix +``` + +两个目录的入口都是： + +```bash +bash run_adaptive_concurrency.sh +``` + +## 2. 先 dry-run + +只打印 TP/DP、ISL/DSL 和并发搜索计划，不加载模型： + +```bash +DRY_RUN=1 bash run_adaptive_concurrency.sh +``` + +## 3. 单组合 smoke + +只测 `TP=8, DP=1, ISL=1K, DSL=128`，并发搜索到 8： + +```bash +RUN_ID=smoke-$(date +%Y%m%d-%H%M%S) \ +TP_LIST="8" \ +ISL_LIST="1024" \ +DSL_LIST="128" \ +GRID_LIMIT=1 \ +SEARCH_MAX_CONCURRENCY=8 \ +RESULT_BASE=/data3/yy/sskj_adaptive_results/smoke \ +bash run_adaptive_concurrency.sh +``` + +## 4. 正式运行 + +长任务放进 tmux。vLLM 示例： + +```bash +tmux new-session -d -s vllm-adaptive \ + "cd /data3/yy/sskj/experiments/dsv4_h200_vllm_tp_dp_matrix && \ + RUN_ID=vllm-\$(date +%Y%m%d-%H%M%S) \ + RESULT_BASE=/data3/yy/sskj_adaptive_results/vllm \ + bash run_adaptive_concurrency.sh" +``` + +SGLang 示例： + +```bash +tmux new-session -d -s sglang-adaptive \ + "cd /data3/yy/sskj/experiments/dsv4_h200_sglang_tp_dp_matrix && \ + RUN_ID=sglang-\$(date +%Y%m%d-%H%M%S) \ + RESULT_BASE=/data3/yy/sskj_adaptive_results/sglang \ + bash run_adaptive_concurrency.sh" +``` + +查看任务： + +```bash +tmux ls +tmux attach -t vllm-adaptive +tmux attach -t sglang-adaptive +``` + +从 tmux 退出但保持任务运行：按 `Ctrl-b`，松开，再按 `d`。 + +## 5. 常用覆盖参数 + +无需修改脚本，可以把变量写在命令前： + +```bash +MODEL_PATH=/path/to/model \ +DATASET_PATH=/path/to/ShareGPT.json \ +DOCKER_IMAGE=your/server:image \ +DOCKER_CLIENT_IMAGE=lmsysorg/sglang:latest \ +CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 \ +SEARCH_MAX_CONCURRENCY=512 \ +TPS_MIN_GAIN_PCT=2.0 \ +PLATEAU_PATIENCE=2 \ +bash run_adaptive_concurrency.sh +``` + +- vLLM 端口用 `VLLM_PORT`，SGLang 端口用 `SGLANG_PORT`。 +- `TP_LIST="8"` 只跑 TP=8；DP 使用 `config.env` 中对应的 TP/DP 组合。 +- `ISL_LIST="1024 4096"`、`DSL_LIST="128 256"` 用来筛选长度。 +- `GRID_LIMIT=1` 表示每组 TP/DP 只跑一个 ISL/DSL shape。 +- `BENCH_WARMUP_MAX_REQUESTS=0` 表示每个测点用完整并发 C warm-up；正数表示上限。 +- 正式基线保持 `BENCH_DATASET_NAME=random` 和 `RANDOM_RANGE_RATIO=1.0`。 +- `random-ids` 只用于检查链路，不与正式 ShareGPT-backed 结果混用。 + +换机器或模型时，至少检查 `MODEL_PATH`、`DATASET_PATH`、Docker 镜像、GPU 编号、 +端口、模型最大上下文和显存参数。需要永久修改时再编辑对应目录的 `config.env`。 + +## 6. 结果 + +每次运行在 `${RESULT_BASE}/${RUN_ID}/` 下生成： + +```text +adaptive_points.csv 每个并发测点的 TPS、TTFT、TPOT 等 +adaptive_summary.csv 每个 ISL/DSL/TP/DP 的饱和并发和最佳 TPS 并发 +adaptive_summary.md 可直接阅读的汇总表 +run_manifest.json 本次模型、数据集和搜索参数 +logs/ 编排与服务日志 +tp*_dp*/raw_outputs/ sglang.bench_serving 原始 JSON +tp*_dp*/gpu_logs/ nvidia-smi 采样 +``` + +快速查看： + +```bash +column -s, -t adaptive_summary.csv | less -S +tail -f logs/orchestrator.log +``` diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/README_adaptive_concurrency.md b/experiments/dsv4_h200_sglang_tp_dp_matrix/README_adaptive_concurrency.md new file mode 100644 index 0000000..0545572 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/README_adaptive_concurrency.md @@ -0,0 +1,109 @@ +# SGLang Adaptive Concurrency Search + +`run_adaptive_concurrency.sh` keeps the existing ISL/DSL and TP/DP matrix, but +searches concurrency at runtime instead of testing only low/high endpoints. + +## Search rule + +For every fixed `(TP, DP, ISL, DSL)`: + +```text +C = 1, 2, 4, 8, ... up to SEARCH_MAX_CONCURRENCY +num_prompts = C * NUM_PROMPTS_MULTIPLIER +``` + +The search stops when Total TPS grows by less than `TPS_MIN_GAIN_PCT` for +`PLATEAU_PATIENCE` consecutive points. Defaults are 2% and two points. + +- `saturation_concurrency`: first point in the final low-gain streak +- `best_tps_concurrency`: tested point with the highest Total TPS +- `stop_probe_concurrency`: last point tested to confirm the stop + +OOM stops only the current shape. The service is restarted before the next +shape. A non-OOM failure is retried once after a service restart. + +Before every measured point, the client sends `C` concurrent warm-up requests. +This keeps lazy kernel compilation and CUDA graph capture out of the measured +TTFT/TPS. `BENCH_WARMUP_MAX_REQUESTS=0` means no cap; use a positive value only +when deliberately limiting warm-up at very high concurrency. + +## Dataset requirement + +The default remains compatible with the fixed matrix baseline: + +```text +--dataset-name random --dataset-path DATASET_PATH +--random-range-ratio 1.0 +``` + +In the installed SGLang client, the upstream default `0.0` samples each length +uniformly from `1` to the requested maximum. This adaptive experiment sets +`1.0` explicitly so every point represents the fixed ISL/DSL shown in the +matrix. Override `RANDOM_RANGE_RATIO` only when a mixed-length workload is +intentional. + +`DATASET_PATH` must have at least +`SEARCH_MAX_CONCURRENCY * NUM_PROMPTS_MULTIPLIER` usable two-turn +conversations. The configured dataset is checked before model loading. + +For setup smoke tests without ShareGPT, explicitly use: + +```bash +BENCH_DATASET_NAME=random-ids +``` + +This sends generated token IDs with `--tokenize-prompt`; it is convenient for +smoke tests but changes the token distribution, so do not mix it with random +ShareGPT-backed baseline results. + +## Commands + +Print the plan without starting a server: + +```bash +DRY_RUN=1 bash run_adaptive_concurrency.sh +``` + +Run one small search: + +```bash +TP_LIST="8" \ +ISL_LIST="1024" \ +DSL_LIST="128" \ +GRID_LIMIT=1 \ +SEARCH_MAX_CONCURRENCY=8 \ +DATASET_PATH=/path/to/ShareGPT.json \ +bash run_adaptive_concurrency.sh +``` + +Persistent-container mode remains supported: + +```bash +bash start_sglang_container.sh +CONTAINER_NAME="dsv4_h200_sglang_tp_dp_matrix_sglang_container" \ +DATASET_PATH=/path/to/ShareGPT.json \ +bash run_adaptive_concurrency.sh +``` + +Run long searches inside tmux. + +## Outputs + +Results are written to `adaptive_results//`: + +```text +adaptive_points.jsonl +adaptive_points.csv +adaptive_summary.jsonl +adaptive_summary.csv +adaptive_summary.md +run_manifest.json +tp*/raw_outputs/ +tp*/metrics/ +tp*/logs/ +tp*/gpu_logs/ +``` + +Every successful point is accepted only when `completed == num_prompts` and +actual input/output lengths match the requested shape within configured +tolerances. Invalid workloads never participate in saturation decisions. diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/adaptive_config.env b/experiments/dsv4_h200_sglang_tp_dp_matrix/adaptive_config.env new file mode 100644 index 0000000..4085397 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/adaptive_config.env @@ -0,0 +1,49 @@ +# Adaptive concurrency search settings. +# +# For each fixed (TP, DP, ISL, DSL), probe: +# C = start, start * multiplier, ... up to max +# and stop after Total TPS has less than TPS_MIN_GAIN_PCT meaningful growth for +# PLATEAU_PATIENCE consecutive points. + +SEARCH_START_CONCURRENCY="${SEARCH_START_CONCURRENCY:-1}" +SEARCH_MAX_CONCURRENCY="${SEARCH_MAX_CONCURRENCY:-512}" +SEARCH_MULTIPLIER="${SEARCH_MULTIPLIER:-2}" +NUM_PROMPTS_MULTIPLIER="${NUM_PROMPTS_MULTIPLIER:-5}" + +# A gain below 2% is treated as throughput saturation. Two consecutive +# low-gain points prevent one noisy measurement from stopping the search. +TPS_MIN_GAIN_PCT="${TPS_MIN_GAIN_PCT:-2.0}" +PLATEAU_PATIENCE="${PLATEAU_PATIENCE:-2}" + +# Keep the same random workload semantics as the fixed matrix baseline. +# DATASET_PATH must contain at least SEARCH_MAX_CONCURRENCY times +# NUM_PROMPTS_MULTIPLIER valid two-turn conversations. Set this explicitly to +# random-ids to use generated token IDs without a ShareGPT seed dataset. +BENCH_DATASET_NAME="${BENCH_DATASET_NAME:-random}" +# SGLang interprets 0.0 as Uniform[1, requested_len]. Use 1.0 for fixed +# ISL/DSL points; lower values intentionally benchmark a length distribution. +RANDOM_RANGE_RATIO="${RANDOM_RANGE_RATIO:-1.0}" +# Before each measured point, warm up with the same concurrency so lazy kernel +# compilation and CUDA graph capture are excluded from TTFT/TPS. 0 means no +# cap; set a positive cap only when very high-concurrency warmup is impractical. +BENCH_WARMUP_MAX_REQUESTS="${BENCH_WARMUP_MAX_REQUESTS:-0}" + +# Reject a point if the completed request count or actual token lengths do not +# match the requested workload. +INPUT_LENGTH_TOLERANCE_PCT="${INPUT_LENGTH_TOLERANCE_PCT:-5.0}" +OUTPUT_LENGTH_TOLERANCE_PCT="${OUTPUT_LENGTH_TOLERANCE_PCT:-10.0}" + +MAX_POINT_RETRIES="${MAX_POINT_RETRIES:-1}" +SERVER_RESTART_COOLDOWN_S="${SERVER_RESTART_COOLDOWN_S:-10}" +SCENARIO_TIMEOUT_S="${SCENARIO_TIMEOUT_S:-1800}" +GPU_MEM_SAMPLE_INTERVAL_S="${GPU_MEM_SAMPLE_INTERVAL_S:-1}" + +# Optional space-separated filters, useful for smoke tests: +# TP_LIST="8" ISL_LIST="1024" DSL_LIST="128" +TP_LIST="${TP_LIST:-}" +ISL_LIST="${ISL_LIST:-}" +DSL_LIST="${DSL_LIST:-}" + +DRY_RUN="${DRY_RUN:-0}" +# Counts ISL/DSL shapes per TP/DP config, not individual concurrency probes. +GRID_LIMIT="${GRID_LIMIT:-0}" diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/config.env b/experiments/dsv4_h200_sglang_tp_dp_matrix/config.env index d5ac289..bddd299 100644 --- a/experiments/dsv4_h200_sglang_tp_dp_matrix/config.env +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/config.env @@ -6,7 +6,7 @@ EXPERIMENT="dsv4_h200_sglang_tp_dp_matrix" MODEL_NAME="DeepSeek-V4-Flash" -MODEL_PATH="/data/models/DeepSeek-V4-Flash" +MODEL_PATH="/data3/hf_models/DeepSeek-V4-Flash" SERVED_MODEL_NAME="deepseek-v4-flash" SGLANG_PORT="${SGLANG_PORT:-30031}" @@ -45,7 +45,7 @@ DOCKER_IMAGE="${DOCKER_IMAGE:-lmsysorg/sglang:latest}" # Dataset used by sglang.bench_serving --dataset-name random. # The random sampler needs a ShareGPT-style JSON file locally; it falls back to # downloading from HuggingFace, which usually fails on offline H200 nodes. -DATASET_PATH="${DATASET_PATH:-/data/user1/yy/datasets/ShareGPT_V4.3_unfiltered_cleaned_split.json}" +DATASET_PATH="${DATASET_PATH:-/data3/yy/sskj/datasets/ShareGPT_V3_unfiltered_cleaned_split.json}" # Matrix and concurrency rules are defined in matrix.json by default. MATRIX_FILE="${MATRIX_FILE:-${SCRIPT_DIR:-.}/matrix.json}" diff --git a/experiments/dsv4_h200_sglang_tp_dp_matrix/run_adaptive_concurrency.sh b/experiments/dsv4_h200_sglang_tp_dp_matrix/run_adaptive_concurrency.sh new file mode 100755 index 0000000..90f7827 --- /dev/null +++ b/experiments/dsv4_h200_sglang_tp_dp_matrix/run_adaptive_concurrency.sh @@ -0,0 +1,181 @@ +#!/usr/bin/env bash +# Find the Total-TPS saturation concurrency for each SGLang TP/DP/ISL/DSL shape. +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" +# shellcheck source=/dev/null +source "${SCRIPT_DIR}/adaptive_config.env" +# shellcheck source=/dev/null +source "${SCRIPT_DIR}/../../scripts/common/adaptive_bench_lib.sh" + +ENGINE="sglang" +ENGINE_PORT="$SGLANG_PORT" +RESULT_BASE="${RESULT_BASE:-${SCRIPT_DIR}/adaptive_results}" +ACTIVE_ENGINE_SERVER_LOG="" + +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}" + +engine_is_healthy() { + curl --fail --silent --show-error --max-time 5 \ + "http://127.0.0.1:${ENGINE_PORT}/health" >/dev/null 2>&1 +} + +engine_stop_server() { + local tp="$1" + local dp="$2" + local pid_file="${RUNTIME_BASE}/${EXPERIMENT}_sglang_tp${tp}_dp${dp}.pid" + + if [[ -f "$pid_file" ]]; then + local pid + pid="$(cat "$pid_file")" + if [[ -n "${CONTAINER_NAME:-}" ]]; then + if docker exec "$CONTAINER_NAME" kill -0 "$pid" 2>/dev/null; then + log "stopping sglang in persistent container pid=${pid} tp=${tp} dp=${dp}" + docker exec "$CONTAINER_NAME" kill "$pid" 2>/dev/null || true + sleep 5 + docker exec "$CONTAINER_NAME" kill -9 "$pid" 2>/dev/null || true + fi + elif 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 + if [[ -z "${CONTAINER_NAME:-}" ]]; then + docker rm -f "${EXPERIMENT}_sglang_tp${tp}_dp${dp}" >/dev/null 2>&1 || true + fi + ACTIVE_ENGINE_SERVER_LOG="" + sleep 2 +} + +engine_build_server_args() { + local tp="$1" + local dp="$2" + local -a args=( + sglang serve --model-path "$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 "$ENGINE_PORT" + ) + if (( dp > 1 )); then + args+=(--dp-size "$dp") + fi + printf '%q ' "${args[@]}" +} + +engine_start_server() { + local tp="$1" + local dp="$2" + local outer_log="${ADAPTIVE_LOG_DIR}/sglang_tp${tp}_dp${dp}.server.outer.log" + log "starting sglang server tp=${tp} dp=${dp}" + if [[ -n "${CONTAINER_NAME:-}" ]]; then + bash "${SCRIPT_DIR}/run_sglang_in_container.sh" "$tp" "$dp" >> "$outer_log" 2>&1 + else + bash "${SCRIPT_DIR}/start_sglang_dp.sh" "$tp" "$dp" >> "$outer_log" 2>&1 + fi + if ! engine_is_healthy; then + log "ERROR: sglang health check failed tp=${tp} dp=${dp}" + return 1 + fi + if [[ -z "${CONTAINER_NAME:-}" ]]; then + ACTIVE_ENGINE_SERVER_LOG="$( + find "${RUNTIME_BASE}/logs" -maxdepth 1 -type f \ + -name "${EXPERIMENT}_sglang*tp${tp}_dp${dp}_*.log" \ + -printf '%T@ %p\n' 2>/dev/null | sort -nr | head -n 1 | cut -d' ' -f2- + )" + fi + log "sglang server healthy tp=${tp} dp=${dp} log=${ACTIVE_ENGINE_SERVER_LOG:-container:/tmp/sglang_server.log}" +} + +engine_detect_oom() { + local detail_log="$1" + local tp="$2" + local dp="$3" + local pattern='CUDA out of memory|torch\.OutOfMemoryError|OutOfMemory|out of memory|OOM|RESOURCE_EXHAUSTED|Failed to allocate memory' + local outer_log="${ADAPTIVE_LOG_DIR}/sglang_tp${tp}_dp${dp}.server.outer.log" + local -a logs=("$detail_log" "$outer_log") + if [[ -n "$ACTIVE_ENGINE_SERVER_LOG" ]]; then + logs+=("$ACTIVE_ENGINE_SERVER_LOG") + fi + if grep -Eiq "$pattern" "${logs[@]}" 2>/dev/null; then + return 0 + fi + if [[ -n "${CONTAINER_NAME:-}" ]]; then + docker exec "$CONTAINER_NAME" grep -Eiq "$pattern" /tmp/sglang_server.log 2>/dev/null + return $? + fi + return 1 +} + +engine_run_bench() { + local isl="$1" + local dsl="$2" + local concurrency="$3" + local num_prompts="$4" + local output_file="$5" + local warmup_requests + warmup_requests="$(adaptive_warmup_request_count "$concurrency")" + local -a bench_args=( + --backend sglang + --host 127.0.0.1 + --port "$ENGINE_PORT" + --dataset-name "$BENCH_DATASET_NAME" + --random-input-len "$isl" + --random-output-len "$dsl" + --random-range-ratio "$RANDOM_RANGE_RATIO" + --num-prompts "$num_prompts" + --max-concurrency "$concurrency" + --request-rate 10000 + --warmup-requests "$warmup_requests" + --output-file "$output_file" + --output-details + --disable-tqdm + ) + if [[ "$BENCH_DATASET_NAME" == "random" ]]; then + bench_args+=(--dataset-path "$DATASET_PATH") + else + bench_args+=(--tokenize-prompt) + fi + + if [[ "$USE_DOCKER_CLIENT" == "1" ]]; then + local -a volume_args=(-v "${MODEL_PATH}:${MODEL_PATH}:ro" -v "${RESULT_BASE}:${RESULT_BASE}") + if [[ "$BENCH_DATASET_NAME" == "random" ]]; then + volume_args+=(-v "${DATASET_PATH}:${DATASET_PATH}:ro") + fi + docker run --rm \ + --network host \ + "${volume_args[@]}" \ + -e PYTHONUNBUFFERED=1 \ + "$DOCKER_IMAGE" \ + python -m sglang.bench_serving "${bench_args[@]}" + else + "$PYTHON" -m sglang.bench_serving "${bench_args[@]}" + fi +} + +export -f engine_run_bench +export ENGINE_PORT MODEL_PATH RESULT_BASE DOCKER_IMAGE USE_DOCKER_CLIENT +export BENCH_DATASET_NAME DATASET_PATH RANDOM_RANGE_RATIO BENCH_WARMUP_MAX_REQUESTS PYTHON + +adaptive_main "$@" diff --git a/experiments/dsv4_h200_vllm_tp_dp_matrix/README_adaptive_concurrency.md b/experiments/dsv4_h200_vllm_tp_dp_matrix/README_adaptive_concurrency.md new file mode 100644 index 0000000..d6a3ced --- /dev/null +++ b/experiments/dsv4_h200_vllm_tp_dp_matrix/README_adaptive_concurrency.md @@ -0,0 +1,115 @@ +# vLLM Adaptive Concurrency Search + +`run_adaptive_concurrency.sh` keeps the existing ISL/DSL and TP/DP matrix, but +searches concurrency at runtime instead of testing only low/high endpoints. + +## Search rule + +For every fixed `(TP, DP, ISL, DSL)`: + +```text +C = 1, 2, 4, 8, ... up to SEARCH_MAX_CONCURRENCY +num_prompts = C * NUM_PROMPTS_MULTIPLIER +``` + +The search stops when Total TPS grows by less than `TPS_MIN_GAIN_PCT` for +`PLATEAU_PATIENCE` consecutive points. Defaults are 2% and two points. + +- `saturation_concurrency`: first point in the final low-gain streak +- `best_tps_concurrency`: tested point with the highest Total TPS +- `stop_probe_concurrency`: last point tested to confirm the stop + +OOM stops only the current shape. The service is restarted before the next +shape. A non-OOM failure is retried once after a service restart. + +Before every measured point, the client sends `C` concurrent warm-up requests. +This keeps lazy kernel compilation and CUDA graph capture out of the measured +TTFT/TPS. `BENCH_WARMUP_MAX_REQUESTS=0` means no cap; use a positive value only +when deliberately limiting warm-up at very high concurrency. + +## Dataset requirement + +The default remains compatible with the fixed matrix baseline: + +```text +--dataset-name random --dataset-path DATASET_PATH +--random-range-ratio 1.0 +``` + +In the installed SGLang client, the upstream default `0.0` samples each length +uniformly from `1` to the requested maximum. This adaptive experiment sets +`1.0` explicitly so every point represents the fixed ISL/DSL shown in the +matrix. Override `RANDOM_RANGE_RATIO` only when a mixed-length workload is +intentional. + +`DATASET_PATH` must have at least +`SEARCH_MAX_CONCURRENCY * NUM_PROMPTS_MULTIPLIER` usable two-turn +conversations. The current `runtime/dummy_sharegpt.json` has only two and is +not valid for a real adaptive run. + +For setup smoke tests without ShareGPT, explicitly use: + +```bash +BENCH_DATASET_NAME=random-ids +``` + +This sends generated token IDs with `--tokenize-prompt`; it is convenient for +smoke tests but changes the token distribution, so do not mix it with random +ShareGPT-backed baseline results. + +## Commands + +Print the plan without starting a server: + +```bash +DRY_RUN=1 bash run_adaptive_concurrency.sh +``` + +Run one small search: + +```bash +TP_LIST="8" \ +ISL_LIST="1024" \ +DSL_LIST="128" \ +GRID_LIMIT=1 \ +SEARCH_MAX_CONCURRENCY=8 \ +DATASET_PATH=/path/to/ShareGPT.json \ +bash run_adaptive_concurrency.sh +``` + +Run the full search in tmux: + +```bash +tmux new -s vllm-adaptive +DATASET_PATH=/path/to/ShareGPT.json bash run_adaptive_concurrency.sh +``` + +Tune the stop rule when needed: + +```bash +TPS_MIN_GAIN_PCT=3.0 \ +PLATEAU_PATIENCE=2 \ +SEARCH_MAX_CONCURRENCY=512 \ +bash run_adaptive_concurrency.sh +``` + +## Outputs + +Results are written to `adaptive_results//`: + +```text +adaptive_points.jsonl +adaptive_points.csv +adaptive_summary.jsonl +adaptive_summary.csv +adaptive_summary.md +run_manifest.json +tp*/raw_outputs/ +tp*/metrics/ +tp*/logs/ +tp*/gpu_logs/ +``` + +Every successful point is accepted only when `completed == num_prompts` and +actual input/output lengths match the requested shape within configured +tolerances. Invalid workloads never participate in saturation decisions. diff --git a/experiments/dsv4_h200_vllm_tp_dp_matrix/adaptive_config.env b/experiments/dsv4_h200_vllm_tp_dp_matrix/adaptive_config.env new file mode 100644 index 0000000..4085397 --- /dev/null +++ b/experiments/dsv4_h200_vllm_tp_dp_matrix/adaptive_config.env @@ -0,0 +1,49 @@ +# Adaptive concurrency search settings. +# +# For each fixed (TP, DP, ISL, DSL), probe: +# C = start, start * multiplier, ... up to max +# and stop after Total TPS has less than TPS_MIN_GAIN_PCT meaningful growth for +# PLATEAU_PATIENCE consecutive points. + +SEARCH_START_CONCURRENCY="${SEARCH_START_CONCURRENCY:-1}" +SEARCH_MAX_CONCURRENCY="${SEARCH_MAX_CONCURRENCY:-512}" +SEARCH_MULTIPLIER="${SEARCH_MULTIPLIER:-2}" +NUM_PROMPTS_MULTIPLIER="${NUM_PROMPTS_MULTIPLIER:-5}" + +# A gain below 2% is treated as throughput saturation. Two consecutive +# low-gain points prevent one noisy measurement from stopping the search. +TPS_MIN_GAIN_PCT="${TPS_MIN_GAIN_PCT:-2.0}" +PLATEAU_PATIENCE="${PLATEAU_PATIENCE:-2}" + +# Keep the same random workload semantics as the fixed matrix baseline. +# DATASET_PATH must contain at least SEARCH_MAX_CONCURRENCY times +# NUM_PROMPTS_MULTIPLIER valid two-turn conversations. Set this explicitly to +# random-ids to use generated token IDs without a ShareGPT seed dataset. +BENCH_DATASET_NAME="${BENCH_DATASET_NAME:-random}" +# SGLang interprets 0.0 as Uniform[1, requested_len]. Use 1.0 for fixed +# ISL/DSL points; lower values intentionally benchmark a length distribution. +RANDOM_RANGE_RATIO="${RANDOM_RANGE_RATIO:-1.0}" +# Before each measured point, warm up with the same concurrency so lazy kernel +# compilation and CUDA graph capture are excluded from TTFT/TPS. 0 means no +# cap; set a positive cap only when very high-concurrency warmup is impractical. +BENCH_WARMUP_MAX_REQUESTS="${BENCH_WARMUP_MAX_REQUESTS:-0}" + +# Reject a point if the completed request count or actual token lengths do not +# match the requested workload. +INPUT_LENGTH_TOLERANCE_PCT="${INPUT_LENGTH_TOLERANCE_PCT:-5.0}" +OUTPUT_LENGTH_TOLERANCE_PCT="${OUTPUT_LENGTH_TOLERANCE_PCT:-10.0}" + +MAX_POINT_RETRIES="${MAX_POINT_RETRIES:-1}" +SERVER_RESTART_COOLDOWN_S="${SERVER_RESTART_COOLDOWN_S:-10}" +SCENARIO_TIMEOUT_S="${SCENARIO_TIMEOUT_S:-1800}" +GPU_MEM_SAMPLE_INTERVAL_S="${GPU_MEM_SAMPLE_INTERVAL_S:-1}" + +# Optional space-separated filters, useful for smoke tests: +# TP_LIST="8" ISL_LIST="1024" DSL_LIST="128" +TP_LIST="${TP_LIST:-}" +ISL_LIST="${ISL_LIST:-}" +DSL_LIST="${DSL_LIST:-}" + +DRY_RUN="${DRY_RUN:-0}" +# Counts ISL/DSL shapes per TP/DP config, not individual concurrency probes. +GRID_LIMIT="${GRID_LIMIT:-0}" diff --git a/experiments/dsv4_h200_vllm_tp_dp_matrix/config.env b/experiments/dsv4_h200_vllm_tp_dp_matrix/config.env index 246c1e1..7644eaf 100644 --- a/experiments/dsv4_h200_vllm_tp_dp_matrix/config.env +++ b/experiments/dsv4_h200_vllm_tp_dp_matrix/config.env @@ -51,7 +51,7 @@ DOCKER_CLIENT_IMAGE="${DOCKER_CLIENT_IMAGE:-lmsysorg/sglang:latest}" # Dataset used by sglang.bench_serving --dataset-name random. # The random sampler needs a ShareGPT-style JSON file locally; it falls back to # downloading from HuggingFace, which usually fails on offline H200 nodes. -DATASET_PATH="${DATASET_PATH:-${SCRIPT_DIR}/runtime/dummy_sharegpt.json}" +DATASET_PATH="${DATASET_PATH:-/data3/yy/sskj/datasets/ShareGPT_V3_unfiltered_cleaned_split.json}" # Matrix and concurrency rules are defined in matrix.json by default. MATRIX_FILE="${MATRIX_FILE:-${SCRIPT_DIR:-.}/matrix.json}" diff --git a/experiments/dsv4_h200_vllm_tp_dp_matrix/run_adaptive_concurrency.sh b/experiments/dsv4_h200_vllm_tp_dp_matrix/run_adaptive_concurrency.sh new file mode 100755 index 0000000..8b1fdd0 --- /dev/null +++ b/experiments/dsv4_h200_vllm_tp_dp_matrix/run_adaptive_concurrency.sh @@ -0,0 +1,164 @@ +#!/usr/bin/env bash +# Find the Total-TPS saturation concurrency for each vLLM TP/DP/ISL/DSL shape. +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" +# shellcheck source=/dev/null +source "${SCRIPT_DIR}/adaptive_config.env" +# shellcheck source=/dev/null +source "${SCRIPT_DIR}/../../scripts/common/adaptive_bench_lib.sh" + +ENGINE="vllm" +ENGINE_PORT="$VLLM_PORT" +RESULT_BASE="${RESULT_BASE:-${SCRIPT_DIR}/adaptive_results}" +ACTIVE_ENGINE_SERVER_LOG="" + +if [[ -x "${VENV_CLIENT}/bin/python" ]]; then + PYTHON="${VENV_CLIENT}/bin/python" +else + PYTHON="$(command -v python3)" +fi + +DOCKER_IMAGE="${DOCKER_IMAGE:-vllm/vllm-openai:latest}" +DOCKER_CLIENT_IMAGE="${DOCKER_CLIENT_IMAGE:-lmsysorg/sglang:latest}" + +engine_is_healthy() { + curl --fail --silent --show-error --max-time 5 \ + "http://127.0.0.1:${ENGINE_PORT}/health" >/dev/null 2>&1 +} + +engine_stop_server() { + local tp="$1" + local dp="$2" + local pid_file="${RUNTIME_BASE}/${EXPERIMENT}_vllm_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 vllm 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 + docker rm -f "${EXPERIMENT}_vllm_tp${tp}_dp${dp}" >/dev/null 2>&1 || true + ACTIVE_ENGINE_SERVER_LOG="" + sleep 2 +} + +engine_build_server_args() { + local tp="$1" + local dp="$2" + local -a args=( + vllm serve "$MODEL_PATH" + --trust-remote-code + --kv-cache-dtype "$KV_CACHE_DTYPE" + --block-size "$BLOCK_SIZE" + --tensor-parallel-size "$tp" + --gpu-memory-utilization "$GPU_MEMORY_UTILIZATION" + --max-model-len "$MAX_MODEL_LEN" + --max-num-seqs "$MAX_NUM_SEQS" + --no-enable-flashinfer-autotune + --host 0.0.0.0 + --port "$ENGINE_PORT" + ) + if (( dp > 1 )); then + args+=(--data-parallel-size "$dp") + fi + printf '%q ' "${args[@]}" +} + +engine_start_server() { + local tp="$1" + local dp="$2" + local outer_log="${ADAPTIVE_LOG_DIR}/vllm_tp${tp}_dp${dp}.server.outer.log" + log "starting vllm server tp=${tp} dp=${dp}" + bash "${SCRIPT_DIR}/start_vllm_dp.sh" "$tp" "$dp" >> "$outer_log" 2>&1 + if ! engine_is_healthy; then + log "ERROR: vllm health check failed tp=${tp} dp=${dp}" + return 1 + fi + ACTIVE_ENGINE_SERVER_LOG="$( + find "${RUNTIME_BASE}/logs" -maxdepth 1 -type f \ + -name "${EXPERIMENT}_vllm*tp${tp}_dp${dp}_*.log" \ + -printf '%T@ %p\n' 2>/dev/null | sort -nr | head -n 1 | cut -d' ' -f2- + )" + log "vllm server healthy tp=${tp} dp=${dp} log=${ACTIVE_ENGINE_SERVER_LOG:-unknown}" +} + +engine_detect_oom() { + local detail_log="$1" + local tp="$2" + local dp="$3" + local pattern='CUDA out of memory|torch\.OutOfMemoryError|OutOfMemory|out of memory|OOM|RESOURCE_EXHAUSTED|Failed to allocate memory' + local outer_log="${ADAPTIVE_LOG_DIR}/vllm_tp${tp}_dp${dp}.server.outer.log" + local -a logs=("$detail_log" "$outer_log") + if [[ -n "$ACTIVE_ENGINE_SERVER_LOG" ]]; then + logs+=("$ACTIVE_ENGINE_SERVER_LOG") + fi + grep -Eiq "$pattern" "${logs[@]}" 2>/dev/null +} + +engine_run_bench() { + local isl="$1" + local dsl="$2" + local concurrency="$3" + local num_prompts="$4" + local output_file="$5" + local warmup_requests + warmup_requests="$(adaptive_warmup_request_count "$concurrency")" + local -a bench_args=( + --backend vllm + --host 127.0.0.1 + --port "$ENGINE_PORT" + --dataset-name "$BENCH_DATASET_NAME" + --random-input-len "$isl" + --random-output-len "$dsl" + --random-range-ratio "$RANDOM_RANGE_RATIO" + --num-prompts "$num_prompts" + --max-concurrency "$concurrency" + --request-rate 10000 + --warmup-requests "$warmup_requests" + --output-file "$output_file" + --output-details + --disable-tqdm + ) + if [[ "$BENCH_DATASET_NAME" == "random" ]]; then + bench_args+=(--dataset-path "$DATASET_PATH") + else + bench_args+=(--tokenize-prompt) + fi + + if [[ "$USE_DOCKER_CLIENT" == "1" ]]; then + local -a volume_args=(-v "${MODEL_PATH}:${MODEL_PATH}:ro" -v "${RESULT_BASE}:${RESULT_BASE}") + if [[ "$BENCH_DATASET_NAME" == "random" ]]; then + volume_args+=(-v "${DATASET_PATH}:${DATASET_PATH}:ro") + fi + docker run --rm \ + --network host \ + "${volume_args[@]}" \ + -e PYTHONUNBUFFERED=1 \ + -e HF_HUB_OFFLINE=1 \ + -e TRANSFORMERS_OFFLINE=1 \ + "$DOCKER_CLIENT_IMAGE" \ + python -m sglang.bench_serving "${bench_args[@]}" + else + "$PYTHON" -m sglang.bench_serving "${bench_args[@]}" + fi +} + +export -f engine_run_bench +export ENGINE_PORT MODEL_PATH RESULT_BASE DOCKER_CLIENT_IMAGE USE_DOCKER_CLIENT +export BENCH_DATASET_NAME DATASET_PATH RANDOM_RANGE_RATIO BENCH_WARMUP_MAX_REQUESTS PYTHON + +adaptive_main "$@" diff --git a/scripts/common/adaptive_bench_lib.sh b/scripts/common/adaptive_bench_lib.sh new file mode 100755 index 0000000..3f8b158 --- /dev/null +++ b/scripts/common/adaptive_bench_lib.sh @@ -0,0 +1,682 @@ +#!/usr/bin/env bash +# Shared adaptive-concurrency search loop for serving benchmarks. + +ADAPTIVE_COMMON_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +ADAPTIVE_HELPER="${ADAPTIVE_COMMON_DIR}/adaptive_concurrency.py" + +adaptive_timestamp() { + date --iso-8601=seconds +} + +adaptive_json_number_or_null() { + local value="${1:-}" + if [[ -n "$value" ]]; then + printf '%s' "$value" + else + printf 'null' + fi +} + +adaptive_warmup_request_count() { + local concurrency="$1" + if (( BENCH_WARMUP_MAX_REQUESTS > 0 && concurrency > BENCH_WARMUP_MAX_REQUESTS )); then + printf '%s' "$BENCH_WARMUP_MAX_REQUESTS" + else + printf '%s' "$concurrency" + fi +} +export -f adaptive_warmup_request_count + +adaptive_validate_config() { + local name + for name in SEARCH_START_CONCURRENCY SEARCH_MAX_CONCURRENCY SEARCH_MULTIPLIER \ + NUM_PROMPTS_MULTIPLIER PLATEAU_PATIENCE MAX_POINT_RETRIES \ + BENCH_WARMUP_MAX_REQUESTS; do + if ! [[ "${!name}" =~ ^[0-9]+$ ]]; then + log "ERROR: ${name} must be a non-negative integer, got ${!name}" + return 1 + fi + done + if (( SEARCH_START_CONCURRENCY < 1 )); then + log "ERROR: SEARCH_START_CONCURRENCY must be at least 1" + return 1 + fi + if (( SEARCH_MAX_CONCURRENCY < SEARCH_START_CONCURRENCY )); then + log "ERROR: SEARCH_MAX_CONCURRENCY must be >= SEARCH_START_CONCURRENCY" + return 1 + fi + if (( SEARCH_MULTIPLIER < 2 )); then + log "ERROR: SEARCH_MULTIPLIER must be at least 2" + return 1 + fi + if (( NUM_PROMPTS_MULTIPLIER < 1 || PLATEAU_PATIENCE < 1 )); then + log "ERROR: NUM_PROMPTS_MULTIPLIER and PLATEAU_PATIENCE must be at least 1" + return 1 + fi + if ! [[ "$TPS_MIN_GAIN_PCT" =~ ^[0-9]+([.][0-9]+)?$ ]]; then + log "ERROR: TPS_MIN_GAIN_PCT must be a non-negative number" + return 1 + fi + if ! [[ "$RANDOM_RANGE_RATIO" =~ ^[0-9]+([.][0-9]+)?$ ]] || \ + ! awk -v ratio="$RANDOM_RANGE_RATIO" 'BEGIN { exit !(ratio >= 0 && ratio <= 1) }'; then + log "ERROR: RANDOM_RANGE_RATIO must be between 0.0 and 1.0" + return 1 + fi +} + +adaptive_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 $! +} + +adaptive_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 +} + +adaptive_warmup() { + local log_path="$1" + log "running synthetic warmup" + timeout "$SCENARIO_TIMEOUT_S" bash -c \ + 'engine_run_bench 1024 128 1 1 /dev/null' \ + >> "$log_path" 2>&1 +} + +adaptive_restart_server() { + local tp="$1" + local dp="$2" + log "restarting ${ENGINE} server tp=${tp} dp=${dp}" + engine_stop_server "$tp" "$dp" + sleep "$SERVER_RESTART_COOLDOWN_S" + engine_start_server "$tp" "$dp" + adaptive_warmup "${ADAPTIVE_LOG_DIR}/${ENGINE}_tp${tp}_dp${dp}.warmup.log" +} + +adaptive_append_point_from_metrics() { + local metrics_file="$1" + local tp="$2" + local dp="$3" + local mark="$4" + local isl="$5" + local dsl="$6" + local concurrency="$7" + local num_prompts="$8" + local attempt="$9" + local gain_pct="${10:-}" + local plateau_streak="${11:-0}" + local raw_file="${12}" + local detail_log="${13}" + local gain_json + gain_json="$(adaptive_json_number_or_null "$gain_pct")" + + jq -c \ + --arg timestamp "$(adaptive_timestamp)" \ + --arg engine "$ENGINE" \ + --argjson tp "$tp" \ + --argjson dp "$dp" \ + --arg mark "$mark" \ + --argjson isl "$isl" \ + --argjson dsl "$dsl" \ + --argjson concurrency "$concurrency" \ + --argjson num_prompts "$num_prompts" \ + --argjson warmup_requests "$(adaptive_warmup_request_count "$concurrency")" \ + --argjson attempt "$attempt" \ + --argjson gain_pct "$gain_json" \ + --argjson plateau_streak "$plateau_streak" \ + --arg raw_file "$raw_file" \ + --arg detail_log "$detail_log" \ + '. + { + timestamp: $timestamp, + engine: $engine, + tp: $tp, + dp: $dp, + mark: $mark, + isl: $isl, + dsl: $dsl, + concurrency: $concurrency, + num_prompts: $num_prompts, + warmup_requests: $warmup_requests, + attempt: $attempt, + gain_pct: $gain_pct, + plateau_streak: $plateau_streak, + error_type: "", + raw_file: $raw_file, + detail_log: $detail_log + }' "$metrics_file" >> "$ADAPTIVE_POINTS_JSONL" +} + +adaptive_append_failed_point() { + local tp="$1" + local dp="$2" + local mark="$3" + local isl="$4" + local dsl="$5" + local concurrency="$6" + local num_prompts="$7" + local attempt="$8" + local status="$9" + local error_type="${10}" + local raw_file="${11}" + local detail_log="${12}" + + jq -nc \ + --arg timestamp "$(adaptive_timestamp)" \ + --arg engine "$ENGINE" \ + --argjson tp "$tp" \ + --argjson dp "$dp" \ + --arg mark "$mark" \ + --argjson isl "$isl" \ + --argjson dsl "$dsl" \ + --argjson concurrency "$concurrency" \ + --argjson num_prompts "$num_prompts" \ + --argjson warmup_requests "$(adaptive_warmup_request_count "$concurrency")" \ + --argjson attempt "$attempt" \ + --arg status "$status" \ + --arg error_type "$error_type" \ + --arg raw_file "$raw_file" \ + --arg detail_log "$detail_log" \ + '{ + timestamp: $timestamp, + engine: $engine, + tp: $tp, + dp: $dp, + mark: $mark, + isl: $isl, + dsl: $dsl, + concurrency: $concurrency, + num_prompts: $num_prompts, + warmup_requests: $warmup_requests, + attempt: $attempt, + status: $status, + completed: 0, + failed: $num_prompts, + error_type: $error_type, + validation_errors: [], + gain_pct: null, + plateau_streak: 0, + raw_file: $raw_file, + detail_log: $detail_log + }' >> "$ADAPTIVE_POINTS_JSONL" +} + +adaptive_append_shape_summary() { + local tp="$1" + local dp="$2" + local mark="$3" + local isl="$4" + local dsl="$5" + local status="$6" + local stop_reason="$7" + local tested_points="$8" + local max_successful_concurrency="$9" + local saturation_concurrency="${10:-}" + local stop_probe_concurrency="${11:-}" + local best_tps_concurrency="${12:-}" + local best_total_tps="${13:-}" + local last_total_tps="${14:-}" + local saturation_json stop_probe_json best_c_json best_tps_json last_tps_json + saturation_json="$(adaptive_json_number_or_null "$saturation_concurrency")" + stop_probe_json="$(adaptive_json_number_or_null "$stop_probe_concurrency")" + best_c_json="$(adaptive_json_number_or_null "$best_tps_concurrency")" + best_tps_json="$(adaptive_json_number_or_null "$best_total_tps")" + last_tps_json="$(adaptive_json_number_or_null "$last_total_tps")" + + jq -nc \ + --arg timestamp "$(adaptive_timestamp)" \ + --arg engine "$ENGINE" \ + --argjson tp "$tp" \ + --argjson dp "$dp" \ + --arg mark "$mark" \ + --argjson isl "$isl" \ + --argjson dsl "$dsl" \ + --arg status "$status" \ + --arg stop_reason "$stop_reason" \ + --argjson tested_points "$tested_points" \ + --argjson search_cap "$SEARCH_MAX_CONCURRENCY" \ + --argjson max_successful_concurrency "$max_successful_concurrency" \ + --argjson saturation_concurrency "$saturation_json" \ + --argjson stop_probe_concurrency "$stop_probe_json" \ + --argjson best_tps_concurrency "$best_c_json" \ + --argjson best_total_tps "$best_tps_json" \ + --argjson last_total_tps "$last_tps_json" \ + '{ + timestamp: $timestamp, + engine: $engine, + tp: $tp, + dp: $dp, + mark: $mark, + isl: $isl, + dsl: $dsl, + status: $status, + stop_reason: $stop_reason, + tested_points: $tested_points, + search_cap: $search_cap, + max_successful_concurrency: $max_successful_concurrency, + saturation_concurrency: $saturation_concurrency, + stop_probe_concurrency: $stop_probe_concurrency, + best_tps_concurrency: $best_tps_concurrency, + best_total_tps: $best_total_tps, + last_total_tps: $last_total_tps + }' >> "$ADAPTIVE_SHAPES_JSONL" +} + +adaptive_run_point() { + local tp="$1" + local dp="$2" + local isl="$3" + local dsl="$4" + local concurrency="$5" + local num_prompts="$6" + local config_root="$7" + local attempt bench_rc parse_rc gpu_pid + + POINT_RAW_FILE="" + POINT_DETAIL_LOG="" + POINT_METRICS_FILE="" + POINT_ATTEMPT=0 + POINT_ERROR_TYPE="" + + for (( attempt = 1; attempt <= MAX_POINT_RETRIES + 1; attempt++ )); do + POINT_ATTEMPT="$attempt" + POINT_RAW_FILE="${config_root}/raw_outputs/${ENGINE}_adaptive_c${concurrency}_i${isl}_o${dsl}_a${attempt}.jsonl" + POINT_DETAIL_LOG="${config_root}/logs/${ENGINE}_c${concurrency}_i${isl}_o${dsl}_a${attempt}.log" + POINT_METRICS_FILE="${config_root}/metrics/${ENGINE}_c${concurrency}_i${isl}_o${dsl}_a${attempt}.json" + local gpu_csv="${config_root}/gpu_logs/gpu_c${concurrency}_i${isl}_o${dsl}_a${attempt}.csv" + + log "probe ${ENGINE} tp=${tp} dp=${dp} isl=${isl} dsl=${dsl} c=${concurrency} warmup=$(adaptive_warmup_request_count "$concurrency") n=${num_prompts} attempt=${attempt}" + gpu_pid="$(adaptive_start_gpu_monitor "$gpu_csv")" + bench_rc=0 + timeout "$SCENARIO_TIMEOUT_S" bash -c \ + 'engine_run_bench "$@"' _ \ + "$isl" "$dsl" "$concurrency" "$num_prompts" "$POINT_RAW_FILE" \ + > "$POINT_DETAIL_LOG" 2>&1 || bench_rc=$? + adaptive_stop_gpu_monitor "$gpu_pid" + + if (( bench_rc == 0 )); then + parse_rc=0 + "$PYTHON" "$ADAPTIVE_HELPER" parse-result \ + --input "$POINT_RAW_FILE" \ + --output "$POINT_METRICS_FILE" \ + --expected-prompts "$num_prompts" \ + --isl "$isl" \ + --dsl "$dsl" \ + --input-tolerance-pct "$INPUT_LENGTH_TOLERANCE_PCT" \ + --output-tolerance-pct "$OUTPUT_LENGTH_TOLERANCE_PCT" \ + >> "$POINT_DETAIL_LOG" 2>&1 || parse_rc=$? + if (( parse_rc == 0 )); then + return 0 + fi + POINT_ERROR_TYPE="workload validation failed" + return 21 + fi + + if engine_detect_oom "$POINT_DETAIL_LOG" "$tp" "$dp"; then + POINT_ERROR_TYPE="detected out of memory" + return 20 + fi + + POINT_ERROR_TYPE="benchmark rc=${bench_rc}" + if (( attempt <= MAX_POINT_RETRIES )); then + log "probe failed without OOM; restarting server before retry" + if ! adaptive_restart_server "$tp" "$dp"; then + POINT_ERROR_TYPE="benchmark failed and server restart failed" + return 22 + fi + continue + fi + return 22 + done +} + +adaptive_run_shape() { + local tp="$1" + local dp="$2" + local mark="$3" + local isl="$4" + local dsl="$5" + local config_root="$6" + local concurrency="$SEARCH_START_CONCURRENCY" + local previous_tps="" + local current_tps="" + local best_tps="" + local best_concurrency="" + local max_successful_concurrency=0 + local last_total_tps="" + local plateau_streak=0 + local plateau_start_concurrency="" + local saturation_concurrency="" + local stop_probe_concurrency="" + local stop_reason="SEARCH_CAP_REACHED" + local shape_status="COMPLETED" + local tested_points=0 + local gain_pct="" + local meaningful_gain=1 + local point_rc=0 + + ADAPTIVE_RESTART_NEEDED=0 + ADAPTIVE_FATAL_WORKLOAD=0 + + while (( concurrency <= SEARCH_MAX_CONCURRENCY )); do + local num_prompts=$((concurrency * NUM_PROMPTS_MULTIPLIER)) + point_rc=0 + adaptive_run_point "$tp" "$dp" "$isl" "$dsl" "$concurrency" "$num_prompts" "$config_root" || point_rc=$? + tested_points=$((tested_points + 1)) + stop_probe_concurrency="$concurrency" + + if (( point_rc == 0 )); then + current_tps="$(jq -r '.total_tps' "$POINT_METRICS_FILE")" + gain_pct="" + meaningful_gain=1 + if [[ -n "$previous_tps" ]]; then + IFS=$'\t' read -r gain_pct meaningful_gain < <( + "$PYTHON" "$ADAPTIVE_HELPER" gain \ + --previous "$previous_tps" \ + --current "$current_tps" \ + --threshold-pct "$TPS_MIN_GAIN_PCT" + ) + if (( meaningful_gain == 1 )); then + plateau_streak=0 + plateau_start_concurrency="" + else + plateau_streak=$((plateau_streak + 1)) + if [[ -z "$plateau_start_concurrency" ]]; then + plateau_start_concurrency="$concurrency" + fi + fi + fi + + if [[ -z "$best_tps" ]] || awk -v current="$current_tps" -v best="$best_tps" 'BEGIN { exit !(current > best) }'; then + best_tps="$current_tps" + best_concurrency="$concurrency" + fi + max_successful_concurrency="$concurrency" + last_total_tps="$current_tps" + adaptive_append_point_from_metrics \ + "$POINT_METRICS_FILE" "$tp" "$dp" "$mark" "$isl" "$dsl" \ + "$concurrency" "$num_prompts" "$POINT_ATTEMPT" "$gain_pct" \ + "$plateau_streak" "$POINT_RAW_FILE" "$POINT_DETAIL_LOG" + + log "probe result c=${concurrency} total_tps=${current_tps} gain_pct=${gain_pct:-first} plateau_streak=${plateau_streak}/${PLATEAU_PATIENCE}" + if (( plateau_streak >= PLATEAU_PATIENCE )); then + saturation_concurrency="$plateau_start_concurrency" + stop_reason="TPS_PLATEAU" + break + fi + previous_tps="$current_tps" + elif (( point_rc == 20 )); then + adaptive_append_failed_point \ + "$tp" "$dp" "$mark" "$isl" "$dsl" "$concurrency" "$num_prompts" \ + "$POINT_ATTEMPT" "OOM" "$POINT_ERROR_TYPE" "$POINT_RAW_FILE" "$POINT_DETAIL_LOG" + shape_status="OOM_BOUNDARY" + stop_reason="OOM" + ADAPTIVE_RESTART_NEEDED=1 + break + elif (( point_rc == 21 )); then + adaptive_append_point_from_metrics \ + "$POINT_METRICS_FILE" "$tp" "$dp" "$mark" "$isl" "$dsl" \ + "$concurrency" "$num_prompts" "$POINT_ATTEMPT" "" 0 \ + "$POINT_RAW_FILE" "$POINT_DETAIL_LOG" + shape_status="INVALID_WORKLOAD" + stop_reason="INVALID_WORKLOAD" + ADAPTIVE_FATAL_WORKLOAD=1 + break + else + adaptive_append_failed_point \ + "$tp" "$dp" "$mark" "$isl" "$dsl" "$concurrency" "$num_prompts" \ + "$POINT_ATTEMPT" "FAILED" "$POINT_ERROR_TYPE" "$POINT_RAW_FILE" "$POINT_DETAIL_LOG" + shape_status="FAILED" + stop_reason="BENCHMARK_FAILED" + ADAPTIVE_RESTART_NEEDED=1 + break + fi + + if (( concurrency == SEARCH_MAX_CONCURRENCY )); then + stop_reason="SEARCH_CAP_REACHED" + break + fi + local next_concurrency=$((concurrency * SEARCH_MULTIPLIER)) + if (( next_concurrency > SEARCH_MAX_CONCURRENCY )); then + next_concurrency="$SEARCH_MAX_CONCURRENCY" + fi + concurrency="$next_concurrency" + done + + adaptive_append_shape_summary \ + "$tp" "$dp" "$mark" "$isl" "$dsl" "$shape_status" "$stop_reason" \ + "$tested_points" "$max_successful_concurrency" "$saturation_concurrency" \ + "$stop_probe_concurrency" "$best_concurrency" "$best_tps" "$last_total_tps" + + log "shape done tp=${tp} dp=${dp} isl=${isl} dsl=${dsl} stop=${stop_reason} saturation_c=${saturation_concurrency:-none} best_c=${best_concurrency:-none} best_total_tps=${best_tps:-none}" +} + +adaptive_matches_filter() { + local value="$1" + local filter="${2:-}" + [[ -z "$filter" || " $filter " == *" $value "* ]] +} + +adaptive_print_sequence() { + local concurrency="$SEARCH_START_CONCURRENCY" + local -a sequence=() + while (( concurrency <= SEARCH_MAX_CONCURRENCY )); do + sequence+=("$concurrency") + if (( concurrency == SEARCH_MAX_CONCURRENCY )); then + break + fi + local next_concurrency=$((concurrency * SEARCH_MULTIPLIER)) + if (( next_concurrency > SEARCH_MAX_CONCURRENCY )); then + next_concurrency="$SEARCH_MAX_CONCURRENCY" + fi + concurrency="$next_concurrency" + done + printf '%s ' "${sequence[@]}" +} + +adaptive_cleanup_active_server() { + if [[ -n "${ADAPTIVE_ACTIVE_TP:-}" && -n "${ADAPTIVE_ACTIVE_DP:-}" ]]; then + engine_stop_server "$ADAPTIVE_ACTIVE_TP" "$ADAPTIVE_ACTIVE_DP" || true + ADAPTIVE_ACTIVE_TP="" + ADAPTIVE_ACTIVE_DP="" + fi +} + +adaptive_main() { + RUN_ID="${RUN_ID:-adaptive_$(date '+%Y%m%d-%H%M%S')}" + RESULT_BASE="${RESULT_BASE:-${SCRIPT_DIR}/adaptive_results}" + MATRIX_FILE="${MATRIX_FILE:-${SCRIPT_DIR}/matrix.json}" + MATRIX_MODE="${MATRIX_MODE:-Y}" + ADAPTIVE_RUN_ROOT="${RESULT_BASE}/${RUN_ID}" + ADAPTIVE_LOG_DIR="${ADAPTIVE_RUN_ROOT}/logs" + ADAPTIVE_POINTS_JSONL="${ADAPTIVE_RUN_ROOT}/adaptive_points.jsonl" + ADAPTIVE_SHAPES_JSONL="${ADAPTIVE_RUN_ROOT}/adaptive_shapes.jsonl" + local shapes_tsv="${ADAPTIVE_RUN_ROOT}/shapes.tsv" + mkdir -p "$ADAPTIVE_LOG_DIR" + : > "$ADAPTIVE_POINTS_JSONL" + : > "$ADAPTIVE_SHAPES_JSONL" + log_init "${ADAPTIVE_LOG_DIR}/orchestrator.log" + + adaptive_validate_config + "$PYTHON" "$ADAPTIVE_HELPER" shapes \ + --matrix "$MATRIX_FILE" \ + --mode "$MATRIX_MODE" \ + > "$shapes_tsv" + + local tokenize_prompt_json=true + if [[ "$BENCH_DATASET_NAME" == "random" ]]; then + local required_dataset_rows=$((SEARCH_MAX_CONCURRENCY * NUM_PROMPTS_MULTIPLIER)) + local dataset_capacity=0 + local dataset_error="" + if [[ ! -f "$DATASET_PATH" ]]; then + dataset_error="DATASET_PATH does not exist: ${DATASET_PATH}" + elif ! dataset_capacity="$( + "$PYTHON" "$ADAPTIVE_HELPER" dataset-capacity --path "$DATASET_PATH" 2>/dev/null + )"; then + dataset_error="DATASET_PATH is not a valid ShareGPT JSON file: ${DATASET_PATH}" + fi + if [[ -n "$dataset_error" ]]; then + if [[ "$DRY_RUN" == "1" ]]; then + log "WARNING: ${dataset_error}" + else + log "ERROR: ${dataset_error}" + return 1 + fi + fi + if [[ -z "$dataset_error" ]] && (( dataset_capacity < required_dataset_rows )); then + if [[ "$DRY_RUN" == "1" ]]; then + log "WARNING: DATASET_PATH has ${dataset_capacity} usable conversations, but a real run may request ${required_dataset_rows}." + else + log "ERROR: DATASET_PATH has ${dataset_capacity} usable conversations, but adaptive search may request ${required_dataset_rows}. Set DATASET_PATH to a larger ShareGPT file or explicitly use BENCH_DATASET_NAME=random-ids." + return 1 + fi + fi + tokenize_prompt_json=false + elif [[ "$BENCH_DATASET_NAME" != random* ]]; then + log "ERROR: BENCH_DATASET_NAME must be random or start with random" + return 1 + fi + + log "experiment=${EXPERIMENT_NAME} engine=${ENGINE} run_id=${RUN_ID}" + log "search_start=${SEARCH_START_CONCURRENCY} search_cap=${SEARCH_MAX_CONCURRENCY} multiplier=${SEARCH_MULTIPLIER} min_gain_pct=${TPS_MIN_GAIN_PCT} patience=${PLATEAU_PATIENCE} prompts_multiplier=${NUM_PROMPTS_MULTIPLIER} warmup_cap=${BENCH_WARMUP_MAX_REQUESTS}" + log "dataset=${BENCH_DATASET_NAME} random_range_ratio=${RANDOM_RANGE_RATIO} tokenize_prompt=${tokenize_prompt_json} matrix=${MATRIX_FILE} mode=${MATRIX_MODE}" + + jq -n \ + --arg experiment "$EXPERIMENT_NAME" \ + --arg engine "$ENGINE" \ + --arg run_id "$RUN_ID" \ + --arg model "$MODEL_PATH" \ + --arg hardware "$HARDWARE" \ + --arg matrix "$MATRIX_FILE" \ + --arg dataset "$BENCH_DATASET_NAME" \ + --argjson tokenize_prompt "$tokenize_prompt_json" \ + --argjson random_range_ratio "$RANDOM_RANGE_RATIO" \ + --argjson search_start "$SEARCH_START_CONCURRENCY" \ + --argjson search_cap "$SEARCH_MAX_CONCURRENCY" \ + --argjson search_multiplier "$SEARCH_MULTIPLIER" \ + --argjson prompts_multiplier "$NUM_PROMPTS_MULTIPLIER" \ + --argjson min_gain_pct "$TPS_MIN_GAIN_PCT" \ + --argjson plateau_patience "$PLATEAU_PATIENCE" \ + --argjson warmup_max_requests "$BENCH_WARMUP_MAX_REQUESTS" \ + '{ + experiment: $experiment, + engine: $engine, + run_id: $run_id, + model: $model, + hardware: $hardware, + matrix: $matrix, + dataset: $dataset, + tokenize_prompt: $tokenize_prompt, + random_range_ratio: $random_range_ratio, + search: { + start_concurrency: $search_start, + max_concurrency: $search_cap, + multiplier: $search_multiplier, + num_prompts_multiplier: $prompts_multiplier, + min_tps_gain_pct: $min_gain_pct, + plateau_patience: $plateau_patience, + warmup_max_requests: $warmup_max_requests + } + }' > "${ADAPTIVE_RUN_ROOT}/run_manifest.json" + + ADAPTIVE_ACTIVE_TP="" + ADAPTIVE_ACTIVE_DP="" + trap adaptive_cleanup_active_server EXIT + trap 'adaptive_cleanup_active_server; exit 130' INT + trap 'adaptive_cleanup_active_server; exit 143' TERM + + local abort_all=0 + local cfg tp dp config_root server_args shape_count mark isl dsl + for cfg in "${PARALLEL_CONFIGS[@]}"; do + read -r tp dp <<< "$cfg" + if ! adaptive_matches_filter "$tp" "$TP_LIST"; then + continue + fi + config_root="${ADAPTIVE_RUN_ROOT}/tp${tp}_dp${dp}" + mkdir -p "${config_root}/raw_outputs" "${config_root}/metrics" \ + "${config_root}/logs" "${config_root}/gpu_logs" + server_args="$(engine_build_server_args "$tp" "$dp")" + printf '%s\n' "$server_args" > "${config_root}/server_cmd.txt" + + if [[ "$DRY_RUN" == "1" ]]; then + log "DRY_RUN server tp=${tp} dp=${dp}: ${server_args}" + shape_count=0 + while IFS=$'\t' read -r mark isl dsl; do + [[ "$mark" == "mark" ]] && continue + adaptive_matches_filter "$isl" "$ISL_LIST" || continue + adaptive_matches_filter "$dsl" "$DSL_LIST" || continue + if (( GRID_LIMIT > 0 && shape_count >= GRID_LIMIT )); then + break + fi + shape_count=$((shape_count + 1)) + log "DRY_RUN tp=${tp} dp=${dp} isl=${isl} dsl=${dsl} C=[$(adaptive_print_sequence)]" + done < "$shapes_tsv" + continue + fi + + ADAPTIVE_ACTIVE_TP="$tp" + ADAPTIVE_ACTIVE_DP="$dp" + if ! engine_start_server "$tp" "$dp"; then + log "ERROR: failed to start ${ENGINE} tp=${tp} dp=${dp}; skipping config" + adaptive_cleanup_active_server + continue + fi + adaptive_warmup "${ADAPTIVE_LOG_DIR}/${ENGINE}_tp${tp}_dp${dp}.warmup.log" || { + log "ERROR: warmup failed for tp=${tp} dp=${dp}; skipping config" + adaptive_cleanup_active_server + continue + } + + shape_count=0 + while IFS=$'\t' read -r mark isl dsl; do + [[ "$mark" == "mark" ]] && continue + adaptive_matches_filter "$isl" "$ISL_LIST" || continue + adaptive_matches_filter "$dsl" "$DSL_LIST" || continue + if (( GRID_LIMIT > 0 && shape_count >= GRID_LIMIT )); then + log "GRID_LIMIT=${GRID_LIMIT} reached for tp=${tp} dp=${dp}" + break + fi + shape_count=$((shape_count + 1)) + log "===== shape START tp=${tp} dp=${dp} isl=${isl} dsl=${dsl} mark=${mark} =====" + adaptive_run_shape "$tp" "$dp" "$mark" "$isl" "$dsl" "$config_root" + + if (( ADAPTIVE_FATAL_WORKLOAD == 1 )); then + log "ERROR: invalid synthetic workload; aborting remaining searches" + abort_all=1 + break + fi + if (( ADAPTIVE_RESTART_NEEDED == 1 )); then + if ! adaptive_restart_server "$tp" "$dp"; then + log "ERROR: server restart failed; skipping remaining shapes for tp=${tp} dp=${dp}" + break + fi + fi + done < "$shapes_tsv" + + adaptive_cleanup_active_server + if (( abort_all == 1 )); then + break + fi + done + + "$PYTHON" "$ADAPTIVE_HELPER" summarize \ + --points "$ADAPTIVE_POINTS_JSONL" \ + --shapes "$ADAPTIVE_SHAPES_JSONL" \ + --output-dir "$ADAPTIVE_RUN_ROOT" \ + >> "${ADAPTIVE_LOG_DIR}/summarize.log" 2>&1 + log "adaptive results saved to ${ADAPTIVE_RUN_ROOT}" + + trap - EXIT INT TERM + adaptive_cleanup_active_server + if (( abort_all == 1 )); then + return 2 + fi +} diff --git a/scripts/common/adaptive_concurrency.py b/scripts/common/adaptive_concurrency.py new file mode 100755 index 0000000..d7a517b --- /dev/null +++ b/scripts/common/adaptive_concurrency.py @@ -0,0 +1,320 @@ +#!/usr/bin/env python3 +"""Helpers for adaptive-concurrency serving benchmarks.""" + +from __future__ import annotations + +import argparse +import csv +import json +from pathlib import Path +from typing import Any, Iterable + + +POINT_FIELDS = [ + "timestamp", + "engine", + "tp", + "dp", + "mark", + "isl", + "dsl", + "concurrency", + "num_prompts", + "warmup_requests", + "attempt", + "status", + "completed", + "failed", + "duration_s", + "request_tps", + "input_tps", + "output_tps", + "total_tps", + "mean_input_tokens", + "mean_output_tokens", + "ttft_p50_ms", + "ttft_p95_ms", + "ttft_p99_ms", + "tpot_p50_ms", + "tpot_p95_ms", + "tpot_p99_ms", + "e2e_p50_ms", + "e2e_p95_ms", + "e2e_p99_ms", + "itl_p50_ms", + "itl_p95_ms", + "itl_p99_ms", + "gain_pct", + "plateau_streak", + "error_type", + "validation_errors", + "raw_file", + "detail_log", +] + + +SHAPE_FIELDS = [ + "timestamp", + "engine", + "tp", + "dp", + "mark", + "isl", + "dsl", + "status", + "stop_reason", + "tested_points", + "search_cap", + "max_successful_concurrency", + "saturation_concurrency", + "stop_probe_concurrency", + "best_tps_concurrency", + "best_total_tps", + "last_total_tps", +] + + +def read_jsonl(path: Path) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + if not path.exists(): + return rows + with path.open("r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + value = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(value, dict): + rows.append(value) + return rows + + +def last_json_object(path: Path) -> dict[str, Any]: + rows = read_jsonl(path) + if not rows: + raise ValueError(f"no valid JSON object in {path}") + return rows[-1] + + +def percentile_value(data: dict[str, Any], prefix: str, percentile: str) -> float: + if percentile == "p50": + key = f"median_{prefix}_ms" + else: + key = f"{percentile}_{prefix}_ms" + return float(data.get(key, 0.0) or 0.0) + + +def command_shapes(args: argparse.Namespace) -> int: + with args.matrix.open("r", encoding="utf-8") as f: + data = json.load(f) + + mode = args.mode or data.get("mode", "Y") + print("mark\tinput_len\toutput_len") + for isl_text in sorted(data["matrix"], key=int): + dsl_map = data["matrix"][isl_text] + for dsl_text in sorted(dsl_map, key=int): + mark = dsl_map[dsl_text] + if mode == "Y" and mark != "Y": + continue + if mode == "Y+P" and mark not in ("Y", "P"): + continue + if mode != "all" and mark == "N": + continue + print(f"{mark}\t{int(isl_text)}\t{int(dsl_text)}") + return 0 + + +def command_dataset_capacity(args: argparse.Namespace) -> int: + try: + with args.path.open("r", encoding="utf-8") as f: + data = json.load(f) + except (OSError, json.JSONDecodeError) as exc: + raise SystemExit(f"invalid dataset {args.path}: {exc}") + if not isinstance(data, list): + raise SystemExit(f"dataset must be a JSON list: {args.path}") + capacity = 0 + for row in data: + if not isinstance(row, dict): + continue + conversations = row.get("conversations", row.get("conversation", [])) + if isinstance(conversations, list) and len(conversations) >= 2: + capacity += 1 + print(capacity) + return 0 + + +def command_parse_result(args: argparse.Namespace) -> int: + validation_errors: list[str] = [] + try: + data = last_json_object(args.input) + except (OSError, ValueError) as exc: + data = {} + validation_errors.append(str(exc)) + + completed = int(data.get("completed", 0) or 0) + input_lens = [int(v) for v in data.get("input_lens", []) if v is not None] + output_lens = [int(v) for v in data.get("output_lens", []) if v is not None] + errors = [str(v) for v in data.get("errors", []) if str(v)] + mean_input = sum(input_lens) / len(input_lens) if input_lens else 0.0 + mean_output = sum(output_lens) / len(output_lens) if output_lens else 0.0 + + if completed != args.expected_prompts: + validation_errors.append( + f"completed={completed}, expected={args.expected_prompts}" + ) + if errors: + validation_errors.append(f"request_errors={len(errors)}") + + input_low = args.isl * (1.0 - args.input_tolerance_pct / 100.0) + input_high = args.isl * (1.0 + args.input_tolerance_pct / 100.0) + if not input_low <= mean_input <= input_high: + validation_errors.append( + f"mean_input_tokens={mean_input:.2f}, expected_range={input_low:.2f}-{input_high:.2f}" + ) + + output_low = args.dsl * (1.0 - args.output_tolerance_pct / 100.0) + output_high = args.dsl * (1.0 + args.output_tolerance_pct / 100.0) + if not output_low <= mean_output <= output_high: + validation_errors.append( + f"mean_output_tokens={mean_output:.2f}, expected_range={output_low:.2f}-{output_high:.2f}" + ) + + status = "COMPLETED" if not validation_errors else "INVALID_WORKLOAD" + result = { + "status": status, + "completed": completed, + "failed": max(args.expected_prompts - completed, len(errors)), + "duration_s": float(data.get("duration", 0.0) or 0.0), + "request_tps": float(data.get("request_throughput", 0.0) or 0.0), + "input_tps": float(data.get("input_throughput", 0.0) or 0.0), + "output_tps": float(data.get("output_throughput", 0.0) or 0.0), + "total_tps": float(data.get("total_throughput", 0.0) or 0.0), + "mean_input_tokens": mean_input, + "mean_output_tokens": mean_output, + "ttft_p50_ms": percentile_value(data, "ttft", "p50"), + "ttft_p95_ms": percentile_value(data, "ttft", "p95"), + "ttft_p99_ms": percentile_value(data, "ttft", "p99"), + "tpot_p50_ms": percentile_value(data, "tpot", "p50"), + "tpot_p95_ms": percentile_value(data, "tpot", "p95"), + "tpot_p99_ms": percentile_value(data, "tpot", "p99"), + "e2e_p50_ms": percentile_value(data, "e2e_latency", "p50"), + "e2e_p95_ms": percentile_value(data, "e2e_latency", "p95"), + "e2e_p99_ms": percentile_value(data, "e2e_latency", "p99"), + "itl_p50_ms": percentile_value(data, "itl", "p50"), + "itl_p95_ms": percentile_value(data, "itl", "p95"), + "itl_p99_ms": percentile_value(data, "itl", "p99"), + "validation_errors": validation_errors, + } + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("w", encoding="utf-8") as f: + json.dump(result, f, indent=2, ensure_ascii=False) + f.write("\n") + print(status) + return 0 if status == "COMPLETED" else 2 + + +def command_gain(args: argparse.Namespace) -> int: + if args.previous <= 0: + gain_pct = float("inf") + meaningful = True + else: + gain_pct = (args.current - args.previous) / args.previous * 100.0 + meaningful = gain_pct >= args.threshold_pct + gain_text = "inf" if gain_pct == float("inf") else f"{gain_pct:.6f}" + print(f"{gain_text}\t{1 if meaningful else 0}") + return 0 + + +def write_csv(path: Path, rows: Iterable[dict[str, Any]], fields: list[str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8", newline="") as f: + writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore") + writer.writeheader() + for row in rows: + output = dict(row) + if isinstance(output.get("validation_errors"), list): + output["validation_errors"] = "; ".join(output["validation_errors"]) + writer.writerow(output) + + +def command_summarize(args: argparse.Namespace) -> int: + points = read_jsonl(args.points) + shapes = read_jsonl(args.shapes) + args.output_dir.mkdir(parents=True, exist_ok=True) + write_csv(args.output_dir / "adaptive_points.csv", points, POINT_FIELDS) + write_csv(args.output_dir / "adaptive_summary.csv", shapes, SHAPE_FIELDS) + + summary_jsonl = args.output_dir / "adaptive_summary.jsonl" + with summary_jsonl.open("w", encoding="utf-8") as f: + for row in shapes: + f.write(json.dumps(row, ensure_ascii=False) + "\n") + + report = args.output_dir / "adaptive_summary.md" + with report.open("w", encoding="utf-8") as f: + f.write("# Adaptive concurrency search summary\n\n") + f.write( + "| Engine | TP | DP | ISL | DSL | Stop | Saturation C | Best TPS C | Best Total TPS | Max successful C |\n" + ) + f.write("|---|---:|---:|---:|---:|---|---:|---:|---:|---:|\n") + for row in shapes: + f.write( + "| {engine} | {tp} | {dp} | {isl} | {dsl} | {stop_reason} | " + "{saturation_concurrency} | {best_tps_concurrency} | {best_total_tps} | " + "{max_successful_concurrency} |\n".format(**{k: row.get(k, "") for k in SHAPE_FIELDS}) + ) + f.write("\n") + f.write( + "`Saturation C` is the first point in the final low-gain streak. " + "`Best TPS C` is the tested point with the highest observed Total TPS.\n" + ) + print(f"wrote summaries under {args.output_dir}") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(dest="command", required=True) + + shapes = subparsers.add_parser("shapes") + shapes.add_argument("--matrix", type=Path, required=True) + shapes.add_argument("--mode", choices=["Y", "Y+P", "all"], default=None) + shapes.set_defaults(func=command_shapes) + + dataset_capacity = subparsers.add_parser("dataset-capacity") + dataset_capacity.add_argument("--path", type=Path, required=True) + dataset_capacity.set_defaults(func=command_dataset_capacity) + + parse_result = subparsers.add_parser("parse-result") + parse_result.add_argument("--input", type=Path, required=True) + parse_result.add_argument("--output", type=Path, required=True) + parse_result.add_argument("--expected-prompts", type=int, required=True) + parse_result.add_argument("--isl", type=int, required=True) + parse_result.add_argument("--dsl", type=int, required=True) + parse_result.add_argument("--input-tolerance-pct", type=float, default=5.0) + parse_result.add_argument("--output-tolerance-pct", type=float, default=10.0) + parse_result.set_defaults(func=command_parse_result) + + gain = subparsers.add_parser("gain") + gain.add_argument("--previous", type=float, required=True) + gain.add_argument("--current", type=float, required=True) + gain.add_argument("--threshold-pct", type=float, required=True) + gain.set_defaults(func=command_gain) + + summarize = subparsers.add_parser("summarize") + summarize.add_argument("--points", type=Path, required=True) + summarize.add_argument("--shapes", type=Path, required=True) + summarize.add_argument("--output-dir", type=Path, required=True) + summarize.set_defaults(func=command_summarize) + return parser + + +def main() -> int: + args = build_parser().parse_args() + return int(args.func(args)) + + +if __name__ == "__main__": + raise SystemExit(main())