From e9ae04d852ea252aa4c44450816a51270173c39f Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Fri, 17 Jul 2026 10:40:36 +0800 Subject: [PATCH] Add exact production-trace real replay client --- .../qwen30_exact_trace_client.py | 251 ++++++++++++++++++ .../run_qwen30_exact_trace_real_anchor.sh | 89 +++++++ .../test_fidelity_envelope.py | 18 ++ 3 files changed, 358 insertions(+) create mode 100644 runs/frontier-fidelity-envelope-v1/qwen30_exact_trace_client.py create mode 100755 runs/frontier-fidelity-envelope-v1/run_qwen30_exact_trace_real_anchor.sh diff --git a/runs/frontier-fidelity-envelope-v1/qwen30_exact_trace_client.py b/runs/frontier-fidelity-envelope-v1/qwen30_exact_trace_client.py new file mode 100644 index 0000000..c23c66d --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/qwen30_exact_trace_client.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""Replay a private exact-trace anchor without emitting prompt text.""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import math +from pathlib import Path +from typing import Any + + +TARGET_PASS_RATE = 0.95 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, required=True) + parser.add_argument("--requests-file", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--tpot-slo-ms", type=float, default=150.0) + parser.add_argument("--timeout-seconds", type=float, default=1800.0) + return parser.parse_args() + + +def percentile(values: list[float], fraction: float) -> float | None: + if not values: + return None + ordered = sorted(values) + return ordered[math.ceil(fraction * len(ordered)) - 1] + + +def ttft_slo_ms(input_tokens: int) -> float: + return 1000.0 + 1000.0 * input_tokens / 8000.0 + + +def row_vector_sha256(rows: list[dict[str, Any]]) -> str: + digest = hashlib.sha256() + for row in rows: + digest.update( + json.dumps( + [ + row["source_index"], + row["arrived_at"], + row["input_length"], + row["output_length"], + row["session_id"], + row["runtime_block_ids"], + ], + separators=(",", ":"), + ).encode() + ) + digest.update(b"\n") + return digest.hexdigest() + + +async def request_one( + session: aiohttp.ClientSession, + row: dict[str, Any], + *, + scheduled_at: float, + benchmark_start: float, + tpot_slo_ms: float, +) -> dict[str, Any]: + loop = asyncio.get_running_loop() + delay = scheduled_at - loop.time() + if delay > 0: + await asyncio.sleep(delay) + admitted_at = loop.time() + record: dict[str, Any] = { + "source_index": int(row["source_index"]), + "session_id": int(row["session_id"]), + "scheduled_s": scheduled_at - benchmark_start, + "admitted_s": admitted_at - benchmark_start, + "admission_lag_ms": (admitted_at - scheduled_at) * 1000.0, + "input_tokens": int(row["input_length"]), + "requested_output_tokens": int(row["output_length"]), + "success": False, + } + body = dict(row["body"]) + body.update( + { + "temperature": 0, + "stream": True, + "stream_options": {"include_usage": True}, + "return_token_ids": True, + } + ) + try: + started = loop.time() + async with session.post("/v1/completions", json=body) as response: + if response.status != 200: + detail = (await response.text())[:1000] + raise RuntimeError(f"HTTP {response.status}: {detail}") + first_token_at = None + last_token_at = None + streamed_tokens = 0 + usage = None + while True: + raw = await response.content.readline() + if not raw: + break + line = raw.decode(errors="replace").strip() + if not line.startswith("data:"): + continue + data = line[5:].strip() + if data == "[DONE]": + break + payload = json.loads(data) + if payload.get("usage"): + usage = payload["usage"] + emitted = 0 + for choice in payload.get("choices") or []: + token_ids = choice.get("token_ids") or [] + emitted += len(token_ids) if token_ids else int(bool(choice.get("text"))) + if emitted: + now = loop.time() + first_token_at = first_token_at or now + last_token_at = now + streamed_tokens += emitted + finished = loop.time() + if first_token_at is None or last_token_at is None or usage is None: + raise RuntimeError("missing streaming token or usage") + actual_input = int(usage["prompt_tokens"]) + actual_output = int(usage["completion_tokens"]) + if actual_input != int(row["input_length"]) or actual_output != int( + row["output_length"] + ): + raise RuntimeError(f"usage mismatch: {actual_input}+{actual_output}") + ttft = (first_token_at - started) * 1000.0 + tpot = ( + (last_token_at - first_token_at) * 1000.0 / (actual_output - 1) + if actual_output > 1 + else None + ) + record.update( + { + "success": True, + "actual_input_tokens": actual_input, + "actual_output_tokens": actual_output, + "streamed_token_count": streamed_tokens, + "ttft_ms": ttft, + "tpot_ms": tpot, + "e2e_ms": (finished - started) * 1000.0, + "slo_pass": ttft <= ttft_slo_ms(actual_input) + and (tpot is None or tpot <= tpot_slo_ms), + } + ) + except Exception as error: + record.update( + { + "error": f"{type(error).__name__}: {error}", + "slo_pass": False, + } + ) + return record + + +async def replay(args: argparse.Namespace, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + import aiohttp + + timeout = aiohttp.ClientTimeout(total=args.timeout_seconds) + connector = aiohttp.TCPConnector(limit=0, ttl_dns_cache=300) + benchmark_start = asyncio.get_running_loop().time() + 2.0 + async with aiohttp.ClientSession( + base_url=f"http://{args.host}:{args.port}", + timeout=timeout, + connector=connector, + ) as session: + tasks = [ + asyncio.create_task( + request_one( + session, + row, + scheduled_at=benchmark_start + float(row["arrived_at"]), + benchmark_start=benchmark_start, + tpot_slo_ms=args.tpot_slo_ms, + ) + ) + for row in rows + ] + return await asyncio.gather(*tasks) + + +def main() -> None: + args = parse_args() + if args.tpot_slo_ms <= 0 or args.timeout_seconds <= 0: + raise ValueError("SLO and timeout must be positive") + rows = [json.loads(line) for line in args.requests_file.open() if line.strip()] + if not rows: + raise ValueError("requests file is empty") + arrivals = [float(row["arrived_at"]) for row in rows] + if any(right < left for left, right in zip(arrivals, arrivals[1:])): + raise ValueError("request arrival order drift") + requests = asyncio.run(replay(args, rows)) + requests.sort(key=lambda row: int(row["source_index"])) + completed = [row for row in requests if row["success"]] + passed = sum(bool(row["slo_pass"]) for row in requests) + ttfts = [float(row["ttft_ms"]) for row in completed] + tpots = [ + float(row["tpot_ms"]) + for row in completed + if row["tpot_ms"] is not None + ] + pass_rate = passed / len(requests) + payload = { + "schema": "qwen30-exact-trace-anchor-v1", + "contract": { + "requests_file": str(args.requests_file.resolve()), + "requests_file_sha256": hashlib.sha256( + args.requests_file.read_bytes() + ).hexdigest(), + "row_vector_sha256": row_vector_sha256(rows), + "requests": len(rows), + "first_arrival_s": arrivals[0], + "last_arrival_s": arrivals[-1], + "arrival": "original_trace_timestamp_and_order", + "input_output_prompt": "exact_source_values", + "ttft_slo": "1000ms + 1000ms * input_tokens / 8000", + "tpot_slo_ms": args.tpot_slo_ms, + "target_pass_rate": TARGET_PASS_RATE, + "prompt_text_emitted": False, + }, + "summary": { + "completed": len(completed), + "failed": len(requests) - len(completed), + "passed": passed, + "pass_rate": pass_rate, + "feasible": pass_rate >= TARGET_PASS_RATE, + "ttft_p50_ms": percentile(ttfts, 0.50), + "ttft_p95_ms": percentile(ttfts, 0.95), + "tpot_p50_ms": percentile(tpots, 0.50), + "tpot_p95_ms": percentile(tpots, 0.95), + "admission_lag_p95_ms": percentile( + [float(row["admission_lag_ms"]) for row in requests], 0.95 + ), + }, + "requests": requests, + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + print(json.dumps(payload["summary"], sort_keys=True), flush=True) + if len(completed) != len(requests): + raise SystemExit(2) + + +if __name__ == "__main__": + main() diff --git a/runs/frontier-fidelity-envelope-v1/run_qwen30_exact_trace_real_anchor.sh b/runs/frontier-fidelity-envelope-v1/run_qwen30_exact_trace_real_anchor.sh new file mode 100755 index 0000000..8c03fe0 --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/run_qwen30_exact_trace_real_anchor.sh @@ -0,0 +1,89 @@ +#!/usr/bin/env bash + +set -euo pipefail + +OUTPUT_ROOT="${OUTPUT_ROOT:?OUTPUT_ROOT is required}" +REQUESTS_FILE="${REQUESTS_FILE:?REQUESTS_FILE is required}" +TP="${TP:?TP is required}" +MNS="${MNS:?MNS is required}" +TRACE_LABEL="${TRACE_LABEL:?TRACE_LABEL is required}" +SERVER_PORT="${SERVER_PORT:?SERVER_PORT is required}" +VENV_ROOT="${VENV_ROOT:-/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1}" +MODEL_ROOT="${MODEL_ROOT:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B}" +SERVED_MODEL="qwen3-30b-exact-trace" +SERVER_PID="" + +mkdir -p "${OUTPUT_ROOT}/logs" "${OUTPUT_ROOT}/provenance" "${OUTPUT_ROOT}/results" +exec > >(tee -a "${OUTPUT_ROOT}/logs/controller.log") 2>&1 + +cleanup() { + if [[ -n "${SERVER_PID}" ]] && kill -0 "${SERVER_PID}" 2>/dev/null; then + kill -TERM -- "-${SERVER_PID}" 2>/dev/null || true + for _ in $(seq 1 30); do + kill -0 "${SERVER_PID}" 2>/dev/null || break + sleep 1 + done + kill -KILL -- "-${SERVER_PID}" 2>/dev/null || true + fi + SERVER_PID="" +} +trap cleanup EXIT INT TERM + +IFS=',' read -r -a GPU_IDS <<< "${CUDA_VISIBLE_DEVICES:?fleet GPU allocation is required}" +if [[ "${#GPU_IDS[@]}" -ne "${TP}" ]]; then + echo "ERROR: expected ${TP} GPUs, got ${CUDA_VISIBLE_DEVICES}" >&2 + exit 1 +fi + +REQUEST_COUNT="$(wc -l < "${REQUESTS_FILE}")" +echo "QWEN30_EXACT_TRACE_REAL_LAUNCH_ECHO host=$(hostname) gpus=${CUDA_VISIBLE_DEVICES} model=${MODEL_ROOT} runtime=vLLM-0.20.0+cu129 dtype=BF16 config=TP${TP}_MNS${MNS}_MBT8192 trace=${TRACE_LABEL} requests=${REQUEST_COUNT} source=${REQUESTS_FILE} arrivals=original600s input_output_prompt=exact prefix=on block=16 tpot_slo=150ms output=${OUTPUT_ROOT} expected_wall=12-35m hard_wall=3600s" +date -u +"START_UTC=%Y-%m-%dT%H:%M:%SZ" +sha256sum qwen30_exact_trace_client.py run_qwen30_exact_trace_real_anchor.sh \ + > "${OUTPUT_ROOT}/provenance/source.sha256" +sha256sum "${REQUESTS_FILE}" > "${OUTPUT_ROOT}/provenance/requests.sha256" +sha256sum "${MODEL_ROOT}/config.json" > "${OUTPUT_ROOT}/provenance/model.sha256" +nvidia-smi --query-gpu=index,name,uuid,driver_version --format=csv,noheader \ + > "${OUTPUT_ROOT}/provenance/gpus.csv" + +export TOKENIZERS_PARALLELISM=false +export VLLM_USE_V1=1 +export TORCH_CUDA_ARCH_LIST=9.0 +export HF_HUB_OFFLINE=1 +export TRANSFORMERS_OFFLINE=1 +ulimit -n 65536 + +setsid "${VENV_ROOT}/bin/vllm" serve "${MODEL_ROOT}" \ + --host 127.0.0.1 --port "${SERVER_PORT}" --served-model-name "${SERVED_MODEL}" \ + --tensor-parallel-size "${TP}" --gpu-memory-utilization 0.92 \ + --max-model-len 40960 --max-num-batched-tokens 8192 --max-num-seqs "${MNS}" \ + --enable-prefix-caching --enable-chunked-prefill --no-enable-log-requests \ + > "${OUTPUT_ROOT}/logs/server.log" 2>&1 & +SERVER_PID=$! +READY=0 +for _ in $(seq 1 120); do + if curl -fsS --max-time 2 "http://127.0.0.1:${SERVER_PORT}/v1/models" \ + > "${OUTPUT_ROOT}/results/models.json" 2>/dev/null; then + READY=1 + break + fi + if ! kill -0 "${SERVER_PID}" 2>/dev/null; then + tail -200 "${OUTPUT_ROOT}/logs/server.log" + exit 1 + fi + sleep 3 +done +if [[ "${READY}" -ne 1 ]]; then + tail -200 "${OUTPUT_ROOT}/logs/server.log" + exit 1 +fi + +"${VENV_ROOT}/bin/python" qwen30_exact_trace_client.py \ + --port "${SERVER_PORT}" --requests-file "${REQUESTS_FILE}" \ + --output "${OUTPUT_ROOT}/results/result.json" --tpot-slo-ms 150 \ + --timeout-seconds 1800 + +cleanup +find "${OUTPUT_ROOT}" -type f ! -path '*/provenance/artifacts.sha256' -print0 \ + | sort -z | xargs -0 sha256sum > "${OUTPUT_ROOT}/provenance/artifacts.sha256" +date -u +"END_UTC=%Y-%m-%dT%H:%M:%SZ" +echo "QWEN30_EXACT_TRACE_REAL_ANCHOR_COMPLETE" diff --git a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py index 4d8d224..5ac9a6b 100644 --- a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py +++ b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py @@ -214,6 +214,24 @@ class FidelityEnvelopeTest(unittest.TestCase): ) self.assertEqual(steady["anchors"][0]["requests"], 20) + def test_exact_real_client_uses_the_prepared_row_vector_contract(self) -> None: + client = load("qwen30_exact_trace_client.py") + prepare = load("prepare_exact_trace.py") + rows = [ + { + "source_index": 7, + "arrived_at": 1.25, + "input_length": 32, + "output_length": 4, + "session_id": 3, + "runtime_block_ids": [11, 12], + } + ] + digest = client.hashlib.sha256() + prepare.update_digest(digest, [7, 1.25, 32, 4, 3, [11, 12]]) + self.assertEqual(client.row_vector_sha256(rows), digest.hexdigest()) + self.assertEqual(client.ttft_slo_ms(512), 1064.0) + if __name__ == "__main__": unittest.main()