diff --git a/.research/experiments/qwen235-tp8-collective-profile-ablation-20260719.md b/.research/experiments/qwen235-tp8-collective-profile-ablation-20260719.md new file mode 100644 index 0000000..4b9365f --- /dev/null +++ b/.research/experiments/qwen235-tp8-collective-profile-ablation-20260719.md @@ -0,0 +1,64 @@ +# 实验 EXP-SIMFID-Q235-CC-TP8:真实 TP4/TP8 collective profile 消融 + +> **状态:** review 通过,执行中 +> +> 本 card 是 SHA、command、config、log 等 provenance 的唯一归宿;本轮只重跑 simulator,不重跑已经冻结的 48 个真机 trial。 + +## Claim 与决策 + +- **Parent claim:** Qwen235 Fixed-PD 的 30%+ TPOT/E2E selection regret,是否主要由 TP8 collective profile 缺失及 TP4 profile 与真实 serving backend 不匹配造成。 +- **目的:** 支持或反驳 mechanism hypothesis;不是用同一 workload 的 E2E calibration 修正 simulator。 +- **Competing hypotheses:** + - H1:collective profile coverage/backend mismatch 是排序反转的必要主因。换成与真机 serving 一致的 TP4/TP8 实测 profile 后,Frontier 的 Fixed-PD TPOT winner 从 TP8 翻到 TP4,mean/p90 TPOT 与 E2E selection regret 降到 10% 以内。 + - H2:collective mismatch 只解释部分误差。换 profile 后 TP8 仍是 Frontier winner,Fixed-PD TPOT/E2E regret 仍超过 10%;下一主因应定位 decode batch/state-conditioned MoE composition。 +- **事前预测:** 当前 Frontier 在 Fixed-PD 上预测 TP4/TP8 mean TPOT 为 87.77/61.59 ms,TP8 有 26.19 ms 优势;真实 TP4/TP8 为 21.04/27.99 ms。若新的 TP4/TP8 collective profile 使这个 26.19 ms 的 simulator margin 反转,则支持 H1;若不能,则支持 H2。 +- **判定规则:** 只以 frozen simulator rerun 的 winner 与真实 frozen surface 计算 selection regret。绝对 latency ratio 作为 secondary metric,不用它替代 selection verdict。 + +## Setup + +- **自变量:** + - A0:当前 `measured-allreduce.csv`;TP4 是 Qwen30 hidden=2048 的旧实测,且 profiler 只检查 FlashInfer 可用、没有证明每个 payload 的实际 dispatch;TP8 无行并静默 analytical fallback。 + - A1:Qwen235 serving-matched piecewise collective profile;TP4/TP8 都在 dash0 H20、vLLM 0.20.0 commit `88d34c640...` 上实测。Frozen server logs 证明真机同时使用 `disable_custom_all_reduce=true` 与 FlashInfer-TRTLLM `allreduce_rms` fusion;profile 对 fusion-eligible payload 测同一 FlashInfer communicator,对阈值外 payload 测真实 PyNCCL/symmetric fallback。 +- **控制变量:** Frontier commit、Qwen235 operator profiles、runtime contract、四类 frozen traces、候选配置、MNS/MBT、prefix policy、real results 与分析脚本全部不变。 +- **System context:** Qwen3-235B-A22B-FP8;vLLM 0.20.0+cu129;dash0 8×H20;`{TP4/EP1, TP8/EP8} × MNS{64,128}`;MBT=8192;Frontier piecewise graph path。 +- **Workload 或 trace:** 重跑四类 simulator surface:Fixed-PD 4096→256 @ 0.2 req/s/GPU、Fixed-PO 4096→1、Trace-PD、Trace-PO;每 cell 沿用原 129-request trace。Fixed-PD 是 primary,另外三类检查 profile 替换是否引入新的 selection regression。 +- **Profile protocol:** payload 覆盖所有 Qwen235 decode graph buckets(1--256,含真实 capture sizes)、fusion 阈值两侧 `{63,64,65}` / `{255,256,257}`,以及 512--8192 prefill sizes;每个 TP 与 payload 先 warmup,再保留 3×20 个 per-rank CUDA-event samples。raw JSON 记录实际 backend dispatch、fusion byte limit、world size、dtype、payload bytes、GPU/runtime/commit 与 source hashes。TP4/TP8 使用相同 payload grid;不把 microbenchmark 直接当作 E2E 结论。 +- **Profile contract:** H20/SM90 上 vLLM 0.20 的 fusion limit 是 TP4 2 MiB、TP8 0.5 MiB,即 Q235 BF16 hidden=4096 时分别为 256/64 tokens。simulator runner 启动前解析 CSV,要求所选 configs 的每个 `TP>1` 都有有限、正值的 measured rows;缺覆盖立即失败。结果 manifest 写入 CSV SHA-256、TP coverage、row counts 与 piecewise backend 集合。决策实验禁止 analytical fallback。 +- **Baselines:** A0 current Frontier、A1 measured-profile Frontier、frozen real hardware surface。 +- **Metrics:** profile latency median/p90 与跨 rank spread;simulated mean/p90 TTFT/TPOT/E2E;winner、selection regret、tau-b(可定义时);每个 TP 的 measured-profile hit/fallback counters。 + +## 预期产物与 review + +- **预期数据:** TP4/TP8 raw collective JSON;materialized Frontier CSV + manifest;四类 A1 simulator surface;A0/A1/real comparison JSON/Markdown;profile cost ledger。 +- **Figure prototype:** `../../runs/frontier-fidelity-envelope-v1/qwen235-collective-ablation-mock.png`;左图固定真实与 A0 TPOT,并为 A1 留待测 series;右图明确“winner flip→0% regret / unchanged→33% regret”的判定。它回答 profile 修复是否足以改变配置选择。 +- **人工 review:** 通过(2026-07-19,用户明确要求“推进实验”)。 +- **Review 意见:** 保留 frozen real surface,只补真实 TP4/TP8 profile 后重跑 simulator;每个 simulator 实验必须使用真实 profile,缺失 coverage 或运行时 analytical fallback 立即失败。 + +## Benchmark design audit(experiment-design-review) + +| Crime | Verdict | Severity | Evidence | Fix / gate | +|---|---|---|---|---| +| 用 microbenchmark 代替 E2E | PASS | — | collective profile 只作为自变量;结论来自完整 simulator surface 对 frozen real surface 的 selection regret | 保留 A0/A1/real 三方结果 | +| calibration set 等于 evaluation set | PASS | — | A1 只测 collective operator,不使用 real E2E latency 拟合参数 | 禁止 E2E scale/calibration | +| selective benchmarking | PASS | — | primary Fixed-PD 外,同时重跑另外三类 workload | 报告所有 16 个 simulator cells | +| 缺失平台/版本 | PASS | — | raw/manifest 绑定 H20、vLLM commit、model、backend 与 hashes | 任一 provenance 缺失则 profile 不可采纳 | +| 缺失方差 | NEEDS EVIDENCE | Major | 尚未执行 profile repeats | raw artifact 必须保留 per-rank repeated samples并报告 spread | +| profile 覆盖静默降级 | FAIL(A0) | Blocking | TP8 无 measured rows,Frontier 使用 analytical fallback | A1 runner fail-fast;fallback count 必须为 0 | +| backend/fusion 阈值未对齐 | FAIL(A0) | Blocking | 真机日志启用 FlashInfer `allreduce_rms`;vLLM 源码规定 H20 TP4/TP8 fusion limit 为 2/0.5 MiB,旧 CSV 未记录这条 piecewise contract | A1 在阈值两侧实测并记录每行 dispatch | + +**总体建议:** 已批准执行;coverage gate、backend match 与 provenance gate 任一不通过则 Block。 + +## 复现信息 + +- **Code:** AITuner branch `feature/sim`;本 card 创建时 HEAD `f4a75aa8e400ead4eb6d305178192e85940c6de6`,后续运行 commit 待填。vLLM source commit `88d34c6409e9fb3c7b8ca0c04756f061d2099eb1`。 +- **Environment:** dash0 8×H20;`/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1`;model `/home/admin/cpfs/wjh/models/Qwen/Qwen3-235B-A22B-FP8`。 +- **产物路径:** 待 review 后冻结;不得覆盖旧 campaign `/home/admin/cpfs/wjh/aituner/qwen235-v020-fourcase-20260719-r1`。 +- **已知 deviation:** `--disable-custom-all-reduce` 只关闭 vLLM custom AR,不关闭编译器的 FlashInfer `allreduce_rms` fusion。真机 TP4/TP8 日志都显示自动选择 `trtllm` workspace;旧 TP4 profiler 没有记录/执行真实 fusion-limit piecewise dispatch。A1 因此必须同时重测 TP4 与 TP8,不能只追加 TP8 行。 + +## 结果 + +- **观察事实:** 待运行。 +- **异常:** 待运行。 +- **含义:** 待运行。 +- **Claim update:** unchanged +- **下一步:** 依次完成 collective profiling、profile materialization、CPU simulator rerun 与 analysis。 diff --git a/.research/ongoing.md b/.research/ongoing.md new file mode 100644 index 0000000..7475282 --- /dev/null +++ b/.research/ongoing.md @@ -0,0 +1,72 @@ +# AITuner 研究当前状态 + +> 2026-07-17|写给未参与项目的读者,可直接作为 presentation 讲稿。历史过程与复现信息见 `../runs/*/` 各 experiment card、`../docs/` 各 campaign 文档。 +> +> **2026-07-19 update:** Qwen235 四类 workload matrix 已完成;Fixed-PD 出现 33.0--37.2% TPOT 与 30.7--34.6% E2E selection regret。当前发现 Q235 simulator 输入的 collective CSV 缺 TP8,并且旧 TP4 profile 未绑定真实 `allreduce_rms` fusion-limit dispatch。active ablation 见 [`experiments/qwen235-tp8-collective-profile-ablation-20260719.md`](experiments/qwen235-tp8-collective-profile-ablation-20260719.md)。 + +## 一眼看懂 + +- **Topic / problem:** LLM serving 的自动、低成本配置调优(AITuner)。当前主线问题:用 simulator 给部署配置(并行度、批量上限等)排序,什么时候可信?需要补多少真机证据?算上这些成本还划算吗? +- **Central claim:** simulator 要能帮助配置调优,必须先满足 scheduler transition 的 liveness/coverage,再满足「配置相关残差小于真机 decision margin」;前者决定 capacity 是否有定义,后者决定排序是否正确。(ID: C0) +- **当前结论:** 早先 35 个 trace stall 不是 Frontier scheduler liveness failure:adapter 为不满 16-token 的 prefix block 错误生成了 cache identity,Frontier 又没有 fail-fast。修正为完整 block、使用真实 graph buckets/KV blocks 和 `piecewise`/`KERNEL_ONLY` profile 后,Qwen30 Trace-PD 的全部 12 个 cell 完成 129/129 request,Frontier 对 TTFT/TPOT/E2E 的 6 个 argmin 均与三次 fresh-server 真机一致;但绝对 latency 仍高估 4--511×。这只证明一个 MoE Trace-PD surface 的 selection fidelity,不能外推到 prefill-only、fixed workload 或 235B。 +- **最大 uncertainty / risk:** 这个正确选择是否依赖 Trace-PD 的 queue/margin 偶然性;Fixed-PD、Trace-P、Fixed-P 与大型 FP8 MoE 是否仍保留正确 config ranking。Q235 还缺少同栈 profile/runtime contract,不能复用旧 vLLM 0.10.2 数据。 +- **下一项 critical action:** 完成 Qwen30 其余三项 fixed/trace × PD/P surface。Fixed case 已先启动无请求 vLLM runtime-state preflight;Trace-P 的 graph-aligned Frontier 12-cell CPU surface 正在运行。Q235 先做 vLLM 0.20 TP4/TP8 compatibility gate,随后才允许新 profile。 +- **停止条件:** T1 出 verdict 且成本账本建立后:pass 且摊销论证成立 → 转向「sim 剪枝 + 真机终选」的 hybrid 机制设计;fail → 转入失败机制归因;两条路都无 insight 增量 → 收敛写作。 + +## 核心概念 + +- **Frontier:** 本项目使用的 simulator,属 Vidur 系(直接使用 vidur backend),加自研 FP8/MoE/EP/decode-profile 兼容补丁。 +- **Regret:** 按 simulator 排序选配置,相对真机最优配置的性能损失百分比(以每 GPU capacity 计)。primary metric:排序选对则 regret=0。 +- **τ-b(Kendall tau-b):** simulator 排序与真机排序的秩相关,1 = 完全一致,−1 = 完全反序,0 = 无关;tie-aware。 +- **Decision margin:** 真机上头部配置之间的性能差距,即 simulator 误差的容忍带。 +- **Action-differential residual:** simulator 误差中随配置(action)不同而不同的部分。Why needed:所有配置统一偏移不影响排序,只有差异化残差才可能穿过 margin 改变选择——这解释了「绝对误差 33%」与「排序全对」为何可以同时成立。 +- **Capacity bracket:** 真机 anchor 为候选配置的 capacity 划出的上下界;「bracket 不反转」指未测的负载点不可能推翻 top 选择。 +- **Decision-valid coverage:** simulator 能从初始状态推进到所有请求完成,并为 config×workload cell 产生合法 SLO metric 的比例。若 reachable nonterminal state 没有 enabled transition/future event,capacity 与 rank 都没有定义,不能把该 cell 当作 infeasible。 +- **Workload realism 阶梯:** prefill-only(无 decode)→ fixed-shape mixed(固定输入输出长度的混合负载)→ trace-faithful mixed(生产 trace 忠实回放)。fidelity 结论不能向更高一级外推。 + +## Claim 层级 + +- **Central claim:** 见「一眼看懂」。(ID: C0) + - **Subclaim:** zero-shot 排序失败是真实现象。(ID: C1;supported) + - 30B 纯 profile 驱动的 regret 为 25.63%(τ-b=0);另一 throughput-proxy 评测口径下为 30.46%。Boundary:均发生在 capacity-point + SLO-gated selection——恰是 Vidur 论文自己声明预测误差会爆炸、评测刻意回避的 regime(见 claim map)。 + - **Subclaim:** 少量结构化的真机证据可以恢复低 regret 排序。(ID: C2) + - **Hypothesis(decision-bearing):** trace-faithful 回放下,同栈 profile + 真机 KV capacity + 兼容补丁、且不做逐案例端到端校准的 Frontier,能满足 gate:regret ≤5% ∧ τ-b ≥0.8 ∧ bracket 不反转。(ID: H1;weakened) + - **Supporting:** 235B prefill-only regret=0;235B fixed-shape mixed 的 top set 全中;30B 加 per-TP 校准后 regret 0.76%(但这是外部端到端 scale 给出的上界,不是原生 profile 保真度)。 + - **Counterevidence:** 修正 prefix trace contract 后的 TP2/MNS16 `none`-graph run 完成但 p50 TPOT 约 96 ms,真机为约 14 ms;然而该比较尚未对齐 real vLLM 的 `FULL_AND_PIECEWISE` graph path。 + - **下一项 discriminative experiment:** 补齐 `KERNEL_ONLY` graph family并以 `piecewise` 重跑相同 trace;若 full surface 仍错,graph omission 不再是可用解释。 + - **Hypothesis(机制,active):** 误差机制是 action-conditioned residual——执行状态的转移(并行拓扑、kernel family、graph mode、batch 组成)使按算子 profile 的组合预测跨配置不可复合;残差大于 margin 时排序失败。(ID: H2;supported) + - **Supporting:** 三个 TP 档的端到端校准系数为 0.72/0.47/0.35,残差确实随配置剧烈变化;235B 的批量上限交互预测错误但被 2× margin 容忍;30B prefill-only 在低负载近似对齐、饱和后按 TP 反向放大,最终 τ-b=−1。 + - **下一步:** 按 collective 通信 → batch 组成 → 调度器逐步轨迹的单变量顺序定位组合性失效点(对应 experiment card 中 A1/A2/A3 消融阶梯)。 + - **Subclaim:** 成本论证只有在摊销前提下成立。(ID: C3) + - **Hypothesis(active):** 每个 model×硬件×runtime 的一次性对齐成本,摊销到大配置面、频繁重调(引擎版本 churn 的频率证据见 claim map)或禁止在线实验的场景后,低于重复真机调优。(ID: H3;untested——分母已实测,分子未入账) + - **下一步:** 建 cost ledger(见「下一步」)。 + +## 当前 critical experiment + +- **Question:** 生产 trace 忠实回放(prefix 打开、原始到达时间与会话结构)下,best-effort Frontier 能否满足 low-regret gate? +- **为什么现在做:** 这是 H1 的判决实验;所有已完成的机制分解都在人工 workload 上,不能替代这个 verdict。 +- **当前状态:** Trace-PD 的 graph-aligned surface 已通过 selection gate,但绝对 latency 不通过 calibration。现在以不共享结果的 Fixed-PD、Trace-P、Fixed-P surface 检验它的泛化边界。 +- **Result → decision:** 若其它 surface 排序失败,保留 Trace-PD success 为条件化 envelope,并按 fixed/trace/prefill/decode 的差异定位 state composition;若都通过,才扩大到 Q235 或寻找 simulator 已解决范围之外的新问题。 +- **Experiment card:** [`../runs/frontier-fidelity-envelope-v1/experiment-card.md`](../runs/frontier-fidelity-envelope-v1/experiment-card.md) + +## Key evidence(最多 3 条) + +- **E1(否证「prefill-only 是充分 easy condition」,支持 H2):** 30B BF16、去掉 decode/prefix/混合 batch 后,真机最优是 TP4(8 vs 7 req/s/GPU),simulator 却把 TP4 排最差(6 vs 8):top set 无交集,regret 12.5%,τ-b=−1。产物:`../runs/frontier-phase-factorial-v0/results/final/`(dash0,12.07 H20-GPUh)。 +- **E2(margin 可掩盖残差,支持 C0/H2):** 235B FP8 的 prefill-only 与 fixed-shape mixed 都选对 top set,但 simulator 漏掉并发上限(max-num-seqs)×批 token 上限(max-num-batched-tokens)的交互棋盘格,并把 10/34 个真机 anchor 误判为不可行;真机头部 margin 有 2×,残差被容忍。产物:`../runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_evidence/`、`../runs/frontier-multicase-sufficiency-v1/results/t0-final/`。 +- **E3(selection 与 calibration 分开):** 用完整 block projection、`piecewise` 和 graph-compatible KERNEL_ONLY profile 后,Qwen30 Trace-PD 的 12/12 sim cells 完成,6 个 mean/p90 latency objective 的 argmin 都与真机一致;但 sim/real latency ratio 仍为 4--511×。产物:[`../runs/frontier-fidelity-envelope-v1/graph-piecewise-experiment-card.md`](../runs/frontier-fidelity-envelope-v1/graph-piecewise-experiment-card.md)。 + +## 下一步(最多 3 项) + +- [ ] **完成 Qwen30 remaining matrix(direct progress):** Fixed-PD、Trace-P、Fixed-P 分别冻结 Frontier/真机 12-cell surface;P-only 的 TPOT 一律为 N/A。 +- [ ] **Q235 portability gate:** 先验证 vLLM0.20 TP4/TP8 FP8 runtime 和 deadc4a profile provenance,再决定是否允许其 Fixed-P sweep。 +- [ ] **建 cost ledger:** parent H3;完成标准 = 每 case 一行(profiling GPU-h、补丁工时、校准探测、sim CPU-h),与已实测的真机调优成本同表,随每个 case 更新。 + +## Blocker 或 anomaly + +- **下一启动已准备:** dash0 8×H20 当前空闲;graph-compatible attention/linear/MoE/router kernel-only profile 会先在一 GPU smoke,成功后以 3 张 GPU 并行完成 attention shards。完整 replay 仍为 CPU-only。 +- **Anomaly(保留):** 235B pilot 中 simulator 把 10/34 个 anchor 误判为不可行——false-infeasible 是 H1 的主要威胁模式,T1 分析时须单独报告。 +- **勘误(已固定):** 统一平台为 dash0-only,早期 dash1 出处是文档错误;fixed-shape pilot 的主 SLO(TPOT 40ms)无判别力,150ms 是事后明示的敏感性分析,不得写成盲选的 primary。 + +## Related work + +- Claim map:[`../docs/simulator-claim-map-20260716.md`](../docs/simulator-claim-map-20260716.md)。核心缺口:capacity-point + SLO-gated selection 的 regret 无人用真机 ground-truth 面验证过;alignment 成本无人与真机调优成本放进同一张表比较。 diff --git a/runs/frontier-fidelity-envelope-v1/analyze_qwen235_v020_campaign.py b/runs/frontier-fidelity-envelope-v1/analyze_qwen235_v020_campaign.py index 25853e6..4f4c739 100644 --- a/runs/frontier-fidelity-envelope-v1/analyze_qwen235_v020_campaign.py +++ b/runs/frontier-fidelity-envelope-v1/analyze_qwen235_v020_campaign.py @@ -97,14 +97,18 @@ def compare_metric(real: dict, sim: dict, metric: str): def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--campaign-root", type=Path, required=True) + parser.add_argument("--real-root", type=Path) + parser.add_argument("--sim-root", type=Path) parser.add_argument("--json-output", type=Path, required=True) parser.add_argument("--markdown-output", type=Path, required=True) args = parser.parse_args() + real_root = args.real_root or args.campaign_root + sim_root = args.sim_root or args.campaign_root cases = {} lines = ["# Qwen235 vLLM 0.20 Frontier vs real", "", "| case | metric | Frontier winner | real winner | match | regret | pair agreement |", "|---|---|---|---|---:|---:|---:|"] for case in CASES: - real = real_surface(args.campaign_root, case) - sim = sim_surface(args.campaign_root, case) + real = real_surface(real_root, case) + sim = sim_surface(sim_root, case) metrics = ["ttft_mean_ms", "ttft_p90_ms", "e2e_mean_ms", "e2e_p90_ms"] if case.endswith("pd"): metrics[2:2] = ["tpot_mean_ms", "tpot_p90_ms"] @@ -118,7 +122,12 @@ def main() -> None: f"{item['pair_direction_agreement']:.1%} |" ) cases[case] = {"real": real, "sim": sim, "comparison": comparisons} - payload = {"schema": "qwen235-v020-simulator-real-comparison-v1", "cases": cases} + payload = { + "schema": "qwen235-v020-simulator-real-comparison-v1", + "real_root": str(real_root.resolve()), + "sim_root": str(sim_root.resolve()), + "cases": cases, + } args.json_output.parent.mkdir(parents=True, exist_ok=True) args.json_output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") args.markdown_output.write_text("\n".join(lines) + "\n") diff --git a/runs/frontier-fidelity-envelope-v1/jobs_qwen235_collective_profile.toml b/runs/frontier-fidelity-envelope-v1/jobs_qwen235_collective_profile.toml new file mode 100644 index 0000000..be41524 --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/jobs_qwen235_collective_profile.toml @@ -0,0 +1,45 @@ +version = 1 + +[[jobs]] +name = "qwen235-vllm020-serving-allreduce-tp4-20260719-v1" +gpus = 4 +gpu_model = "H20" +hosts = ["dash0"] +command = "cd /home/admin/cpfs/wjh/aituner/aituner/runs/frontier-qwen30-vllm020-profile-v1 && timeout --signal=TERM --kill-after=30s 1020 bash run_allreduce_profile.sh" +artifacts = ["artifacts/qwen235-serving-allreduce-tp4-20260719-v1"] + +[jobs.env] +XDG_CACHE_HOME = "/tmp/wjh/.cache" +VLLM_CACHE_ROOT = "/tmp/wjh/.cache/vllm" +TP = "4" +NUM_TOKENS = "1 2 4 8 16 24 32 40 48 56 63 64 65 72 80 88 96 104 112 120 128 136 144 152 160 168 176 184 192 200 208 216 224 232 240 248 255 256 257 512 1024 2048 4096 8192" +HIDDEN_DIM = "4096" +COLLECTIVE_CONTRACT = "qwen235-serving-projected" +TRIALS = "3" +REPEATS = "20" +MODEL = "/home/admin/cpfs/wjh/models/Qwen/Qwen3-235B-A22B-FP8" +OUTPUT_ROOT = "/home/admin/cpfs/wjh/aituner/gpu-fleet-fidelity-envelope-v1/artifacts/qwen235-serving-allreduce-tp4-20260719-v1" +VENV_ROOT = "/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1" +VLLM_SOURCE = "/home/admin/cpfs/wjh/agentic-kv/third_party/vllm_v20_build" + +[[jobs]] +name = "qwen235-vllm020-serving-allreduce-tp8-20260719-v1" +gpus = 8 +gpu_model = "H20" +hosts = ["dash0"] +command = "cd /home/admin/cpfs/wjh/aituner/aituner/runs/frontier-qwen30-vllm020-profile-v1 && timeout --signal=TERM --kill-after=30s 1020 bash run_allreduce_profile.sh" +artifacts = ["artifacts/qwen235-serving-allreduce-tp8-20260719-v1"] + +[jobs.env] +XDG_CACHE_HOME = "/tmp/wjh/.cache" +VLLM_CACHE_ROOT = "/tmp/wjh/.cache/vllm" +TP = "8" +NUM_TOKENS = "1 2 4 8 16 24 32 40 48 56 63 64 65 72 80 88 96 104 112 120 128 136 144 152 160 168 176 184 192 200 208 216 224 232 240 248 255 256 257 512 1024 2048 4096 8192" +HIDDEN_DIM = "4096" +COLLECTIVE_CONTRACT = "qwen235-serving-projected" +TRIALS = "3" +REPEATS = "20" +MODEL = "/home/admin/cpfs/wjh/models/Qwen/Qwen3-235B-A22B-FP8" +OUTPUT_ROOT = "/home/admin/cpfs/wjh/aituner/gpu-fleet-fidelity-envelope-v1/artifacts/qwen235-serving-allreduce-tp8-20260719-v1" +VENV_ROOT = "/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1" +VLLM_SOURCE = "/home/admin/cpfs/wjh/agentic-kv/third_party/vllm_v20_build" diff --git a/runs/frontier-fidelity-envelope-v1/materialize_qwen235_v020_allreduce.py b/runs/frontier-fidelity-envelope-v1/materialize_qwen235_v020_allreduce.py new file mode 100644 index 0000000..e1ca264 --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/materialize_qwen235_v020_allreduce.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""Validate Qwen235 serving collectives and materialize Frontier's CC CSV.""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import math +from pathlib import Path +from typing import Any + + +VLLM_COMMIT = "88d34c6409e9fb3c7b8ca0c04756f061d2099eb1" +TOKEN_POINTS = ( + 1, 2, 4, 8, 16, 24, 32, 40, 48, 56, 63, 64, 65, 72, 80, 88, 96, + 104, 112, 120, 128, 136, 144, 152, 160, 168, 176, 184, 192, 200, + 208, 216, 224, 232, 240, 248, 255, 256, 257, 512, 1024, 2048, 4096, + 8192, +) +TP_SIZES = (4, 8) +HIDDEN_DIM = 4096 +ALLOWED_BACKENDS = { + "flashinfer_trtllm_fused_projection", + "pynccl_symmetric_with_copy", + "torch_symmetric_memory", + "pynccl", +} +FIELDS = ( + "time_stats.all_reduce.min", + "time_stats.all_reduce.max", + "time_stats.all_reduce.mean", + "time_stats.all_reduce.median", + "time_stats.all_reduce.std", + "rank", + "num_workers", + "size", + "collective", + "devices_per_node", + "max_devices_per_node", +) + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def materialize( + input_paths: list[Path], output_path: Path, manifest_path: Path +) -> dict[str, Any]: + if len(input_paths) != 2: + raise ValueError("exactly two raw inputs are required for TP4 and TP8") + rows = [] + seen = set() + backends_by_tp: dict[int, set[str]] = {} + environments = [] + for path in input_paths: + payload = json.loads(path.read_text()) + if payload.get("schema_version") != "vllm020_allreduce_raw.v2": + raise ValueError(f"unexpected raw collective schema in {path}") + environment = payload["environment"] + if environment.get("vllm_version") != "0.20.0": + raise ValueError(f"vLLM version mismatch in {path}") + if environment.get("vllm_source_commit") != VLLM_COMMIT: + raise ValueError(f"vLLM commit mismatch in {path}") + if environment.get("gpu") != "NVIDIA H20": + raise ValueError(f"GPU mismatch in {path}: {environment.get('gpu')!r}") + if Path(environment.get("model", "")).name != "Qwen3-235B-A22B-FP8": + raise ValueError(f"model mismatch in {path}") + if environment.get("collective_contract") != "qwen235-serving-projected": + raise ValueError(f"collective contract mismatch in {path}") + if environment.get("disable_custom_all_reduce") is not True: + raise ValueError(f"custom all-reduce was not disabled in {path}") + backend_env = environment.get("backend_env", {}) + if backend_env.get("VLLM_ALLREDUCE_USE_FLASHINFER") != "1": + raise ValueError(f"FlashInfer projection was not enabled in {path}") + if backend_env.get("VLLM_ALLREDUCE_USE_SYMM_MEM") != "1": + raise ValueError(f"symmetric-memory contract mismatch in {path}") + environments.append(environment) + + for source in payload["rows"]: + tp = int(source["tensor_parallel_size"]) + tokens = int(source["num_tokens"]) + key = (tp, tokens) + if key in seen: + raise ValueError(f"duplicate collective row: {key}") + seen.add(key) + if tp not in TP_SIZES: + raise ValueError(f"unexpected TP: {tp}") + if int(source["hidden_dim"]) != HIDDEN_DIM: + raise ValueError(f"hidden-dim mismatch for {key}") + expected_bytes = tokens * HIDDEN_DIM * 2 + if int(source["payload_bytes"]) != expected_bytes: + raise ValueError(f"payload mismatch for {key}") + backend = source["selected_backend"] + if backend not in ALLOWED_BACKENDS: + raise ValueError(f"unexpected backend for {key}: {backend!r}") + fusion_limit = {4: 2 * 1024 * 1024, 8: 512 * 1024}[tp] + if int(source["real_fusion_limit_bytes"]) != fusion_limit: + raise ValueError(f"fusion limit mismatch for {key}") + expected_backend = ( + "flashinfer_trtllm_fused_projection" + if expected_bytes <= fusion_limit + else None + ) + if expected_backend is not None and backend != expected_backend: + raise ValueError(f"missing fused projection for {key}") + if expected_backend is None and backend == "flashinfer_trtllm_fused_projection": + raise ValueError(f"fused projection exceeds runtime limit for {key}") + backends_by_tp.setdefault(tp, set()).add(backend) + median = float(source["critical_path_median_ms"]) + if not math.isfinite(median) or median <= 0: + raise ValueError(f"invalid critical-path median for {key}") + if int(source.get("trials", 0)) != 3 or int( + source.get("repeats_per_trial", 0) + ) != 20: + raise ValueError(f"repeat contract mismatch for {key}") + samples = source.get("per_trial_rank_samples_ms", []) + if ( + len(samples) != 3 + or any(len(trial) != tp for trial in samples) + or any( + len(rank_samples) != 20 + or any( + not math.isfinite(float(value)) or float(value) <= 0 + for value in rank_samples + ) + for trial in samples + for rank_samples in trial + ) + ): + raise ValueError(f"raw sample coverage mismatch for {key}") + critical_samples = [ + max(float(trial[rank][repeat]) for rank in range(tp)) + for trial in samples + for repeat in range(20) + ] + ordered = sorted(critical_samples) + measured_median = (ordered[29] + ordered[30]) / 2 + if not math.isclose(median, measured_median, rel_tol=1e-6, abs_tol=1e-9): + raise ValueError(f"critical-path median/sample mismatch for {key}") + rows.append( + { + "time_stats.all_reduce.min": median, + "time_stats.all_reduce.max": median, + "time_stats.all_reduce.mean": median, + "time_stats.all_reduce.median": median, + "time_stats.all_reduce.std": 0.0, + "rank": 0, + "num_workers": tp, + "size": expected_bytes, + "collective": "all_reduce", + "devices_per_node": tp, + "max_devices_per_node": 8, + } + ) + + expected = {(tp, tokens) for tp in TP_SIZES for tokens in TOKEN_POINTS} + if seen != expected: + raise ValueError( + f"collective coverage mismatch: missing={expected - seen}, extra={seen - expected}" + ) + if output_path.exists() or manifest_path.exists(): + raise FileExistsError("refusing to overwrite immutable collective profile") + output_path.parent.mkdir(parents=True, exist_ok=True) + with output_path.open("w", newline="") as output: + writer = csv.DictWriter(output, fieldnames=FIELDS, lineterminator="\n") + writer.writeheader() + writer.writerows(sorted(rows, key=lambda row: (row["num_workers"], row["size"]))) + + manifest = { + "schema": "qwen235-v020-serving-allreduce-profile-v1", + "inputs": {str(path.resolve()): sha256(path) for path in input_paths}, + "output": str(output_path.resolve()), + "output_sha256": sha256(output_path), + "vllm_source_commit": VLLM_COMMIT, + "hardware": "NVIDIA H20", + "model": "Qwen3-235B-A22B-FP8", + "dtype": "bfloat16", + "hidden_dim": HIDDEN_DIM, + "collective_contract": "qwen235-serving-projected", + "tp_coverage": list(TP_SIZES), + "token_points": list(TOKEN_POINTS), + "rows": len(rows), + "observed_backends_by_tp": { + str(tp): sorted(backends) for tp, backends in sorted(backends_by_tp.items()) + }, + "frontier_target": "time_stats.all_reduce.median", + "unused_stat_policy": "repeat critical_path_median; std=0", + "raw_environments": environments, + } + manifest_path.parent.mkdir(parents=True, exist_ok=True) + manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") + return manifest + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--input", type=Path, action="append", required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--manifest", type=Path, required=True) + args = parser.parse_args() + manifest = materialize(args.input, args.output, args.manifest) + print(json.dumps(manifest, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/runs/frontier-fidelity-envelope-v1/plot_qwen235_collective_ablation_mock.py b/runs/frontier-fidelity-envelope-v1/plot_qwen235_collective_ablation_mock.py new file mode 100644 index 0000000..6b1a208 --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/plot_qwen235_collective_ablation_mock.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Render the preregistered Qwen235 collective-profile ablation layout.""" + +from pathlib import Path + +import matplotlib.pyplot as plt +import numpy as np + + +OUT = Path(__file__).with_name("qwen235-collective-ablation-mock.png") + + +def main() -> None: + configs = ["TP4 / EP1", "TP8 / EP8"] + real_tpot = np.array([21.0442, 27.9942]) + current_frontier_tpot = np.array([87.7735, 61.5878]) + + fig, (ax0, ax1) = plt.subplots(1, 2, figsize=(10.8, 4.4)) + x = np.arange(len(configs)) + width = 0.34 + ax0.bar( + x - width / 2, + real_tpot, + width, + color="#4c78a8", + label="Frozen real", + ) + ax0.bar( + x + width / 2, + current_frontier_tpot, + width, + color="#e45756", + label="A0 current Frontier", + ) + for index in x: + ax0.text( + index, + max(real_tpot[index], current_frontier_tpot[index]) + 4, + "A1 measured-profile\nresult: TBD", + ha="center", + va="bottom", + fontsize=8, + color="#222222", + ) + ax0.set_xticks(x, configs) + ax0.set_ylim(0, 112) + ax0.set_ylabel("Fixed-PD mean TPOT (ms)") + ax0.set_title("A. Does A1 reverse the simulated TP margin?") + ax0.legend(frameon=False, fontsize=8) + ax0.grid(axis="y", alpha=0.25) + + outcomes = ["A0 current", "A1 flips\nto TP4", "A1 stays\nat TP8"] + regret = [33.0, 0.0, 33.0] + colors = ["#e45756", "#54a24b", "#f2cf5b"] + bars = ax1.bar(outcomes, regret, color=colors, width=0.68) + ax1.axhline(10.0, color="#222222", linestyle="--", linewidth=1.2, label="10% gate") + ax1.bar_label(bars, labels=["33%", "0%", "33%"], padding=3, fontsize=9) + ax1.set_ylim(0, 42) + ax1.set_ylabel("Real TPOT selection regret (%)") + ax1.set_title("B. Preregistered decision outcomes") + ax1.legend(frameon=False, fontsize=8) + ax1.grid(axis="y", alpha=0.25) + + fig.suptitle( + "SCHEMATIC / preregistered layout — A1 values are not yet measured", + fontsize=11, + ) + fig.tight_layout() + fig.savefig(OUT, dpi=180, bbox_inches="tight") + print(OUT) + + +if __name__ == "__main__": + main() diff --git a/runs/frontier-fidelity-envelope-v1/qwen235-collective-ablation-mock.png b/runs/frontier-fidelity-envelope-v1/qwen235-collective-ablation-mock.png new file mode 100644 index 0000000..e6fb39b Binary files /dev/null and b/runs/frontier-fidelity-envelope-v1/qwen235-collective-ablation-mock.png differ diff --git a/runs/frontier-fidelity-envelope-v1/run_frontier_qwen235_v020_surface.py b/runs/frontier-fidelity-envelope-v1/run_frontier_qwen235_v020_surface.py index e275421..2d85d9c 100644 --- a/runs/frontier-fidelity-envelope-v1/run_frontier_qwen235_v020_surface.py +++ b/runs/frontier-fidelity-envelope-v1/run_frontier_qwen235_v020_surface.py @@ -158,6 +158,10 @@ def main() -> None: selected = [config for config in GRID if config.name in wanted] if {config.name for config in selected} != wanted: raise ValueError(f"unknown configs: {wanted - {config.name for config in selected}}") + allreduce_profile = Q30.validate_allreduce_profile( + args.allreduce_csv, {config.tp for config in selected if config.tp > 1} + ) + allreduce_csv_sha256 = BASE.sha256(args.allreduce_csv) builder = BASE.load_module( "qwen235_frontier_builder", args.replayserve_root / "tools/run_frontier_sweep.py" @@ -182,7 +186,11 @@ def main() -> None: result_path = run_dir / "result.json" if args.resume and result_path.is_file(): previous = json.loads(result_path.read_text()) - if previous.get("status") == "completed": + if ( + previous.get("status") == "completed" + and previous.get("allreduce_csv_sha256") + == allreduce_csv_sha256 + ): results.append(previous) continue run_dir.mkdir(parents=True, exist_ok=True) @@ -217,8 +225,16 @@ def main() -> None: "trace": Q30.trace_manifest_entry(trace), "elapsed_seconds": time.time() - started, "returncode": completed.returncode, + "allreduce_csv_sha256": allreduce_csv_sha256, } - if completed.returncode == 0: + fallback_evidence = Q30.collective_fallback_evidence(run_dir) + if completed.returncode == 0 and fallback_evidence: + result.update( + status="failed", + failure_class="collective_profile_fallback", + collective_fallback_evidence=fallback_evidence, + ) + elif completed.returncode == 0: metrics = Q30.find_request_metrics(run_dir) result.update(status="completed", metrics=Q30.score(metrics, trace["shapes"]), request_metrics_sha256=BASE.sha256(metrics)) else: @@ -233,6 +249,12 @@ def main() -> None: "frontier_commit": subprocess.check_output(["git", "-C", str(args.frontier_source), "rev-parse", "HEAD"], text=True).strip(), "profiles": {name: BASE.sha256(path) for name, path in paths.items()}, "runtime_contract_sha256": BASE.sha256(args.runtime_contract), + "collective": { + "backend": "vidur", + "allreduce_csv": str(args.allreduce_csv), + "allreduce_csv_sha256": allreduce_csv_sha256, + "measured_profile_validation": allreduce_profile, + }, "prefix_caching": args.prefix_caching, "results": results, } diff --git a/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py b/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py index a2c39a0..5590ed8 100644 --- a/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py +++ b/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py @@ -121,6 +121,91 @@ def percentile(values: list[float], fraction: float) -> float | None: return ordered[math.ceil(fraction * len(ordered)) - 1] +def validate_allreduce_profile( + path: Path | None, required_tps: set[int] +) -> dict[str, Any] | None: + """Fail before simulation when a measured collective profile misses a TP.""" + if not required_tps and path is None: + return None + if path is None: + raise ValueError( + f"measured all-reduce profile is required for TP coverage {sorted(required_tps)}" + ) + if not path.is_file(): + raise FileNotFoundError(path) + + required_fields = { + "time_stats.all_reduce.median", + "num_workers", + "size", + "collective", + } + rows_by_tp: dict[int, int] = {} + sizes_by_tp: dict[int, set[int]] = {} + seen: set[tuple[int, int]] = set() + with path.open(newline="") as handle: + reader = csv.DictReader(handle) + missing_fields = required_fields - set(reader.fieldnames or ()) + if missing_fields: + raise ValueError( + f"all-reduce profile schema missing fields: {sorted(missing_fields)}" + ) + for line_number, row in enumerate(reader, start=2): + if row["collective"] != "all_reduce": + raise ValueError( + f"unexpected collective at {path}:{line_number}: {row['collective']!r}" + ) + tp = int(row["num_workers"]) + size = int(row["size"]) + median = float(row["time_stats.all_reduce.median"]) + if size <= 0 or not math.isfinite(median) or median <= 0: + raise ValueError( + f"invalid measured all-reduce row at {path}:{line_number}" + ) + key = (tp, size) + if key in seen: + raise ValueError(f"duplicate measured all-reduce row: TP{tp}, size={size}") + seen.add(key) + rows_by_tp[tp] = rows_by_tp.get(tp, 0) + 1 + sizes_by_tp.setdefault(tp, set()).add(size) + + missing_tps = required_tps - set(rows_by_tp) + if missing_tps: + raise ValueError( + "measured all-reduce profile coverage missing " + f"TPs {sorted(missing_tps)}; available={sorted(rows_by_tp)}; " + "analytical fallback is forbidden" + ) + return { + "required_tp_coverage": sorted(required_tps), + "available_tp_coverage": sorted(rows_by_tp), + "rows_by_tp": {str(tp): rows_by_tp[tp] for tp in sorted(rows_by_tp)}, + "size_range_by_tp": { + str(tp): [min(sizes), max(sizes)] + for tp, sizes in sorted(sizes_by_tp.items()) + }, + } + + +def collective_fallback_evidence(run_dir: Path) -> list[dict[str, Any]]: + patterns = ( + "all-reduce profiling data is empty after filtering", + "analytical fallback", + "fallback to analytical", + ) + evidence = [] + for name in ("stdout.log", "stderr.log"): + path = run_dir / name + for line_number, line in enumerate( + path.read_text(errors="replace").splitlines(), start=1 + ): + if any(pattern in line.lower() for pattern in patterns): + evidence.append( + {"log": name, "line_number": line_number, "text": line.strip()} + ) + return evidence + + def manifest_offered_rate(path: Path) -> tuple[float | None, str | None]: manifest_path = path.with_name("manifest.json") if not manifest_path.is_file(): @@ -415,6 +500,14 @@ def main() -> None: selected = [config for config in BASE.GRID if config.name in wanted] if {config.name for config in selected} != wanted: raise ValueError(f"unknown configs: {wanted - {config.name for config in selected}}") + allreduce_profile = None + if args.cc_backend == "vidur": + allreduce_profile = validate_allreduce_profile( + args.allreduce_csv, {config.tp for config in selected if config.tp > 1} + ) + allreduce_csv_sha256 = ( + BASE.sha256(args.allreduce_csv) if args.allreduce_csv else None + ) paths = BASE.profile_paths(args.profile_root) coverage = BASE.validate_profile(paths) kernel_paths = None @@ -477,8 +570,14 @@ def main() -> None: run_dir = args.output_root / "runs" / config.name / trace["label"] result_path = run_dir / "result.json" if args.resume and result_path.is_file(): - loads.append(json.loads(result_path.read_text())) - continue + previous = json.loads(result_path.read_text()) + if ( + previous.get("status") == "completed" + and previous.get("allreduce_csv_sha256") + == allreduce_csv_sha256 + ): + loads.append(previous) + continue run_dir.mkdir(parents=True, exist_ok=True) command = builder.build_frontier_command( python_bin="/usr/bin/python3", @@ -520,6 +619,39 @@ def main() -> None: timeout=args.timeout_seconds, check=False, ) + fallback_evidence = ( + collective_fallback_evidence(run_dir) + if args.cc_backend == "vidur" + else [] + ) + if completed.returncode == 0 and fallback_evidence: + result = { + "status": "frontier_failed", + "failure_kind": "collective_profile_fallback", + "returncode": completed.returncode, + "config": { + "tp": config.tp, + "mns": config.mns, + "name": config.name, + }, + "trace_label": trace["label"], + "offered_request_rate": trace["offered_request_rate"], + "offered_request_rate_per_gpu": ( + trace["offered_request_rate"] / config.tp + ), + "request_count": trace["requests"], + "elapsed_seconds": time.time() - started, + "trace_sha256": trace["sha256"], + "collective_fallback_evidence": fallback_evidence, + "allreduce_csv_sha256": allreduce_csv_sha256, + } + BASE.write_json(result_path, result) + loads.append(result) + if not args.continue_on_failure: + raise RuntimeError( + f"measured collective fallback for {config.name}/{trace['label']}" + ) + continue if completed.returncode != 0: stderr_path = run_dir / "stderr.log" result = { @@ -542,6 +674,7 @@ def main() -> None: "elapsed_seconds": time.time() - started, "trace_sha256": trace["sha256"], "stderr_sha256": BASE.sha256(stderr_path), + "allreduce_csv_sha256": allreduce_csv_sha256, } BASE.write_json(result_path, result) loads.append(result) @@ -575,6 +708,7 @@ def main() -> None: "trace_sha256": trace["sha256"], "request_metrics_sha256": BASE.sha256(metrics), "score": score(metrics, trace["shapes"]), + "allreduce_csv_sha256": allreduce_csv_sha256, } BASE.write_json(result_path, result) loads.append(result) @@ -720,7 +854,8 @@ def main() -> None: "collective": { "backend": args.cc_backend, "allreduce_csv": str(args.allreduce_csv) if args.allreduce_csv else None, - "allreduce_csv_sha256": BASE.sha256(args.allreduce_csv) if args.allreduce_csv else None, + "allreduce_csv_sha256": allreduce_csv_sha256, + "measured_profile_validation": allreduce_profile, }, "traces": [trace_manifest_entry(trace) for trace in traces], "config_results": config_results, diff --git a/runs/frontier-fidelity-envelope-v1/run_qwen235_collective_profile_ablation.sh b/runs/frontier-fidelity-envelope-v1/run_qwen235_collective_profile_ablation.sh new file mode 100644 index 0000000..f38c5c7 --- /dev/null +++ b/runs/frontier-fidelity-envelope-v1/run_qwen235_collective_profile_ablation.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash + +set -euo pipefail + +BASE_CAMPAIGN_ROOT="${BASE_CAMPAIGN_ROOT:?BASE_CAMPAIGN_ROOT is required}" +OUTPUT_ROOT="${OUTPUT_ROOT:?OUTPUT_ROOT is required}" +ALLREDUCE_CSV="${ALLREDUCE_CSV:?ALLREDUCE_CSV is required}" +RUNNER_DIR="${RUNNER_DIR:-$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)}" +PROFILE_ROOT="${PROFILE_ROOT:-/home/admin/cpfs/wjh/aituner/qwen235-v020-profiles-20260719-r1/frozen}" +FRONTIER_SOURCE="${FRONTIER_SOURCE:-/home/admin/cpfs/wjh/aituner/frontier-q235-v020-5b953f5}" +REPLAYSERVE_ROOT="${REPLAYSERVE_ROOT:-/home/admin/cpfs/wjh/replayserve}" +VENV_ROOT="${VENV_ROOT:-/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1}" +PYTHON_DEPS="${PYTHON_DEPS:-${VENV_ROOT}/lib/python3.12/site-packages}" + +mkdir -p "${OUTPUT_ROOT}"/{provenance,sim,analysis} +exec > >(tee -a "${OUTPUT_ROOT}/controller.log") 2>&1 +echo "Q235_COLLECTIVE_ABLATION_LAUNCH_ECHO host=$(hostname) base=${BASE_CAMPAIGN_ROOT} simulator=Frontier-6e8e0d8 configs={TP4/EP1,TP8/EP8}xMNS{64,128} cases={fixed-pd,fixed-po,trace-pd,trace-po} real_rerun=false measured_collective=${ALLREDUCE_CSV} analytical_fallback=forbidden expected_wall=1-6h output=${OUTPUT_ROOT}" +date -u +START_UTC=%Y-%m-%dT%H:%M:%SZ +test "$(git -C "${FRONTIER_SOURCE}" rev-parse HEAD)" = "6e8e0d845bceff11b0b62cb29df3a1a93411fdd4" +sha256sum "${BASH_SOURCE[0]}" "${ALLREDUCE_CSV}" \ + "${PROFILE_ROOT}/manifest.json" \ + "${BASE_CAMPAIGN_ROOT}/provenance/runtime-contract.json" \ + > "${OUTPUT_ROOT}/provenance/input.sha256" +git -C "${RUNNER_DIR}" rev-parse HEAD > "${OUTPUT_ROOT}/provenance/aituner.commit" +git -C "${FRONTIER_SOURCE}" rev-parse HEAD > "${OUTPUT_ROOT}/provenance/frontier.commit" + +run_sim() { + local case_name="$1" prefix_flag="$2" + "${VENV_ROOT}/bin/python" "${RUNNER_DIR}/run_frontier_qwen235_v020_surface.py" \ + --frontier-source "${FRONTIER_SOURCE}" --replayserve-root "${REPLAYSERVE_ROOT}" \ + --profile-root "${PROFILE_ROOT}" --python-deps "${PYTHON_DEPS}" \ + --output-root "${OUTPUT_ROOT}/sim/${case_name}" \ + --runtime-contract "${BASE_CAMPAIGN_ROOT}/provenance/runtime-contract.json" \ + --trace-tp "4=${BASE_CAMPAIGN_ROOT}/traces/${case_name}/tp4/public/frontier.csv" \ + --trace-tp "8=${BASE_CAMPAIGN_ROOT}/traces/${case_name}/tp8/public/frontier.csv" \ + "${prefix_flag}" --allreduce-csv "${ALLREDUCE_CSV}" --resume +} + +run_sim fixed-pd --no-prefix-caching +run_sim fixed-po --no-prefix-caching +run_sim trace-pd --prefix-caching +run_sim trace-po --prefix-caching + +"${VENV_ROOT}/bin/python" "${RUNNER_DIR}/analyze_qwen235_v020_campaign.py" \ + --campaign-root "${BASE_CAMPAIGN_ROOT}" --sim-root "${OUTPUT_ROOT}" \ + --json-output "${OUTPUT_ROOT}/analysis/comparison.json" \ + --markdown-output "${OUTPUT_ROOT}/analysis/comparison.md" +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 Q235_COLLECTIVE_ABLATION_COMPLETE diff --git a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py index c48ba2b..0c4c298 100644 --- a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py +++ b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py @@ -26,6 +26,28 @@ def load(name: str): class FidelityEnvelopeTest(unittest.TestCase): + @staticmethod + def write_allreduce_csv(path: Path, tps: tuple[int, ...]) -> None: + fields = [ + "time_stats.all_reduce.median", + "num_workers", + "size", + "collective", + ] + with path.open("w", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fields, lineterminator="\n") + writer.writeheader() + for tp in tps: + for size in (8192, 65536): + writer.writerow( + { + "time_stats.all_reduce.median": 0.1 * tp, + "num_workers": tp, + "size": size, + "collective": "all_reduce", + } + ) + def test_block_identities_are_parent_sensitive_and_prefix_stable(self) -> None: module = load("prepare_exact_trace.py") prefix = list(range(32)) @@ -86,6 +108,103 @@ class FidelityEnvelopeTest(unittest.TestCase): str(4 + 8192 / 1000), ) + def test_allreduce_profile_requires_every_selected_tp(self) -> None: + module = load("run_frontier_qwen30_exact_trace_surface.py") + with tempfile.TemporaryDirectory() as temporary: + profile = Path(temporary) / "allreduce.csv" + self.write_allreduce_csv(profile, (4,)) + with self.assertRaisesRegex( + ValueError, "coverage missing TPs \\[8\\].*fallback is forbidden" + ): + module.validate_allreduce_profile(profile, {4, 8}) + + self.write_allreduce_csv(profile, (4, 8)) + validated = module.validate_allreduce_profile(profile, {4, 8}) + self.assertEqual(validated["required_tp_coverage"], [4, 8]) + self.assertEqual(validated["rows_by_tp"], {"4": 2, "8": 2}) + + def test_collective_analytical_fallback_is_runtime_failure(self) -> None: + module = load("run_frontier_qwen30_exact_trace_surface.py") + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + (root / "stdout.log").write_text( + "All-reduce profiling data is empty after filtering\n" + ) + (root / "stderr.log").write_text("unrelated warning\n") + evidence = module.collective_fallback_evidence(root) + self.assertEqual(len(evidence), 1) + self.assertEqual(evidence[0]["log"], "stdout.log") + + def test_materialize_qwen235_allreduce_requires_serving_contract(self) -> None: + module = load("materialize_qwen235_v020_allreduce.py") + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + inputs = [] + for tp in (4, 8): + path = root / f"tp{tp}.json" + rows = [] + for tokens in module.TOKEN_POINTS: + fusion_limit = {4: 2 * 1024 * 1024, 8: 512 * 1024}[tp] + payload_bytes = tokens * module.HIDDEN_DIM * 2 + median = tp / 10 + tokens / 10000 + rows.append( + { + "tensor_parallel_size": tp, + "num_tokens": tokens, + "hidden_dim": module.HIDDEN_DIM, + "payload_bytes": payload_bytes, + "selected_backend": ( + "flashinfer_trtllm_fused_projection" + if payload_bytes <= fusion_limit + else "pynccl" + ), + "real_fusion_limit_bytes": fusion_limit, + "trials": 3, + "repeats_per_trial": 20, + "per_trial_rank_samples_ms": [ + [[median] * 20 for _ in range(tp)] for _ in range(3) + ], + "critical_path_median_ms": median, + } + ) + path.write_text( + json.dumps( + { + "schema_version": "vllm020_allreduce_raw.v2", + "environment": { + "vllm_version": "0.20.0", + "vllm_source_commit": module.VLLM_COMMIT, + "gpu": "NVIDIA H20", + "model": "/models/Qwen3-235B-A22B-FP8", + "collective_contract": "qwen235-serving-projected", + "disable_custom_all_reduce": True, + "backend_env": { + "VLLM_ALLREDUCE_USE_FLASHINFER": "1", + "VLLM_ALLREDUCE_USE_SYMM_MEM": "1", + }, + }, + "rows": rows, + } + ) + ) + inputs.append(path) + output = root / "allreduce.csv" + manifest_path = root / "manifest.json" + manifest = module.materialize(inputs, output, manifest_path) + self.assertEqual(manifest["tp_coverage"], [4, 8]) + self.assertEqual(manifest["rows"], 2 * len(module.TOKEN_POINTS)) + self.assertEqual( + manifest["observed_backends_by_tp"], + { + "4": ["flashinfer_trtllm_fused_projection", "pynccl"], + "8": ["flashinfer_trtllm_fused_projection", "pynccl"], + }, + ) + with output.open(newline="") as handle: + rows = list(csv.DictReader(handle)) + self.assertEqual(rows[0]["num_workers"], "4") + self.assertEqual(rows[-1]["num_workers"], "8") + def test_batch_profile_retains_one_single_request_anchor_per_tp(self) -> None: runner = REPO_ROOT / "runs/frontier-phase-factorial-v0/run_frontier_qwen30_prefill_surface.py" spec = importlib.util.spec_from_file_location("qwen30_surface_runner", runner) diff --git a/runs/frontier-qwen30-vllm020-profile-v1/profile_vllm020_allreduce.py b/runs/frontier-qwen30-vllm020-profile-v1/profile_vllm020_allreduce.py index 09b3752..63309a1 100644 --- a/runs/frontier-qwen30-vllm020-profile-v1/profile_vllm020_allreduce.py +++ b/runs/frontier-qwen30-vllm020-profile-v1/profile_vllm020_allreduce.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Profile vLLM 0.20 TP all-reduce and assert FlashInfer TRTLLM dispatch.""" +"""Profile the vLLM 0.20 TP all-reduce path under a frozen runtime contract.""" from __future__ import annotations @@ -27,8 +27,14 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--output", type=Path, required=True) parser.add_argument("--num-tokens", type=int, nargs="+", default=[8]) parser.add_argument("--hidden-dim", type=int, default=2048) + parser.add_argument( + "--collective-contract", + choices=("flashinfer-trtllm", "qwen235-serving-projected"), + default="flashinfer-trtllm", + ) parser.add_argument("--warmup-iters", type=int, default=3) parser.add_argument("--repeats", type=int, default=10) + parser.add_argument("--trials", type=int, default=1) return parser.parse_args() @@ -55,10 +61,19 @@ def main() -> None: source_head = git_head(args.vllm_source) if source_head != VLLM_COMMIT: raise SystemExit(f"expected vLLM source {VLLM_COMMIT}, got {source_head}") + if args.repeats <= 0 or args.trials <= 0 or args.warmup_iters < 0: + raise SystemExit("warmup/repeats/trials must be non-negative/positive") if os.getenv("VLLM_ALLREDUCE_USE_FLASHINFER") != "1": raise SystemExit("VLLM_ALLREDUCE_USE_FLASHINFER must equal 1") if os.getenv("VLLM_FLASHINFER_ALLREDUCE_BACKEND") != "trtllm": raise SystemExit("VLLM_FLASHINFER_ALLREDUCE_BACKEND must equal trtllm") + expected_symm_mem = ( + "1" if args.collective_contract == "qwen235-serving-projected" else "0" + ) + if os.getenv("VLLM_ALLREDUCE_USE_SYMM_MEM") != expected_symm_mem: + raise SystemExit( + f"VLLM_ALLREDUCE_USE_SYMM_MEM must equal {expected_symm_mem}" + ) if "RANK" not in os.environ or "WORLD_SIZE" not in os.environ: raise SystemExit("launch with torchrun") @@ -67,6 +82,7 @@ def main() -> None: destroy_model_parallel, init_distributed_environment, initialize_model_parallel, + set_custom_all_reduce, tensor_model_parallel_all_reduce, ) from vllm.distributed.parallel_state import get_tp_group @@ -80,11 +96,15 @@ def main() -> None: rank = int(os.environ["RANK"]) local_rank = int(os.environ["LOCAL_RANK"]) world_size = int(os.environ["WORLD_SIZE"]) - if world_size not in (2, 4): - raise SystemExit(f"expected TP world size 2 or 4, got {world_size}") + if world_size not in (2, 4, 8): + raise SystemExit(f"expected TP world size 2, 4, or 8, got {world_size}") device = torch.device(f"cuda:{local_rank}") torch.accelerator.set_device_index(device) torch.set_default_device(device) + disable_custom_all_reduce = ( + args.collective_contract == "qwen235-serving-projected" + ) + set_custom_all_reduce(not disable_custom_all_reduce) init_distributed_environment() model_config = ModelConfig( model=str(args.model), @@ -95,11 +115,80 @@ def main() -> None: ) vllm_config = VllmConfig( model_config=model_config, - parallel_config=ParallelConfig(tensor_parallel_size=world_size) + parallel_config=ParallelConfig( + tensor_parallel_size=world_size, + disable_custom_all_reduce=disable_custom_all_reduce, + ) ) with set_current_vllm_config(vllm_config): initialize_model_parallel(tensor_model_parallel_size=world_size) + from vllm.distributed.device_communicators.all_reduce_utils import ( + should_nccl_symm_mem_allreduce, + ) + + def resolve_profile_operation(input_tensor: torch.Tensor): + communicator = get_tp_group().device_communicator + if args.collective_contract == "qwen235-serving-projected": + from vllm.compilation.passes.fusion.allreduce_rms_fusion import ( + FI_ALLREDUCE_FUSION_MAX_SIZE_MB, + ) + + major, minor = torch.cuda.get_device_capability(device) + capability = major * 10 + minor + try: + fusion_limit_bytes = int( + FI_ALLREDUCE_FUSION_MAX_SIZE_MB[capability][world_size] + * 1024 + * 1024 + ) + except KeyError as error: + raise SystemExit( + f"no FlashInfer fusion limit for SM{capability}, TP{world_size}" + ) from error + if input_tensor.nbytes <= fusion_limit_bytes: + fi_comm = communicator.fi_ar_comm + if fi_comm is None or fi_comm.disabled: + raise SystemExit("FlashInfer all-reduce communicator is unavailable") + if not fi_comm.should_use_fi_ar(input_tensor): + raise SystemExit("FlashInfer rejected a fusion-eligible payload") + return ( + "flashinfer_trtllm_fused_projection", + fi_comm.all_reduce, + fusion_limit_bytes, + ) + if ( + communicator.pynccl_comm is not None + and should_nccl_symm_mem_allreduce(world_size, input_tensor) + ): + return ( + "pynccl_symmetric_with_copy", + torch.ops.vllm.all_reduce_symmetric_with_copy, + fusion_limit_bytes, + ) + if ( + communicator.symm_mem_comm is not None + and communicator.symm_mem_comm.should_use_symm_mem(input_tensor) + ): + return ( + "torch_symmetric_memory", + communicator.symm_mem_comm.all_reduce, + fusion_limit_bytes, + ) + if communicator.pynccl_comm is not None: + return "pynccl", communicator.pynccl_comm.all_reduce, fusion_limit_bytes + return "torch_distributed", tensor_model_parallel_all_reduce, fusion_limit_bytes + + fi_comm = communicator.fi_ar_comm + selected_backend = ( + "flashinfer_trtllm" + if fi_comm is not None + and not fi_comm.disabled + and fi_comm.should_use_fi_ar(input_tensor) + else "non_flashinfer_fallback" + ) + return selected_backend, tensor_model_parallel_all_reduce, None + rows: list[dict[str, Any]] = [] expected_sum = world_size * (world_size + 1) / 2 try: @@ -110,8 +199,11 @@ def main() -> None: dtype=torch.bfloat16, device=device, ) + selected_backend, profile_operation, fusion_limit_bytes = ( + resolve_profile_operation(input_tensor) + ) for _ in range(args.warmup_iters): - output = tensor_model_parallel_all_reduce(input_tensor) + output = profile_operation(input_tensor) torch.accelerator.synchronize() torch.testing.assert_close( output, @@ -120,30 +212,55 @@ def main() -> None: rtol=0.0, ) - communicator = get_tp_group().device_communicator - fi_comm = communicator.fi_ar_comm - if fi_comm is None or fi_comm.disabled: + if args.collective_contract == "flashinfer-trtllm": + if selected_backend != "flashinfer_trtllm": + raise SystemExit( + f"expected FlashInfer TRTLLM, got {selected_backend} at " + f"TP={world_size}, tokens={num_tokens}" + ) + elif selected_backend not in { + "flashinfer_trtllm_fused_projection", + "pynccl_symmetric_with_copy", + "torch_symmetric_memory", + "pynccl", + }: raise SystemExit( - f"FlashInfer all-reduce was not selected at TP={world_size}, " - f"tokens={num_tokens}" + f"unexpected Qwen235 serving backend: {selected_backend}" ) - uses_flashinfer = fi_comm.should_use_fi_ar(input_tensor) - samples: list[float] = [] - for _ in range(args.repeats): - dist.barrier() - start = torch.cuda.Event(enable_timing=True) - end = torch.cuda.Event(enable_timing=True) - start.record() - output = tensor_model_parallel_all_reduce(input_tensor) - end.record() - torch.accelerator.synchronize() - samples.append(float(start.elapsed_time(end))) + all_trial_rank_samples: list[list[list[float]]] = [] + critical_path_samples: list[float] = [] + for _ in range(args.trials): + samples: list[float] = [] + for _ in range(args.repeats): + dist.barrier() + start = torch.cuda.Event(enable_timing=True) + end = torch.cuda.Event(enable_timing=True) + start.record() + output = profile_operation(input_tensor) + end.record() + torch.accelerator.synchronize() + samples.append(float(start.elapsed_time(end))) - gathered: list[list[float] | None] = [None] * world_size - dist.all_gather_object(gathered, samples) + gathered: list[list[float] | None] = [None] * world_size + dist.all_gather_object(gathered, samples) + if rank == 0: + rank_samples = [item for item in gathered if item is not None] + all_trial_rank_samples.append(rank_samples) + critical_path_samples.extend( + max(per_rank[index] for per_rank in rank_samples) + for index in range(args.repeats) + ) if rank == 0: - per_rank = [stats_ms(item) for item in gathered if item is not None] + flattened_by_rank = [ + [ + sample + for trial in all_trial_rank_samples + for sample in trial[rank_index] + ] + for rank_index in range(world_size) + ] + per_rank = [stats_ms(samples) for samples in flattened_by_rank] row = { "tensor_parallel_size": world_size, "num_tokens": num_tokens, @@ -153,12 +270,16 @@ def main() -> None: * torch.tensor([], dtype=torch.bfloat16).element_size(), "dtype": "bfloat16", "communicator": "vllm.tensor_model_parallel_all_reduce", - "selected_backend": ( - "flashinfer_trtllm" if uses_flashinfer else "nccl_fallback" - ), + "collective_contract": args.collective_contract, + "selected_backend": selected_backend, + "real_fusion_limit_bytes": fusion_limit_bytes, + "trials": args.trials, + "repeats_per_trial": args.repeats, + "per_trial_rank_samples_ms": all_trial_rank_samples, "per_rank_time_ms": per_rank, - "critical_path_median_ms": max( - rank_stats["median"] for rank_stats in per_rank + "critical_path_time_ms": stats_ms(critical_path_samples), + "critical_path_median_ms": statistics.median( + critical_path_samples ), } rows.append(row) @@ -169,7 +290,11 @@ def main() -> None: if rank == 0: payload = { - "schema_version": "qwen30_vllm020_allreduce_raw.v1", + "schema_version": ( + "qwen30_vllm020_allreduce_raw.v1" + if args.collective_contract == "flashinfer-trtllm" + else "vllm020_allreduce_raw.v2" + ), "environment": { "vllm_version": vllm.__version__, "vllm_source_commit": source_head, @@ -178,9 +303,18 @@ def main() -> None: "gpu": torch.cuda.get_device_name(device), "model": str(args.model), "backend_env": { - "VLLM_ALLREDUCE_USE_FLASHINFER": "1", - "VLLM_FLASHINFER_ALLREDUCE_BACKEND": "trtllm", + "VLLM_ALLREDUCE_USE_FLASHINFER": os.getenv( + "VLLM_ALLREDUCE_USE_FLASHINFER", "0" + ), + "VLLM_FLASHINFER_ALLREDUCE_BACKEND": os.getenv( + "VLLM_FLASHINFER_ALLREDUCE_BACKEND" + ), + "VLLM_ALLREDUCE_USE_SYMM_MEM": os.getenv( + "VLLM_ALLREDUCE_USE_SYMM_MEM", "1" + ), }, + "collective_contract": args.collective_contract, + "disable_custom_all_reduce": disable_custom_all_reduce, }, "rows": rows, } diff --git a/runs/frontier-qwen30-vllm020-profile-v1/run_allreduce_profile.sh b/runs/frontier-qwen30-vllm020-profile-v1/run_allreduce_profile.sh index 0cf4565..fef602d 100644 --- a/runs/frontier-qwen30-vllm020-profile-v1/run_allreduce_profile.sh +++ b/runs/frontier-qwen30-vllm020-profile-v1/run_allreduce_profile.sh @@ -2,10 +2,11 @@ set -euo pipefail -TP="${TP:?TP must be set to 2 or 4}" +TP="${TP:?TP must be set to 2, 4, or 8}" case "${TP}" in 2) HARD_GPU_CAP="0.40_H20h" ;; 4) HARD_GPU_CAP="0.80_H20h" ;; + 8) HARD_GPU_CAP="1.60_H20h" ;; *) echo "ERROR: invalid TP=${TP}" >&2; exit 1 ;; esac OUTPUT_ROOT="${OUTPUT_ROOT:?OUTPUT_ROOT must be set}" @@ -13,6 +14,10 @@ VENV_ROOT="${VENV_ROOT:-/tmp/wjh/venvs/vllm-0.20.0-cu129-profiler-v1}" VLLM_SOURCE="${VLLM_SOURCE:-/home/admin/cpfs/wjh/agentic-kv/third_party/vllm_v20_build}" MODEL="${MODEL:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B}" NUM_TOKENS="${NUM_TOKENS:-8}" +HIDDEN_DIM="${HIDDEN_DIM:-2048}" +COLLECTIVE_CONTRACT="${COLLECTIVE_CONTRACT:-flashinfer-trtllm}" +TRIALS="${TRIALS:-1}" +REPEATS="${REPEATS:-10}" mkdir -p "${OUTPUT_ROOT}/logs" "${OUTPUT_ROOT}/provenance" "${OUTPUT_ROOT}/raw" exec > >(tee -a "${OUTPUT_ROOT}/logs/profile.log") 2>&1 @@ -22,9 +27,20 @@ if [[ "${#GPU_IDS[@]}" -ne "${TP}" ]]; then exit 1 fi -export VLLM_ALLREDUCE_USE_FLASHINFER=1 -export VLLM_FLASHINFER_ALLREDUCE_BACKEND=trtllm -echo "PROFILE_LAUNCH_ECHO host=$(hostname) gpus=${CUDA_VISIBLE_DEVICES} model=${MODEL} runtime=vLLM-0.20.0+cu129 operator=tensor_model_parallel_all_reduce backend=FlashInfer-TRTLLM tp=${TP} tokens=${NUM_TOKENS} hidden=2048 dtype=BF16 output=${OUTPUT_ROOT} expected_wall=2-6m hard_wall=720s hard_gpu_cap=${HARD_GPU_CAP}" +case "${COLLECTIVE_CONTRACT}" in + flashinfer-trtllm) + export VLLM_ALLREDUCE_USE_FLASHINFER=1 + export VLLM_FLASHINFER_ALLREDUCE_BACKEND=trtllm + export VLLM_ALLREDUCE_USE_SYMM_MEM=0 + ;; + qwen235-serving-projected) + export VLLM_ALLREDUCE_USE_FLASHINFER=1 + export VLLM_FLASHINFER_ALLREDUCE_BACKEND=trtllm + export VLLM_ALLREDUCE_USE_SYMM_MEM=1 + ;; + *) echo "ERROR: invalid COLLECTIVE_CONTRACT=${COLLECTIVE_CONTRACT}" >&2; exit 1 ;; +esac +echo "PROFILE_LAUNCH_ECHO host=$(hostname) gpus=${CUDA_VISIBLE_DEVICES} model=${MODEL} runtime=vLLM-0.20.0+cu129 operator=tensor_model_parallel_all_reduce contract=${COLLECTIVE_CONTRACT} tp=${TP} tokens=${NUM_TOKENS} hidden=${HIDDEN_DIM} dtype=BF16 trials=${TRIALS} repeats=${REPEATS} output=${OUTPUT_ROOT} expected_wall=2-8m hard_wall=720s hard_gpu_cap=${HARD_GPU_CAP}" date -u +"START_UTC=%Y-%m-%dT%H:%M:%SZ" nvidia-smi --query-gpu=index,name,driver_version,memory.used,utilization.gpu --format=csv,noheader @@ -46,8 +62,11 @@ timeout --signal=TERM --kill-after=30s 600 \ --model "${MODEL}" \ --output "${OUTPUT_ROOT}/raw/allreduce-tp${TP}.json" \ --num-tokens "${TOKEN_ARGS[@]}" \ + --hidden-dim "${HIDDEN_DIM}" \ + --collective-contract "${COLLECTIVE_CONTRACT}" \ --warmup-iters 3 \ - --repeats 10 + --repeats "${REPEATS}" \ + --trials "${TRIALS}" test -s "${OUTPUT_ROOT}/raw/allreduce-tp${TP}.json" sha256sum "${OUTPUT_ROOT}/raw/allreduce-tp${TP}.json" \