diff --git a/.github/workflows/benchmark-tmpl.yml b/.github/workflows/benchmark-tmpl.yml index 8db71997d4..be89e4ef6e 100644 --- a/.github/workflows/benchmark-tmpl.yml +++ b/.github/workflows/benchmark-tmpl.yml @@ -15,6 +15,10 @@ on: # recipe ignores it, and the SPEED-Bench path falls back to HF_TOKEN. SPEEDBENCH_HF_TOKEN: required: false + # Read-only token for the private GreenSpec repo pulled by + # resolve_greenspec_dataset (benchmark_lib.sh). Optional, falls back to HF_TOKEN. + GREENSPEC_HF_TOKEN: + required: false inputs: config: description: "Benchmark configuration as JSON" @@ -169,6 +173,7 @@ env: HF_TOKEN: ${{ secrets.INFERENCEX_OFFICIAL_RO_HF_TOKEN }} HF_HUB_CACHE: '/mnt/hf_hub_cache/' SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} EXP_NAME: ${{ fromJSON(inputs.config).exp-name }} RECIPE_FINGERPRINT: ${{ fromJSON(inputs.config).recipe-fingerprint || '' }} MODEL: ${{ fromJSON(inputs.config).model }} @@ -535,6 +540,23 @@ jobs: !results/speedbench_aiperf/**/inputs.json if-no-files-found: ignore + # GreenSpec recipes (run_greenspec_aiperf) hold internal prod traffic. The + # client's guard already pruned results/greenspec_aiperf to aggregate files + # and checked them for row ids; upload by allowlist all the same, so a file + # a future aiperf adds never ships by default. + - name: Upload GreenSpec aiperf artifacts + if: ${{ always() && inputs.scenario-type != 'agentic-coding' }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: greenspec_aiperf_${{ env.RESULT_FILENAME }} + path: | + results/greenspec_aiperf/profile_export_aiperf.json + results/greenspec_aiperf/result.json + results/greenspec_aiperf/spec_metrics_before.json + results/greenspec_aiperf/spec_metrics_after.json + results/greenspec_aiperf/benchmark_command.txt + if-no-files-found: ignore + - name: Upload GPU metrics if: always() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index 78fcd37d18..256085cbb8 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.github/workflows/e2e-tests.yml @@ -118,6 +118,8 @@ on: # zizmor: ignore[concurrency-limits] required: false SPEEDBENCH_HF_TOKEN: required: false + GREENSPEC_HF_TOKEN: + required: false inputs: generate-cli-command: description: "Command passed to generate matrix script" @@ -438,6 +440,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: config: ${{ toJSON(matrix.config) }} serving: ${{ needs.get-jobs.outputs.serving }} @@ -468,6 +471,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: config: ${{ toJSON(matrix.config) }} klaud-run: ${{ inputs.klaud-run }} @@ -566,6 +570,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: config: ${{ toJSON(matrix.config) }} klaud-run: ${{ inputs.klaud-run }} @@ -592,6 +597,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: config: ${{ toJSON(matrix.config) }} klaud-run: ${{ inputs.klaud-run }} diff --git a/.github/workflows/run-sweep.yml b/.github/workflows/run-sweep.yml index 2140f2a690..69e2f0259a 100644 --- a/.github/workflows/run-sweep.yml +++ b/.github/workflows/run-sweep.yml @@ -434,6 +434,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: config: ${{ toJSON(matrix.config) }} # Only same-repository PRs owned by Klaud receive background priority. @@ -525,6 +526,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: &single-node-inputs config: ${{ toJSON(matrix.config) }} klaud-run: *klaud-run @@ -559,6 +561,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: *single-node-inputs sweep-agentic: @@ -583,6 +586,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: config: ${{ toJSON(matrix.config) }} klaud-run: *klaud-run @@ -656,6 +660,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: config: ${{ toJSON(matrix.config) }} klaud-run: *klaud-run @@ -695,6 +700,7 @@ jobs: MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }} + GREENSPEC_HF_TOKEN: ${{ secrets.GREENSPEC_HF_TOKEN }} with: config: ${{ toJSON(matrix.config) }} klaud-run: *klaud-run diff --git a/.gitmodules b/.gitmodules index f7635a307b..b39bc96d07 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,6 +1,7 @@ [submodule "utils/aiperf"] path = utils/aiperf - url = https://github.com/SemiAnalysisAI/aiperf.git + url = https://github.com/Noridom1/aiperf.git + branch = feat/greenspec-loader [submodule "utils/srt-slurm"] path = utils/srt-slurm url = https://github.com/NVIDIA/srt-slurm.git diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index bd5d1a57eb..31f6f71e3a 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -4049,3 +4049,481 @@ run_agentic_replay_and_write_outputs() ( validate_required_agentic_server_metrics "$result_dir" fi ) + + +# SPEED-Bench helpers (aiperf driver, fixed-seq-len arms) +# +# One shared client for every SPEED-Bench recipe: the pinned aiperf fork loads +# the prepared jsonl through its speed_bench_* custom-dataset loaders (full +# multi-turn, category filter, placeholder rows rejected), and +# infx.bench_serving.speedbench_aiperf maps the export onto the +# benchmark_serving result keys, so these arms ride the normal fixed-seq-len +# process_result/collect path. Engine launch stays in the recipe. + +# Prepared SPEED-Bench (all six splits, prompts resolved from their sources, +# no SPECDEC_BENCH placeholders). Pinned by commit so every arm reads identical +# bytes; SHA256SUMS in the repo is checked after download. +SPEEDBENCH_HF_REPO="${SPEEDBENCH_HF_REPO:-Noridom1/speed-bench-prepared}" +SPEEDBENCH_HF_REVISION="${SPEEDBENCH_HF_REVISION:-6e5c3a81f3d529e4daba914e1ff1046823b0165b}" + +# Resolve $SPEEDBENCH_CONFIG.jsonl (qualitative | throughput_{1k,2k,8k,16k,32k}) +# and export SPEEDBENCH_FILE. A pre-staged $SPEEDBENCH_DIR wins; otherwise the +# file comes from the pinned HF snapshot, cache first (HF_HUB_CACHE is a +# persistent mount on the runners), network second. The repo is private: +# SPEEDBENCH_HF_TOKEN is preferred, falling back to the ambient HF_TOKEN. +resolve_speedbench_dataset() { + check_env_vars SPEEDBENCH_CONFIG + case "$SPEEDBENCH_CONFIG" in + qualitative|throughput_1k|throughput_2k|throughput_8k|throughput_16k|throughput_32k) ;; + *) + echo "ERROR: unknown SPEEDBENCH_CONFIG='$SPEEDBENCH_CONFIG'" >&2 + return 1 + ;; + esac + local name="$SPEEDBENCH_CONFIG.jsonl" + + if [ -n "${SPEEDBENCH_DIR:-}" ] && [ -f "$SPEEDBENCH_DIR/$name" ]; then + export SPEEDBENCH_FILE="$SPEEDBENCH_DIR/$name" + echo "SPEED-Bench: using pre-staged $SPEEDBENCH_FILE" + return 0 + fi + + install_agentic_deps || return $? + local snapshot_dir + # Subshell with tracing off so neither token reaches the job log. + snapshot_dir=$( + set +x + export HF_TOKEN="${SPEEDBENCH_HF_TOKEN:-${HF_TOKEN:-}}" + "$AIPERF_PYTHON" - "$SPEEDBENCH_HF_REPO" "$SPEEDBENCH_HF_REVISION" "$name" <<'PY' +import sys +from huggingface_hub import snapshot_download + +repo, revision, name = sys.argv[1:4] +kwargs = dict(repo_id=repo, repo_type="dataset", revision=revision, + allow_patterns=[name, "SHA256SUMS"]) +try: + path = snapshot_download(local_files_only=True, **kwargs) +except Exception: + path = snapshot_download(**kwargs) +print(path) +PY + ) || { + echo "ERROR: could not fetch $name from $SPEEDBENCH_HF_REPO@$SPEEDBENCH_HF_REVISION;" \ + "check that SPEEDBENCH_HF_TOKEN (or HF_TOKEN) can read the repo" >&2 + return 1 + } + + if ! (cd "$snapshot_dir" && grep -E "[[:space:]]\\*?${name}\$" SHA256SUMS | sha256sum -c -); then + echo "ERROR: $snapshot_dir/$name does not match SHA256SUMS" >&2 + return 1 + fi + export SPEEDBENCH_FILE="$snapshot_dir/$name" + echo "SPEED-Bench: $SPEEDBENCH_FILE ($SPEEDBENCH_HF_REPO@$SPEEDBENCH_HF_REVISION)" +} + +# Run one SPEED-Bench cell with aiperf and write $RESULT_FILENAME.json in the +# benchmark_serving shape. Call after the server is ready. +# +# Required: MODEL PORT CONC OSL RESULT_FILENAME SPEEDBENCH_CONFIG +# Optional: +# SPEEDBENCH_CATEGORY qualitative: coding|humanities|math|multilingual|qa| +# rag|reasoning|roleplay|stem|summarization|writing; +# throughput_*: low_entropy|mixed|high_entropy. +# Unset = the whole split. +# SPEEDBENCH_NUM_PROMPTS conversations to send; default CONC*10 clamped to +# [64, rows in the selected pool]. Rows are taken in +# file order, so every arm at one concurrency sees +# the identical prompt set. +# SPEEDBENCH_IGNORE_EOS 1 (default) forces OSL output tokens per turn. +# SPEEDBENCH_EXTRA_INPUTS JSON object merged into every request body, e.g. +# '{"temperature": 0, "chat_template_kwargs": {"enable_thinking": false}}'. +# SPEEDBENCH_WARMUP_REQUESTS default 0. +# SPEEDBENCH_METRICS_URL Prometheus endpoint for acceptance counters; +# default http://localhost:$PORT/metrics. +# SPEEDBENCH_SERVER_PID stop the client if this server dies (server_watch). +# SPEEDBENCH_META whitespace-separated KEY=VALUE pairs copied into +# the result (VALUE parsed as JSON when it is). +# SERVED_MODEL_NAME, AIPERF_TOKENIZER, AIPERF_GPU_TELEMETRY_URL, RESULT_DIR. +run_speedbench_aiperf() { + if [ "${EVAL_ONLY}" = "true" ]; then + echo "EVAL_ONLY mode: skipping SPEED-Bench throughput benchmark" + return 0 + fi + check_env_vars MODEL PORT CONC OSL RESULT_FILENAME SPEEDBENCH_CONFIG INFMAX_CONTAINER_WORKSPACE + resolve_speedbench_dataset || return $? + + local category="${SPEEDBENCH_CATEGORY:-}" + local preset="speed_bench_${SPEEDBENCH_CONFIG}" + if [ -n "$category" ]; then + if [ "$SPEEDBENCH_CONFIG" = "qualitative" ]; then + preset="speed_bench_${category}" + else + preset="speed_bench_${SPEEDBENCH_CONFIG}_${category}" + fi + fi + + # The pinned loader dies ~30s in with an opaque error when a category + # filter matches nothing, so size the pool up front and fail loudly. + local pool + pool=$("$AIPERF_PYTHON" - "$SPEEDBENCH_FILE" "$category" <<'PY' +import json, sys +path, category = sys.argv[1], sys.argv[2] +with open(path) as f: + print(sum(1 for line in f if line.strip() + and (not category or json.loads(line).get("category") == category))) +PY + ) || return 1 + if [ "$pool" -le 0 ]; then + echo "ERROR: SPEED-Bench category '$category' matches no rows in $SPEEDBENCH_FILE" >&2 + return 1 + fi + local num_prompts="${SPEEDBENCH_NUM_PROMPTS:-$((CONC * 10))}" + if [ -z "${SPEEDBENCH_NUM_PROMPTS:-}" ] && [ "$num_prompts" -lt 64 ]; then num_prompts=64; fi + if [ "$num_prompts" -gt "$pool" ]; then num_prompts="$pool"; fi + + local extra_inputs + extra_inputs=$(SPEEDBENCH_IGNORE_EOS="${SPEEDBENCH_IGNORE_EOS:-1}" "$AIPERF_PYTHON" -c ' +import json, os +extra = {"ignore_eos": True} if os.environ["SPEEDBENCH_IGNORE_EOS"] == "1" else {} +extra.update(json.loads(os.environ.get("SPEEDBENCH_EXTRA_INPUTS") or "{}")) +print(json.dumps(extra, separators=(",", ":")))') || return 1 + + local result_dir="${RESULT_DIR:-$INFMAX_CONTAINER_WORKSPACE/results}" + local artifact_dir="$result_dir/speedbench_aiperf" + local metrics_url="${SPEEDBENCH_METRICS_URL:-http://localhost:$PORT/metrics}" + mkdir -p "$artifact_dir" + + local -a cmd=( + "$AIPERF_CLI" profile + --model "${SERVED_MODEL_NAME:-$MODEL}" + --tokenizer "${AIPERF_TOKENIZER:-$MODEL}" + --tokenizer-trust-remote-code + --url "http://localhost:$PORT" + --endpoint-type chat + --streaming + --custom-dataset-type "$preset" + --input-file "$SPEEDBENCH_FILE" + --dataset-sampling-strategy sequential + --conversation-num "$num_prompts" + --concurrency "$CONC" + --osl "$OSL" + --extra-inputs "$extra_inputs" + --random-seed 42 + --use-server-token-count + --server-metrics "$metrics_url" + --ui-type simple + --output-artifact-dir "$artifact_dir" + ) + if [ "${SPEEDBENCH_WARMUP_REQUESTS:-0}" -gt 0 ]; then + cmd+=(--warmup-request-count "$SPEEDBENCH_WARMUP_REQUESTS") + fi + if [ -n "${AIPERF_GPU_TELEMETRY_URL:-}" ]; then + cmd+=(--gpu-telemetry "$AIPERF_GPU_TELEMETRY_URL") + else + cmd+=(--no-gpu-telemetry) + fi + + echo "===== SPEED-Bench cell =====" + echo "preset=$preset file=$SPEEDBENCH_FILE pool=$pool num_prompts=$num_prompts" + echo "conc=$CONC osl=$OSL extra_inputs=$extra_inputs" + echo "============================" + write_command "$artifact_dir/benchmark_command.txt" "${cmd[@]}" + + local -a meta_args=() + local pair + for pair in ${SPEEDBENCH_META:-}; do + meta_args+=(--meta "$pair") + done + local py=(env "PYTHONPATH=$INFERENCEX_REPO_ROOT${PYTHONPATH:+:$PYTHONPATH}" "$AIPERF_PYTHON" -m infx.bench_serving.speedbench_aiperf) + "${py[@]}" snapshot --metrics-url "$metrics_url" --out "$artifact_dir/spec_metrics_before.json" + + if [[ -n "${SPEEDBENCH_SERVER_PID:-}" && "$SPEEDBENCH_SERVER_PID" != "${INFERENCEX_SERVER_PID:-}" ]]; then + INFERENCEX_SERVER_STATE=$(mktemp /tmp/inferencex-server-state.XXXXXX) || return 1 + PYTHONPATH="$INFERENCEX_REPO_ROOT${PYTHONPATH:+:$PYTHONPATH}" python3 -m infx.bench_serving.server_watch capture --pid "$SPEEDBENCH_SERVER_PID" \ + > "$INFERENCEX_SERVER_STATE" || return 1 + INFERENCEX_SERVER_PID="$SPEEDBENCH_SERVER_PID" + fi + local aiperf_rc=0 + run_server_client "${cmd[@]}" || aiperf_rc=$? + + "${py[@]}" snapshot --metrics-url "$metrics_url" --out "$artifact_dir/spec_metrics_after.json" + + if [ ! -f "$artifact_dir/profile_export_aiperf.json" ]; then + echo "ERROR: aiperf exited with code $aiperf_rc and wrote no profile_export_aiperf.json" >&2 + [ "$aiperf_rc" -ne 0 ] && return "$aiperf_rc" + return 1 + fi + "${py[@]}" convert \ + --aiperf-json "$artifact_dir/profile_export_aiperf.json" \ + --before "$artifact_dir/spec_metrics_before.json" \ + --after "$artifact_dir/spec_metrics_after.json" \ + --out "$INFMAX_CONTAINER_WORKSPACE/$RESULT_FILENAME.json" \ + --model "$MODEL" \ + --concurrency "$CONC" \ + --meta "sb_dataset_subset=\"$SPEEDBENCH_CONFIG\"" \ + --meta "sb_category=\"$category\"" \ + --meta "sb_aiperf_preset=\"$preset\"" \ + --meta "sb_num_conversations=$num_prompts" \ + --meta "sb_ignore_eos=${SPEEDBENCH_IGNORE_EOS:-1}" \ + --meta "sb_dataset_repo=\"$SPEEDBENCH_HF_REPO@$SPEEDBENCH_HF_REVISION\"" \ + --meta "sb_extra_inputs=$extra_inputs" \ + "${meta_args[@]}" || return $? + cp "$INFMAX_CONTAINER_WORKSPACE/$RESULT_FILENAME.json" "$artifact_dir/result.json" + return "$aiperf_rc" +} + + +# GreenSpec helpers (aiperf driver, natural-stop arms) +# +# GreenSpec is an internal speculative-decoding bench cut from production chat +# traffic (tools, multi-turn history, real system prompts). One flat +# greenspec.jsonl holds every subset; the pinned aiperf fork selects one through +# its greenspec_* presets, and infx.bench_serving.speedbench_aiperf maps the +# export onto the benchmark_serving keys as for SPEED-Bench. Unlike SPEED-Bench, +# output length is not forced: no --osl and no ignore_eos, each request stops on +# its own (capped by GREENSPEC_MAX_TOKENS). +# +# The data is internal and this repo's artifacts are public: the run ends with +# `greenspec_aiperf guard`, which keeps only an allowlist of aggregate files, +# withholds the server log and fails the job if a row id is left anywhere it +# would be uploaded. Never add --export-level raw or vLLM request logging to a +# GreenSpec recipe. + +# Private, pinned by commit (the v1.1.0 tag); SHA256SUMS in the repo is checked +# after download. +GREENSPEC_HF_REPO="${GREENSPEC_HF_REPO:-Green-Lab/GreenSpec}" +GREENSPEC_HF_REVISION="${GREENSPEC_HF_REVISION:-f76978679a941e21eb701e7e046a0038a94efa7b}" + +# Resolve greenspec.jsonl and export GREENSPEC_FILE. A pre-staged $GREENSPEC_DIR +# (holding greenspec.jsonl) wins; otherwise the pinned HF snapshot, cache first, +# network second. Token: GREENSPEC_HF_TOKEN, falling back to HF_TOKEN. +resolve_greenspec_dataset() { + local name="greenspec.jsonl" + if [ -n "${GREENSPEC_DIR:-}" ] && [ -f "$GREENSPEC_DIR/$name" ]; then + export GREENSPEC_FILE="$GREENSPEC_DIR/$name" + echo "GreenSpec: using pre-staged $GREENSPEC_FILE" + return 0 + fi + + install_agentic_deps || return $? + local snapshot_dir + # Subshell with tracing off so neither token reaches the job log. + snapshot_dir=$( + set +x + export HF_TOKEN="${GREENSPEC_HF_TOKEN:-${HF_TOKEN:-}}" + "$AIPERF_PYTHON" - "$GREENSPEC_HF_REPO" "$GREENSPEC_HF_REVISION" "$name" <<'PY' +import sys +from huggingface_hub import snapshot_download + +repo, revision, name = sys.argv[1:4] +kwargs = dict(repo_id=repo, repo_type="dataset", revision=revision, + allow_patterns=[name, "SHA256SUMS"]) +try: + path = snapshot_download(local_files_only=True, **kwargs) +except Exception: + path = snapshot_download(**kwargs) +print(path) +PY + ) || { + echo "ERROR: could not fetch $name from $GREENSPEC_HF_REPO@$GREENSPEC_HF_REVISION;" \ + "check that GREENSPEC_HF_TOKEN (or HF_TOKEN) can read the repo" >&2 + return 1 + } + + if ! (cd "$snapshot_dir" && grep -E "[[:space:]]\\*?${name}\$" SHA256SUMS | sha256sum -c -); then + echo "ERROR: $snapshot_dir/$name does not match SHA256SUMS" >&2 + return 1 + fi + export GREENSPEC_FILE="$snapshot_dir/$name" + echo "GreenSpec: $GREENSPEC_FILE ($GREENSPEC_HF_REPO@$GREENSPEC_HF_REVISION)" +} + +# Run one GreenSpec cell with aiperf and write $RESULT_FILENAME.json in the +# benchmark_serving shape. Call after the server is ready. +# +# Required: MODEL PORT CONC RESULT_FILENAME +# Optional: +# GREENSPEC_SUBSET diverse (default) | agentic_coding | agentic_tools | llm_task +# GREENSPEC_LITE 1 = the lite quarter of every bucket; default 0. +# GREENSPEC_BUCKET one bucket, e.g. agentic_tools__4k-16k. +# GREENSPEC_MODE target-default (default): messages, tools and +# response_format only, decode params from the +# server. recorded: each request's own prod params. +# GREENSPEC_MAX_TOKENS output cap per request, default 8192 (lowers a +# recorded cap, never raises it). +# GREENSPEC_EXTRA_INPUTS JSON object merged into every request body. +# GREENSPEC_NUM_PROMPTS requests to send; default the whole selection. +# Rows are interleaved across buckets, so a prefix +# stays balanced and every arm sees the same rows. +# GREENSPEC_WARMUP_REQUESTS default 0. +# GREENSPEC_METRICS_URL Prometheus endpoint; default http://localhost:$PORT/metrics. +# GREENSPEC_SERVER_PID stop the client if this server dies (server_watch). +# GREENSPEC_SERVER_LOG server log the guard withholds; default /workspace/server.log. +# GREENSPEC_KEEP_LOG_DIR runner-local dir to move that log to instead of dropping it. +# GREENSPEC_PROGRESS_INTERVAL seconds between progress lines in the job log +# (numbers only); default 60, 0 = off. +# GREENSPEC_META whitespace-separated KEY=VALUE pairs copied into the result. +# SERVED_MODEL_NAME, AIPERF_TOKENIZER, AIPERF_GPU_TELEMETRY_URL, RESULT_DIR. +run_greenspec_aiperf() { + if [ "${EVAL_ONLY}" = "true" ]; then + echo "EVAL_ONLY mode: skipping GreenSpec benchmark" + return 0 + fi + check_env_vars MODEL PORT CONC RESULT_FILENAME INFMAX_CONTAINER_WORKSPACE + resolve_greenspec_dataset || return $? + + local subset="${GREENSPEC_SUBSET:-diverse}" + local lite="${GREENSPEC_LITE:-0}" + local bucket="${GREENSPEC_BUCKET:-}" + local mode="${GREENSPEC_MODE:-target-default}" + local max_tokens="${GREENSPEC_MAX_TOKENS:-8192}" + local model_name="${SERVED_MODEL_NAME:-$MODEL}" + local py=(env "PYTHONPATH=$INFERENCEX_REPO_ROOT${PYTHONPATH:+:$PYTHONPATH}" "$AIPERF_PYTHON") + local gs=("${py[@]}" -m infx.bench_serving.greenspec_aiperf) + local sel=(--file "$GREENSPEC_FILE" --subset "$subset" --lite "$lite" --bucket "$bucket") + + local pool + pool=$("${gs[@]}" pool "${sel[@]}") || return 1 + if [ "$pool" -le 0 ]; then + echo "ERROR: GreenSpec subset=$subset lite=$lite bucket=$bucket matches no rows" >&2 + return 1 + fi + local num_prompts="${GREENSPEC_NUM_PROMPTS:-$pool}" + if [ "$num_prompts" -gt "$pool" ]; then num_prompts="$pool"; fi + + local extra_inputs + extra_inputs=$(GREENSPEC_MAX_TOKENS="$max_tokens" "$AIPERF_PYTHON" -c ' +import json, os +extra = {"max_tokens": int(os.environ["GREENSPEC_MAX_TOKENS"])} +extra.update(json.loads(os.environ.get("GREENSPEC_EXTRA_INPUTS") or "{}")) +print(json.dumps(extra, separators=(",", ":")))') || return 1 + + local result_dir="${RESULT_DIR:-$INFMAX_CONTAINER_WORKSPACE/results}" + local artifact_dir="$result_dir/greenspec_aiperf" + local metrics_url="${GREENSPEC_METRICS_URL:-http://localhost:$PORT/metrics}" + local tmp_dir + tmp_dir=$(mktemp -d /tmp/greenspec.XXXXXX) || return 1 + mkdir -p "$artifact_dir" + + # The preset when the pinned aiperf has it; otherwise a pre-filtered + # raw_payload copy in a temp dir that never reaches results/. + local preset input_file loader + preset=$("${gs[@]}" preset --subset "$subset" --lite "$lite" --bucket "$bucket" --mode "$mode") || return 1 + if [ -n "$preset" ] && "$AIPERF_PYTHON" -c 'import sys +from aiperf.plugin import plugins +plugins.get_dataset_loader_metadata(sys.argv[1])' "$preset" 2>/dev/null; then + loader="$preset" + input_file="$GREENSPEC_FILE" + else + loader="raw_payload" + input_file="$tmp_dir/greenspec_selection.jsonl" + "${gs[@]}" prepare "${sel[@]}" --out "$input_file" --model "$model_name" \ + --mode "$mode" --extra "$extra_inputs" >/dev/null || { rm -rf "$tmp_dir"; return 1; } + fi + + local -a cmd=( + "$AIPERF_CLI" profile + --model "$model_name" + --tokenizer "${AIPERF_TOKENIZER:-$MODEL}" + --tokenizer-trust-remote-code + --url "http://localhost:$PORT" + --endpoint-type chat + --streaming + --custom-dataset-type "$loader" + --input-file "$input_file" + --dataset-sampling-strategy sequential + --conversation-num "$num_prompts" + --concurrency "$CONC" + --extra-inputs "$extra_inputs" + --random-seed 42 + --use-server-token-count + --server-metrics "$metrics_url" + --ui-type simple + --output-artifact-dir "$artifact_dir" + ) + if [ "${GREENSPEC_WARMUP_REQUESTS:-0}" -gt 0 ]; then + cmd+=(--warmup-request-count "$GREENSPEC_WARMUP_REQUESTS") + fi + if [ -n "${AIPERF_GPU_TELEMETRY_URL:-}" ]; then + cmd+=(--gpu-telemetry "$AIPERF_GPU_TELEMETRY_URL") + else + cmd+=(--no-gpu-telemetry) + fi + + echo "===== GreenSpec cell =====" + echo "subset=$subset lite=$lite bucket=${bucket:-all} mode=$mode loader=$loader" + echo "pool=$pool num_prompts=$num_prompts conc=$CONC extra_inputs=$extra_inputs" + echo "==========================" + write_command "$artifact_dir/benchmark_command.txt" "${cmd[@]}" >/dev/null + + local -a meta_args=() + local pair + for pair in ${GREENSPEC_META:-}; do + meta_args+=(--meta "$pair") + done + local sb=("${py[@]}" -m infx.bench_serving.speedbench_aiperf) + "${sb[@]}" snapshot --metrics-url "$metrics_url" --out "$artifact_dir/spec_metrics_before.json" + + if [[ -n "${GREENSPEC_SERVER_PID:-}" && "$GREENSPEC_SERVER_PID" != "${INFERENCEX_SERVER_PID:-}" ]]; then + INFERENCEX_SERVER_STATE=$(mktemp /tmp/inferencex-server-state.XXXXXX) || return 1 + PYTHONPATH="$INFERENCEX_REPO_ROOT${PYTHONPATH:+:$PYTHONPATH}" python3 -m infx.bench_serving.server_watch capture --pid "$GREENSPEC_SERVER_PID" \ + > "$INFERENCEX_SERVER_STATE" || return 1 + INFERENCEX_SERVER_PID="$GREENSPEC_SERVER_PID" + fi + # aiperf's console reaches the job log through `greenspec_aiperf console`, + # which drops what can echo server error text (the Error Summary table, + # ERROR records) and prints each progress bar at most every + # GREENSPEC_PROGRESS_INTERVAL seconds (default 60). Unbuffered, so the job + # log follows the run. + local aiperf_rc=0 + { run_server_client env PYTHONUNBUFFERED=1 "${cmd[@]}" 2>&1 \ + || echo "$?" > "$tmp_dir/aiperf_rc"; } \ + | "${gs[@]}" console --interval "${GREENSPEC_PROGRESS_INTERVAL:-60}" || true + if [ -f "$tmp_dir/aiperf_rc" ]; then aiperf_rc=$(<"$tmp_dir/aiperf_rc"); fi + echo "aiperf exited with code $aiperf_rc" + + "${sb[@]}" snapshot --metrics-url "$metrics_url" --out "$artifact_dir/spec_metrics_after.json" + + local rc=0 + if [ ! -f "$artifact_dir/profile_export_aiperf.json" ]; then + echo "ERROR: aiperf exited with code $aiperf_rc and wrote no profile_export_aiperf.json" >&2 + rc=$(( aiperf_rc != 0 ? aiperf_rc : 1 )) + else + "${sb[@]}" convert \ + --aiperf-json "$artifact_dir/profile_export_aiperf.json" \ + --before "$artifact_dir/spec_metrics_before.json" \ + --after "$artifact_dir/spec_metrics_after.json" \ + --out "$INFMAX_CONTAINER_WORKSPACE/$RESULT_FILENAME.json" \ + --model "$MODEL" \ + --concurrency "$CONC" \ + --meta "gs_subset=\"$subset\"" \ + --meta "gs_lite=$lite" \ + --meta "gs_bucket=\"$bucket\"" \ + --meta "gs_mode=\"$mode\"" \ + --meta "gs_aiperf_loader=\"$loader\"" \ + --meta "gs_num_requests=$num_prompts" \ + --meta "gs_max_tokens=$max_tokens" \ + --meta "gs_dataset_repo=\"$GREENSPEC_HF_REPO@$GREENSPEC_HF_REVISION\"" \ + --meta "gs_extra_inputs=$extra_inputs" \ + "${meta_args[@]}" || rc=$? + [ -f "$INFMAX_CONTAINER_WORKSPACE/$RESULT_FILENAME.json" ] \ + && cp "$INFMAX_CONTAINER_WORKSPACE/$RESULT_FILENAME.json" "$artifact_dir/result.json" + [ "$rc" -eq 0 ] && rc="$aiperf_rc" + fi + + local -a guard_args=( + --file "$GREENSPEC_FILE" --artifact-dir "$artifact_dir" + --upload "$INFMAX_CONTAINER_WORKSPACE/$RESULT_FILENAME.json" + --server-log "${GREENSPEC_SERVER_LOG:-$INFMAX_CONTAINER_WORKSPACE/server.log}" + ) + if [ -n "${GREENSPEC_KEEP_LOG_DIR:-}" ]; then + guard_args+=(--keep-log-dir "$GREENSPEC_KEEP_LOG_DIR") + fi + "${gs[@]}" guard "${guard_args[@]}" || { + echo "ERROR: GreenSpec guard failed; nothing from this run may be uploaded" >&2 + rm -rf "$artifact_dir" "$INFMAX_CONTAINER_WORKSPACE/$RESULT_FILENAME.json" + rc=1 + } + rm -rf "$tmp_dir" + return "$rc" +} diff --git a/docs/greenspec-procedures.md b/docs/greenspec-procedures.md new file mode 100644 index 0000000000..bddff5f4bb --- /dev/null +++ b/docs/greenspec-procedures.md @@ -0,0 +1,147 @@ +# GreenSpec Procedures + +Use this guide to benchmark speculative decoding on GreenSpec: add a cell, pick a subset, dispatch it, read the result, and keep the data off public artifacts. GreenSpec is an internal speculative-decoding bench cut from production chat traffic (tools, multi-turn history, real system prompts), released as one private dataset on Hugging Face. [`benchmarks/benchmark_lib.sh`](../benchmarks/benchmark_lib.sh) (`run_greenspec_aiperf`) and [`infx/bench_serving/greenspec_aiperf.py`](../infx/bench_serving/greenspec_aiperf.py) are authoritative; this page explains how to use them. + +> **Required:** GreenSpec rows are internal and this repository's artifacts are public. Never add `--export-level raw`, aiperf `inputs.json` uploads, or vLLM request logging (`--enable-log-requests`, `--enable-log-outputs`) to a GreenSpec recipe, and never upload `results/greenspec_aiperf/` or the server log by another path. The guard at the end of `run_greenspec_aiperf` is the last line of defense, not a license to relax the recipe. + +## Procedure index + +| Procedure | Use when | +| --- | --- | +| [Understand the flow](#how-a-greenspec-cell-runs) | Reading a run's log or changing the client | +| [Check prerequisites](#prerequisites) | Before the first GreenSpec cell on a runner | +| [Choose a selection](#choose-a-selection) | Picking the subset, size, and decode parameters | +| [Add a GreenSpec cell](#add-a-greenspec-cell) | Benchmarking a new model, engine, or draft | +| [Dispatch and read results](#dispatch-and-read-results) | Running cells in CI and comparing arms | +| [Troubleshoot](#troubleshooting) | A cell fails before, during, or after the client | + +## Governing sources + +| Source | What it governs | +| --- | --- | +| [`benchmarks/benchmark_lib.sh`](../benchmarks/benchmark_lib.sh) | `resolve_greenspec_dataset`, `run_greenspec_aiperf`, the pinned dataset revision, every `GREENSPEC_*` variable | +| [`infx/bench_serving/greenspec_aiperf.py`](../infx/bench_serving/greenspec_aiperf.py) | Subset selection, preset names, the `raw_payload` fallback, the artifact guard and its allowlist | +| [`infx/bench_serving/speedbench_aiperf.py`](../infx/bench_serving/speedbench_aiperf.py) | Mapping the aiperf export and spec-decode counters onto the `benchmark_serving` result keys | +| [`utils/aiperf`](../.gitmodules) | The pinned aiperf fork that ships `GreenSpecLoader` and the `greenspec_*` presets | +| [`.github/workflows/benchmark-tmpl.yml`](../.github/workflows/benchmark-tmpl.yml) | `GREENSPEC_HF_TOKEN` wiring and the allowlisted `greenspec_aiperf_*` artifact upload | +| [`utils/test_greenspec_aiperf.py`](../utils/test_greenspec_aiperf.py) | Selection, preset, payload, and guard contracts | + +## How a GreenSpec cell runs + +1. The recipe calls `resolve_greenspec_dataset` before starting the server. It installs the pinned aiperf venv and fetches `greenspec.jsonl` from `Green-Lab/GreenSpec` at the commit pinned in `GREENSPEC_HF_REVISION` (`HF_HUB_CACHE` first, then the network), then checks `SHA256SUMS`. A missing token or a bad pin fails here, before the GPU is held. +2. The recipe starts the engine and calls `run_greenspec_aiperf` once the server is ready. +3. `run_greenspec_aiperf` counts the selection (`greenspec_aiperf pool`) and fails on an empty one. It then picks the aiperf loader: the `greenspec_*` preset that covers the selection when the pinned aiperf has it, otherwise a pre-filtered `raw_payload` file written to a temp dir outside `results/`. Both paths send the same request bodies. +4. aiperf replays the selection in file order (`--dataset-sampling-strategy sequential`, seed 42) at `CONC`, streaming, with `--use-server-token-count`. Output length is **not** forced: no `--osl`, no `ignore_eos`; each request stops on its own, capped by `GREENSPEC_MAX_TOKENS`. The engine's spec-decode counters are scraped from `/metrics` before and after. aiperf's console streams to the job log through `greenspec_aiperf console`, which drops what can echo server error text: the Error Summary table becomes one line of codes and counts, and an `ERROR` record keeps only its timestamp and level. Progress bars print as `GreenSpec progress: Profiling (52%) elapsed 02:10 eta 02:00`, at most once per interval. +5. `speedbench_aiperf` writes `$RESULT_FILENAME.json` in the `benchmark_serving` shape, adds `acceptance_length` and `acceptance_rate` from the counters, and records the selection as `gs_*` keys. The cell then flows through the normal fixed-seq-len collect path. +6. `greenspec_aiperf guard` runs last. It deletes everything in `results/greenspec_aiperf/` except the allowlist, replaces error messages in the aiperf export with ``, replaces the server log with a stub (or moves it to `GREENSPEC_KEEP_LOG_DIR`), and fails the job if any GreenSpec `question_id` or any file over 2 MiB is left in what gets uploaded. + +## Prerequisites + +| Requirement | Why | +| --- | --- | +| Repository secret `GREENSPEC_HF_TOKEN`, a read token for the private `Green-Lab/GreenSpec` repo | `resolve_greenspec_dataset` falls back to `HF_TOKEN`, which normally cannot read it | +| The runner launcher forwards `GREENSPEC_HF_TOKEN` into the container | `launch_h200-greennode.sh` lists it in `RUN_ENV`; `launch_hardware-hcm.sh` uses `--export=ALL`. Add it to any launcher with an explicit list | +| The `utils/aiperf` pin includes the `greenspec_*` presets | Without them every cell takes the `raw_payload` fallback; results stay valid, and `gs_aiperf_loader` in the result shows which path ran | +| The engine exposes spec-decode counters on `/metrics` (vLLM `vllm:spec_decode_*`) | `acceptance_length` is `null` otherwise | +| `--max-model-len` covers the longest bucket you select | Rows reach the 64k+ buckets; a short context turns them into failed requests | + +## Choose a selection + +The release has four subsets. `diverse` is every use case across every input-length bucket with equal-size buckets; the other three are the rows of one use case. `lite` is a fixed quarter of every bucket, for fast iteration; use the full set for headline numbers. + +| `GREENSPEC_SUBSET` | Content | Buckets (`GREENSPEC_BUCKET`) | +| --- | --- | --- | +| `diverse` (default) | All three use cases | Any bucket below | +| `agentic_coding` | Coding-agent traffic | `agentic_coding__0-8k`, `__8k-32k`, `__32k-64k`, `__64k+` | +| `agentic_tools` | Tool-calling traffic | `agentic_tools__0-4k`, `__4k-16k`, `__16k-32k`, `__32k+` | +| `llm_task` | Plain chat, no tools | `llm_task__0-1k`, `__1k-4k`, `__4k-16k`, `__16k+` | + +| Variable | Default | Meaning | +| --- | --- | --- | +| `GREENSPEC_LITE` | `0` | `1` keeps the lite quarter of every bucket | +| `GREENSPEC_BUCKET` | unset | One bucket. The bucket presets cover the full bucket, so a lite or `recorded` bucket selection runs through `raw_payload` | +| `GREENSPEC_MODE` | `target-default` | `target-default` sends messages, tools, and `response_format` only, so decode parameters come from the server. `recorded` adds each request's own production parameters (temperature, top-p, its own output cap); with a bucket it runs through `raw_payload` | +| `GREENSPEC_MAX_TOKENS` | `8192` | Output cap per request. In `recorded` mode it lowers a request's own cap, never raises it | +| `GREENSPEC_EXTRA_INPUTS` | `{}` | JSON object merged into every request body, e.g. `{"temperature": 0}` | +| `GREENSPEC_NUM_PROMPTS` | whole selection | Requests per cell. Rows are interleaved across buckets, so any prefix stays balanced and every arm sees the same rows | +| `GREENSPEC_WARMUP_REQUESTS` | `0` | Warmup requests excluded from the metrics | +| `GREENSPEC_PROGRESS_INTERVAL` | `60` | Seconds between `GreenSpec progress:` lines per bar in the job log; `0` prints every update | + +The full list, including `GREENSPEC_METRICS_URL`, `GREENSPEC_SERVER_PID`, `GREENSPEC_SERVER_LOG`, `GREENSPEC_KEEP_LOG_DIR`, and `GREENSPEC_META`, is in the comment above `run_greenspec_aiperf`. + +For a speculative-decoding comparison, keep everything but the draft identical between arms: same subset, mode, cap, `CONC`, engine flags, and image. Comparing acceptance across `target-default` and `recorded` measures the sampling change, not the draft. + +## Add a GreenSpec cell + +Follow [Configuration procedures](./configuration-procedures.md#add-a-model--hardware-recipe) for the general recipe rules. GreenSpec-specific steps: + +1. **Write a shared body** under `benchmarks/single_node/fixed_seq_len/`, e.g. `gs_body.sh`. It sources `benchmark_lib.sh`, maps the wrapper's variables to `GREENSPEC_*`, calls `resolve_greenspec_dataset` before the server starts, starts the engine with logs in `/workspace/server.log`, waits for readiness, then runs: + + ```bash + GREENSPEC_SERVER_PID="$SERVER_PID" \ + GREENSPEC_SERVER_LOG="$SERVER_LOG" \ + GREENSPEC_META="gs_arm=\"$GS_ARM\" num_speculative_tokens=$NUM_SPEC_TOKENS" \ + run_greenspec_aiperf + ``` + + Exit with its return code: a guard failure must fail the job. Keep the engine flags of the arm's non-GreenSpec sibling so results are comparable; the only engine difference between arms is `--speculative-config`. +2. **Add one wrapper per arm and selection.** Each sets the arm and selection and sources the body: + + ```bash + GS_ARM=mtp + GS_SUBSET=agentic_tools + GS_LITE=1 + NUM_SPEC_TOKENS=4 + source "$(dirname "$0")/gs_body.sh" + ``` + + The wrapper name must match the master-config key, as for any fixed-seq-len recipe. +3. **Add the master-config keys**, one per arm, with `spec-decoding` set on the draft arm. `isl` and `osl` are labels only: every row brings its own prompt, and output is capped by `GREENSPEC_MAX_TOKENS`. Say so in a comment above the keys. +4. **Add a `perf-changelog.yaml` entry** naming the image, draft, subset, mode, `CONC`, and output cap. +5. **Run the local checks:** `python -m pytest utils/test_greenspec_aiperf.py utils/test_speedbench_aiperf.py` with the aiperf venv, plus the matrix generation in [CI procedures](./ci-procedures.md#local-matrix-generation). + +A worked example (Gemma-4 31B, base vs MTP depth 4, `agentic_tools` lite, conc 8) lives on the branch `bench/gemma4-greenspec-hcm`: `gemma4gs_body.sh`, `gemma4gsatlbase_fp8block_h200.sh`, `gemma4gsatlmtp_fp8block_h200_mtp.sh`, and the keys `gemma4gsatl{base,mtp}-fp8block-hardware-hcm-vllm`. + +## Dispatch and read results + +Dispatch both arms in one run so they land on the same node and image (see [CI procedures](./ci-procedures.md#manual-end-to-end-dispatch)): + +```bash +gh workflow run e2e-tests.yml -R vngcloud/InferenceX --ref \ + -f generate-cli-command="test-config --config-files configs/nvidia-master.yaml \ + --config-keys --runner-node-filter --no-evals" \ + -f test-name="greenspec base vs " +``` + +Each cell uploads `greenspec_aiperf_` with exactly these files: + +| File | Holds | +| --- | --- | +| `result.json` | The `benchmark_serving`-shaped result, including `acceptance_length`, `acceptance_rate`, and the `gs_*` selection keys | +| `profile_export_aiperf.json` | aiperf's aggregate export, error messages withheld | +| `spec_metrics_before.json`, `spec_metrics_after.json` | Spec-decode counter snapshots | +| `benchmark_command.txt` | The aiperf command line | + +The same result also goes through the normal fixed-seq-len collect path. Compare arms on `output_throughput` (the speedup), `median_ttft_ms`, `median_tpot_ms`, `median_itl_ms`, and `acceptance_length`. Check `completed` equals `num_prompts` in both arms before reading any ratio; a cell with failed requests compares different work. `gs_aiperf_loader`, `gs_dataset_repo`, `gs_mode`, and `gs_max_tokens` show what actually ran. + +## Troubleshooting + +| Symptom | Cause and fix | +| --- | --- | +| `could not fetch greenspec.jsonl from Green-Lab/GreenSpec@...` | The token cannot read the repo: check `GREENSPEC_HF_TOKEN` exists and the launcher forwards it | +| `GreenSpec subset=... matches no rows` | Typo in `GREENSPEC_SUBSET` or `GREENSPEC_BUCKET`, or a bucket from another use case. Use the names in [Choose a selection](#choose-a-selection) | +| `gs_aiperf_loader` is `raw_payload` when a preset was expected | The `utils/aiperf` pin lacks the preset, or the selection has none (a bucket with `GREENSPEC_LITE=1` or `recorded`). Results are still valid | +| Many failed requests in long buckets | `--max-model-len` below the bucket's input length, or the server ran out of KV cache. Check the kept server log on the runner | +| `GreenSpec guard failed; nothing from this run may be uploaded` | A row id or an oversized file is in an upload path. Do not bypass the guard: find what wrote it (a new aiperf export file, an extra upload path, a debug dump) and remove it from the recipe | +| `acceptance_length` is `null` | The engine does not expose spec-decode counters at `GREENSPEC_METRICS_URL`, or it is the base arm | + +The server log is withheld from artifacts by design. To read it, set `GREENSPEC_KEEP_LOG_DIR` to a runner-local path in the recipe and read it on the runner; never attach it to an issue or PR. + +## Review checklist + +- [ ] Arms differ only in the draft: same image, engine flags, subset, mode, cap, and `CONC`. +- [ ] No raw export, request logging, or extra upload path for `results/greenspec_aiperf/` or the server log. +- [ ] The recipe exits with `run_greenspec_aiperf`'s return code. +- [ ] `--max-model-len` covers the selected buckets. +- [ ] Master-config comment states that `isl`/`osl` are labels; `perf-changelog.yaml` names the selection. +- [ ] Both arms report `completed == num_prompts` before any comparison is quoted. diff --git a/docs/index.md b/docs/index.md index 8657986029..f8358a400a 100644 --- a/docs/index.md +++ b/docs/index.md @@ -20,6 +20,7 @@ This is the mandatory low-context router for InferenceX work. Pick the one page | [`ci-procedures.md`](./ci-procedures.md) / [`ci-procedures_zh.md`](./ci-procedures_zh.md) | Matrix generation, validation, dispatch, PR sweeps, reuse, staging, and artifact downloads | | [`eval-agentx-procedures.md`](./eval-agentx-procedures.md) / [`eval-agentx-procedures_zh.md`](./eval-agentx-procedures_zh.md) | Eval and AgentX selection, execution, scoring, evidence, and live-run diagnosis | | [`serving-yaml-procedures.md`](./serving-yaml-procedures.md) / [`serving-yaml-procedures_zh.md`](./serving-yaml-procedures_zh.md) | Benchmarking your own serving config from one YAML on a VNG runner: prerequisites, runner-provided image/model access, dispatch, escalation | +| [`greenspec-procedures.md`](./greenspec-procedures.md) | Benchmarking speculative decoding on the internal GreenSpec dataset: selection, cells, dispatch, results, and data safety | | [`results-and-ingestion.md`](./results-and-ingestion.md) / [`results-and-ingestion_zh.md`](./results-and-ingestion_zh.md) | Published-result lookup, artifact identities and schemas, app ingestion, dedupe, and provenance | | [`recovery-results-procedures.md`](./recovery-results-procedures.md) / [`recovery-results-procedures_zh.md`](./recovery-results-procedures_zh.md) | Result processing, ingest verification and recovery, runner cleanup, and failure classification | | [`testing.md`](./testing.md) / [`testing_zh.md`](./testing_zh.md) | Local checks, smoke runs, evidence standards, and review gates | diff --git a/docs/procedures.md b/docs/procedures.md index 31784647ba..0c888384f4 100644 --- a/docs/procedures.md +++ b/docs/procedures.md @@ -23,6 +23,7 @@ The repository source, workflow YAML, launcher scripts, and result collectors ar | Dispatch or monitor a run | [CI procedures](./ci-procedures.md#manual-end-to-end-dispatch) | `e2e-tests.yml`, run logs, artifacts | | Prepare a PR sweep | [CI procedures](./ci-procedures.md#pr-primary-and-modifier-labels) | `run-sweep.yml`, labels, changelog delta | | Reuse a green sweep | [CI procedures](./ci-procedures.md#artifact-reuse-and-merge-with-reuse) | reuse gate, source artifacts, merge helper | +| Run a GreenSpec cell | [GreenSpec procedures](./greenspec-procedures.md#add-a-greenspec-cell) | `run_greenspec_aiperf`, `greenspec_aiperf.py`, guard allowlist | | Add or debug evals | [Eval and AgentX procedures](./eval-agentx-procedures.md#2-add-a-graded-eval) | `EVALS.md`, eval templates, score validator | | Run AgentX | [Eval and AgentX procedures](./eval-agentx-procedures.md#7-run-agentx-fast-feedback-versus-canonical-evidence) | agentic config, trace source, live-run skill | | Inspect a result or ingest | [Recovery and results procedures](./recovery-results-procedures.md#result-pipeline-know-what-should-exist) | artifact schema, collector, app ingest workflow | diff --git a/infx/bench_serving/greenspec_aiperf.py b/infx/bench_serving/greenspec_aiperf.py new file mode 100644 index 0000000000..71340302ff --- /dev/null +++ b/infx/bench_serving/greenspec_aiperf.py @@ -0,0 +1,366 @@ +"""GreenSpec on aiperf: subset selection, the raw_payload fallback, and the artifact guard. + +GreenSpec is an internal speculative-decoding bench cut from production chat +traffic. A release ships every subset in one flat ``greenspec.jsonl``: each line +is a fixed set of metadata keys (``META_KEYS``) followed by a complete +chat-completions body with no ``model`` and no ``stream``. ``run_greenspec_aiperf`` +(``benchmarks/benchmark_lib.sh``) drives aiperf over one subset and uses this +module for three steps: + +``pool`` count the rows a selection keeps (fail fast on an empty filter). +``prepare`` write the selection as an aiperf ``raw_payload`` file, for an aiperf + without the ``greenspec_*`` presets or a selection no preset covers. + It builds each body exactly like aiperf's ``GreenSpecLoader``. +``console`` pass aiperf's console through to the job log, minus what can echo + server error text (the Error Summary table and ERROR records), + with each progress bar printed at most once per interval. +``guard`` run after the client, before any upload. This repo is public and + its artifacts are downloadable, so nothing derived from row content + may leave the runner: keep only ``ALLOWLIST`` in the artifact dir, + strip error messages from the aiperf export (a server error can + echo request content), withhold the server log, then fail if any + ``question_id`` of the release, or any file over ``MAX_BYTES``, + is still in what gets uploaded. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import sys +import time +from collections.abc import Callable, Iterator +from pathlib import Path +from typing import Any + +META_KEYS = ( + "question_id", + "category", + "bucket", + "lite", + "isl", + "osl", + "thinking_src", + "recorded", +) +MAX_TOKEN_KEYS = ("max_tokens", "max_completion_tokens") +SUBSETS = ("diverse", "agentic_coding", "agentic_tools", "llm_task") +ALLOWLIST = ( + "profile_export_aiperf.json", + "result.json", + "spec_metrics_before.json", + "spec_metrics_after.json", + "benchmark_command.txt", +) +MAX_BYTES = 2 * 1024 * 1024 +# aiperf's tqdm bars: "{desc}: {n}/{total} |{bar}| {pct}% [{elapsed}<{remaining}]", or +# "{desc}: {pct}% |{bar}| [{elapsed}<{remaining}]" when the total is unknown. +BAR_RE = re.compile( + r"(?P[A-Za-z][A-Za-z _-]{0,30}): +(?:(?P[\d,]+)/(?P[\d,]+) +)?" + r"(?:(?P\d+)% +)?\|[^|]*\| *(?:(?P\d+)% +)?" + r"\[(?P[\d:]+)<(?P[\d:?]+)\]" +) +# aiperf's log record: "14:27:38.612 INFO message (file.py:64)". +LOG_RE = re.compile(r"\d\d:\d\d:\d\d\.\d{3} +([A-Z]+) +") +ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") +WITHHELD_LEVELS = ("ERROR", "CRITICAL") +SERVER_LOG_STUB = ( + "server log withheld: GreenSpec rows are internal and a server log can echo them\n" +) + + +def iter_rows(path: Path) -> Iterator[dict[str, Any]]: + with open(path, encoding="utf-8") as f: + for line in f: + if line.strip(): + yield json.loads(line) + + +def keep(row: dict[str, Any], subset: str, lite: bool, bucket: str | None) -> bool: + return ( + (subset == "diverse" or row["category"] == subset) + and (not lite or row["lite"]) + and (bucket is None or row["bucket"] == bucket) + ) + + +def select(path: Path, subset: str, lite: bool, bucket: str | None) -> list[dict[str, Any]]: + if subset not in SUBSETS: + raise ValueError(f"unknown GreenSpec subset {subset!r}; one of {SUBSETS}") + return [r for r in iter_rows(path) if keep(r, subset, lite, bucket)] + + +def preset(subset: str, lite: bool, bucket: str | None, recorded: bool) -> str | None: + """The aiperf preset for a selection, or None when no preset covers it.""" + if bucket is not None: + # Bucket presets cover the full bucket only: no lite or recorded variants. + if lite or recorded: + return None + return "greenspec_" + bucket.replace("__", "_").replace("-", "_").replace("+", "_plus") + name = "greenspec" if subset == "diverse" else f"greenspec_{subset}" + return name + ("_lite" if lite else "") + ("_recorded" if recorded else "") + + +def payload( + row: dict[str, Any], + model: str, + *, + recorded: bool = False, + extra: dict[str, Any] | None = None, + stream: bool = True, +) -> dict[str, Any]: + """One row -> the request body aiperf's GreenSpecLoader sends for it.""" + body = {k: v for k, v in row.items() if k not in META_KEYS} + if recorded: + body.update(row.get("recorded") or {}) + for key, value in (extra or {}).items(): + capped = [k for k in MAX_TOKEN_KEYS if k in body] + if key in MAX_TOKEN_KEYS and capped and isinstance(value, int): + for k in capped: + body[k] = min(body[k], value) + else: + body[key] = value + out: dict[str, Any] = {"model": model, **body} + if stream: + out["stream"] = True + out["stream_options"] = {"include_usage": True} + return out + + +class ConsoleFilter: + """aiperf's console for the job log, minus what can echo server error text. + + The Error Summary table is dropped (its Message column is the raw server + response) and replaced by its codes and counts; an ERROR or CRITICAL record + keeps its timestamp and level and loses its text and continuation lines (a + traceback, wrapped text). An unterminated table stays dropped. Each progress + bar is rebuilt from its numbers and printed at most once per ``interval`` + seconds, plus once at 100%. + """ + + def __init__( + self, + write: Callable[[str], None], + interval: float, + clock: Callable[[], float] = time.monotonic, + ) -> None: + self._write = write + self._interval = interval + self._clock = clock + self._bars: dict[str, tuple[float, str]] = {} + self._in_table = False + self._dropping = False + self._errors: list[str] = [] + + def feed(self, segment: str) -> None: + text = ANSI_RE.sub("", segment).rstrip() + if not text: # tqdm's \r next to a record's \n leaves empty segments + return + bar = BAR_RE.match(text.lstrip()) + if bar: + self._dropping = False # a bar ends the continuation of an ERROR record + self._bar(bar) + # tqdm writes no newline, so a log record can follow on the same line. + text = text.lstrip()[bar.end():] + if not text.strip(): + return + if self._in_table: + self._table_line(text) + return + log = LOG_RE.match(text) + if log: + self._dropping = log.group(1) in WITHHELD_LEVELS + if self._dropping: + self._write(f"{text[:log.end()]}") + return + elif "Error Summary" in text: + self._in_table, self._dropping = True, False + return + elif self._dropping: + return + self._write(text) + + def close(self) -> None: + if self._in_table: + self._in_table = False + self._report_errors() + + def _bar(self, m: re.Match[str]) -> None: + desc, pct = m["desc"].strip(), m["pct"] or m["pct0"] + now, last = self._clock(), self._bars.get(desc) + if last and (last[1] == pct == "100" or (pct != "100" and now - last[0] < self._interval)): + return + self._bars[desc] = (now, pct) + count = f" {m['n']}/{m['total']}" if m["n"] else "" + self._write( + f"GreenSpec progress: {desc}{count} ({pct}%) elapsed {m['elapsed']} eta {m['eta']}" + ) + + def _table_line(self, text: str) -> None: + if text.lstrip().startswith(("└", "╰")): + self._in_table = False + self._report_errors() + return + cells = [c.strip() for c in text.split("│")] + if len(cells) >= 4 and (cells[1].isdigit() or cells[1] == "N/A"): + count = cells[-2].replace(",", "") + if count.isdigit(): + self._errors.append(f"{cells[1]} x{count}") + + def _report_errors(self) -> None: + listed = ", ".join(self._errors) or "no rows parsed" + self._write(f"GreenSpec: aiperf error summary withheld (messages can echo requests): {listed}") + self._errors = [] + + +def filter_console(stream: Any, write: Callable[[str], None], interval: float) -> None: + """Feed a binary stream to ``ConsoleFilter`` one \\r- or \\n-delimited segment at a time. + + A segment the filter fails on is dropped rather than ending the read: aiperf + writes into this pipe and would die of a broken pipe. + """ + console = ConsoleFilter(write, interval) + + def feed(raw: bytes) -> None: + try: + console.feed(raw.decode("utf-8", errors="replace")) + except Exception: # noqa: BLE001 + write("GreenSpec: console line dropped (filter error)") + + pending = b"" + while chunk := stream.read1(65536): + *segments, pending = re.split(rb"[\r\n]", pending + chunk) + for segment in segments: + feed(segment) + if pending: + feed(pending) + console.close() + + +def _redact_errors(export_path: Path) -> None: + export = json.loads(export_path.read_text()) + for entry in export.get("error_summary") or []: + details = entry.get("error_details") + if isinstance(details, dict) and "message" in details: + details["message"] = "" + export_path.write_text(json.dumps(export, indent=2)) + + +def guard( + dataset: Path, + artifact_dir: Path, + uploads: list[Path], + server_logs: list[Path], + keep_log_dir: Path | None = None, +) -> list[str]: + """Prune, redact and scan everything a GreenSpec job uploads. Returns the problems.""" + if artifact_dir.is_dir(): + for p in sorted(artifact_dir.rglob("*"), reverse=True): + rel = p.relative_to(artifact_dir).as_posix() + if p.is_dir(): + if not any(p.iterdir()): + p.rmdir() + elif rel not in ALLOWLIST: + p.unlink() + export = artifact_dir / "profile_export_aiperf.json" + if export.is_file(): + _redact_errors(export) + for log in server_logs: + if log.is_file(): + if keep_log_dir is not None: + keep_log_dir.mkdir(parents=True, exist_ok=True) + shutil.move(str(log), str(keep_log_dir / log.name)) + # Swap in a new inode: the server may still hold the log open, and a truncate + # would let its next writes land in the stub (after a hole of NULs). + tmp = log.with_name(log.name + ".stub") + tmp.write_text(SERVER_LOG_STUB) + os.replace(tmp, log) + qids = {r["question_id"] for r in iter_rows(dataset)} + files = [p for p in artifact_dir.rglob("*") if p.is_file()] if artifact_dir.is_dir() else [] + files += [p for p in uploads if p.is_file()] + problems = [] + for p in files: + size = p.stat().st_size + if size > MAX_BYTES: + problems.append(f"{p}: {size} bytes > {MAX_BYTES}") + continue + text = p.read_text(errors="replace") + if any(q in text for q in qids): + problems.append(f"{p}: contains a GreenSpec question_id") + return problems + + +def _selection(args: argparse.Namespace) -> tuple[str, bool, str | None]: + return args.subset, args.lite == "1", args.bucket or None + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="operation", required=True) + + def selection(p: argparse.ArgumentParser) -> None: + p.add_argument("--file", type=Path, required=True) + p.add_argument("--subset", default="diverse") + p.add_argument("--lite", default="0", choices=("0", "1")) + p.add_argument("--bucket", default="") + + pool = commands.add_parser("pool", help="count the rows a selection keeps") + selection(pool) + name = commands.add_parser("preset", help="print the aiperf preset, empty when none") + name.add_argument("--subset", default="diverse") + name.add_argument("--lite", default="0", choices=("0", "1")) + name.add_argument("--bucket", default="") + name.add_argument("--mode", default="target-default", choices=("target-default", "recorded")) + prep = commands.add_parser("prepare", help="write a selection as a raw_payload file") + selection(prep) + prep.add_argument("--out", type=Path, required=True) + prep.add_argument("--model", required=True) + prep.add_argument("--mode", default="target-default", choices=("target-default", "recorded")) + prep.add_argument("--extra", default="{}", help="JSON object, as --extra-inputs") + prep.add_argument("--no-stream", action="store_true") + cons = commands.add_parser("console", help="filter aiperf's console (stdin) for the job log") + cons.add_argument("--interval", type=float, default=60.0, help="seconds between bar lines") + grd = commands.add_parser("guard", help="prune, redact and scan before upload") + grd.add_argument("--file", type=Path, required=True) + grd.add_argument("--artifact-dir", type=Path, required=True) + grd.add_argument("--upload", type=Path, action="append", default=[]) + grd.add_argument("--server-log", type=Path, action="append", default=[]) + grd.add_argument("--keep-log-dir", type=Path, default=None) + args = parser.parse_args() + + if args.operation == "pool": + print(len(select(args.file, *_selection(args)))) + elif args.operation == "preset": + lite, bucket = args.lite == "1", args.bucket or None + print(preset(args.subset, lite, bucket, args.mode == "recorded") or "") + elif args.operation == "prepare": + rows = select(args.file, *_selection(args)) + extra = json.loads(args.extra) + with open(args.out, "w", encoding="utf-8") as f: + for row in rows: + body = payload( + row, + args.model, + recorded=args.mode == "recorded", + extra=extra, + stream=not args.no_stream, + ) + f.write(json.dumps(body, ensure_ascii=False) + "\n") + print(len(rows)) + elif args.operation == "console": + filter_console(sys.stdin.buffer, lambda line: print(line, flush=True), args.interval) + else: + problems = guard( + args.file, args.artifact_dir, args.upload, args.server_log, args.keep_log_dir + ) + for problem in problems: + print(f"GreenSpec guard: {problem}", file=sys.stderr) + return 1 if problems else 0 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/infx/bench_serving/speedbench_aiperf.py b/infx/bench_serving/speedbench_aiperf.py new file mode 100644 index 0000000000..e4a6c7e47c --- /dev/null +++ b/infx/bench_serving/speedbench_aiperf.py @@ -0,0 +1,221 @@ +"""Turn an aiperf SPEED-Bench run into a fixed-seq-len benchmark result. + +SPEED-Bench recipes drive load with aiperf (``run_speedbench_aiperf`` in +``benchmarks/benchmark_lib.sh``) but land in the same fixed-seq-len pipeline +as every ``benchmark_serving.py`` arm, so the raw result must keep that +client's shape: ``infx.results.fixed_sequence`` reads ``model_id``, +``max_concurrency``, ``total_token_throughput``, ``output_throughput`` and every +``*_ms`` key. This module maps aiperf's ``profile_export_aiperf.json`` onto +those keys and attaches the speculative-decoding acceptance measured from the +server's Prometheus counters around the run. + +Metric mapping (aiperf -> benchmark_serving): + time_to_first_token -> ttft (identical definition) + inter_token_latency -> tpot ((e2e - ttft) / (osl - 1), per request) + inter_chunk_latency -> itl (gap between streamed chunks; under speculative + decoding one chunk may carry several tokens, + exactly as in benchmark_serving's itl) + request_latency -> e2el +""" + +from __future__ import annotations + +import argparse +import json +import math +import re +import sys +import urllib.request +from pathlib import Path +from typing import Any + +from infx.bench_serving.benchmark_outcome import benchmark_outcome + +PERCENTILES = ("p90", "p99") +LATENCY_METRICS = { + "ttft": "time_to_first_token", + "tpot": "inter_token_latency", + "itl": "inter_chunk_latency", + "e2el": "request_latency", +} +# vLLM exports monotonic counters, so acceptance over the run is a delta. +VLLM_COUNTERS = { + "accepted_tokens": "vllm:spec_decode_num_accepted_tokens_total", + "draft_events": "vllm:spec_decode_num_drafts_total", + "drafted_tokens": "vllm:spec_decode_num_draft_tokens_total", +} +# SGLang exports only a running-average gauge; it is recorded as read after the run. +SGLANG_ACCEPT_LENGTH = "sglang:spec_accept_length" +_SAMPLE = re.compile(r"^(?P[A-Za-z_:][\w:]*)(?:\{[^}]*\})?\s+(?P\S+)") + + +def parse_prometheus(text: str) -> dict[str, float]: + """Sum every sample of each metric across label sets (e.g. per-engine series).""" + totals: dict[str, float] = {} + for line in text.splitlines(): + match = _SAMPLE.match(line) + if match is None: + continue + try: + value = float(match["value"]) + except ValueError: + continue + if math.isfinite(value): + totals[match["name"]] = totals.get(match["name"], 0.0) + value + return totals + + +def snapshot(metrics_url: str, timeout: float = 10.0) -> dict[str, float]: + """Read the speculative-decoding series this module consumes; missing ones are omitted.""" + with urllib.request.urlopen(metrics_url, timeout=timeout) as response: # noqa: S310 + totals = parse_prometheus(response.read().decode()) + wanted = {*VLLM_COUNTERS.values(), SGLANG_ACCEPT_LENGTH} + return {name: value for name, value in totals.items() if name in wanted} + + +def acceptance(before: dict[str, float], after: dict[str, float]) -> dict[str, Any]: + """Acceptance length AL = 1 + accepted / draft events, independent of draft depth.""" + delta = { + key: round(after.get(name, 0.0) - before.get(name, 0.0)) + for key, name in VLLM_COUNTERS.items() + } + result: dict[str, Any] = {f"spec_{key}": value for key, value in delta.items()} + result["acceptance_length"] = ( + 1 + delta["accepted_tokens"] / delta["draft_events"] if delta["draft_events"] > 0 else None + ) + result["acceptance_rate"] = ( + delta["accepted_tokens"] / delta["drafted_tokens"] if delta["drafted_tokens"] > 0 else None + ) + if result["acceptance_length"] is None and SGLANG_ACCEPT_LENGTH in after: + result["acceptance_length"] = after[SGLANG_ACCEPT_LENGTH] + return result + + +def _stat(export: dict[str, Any], metric: str, stat: str = "avg") -> float: + value = export.get(metric, {}).get(stat) + if value is None: + raise KeyError(f"aiperf export lacks {metric}.{stat}") + return float(value) + + +def convert( + export: dict[str, Any], + *, + model: str, + concurrency: int, + spec: dict[str, Any] | None = None, + metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Build a benchmark_serving-shaped result from one aiperf profile export.""" + if export.get("was_cancelled"): + raise ValueError("aiperf run was cancelled; refusing to publish a partial result") + completed = int(_stat(export, "request_count")) + failed = sum(int(entry.get("count", 0)) for entry in export.get("error_summary") or []) + requested = completed + failed + result: dict[str, Any] = { + "backend": "aiperf", + "model_id": model, + "max_concurrency": concurrency, + "num_prompts": requested, + "completed": completed, + "duration": _stat(export, "benchmark_duration"), + "total_input_tokens": int(_stat(export, "total_isl")), + "total_output_tokens": int(_stat(export, "total_osl")), + "request_throughput": _stat(export, "request_throughput"), + "output_throughput": _stat(export, "output_token_throughput"), + "total_token_throughput": _stat(export, "total_token_throughput"), + "benchmark_outcome": benchmark_outcome(requested, completed), + "aiperf_version": export.get("aiperf_version"), + "aiperf_schema_version": export.get("schema_version"), + } + for short, metric in LATENCY_METRICS.items(): + result[f"mean_{short}_ms"] = _stat(export, metric) + result[f"median_{short}_ms"] = _stat(export, metric, "p50") + result[f"std_{short}_ms"] = _stat(export, metric, "std") + for percentile in PERCENTILES: + result[f"{percentile}_{short}_ms"] = _stat(export, metric, percentile) + for key in ("input_sequence_length", "output_sequence_length"): + if key in export: + result[f"mean_{key}"] = _stat(export, key) + result.update(spec or {}) + result.update(metadata or {}) + return result + + +def _metadata_pairs(pairs: list[str]) -> dict[str, Any]: + metadata: dict[str, Any] = {} + for pair in pairs: + key, sep, raw = pair.partition("=") + if not sep or not key: + raise argparse.ArgumentTypeError(f"--meta expects key=value, got {pair!r}") + try: + metadata[key] = json.loads(raw) + except json.JSONDecodeError: + metadata[key] = raw + return metadata + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="operation", required=True) + + snap = commands.add_parser("snapshot", help="Save the server's spec-decode series as JSON") + snap.add_argument("--metrics-url", required=True) + snap.add_argument("--out", type=Path, required=True) + + conv = commands.add_parser("convert", help="Write the fixed-seq-len raw result JSON") + conv.add_argument("--aiperf-json", type=Path, required=True) + conv.add_argument("--out", type=Path, required=True) + conv.add_argument("--model", required=True) + conv.add_argument("--concurrency", type=int, required=True) + conv.add_argument("--before", type=Path, help="snapshot taken before the run") + conv.add_argument("--after", type=Path, help="snapshot taken after the run") + conv.add_argument("--meta", action="append", default=[], metavar="KEY=VALUE") + + args = parser.parse_args() + if args.operation == "snapshot": + try: + values = snapshot(args.metrics_url) + except OSError as exc: + # A server without /metrics must not fail the benchmark; AL is then unknown. + print(f"[speedbench_aiperf] metrics snapshot failed: {exc}", file=sys.stderr) + values = {} + args.out.write_text(json.dumps(values, indent=2)) + return 0 + + spec = None + if args.before and args.after and args.before.is_file() and args.after.is_file(): + spec = acceptance(json.loads(args.before.read_text()), json.loads(args.after.read_text())) + result = convert( + json.loads(args.aiperf_json.read_text()), + model=args.model, + concurrency=args.concurrency, + spec=spec, + metadata=_metadata_pairs(args.meta), + ) + args.out.write_text(json.dumps(result, indent=2)) + print( + json.dumps( + { + k: result.get(k) + for k in ( + "completed", + "num_prompts", + "output_throughput", + "total_token_throughput", + "mean_ttft_ms", + "mean_tpot_ms", + "acceptance_length", + "acceptance_rate", + ) + }, + indent=2, + ) + ) + # The request gate travels in benchmark_outcome; process_result fails the job on it, + # after the raw result and aiperf artifacts have been kept for diagnosis. + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/runners/launch_h200-greennode.sh b/runners/launch_h200-greennode.sh index 3ff8f7e7f6..e380706493 100644 --- a/runners/launch_h200-greennode.sh +++ b/runners/launch_h200-greennode.sh @@ -36,6 +36,12 @@ fi DCGM_NAME="dcgm-exporter-${RUNNER_NAME:-h200-greennode_01}" RUN_ENV=( HF_TOKEN HF_HUB_CACHE PORT + # Optional read token for the private prepared SPEED-Bench repo; see + # resolve_speedbench_dataset in benchmarks/benchmark_lib.sh. + SPEEDBENCH_HF_TOKEN + # Optional read token for the private GreenSpec repo; see + # resolve_greenspec_dataset in benchmarks/benchmark_lib.sh. + GREENSPEC_HF_TOKEN EXP_NAME MODEL MODEL_PREFIX IMAGE FRAMEWORK PRECISION TP EP_SIZE DP_ATTENTION PP_SIZE DCP_SIZE PCP_SIZE CONC SPEC_DECODING SCENARIO_TYPE SCENARIO_SUBDIR IS_AGENTIC diff --git a/utils/aiperf b/utils/aiperf index faf3f8b357..8f69bd6f89 160000 --- a/utils/aiperf +++ b/utils/aiperf @@ -1 +1 @@ -Subproject commit faf3f8b35760a7e2f9eadcc04fae66ebdbd3b765 +Subproject commit 8f69bd6f89e5855828ce6038ced9a16b34df4138 diff --git a/utils/test_greenspec_aiperf.py b/utils/test_greenspec_aiperf.py new file mode 100644 index 0000000000..08b819dd48 --- /dev/null +++ b/utils/test_greenspec_aiperf.py @@ -0,0 +1,222 @@ +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from infx.bench_serving.greenspec_aiperf import ( + ALLOWLIST, + MAX_BYTES, + META_KEYS, + SERVER_LOG_STUB, + ConsoleFilter, + guard, + payload, + preset, + select, +) + + +def _row(qid: str, category: str, bucket: str, lite: bool, **body) -> dict: + return { + "question_id": qid, + "category": category, + "bucket": bucket, + "lite": lite, + "isl": 100, + "osl": 10, + "thinking_src": "none", + "recorded": {"temperature": 0.6, "max_completion_tokens": 4096}, + "messages": [{"role": "user", "content": f"q {qid}"}], + **body, + } + + +ROWS = [ + _row("m/aaaa", "agentic_tools", "agentic_tools__0-4k", True, tools=[{"type": "function"}]), + _row("m/bbbb", "agentic_tools", "agentic_tools__4k-16k", False), + _row("m/cccc", "llm_task", "llm_task__16k+", True), +] + + +@pytest.fixture +def dataset(tmp_path: Path) -> Path: + p = tmp_path / "greenspec.jsonl" + p.write_text("".join(json.dumps(r) + "\n" for r in ROWS)) + return p + + +@pytest.mark.parametrize( + ("args", "expected"), + [ + (("diverse", False, None), ["m/aaaa", "m/bbbb", "m/cccc"]), + (("diverse", True, None), ["m/aaaa", "m/cccc"]), + (("agentic_tools", False, None), ["m/aaaa", "m/bbbb"]), + (("agentic_tools", True, None), ["m/aaaa"]), + (("diverse", False, "llm_task__16k+"), ["m/cccc"]), + ], +) +def test_select(dataset: Path, args: tuple, expected: list[str]) -> None: + assert [r["question_id"] for r in select(dataset, *args)] == expected + + +def test_unknown_subset(dataset: Path) -> None: + with pytest.raises(ValueError, match="unknown GreenSpec subset"): + select(dataset, "coding", False, None) + + +@pytest.mark.parametrize( + ("args", "expected"), + [ + (("diverse", False, None, False), "greenspec"), + (("diverse", True, None, True), "greenspec_lite_recorded"), + (("agentic_tools", True, None, False), "greenspec_agentic_tools_lite"), + (("diverse", False, "agentic_coding__64k+", False), "greenspec_agentic_coding_64k_plus"), + (("diverse", False, "llm_task__0-1k", False), "greenspec_llm_task_0_1k"), + (("diverse", False, "llm_task__0-1k", True), None), # no recorded bucket presets + (("diverse", True, "llm_task__0-1k", False), None), # no lite bucket presets + ], +) +def test_preset(args: tuple, expected: str | None) -> None: + assert preset(*args) == expected + + +def test_payload_target_default() -> None: + body = payload(ROWS[0], "served", extra={"max_tokens": 8192}) + assert list(body)[0] == "model" and body["model"] == "served" + assert not set(META_KEYS) & set(body) + assert body["max_tokens"] == 8192 and "temperature" not in body + assert body["stream"] is True and body["stream_options"] == {"include_usage": True} + assert body["tools"] == [{"type": "function"}] + + +def test_payload_recorded_cap_only_lowers() -> None: + low = payload(ROWS[0], "m", recorded=True, extra={"max_tokens": 2048}) + assert low["max_completion_tokens"] == 2048 and "max_tokens" not in low + high = payload(ROWS[0], "m", recorded=True, extra={"max_tokens": 8192}) + assert high["max_completion_tokens"] == 4096 and high["temperature"] == 0.6 + + +CONSOLE = ( + "14:27:38.612 INFO Starting AIPerf System (_single_run.py:64)\n" + "Profiling: 0% | | [00:00 list[str]: + out: list[str] = [] + ticks = iter(range(0, 1000, 10)) # each bar line is 10 s after the previous one + console = ConsoleFilter(out.append, interval, clock=lambda: next(ticks)) + for segment in text.replace("\r", "\n").split("\n"): + console.feed(segment) + console.close() + return out + + +def test_console_withholds_error_text_and_throttles_bars() -> None: + out = _filtered(CONSOLE, interval=15) + assert "secret" not in "\n".join(out) and "aaaa" not in "\n".join(out) + assert out == [ + "14:27:38.612 INFO Starting AIPerf System (_single_run.py:64)", + "GreenSpec progress: Profiling (0%) elapsed 00:00 eta ?", + "GreenSpec progress: Profiling (90%) elapsed 00:50 eta 00:05", + "14:27:46.424 WARNING OSL mismatch: got 18 tokens", + "14:27:46.500 ERROR ", + "GreenSpec progress: Profiling (100%) elapsed 00:55 eta 00:00", + "GreenSpec progress: Processing Records 40/40 (100%) elapsed 00:01 eta 00:00", + "GreenSpec: aiperf error summary withheld (messages can echo requests): 400 x1203, N/A x2", + " NVIDIA AIPerf | LLM Metrics", + "│ Request Latency (ms) │ 12.00 │", + ] + + +def test_console_keeps_an_unterminated_error_table_dropped() -> None: + cut = CONSOLE[: CONSOLE.index("│ N/A")] + out = _filtered(cut, interval=0) + assert "secret" not in "\n".join(out) + assert out[-1] == "GreenSpec: aiperf error summary withheld (messages can echo requests): 400 x1203" + + +def test_cli_console_filters_stdin() -> None: + cmd = [sys.executable, "-m", "infx.bench_serving.greenspec_aiperf", "console", "--interval", "0"] + res = subprocess.run(cmd, input=CONSOLE.encode(), capture_output=True, check=True, + cwd=Path(__file__).resolve().parents[1]) + lines = res.stdout.decode().splitlines() + assert "secret" not in res.stdout.decode() + assert "GreenSpec progress: Profiling (40%) elapsed 00:20 eta 00:30" in lines + assert lines[-1] == "│ Request Latency (ms) │ 12.00 │" + + +def test_guard_prunes_redacts_and_withholds_server_log(dataset: Path, tmp_path: Path) -> None: + art = tmp_path / "results" / "greenspec_aiperf" + (art / "logs").mkdir(parents=True) + for name in ALLOWLIST: + (art / name).write_text("{}") + export = {"error_summary": [{"count": 1, "error_details": {"code": 400, "message": "x"}}]} + (art / "profile_export_aiperf.json").write_text(json.dumps(export)) + (art / "inputs.json").write_text("prompts") + (art / "profile_export.jsonl").write_text("{}") + (art / "logs" / "aiperf.log").write_text("log") + log = tmp_path / "server.log" + log.write_text("engine output") + keep = tmp_path / "kept" + + assert guard(dataset, art, [], [log], keep_log_dir=keep) == [] + assert sorted(p.name for p in art.rglob("*")) == sorted(ALLOWLIST) + redacted = json.loads((art / "profile_export_aiperf.json").read_text()) + assert redacted["error_summary"][0]["error_details"] == {"code": 400, "message": ""} + assert log.read_text() == SERVER_LOG_STUB + assert (keep / "server.log").read_text() == "engine output" + + +def test_guard_stub_survives_a_server_still_writing(dataset: Path, tmp_path: Path) -> None: + log = tmp_path / "server.log" + log.write_text("engine output") + with log.open("r+") as server: # like `> server.log`: no O_APPEND + server.seek(0, 2) + assert guard(dataset, tmp_path / "none", [], [log]) == [] + server.write("late engine output") + assert log.read_text() == SERVER_LOG_STUB + + +def test_guard_fails_on_a_row_id_or_a_big_file(dataset: Path, tmp_path: Path) -> None: + art = tmp_path / "greenspec_aiperf" + art.mkdir() + (art / "result.json").write_text(json.dumps({"note": "m/bbbb"})) + big = tmp_path / "result_cell.json" + big.write_text("x" * (MAX_BYTES + 1)) + problems = guard(dataset, art, [big], []) + assert any("question_id" in p for p in problems) + assert any("bytes" in p for p in problems) + + +def test_cli_prepare_writes_raw_payload_lines(dataset: Path, tmp_path: Path) -> None: + out = tmp_path / "sel.jsonl" + cmd = [ + sys.executable, "-m", "infx.bench_serving.greenspec_aiperf", "prepare", + "--file", str(dataset), "--subset", "agentic_tools", "--lite", "1", + "--out", str(out), "--model", "served", "--extra", '{"max_tokens": 1024}', + ] + res = subprocess.run(cmd, capture_output=True, text=True, check=True, + cwd=Path(__file__).resolve().parents[1]) + assert res.stdout.strip() == "1" + lines = [json.loads(line) for line in out.read_text().splitlines()] + assert lines == [payload(ROWS[0], "served", extra={"max_tokens": 1024})] diff --git a/utils/test_speedbench_aiperf.py b/utils/test_speedbench_aiperf.py new file mode 100644 index 0000000000..8b1c57bea2 --- /dev/null +++ b/utils/test_speedbench_aiperf.py @@ -0,0 +1,232 @@ +import json +import re +import subprocess +import sys +from pathlib import Path + +import pytest + +from infx.bench_serving.benchmark_outcome import benchmark_outcome +from infx.bench_serving.speedbench_aiperf import acceptance, convert, parse_prometheus + + +def _stat(avg: float, **extra: float) -> dict: + return {"unit": "ms", "avg": avg, "p50": avg, "p90": avg * 1.5, "p99": avg * 2, "std": 1.0, + **extra} + + +def _export(**overrides) -> dict: + export = { + "schema_version": "1.4", + "aiperf_version": "0.12.0", + "was_cancelled": False, + "error_summary": [], + "request_count": {"unit": "requests", "avg": 80.0}, + "benchmark_duration": {"unit": "sec", "avg": 100.0}, + "total_isl": {"unit": "tokens", "avg": 655360.0}, + "total_osl": {"unit": "tokens", "avg": 81920.0}, + "request_throughput": {"unit": "requests/sec", "avg": 0.8}, + "output_token_throughput": {"unit": "tokens/sec", "avg": 819.2}, + "total_token_throughput": {"unit": "tokens/sec", "avg": 7372.8}, + "time_to_first_token": _stat(300.0), + "inter_token_latency": _stat(10.0), + "inter_chunk_latency": _stat(25.0), + "request_latency": _stat(10530.0), + "input_sequence_length": {"unit": "tokens", "avg": 8192.0}, + "output_sequence_length": {"unit": "tokens", "avg": 1024.0}, + } + export.update(overrides) + return export + + +def test_convert_emits_the_keys_fixed_sequence_reads() -> None: + result = convert(_export(), model="org/model", concurrency=8) + + assert result["model_id"] == "org/model" + assert result["max_concurrency"] == 8 + assert result["total_token_throughput"] == 7372.8 + assert result["output_throughput"] == 819.2 + assert result["mean_ttft_ms"] == 300.0 + assert result["p99_ttft_ms"] == 600.0 + assert result["median_tpot_ms"] == 10.0 + assert result["mean_itl_ms"] == 25.0 + assert result["p90_e2el_ms"] == 10530.0 * 1.5 + assert result["total_input_tokens"] == 655360 + assert result["num_prompts"] == result["completed"] == 80 + assert result["benchmark_outcome"] == benchmark_outcome(80, 80) + + +def test_convert_counts_errors_into_the_request_gate() -> None: + export = _export(error_summary=[{"error_details": {"code": 500}, "count": 10}]) + + result = convert(export, model="m", concurrency=1) + + assert result["num_prompts"] == 90 + assert result["completed"] == 80 + assert result["benchmark_outcome"]["status"] == "failed" + + +def test_convert_refuses_a_cancelled_run() -> None: + with pytest.raises(ValueError, match="cancelled"): + convert(_export(was_cancelled=True), model="m", concurrency=1) + + +def test_convert_result_passes_fixed_sequence_processing() -> None: + from infx.results.fixed_sequence import build_result + + result = convert(_export(), model="org/model", concurrency=8, + metadata={"sb_category": "high_entropy", "sb_extra_inputs": {"ignore_eos": True}}) + env = {"RUNNER_TYPE": "h200", "IMAGE": "img", "MODEL_PREFIX": "p", "FRAMEWORK": "vllm", + "PRECISION": "fp8", "SPEC_DECODING": "mtp", "ISL": "8192", "OSL": "1024", + "DISAGG": "false", "IS_MULTINODE": "false", "TP": "1", "EP_SIZE": "1", + "DP_ATTENTION": "false"} + + data = build_result(result, env) + + assert data["conc"] == 8 + assert data["tput_per_gpu"] == 7372.8 + assert data["mean_ttft"] == pytest.approx(0.3) + assert data["p99_itl"] == pytest.approx(0.05) + assert data["mean_intvty"] == pytest.approx(100.0) + assert data["benchmark_outcome"]["status"] == "passed" + + +def test_acceptance_uses_counter_deltas() -> None: + before = {"vllm:spec_decode_num_accepted_tokens_total": 100.0, + "vllm:spec_decode_num_drafts_total": 50.0, + "vllm:spec_decode_num_draft_tokens_total": 200.0} + after = {"vllm:spec_decode_num_accepted_tokens_total": 1100.0, + "vllm:spec_decode_num_drafts_total": 550.0, + "vllm:spec_decode_num_draft_tokens_total": 2200.0} + + spec = acceptance(before, after) + + assert spec["spec_accepted_tokens"] == 1000 + assert spec["acceptance_length"] == pytest.approx(3.0) + assert spec["acceptance_rate"] == pytest.approx(0.5) + + +def test_acceptance_is_null_without_drafts_and_falls_back_to_sglang_gauge() -> None: + assert acceptance({}, {})["acceptance_length"] is None + assert acceptance({}, {"sglang:spec_accept_length": 2.4})["acceptance_length"] == 2.4 + + +def test_parse_prometheus_sums_label_sets_and_skips_comments() -> None: + text = "\n".join([ + "# HELP vllm:spec_decode_num_drafts_total drafts", + "# TYPE vllm:spec_decode_num_drafts_total counter", + 'vllm:spec_decode_num_drafts_total{engine="0",model_name="m"} 1.5e+03', + 'vllm:spec_decode_num_drafts_total{engine="1",model_name="m"} 500', + "vllm:num_requests_running 3", + ]) + + totals = parse_prometheus(text) + + assert totals["vllm:spec_decode_num_drafts_total"] == 2000.0 + assert totals["vllm:num_requests_running"] == 3.0 + + +_ROWS = [ + {"question_id": f"{i:032d}", "category": cat, "messages": [{"role": "user", "content": "hi"}]} + for i, cat in enumerate(["high_entropy"] * 3 + ["low_entropy"] * 2) +] + + +def _fake_aiperf(tmp_path: Path) -> Path: + export = json.dumps(_export(request_count={"unit": "requests", "avg": 3.0})) + script = tmp_path / "aiperf" + script.write_text(f"""#!/usr/bin/env bash +printf '%s\\n' "$@" > {tmp_path}/argv +while [ $# -gt 0 ]; do + if [ "$1" = --output-artifact-dir ]; then out="$2"; fi + shift +done +mkdir -p "$out" +cat > "$out/profile_export_aiperf.json" <<'JSON' +{export} +JSON +""") + script.chmod(0o755) + return script + + +def _run_cell(tmp_path: Path, env: str) -> subprocess.CompletedProcess: + data = tmp_path / "data" + data.mkdir() + (data / "throughput_8k.jsonl").write_text("".join(json.dumps(r) + "\n" for r in _ROWS)) + workspace = tmp_path / "workspace" + workspace.mkdir() + script = f""" + set -e + export INFMAX_CONTAINER_WORKSPACE={workspace} RESULT_DIR={workspace}/results + export MODEL=org/model PORT=1 CONC=8 OSL=1024 RESULT_FILENAME=cell + export SPEEDBENCH_CONFIG=throughput_8k SPEEDBENCH_DIR={data} + {env} + source benchmarks/benchmark_lib.sh + AIPERF_PYTHON={sys.executable} + AIPERF_CLI={_fake_aiperf(tmp_path)} + run_speedbench_aiperf + """ + return subprocess.run(["bash", "-c", script], capture_output=True, text=True, check=False) + + +def test_run_speedbench_aiperf_writes_a_fixed_seq_result(tmp_path) -> None: + result = _run_cell(tmp_path, "export SPEEDBENCH_CATEGORY=high_entropy SPEEDBENCH_META='sb_arm=mtp'") + + assert result.returncode == 0, result.stderr + argv = (tmp_path / "argv").read_text().splitlines() + assert argv[argv.index("--custom-dataset-type") + 1] == "speed_bench_throughput_8k_high_entropy" + # CONC*10 = 80, floored at 64, then clamped to the three matching rows. + assert argv[argv.index("--conversation-num") + 1] == "3" + assert json.loads(argv[argv.index("--extra-inputs") + 1]) == {"ignore_eos": True} + out = json.loads((tmp_path / "workspace" / "cell.json").read_text()) + assert out["model_id"] == "org/model" + assert out["max_concurrency"] == 8 + assert out["sb_category"] == "high_entropy" + assert out["sb_arm"] == "mtp" + assert out["acceptance_length"] is None + assert (tmp_path / "workspace/results/speedbench_aiperf/benchmark_command.txt").is_file() + + +def test_run_speedbench_aiperf_merges_extra_inputs_and_honours_eos(tmp_path) -> None: + result = _run_cell( + tmp_path, + "export SPEEDBENCH_IGNORE_EOS=0 " + "SPEEDBENCH_EXTRA_INPUTS='{\"temperature\": 0, \"chat_template_kwargs\": {\"enable_thinking\": false}}'", + ) + + assert result.returncode == 0, result.stderr + argv = (tmp_path / "argv").read_text().splitlines() + assert argv[argv.index("--custom-dataset-type") + 1] == "speed_bench_throughput_8k" + assert json.loads(argv[argv.index("--extra-inputs") + 1]) == { + "temperature": 0, "chat_template_kwargs": {"enable_thinking": False}} + + +def test_run_speedbench_aiperf_rejects_an_empty_category(tmp_path) -> None: + result = _run_cell(tmp_path, "export SPEEDBENCH_CATEGORY=mixed") + + assert result.returncode != 0 + assert "matches no rows" in result.stderr + assert not (tmp_path / "argv").exists() + + +def test_speedbench_token_reaches_the_greennode_container() -> None: + script = Path("runners/launch_h200-greennode.sh").read_text() + run_env = script.split("RUN_ENV=(", 1)[1].split(")", 1)[0] + + assert "SPEEDBENCH_HF_TOKEN" in run_env.split() + + +def test_speedbench_token_is_wired_only_into_benchmark_tmpl_calls() -> None: + tmpl = Path(".github/workflows/benchmark-tmpl.yml").read_text() + assert " SPEEDBENCH_HF_TOKEN:\n required: false\n" in tmpl + assert " SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }}\n" in tmpl + + for name in ("e2e-tests", "run-sweep"): + text = Path(f".github/workflows/{name}.yml").read_text() + for job in re.split(r"\n (?=[\w-]+:\n)", text): + uses = re.search(r"uses: .*workflows/([\w-]+)\.yml", job) + if not uses or "MODAL_TOKEN_SECRET: ${{" not in job: + continue + passes = "SPEEDBENCH_HF_TOKEN: ${{ secrets.SPEEDBENCH_HF_TOKEN }}" in job + assert passes == (uses.group(1) == "benchmark-tmpl"), (name, uses.group(1))