From 0292096022bafc8f43641fc6f78e98500ad36e5b Mon Sep 17 00:00:00 2001 From: phucnlt2 Date: Tue, 29 Sep 2026 10:50:56 +0700 Subject: [PATCH 1/8] feat(speedbench): shared SPEED-Bench client on aiperf MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add run_speedbench_aiperf and resolve_speedbench_dataset to benchmark_lib.sh. The pinned utils/aiperf speed_bench_* loaders read the prepared SPEED-Bench snapshot (Noridom1/speed-bench-prepared@6e5c3a81, SHA256-checked, cache first); infx.bench_serving.speedbench_aiperf maps the aiperf export onto the benchmark_serving result keys and attaches acceptance length from the vllm:spec_decode_* counters, so SPEED-Bench arms use the normal fixed-seq-len process_result/collect path. Wire an optional SPEEDBENCH_HF_TOKEN secret (benchmark-tmpl callers only, forwarded by launch_h200-greennode.sh; falls back to HF_TOKEN) and upload results/speedbench_aiperf as its own artifact. 在 benchmark_lib.sh 中新增 run_speedbench_aiperf 与 resolve_speedbench_dataset。固定版本 utils/aiperf 的 speed_bench_* loader 读取预处理好的 SPEED-Bench 快照（Noridom1/speed-bench-prepared@6e5c3a81， 校验 SHA256，优先使用缓存）；infx.bench_serving.speedbench_aiperf 将 aiperf 导出转换为 benchmark_serving 结果格式，并从 vllm:spec_decode_* 计数器计算 接受长度，使 SPEED-Bench 走标准 fixed-seq-len 处理流程。 新增可选 secret SPEEDBENCH_HF_TOKEN（仅传给 benchmark-tmpl 调用，由 launch_h200-greennode.sh 转发进容器，缺省回退到 HF_TOKEN），并将 results/speedbench_aiperf 作为独立 artifact 上传。 Co-Authored-By: Claude Opus 5.5 --- benchmarks/benchmark_lib.sh | 222 +++++++++++++++++++++++ infx/bench_serving/speedbench_aiperf.py | 221 ++++++++++++++++++++++ runners/launch_h200-greennode.sh | 3 + utils/test_speedbench_aiperf.py | 232 ++++++++++++++++++++++++ 4 files changed, 678 insertions(+) create mode 100644 infx/bench_serving/speedbench_aiperf.py create mode 100644 utils/test_speedbench_aiperf.py diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index bd5d1a57eb..14bf647f1c 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -4049,3 +4049,225 @@ 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" +} 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..6050ca8f77 100644 --- a/runners/launch_h200-greennode.sh +++ b/runners/launch_h200-greennode.sh @@ -36,6 +36,9 @@ 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 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/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)) From 4f0bd067f9c378a40df44455d16d8d35dab18a2c Mon Sep 17 00:00:00 2001 From: Noridom1 Date: Wed, 28 Oct 2026 12:00:23 +0700 Subject: [PATCH 2/8] feat(greenspec): GreenSpec arm on aiperf run_greenspec_aiperf (benchmark_lib.sh) benchmarks one GreenSpec subset with aiperf: it uses the greenspec_* preset when the installed aiperf has it, else writes a raw_payload file. The dataset is fetched from a private HF repo pinned by commit and checked against SHA256SUMS. A guard runs before any upload: it keeps an allowlist of result files, strips error messages, withholds the server log, and fails on any row id or oversized file. Workflows gain the greenspec inputs; the aiperf submodule points at a fork branch with the GreenSpec loader. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/benchmark-tmpl.yml | 22 +++ .github/workflows/e2e-tests.yml | 6 + .github/workflows/run-sweep.yml | 6 + .gitmodules | 3 +- benchmarks/benchmark_lib.sh | 247 +++++++++++++++++++++++++ infx/bench_serving/greenspec_aiperf.py | 230 +++++++++++++++++++++++ runners/launch_h200-greennode.sh | 2 + utils/aiperf | 2 +- utils/test_greenspec_aiperf.py | 143 ++++++++++++++ 9 files changed, 659 insertions(+), 2 deletions(-) create mode 100644 infx/bench_serving/greenspec_aiperf.py create mode 100644 utils/test_greenspec_aiperf.py 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 14bf647f1c..2e4d962cd3 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -4271,3 +4271,250 @@ print(json.dumps(extra, separators=(",", ":")))') || return 1 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_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 can print server error text; keep it off the job log. + local aiperf_rc=0 + run_server_client "${cmd[@]}" > "$tmp_dir/aiperf_console.log" 2>&1 || aiperf_rc=$? + echo "aiperf exited with code $aiperf_rc (console kept off the job log)" + + "${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/infx/bench_serving/greenspec_aiperf.py b/infx/bench_serving/greenspec_aiperf.py new file mode 100644 index 0000000000..77d5bca604 --- /dev/null +++ b/infx/bench_serving/greenspec_aiperf.py @@ -0,0 +1,230 @@ +"""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``. +``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 shutil +import sys +from collections.abc import 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 +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: + if 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 + + +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)) + log.write_text(SERVER_LOG_STUB) + 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") + 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)) + 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/runners/launch_h200-greennode.sh b/runners/launch_h200-greennode.sh index 6050ca8f77..4123fb8bd5 100644 --- a/runners/launch_h200-greennode.sh +++ b/runners/launch_h200-greennode.sh @@ -39,6 +39,8 @@ RUN_ENV=( # 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 (resolve_greenspec_dataset). + 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..b9c27f985b --- /dev/null +++ b/utils/test_greenspec_aiperf.py @@ -0,0 +1,143 @@ +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, + 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 + ], +) +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 + + +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_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})] From 60c5a2fe051b2d2775acba31821a05d105234a9c Mon Sep 17 00:00:00 2001 From: Noridom1 Date: Wed, 28 Oct 2026 13:23:56 +0700 Subject: [PATCH 3/8] fix(greenspec): no preset for a lite bucket selection Bucket presets cover the full bucket, so a lite bucket selection took the preset and sent the first N rows of the full bucket instead of its lite rows. Return no preset there, as for recorded, so it runs through raw_payload. Co-Authored-By: Claude Opus 5.5 --- infx/bench_serving/greenspec_aiperf.py | 3 ++- utils/test_greenspec_aiperf.py | 1 + 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/infx/bench_serving/greenspec_aiperf.py b/infx/bench_serving/greenspec_aiperf.py index 77d5bca604..e92e6b04a3 100644 --- a/infx/bench_serving/greenspec_aiperf.py +++ b/infx/bench_serving/greenspec_aiperf.py @@ -79,7 +79,8 @@ def select(path: Path, subset: str, lite: bool, bucket: str | None) -> list[dict 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: - if recorded: + # 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}" diff --git a/utils/test_greenspec_aiperf.py b/utils/test_greenspec_aiperf.py index b9c27f985b..7fcd1f1785 100644 --- a/utils/test_greenspec_aiperf.py +++ b/utils/test_greenspec_aiperf.py @@ -74,6 +74,7 @@ def test_unknown_subset(dataset: Path) -> None: (("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: From 0c6498ac0f0bc4deae02132a804a4d449f634a21 Mon Sep 17 00:00:00 2001 From: Noridom1 Date: Wed, 28 Oct 2026 13:23:56 +0700 Subject: [PATCH 4/8] docs(greenspec): GreenSpec procedures How a GreenSpec cell runs, prerequisites, subset selection, adding a cell, dispatch and results, troubleshooting, and the data-safety rules. English only for now. Co-Authored-By: Claude Opus 5.5 --- docs/greenspec-procedures.md | 146 +++++++++++++++++++++++++++++++++++ docs/index.md | 1 + docs/procedures.md | 1 + 3 files changed, 148 insertions(+) create mode 100644 docs/greenspec-procedures.md diff --git a/docs/greenspec-procedures.md b/docs/greenspec-procedures.md new file mode 100644 index 0000000000..0e2015522f --- /dev/null +++ b/docs/greenspec-procedures.md @@ -0,0 +1,146 @@ +# 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. +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 | + +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 | From 8e8d43438302b28b187de30a04e42270443674b4 Mon Sep 17 00:00:00 2001 From: Noridom1 Date: Wed, 28 Oct 2026 13:30:50 +0700 Subject: [PATCH 5/8] feat(greenspec): progress lines in the job log aiperf's console stays off the job log because it can echo server error text, which left long cells with no progress. A watcher now reads that console every GREENSPEC_PROGRESS_INTERVAL seconds (default 60, 0 = off) and prints only the numbers of its progress bars: 'GreenSpec progress: Profiling 33/64 (52%) elapsed 02:10 eta 02:00'. Co-Authored-By: Claude Opus 5.5 --- benchmarks/benchmark_lib.sh | 20 ++++++++++++++- docs/greenspec-procedures.md | 3 ++- infx/bench_serving/greenspec_aiperf.py | 34 ++++++++++++++++++++++++++ utils/test_greenspec_aiperf.py | 16 ++++++++++++ 4 files changed, 71 insertions(+), 2 deletions(-) diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index 2e4d962cd3..3c6048f3af 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -4360,6 +4360,8 @@ PY # 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() { @@ -4469,8 +4471,24 @@ plugins.get_dataset_loader_metadata(sys.argv[1])' "$preset" 2>/dev/null; then INFERENCEX_SERVER_PID="$GREENSPEC_SERVER_PID" fi # aiperf's console can print server error text; keep it off the job log. - local aiperf_rc=0 + # A watcher reprints only the numbers of its progress bars, every + # GREENSPEC_PROGRESS_INTERVAL seconds (default 60, 0 = off). + local aiperf_rc=0 progress_pid="" + local progress_interval="${GREENSPEC_PROGRESS_INTERVAL:-60}" + if [ "$progress_interval" -gt 0 ]; then + ( + set +x + while sleep "$progress_interval"; do + "${gs[@]}" progress --console-log "$tmp_dir/aiperf_console.log" || true + done + ) & + progress_pid=$! + fi run_server_client "${cmd[@]}" > "$tmp_dir/aiperf_console.log" 2>&1 || aiperf_rc=$? + if [ -n "$progress_pid" ]; then + kill "$progress_pid" 2>/dev/null + wait "$progress_pid" 2>/dev/null + fi echo "aiperf exited with code $aiperf_rc (console kept off the job log)" "${sb[@]}" snapshot --metrics-url "$metrics_url" --out "$artifact_dir/spec_metrics_after.json" diff --git a/docs/greenspec-procedures.md b/docs/greenspec-procedures.md index 0e2015522f..1c3e852055 100644 --- a/docs/greenspec-procedures.md +++ b/docs/greenspec-procedures.md @@ -31,7 +31,7 @@ Use this guide to benchmark speculative decoding on GreenSpec: add a cell, pick 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. +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 goes to a temp file, since it can echo server error text; a watcher prints only the numbers of its progress bars to the job log as `GreenSpec progress: Profiling 33/64 (52%) elapsed 02:10 eta 02:00`. 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. @@ -65,6 +65,7 @@ The release has four subsets. `diverse` is every use case across every input-len | `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 in the job log; `0` turns them off | 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`. diff --git a/infx/bench_serving/greenspec_aiperf.py b/infx/bench_serving/greenspec_aiperf.py index e92e6b04a3..620a0901ff 100644 --- a/infx/bench_serving/greenspec_aiperf.py +++ b/infx/bench_serving/greenspec_aiperf.py @@ -11,6 +11,9 @@ ``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``. +``progress`` print the numbers of aiperf's progress bars from its console log, + which the job log never sees (the console can echo server error + text), so a long cell can be followed. ``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, @@ -24,6 +27,7 @@ import argparse import json +import re import shutil import sys from collections.abc import Iterator @@ -50,6 +54,13 @@ "benchmark_command.txt", ) MAX_BYTES = 2 * 1024 * 1024 +# aiperf's tqdm bar: "{desc}: {n}/{total} |{bar}| {pct}% [{elapsed}<{remaining}]". +PROGRESS_RE = re.compile( + r"([A-Za-z][A-Za-z _-]{0,30}): +([\d,]+)/([\d,]+) \|[^|]*\| *(\d+)% " + r"\[([\d:]+)<([\d:?]+)\]" +) +ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") +PROGRESS_TAIL_BYTES = 64 * 1024 SERVER_LOG_STUB = ( "server log withheld: GreenSpec rows are internal and a server log can echo them\n" ) @@ -113,6 +124,24 @@ def payload( return out +def progress(console_log: Path) -> list[str]: + """The latest state of each aiperf progress bar, rebuilt from its numbers only.""" + if not console_log.is_file(): + return [] + with open(console_log, "rb") as f: + f.seek(max(0, console_log.stat().st_size - PROGRESS_TAIL_BYTES)) + text = f.read().decode("utf-8", errors="replace") + latest: dict[str, str] = {} + for line in re.split(r"[\r\n]", ANSI_RE.sub("", text)): + m = PROGRESS_RE.search(line) + if m: + desc, n, total, pct, elapsed, remaining = m.groups() + latest[desc.strip()] = ( + f"{desc.strip()} {n}/{total} ({pct}%) elapsed {elapsed} eta {remaining}" + ) + return list(latest.values()) + + def _redact_errors(export_path: Path) -> None: export = json.loads(export_path.read_text()) for entry in export.get("error_summary") or []: @@ -190,6 +219,8 @@ def selection(p: argparse.ArgumentParser) -> None: 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") + prog = commands.add_parser("progress", help="print aiperf progress-bar numbers") + prog.add_argument("--console-log", type=Path, required=True) 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) @@ -217,6 +248,9 @@ def selection(p: argparse.ArgumentParser) -> None: ) f.write(json.dumps(body, ensure_ascii=False) + "\n") print(len(rows)) + elif args.operation == "progress": + for state in progress(args.console_log): + print(f"GreenSpec progress: {state}") else: problems = guard( args.file, args.artifact_dir, args.upload, args.server_log, args.keep_log_dir diff --git a/utils/test_greenspec_aiperf.py b/utils/test_greenspec_aiperf.py index 7fcd1f1785..5bdf787a02 100644 --- a/utils/test_greenspec_aiperf.py +++ b/utils/test_greenspec_aiperf.py @@ -13,6 +13,7 @@ guard, payload, preset, + progress, select, ) @@ -97,6 +98,21 @@ def test_payload_recorded_cap_only_lowers() -> None: assert high["max_completion_tokens"] == 4096 and high["temperature"] == 0.6 +def test_progress_keeps_only_bar_numbers(tmp_path: Path) -> None: + console = tmp_path / "aiperf_console.log" + console.write_text( + "Error: 400 secret prompt text m/aaaa\n" + "\x1b[2KWarmup: 4/4 |\x1b[32m####\x1b[0m| 100% [00:03<00:00]\n" + "Profiling: 10/64 |## | 16% [00:40<03:30]\r" + "Profiling: 33/64 |#####| 52% [02:10<02:00]\r" + ) + assert progress(console) == [ + "Warmup 4/4 (100%) elapsed 00:03 eta 00:00", + "Profiling 33/64 (52%) elapsed 02:10 eta 02:00", + ] + assert progress(tmp_path / "missing.log") == [] + + 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) From 02e95614c1b1bfe2875479b8db56ec627270b0bc Mon Sep 17 00:00:00 2001 From: Noridom1 Date: Wed, 28 Oct 2026 14:37:39 +0700 Subject: [PATCH 6/8] feat(greenspec): stream aiperf's console to the job log, filtered The watcher only reprinted progress-bar numbers from a hidden console log, and missed aiperf's percent-only Profiling bar. aiperf's console now streams to the job log through `greenspec_aiperf console`, unbuffered. The filter drops what can echo server error text: the Error Summary table (its Message column is the raw server response) becomes one line of codes and counts, and an ERROR or CRITICAL record keeps only its timestamp and level. Each progress bar, with or without a total, prints at most once per GREENSPEC_PROGRESS_INTERVAL seconds. Co-Authored-By: Claude Opus 5.5 --- benchmarks/benchmark_lib.sh | 31 ++--- docs/greenspec-procedures.md | 4 +- infx/bench_serving/greenspec_aiperf.py | 156 ++++++++++++++++++++----- utils/test_greenspec_aiperf.py | 78 ++++++++++--- 4 files changed, 204 insertions(+), 65 deletions(-) diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index 3c6048f3af..31f6f71e3a 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -4470,26 +4470,17 @@ plugins.get_dataset_loader_metadata(sys.argv[1])' "$preset" 2>/dev/null; then > "$INFERENCEX_SERVER_STATE" || return 1 INFERENCEX_SERVER_PID="$GREENSPEC_SERVER_PID" fi - # aiperf's console can print server error text; keep it off the job log. - # A watcher reprints only the numbers of its progress bars, every - # GREENSPEC_PROGRESS_INTERVAL seconds (default 60, 0 = off). - local aiperf_rc=0 progress_pid="" - local progress_interval="${GREENSPEC_PROGRESS_INTERVAL:-60}" - if [ "$progress_interval" -gt 0 ]; then - ( - set +x - while sleep "$progress_interval"; do - "${gs[@]}" progress --console-log "$tmp_dir/aiperf_console.log" || true - done - ) & - progress_pid=$! - fi - run_server_client "${cmd[@]}" > "$tmp_dir/aiperf_console.log" 2>&1 || aiperf_rc=$? - if [ -n "$progress_pid" ]; then - kill "$progress_pid" 2>/dev/null - wait "$progress_pid" 2>/dev/null - fi - echo "aiperf exited with code $aiperf_rc (console kept off the job log)" + # 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" diff --git a/docs/greenspec-procedures.md b/docs/greenspec-procedures.md index 1c3e852055..bddff5f4bb 100644 --- a/docs/greenspec-procedures.md +++ b/docs/greenspec-procedures.md @@ -31,7 +31,7 @@ Use this guide to benchmark speculative decoding on GreenSpec: add a cell, pick 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 goes to a temp file, since it can echo server error text; a watcher prints only the numbers of its progress bars to the job log as `GreenSpec progress: Profiling 33/64 (52%) elapsed 02:10 eta 02:00`. +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. @@ -65,7 +65,7 @@ The release has four subsets. `diverse` is every use case across every input-len | `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 in the job log; `0` turns them off | +| `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`. diff --git a/infx/bench_serving/greenspec_aiperf.py b/infx/bench_serving/greenspec_aiperf.py index 620a0901ff..18ecd2cc8c 100644 --- a/infx/bench_serving/greenspec_aiperf.py +++ b/infx/bench_serving/greenspec_aiperf.py @@ -11,9 +11,9 @@ ``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``. -``progress`` print the numbers of aiperf's progress bars from its console log, - which the job log never sees (the console can echo server error - text), so a long cell can be followed. +``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, @@ -30,7 +30,8 @@ import re import shutil import sys -from collections.abc import Iterator +import time +from collections.abc import Callable, Iterator from pathlib import Path from typing import Any @@ -54,13 +55,17 @@ "benchmark_command.txt", ) MAX_BYTES = 2 * 1024 * 1024 -# aiperf's tqdm bar: "{desc}: {n}/{total} |{bar}| {pct}% [{elapsed}<{remaining}]". -PROGRESS_RE = re.compile( - r"([A-Za-z][A-Za-z _-]{0,30}): +([\d,]+)/([\d,]+) \|[^|]*\| *(\d+)% " - r"\[([\d:]+)<([\d:?]+)\]" +# 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]") -PROGRESS_TAIL_BYTES = 64 * 1024 +WITHHELD_LEVELS = ("ERROR", "CRITICAL") SERVER_LOG_STUB = ( "server log withheld: GreenSpec rows are internal and a server log can echo them\n" ) @@ -124,22 +129,114 @@ def payload( return out -def progress(console_log: Path) -> list[str]: - """The latest state of each aiperf progress bar, rebuilt from its numbers only.""" - if not console_log.is_file(): - return [] - with open(console_log, "rb") as f: - f.seek(max(0, console_log.stat().st_size - PROGRESS_TAIL_BYTES)) - text = f.read().decode("utf-8", errors="replace") - latest: dict[str, str] = {} - for line in re.split(r"[\r\n]", ANSI_RE.sub("", text)): - m = PROGRESS_RE.search(line) - if m: - desc, n, total, pct, elapsed, remaining = m.groups() - latest[desc.strip()] = ( - f"{desc.strip()} {n}/{total} ({pct}%) elapsed {elapsed} eta {remaining}" - ) - return list(latest.values()) +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: @@ -219,8 +316,8 @@ def selection(p: argparse.ArgumentParser) -> None: 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") - prog = commands.add_parser("progress", help="print aiperf progress-bar numbers") - prog.add_argument("--console-log", type=Path, required=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) @@ -248,9 +345,8 @@ def selection(p: argparse.ArgumentParser) -> None: ) f.write(json.dumps(body, ensure_ascii=False) + "\n") print(len(rows)) - elif args.operation == "progress": - for state in progress(args.console_log): - print(f"GreenSpec progress: {state}") + 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 diff --git a/utils/test_greenspec_aiperf.py b/utils/test_greenspec_aiperf.py index 5bdf787a02..dd9c322874 100644 --- a/utils/test_greenspec_aiperf.py +++ b/utils/test_greenspec_aiperf.py @@ -10,10 +10,10 @@ MAX_BYTES, META_KEYS, SERVER_LOG_STUB, + ConsoleFilter, guard, payload, preset, - progress, select, ) @@ -98,19 +98,71 @@ def test_payload_recorded_cap_only_lowers() -> None: assert high["max_completion_tokens"] == 4096 and high["temperature"] == 0.6 -def test_progress_keeps_only_bar_numbers(tmp_path: Path) -> None: - console = tmp_path / "aiperf_console.log" - console.write_text( - "Error: 400 secret prompt text m/aaaa\n" - "\x1b[2KWarmup: 4/4 |\x1b[32m####\x1b[0m| 100% [00:03<00:00]\n" - "Profiling: 10/64 |## | 16% [00:40<03:30]\r" - "Profiling: 33/64 |#####| 52% [02:10<02:00]\r" - ) - assert progress(console) == [ - "Warmup 4/4 (100%) elapsed 00:03 eta 00:00", - "Profiling 33/64 (52%) elapsed 02:10 eta 02:00", +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 │", ] - assert progress(tmp_path / "missing.log") == [] + + +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: From 611667bdda0ccbdc1ebee4df9cbe7b02fb75c9ea Mon Sep 17 00:00:00 2001 From: Noridom1 Date: Wed, 28 Oct 2026 14:44:32 +0700 Subject: [PATCH 7/8] fix(greenspec): swap the server log stub in, don't truncate in place The server still holds its log open when the guard runs. Truncating and writing the stub let later engine writes land after it, behind a hole of NULs, so the uploaded log was no longer just the stub. Write the stub to a sibling file and os.replace it, so the server keeps writing to the unlinked inode. Co-Authored-By: Claude Opus 5.5 --- infx/bench_serving/greenspec_aiperf.py | 7 ++++++- utils/test_greenspec_aiperf.py | 10 ++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/infx/bench_serving/greenspec_aiperf.py b/infx/bench_serving/greenspec_aiperf.py index 18ecd2cc8c..71340302ff 100644 --- a/infx/bench_serving/greenspec_aiperf.py +++ b/infx/bench_serving/greenspec_aiperf.py @@ -27,6 +27,7 @@ import argparse import json +import os import re import shutil import sys @@ -272,7 +273,11 @@ def guard( 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)) - log.write_text(SERVER_LOG_STUB) + # 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()] diff --git a/utils/test_greenspec_aiperf.py b/utils/test_greenspec_aiperf.py index dd9c322874..08b819dd48 100644 --- a/utils/test_greenspec_aiperf.py +++ b/utils/test_greenspec_aiperf.py @@ -187,6 +187,16 @@ def test_guard_prunes_redacts_and_withholds_server_log(dataset: Path, tmp_path: 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() From 4b2884284e1a8bea9fe6760085741825075657f6 Mon Sep 17 00:00:00 2001 From: Noridom1 Date: Wed, 28 Oct 2026 14:44:32 +0700 Subject: [PATCH 8/8] fix(greennode): keep RUN_ENV comment free of parens Co-Authored-By: Claude Opus 5.5 --- runners/launch_h200-greennode.sh | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/runners/launch_h200-greennode.sh b/runners/launch_h200-greennode.sh index 4123fb8bd5..e380706493 100644 --- a/runners/launch_h200-greennode.sh +++ b/runners/launch_h200-greennode.sh @@ -39,7 +39,8 @@ RUN_ENV=( # 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 (resolve_greenspec_dataset). + # 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