From 9b3a2eab80faa79bf06596d3559df5a3e216ac39 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Mon, 13 Jul 2026 15:09:01 +0800 Subject: [PATCH] Add reproducible CollectiveSpec opportunity screen --- docs/collectivespec-pilot-design-20260713.md | 157 ++++++++++++ .../create_controlled_greedy_window.py | 175 +++++++++++++ scripts/collectivespec/run_static_k_pilot.py | 237 ++++++++++++++++++ .../summarize_engine_metrics.py | 135 ++++++++++ .../collectivespec/summarize_fixed_grid.py | 215 ++++++++++++++++ .../summarize_static_k_pilot.py | 136 ++++++++++ 6 files changed, 1055 insertions(+) create mode 100644 docs/collectivespec-pilot-design-20260713.md create mode 100644 scripts/collectivespec/create_controlled_greedy_window.py create mode 100644 scripts/collectivespec/run_static_k_pilot.py create mode 100644 scripts/collectivespec/summarize_engine_metrics.py create mode 100644 scripts/collectivespec/summarize_fixed_grid.py create mode 100644 scripts/collectivespec/summarize_static_k_pilot.py diff --git a/docs/collectivespec-pilot-design-20260713.md b/docs/collectivespec-pilot-design-20260713.md new file mode 100644 index 0000000..8711a3a --- /dev/null +++ b/docs/collectivespec-pilot-design-20260713.md @@ -0,0 +1,157 @@ +# CollectiveSpec：先做机会判定的实验设计 + +## 结论先行 + +当前不应先实现一个“按请求动态 K、再做一次 DP collective 同步”的原型。该路径的 +工程修复很薄，而且相邻公开工作已经覆盖了 request-level dynamic speculation 与 +ragged verification 的大量空间。CollectiveSpec 只有在 +一个更强、可证伪的事实成立时才值得继续：**在 wide-EP MoE、DP>1 的生产负载中， +不同 DP rank / request 所需的投机深度确实不同，并且全局 static K 明显浪费了 SLO +可行 goodput。** + +本文件把第一轮实验定义为一个机会判定（opportunity gate），不是最终性能主张。 + +## 固定条件 + +- Host: `dash0`, 8x NVIDIA H20。 +- Model: Qwen3-235B-A22B FP8；draft: EAGLE3。 +- Deployment: TP=4, DP=2, EP=8，`VLLM_MOE_USE_DEEPEP=1`。 +- Trace: `thinking_w20260327_1000`，600 秒 decode-only 窗口。 +- SLO: TPOT <= 40 ms，pass rate >= 0.95。 +- 同一 engine revision、同一模型/trace 路径、同一环境变量；实验串行执行，避免 GPU + 互相干扰。 + +这里的 resolved topology 来自远端实际 StudySpec，而不是仓库 README 中可能已过期的 +配置描述。 + +## 假设与可证伪指标 + +### G0：static-K 是否有足够可利用的空间？ + +- H0：在该固定拓扑和负载下，NoSpec/K=1/2/3 的 SLO-goodput 差异很小；最佳固定 K 已经 + 足够好。此时停止 CollectiveSpec。 +- H1：不同 static K 的可行前沿存在实质差异，且最优 K 对负载区间敏感。只有 H1 才 + 说明 dynamic policy 可能有直接性能价值。 + +主要指标：每个 K 在相同 SLO 下可达到的最大 `sampling_u`，以及对应请求率、TPOT +pass rate、p50/p95 TPOT、成功/失败原因。`sampling_u` 是现有 replay 使用的一致 trace +抽样旋钮，因此只能作为此 trace 的 SLO-goodput proxy，不能直接外推为线上 QPS。 + +本 trace 的 output-length 分布很重尾；replayer 的 drain deadline 可能在长输出尚未完成 +时终止 probe。故每个结果必须同时报告 completion-success count 和 deadline failure；不能 +只因 TPOT pass rate 达标就把截断 request 当作“无成本”。本轮的原始 static screen 仍沿用 +现有 SLO 以便和项目 baseline 可比，但它不能替代完整 completion 的确认实验。 + +判定门槛（预注册）： + +1. 在 K=1/2/3 之间，最优 K 相对次优 K 的最大可行 `sampling_u` 小于 5%，或 + 置信区间/重复实验重叠很大： + **停止**。 +2. 最优 speculative K 比次优 speculative K 至少高 10%，且在两个独立重复中方向一致： + 进入 G1。NoSpec 仅作为“是否值得用 draft model”的部署对照，不能替代这条判据。 +3. 如果 K=3 不受当前 engine 支持、任一配置启动失败，记录为兼容性结果，不把它误作 + 性能差。 + +## 第一阶段：static-K screening + +配置为 `{NoSpec, K=1, K=2, K=3}`。NoSpec 会删除 `--speculative-config`，而不是传 +非法的 `num_speculative_tokens=0`。它释放 draft model 相关资源，因此不等于 +same-stack 的 logical K=0；后者在 EAGLE 类实现中仍可能需要一次 draft forward 来保持 +KV 同步。当前 dash0 binary 的 MLA indexer 明确限制 `num_speculative_tokens <= 3`， +故这已穷尽该 binary 的合法 static horizon。原计划每个配置： + +- 搜索范围 `sampling_u in [0.005, 0.020]`； +- 最多 3 次 probe、tolerance=0.003； +- 每个 probe 使用完整 600 秒 trace replay（不会使用 `max_requests_per_probe` 的 + 截断模式）； +- 启动顺序 `2,1,3,0`，降低冷启动或时间漂移与 K 单调对应的风险； +- 每个 K 都使用独立 Store、不可变派生 StudySpec、完整 stdout/stderr log。 + +这是一轮筛选而非 final frontier。它一旦显示值得继续，才对 top-2 K 做交叉顺序的完整 +搜索与至少两次重复。 + +### 2026-07-13 数据质量修正：controlled screen + +原始 trace 第一个 K=2 probe 暴露了一个不应隐藏的 measurement red flag：worker 的 +drain deadline 按 selected set 的 p99 output length 计算，而该 set 含一个 36,034-token +completion。该 request 因 deadline 被裁掉；尽管 TPOT-only pass rate 仍可达标，censoring +会随 `sampling_u` 改变 selected set，不能用于比较 static-K frontier。 + +因此原始 run 只保留为诊断 artifact，停止后改跑一个明确标为 **controlled** 的 screen： +保持相同 arrival、prompt、sampling seed 和 topology，但把每个 request 的 +`min_tokens=max_tokens=4096`。4096 接近原始 output mean 3,924.6，且使 p99 deadline 覆盖 +每个 request 的完整 completion。它回答的是“在相同输入/到达条件下，static K 是否留下 +可利用空间”，不是 production trace 的最终 goodput 结论。最终论文实验必须同时有： + +1. 该受控 curve（机制和可重复性）； +2. 原始长度 trace 的完整-completion 版本（不能使用 p99 censoring）； +3. 至少一个 held-out window。 + +### 2026-07-13 数据质量修正 #2：fresh-engine fixed grid + +受控 screen 的第一条 K=2、`u=0.0125` probe 本身通过了完整性检查（152/152 success、 +usage 返回的 completion token 均精确为 4096、无 early stop）。但随后发现原二分搜索会 +在同一 engine 内连续执行多个 probe；第二、三 probe 继承前一 probe 的 prefix/KV cache， +也继承全局 RNG 的已消耗状态。不同 K 的二分分支/中止路径不同，因此不能把该搜索输出的 +`best_sampling_u` 差异直接归因于 K。该 run 在第二 probe 中主动停止，**不作为 G0 结果**。 + +替代协议是每个 `(offered-load, K)` 只运行一次、每次均启动一个 fresh engine。两个固定 +负载由原始 immutable trace 的统一 `sampling_u` 阈值物化：`u=0.0125`（152 requests, +0.2533 req/s）和 `u=0.0200`（263 requests, 0.4383 req/s）。物化后的 request 仍保留原 +prompt、arrival 和 sampling provenance，但强制 `temperature=0`、显式 engine `seed=0`， +并用统一的 4096 completion override。每个 K 只有一个 probe，所以 accelerator KV/prefix +cache 为空且 RNG 从相同 seed 开始；每次都从 `probe_details.jsonl` 验证： + +1. `early_stopped=false`； +2. outcome count = selected count，且每个 request success； +3. TTFT/TPOT 均非空； +4. completion token 的 source 为 usage，且实际/预期均为 4096； +5. result 无 partial-probe failure，且只包含一个 primary probe。 + +运行顺序在两个负载间反向（ABBA），避免 K 与时间漂移完全共线。这个 grid 仍然只回答 +static-K 是否存在足够大的机会；它不估计 production sampling goodput，也不证明 rank-local +K 的上界。当前 vLLM deployment 的 `reasoning_parser=''`；其 SSE 实现将生成文本放在 +`delta.content`，所以本协议中的 token-time 定义覆盖当前 `` 输出。若以后启用 +reasoning parser，客户端必须同时记录 `reasoning_content` 后才可复用此指标。 + +## G1：只有在 G0 通过后才做的直接验证 + +目标不是“不同请求有不同 K”这种已经很常见的说法，而是验证下面的系统命题： + +> 在 DP+EP MoE 下，局部独立的 K 决策会让 collective 序列分歧；把它们编译为 +> rank-agreement 的 ragged execution plan，可保留异质请求的计算节省，同时不改变 +> collective order。 + +当前 dash0 vLLM 已经有一个很好的切入点：每个 DP step 会 all-reduce 一段 metadata， +并把各 rank 的 total token count padding 到最大值；CUDA graph mode 也会取跨 rank 的 +共同模式。这说明论文的最小机制不应另造一个 scheduler，而应把现有 scalar +`(num_tokens, num_reqs, graph_mode)` agreement 扩展为 canonical speculative-plan header。 +关键增量是让 header 描述真实 active frontiers，并保证后续 verifier/EP split vector 的 +collective ordinal 相同；若最后仍 padding 到 global max，就没有可主张的性能机制。 + +需要实现/测量： + +1. **oracle trace replayer**：利用 G0 的 per-K service curve，为每个到达时刻选择 + SLO-feasible K，比较 best-static K 与 oracle 的 upper bound。若 oracle gain <10%， + 停止，避免把噪声当论文方向。 +2. **collective trace**：按 DP rank 记录每个 decode step 的 collective 序列、token + shape、active-sequence mask、MoE all-to-all bytes 和 rank idle time。验证“local K + 不同”是否真的导致 sequence divergence，而不是仅是一个 API 限制。 +3. **CollectiveSpec prototype**：固定 collective order，用全局 agreement header 和 + ragged/padded verification plan；对比 `best static K`、global-max-K、oracle 和当前 + upstream dynamic-spec baseline（包括 DSpark/FASER 能实现的部分）。 +4. **ablation**：去掉 agreement、去掉 ragged packing、去掉 queue/SLO policy；报告 + goodput、p50/p95/p99 TPOT、acceptance、MoE communication bytes、GPU SM/HBM util、 + rank skew。 + +## 主要风险 + +- 最新 upstream 动态投机对 DP>1 的处理可能本身只需一个 global-K broadcast；那是 + feature patch，不构成研究贡献。 +- 当前 dash0 runtime 已验证 DP=2 + static EAGLE 可以工作；尚未在这个 binary 上证明 + “local dynamic K 会 deadlock”。因此研究动机必须写成固定 EAGLE horizon 的执行限制， + 不能把未运行的 dynamic-K 路径当作既成故障。 +- FASER/DSpark 等相邻工作会把“dynamic K + ragged verify”作为强 baseline；必须在 + 做任何大实现前进行逐项复现/排除。 +- trace 的 `sampling_u` 是 proxy；最终结论必须在固定 arrival trace、真实请求长度和 + 至少一个不同 workload 上复现。 diff --git a/scripts/collectivespec/create_controlled_greedy_window.py b/scripts/collectivespec/create_controlled_greedy_window.py new file mode 100644 index 0000000..148494d --- /dev/null +++ b/scripts/collectivespec/create_controlled_greedy_window.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python3 +"""Materialize a fixed-load, greedy subset of an immutable replay window. + +The output preserves prompt/arrival/sampling provenance while setting every +selected request's temperature to zero. It is intentionally a controlled +mechanism workload, not a replacement for a production sampling trace. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +from pathlib import Path +from typing import Any + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def resolve_trace_path(windows_path: Path, trace_file: str) -> Path: + candidate = Path(trace_file) + if candidate.is_absolute(): + return candidate + direct = (windows_path.parent / candidate).resolve() + if direct.exists(): + return direct + if candidate.parts and candidate.parts[0] == "trace_windows": + return (windows_path.parent / Path(*candidate.parts[1:])).resolve() + return direct + + +def numeric(value: Any, *, field: str, row_index: int) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise SystemExit(f"row {row_index}: {field} must be numeric") + value = float(value) + if not math.isfinite(value): + raise SystemExit(f"row {row_index}: {field} must be finite") + return value + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--windows-path", type=Path, required=True) + parser.add_argument("--window-id", required=True) + parser.add_argument("--sampling-u-threshold", type=float, required=True) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--output-window-id", required=True) + args = parser.parse_args() + + if not args.windows_path.is_file(): + raise SystemExit(f"windows file does not exist: {args.windows_path}") + threshold = float(args.sampling_u_threshold) + if not math.isfinite(threshold) or not 0.0 <= threshold <= 1.0: + raise SystemExit("--sampling-u-threshold must be finite and in [0, 1]") + + source_windows = json.loads(args.windows_path.read_text(encoding="utf-8")) + windows = source_windows.get("windows") if isinstance(source_windows, dict) else source_windows + if not isinstance(windows, list): + raise SystemExit("windows payload must contain a list") + source_window = next( + ( + item + for item in windows + if isinstance(item, dict) and item.get("window_id") == args.window_id + ), + None, + ) + if source_window is None: + raise SystemExit(f"window id not found: {args.window_id}") + trace_file = source_window.get("trace_file") + if not isinstance(trace_file, str) or not trace_file: + raise SystemExit(f"window {args.window_id} has no trace_file") + source_trace = resolve_trace_path(args.windows_path, trace_file) + if not source_trace.is_file(): + raise SystemExit(f"source trace does not exist: {source_trace}") + + args.output_dir.mkdir(parents=True, exist_ok=True) + output_trace = args.output_dir / f"{args.output_window_id}.jsonl" + selected = 0 + sampling_values: list[float] = [] + timestamps: list[float] = [] + with source_trace.open("r", encoding="utf-8") as src, output_trace.open( + "w", encoding="utf-8" + ) as dest: + for row_index, line in enumerate(src): + if not line.strip(): + continue + row = json.loads(line) + if not isinstance(row, dict): + raise SystemExit(f"row {row_index}: expected an object") + sampling_u = numeric(row.get("sampling_u"), field="sampling_u", row_index=row_index) + if sampling_u > threshold: + continue + timestamp = numeric(row.get("timestamp"), field="timestamp", row_index=row_index) + row["temperature"] = 0.0 + dest.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") + selected += 1 + sampling_values.append(sampling_u) + timestamps.append(timestamp) + if selected == 0: + output_trace.unlink(missing_ok=True) + raise SystemExit("threshold selected zero requests") + + output_window = dict(source_window) + output_window.update( + { + "window_id": args.output_window_id, + "trace_file": str(output_trace.resolve()), + "num_requests": selected, + "sampling_strategy": "fixed_uniform_score_then_greedy_temperature", + "collectivespec_source_window_id": args.window_id, + "collectivespec_sampling_u_threshold": threshold, + "collectivespec_temperature_override": 0.0, + "collectivespec_source_trace_path": str(source_trace), + "collectivespec_source_trace_sha256": sha256(source_trace), + } + ) + output_windows = { + "kind": "collectivespec_controlled_greedy_window", + "source_windows_path": str(args.windows_path.resolve()), + "source_window_id": args.window_id, + "sampling_u_threshold": threshold, + "temperature_override": 0.0, + "windows": [output_window], + } + output_manifest = args.output_dir / "windows.json" + output_manifest.write_text( + json.dumps(output_windows, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + sanity = { + "n": selected, + "sampling_u": { + "min": min(sampling_values), + "max": max(sampling_values), + "distinct_value_count": len(set(sampling_values)), + "all_at_or_below_threshold": all(value <= threshold for value in sampling_values), + }, + "timestamp": { + "min": min(timestamps), + "max": max(timestamps), + "non_negative": all(value >= 0 for value in timestamps), + "nondecreasing": all( + timestamps[index] >= timestamps[index - 1] + for index in range(1, len(timestamps)) + ), + }, + "temperature": {"distinct_value_count": 1, "all_zero": True}, + "source_trace_sha256": sha256(source_trace), + "output_trace_sha256": sha256(output_trace), + } + print( + json.dumps( + { + "windows_path": str(output_manifest), + "trace_path": str(output_trace), + "data_sanity": sanity, + }, + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/collectivespec/run_static_k_pilot.py b/scripts/collectivespec/run_static_k_pilot.py new file mode 100644 index 0000000..6b52f1f --- /dev/null +++ b/scripts/collectivespec/run_static_k_pilot.py @@ -0,0 +1,237 @@ +#!/usr/bin/env python3 +"""Run an isolated, reproducible static-speculation screening study. + +This driver deliberately does not implement CollectiveSpec. It establishes +whether the current static-K deployment has enough headroom to justify such a +system. Each K gets an immutable derived StudySpec and a separate store. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import hashlib +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import Any + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base-spec", type=Path, required=True) + parser.add_argument("--source-root", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + parser.add_argument("--run-id", required=True) + parser.add_argument("--k-values", nargs="+", type=int, default=[0, 1, 2, 3]) + parser.add_argument( + "--order", + nargs="+", + type=int, + help="Sequential launch order. Defaults to --k-values order.", + ) + parser.add_argument("--sampling-u-low", type=float, default=0.005) + parser.add_argument("--sampling-u-high", type=float, default=0.020) + parser.add_argument("--tolerance", type=float, default=0.003) + parser.add_argument("--max-probes", type=int, default=3) + parser.add_argument( + "--completion-tokens-override", + type=int, + help="Force every request to this exact completion length for a controlled screen.", + ) + parser.add_argument( + "--trace-windows-path", + type=Path, + help="Override trace.windows_path in every derived StudySpec.", + ) + parser.add_argument( + "--trace-window-id", + help="Override trace.window_id in every derived StudySpec.", + ) + parser.add_argument( + "--seed", + type=int, + help="Explicit engine seed recorded in every derived StudySpec.", + ) + parser.add_argument( + "--port", + type=int, + help="Serving port. Defaults to the base StudySpec's engine/base flag port.", + ) + parser.add_argument( + "--continue-on-error", + action="store_true", + help="Record a failed K and proceed to the next independent K.", + ) + parser.add_argument("--dry-run", action="store_true") + return parser.parse_args() + + +def derived_spec( + base: dict[str, Any], args: argparse.Namespace, k: int +) -> dict[str, Any]: + spec = json.loads(json.dumps(base)) + label = f"collectivespec-static-k{k}-screen-{args.run_id}" + spec["study_id"] = label + spec["engine"]["cwd"] = str(args.source_root) + flags = spec["engine"]["base_flags"] + port = args.port + if port is None: + port = int(flags.get("port", spec["engine"]["port"])) + spec["engine"]["port"] = port + flags["port"] = port + + if k == 0: + # vLLM's EAGLE config rejects num_speculative_tokens=0. Removing the + # entire flag is the real no-speculation baseline. + flags.pop("speculative-config", None) + else: + raw_config = flags.get("speculative-config") + if raw_config is None: + raise ValueError("base spec has no speculative-config for K > 0") + config = json.loads(raw_config) + config["num_speculative_tokens"] = k + flags["speculative-config"] = json.dumps(config, separators=(",", ":")) + + search = spec["search"] + search["low"] = args.sampling_u_low + search["high"] = args.sampling_u_high + search["tolerance"] = args.tolerance + search["max_probes"] = args.max_probes + if args.completion_tokens_override is not None: + spec["trace"]["completion_tokens_override"] = args.completion_tokens_override + if args.trace_windows_path is not None: + spec["trace"]["windows_path"] = str(args.trace_windows_path) + if args.trace_window_id is not None: + spec["trace"]["window_id"] = args.trace_window_id + if args.seed is not None: + flags["seed"] = args.seed + return spec + + +def main() -> int: + args = parse_args() + if not args.base_spec.is_file(): + raise SystemExit(f"base spec does not exist: {args.base_spec}") + if not args.source_root.is_dir(): + raise SystemExit(f"source root does not exist: {args.source_root}") + if any(k < 0 for k in args.k_values): + raise SystemExit("K must be non-negative") + if not 0 <= args.sampling_u_low < args.sampling_u_high <= 1: + raise SystemExit("sampling-u bounds must satisfy 0 <= low < high <= 1") + if args.max_probes < 1: + raise SystemExit("max-probes must be positive") + if args.completion_tokens_override is not None and args.completion_tokens_override < 0: + raise SystemExit("completion token override must be non-negative") + if (args.trace_windows_path is None) != (args.trace_window_id is None): + raise SystemExit( + "--trace-windows-path and --trace-window-id must be provided together" + ) + if args.trace_windows_path is not None and not args.trace_windows_path.is_file(): + raise SystemExit(f"trace windows file does not exist: {args.trace_windows_path}") + + order = args.order or args.k_values + if sorted(order) != sorted(args.k_values) or len(order) != len(args.k_values): + raise SystemExit("--order must be a permutation of --k-values") + + base = json.loads(args.base_spec.read_text()) + args.output_root.mkdir(parents=True, exist_ok=True) + specs_dir = args.output_root / "specs" + specs_dir.mkdir(exist_ok=True) + + manifest = { + "kind": "collectivespec_static_k_screen", + "created_at_utc": dt.datetime.now(dt.timezone.utc).isoformat(), + "base_spec": str(args.base_spec), + "base_spec_sha256": sha256(args.base_spec), + "source_root": str(args.source_root), + "run_id": args.run_id, + "k_values": args.k_values, + "launch_order": order, + "sampling_u": [args.sampling_u_low, args.sampling_u_high], + "tolerance": args.tolerance, + "max_probes": args.max_probes, + "completion_tokens_override": args.completion_tokens_override, + "trace_windows_path": ( + str(args.trace_windows_path) if args.trace_windows_path is not None else None + ), + "trace_window_id": args.trace_window_id, + "seed": args.seed, + "port": args.port + if args.port is not None + else int(base["engine"]["base_flags"].get("port", base["engine"]["port"])), + "python": sys.executable, + } + (args.output_root / "manifest.json").write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n" + ) + + failures: list[dict[str, Any]] = [] + for ordinal, k in enumerate(order, start=1): + spec = derived_spec(base, args, k) + spec_path = specs_dir / f"{ordinal:02d}_k{k}.json" + spec_path.write_text(json.dumps(spec, indent=2, sort_keys=True) + "\n") + store_root = args.output_root / "stores" / f"k{k}" + command = [ + sys.executable, + "-m", + "aituner.cli", + "study", + "tune", + "--spec", + str(spec_path), + "--store-root", + str(store_root), + "--max-trials", + "1", + ] + print(f"[{ordinal}/{len(order)}] K={k} command={json.dumps(command)}", flush=True) + if args.dry_run: + continue + + log_path = args.output_root / "logs" / f"{ordinal:02d}_k{k}.log" + log_path.parent.mkdir(exist_ok=True) + env = os.environ.copy() + env["PYTHONPATH"] = str(args.source_root / "src") + env["PYTHONDONTWRITEBYTECODE"] = "1" + with log_path.open("w") as log: + completed = subprocess.run( + command, + cwd=args.source_root, + env=env, + text=True, + stdout=log, + stderr=subprocess.STDOUT, + ) + print( + f"[{ordinal}/{len(order)}] K={k} returncode={completed.returncode} log={log_path}", + flush=True, + ) + if completed.returncode: + failures.append({"k": k, "returncode": completed.returncode, "log": str(log_path)}) + if not args.continue_on_error: + break + + result = { + "finished_at_utc": dt.datetime.now(dt.timezone.utc).isoformat(), + "failures": failures, + "status": "ok" if not failures else "completed_with_failures", + } + (args.output_root / "driver_result.json").write_text( + json.dumps(result, indent=2, sort_keys=True) + "\n" + ) + return 0 if not failures else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/collectivespec/summarize_engine_metrics.py b/scripts/collectivespec/summarize_engine_metrics.py new file mode 100644 index 0000000..559c092 --- /dev/null +++ b/scripts/collectivespec/summarize_engine_metrics.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +"""Extract per-DP-window speculative metrics from a vLLM engine log. + +vLLM emits an ``Engine NNN`` throughput line followed by the corresponding +``SpecDecoding metrics`` line. This script keeps that adjacency explicit and +does not infer request-level outcomes from the aggregate metrics. +""" + +from __future__ import annotations + +import argparse +import json +import math +import re +import statistics +from collections import defaultdict +from pathlib import Path +from typing import Any + + +ENGINE = re.compile( + r"INFO (?P\d\d-\d\d \d\d:\d\d:\d\d\.\d+) .*?" + r"Engine (?P\d+): .*?" + r"Avg decode step duration: (?P[0-9.]+) ms .*?" + r"Running: (?P\d+) reqs, Waiting: (?P\d+) reqs" +) +SPEC = re.compile( + r"INFO (?P\d\d-\d\d \d\d:\d\d:\d\d\.\d+) .*?" + r"SpecDecoding metrics: Mean acceptance length: (?P[0-9.]+), .*?" + r"Avg Draft acceptance rate: (?P[0-9.]+)%" +) + + +def percentile(values: list[float], q: float) -> float | None: + if not values: + return None + values = sorted(values) + index = (len(values) - 1) * q + lower, upper = math.floor(index), math.ceil(index) + if lower == upper: + return values[lower] + return values[lower] + (values[upper] - values[lower]) * (index - lower) + + +def describe(values: list[float]) -> dict[str, Any]: + return { + "n": len(values), + "mean": statistics.fmean(values) if values else None, + "min": min(values) if values else None, + "p50": percentile(values, 0.5), + "p95": percentile(values, 0.95), + "max": max(values) if values else None, + "distinct_value_count": len(set(values)), + } + + +def extract(path: Path) -> dict[str, Any]: + pending: tuple[str, dict[str, Any]] | None = None + rows: list[dict[str, Any]] = [] + for line in path.read_text(errors="replace").splitlines(): + engine = ENGINE.search(line) + if engine: + pending = ( + engine.group("clock")[:17], + { + "clock": engine.group("clock"), + "engine": int(engine.group("engine")), + "decode_ms": float(engine.group("decode_ms")), + "running": int(engine.group("running")), + "waiting": int(engine.group("waiting")), + }, + ) + continue + spec = SPEC.search(line) + if not spec or pending is None: + continue + clock, row = pending + if spec.group("clock")[:17] != clock: + pending = None + continue + row["mean_accept_length"] = float(spec.group("mean_accept")) + row["accept_pct"] = float(spec.group("accept_pct")) + rows.append(row) + pending = None + + by_engine: dict[int, list[dict[str, Any]]] = defaultdict(list) + for row in rows: + by_engine[row["engine"]].append(row) + summaries: list[dict[str, Any]] = [] + for engine, items in sorted(by_engine.items()): + summaries.append( + { + "engine": engine, + "window_count": len(items), + "accept_pct": describe([item["accept_pct"] for item in items]), + "mean_accept_length": describe([item["mean_accept_length"] for item in items]), + "decode_ms": describe([item["decode_ms"] for item in items]), + "running_requests": describe([float(item["running"]) for item in items]), + } + ) + acceptance = [row["accept_pct"] for row in rows] + return { + "source": str(path), + "records": rows, + "per_engine": summaries, + "data_sanity": { + "accept_pct": { + "n": len(acceptance), + "min": min(acceptance) if acceptance else None, + "max": max(acceptance) if acceptance else None, + "distinct_value_count": len(set(acceptance)), + "within_0_100": all(0 <= item <= 100 for item in acceptance), + }, + "invariants": { + "non_negative_decode_ms": all(row["decode_ms"] >= 0 for row in rows), + "non_negative_running": all(row["running"] >= 0 for row in rows), + "all_records_have_engine": all("engine" in row for row in rows), + }, + }, + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--engine-log", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + result = extract(args.engine_log) + args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n") + print(json.dumps({"per_engine": result["per_engine"], "data_sanity": result["data_sanity"]}, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/collectivespec/summarize_fixed_grid.py b/scripts/collectivespec/summarize_fixed_grid.py new file mode 100644 index 0000000..5bb665f --- /dev/null +++ b/scripts/collectivespec/summarize_fixed_grid.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +"""Summarize a one-probe-per-K fresh-engine static-K grid conservatively. + +Unlike the frontier helper, this reader treats ``probe_details.jsonl`` as the +source of completion and metric-integrity checks. A missing or partial probe +is reported as invalid rather than silently summarized. +""" + +from __future__ import annotations + +import argparse +import json +import math +from pathlib import Path +from typing import Any + + +def load_json(path: Path) -> Any: + return json.loads(path.read_text(encoding="utf-8")) + + +def first(path: Path, pattern: str) -> Path | None: + values = sorted(path.glob(pattern)) + return values[0] if values else None + + +def num(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + value = float(value) + return value if math.isfinite(value) else None + + +def read_one(root: Path, k: int, expected_tokens: int | None) -> dict[str, Any]: + record: dict[str, Any] = {"k": k, "label": "NoSpec" if k == 0 else f"K={k}"} + store = root / "stores" / f"k{k}" + result_path = first(store, "*/trials/trial-*/result.json") + history_path = first(store, "*/trials/trial-*/probe_history.json") + details_path = first(store, "*/trials/trial-*/probe_details.jsonl") + record.update( + { + "result_path": str(result_path) if result_path else None, + "probe_history_path": str(history_path) if history_path else None, + "probe_details_path": str(details_path) if details_path else None, + } + ) + failures: list[str] = [] + if result_path is None or history_path is None or details_path is None: + failures.append("missing_artifact") + record.update({"valid": False, "integrity_failures": failures}) + return record + + result = load_json(result_path) + history = load_json(history_path) + detail_rows = [json.loads(line) for line in details_path.read_text(encoding="utf-8").splitlines() if line] + if result.get("status") != "completed": + failures.append(f"result_status={result.get('status')}") + if result.get("best_source") != "primary_search": + failures.append(f"best_source={result.get('best_source')}") + if result.get("completed_with_probe_failure"): + failures.append("completed_with_probe_failure") + if not isinstance(history, list) or len(history) != 1: + failures.append("history_not_exactly_one_probe") + if len(detail_rows) != 1: + failures.append("details_not_exactly_one_probe") + if failures: + record.update({"valid": False, "integrity_failures": failures}) + return record + + probe = history[0] + details = detail_rows[0] + outcomes = details.get("outcomes") if isinstance(details.get("outcomes"), list) else [] + request_count = int(probe.get("request_count", -1)) + if details.get("early_stopped") or probe.get("early_stopped"): + failures.append("early_stopped") + if len(outcomes) != request_count: + failures.append(f"outcome_count={len(outcomes)}_expected={request_count}") + successful = sum(bool(item.get("success")) for item in outcomes if isinstance(item, dict)) + tpot_present = sum(item.get("tpot_ms") is not None for item in outcomes if isinstance(item, dict)) + ttft_present = sum(item.get("ttft_ms") is not None for item in outcomes if isinstance(item, dict)) + tokens_verified = all( + isinstance(item, dict) + and item.get("completion_tokens_source") == "usage" + and item.get("completion_tokens") == expected_tokens + and item.get("expected_completion_tokens") == expected_tokens + for item in outcomes + ) + if successful != request_count: + failures.append(f"success_count={successful}_expected={request_count}") + if tpot_present != request_count: + failures.append(f"tpot_count={tpot_present}_expected={request_count}") + if ttft_present != request_count: + failures.append(f"ttft_count={ttft_present}_expected={request_count}") + if expected_tokens is not None and not tokens_verified: + failures.append("completion_tokens_not_all_usage_verified") + latency = probe.get("latency_summary") if isinstance(probe.get("latency_summary"), dict) else {} + tpot = latency.get("tpot_ms") if isinstance(latency.get("tpot_ms"), dict) else {} + record.update( + { + "valid": not failures, + "integrity_failures": failures, + "threshold": num(probe.get("threshold")), + "request_count": request_count, + "request_rate": num(probe.get("request_rate")), + "feasible": bool(probe.get("feasible")), + "pass_rate": num(probe.get("pass_rate")), + "success_count": successful, + "ttft_count": ttft_present, + "tpot_count": tpot_present, + "tpot_ms": {name: num(tpot.get(name)) for name in ("mean", "p50", "p90", "p95", "p99")}, + } + ) + return record + + +def data_sanity(records: list[dict[str, Any]]) -> dict[str, Any]: + valid = [record for record in records if record.get("valid")] + p95 = [num(record.get("tpot_ms", {}).get("p95")) for record in valid] + p95 = [value for value in p95 if value is not None] + pass_rates = [num(record.get("pass_rate")) for record in valid] + pass_rates = [value for value in pass_rates if value is not None] + request_rates = [num(record.get("request_rate")) for record in valid] + request_rates = [value for value in request_rates if value is not None] + return { + "n_configurations": len(records), + "n_valid": len(valid), + "all_valid": len(valid) == len(records), + "p95_tpot_ms": { + "n": len(p95), + "min": min(p95) if p95 else None, + "max": max(p95) if p95 else None, + "distinct_value_count": len(set(p95)), + "non_negative": all(value >= 0 for value in p95), + }, + "pass_rate": { + "n": len(pass_rates), + "min": min(pass_rates) if pass_rates else None, + "max": max(pass_rates) if pass_rates else None, + "distinct_value_count": len(set(pass_rates)), + "within_0_1": all(0 <= value <= 1 for value in pass_rates), + }, + "request_rate": { + "n": len(request_rates), + "min": min(request_rates) if request_rates else None, + "max": max(request_rates) if request_rates else None, + "distinct_value_count": len(set(request_rates)), + "non_negative": all(value >= 0 for value in request_rates), + }, + "invariants": { + "one_record_per_k": len({record["k"] for record in records}) == len(records), + "all_full_completions": all( + record.get("success_count") == record.get("request_count") + for record in valid + ), + "all_latency_values_present": all( + record.get("ttft_count") == record.get("request_count") + and record.get("tpot_count") == record.get("request_count") + for record in valid + ), + }, + } + + +def markdown(records: list[dict[str, Any]], sanity: dict[str, Any]) -> str: + lines = [ + "# CollectiveSpec fresh-engine fixed-grid summary", + "", + "| configuration | valid | feasible | offered req/s | completed | pass rate | p95 TPOT (ms) | p99 TPOT (ms) | integrity failures |", + "|---|---:|---:|---:|---:|---:|---:|---:|---|", + ] + for record in records: + tpot = record.get("tpot_ms") or {} + fmt = lambda value: "—" if value is None else f"{value:.6g}" + lines.append( + "| {label} | {valid} | {feasible} | {rate} | {completed}/{count} | {pass_rate} | {p95} | {p99} | {failures} |".format( + label=record["label"], + valid=record.get("valid", False), + feasible=record.get("feasible", "—"), + rate=fmt(record.get("request_rate")), + completed=record.get("success_count", "—"), + count=record.get("request_count", "—"), + pass_rate=fmt(record.get("pass_rate")), + p95=fmt(tpot.get("p95")), + p99=fmt(tpot.get("p99")), + failures=", ".join(record.get("integrity_failures") or []) or "—", + ) + ) + lines.extend(["", "## Data sanity", "", "```json", json.dumps(sanity, indent=2, sort_keys=True), "``", ""]) + return "\n".join(lines) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", type=Path, required=True) + parser.add_argument("--output-json", type=Path, required=True) + parser.add_argument("--output-md", type=Path, required=True) + args = parser.parse_args() + manifest = load_json(args.root / "manifest.json") + expected_tokens = manifest.get("completion_tokens_override") + if isinstance(expected_tokens, bool) or not isinstance(expected_tokens, int): + expected_tokens = None + records = [ + read_one(args.root, int(k), expected_tokens) + for k in sorted(int(k) for k in manifest["k_values"]) + ] + sanity = data_sanity(records) + output = {"manifest": manifest, "records": records, "data_sanity": sanity} + args.output_json.write_text(json.dumps(output, indent=2, sort_keys=True) + "\n", encoding="utf-8") + args.output_md.write_text(markdown(records, sanity), encoding="utf-8") + print(json.dumps(output, indent=2, sort_keys=True)) + return 0 if sanity["all_valid"] else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/collectivespec/summarize_static_k_pilot.py b/scripts/collectivespec/summarize_static_k_pilot.py new file mode 100644 index 0000000..f1a1262 --- /dev/null +++ b/scripts/collectivespec/summarize_static_k_pilot.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""Summarize immutable artifacts from ``run_static_k_pilot.py``. + +The output intentionally distinguishes an absent/failed result from an +SLO-infeasible probe, and includes a small data-sanity block for review. +""" + +from __future__ import annotations + +import argparse +import json +import math +from pathlib import Path +from typing import Any + + +def number(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + value = float(value) + return value if math.isfinite(value) else None + + +def first_result(store: Path) -> Path | None: + candidates = sorted(store.glob("*/trials/trial-*/result.json")) + return candidates[0] if candidates else None + + +def summarize_result(k: int, path: Path | None) -> dict[str, Any]: + record: dict[str, Any] = {"k": k, "label": "NoSpec" if k == 0 else f"K={k}"} + if path is None: + record.update({"status": "missing_result"}) + return record + + payload = json.loads(path.read_text()) + probes = payload.get("probes") if isinstance(payload.get("probes"), list) else [] + feasible = [probe for probe in probes if isinstance(probe, dict) and probe.get("feasible")] + best_payload = feasible[-1].get("payload", {}) if feasible else {} + latency = best_payload.get("latency_summary", {}) if isinstance(best_payload, dict) else {} + tpot = latency.get("tpot_ms", {}) if isinstance(latency, dict) else {} + record.update( + { + "status": str(payload.get("status", "unknown")), + "result_path": str(path), + "best_sampling_u": number(payload.get("best_sampling_u")), + "best_request_rate": number(payload.get("best_request_rate")), + "best_pass_rate": number(payload.get("best_pass_rate")), + "best_request_count": number(payload.get("best_request_count")), + "probe_count": len(probes), + "feasible_probe_count": len(feasible), + "probe_thresholds": [number(probe.get("threshold")) for probe in probes if isinstance(probe, dict)], + "best_tpot_ms": { + metric: number(tpot.get(metric)) for metric in ("mean", "p50", "p90", "p95", "p99") + }, + } + ) + return record + + +def sanity(records: list[dict[str, Any]]) -> dict[str, Any]: + completed = [r for r in records if r.get("status") == "completed"] + values = [r["best_sampling_u"] for r in completed if number(r.get("best_sampling_u")) is not None] + pass_rates = [r["best_pass_rate"] for r in completed if number(r.get("best_pass_rate")) is not None] + distinct = len(set(values)) + return { + "n_configurations": len(records), + "n_completed": len(completed), + "sampling_u": { + "n": len(values), + "min": min(values) if values else None, + "max": max(values) if values else None, + "distinct_value_count": distinct, + "all_identical": len(values) > 1 and distinct == 1, + "non_negative": all(value >= 0 for value in values), + }, + "pass_rate": { + "n": len(pass_rates), + "min": min(pass_rates) if pass_rates else None, + "max": max(pass_rates) if pass_rates else None, + "distinct_value_count": len(set(pass_rates)), + "within_0_1": all(0 <= value <= 1 for value in pass_rates), + }, + "invariants": { + "one_result_per_k": len({r["k"] for r in records}) == len(records), + "all_best_request_rates_non_negative": all( + (value := number(r.get("best_request_rate"))) is None or value >= 0 for r in completed + ), + }, + } + + +def to_markdown(records: list[dict[str, Any]], checks: dict[str, Any]) -> str: + lines = [ + "# CollectiveSpec static-K screening summary", + "", + "| configuration | status | max feasible sampling_u | request rate | pass rate | p95 TPOT (ms) | probes |", + "|---|---:|---:|---:|---:|---:|---:|", + ] + for r in records: + tpot = r.get("best_tpot_ms") or {} + fmt = lambda value: "—" if value is None else f"{value:.6g}" + lines.append( + "| {label} | {status} | {u} | {rate} | {pass_rate} | {p95} | {probes} |".format( + label=r["label"], + status=r["status"], + u=fmt(r.get("best_sampling_u")), + rate=fmt(r.get("best_request_rate")), + pass_rate=fmt(r.get("best_pass_rate")), + p95=fmt(tpot.get("p95")), + probes=r.get("probe_count", "—"), + ) + ) + lines.extend(["", "## Data sanity", "", "```json", json.dumps(checks, indent=2, sort_keys=True), "``", ""]) + return "\n".join(lines) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", type=Path, required=True) + parser.add_argument("--output-json", type=Path, required=True) + parser.add_argument("--output-md", type=Path, required=True) + args = parser.parse_args() + + manifest = json.loads((args.root / "manifest.json").read_text()) + k_values = [int(k) for k in manifest["k_values"]] + records = [summarize_result(k, first_result(args.root / "stores" / f"k{k}")) for k in sorted(k_values)] + checks = sanity(records) + output = {"manifest": manifest, "records": records, "data_sanity": checks} + args.output_json.write_text(json.dumps(output, indent=2, sort_keys=True) + "\n") + args.output_md.write_text(to_markdown(records, checks)) + print(json.dumps(output, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())