feat(pro6000/GLM-5.3): 部署方案入库（TP8+EAGLE 生产标准 / 场景二高并发变体 / TP4PP2+IndexCache / E7b CAR 实验补丁 + deploy profiles）

- deploy_glm53_605.sh：配置 A 生产标准（60.5 在役，全 8 台 md5 fcd9109b 一致），
  支持 MEMFRAC/STEPS/TOPK/DRAFT/CHUNK/EXTRA/RESTART 调参；场景二高并发变体
  仅改 mrr32 + decode 图 bs{4,8,12,16}（KV 池 16.4 驻留上限）
- deploy_glm53_optimal(_s1).sh：配置 B TP4PP2+IndexCache（场景二最优 +41~79%；
  s1 形态唯一差异 radix-on）
- deploy_glm53_607_exp.sh + car_patch/：E7b custom-AR 1stage 补丁（cc1 decode
  每步 -14~-16%，实验性仅 cc1-2 验证；补丁文件与 60.7:/root/patches md5 一致）
- deploy/profiles/pro6000/：两个标准 profile（sskj.deploy 可消费），关键踩坑
  与场景二变体、parser 缺口均在注释中标注

README.md

@@ -60,6 +60,7 @@
 | `experiments/h200/dsv4_h200_vllm_tp{2,4,8}_custom_bench/` | 自定义压测客户端（多服务负载均衡） |
 | `experiments/h200/dsv4_h200_max_context_length/` `…/dsv4_h200_long_context_matrix/` `…/dsv4_h200_256k_4k_probe/` `…/dsv4_h200_64k_sglang_vs_vllm/` | 长上下文专项 |
 | `experiments/p800/dsv4_p800_sglang/` `…/dsv4_p800_max_context_length/` `…/dsv4_p800_long_context_matrix/` `…/dsv4_p800_256k_4k_probe/` | P800 baseline 与长上下文专项 |
+| `experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/` | RTX 6000D + SGLang，GLM-5.3-NVFP4 双场景压测标准（16k/512 高并发 + 128k/64k 90% 命中低并发；真实语料 run-id 窗口纪律 + 三套部署配置 A/B/CAR，基线与口径见目录 README；部署 profile 见 `deploy/profiles/pro6000/glm53_nvfp4_*`） |
 | `experiments/TEMPLATE/` | 老式固定场景实验模板 |
 
 ## 快速复现

deploy/profiles/pro6000/glm53_nvfp4_pro6000_sglang_tp4pp2_indexcache.env

@@ -0,0 +1,43 @@
+# GLM-5.3-NVFP4 SGLang TP=4 PP=2 + IndexCache deployment profile (single RTX 6000D node).
+# 场景二（16k 独立输入、cc8-32、prefill 主导）最优配置：吞吐 +41~79%、cc32 TTFT 减半
+# （对比 TP8+EAGLE 生产配置，2026-09-07 真实语料基线）。
+# 可执行部署脚本：experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/deploy_glm53_optimal.sh
+#
+# 关键点（实测踩坑，勿随意改）:
+#   - index_topk_freq=4 为模型原生默认（层轴索引复用省 75% indexer，无质量损失）
+#   - 禁投机解码：PP2 与投机框架不兼容（已实测）
+#   - 本文件为场景二形态（--disable-radix-cache，独立输入无前缀复用）；
+#     场景一 90% 命中对比须启用 radix（唯一差异：去掉 --disable-radix-cache，
+#     见 deploy_glm53_optimal_s1.sh）
+#   - mem 0.85：0.90 下 cuda graph 捕获余量不足会运行时 OOM（KV 池 569,600 = TP8 的 2.06 倍）
+#   - --disable-custom-all-reduce：TP4 over PCIe 用自定义 AR 在本栈无收益
+#   - 已知缺口：未带 --tool-call-parser glm47 --reasoning-parser glm45，质量门 6/7
+#     （tool call 失败纯属参数缺失，非模型问题）；上生产必须补 parser
+#   - 场景一（90% 命中低并发）该配置全面劣于 TP8+EAGLE（输出吞吐 −25~−65%），勿混用选型
+
+PLATFORM=pro6000
+EXPERIMENT=glm53_nvfp4_pro6000_sglang_tp4pp2_indexcache
+MODEL_NAME=GLM-5.3-NVFP4
+ENGINE=sglang
+RUNTIME=docker
+DOCKER_IMAGE=lmsysorg/sglang:nightly-dev-20260828-daf63171
+CONTAINER_NAME=glm53-nvfp4
+MODEL_PATH=/data/hf_models/GLM-5.3-NVFP4
+SERVED_MODEL_NAME=/data/hf_models/GLM-5.3-NVFP4
+PORT=30000
+HEALTH_PATH=/health
+HEALTH_WAIT_S=2400
+CONTAINER_PYTHON=python3
+
+TP=4
+DP=1
+
+DEVICE_VARS="CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7"
+ENGINE_ENV="PYTHONUNBUFFERED=1 HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1"
+
+DOCKER_FLAGS="--gpus all --ipc=host --shm-size 64g --entrypoint '' -p ${PORT}:${PORT}"
+VOLUMES="/data/hf_models:/data/hf_models:ro"
+
+BOOTSTRAP="exec python3 -m sglang.launch_server ${LAUNCH_ARGS}"
+
+LAUNCH_ARGS="--model-path ${MODEL_PATH} --tp-size ${TP} --pp-size 2 --mem-fraction-static 0.85 --max-running-requests 48 --disable-radix-cache --disable-shared-experts-fusion --moe-runner-backend flashinfer_cutlass --disable-flashinfer-autotune --disable-custom-all-reduce --chunked-prefill-size 16384 --json-model-override-args '{\"index_topk_freq\": 4}' --host 0.0.0.0 --port ${PORT}"

deploy/profiles/pro6000/glm53_nvfp4_pro6000_sglang_tp8eagle.env

@@ -0,0 +1,44 @@
+# GLM-5.3-NVFP4 SGLang TP=8 + EAGLE deployment profile (single RTX 6000D node, 8 GPUs).
+# 生产标准配置（174.1.60.5 团队自用在役，2026-09-07 定稿）。
+# 可执行部署脚本：experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/deploy_glm53_605.sh
+#
+# 关键点（实测踩坑，勿随意改）:
+#   - SM120 必需三项：--disable-shared-experts-fusion --moe-runner-backend
+#     flashinfer_cutlass --disable-flashinfer-autotune
+#   - EAGLE 4/1/5 为 cc1-2 均衡点（3/1/4、5/1/6 在 ±10% 噪声内；树式 topk>1 在 DSA 上不可用；
+#     EAGLE3 本 nightly 不支持 GLM-5.3）；draft 模型自动从主权重加载
+#   - KV fp8_e4m3 + hicache-ratio 3 是长上下文 90% 前缀命中场景的容量前提（KV 池 276,864 token）
+#   - chunk 8192：16384 已 A/B 证伪（MoE 工作区 OOM 风险，且 AR 带宽受限无次数红利）
+#   - 容器入口必须 python3 -m sglang.launch_server（镜像 entrypoint 无 shebang）
+#   - 重新部署前必须等显存排空（docker rm -f 异步滞留数分钟，不等会把新 KV 池压小）
+#   - 场景二高并发变体（16k 独立输入 cc8-32）：仅改 --max-running-requests 32 +
+#     --cuda-graph-max-bs-decode 16 --cuda-graph-bs-decode 4 8 12 16（KV 池 276,864 ÷
+#     16,896/req = 16.4 驻留上限，decode 批自然 ≤16，图覆盖到 bs16 即可，bs24/32 纯耗显存）
+#   - 上线前后过质量门 quality_gate_605.sh（PASS=7/7）
+
+PLATFORM=pro6000
+EXPERIMENT=glm53_nvfp4_pro6000_sglang_tp8eagle
+MODEL_NAME=GLM-5.3-NVFP4
+ENGINE=sglang
+RUNTIME=docker
+DOCKER_IMAGE=lmsysorg/sglang:nightly-dev-20260828-daf63171
+CONTAINER_NAME=glm53-nvfp4
+MODEL_PATH=/data/hf_models/GLM-5.3-NVFP4
+SERVED_MODEL_NAME=/data/hf_models/GLM-5.3-NVFP4
+PORT=30000
+HEALTH_PATH=/health
+HEALTH_WAIT_S=2400
+CONTAINER_PYTHON=python3
+
+TP=8
+DP=1
+
+DEVICE_VARS="CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7"
+ENGINE_ENV="PYTHONUNBUFFERED=1 HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1"
+
+DOCKER_FLAGS="--gpus all --ipc=host --shm-size 64g --entrypoint '' -p ${PORT}:${PORT}"
+VOLUMES="/data/hf_models:/data/hf_models:ro"
+
+BOOTSTRAP="exec python3 -m sglang.launch_server ${LAUNCH_ARGS}"
+
+LAUNCH_ARGS="--model-path ${MODEL_PATH} --tp-size ${TP} --mem-fraction-static 0.90 --max-running-requests 16 --chunked-prefill-size 8192 --max-prefill-tokens 16384 --disable-shared-experts-fusion --moe-runner-backend flashinfer_cutlass --disable-flashinfer-autotune --speculative-algorithm EAGLE --speculative-num-steps 4 --speculative-eagle-topk 1 --speculative-num-draft-tokens 5 --kv-cache-dtype fp8_e4m3 --enable-hierarchical-cache --hicache-ratio 3 --cuda-graph-max-bs-decode 8 --cuda-graph-bs-decode 1 2 3 4 6 8 --cuda-graph-max-bs-prefill 8 --context-length 270336 --reasoning-parser glm45 --tool-call-parser glm47 --host 0.0.0.0 --port ${PORT}"

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/README.md

@@ -0,0 +1,223 @@
+# GLM-5.3-NVFP4 (Pro6000D×8) SGLang 双场景压测标准
+
+- 日期：2026-09-04 ~ 09-08（09-07 真实语料修正版定稿）　机器：174.1.60.5（生产）/ 174.1.60.7（实验）
+- 硬件：8×RTX 6000D 96GB（SM120，PCIe Gen5，无 NVLink）　模型：`/data/hf_models/GLM-5.3-NVFP4`
+- 推理栈：SGLang `lmsysorg/sglang:nightly-dev-20260828-daf63171`（容器入口必须 `python3 -m sglang.launch_server`，镜像 entrypoint 无 shebang）
+- 完整报告：飞书 wiki https://gcn673xpgdxn.feishu.cn/wiki/NzbMwzmKviYidRkZYRrc8GYQnrf （revision 20，真实语料版）；本地副本 `D:\sskj\reports\GLM53_NVFP4_双场景压测报告_2026-09-07.md`
+
+## 目的
+
+把 09-04~09-08 多轮压测沉淀为**可复用的双场景测试标准**：任何同事按本目录脚本与纪律执行，
+即可得到与基线可比的数字。压测核心是自研 `bench_corpus.py`（真实 PG19 语料，input_ids 直发
+原生 `/generate`），配套固定的语料窗口分配（run-id）、命中率日志核验、质量门禁与重跑纪律。
+
+## 两场景定义
+
+| | 场景一：长上下文低并发 | 场景二：16k 独立输入高并发 |
+|---|---|---|
+| 输入 | 131072 (128k) / 65536 (64k) token | 16384 (16k) token，全独立 |
+| 输出 | 512 token（`ignore_eos`） | 同左 |
+| 并发 | cc 1/2/3/4 | cc 8/16/32（扩展点 cc40/64） |
+| 每点请求数 | 8 | cc8→16、cc16→32、cc32→32 |
+| 前缀命中 | `--shared-frac 0.9`（90% 共享前缀 + 10% 唯一后缀，页 16 对齐） | `--shared-frac 0`（命中率必须为 0） |
+| 实测命中率 | 128k 89.99% / 64k 89.94%（日志核验） | 0.0 |
+| 负载形态 | prefill 占 10%，decode（EAGLE）主导 | 100% 新 token，串行 chunk prefill 主导 |
+| 推荐配置 | **配置 A**（TP8+EAGLE 生产标准） | **配置 B**（TP4PP2+IndexCache） |
+
+标准命令（= `run_s1_corpus.sh` / `run_s2_corpus.sh` 的内容）：
+
+```bash
+# 场景一（8 点，run-id 9301-9308 与语料窗口一一绑定）
+python3 scripts/bench_corpus.py --corpus /root/corpus_ids.json \
+  --input-len 131072 --concurrency 1 --num-requests 8 --run-id 9301 \
+  --shared-frac 0.9 --output-len 512        # 128k cc1；其余点换 cc 与 run-id
+
+# 场景二（3 点，run-id 9311-9313）
+python3 scripts/bench_corpus.py --corpus /root/corpus_ids.json \
+  --input-len 16384 --concurrency 8 --num-requests 16 --run-id 9311 \
+  --shared-frac 0 --output-len 512
+```
+
+`bench_hit90.py` / `bench_report.sh` / `bench_matrix.sh` 是合成随机 id 的快速对照工具：
+随机 id 语义不通顺会使 EAGLE accept 虚高（3.46/3.88 vs 自然文本 2.29），**decode 类指标只作
+参考，正式口径一律用真实语料版**。
+
+## 测量口径（必须遵守，否则数字不可比）
+
+1. **输出吞吐 = 服务端 `meta_info.completion_tokens` 总和 / 墙钟**。EAGLE 投机解码下客户端
+   流式 chunk 计数会低估，一律以服务端计数为准（bench_corpus.py 已按此实现）。
+2. **TTFT = 客户端发出到首个流式响应**，并发 >1 时含排队；**TPOT = (首响应→末响应)/(completion_tokens−1)**，
+   并发下含交错停转，非纯 decode 步时。
+3. **命中率只能从容器日志 TP0 "Prefill batch" 行核验**：正则 `#new-token: (\d+).*?#cached-token: (\d+)`，
+   按运行窗口 `docker logs --since/--until`（RFC3339，结束 +2s 边距）截取，剔除每分钟 64-token
+   监控心跳（特征 `#new-token: 64`）。响应内 `cached_tokens` 字段恒为 0，不可用。
+4. **配置对比用 TPOT×accept（步时）**，抵消 accept 长度的内容运气（真实语料 accept 范围 2.13-2.92）；
+   吞吐/e2e 跨语料窗口噪声带 ±8%，同窗口背靠背 A/B（交替顺序）用于终局对比。
+5. 单点验收：`ok=nreq, failed=0, retractions=0`，SUMMARY 字段完整；命中率偏离预期、retractions>0、
+   OOM 或机器有外部负载（见下）→ 停下排查，数据作废。
+
+## 语料（corpus_ids.json，不入库 ~97MB）
+
+真实语料 = PG19 英文长书（GCS `deepmind-gutenberg`，路径 `train/{id}.txt`）用**服役模型自己的
+tokenizer** 分词得到的平坦 token 序列：`{"ids": [...21,296,780 tokens...], "books": [[start,end),...]}`。
+两种获取方式：
+
+```bash
+# 方式一（推荐）：从 60.7 拷贝现成语料（与基线逐字节一致）
+scp root@174.1.60.7:/root/corpus_ids.json /root/     # md5 a8f3909211adb92501cf007ab71d3b29
+
+# 方式二：重建（换 tokenizer / 语料损坏时）
+python3 scripts/fetch_pg19_v2.py                      # 下载最长 12 本书 ~85MB 到 /root/bench_corpus/books/
+docker cp /root/bench_corpus glm53-nvfp4:/tmp/        # 需一个运行中的 GLM-5.3-NVFP4 容器
+docker exec -i glm53-nvfp4 python3 - < scripts/tokenize_corpus.py \
+    > /root/corpus_ids.json 2> /root/tokenize_progress.log
+```
+
+注意：`bench_corpus.py` 的窗口布局是固定 token 偏移（见下表），语料长度必须 ≥ 2,071,040 + 余量；
+换模型/分词器重建后长度不同，超界会报错（此时用 `--pool-override` 指定窗口起点）。
+
+## run-id 窗口与重跑纪律
+
+| 窗口 | token 区间 | 绑定 run-id | 用途 |
+|---|---|---|---|
+| 共享前缀 | [0, 117968) / [0, 58976) | — | 128k / 64k 场景一的 90% 共享前缀 |
+| pool A | [131072, 550400) | 9301-9304 | 128k 唯一后缀（每点 8×13104） |
+| pool B | [550400, 760320) | 9305-9308 | 64k 唯一后缀（每点 8×6560） |
+| pool C | [760320, 2071040) | 9311-9313 | 16k 全独立输入（16/32/32×16384） |
+| spare | [2071040, 语料尾) | — | warmup 切片 + 重跑余量 |
+
+**纪律**：每个 run-id 绑定唯一不重叠窗口，**复用窗口会虚高命中率 → 数据无效**。重跑某点取新鲜
+文本时用 `--pool-override <起始偏移>`（run-id 退化为标签），惯例起点 = 2200000 起步、每轮 +700000。
+bench tag 每轮必换、绝不覆盖旧日志（日志目录默认 `/root/bench_logs`，可用 `LOG_DIR` 覆盖）。
+
+## 部署配置（三套正式 + 一个实验补丁）
+
+| 配置 | 脚本 | 要点 | 适用 |
+|---|---|---|---|
+| **A：TP8+EAGLE 生产标准** | `scripts/deploy_glm53_605.sh` | TP8；EAGLE 4/1/5；KV fp8_e4m3 + hicache 3；mem 0.90（池 276,864）；mrr 16；chunk 8192；max-prefill 16384；decode 图 bs{1,2,3,4,6,8}；ctx 270336；parser glm45/glm47 | 场景一；两类负载混跑的折中（60.5 在役） |
+| **A-s2：场景二高并发变体** | 同上脚本 + `EXTRA` | 仅两处不同：mrr 16→32、decode 图 bs{4,8,12,16}（池 276,864÷16,896=16.4 驻留上限，decode 批自然 ≤16，图覆盖到 bs16 即可；bs24/32 图纯耗显存且会运行时 OOM） | 场景二用配置 A 跑时 |
+| **B：TP4PP2+IndexCache** | `scripts/deploy_glm53_optimal.sh` | TP4 PP2；mem 0.85（池 569,600=2.06×）；mrr 48；chunk 16384；`index_topk_freq=4`；禁 radix；禁投机（PP2 与投机框架不兼容）；`--disable-custom-all-reduce` | **场景二最优**（吞吐 +41~79%） |
+| **B-s1：B 的场景一形态** | `scripts/deploy_glm53_optimal_s1.sh` | 与 B 唯一差异：启用 radix cache（90% 命中前提） | 场景一 A/B 对比时 |
+| **CAR 补丁（实验性）** | `scripts/deploy_glm53_607_exp.sh` + `scripts/car_patch/` | E7b custom-AR 1stage：小 AR（≤8MB，decode AR ≤1MB）走 one-shot 自定义核而非 NCCL RING_LL；cc1 decode 每步 −14~−16%，cc3-4 中性；**仅 cc1-2 验证过，未上生产** | 低并发场景一的实验优化 |
+
+SM120 必需三项（所有配置一致）：`--disable-shared-experts-fusion --moe-runner-backend
+flashinfer_cutlass --disable-flashinfer-autotune`。EAGLE draft 模型自动从主权重加载，无需额外参数。
+
+配置 B 已知缺口：脚本未带 `--tool-call-parser glm47 --reasoning-parser glm45`，质量门 6/7
+（tool call 失败纯属参数缺失，非模型问题）；**上生产必须补 parser**。
+
+deploy 脚本内置集群踩坑防护：`docker rm -f` 异步滞留 → 轮询等容器对象消失 + 等显存排空
+（<2000 MiB，最长 15 分钟）。**不等显存归零就重部署会把新容器 KV 池压小**。脚本支持环境变量
+调参（`MEMFRAC/STEPS/TOPK/DRAFT/CTXLEN/CHUNK/MAXPRE/EXTRA/RESTART`），用法见脚本头注释。
+
+## 基线数字（2026-09-07，真实语料，60.7 干净机器）
+
+**场景一 · 配置 A**（run-id 9301-9308；命中率 89.99%/89.94%，0 失败 0 回撤）：
+
+| 场景 | cc | 输入 tok/s | 输出 tok/s | 单请求 decode tok/s | TTFT mean/p50 s | TPOT ms | accept |
+|---|---|---|---|---|---|---|---|
+| 128k | 1 | 12,581 | 49.1 | 89.8 | 4.51 / 4.51 | 11.6 | 2.84 |
+| 128k | 2 | 14,622 | 57.1 | 48.7 | 5.68 / 4.54 | 22.8 | 2.45 |
+| 128k | 3 | 16,972 | 66.3 | 40.4 | 7.14 / 4.52 | 29.3 | 2.55 |
+| 128k | 4 | 17,625 | 68.9 | 28.8 | 9.12 / 7.64 | 37.8 | 2.32 |
+| 64k | 1 | 7,090 | 55.4 | 72.3 | 1.96 / 1.96 | 14.2 | 2.22 |
+| 64k | 2 | 10,236 | 80.0 | 50.1 | 2.41 / 2.01 | 20.3 | 2.13 |
+| 64k | 3 | 13,803 | 107.8 | 54.3 | 3.11 / 2.03 | 19.4 | 2.92 |
+| 64k | 4 | 13,927 | 108.8 | 38.2 | 4.11 / 3.76 | 27.0 | 2.31 |
+
+**场景二 · 配置 A（真实语料）vs 配置 B**（A：run-id 9311-9313；B 无投机解码，合成/真实语料等价）：
+
+| cc | A 输出 tok/s | B 输出 tok/s | B 增益 | A 输入 tok/s | B 输入 tok/s | A TTFT mean s | B TTFT mean s | A/B TPOT ms |
+|---|---|---|---|---|---|---|---|---|
+| 8 | 77.1 | 108.6 | +41% | 2,468 | 3,474 | 18.8 | 12.3 | 64.5 / 49.6 |
+| 16 | 95.3 | 146.2 | +53% | 3,050 | 4,678 | 20.1 | 18.9 | 123.5 / 72.7 |
+| 32 | 96.2 | 171.9 | +79% | 3,078 | 5,501 | 75.6 | 35.0 | 123.7 / 117.7 |
+
+**场景一 · A vs B 输出吞吐（tok/s）**：A 为真实语料，B 无投机解码口径等价——A 7/8 占优，
+唯一例外 128k cc4（B 75.9 vs A 68.9，B 略优 10%，cc4 聚合被四路交错 prefill 主导，换窗口复测稳定）：
+
+| 场景 | cc1 | cc2 | cc3 | cc4 |
+|---|---|---|---|---|
+| 128k A / B | 49.1 / 24.2 | 57.1 / 48.5 | 66.3 / 58.7 | **68.9 / 75.9** |
+| 64k A / B | 55.4 / 27.8 | 80.0 / 59.4 | 107.8 / 74.0 | 108.8 / 95.2 |
+
+**选型结论**：场景一用配置 A、场景二用配置 B；混跑负载用 A 折中。共同瓶颈是 8192-chunk 串行
+prefill 有效速率 ~2.8-2.9k tok/s（DSA 计算 ~65% + TP8 AR over PCIe ~30%），场景一 90% 命中把
+prefill 降到 10% 所以 TTFT 只有 2-4.6s；场景二该墙被 B（TP4 AR + IndexCache）推到 ~5.5k tok/s。
+
+CAR 补丁（E7b）单独基线：cc1 decode 每步 −14~−16%（步时口径），e2e −6~−14%；cc3-4 中性。
+
+## 质量门禁与运维纪律
+
+- **每次部署后、压测前**跑 `bash scripts/quality_gate_605.sh`（GSM8K×5 + 中文推理鸡兔同笼 + tool
+  call），**PASS=7 FAIL=0 方可用/可压**；实验结束后恢复生产配置并再次 7/7。
+- 压测机必须干净：60.2 曾因他人训练任务混跑出现 prefill 双峰，数据污染弃用——压测前确认
+  `nvidia-smi` 无外部进程。
+- 实验后收尾：恢复生产配置 + 质量门 7/7 + （可选）清空 GPU（`docker rm` 后等显存 0 MiB）。
+- 结果日志：`/root/bench_logs/r1_run*.log`（场景一）、`r2_run*.log`（场景二）；汇总表用
+  `python3 scripts/summarize_bench.py '/root/bench_logs/r1_run*.log'`，字段核对用 `bench_fields.py`。
+
+## 复现步骤
+
+```bash
+# 0) 语料（见上节）+ 脚本就位（本目录 = 服务器上的实验目录，或整体拷到 /root/）
+# 1) 部署生产配置 A 并过质量门
+bash scripts/deploy_glm53_605.sh && bash scripts/quality_gate_605.sh
+# 2) 场景一：8 点（约 25 分钟）
+bash scripts/run_s1_corpus.sh
+# 3) 场景二：换高并发变体 → 3 点 → 恢复生产
+MEMFRAC=0.90 EXTRA='--max-running-requests 32 --cuda-graph-max-bs-decode 16 --cuda-graph-bs-decode 4 8 12 16' \
+  RESTART=yes bash scripts/deploy_glm53_605.sh
+bash scripts/run_s2_corpus.sh
+RESTART=yes bash scripts/deploy_glm53_605.sh && bash scripts/quality_gate_605.sh
+# 4) 命中率核验（以 9301 为例，窗口 = 该点起止时间 +2s 边距）
+docker logs glm53-nvfp4 --since '<start RFC3339>' --until '<end RFC3339>' 2>&1 \
+  | grep 'Prefill batch' | grep -oP '#new-token: \d+.*?#cached-token: \d+'
+# 期望：场景一 89.99%/89.94%，场景二 0.0；心跳行（#new-token: 64）不计
+# 5) 配置 B（场景二专用）
+bash scripts/deploy_glm53_optimal.sh && bash scripts/quality_gate_605.sh   # 预期 6/7（parser 缺口）
+bash scripts/run_s2_corpus.sh
+```
+
+CAR 补丁（实验性）用法：先在**运行中的原生容器**上生成补丁 `bash scripts/car_patch/prep_car_patch.sh`
+（docker cp 原件 + sed 注入 + py_compile 校验，产物默认 `/root/patches/`），再用
+`bash scripts/deploy_glm53_607_exp.sh` 部署（默认 `CAR_PATCH=1` 自动注入后重启容器；`CAR_PATCH=0`
+部署原生）。验证：容器日志出现 `SSKJ_CAR_PATCH_ACTIVE ws=8 full_nvlink=False`。补丁随容器消亡，
+镜像不受影响。
+
+## 目录与文件对照（原版 md5 = 服务器实测，仓库版即标准）
+
+| 文件 | 60.7/集群原版 md5 | 仓库版处理 |
+|---|---|---|
+| `scripts/bench_corpus.py` | c6d92126ab518c5b1936841f869e4b35 | 逐字保留（压测核心） |
+| `scripts/bench_hit90.py` | fd1903969fa101ca43347e6d91ba05e0 | 逐字保留 |
+| `scripts/summarize_bench.py` | 1a265978538dec9eefc6cfb0e6fad281 | 逐字保留 |
+| `scripts/bench_fields.py` | aba7adcd21448b8e21ba8974e5326cf6 | 逐字保留 |
+| `scripts/fetch_pg19_v2.py` | 3dccd45d7fa3b556094d55f3141d7249 | 逐字保留 |
+| `scripts/tokenize_corpus.py` | 266097bfa08d02b9d7d7e87e77d77b97 | 逐字保留 |
+| `scripts/quality_gate_605.sh` | 0bada531dda12cb982d9b2b36695b809 | 逐字保留 |
+| `scripts/deploy_glm53_605.sh` | fcd9109b2121c6aea2e913ce612fa63f（全 8 台一致） | 逐字保留 |
+| `scripts/deploy_glm53_optimal.sh` | 22685f56b28a54216accbe479fba86b4 | 逐字保留 |
+| `scripts/deploy_glm53_optimal_s1.sh` | 6f1ee9ae5d404914f914320dd218bea4 | 逐字保留 |
+| `scripts/deploy_glm53_607_exp.sh` | 21db641f1d997e26fcf1ddc97163b87e | 逐字保留 |
+| `scripts/car_patch/custom_all_reduce.py` | a8fc9a504c041acc40831a848b4985d8 | 逐字保留（60.7:/root/patches） |
+| `scripts/car_patch/custom_all_reduce_utils.py` | 65a4d22bd87ae343e7d68c4879c14f98 | 逐字保留 |
+| `scripts/car_patch/prep_car_patch.sh` | 9578fe4b1f2443ccecd2cc3e0b660fb0 | 逐字保留 |
+| `scripts/run_s1_corpus.sh` | 5b2534f4c2e2f5760c453abb00dc32a2 | 仅路径适配¹ |
+| `scripts/run_s2_corpus.sh` | 8965a71edef6b1cf406496f0362a5d05 | 仅路径适配¹ |
+| `scripts/bench_report.sh` | 282f8cdf5873d3a3eb8a950a1564acae | 仅路径适配¹ |
+| `scripts/bench_matrix.sh` | b961a56bc7c8f26c519fdc7f5f627ba1 | 仅路径适配¹ |
+
+¹ 仓库规范（README 注意事项）要求脚本不写绝对路径：bench 脚本改按脚本所在目录解析，
+语料/日志目录可用 `CORPUS`/`LOG_DIR` 环境变量覆盖，默认仍为 `/root`（服务器原布局，行为不变）。
+
+语料 `corpus_ids.json`（97MB）不入库；历史原始日志不入库（`.gitignore` 排除 `*.log`），
+基线以本 README 表格 + 飞书报告为准。
+
+## 关联
+
+- 部署 profile（`python -m sskj.deploy` 消费）：`deploy/profiles/pro6000/glm53_nvfp4_pro6000_sglang_tp8eagle.env`、
+  `deploy/profiles/pro6000/glm53_nvfp4_pro6000_sglang_tp4pp2_indexcache.env`
+- 前序实验：`experiments/pro6000/glm53_nvfp4_pro6000d_sglang_tp4pp2_profile/`（profile 与配置级证伪）、
+  `experiments/pro6000/glm53_nvfp4_pro6000d_sglang_ppmtp_deepdive/`（PP+MTP 深挖，均在 hzy 分支）
+- 场景一深度优化报告（E7b CAR 补丁出处）：`D:\sskj\reports\GLM53_NVFP4_60.7_场景一优化实验报告_2026-09-07.md`

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/config.env

@@ -0,0 +1,26 @@
+# GLM-5.3-NVFP4 双场景压测标准实验配置（174.1.60.5/60.7, 定稿 2026-09-07 真实语料版）
+HOST=174.1.60.5          # 生产在役机（实验用 60.7，压测前须确认机器干净）
+MODEL=/data/hf_models/GLM-5.3-NVFP4
+IMAGE=lmsysorg/sglang:nightly-dev-20260828-daf63171
+CONTAINER=glm53-nvfp4
+PORT=30000
+
+# ---- 部署配置（详见 README「部署配置」，脚本支持环境变量调参）----
+DEPLOY_A="bash scripts/deploy_glm53_605.sh"                 # 生产标准：TP8 EAGLE 4/1/5 fp8KV hicache3 mem0.90 mrr16 chunk8192
+DEPLOY_A_S2="MEMFRAC=0.90 EXTRA='--max-running-requests 32 --cuda-graph-max-bs-decode 16 --cuda-graph-bs-decode 4 8 12 16' RESTART=yes bash scripts/deploy_glm53_605.sh"
+DEPLOY_B="bash scripts/deploy_glm53_optimal.sh"             # TP4PP2+IndexCache freq4 mem0.85 mrr48 chunk16384 radix-off 无投机
+DEPLOY_B_S1="bash scripts/deploy_glm53_optimal_s1.sh"       # B 的场景一形态：唯一差异 radix-on
+DEPLOY_CAR="bash scripts/deploy_glm53_607_exp.sh"           # A + CAR 1stage 补丁注入（CAR_PATCH=0 关闭）— 实验性，仅 cc1-2 验证
+
+# ---- 场景矩阵（run-id 与语料窗口一一绑定，严禁复用）----
+SCENARIO_1="{128k,64k} x cc{1,2,3,4}, nreq=8, shared-frac 0.9, run-id 9301-9308（runner: run_s1_corpus.sh）"
+SCENARIO_2="16k 全独立, cc{8,16,32}, nreq{16,32,32}, shared-frac 0, run-id 9311-9313（runner: run_s2_corpus.sh）"
+SCENARIO_EXT="cc40/cc64 扩展点：nreq=cc，须用 --pool-override 取新鲜窗口（≥2200000，每轮 +700000）"
+
+# ---- 语料 ----
+CORPUS=/root/corpus_ids.json        # PG19 21,296,780 token；60.7 现成 md5 a8f3909211adb92501cf007ab71d3b29；重建见 README
+CORPUS_BUILD="python3 scripts/fetch_pg19_v2.py && docker exec -i glm53-nvfp4 python3 - < scripts/tokenize_corpus.py > /root/corpus_ids.json"
+
+# ---- 口径与门禁（摘要，全文见 README「测量口径」）----
+METRICS="输出吞吐=服务端 completion_tokens/墙钟；命中率仅从 TP0 Prefill 日志核验；配置对比用 TPOT×accept；噪声带 ±8%"
+QUALITY_GATE="bash scripts/quality_gate_605.sh → PASS=7 FAIL=0（配置 B 预期 6/7：缺 --tool-call-parser glm47）"

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/bench_corpus.py

@@ -0,0 +1,298 @@
+#!/usr/bin/env python3
+"""Real-corpus (PG19) benchmark for sglang GLM-5.3-NVFP4.
+
+Same methodology as bench_hit90.py (input_ids direct to /generate, temp 0,
+ignore_eos, stream, server-side completion_tokens counting, hit-rate verified
+from scheduler logs), but prompts are token slices of REAL book text tokenized
+with the served model's own tokenizer, replacing random ids.
+
+Corpus file: JSON {"ids": [flat token ids], "books": [[start, end), ...]}
+
+Fixed token-offset layout into the flat id array:
+  [0, 117968)      shared prefix for 128k points (90% of 131072)
+  [0, 58976)       shared prefix for 64k points (same region, shorter cut)
+  pool A [131072, +4*8*13104)   128k unique suffixes, run-ids 9301-9304
+  pool B [550400, +4*8*6560)    64k unique suffixes, run-ids 9305-9308
+  pool C [760320, 262144+524288+524288)  16k fully-unique prompts, run-ids 9311-9313
+  spare [2071040, end)          warmup slices / re-run margin
+
+Each run-id maps to one non-overlapping window (one window = one bench point);
+re-running a point with fresh text = bump --pool-override past the spare base.
+
+Usage:
+  python3 bench_corpus.py --corpus /root/corpus_ids.json --input-len 131072 \
+      --concurrency 1 --num-requests 8 --run-id 9301 --shared-frac 0.9
+"""
+import argparse
+import datetime
+import json
+import re
+import statistics
+import subprocess
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+import requests
+
+OUTPUT_LEN_DEFAULT = 512
+CORPUS_DEFAULT = "/root/corpus_ids.json"
+
+# fixed pool layout (see docstring)
+S1_128K_RID0, S1_64K_RID0, S2_RID0 = 9301, 9305, 9311
+POOL_A_BASE, POOL_A_PER = 131072, 8 * 13104          # 128k suffix windows
+POOL_B_BASE = POOL_A_BASE + 4 * POOL_A_PER           # 550400
+POOL_B_PER = 8 * 6560                                 # 64k suffix windows
+POOL_C_BASE = POOL_B_BASE + 4 * POOL_B_PER           # 760320
+POOL_C_SIZES = [16 * 16384, 32 * 16384, 32 * 16384]  # cc8 / cc16 / cc32
+SPARE_BASE = POOL_C_BASE + sum(POOL_C_SIZES)         # 2071040
+
+sess = requests.Session()
+sess.trust_env = False  # bypass any proxy env on the host
+
+
+def split_lens(input_len, shared_frac):
+    # unique suffix = (1 - shared_frac) of the prompt, page-16 aligned
+    # (128k @0.9 -> 13104 unique; 16k @0.0 -> fully unique prompts)
+    unique = round(input_len * (1.0 - shared_frac) / 16) * 16
+    return input_len - unique, unique
+
+
+def pool_start_for(input_len, shared_frac, run_id, override):
+    if override is not None:
+        return override
+    if shared_frac > 0:
+        if input_len == 131072:
+            idx = run_id - S1_128K_RID0
+            if not 0 <= idx < 4:
+                sys_exit_bad_runid(run_id, "128k points use run-ids 9301-9304")
+            return POOL_A_BASE + idx * POOL_A_PER
+        if input_len == 65536:
+            idx = run_id - S1_64K_RID0
+            if not 0 <= idx < 4:
+                sys_exit_bad_runid(run_id, "64k points use run-ids 9305-9308")
+            return POOL_B_BASE + idx * POOL_B_PER
+        sys_exit_bad_runid(run_id, "shared-frac>0 supports 131072/65536 only")
+    idx = run_id - S2_RID0
+    if not 0 <= idx < 3:
+        sys_exit_bad_runid(run_id, "16k unique points use run-ids 9311-9313")
+    return POOL_C_BASE + sum(POOL_C_SIZES[:idx])
+
+
+def sys_exit_bad_runid(run_id, msg):
+    raise SystemExit(f"[pool] run-id {run_id} outside expected set: {msg}")
+
+
+def build_prompts(ids, shared_len, unique_len, num_requests, pool_start):
+    if shared_len:
+        shared = ids[0:shared_len]
+    else:
+        shared = []
+    end = pool_start + num_requests * unique_len
+    if end > len(ids):
+        raise SystemExit(
+            f"[pool] window [{pool_start}, {end}) exceeds corpus ({len(ids)} ids); "
+            f"use --pool-override or a larger corpus")
+    prompts = []
+    for i in range(num_requests):
+        s = pool_start + i * unique_len
+        prompts.append(shared + ids[s:s + unique_len])
+    return shared, prompts, (pool_start, end)
+
+
+def warmup(url, ids, shared_len):
+    # primes the radix cache with the shared prefix (same role as in bench_hit90);
+    # warm slice comes from the spare region so it never collides with a pool window
+    if len(ids) >= SPARE_BASE + 64:
+        warm_slice = ids[SPARE_BASE:SPARE_BASE + 64]
+    else:
+        warm_slice = ids[-64:]
+    payload = {
+        "input_ids": ids[0:shared_len] + warm_slice if shared_len else warm_slice,
+        "sampling_params": {"max_new_tokens": 8, "temperature": 0.0, "ignore_eos": True},
+    }
+    t0 = time.perf_counter()
+    r = sess.post(url, json=payload, timeout=1800)
+    dt = time.perf_counter() - t0
+    print(f"[warmup] http={r.status_code} wall={dt:.2f}s", flush=True)
+
+
+def bench_one(url, prompt, output_len, idx, results):
+    payload = {
+        "input_ids": prompt,
+        "sampling_params": {"max_new_tokens": output_len, "temperature": 0.0, "ignore_eos": True},
+        "stream": True,
+    }
+    rec = {"idx": idx}
+    t0 = time.perf_counter()
+    first = last = None
+    first_ct = None
+    final_meta = None
+    max_ct = 0
+    try:
+        with sess.post(url, json=payload, stream=True, timeout=3600) as resp:
+            for raw in resp.iter_lines():
+                if not raw or not raw.startswith(b"data:"):
+                    continue
+                body = raw[5:].strip()
+                if body == b"[DONE]":
+                    continue
+                now = time.perf_counter()
+                try:
+                    d = json.loads(body)
+                except Exception:
+                    continue
+                mi = d.get("meta_info") or {}
+                ct = mi.get("completion_tokens") or 0
+                if ct:
+                    max_ct = max(max_ct, ct)
+                if first is None:
+                    first = now
+                    first_ct = ct
+                last = now
+                if mi.get("finish_reason"):
+                    final_meta = mi
+        t_end = time.perf_counter()
+        n_out = max_ct or ((final_meta or {}).get("completion_tokens") or 0)
+        decode_span = (last - first) if (first and last and last > first) else 0.0
+        rec.update(
+            ok=n_out > 0,
+            ttft=(first - t0) if first else None,
+            e2e=t_end - t0,
+            n_out=n_out,
+            first_chunk_tokens=first_ct,
+            decode_span=decode_span,
+            tpot=(decode_span / (n_out - 1)) if n_out > 1 else None,
+            per_req_decode_tok_s=(n_out / decode_span) if decode_span > 0 else None,
+            retractions=(final_meta or {}).get("num_retractions"),
+            spec_accept_len=(final_meta or {}).get("spec_accept_length"),
+        )
+    except Exception as e:
+        rec.update(ok=False, error=repr(e))
+    results[idx] = rec
+
+
+def verify_hit_rate(container, t_start, t_end):
+    def rfc3339(epoch):
+        return (datetime.datetime.fromtimestamp(epoch, tz=datetime.timezone.utc)
+                .isoformat().replace("+00:00", "Z"))
+
+    try:
+        # No margin before t_start: warmup's prefill lines end strictly before it,
+        # and catching them would deflate the measured hit rate.
+        p = subprocess.run(
+            ["docker", "logs", container, "--since", rfc3339(t_start), "--until", rfc3339(t_end + 2)],
+            capture_output=True, text=True, timeout=120)
+        text = p.stdout + p.stderr
+    except Exception as e:
+        return {"error": repr(e)}
+    pat = re.compile(r"#new-token: (\d+).*?#cached-token: (\d+)")
+    n_batches = new_tok = cached_tok = 0
+    for line in text.splitlines():
+        if "TP0]" not in line or "Prefill batch" not in line:
+            continue
+        m = pat.search(line)
+        if m:
+            n_batches += 1
+            new_tok += int(m.group(1))
+            cached_tok += int(m.group(2))
+    total = new_tok + cached_tok
+    return {
+        "prefill_batches": n_batches,
+        "new_tokens": new_tok,
+        "cached_tokens": cached_tok,
+        "hit_rate": round(cached_tok / total, 4) if total else None,
+    }
+
+
+def stats(vals):
+    vals = [v for v in vals if v is not None]
+    if not vals:
+        return {"mean": None, "p50": None, "max": None, "min": None}
+    s = sorted(vals)
+    return {
+        "mean": round(statistics.fmean(vals), 4),
+        "p50": round(s[len(s) // 2], 4),
+        "max": round(s[-1], 4),
+        "min": round(s[0], 4),
+    }
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--concurrency", type=int, required=True)
+    ap.add_argument("--num-requests", type=int, required=True)
+    ap.add_argument("--run-id", type=int, required=True)
+    ap.add_argument("--input-len", type=int, required=True, help="16384 / 65536 / 131072")
+    ap.add_argument("--output-len", type=int, default=OUTPUT_LEN_DEFAULT)
+    ap.add_argument("--shared-frac", type=float, default=0.9)
+    ap.add_argument("--corpus", default=CORPUS_DEFAULT)
+    ap.add_argument("--pool-override", type=int, default=None,
+                    help="explicit corpus offset for the unique-suffix window (re-runs)")
+    ap.add_argument("--url", default="http://127.0.0.1:30000/generate")
+    ap.add_argument("--container", default="glm53-nvfp4")
+    args = ap.parse_args()
+
+    with open(args.corpus) as f:
+        corpus = json.load(f)
+    ids = corpus["ids"]
+
+    shared_len, unique_len = split_lens(args.input_len, args.shared_frac)
+    pool_start = pool_start_for(args.input_len, args.shared_frac, args.run_id, args.pool_override)
+    shared, prompts, window = build_prompts(ids, shared_len, unique_len, args.num_requests, pool_start)
+    print(f"[pool] window={window} shared_len={shared_len} unique_len={unique_len} "
+          f"corpus_total={len(ids)}", flush=True)
+    warmup(args.url, ids, shared_len)
+
+    results = {}
+    t_start = time.time()
+    t0 = time.perf_counter()
+    with ThreadPoolExecutor(max_workers=args.concurrency) as ex:
+        futs = [ex.submit(bench_one, args.url, p, args.output_len, i, results)
+                for i, p in enumerate(prompts)]
+        for f in futs:
+            f.result()
+    wall = time.perf_counter() - t0
+    t_end = time.time()
+
+    hit = verify_hit_rate(args.container, t_start, t_end)
+
+    ok = [r for r in results.values() if r.get("ok")]
+    n_out_total = sum(r["n_out"] for r in ok)
+    out_tps = [r["n_out"] / r["e2e"] for r in ok if r.get("e2e")]
+    ttft = stats([r.get("ttft") for r in ok])
+    tpot = stats([r.get("tpot") for r in ok])
+    e2e = stats([r.get("e2e") for r in ok])
+    dec = stats([r.get("per_req_decode_tok_s") for r in ok])
+    spec = [v for v in (r.get("spec_accept_len") for r in ok) if v is not None]
+    retr = sum(r.get("retractions") or 0 for r in ok)
+
+    summary = {
+        "concurrency": args.concurrency,
+        "num_requests": args.num_requests,
+        "run_id": args.run_id,
+        "corpus_window": {"start": window[0], "end": window[1]},
+        "ok": len(ok),
+        "failed": args.num_requests - len(ok),
+        "wall_s": round(wall, 2),
+        "input_len": args.input_len,
+        "shared_len": shared_len,
+        "unique_len": unique_len,
+        "output_len": args.output_len,
+        "output_tokens_total": n_out_total,
+        "output_throughput_tok_s": round(n_out_total / wall, 2) if wall else None,
+        "input_throughput_tok_s": round(args.input_len * len(ok) / wall, 2) if wall else None,
+        "ttft_s": ttft,
+        "tpot_s": tpot,
+        "e2e_s": e2e,
+        "per_req_out_tok_s_e2e": stats(out_tps),
+        "per_req_decode_tok_s": dec,
+        "spec_accept_length_mean": round(statistics.fmean(spec), 3) if spec else None,
+        "retractions_total": retr,
+        "cache_hit_from_logs": hit,
+    }
+    print("\n===== SUMMARY =====")
+    print(json.dumps(summary, indent=2), flush=True)
+
+
+if __name__ == "__main__":
+    main()

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/bench_fields.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python3
+"""Print full stats fields from bench logs to pin down conventions."""
+import glob
+import json
+import sys
+
+pat = sys.argv[1]
+for f in sorted(glob.glob(pat)):
+    txt = open(f).read()
+    if "===== SUMMARY =====" not in txt:
+        continue
+    d = json.loads(txt.split("===== SUMMARY =====")[1])
+    print(
+        f"{f.split('/')[-1]} run={d['run_id']} il={d['input_len']} cc={d['concurrency']} "
+        f"dec_mean={d['per_req_decode_tok_s']['mean']} dec_p50={d['per_req_decode_tok_s']['p50']} "
+        f"tpot_mean={d['tpot_s']['mean']} tpot_p50={d['tpot_s']['p50']} "
+        f"ttft_mean={d['ttft_s']['mean']} ttft_p50={d['ttft_s']['p50']} ttft_max={d['ttft_s']['max']} "
+        f"e2e_mean={d['e2e_s']['mean']} accept={d['spec_accept_length_mean']}"
+    )

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/bench_hit90.py

@@ -0,0 +1,236 @@
+#!/usr/bin/env python3
+"""Parameterized cache-hit benchmark for sglang GLM-5.3-NVFP4.
+
+Scenario: N-token prompts (default 128k), --shared-frac of each prompt is a
+shared prefix (radix/hicache hit, default 0.9), the rest is a unique
+per-request suffix. --shared-frac 0 = fully unique prompts (no prefix reuse).
+Measures TTFT / TPOT / output throughput at a given client concurrency.
+Hit rate is verified from scheduler logs (docker) for the exact run window
+("#new-token / #cached-token" per prefill batch).
+
+Usage:
+  python3 bench_hit90.py --concurrency 1 --num-requests 4 --run-id 101
+  python3 bench_hit90.py --input-len 65536 --concurrency 2 --num-requests 4 --run-id 102
+  python3 bench_hit90.py --input-len 16384 --shared-frac 0 --concurrency 16 --num-requests 32 --run-id 103
+"""
+import argparse
+import datetime
+import json
+import random
+import re
+import statistics
+import subprocess
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+import requests
+
+SEED = 20260904
+OUTPUT_LEN_DEFAULT = 512
+VOCAB_LO, VOCAB_HI = 1000, 100000
+
+sess = requests.Session()
+sess.trust_env = False  # bypass any proxy env on the host
+
+
+def split_lens(input_len, shared_frac):
+    # unique suffix = (1 - shared_frac) of the prompt, page-16 aligned
+    # (128k @0.9 -> 13104 unique; 16k @0.0 -> fully unique prompts)
+    unique = round(input_len * (1.0 - shared_frac) / 16) * 16
+    return input_len - unique, unique
+
+
+def build_prompts(shared_len, unique_len, num_requests, run_id):
+    rng = random.Random(SEED)
+    shared = [rng.randint(VOCAB_LO, VOCAB_HI) for _ in range(shared_len)]
+    prompts = []
+    for i in range(num_requests):
+        # Suffix seed includes run_id so each run generates fresh suffixes;
+        # reusing suffixes from an earlier run would inflate the cache hit rate.
+        r = random.Random(SEED * 100003 + run_id * 1000003 + i)
+        prompts.append(shared + [r.randint(VOCAB_LO, VOCAB_HI) for _ in range(unique_len)])
+    return shared, prompts
+
+
+def warmup(url, shared):
+    suffix = [random.Random(777).randint(VOCAB_LO, VOCAB_HI) for _ in range(64)]
+    payload = {
+        "input_ids": shared + suffix,
+        "sampling_params": {"max_new_tokens": 8, "temperature": 0.0, "ignore_eos": True},
+    }
+    t0 = time.perf_counter()
+    r = sess.post(url, json=payload, timeout=1800)
+    dt = time.perf_counter() - t0
+    print(f"[warmup] http={r.status_code} wall={dt:.2f}s", flush=True)
+
+
+def bench_one(url, prompt, output_len, idx, results):
+    payload = {
+        "input_ids": prompt,
+        "sampling_params": {"max_new_tokens": output_len, "temperature": 0.0, "ignore_eos": True},
+        "stream": True,
+    }
+    rec = {"idx": idx}
+    t0 = time.perf_counter()
+    first = last = None
+    first_ct = None
+    final_meta = None
+    max_ct = 0
+    try:
+        with sess.post(url, json=payload, stream=True, timeout=3600) as resp:
+            for raw in resp.iter_lines():
+                if not raw or not raw.startswith(b"data:"):
+                    continue
+                body = raw[5:].strip()
+                if body == b"[DONE]":
+                    continue
+                now = time.perf_counter()
+                try:
+                    d = json.loads(body)
+                except Exception:
+                    continue
+                mi = d.get("meta_info") or {}
+                ct = mi.get("completion_tokens") or 0
+                if ct:
+                    max_ct = max(max_ct, ct)
+                if first is None:
+                    first = now
+                    first_ct = ct
+                last = now
+                if mi.get("finish_reason"):
+                    final_meta = mi
+        t_end = time.perf_counter()
+        n_out = max_ct or ((final_meta or {}).get("completion_tokens") or 0)
+        decode_span = (last - first) if (first and last and last > first) else 0.0
+        rec.update(
+            ok=n_out > 0,
+            ttft=(first - t0) if first else None,
+            e2e=t_end - t0,
+            n_out=n_out,
+            first_chunk_tokens=first_ct,
+            decode_span=decode_span,
+            tpot=(decode_span / (n_out - 1)) if n_out > 1 else None,
+            per_req_decode_tok_s=(n_out / decode_span) if decode_span > 0 else None,
+            retractions=(final_meta or {}).get("num_retractions"),
+            spec_accept_len=(final_meta or {}).get("spec_accept_length"),
+        )
+    except Exception as e:
+        rec.update(ok=False, error=repr(e))
+    results[idx] = rec
+
+
+def verify_hit_rate(container, t_start, t_end):
+    def rfc3339(epoch):
+        return (datetime.datetime.fromtimestamp(epoch, tz=datetime.timezone.utc)
+                .isoformat().replace("+00:00", "Z"))
+
+    try:
+        # No margin before t_start: warmup's prefill lines end strictly before it,
+        # and catching them would deflate the measured hit rate.
+        p = subprocess.run(
+            ["docker", "logs", container, "--since", rfc3339(t_start), "--until", rfc3339(t_end + 2)],
+            capture_output=True, text=True, timeout=120)
+        text = p.stdout + p.stderr
+    except Exception as e:
+        return {"error": repr(e)}
+    pat = re.compile(r"#new-token: (\d+).*?#cached-token: (\d+)")
+    n_batches = new_tok = cached_tok = 0
+    for line in text.splitlines():
+        if "TP0]" not in line or "Prefill batch" not in line:
+            continue
+        m = pat.search(line)
+        if m:
+            n_batches += 1
+            new_tok += int(m.group(1))
+            cached_tok += int(m.group(2))
+    total = new_tok + cached_tok
+    return {
+        "prefill_batches": n_batches,
+        "new_tokens": new_tok,
+        "cached_tokens": cached_tok,
+        "hit_rate": round(cached_tok / total, 4) if total else None,
+    }
+
+
+def stats(vals):
+    vals = [v for v in vals if v is not None]
+    if not vals:
+        return {"mean": None, "p50": None, "max": None, "min": None}
+    s = sorted(vals)
+    return {
+        "mean": round(statistics.fmean(vals), 4),
+        "p50": round(s[len(s) // 2], 4),
+        "max": round(s[-1], 4),
+        "min": round(s[0], 4),
+    }
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--concurrency", type=int, required=True)
+    ap.add_argument("--num-requests", type=int, required=True)
+    ap.add_argument("--run-id", type=int, default=0,
+                    help="unique id per run so suffix prompts are never reused across runs")
+    ap.add_argument("--input-len", type=int, default=131072, help="total prompt tokens (16k=16384, 64k=65536, 128k=131072)")
+    ap.add_argument("--output-len", type=int, default=OUTPUT_LEN_DEFAULT)
+    ap.add_argument("--shared-frac", type=float, default=0.9,
+                    help="fraction of each prompt that is a shared prefix (radix hit); 0 = fully unique prompts")
+    ap.add_argument("--url", default="http://127.0.0.1:30000/generate")
+    ap.add_argument("--container", default="glm53-nvfp4")
+    args = ap.parse_args()
+
+    shared_len, unique_len = split_lens(args.input_len, args.shared_frac)
+    shared, prompts = build_prompts(shared_len, unique_len, args.num_requests, args.run_id)
+    warmup(args.url, shared)
+
+    results = {}
+    t_start = time.time()
+    t0 = time.perf_counter()
+    with ThreadPoolExecutor(max_workers=args.concurrency) as ex:
+        futs = [ex.submit(bench_one, args.url, p, args.output_len, i, results) for i, p in enumerate(prompts)]
+        for f in futs:
+            f.result()
+    wall = time.perf_counter() - t0
+    t_end = time.time()
+
+    hit = verify_hit_rate(args.container, t_start, t_end)
+
+    ok = [r for r in results.values() if r.get("ok")]
+    n_out_total = sum(r["n_out"] for r in ok)
+    out_tps = [r["n_out"] / r["e2e"] for r in ok if r.get("e2e")]
+    ttft = stats([r.get("ttft") for r in ok])
+    tpot = stats([r.get("tpot") for r in ok])
+    e2e = stats([r.get("e2e") for r in ok])
+    dec = stats([r.get("per_req_decode_tok_s") for r in ok])
+    spec = [v for v in (r.get("spec_accept_len") for r in ok) if v is not None]
+    retr = sum(r.get("retractions") or 0 for r in ok)
+
+    summary = {
+        "concurrency": args.concurrency,
+        "num_requests": args.num_requests,
+        "run_id": args.run_id,
+        "ok": len(ok),
+        "failed": args.num_requests - len(ok),
+        "wall_s": round(wall, 2),
+        "input_len": args.input_len,
+        "shared_len": shared_len,
+        "unique_len": unique_len,
+        "output_len": args.output_len,
+        "output_tokens_total": n_out_total,
+        "output_throughput_tok_s": round(n_out_total / wall, 2) if wall else None,
+        "input_throughput_tok_s": round(args.input_len * len(ok) / wall, 2) if wall else None,
+        "ttft_s": ttft,
+        "tpot_s": tpot,
+        "e2e_s": e2e,
+        "per_req_out_tok_s_e2e": stats(out_tps),
+        "per_req_decode_tok_s": dec,
+        "spec_accept_length_mean": round(statistics.fmean(spec), 3) if spec else None,
+        "retractions_total": retr,
+        "cache_hit_from_logs": hit,
+    }
+    print("\n===== SUMMARY =====")
+    print(json.dumps(summary, indent=2), flush=True)
+
+
+if __name__ == "__main__":
+    main()

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/bench_matrix.sh

@@ -0,0 +1,26 @@
+#!/bin/bash
+# Usage: bench_matrix.sh <tag> <runid_base> [url]
+# Runs the 4-scenario matrix serially in background-safe fashion:
+#   128k/90%hit x cc1, cc2 ; 64k/90%hit x cc1, cc2   (NUMREQ requests each, default 4)
+#
+# Repo 版与 60.7 原版（md5 b961a56bc7c8f26c519fdc7f5f627ba1）唯一差异：
+#   bench 脚本按本目录解析，日志目录可用 LOG_DIR 覆盖（默认 /root/bench_logs）。
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+LOG_DIR="${LOG_DIR:-/root/bench_logs}"
+TAG=$1
+RID=$2
+URL=${3:-http://127.0.0.1:30000/generate}
+NUMREQ=${NUMREQ:-4}
+mkdir -p "$LOG_DIR"
+for spec in "131072 1" "131072 2" "65536 1" "65536 2"; do
+  set -- $spec
+  IL=$1; CC=$2
+  if [ "$IL" -eq 131072 ]; then K=128k; else K=64k; fi
+  LOG=$LOG_DIR/${TAG}_${K}_cc${CC}.log
+  echo "[$(date +%T)] START ${TAG} ${K} cc${CC} run-id ${RID} numreq ${NUMREQ}"
+  python3 "${SCRIPT_DIR}/bench_hit90.py" --input-len $IL --concurrency $CC --num-requests $NUMREQ \
+      --run-id $RID --url $URL > "$LOG" 2>&1
+  RID=$((RID+1))
+  echo "[$(date +%T)] DONE ${TAG} ${K} cc${CC} (exit $?)"
+done
+echo ALL_DONE

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/bench_report.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+# One bench point for the two-scenario report (synthetic random-id variant,
+# bench_hit90.py — 仅作快速对照用，正式口径以 bench_corpus.py 真实语料为准).
+# Usage: bench_report.sh <tag> <input_len> <cc> <numreq> <run_id> [shared_frac]
+# Log: ${LOG_DIR}/<tag>_i<LEN>k_cc<CC>.log
+#
+# Repo 版与 60.7 原版（md5 282f8cdf5873d3a3eb8a950a1564acae）唯一差异：
+#   bench 脚本按本目录解析，日志目录可用 LOG_DIR 覆盖（默认 /root/bench_logs）。
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+LOG_DIR="${LOG_DIR:-/root/bench_logs}"
+TAG=$1; ILEN=$2; CC=$3; NREQ=$4; RID=$5; SF=${6:-0.9}
+mkdir -p "$LOG_DIR"
+LOG=$LOG_DIR/${TAG}_i$((ILEN/1024))k_cc${CC}.log
+echo "[$(date +%H:%M:%S)] START $TAG i$ILEN cc$CC nreq$NREQ run$RID sf$SF"
+python3 "${SCRIPT_DIR}/bench_hit90.py" --concurrency "$CC" --num-requests "$NREQ" \
+  --run-id "$RID" --input-len "$ILEN" --output-len 512 --shared-frac "$SF" \
+  > "$LOG" 2>&1
+RC=$?
+echo "[$(date +%H:%M:%S)] DONE $TAG i$ILEN cc$CC run$RID exit=$RC"
+exit $RC

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/car_patch/custom_all_reduce.py

@@ -0,0 +1,424 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# Adapted from https://github.com/vllm-project/vllm/blob/v0.6.4.post1/vllm/distributed/device_communicators/custom_all_reduce.py
+
+import ctypes
+import logging
+import os
+from contextlib import contextmanager
+from functools import partial
+from typing import Any, List, Optional, Union
+
+import torch
+import torch.distributed as dist
+from torch.distributed import ProcessGroup
+
+import sglang.srt.distributed.device_communicators.custom_all_reduce_ops as ops
+from sglang.srt.distributed.device_communicators.cuda_wrapper import CudaRTLibrary
+from sglang.srt.distributed.device_communicators.custom_all_reduce_utils import (
+    can_use_custom_all_reduce_with_nvlink,
+    is_weak_contiguous,
+)
+from sglang.srt.environ import envs
+from sglang.srt.model_executor.runner_backend_utils.tc_piecewise_cuda_graph import (
+    is_in_tc_piecewise_cuda_graph,
+)
+from sglang.srt.utils import (
+    get_bool_env_var,
+    is_cuda,
+    is_hip,
+    is_musa,
+    log_info_on_rank0,
+)
+
+_is_cuda = is_cuda()
+_is_hip = is_hip()
+_is_musa = is_musa()
+
+logger = logging.getLogger(__name__)
+os.environ.setdefault("SGLANG_CUSTOM_ALLREDUCE_ALGO", "1stage")  # SSKJ-PATCH: C++ dispatch no-ops when full_nvlink=False; force kernel launch
+
+
+class CustomAllreduce:
+    _SUPPORTED_WORLD_SIZES = [2, 4, 6, 8]
+    _MAX_CAR_SIZE = 8192 * 1024
+    if _is_hip:
+        # crossover is at 16MB buffer size for ROCm
+        _MAX_CAR_SIZE = 2 * 8192 * 1024
+    if _is_musa:
+        # crossover is at 128MB buffer size for MUSA
+        _MAX_CAR_SIZE = 16 * 8196 * 1024
+
+    # max_size: max supported allreduce size
+    def __init__(
+        self,
+        group: ProcessGroup,
+        device: Union[int, str, torch.device],
+        max_size=_MAX_CAR_SIZE,
+    ) -> None:
+        """
+        Args:
+            group: the process group to work on. If None, it will use the
+                default process group.
+            device: the device to bind the CustomAllreduce to. If None,
+                it will be bind to f"cuda:{local_rank}".
+        It is the caller's responsibility to make sure each communicator
+        is bind to a unique device, and all communicators in this group
+        are in the same node.
+        """
+        self._IS_CAPTURING = False
+        self.disabled = True  # This can be modified in-place by context manager in piecewise cuda graph runner
+        self.original_disabled = True  # To store the original state
+        self.use_amd_deterministic_impl = _use_amd_deterministic_impl()
+
+        if not ops.IS_CUSTOM_AR_AVAILABLE:
+            # disable because of missing custom allreduce library
+            # e.g. in a non-cuda environment
+            return
+
+        rank = dist.get_rank(group=group)
+        world_size = dist.get_world_size(group=group)
+
+        if isinstance(device, int):
+            device = torch.device(f"cuda:{device}")
+        elif isinstance(device, str):
+            device = torch.device(device)
+        # now `device` is a `torch.device` object
+        assert isinstance(device, torch.device)
+        self.device = device
+        full_nvlink = can_use_custom_all_reduce_with_nvlink(
+            group=group,
+            device=device,
+            supported_world_size=self._SUPPORTED_WORLD_SIZES,
+            cls_name="CustomAllreduce",
+        )
+        if full_nvlink is None:
+            return  # fail to get nvlink status
+
+        self.group = group
+        self.max_size = max_size
+        self.rank = rank
+        self.world_size = world_size
+        self.full_nvlink = full_nvlink
+
+        if not _is_hip:
+            # Buffers memory are owned by this Python class and passed to C++.
+            # Meta data composes of two parts: meta data for synchronization and a
+            # temporary buffer for storing intermediate allreduce results.
+            self.meta_ptrs = self.create_shared_buffer(
+                ops.meta_size() + max_size, group=group
+            )
+            # This is a pre-registered IPC buffer. In eager mode, input tensors
+            # are first copied into this buffer before allreduce is performed
+            self.buffer_ptrs = self.create_shared_buffer(max_size, group=group)
+            # This is a buffer for storing the tuples of pointers pointing to
+            # IPC buffers from all ranks. Each registered tuple has size of
+            # 8*world_size bytes where world_size is at most 8. Allocating 8MB
+            # is enough for 131072 such tuples. The largest model I've seen only
+            # needs less than 10000 of registered tuples.
+            self.rank_data = torch.empty(
+                max_size, dtype=torch.uint8, device=self.device
+            )
+            self._ptr = ops.init_custom_ar(
+                self.meta_ptrs, self.rank_data, rank, self.full_nvlink
+            )
+            ops.register_buffer(self._ptr, self.buffer_ptrs)
+        else:
+            # meta data buffers need to be "uncached" for signal on MI200
+            self.meta = ops.allocate_meta_buffer(ops.meta_size() + max_size)
+            self.buffer = torch.empty(max_size, dtype=torch.uint8, device=self.device)
+            handle = ops.get_meta_buffer_ipc_handle(self.meta)
+            shard_data = (
+                bytes(handle),  # ipc handle to base ptr
+                0,  # offset of base ptr
+            )
+            handles, offsets = self._gather_ipc_meta(shard_data)
+            self.rank_data = torch.empty(
+                max_size, dtype=torch.uint8, device=self.device
+            )
+            self._ptr = ops.init_custom_ar(
+                self.meta, self.rank_data, handles, offsets, rank, self.full_nvlink
+            )
+            self.register_buffer(self.buffer)
+
+        self.disabled = False
+        self.original_disabled = False  # Ensure original_disabled == disabled
+        logger.warning(f"SSKJ_CAR_PATCH_ACTIVE ws={self.world_size} full_nvlink={self.full_nvlink}")
+        self.tms_cudagraph = envs.SGLANG_MEMORY_SAVER_CUDA_GRAPH.get()
+
+    @staticmethod
+    def create_shared_buffer(
+        size_in_bytes: int, group: Optional[ProcessGroup] = None
+    ) -> List[int]:
+        """
+        Creates a shared buffer and returns a list of pointers
+        representing the buffer on all processes in the group.
+        """
+        lib = CudaRTLibrary()
+        pointer = lib.cudaMalloc(size_in_bytes)
+        if _is_musa:
+            lib.cudaMemset(pointer, 0, size_in_bytes)
+        handle = lib.cudaIpcGetMemHandle(pointer)
+        world_size = dist.get_world_size(group=group)
+        rank = dist.get_rank(group=group)
+        handles = [None] * world_size
+        dist.all_gather_object(handles, handle, group=group)
+
+        pointers: List[int] = []
+        for i, h in enumerate(handles):
+            if i == rank:
+                pointers.append(pointer.value)  # type: ignore
+            else:
+                pointers.append(lib.cudaIpcOpenMemHandle(h).value)  # type: ignore
+
+        return pointers
+
+    @staticmethod
+    def free_shared_buffer(
+        pointers: List[int], group: Optional[ProcessGroup] = None
+    ) -> None:
+        rank = dist.get_rank(group=group)
+        lib = CudaRTLibrary()
+        lib.cudaFree(ctypes.c_void_p(pointers[rank]))
+
+    @contextmanager
+    def capture(self):
+        """
+        The main responsibility of this context manager is the
+        `register_graph_buffers` call at the end of the context.
+        It records all the buffer addresses used in the CUDA graph.
+        """
+        try:
+            self._IS_CAPTURING = True
+            yield
+        finally:
+            self._IS_CAPTURING = False
+            if not self.disabled:
+                self.register_graph_buffers()
+
+    def _get_ipc_meta(self, inp: torch.Tensor):
+        # _share_cuda_() doesn't accept meta buffer not allocated from
+        # PyTorch cache allocator, use direct HIP call to get IPC handle
+        handle = ops.get_meta_buffer_ipc_handle(inp)
+        shard_data = (
+            bytes(handle),  # ipc handle to base ptr
+            0,  # offset of base ptr
+        )
+        return self._gather_ipc_meta(shard_data)
+
+    def _gather_ipc_meta(self, shard_data):
+        # Note: don't use `[[None]] * self.world_size` here
+        # because it will create a list of the same reference
+        all_data: List[Optional[Any]] = [[None] for i in range(self.world_size)]
+        all_data[self.rank][0] = shard_data
+
+        ranks = dist.get_process_group_ranks(group=self.group)
+        ranks.sort()
+        for i, rank in enumerate(ranks):
+            dist.broadcast_object_list(
+                all_data[i], src=rank, group=self.group, device="cpu"
+            )
+
+        # we cannot directly use `dist.all_gather_object` here
+        # because it is incompatible with `gloo` backend under inference mode.
+        # see https://github.com/pytorch/pytorch/issues/126032 for details.
+
+        handles = []
+        offsets = []
+        for i in range(len(all_data)):
+            handles.append(all_data[i][0][0])  # type: ignore
+            offsets.append(all_data[i][0][1])  # type: ignore
+        return handles, offsets
+
+    def register_buffer(self, inp: torch.Tensor):
+        handles, offsets = self._get_ipc_meta(inp)
+        ops.register_buffer(self._ptr, inp, handles, offsets)
+
+    def register_graph_buffers(self):
+        if _is_hip:
+            handle, offset = ops.get_graph_buffer_ipc_meta(self._ptr)
+            handles, offsets = self._gather_ipc_meta((bytes(handle), offset))
+            log_info_on_rank0(logger, f"Registering {len(offset)} cuda graph addresses")
+            ops.register_graph_buffers(self._ptr, handles, offsets)
+        else:
+            handle, offset = ops.get_graph_buffer_ipc_meta(self._ptr)
+            log_info_on_rank0(logger, f"Registering {len(offset)} cuda graph addresses")
+            # We cannot directly use `dist.all_gather_object` here
+            # because it is incompatible with `gloo` backend under inference mode.
+            # see https://github.com/pytorch/pytorch/issues/126032 for details.
+            all_data = [
+                [None, None] for _ in range(dist.get_world_size(group=self.group))
+            ]
+            all_data[self.rank] = [handle, offset]
+            ranks = sorted(dist.get_process_group_ranks(group=self.group))
+            for i, rank in enumerate(ranks):
+                dist.broadcast_object_list(
+                    all_data[i], src=rank, group=self.group, device="cpu"
+                )
+            # Unpack list of tuples to tuple of lists.
+            handles = [d[0] for d in all_data]  # type: ignore
+            offsets = [d[1] for d in all_data]  # type: ignore
+            ops.register_graph_buffers(self._ptr, handles, offsets)
+
+    def should_custom_ar(self, inp: torch.Tensor):
+        if self.disabled:
+            return False
+        inp_size = inp.numel() * inp.element_size()
+        # custom allreduce requires input byte size to be multiples of 16
+        if inp_size % 16 != 0:
+            return False
+        if not is_weak_contiguous(inp):
+            return False
+        # for 4 or more non NVLink-capable GPUs, custom allreduce provides
+        # little performance improvement over NCCL.
+        if not _is_hip:
+            if True:
+                return inp_size <= self.max_size
+            return False
+
+        if _is_hip:
+            if self.use_amd_deterministic_impl:
+                return True
+            if self.full_nvlink:
+                return inp_size <= self.max_size
+            return False
+
+        return False
+
+    def _all_reduce_impl(self, inp: torch.Tensor, registered: bool):
+        out = torch.empty_like(inp)
+        if not _is_hip:  # CUDA-like
+            if registered:
+                ops.all_reduce(self._ptr, inp, out, 0, 0)
+            else:
+                ops.all_reduce(
+                    self._ptr, inp, out, self.buffer_ptrs[self.rank], self.max_size
+                )
+        elif self.use_amd_deterministic_impl:
+            inp_size = inp.numel() * inp.element_size()
+            if inp_size < self.max_size:
+                reg_buffer = self.buffer.view(inp.dtype)[: inp.numel()]
+                ops.deterministic_all_reduce_unreg(self._ptr, inp, reg_buffer, out)
+            else:
+                self.register_buffer(inp)
+                ops.deterministic_all_reduce_reg(self._ptr, inp, out)
+        else:  # normal AMD ROCm path
+            if registered:
+                ops.all_reduce_reg(self._ptr, inp, out)
+            else:
+                ops.all_reduce_unreg(self._ptr, inp, self.buffer, out)
+        return out
+
+    def custom_all_reduce(self, input: torch.Tensor) -> Optional[torch.Tensor]:
+        """The main allreduce API that provides support for cuda graph."""
+        # When custom allreduce is disabled, this will be None.
+        if self.disabled or not self.should_custom_ar(input):
+            return None
+        if self._IS_CAPTURING:
+            if torch.cuda.is_current_stream_capturing():
+                return self._all_reduce_impl(input, registered=not self.tms_cudagraph)
+            else:
+                # Could be warmup OR piecewise cuda graph split op execution.
+                # In piecewise cuda graph, split ops run eagerly outside the graph
+                # but _IS_CAPTURING is still True. We need to do real all-reduce.
+                if is_in_tc_piecewise_cuda_graph():
+                    # Split op execution - do real all-reduce
+                    return self._all_reduce_impl(input, registered=False)
+                else:
+                    # True warmup - mimic the allocation pattern since custom
+                    # allreduce is out-of-place.
+                    return torch.zeros_like(input)
+        else:
+            return self._all_reduce_impl(input, registered=False)
+
+    def close(self):
+        if not self.disabled and self._ptr:
+            if ops is not None:
+                ops.dispose(self._ptr)
+            if _is_cuda:
+                self.free_shared_buffer(self.meta_ptrs)
+                self.free_shared_buffer(self.buffer_ptrs)
+            self._ptr = 0
+
+    def __del__(self):
+        self.close()
+
+
+def dispatch_custom_allreduce(
+    group: ProcessGroup,
+    device: torch.device,
+):
+    """Return the CustomAllreduce class to use (aiter on ROCm if enabled).
+
+    On AMD with 1-stage AR enabled, use sglang's CustomAllreduce.
+    Otherwise use AiterCustomAllreduce if available.
+
+    On CUDA, the JIT-compiled v2 implementation is used by default.
+    Set SGLANG_OPT_USE_CUSTOM_ALL_REDUCE_V2=0 to fall back to the legacy CustomAllreduce.
+    Multi-node v2 is admitted only for a single NVLink clique (see
+    can_use_custom_all_reduce_v2); other cross-node groups fall back to NCCL.
+    """
+    if _is_cuda and envs.SGLANG_OPT_USE_CUSTOM_ALL_REDUCE_V2.get():
+        from .custom_all_reduce_v2 import (
+            CustomAllReduceV2,
+            can_use_custom_all_reduce_v2,
+        )
+
+        if can_use_custom_all_reduce_v2(group=group, device=device):
+            logger.debug("[AR] Using CustomAllReduceV2 (JIT-compiled)")
+            return CustomAllReduceV2
+
+    if _is_cuda or _is_musa:
+        return CustomAllreduce
+
+    assert _is_hip
+
+    if envs.SGLANG_USE_1STAGE_ALLREDUCE.is_set():
+        if envs.SGLANG_USE_1STAGE_ALLREDUCE.get():
+            logger.debug(
+                "[AR] All-reduce: 1-stage kernel (SGLANG_USE_1STAGE_ALLREDUCE=1)"
+            )
+        else:
+            logger.debug("[AR] All-reduce: default (SGLANG_USE_1STAGE_ALLREDUCE=0)")
+    elif envs.SGLANG_ENABLE_DETERMINISTIC_INFERENCE.get():
+        logger.debug(
+            "[AR] All-reduce: 1-stage kernel (deterministic inference enabled)"
+        )
+    else:
+        logger.debug("[AR] All-reduce: default")
+
+    # On AMD with 1-stage AR, use sglang's CustomAllreduce
+    # (AiterCustomAllreduce doesn't have deterministic_all_reduce method)
+    if _use_amd_deterministic_impl():
+        return CustomAllreduce
+
+    if get_bool_env_var("SGLANG_USE_AITER_AR", default="true"):
+        try:
+            from aiter.dist.device_communicators.custom_all_reduce import (
+                CustomAllreduce as AiterCustomAllreduce,
+            )
+
+            logger.info("[AR] Using AiterCustomAllreduce (AMD default)")
+            tms_cudagraph = envs.SGLANG_MEMORY_SAVER_CUDA_GRAPH.get()
+            return partial(
+                AiterCustomAllreduce,
+                enable_register_for_capturing=not tms_cudagraph,
+            )
+        except ImportError as e:
+            logger.warning(
+                "[AR] Aiter custom all-reduce not available; "
+                "falling back to sglang CustomAllreduce. Details: %s",
+                e,
+            )
+            return CustomAllreduce
+
+    return CustomAllreduce
+
+
+def _use_amd_deterministic_impl() -> bool:
+    if not _is_hip:  # CUDA is always deterministic
+        return False
+    if envs.SGLANG_USE_1STAGE_ALLREDUCE.is_set():
+        return envs.SGLANG_USE_1STAGE_ALLREDUCE.get()
+    else:
+        return envs.SGLANG_ENABLE_DETERMINISTIC_INFERENCE.get()

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/car_patch/custom_all_reduce_utils.py

@@ -0,0 +1,519 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# Adapted from https://github.com/vllm-project/vllm/blob/v0.6.4.post1/vllm/distributed/device_communicators/custom_all_reduce_utils.py
+
+import ctypes
+import json
+import logging
+import os
+import pickle
+import subprocess
+import sys
+import tempfile
+from functools import wraps
+from itertools import product
+from typing import Callable, Dict, List, Optional, Sequence, TypeVar
+
+import torch
+import torch.distributed as dist
+import torch.multiprocessing as mp
+from typing_extensions import ParamSpec
+
+from sglang.srt.distributed.device_communicators.cuda_wrapper import CudaRTLibrary
+from sglang.srt.distributed.parallel_state import in_the_same_node_as
+from sglang.srt.environ import envs as sglang_envs
+from sglang.srt.utils import is_cuda, is_hip, is_musa
+from sglang.srt.utils.cuda_vmm_utils import _gpu_fabric_clique
+
+logger = logging.getLogger(__name__)
+
+_is_cuda = is_cuda()
+_is_hip = is_hip()
+_is_musa = is_musa()
+
+if _is_cuda:
+    try:
+        import pynvml
+    except ImportError as e:
+        logger.warning("Failed to import pynvml with %r", e)
+
+if _is_musa:
+    try:
+        import pymtml as pynvml
+    except ImportError as e:
+        logger.warning("Failed to import pymtml with %r", e)
+
+if _is_hip:
+    try:
+        from amdsmi import (
+            AmdSmiException,
+            amdsmi_get_processor_handles,
+            amdsmi_init,
+            amdsmi_shut_down,
+            amdsmi_topo_get_link_type,
+        )
+    except ImportError as e:
+        logger.warning("Failed to import amdsmi with %r", e)
+
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
+
+
+def update_environment_variables(envs: Dict[str, str]):
+    for k, v in envs.items():
+        if k in os.environ and os.environ[k] != v:
+            logger.warning(
+                "Overwriting environment variable %s " "from '%s' to '%s'",
+                k,
+                os.environ[k],
+                v,
+            )
+        os.environ[k] = v
+
+
+def producer(
+    batch_src: Sequence[int],
+    producer_queue,
+    consumer_queue,
+    result_queue,
+    cuda_visible_devices: Optional[str] = None,
+):
+    if cuda_visible_devices is not None:
+        update_environment_variables({"CUDA_VISIBLE_DEVICES": cuda_visible_devices})
+
+    lib = CudaRTLibrary()
+    for i in batch_src:
+        lib.cudaSetDevice(i)
+        pointer = lib.cudaMalloc(1024)
+        lib.cudaMemset(pointer, 1, 1024)
+        lib.cudaDeviceSynchronize()
+        handle = lib.cudaIpcGetMemHandle(pointer)
+        producer_queue.put(handle)
+        open_success = consumer_queue.get()
+        if open_success:
+            # use two queues to simulate barrier
+            producer_queue.put(0)
+            consumer_queue.get()
+            # check if the memory is modified
+            host_data = (ctypes.c_char * 1024)()
+            lib.cudaMemcpy(host_data, pointer, 1024)  # type: ignore
+            for i in range(1024):
+                if ord(host_data[i]) != 2:
+                    open_success = False
+                    break
+        result_queue.put(open_success)
+        lib.cudaDeviceReset()
+
+
+def consumer(
+    batch_tgt: Sequence[int],
+    producer_queue,
+    consumer_queue,
+    result_queue,
+    cuda_visible_devices: Optional[str] = None,
+):
+    if cuda_visible_devices is not None:
+        update_environment_variables({"CUDA_VISIBLE_DEVICES": cuda_visible_devices})
+
+    lib = CudaRTLibrary()
+    for j in batch_tgt:
+        lib.cudaSetDevice(j)
+        handle = producer_queue.get()
+        open_success = False
+        try:
+            pointer = lib.cudaIpcOpenMemHandle(handle)  # type: ignore
+            open_success = True
+        except RuntimeError:
+            # cannot error out here, because the producer process
+            # is still waiting for the response.
+            pass
+        consumer_queue.put(open_success)
+        if open_success:
+            # modify the memory
+            lib.cudaMemset(pointer, 2, 1024)
+            lib.cudaDeviceSynchronize()
+            # use two queues to simulate barrier
+            producer_queue.get()
+            consumer_queue.put(0)
+            # check if the memory is modified
+            host_data = (ctypes.c_char * 1024)()
+            lib.cudaMemcpy(host_data, pointer, 1024)  # type: ignore
+            for i in range(1024):
+                if ord(host_data[i]) != 2:
+                    open_success = False
+                    break
+        result_queue.put(open_success)
+        lib.cudaDeviceReset()
+
+
+def can_actually_p2p(
+    batch_src: Sequence[int],
+    batch_tgt: Sequence[int],
+) -> Sequence[bool]:
+    """
+    Usually, checking if P2P access is enabled can be done by
+    `torch.cuda.can_device_access_peer(src, tgt)`. However, sometimes
+    the driver might be broken, and `torch.cuda.can_device_access_peer(src, tgt)`
+    returns `True` even if P2P access is not actually possible.
+    See https://github.com/vllm-project/vllm/issues/2728 and
+    https://forums.developer.nvidia.com/t/direct-gpu-gpu-communication-does-not-seem-to-work-properly/283264/10
+    Therefore, we have to perform a real P2P access to check if it is actually
+    possible.
+
+    Note on p2p and cuda IPC:
+    Usually, one process uses one GPU:
+    GPU src --> cuda context src --> tensor src --> process src
+
+    We need to combine p2p and cuda IPC, so that:
+    GPU src --> cuda context src --> tensor src --> process src
+                                      |shared|
+    GPU tgt --> cuda context tgt --> tensor tgt --> process tgt
+    That is to say, process src creates a tensor in GPU src, passes IPC handle to
+    process tgt, and process tgt accesses the tensor in GPU tgt. Any operation on the
+    tensor in process tgt will be reflected in the tensor in process src, because
+    they are the same memory segment.
+    It is important to note that process tgt accesses the tensor in GPU tgt, not
+    GPU src. That's why we need p2p access.
+
+    The most time-consuming part is the process creation. To avoid creating
+    processes for every pair of GPUs, we use batched testing. We create two
+    processes for testing all pairs of GPUs in batch. The trick is to reset
+    the device after each test (which is not available in PyTorch).
+    """  # noqa
+    cuda_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES", None)
+    # pass the CUDA_VISIBLE_DEVICES to the child process
+    # to make sure they see the same set of GPUs
+
+    # make sure the processes are spawned
+    smp = mp.get_context("spawn")
+    producer_queue = smp.Queue()
+    consumer_queue = smp.Queue()
+    result_queue = smp.Queue()
+    p_src = smp.Process(
+        target=producer,
+        args=(
+            batch_src,
+            producer_queue,
+            consumer_queue,
+            result_queue,
+            cuda_visible_devices,
+        ),
+    )
+    p_tgt = smp.Process(
+        target=consumer,
+        args=(
+            batch_tgt,
+            producer_queue,
+            consumer_queue,
+            result_queue,
+            cuda_visible_devices,
+        ),
+    )
+    p_src.start()
+    p_tgt.start()
+    p_src.join()
+    p_tgt.join()
+    assert p_src.exitcode == 0 and p_tgt.exitcode == 0
+    result: List[bool] = []
+    for src, tgt in zip(batch_src, batch_tgt):
+        a = result_queue.get()
+        b = result_queue.get()
+        if a != b:
+            logger.warning(
+                "Two processes do not agree on the P2P access"
+                " status on %d -> %d, treat as disabled.",
+                src,
+                tgt,
+            )
+            result.append(False)
+        else:
+            result.append(a)
+    return result
+
+
+# why do we need this cache?
+# we are testing peer-to-peer (p2p) access between GPUs,across processes.
+# if we test it every time, it will be very slow, because we need to create
+#  N * N * 2 processes, where N is the world size. This is very slow.
+# to reduce the time, we use a cache file to store the p2p access status.
+# the cache file is generated by the master process if it does not exist.
+# then all the processes can read the cache file to check the p2p access status.
+# Note that the cache file is suffixed by the CUDA_VISIBLE_DEVICES, so that we
+#  can have different cache files for different CUDA_VISIBLE_DEVICES settings,
+#  e.g. used by different vllm engines. The device id in the cache file is a
+#  **local** device id, i.e. from 0 to num_dev-1, where num_dev is the number
+#  of visible devices in the vllm engine.
+_gpu_p2p_access_cache: Optional[Dict[str, bool]] = None
+
+
+def gpu_p2p_access_check(src: int, tgt: int) -> bool:
+    """Check if GPU src can access GPU tgt."""
+
+    # if the cache variable is already calculated,
+    # read from the cache instead of checking it again
+    global _gpu_p2p_access_cache
+    if _gpu_p2p_access_cache is not None:
+        return _gpu_p2p_access_cache[f"{src}->{tgt}"]
+
+    is_distributed = dist.is_initialized()
+
+    num_dev = torch.cuda.device_count()
+    cuda_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES", None)
+    if cuda_visible_devices is None:
+        cuda_visible_devices = ",".join(str(i) for i in range(num_dev))
+
+    # VLLM_CACHE_ROOT -> SGLANG_CACHE_ROOT
+    # "~/.cache/vllm" -> envs.SGLANG_CACHE_DIR
+    SGLANG_CACHE_ROOT = os.path.expanduser(sglang_envs.SGLANG_CACHE_DIR.get())
+    path = os.path.join(
+        SGLANG_CACHE_ROOT, f"gpu_p2p_access_cache_for_{cuda_visible_devices}.json"
+    )
+    cache_dir = os.path.dirname(path)
+    try:
+        os.makedirs(cache_dir, exist_ok=True)
+    except (FileExistsError, NotADirectoryError):
+        if not os.path.isdir(cache_dir):
+            # Path exists as a file (stale cache/lock). Remove and retry.
+            try:
+                os.remove(cache_dir)
+            except OSError:
+                pass
+            os.makedirs(cache_dir, exist_ok=True)
+    from sglang.srt.distributed.parallel_state import get_world_group
+
+    if (not is_distributed or get_world_group().local_rank == 0) and (
+        not os.path.exists(path)
+    ):
+        # only the local master process (with local_rank == 0) can
+        #  enter this block to calculate the cache
+        logger.info("generating GPU P2P access cache in %s", path)
+        cache: Dict[str, bool] = {}
+        ids = list(range(num_dev))
+        # batch of all pairs of GPUs
+        batch_src, batch_tgt = zip(*list(product(ids, ids)))
+        # NOTE: we use `subprocess` rather than `multiprocessing` here
+        # because the caller might not have `if __name__ == "__main__":`,
+        # in that case we cannot use spawn method in multiprocessing.
+        # However, `can_actually_p2p` requires spawn method.
+        # The fix is, we use `subprocess` to call the function,
+        # where we have `if __name__ == "__main__":` in this file.
+
+        # use a temporary file to store the result
+        # we don't use the output of the subprocess directly,
+        # because the subprocess might produce logging output
+        with tempfile.NamedTemporaryFile() as output_file:
+            input_bytes = pickle.dumps((batch_src, batch_tgt, output_file.name))
+            returned = subprocess.run(
+                [sys.executable, __file__], input=input_bytes, capture_output=True
+            )
+            # check if the subprocess is successful
+            try:
+                returned.check_returncode()
+            except Exception as e:
+                # wrap raised exception to provide more information
+                raise RuntimeError(
+                    f"Error happened when batch testing "
+                    f"peer-to-peer access from {batch_src} to {batch_tgt}:\n"
+                    f"{returned.stderr.decode()}"
+                ) from e
+            with open(output_file.name, "rb") as f:
+                result = pickle.load(f)
+        for _i, _j, r in zip(batch_src, batch_tgt, result):
+            cache[f"{_i}->{_j}"] = r
+        with open(path, "w") as f:
+            json.dump(cache, f, indent=4)
+    if is_distributed:
+        get_world_group().barrier()
+    logger.info("reading GPU P2P access cache from %s", path)
+    with open(path) as f:
+        cache = json.load(f)
+    _gpu_p2p_access_cache = cache
+    return _gpu_p2p_access_cache[f"{src}->{tgt}"]
+
+
+def with_nvml_context(fn: Callable[_P, _R]) -> Callable[_P, _R]:
+    @wraps(fn)
+    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
+        if _is_hip:
+            try:
+                amdsmi_init()
+                return fn(*args, **kwargs)
+            finally:
+                amdsmi_shut_down()
+        else:
+            pynvml.nvmlInit()
+            try:
+                return fn(*args, **kwargs)
+            finally:
+                pynvml.nvmlShutdown()
+
+    return wrapper
+
+
+@with_nvml_context
+def is_full_nvlink(physical_device_ids: List[int], world_size: int) -> bool:
+    if _is_hip:
+        """
+        query if the set of gpus are fully connected by xgmi (1 hop)
+        """
+        handles = [amdsmi_get_processor_handles()[i] for i in physical_device_ids]
+        for i, handle in enumerate(handles):
+            for j, peer_handle in enumerate(handles):
+                if i < j:
+                    try:
+                        link_type = amdsmi_topo_get_link_type(handle, peer_handle)
+                        # type is 2 for XGMI
+                        if link_type["hops"] != 1 or link_type["type"] != 2:
+                            return False
+                    except AmdSmiException as error:
+                        logger.error("AMD 1 hop XGMI detection failed.", exc_info=error)
+                        return False
+        return True
+    else:
+        """
+        query if the set of gpus are fully connected by nvlink (1 hop)
+        """
+        handles = [pynvml.nvmlDeviceGetHandleByIndex(i) for i in physical_device_ids]
+        for i, handle in enumerate(handles):
+            for j, peer_handle in enumerate(handles):
+                if i < j:
+                    try:
+                        p2p_status = pynvml.nvmlDeviceGetP2PStatus(
+                            handle, peer_handle, pynvml.NVML_P2P_CAPS_INDEX_NVLINK
+                        )
+                        if p2p_status != pynvml.NVML_P2P_STATUS_OK:
+                            return False
+                    except pynvml.NVMLError:
+                        logger.exception(
+                            "NVLink detection failed. This is normal if your"
+                            " machine has no NVLink equipped."
+                        )
+                        return False
+        return True
+
+
+@with_nvml_context
+def is_one_nvlink_clique(
+    group: torch.distributed.ProcessGroup, device: torch.device
+) -> bool:
+    """True iff every rank's GPU is in the same NVLink fabric clique (one NVL72 /
+    MNNVL domain). Such a clique shares a single NVLink address space even across
+    nodes, so custom-AR v2's symm-mem storage + fabric peer VAs are valid group-wide."""
+    if _is_hip:
+        return False
+    try:
+        clique = _gpu_fabric_clique(device)
+    except Exception as e:
+        logger.warning(
+            "GPU fabric clique query failed (%r); custom-AR stays intra-node.", e
+        )
+        clique = None
+    # Always all-gather (every rank calls it once) so a failed query on any rank
+    # resolves to a clean False rather than a collective mismatch.
+    world_size = dist.get_world_size(group=group)
+    gathered: List[object] = [None] * world_size
+    dist.all_gather_object(gathered, clique, group=group)
+    if any(c is None for c in gathered):
+        return False
+    return len(set(gathered)) == 1
+
+
+def is_weak_contiguous(inp: torch.Tensor):
+    return inp.is_contiguous() or (
+        inp.storage().nbytes() - inp.storage_offset() * inp.element_size()
+        == inp.numel() * inp.element_size()
+    )
+
+
+def can_p2p(rank: int, world_size: int) -> bool:
+    # SGLANG_SKIP_P2P_CHECK can be set to False in sglang
+    SGLANG_SKIP_P2P_CHECK = os.getenv("SGLANG_SKIP_P2P_CHECK", "0") == "1"
+    for i in range(world_size):
+        if i == rank:
+            continue
+        if SGLANG_SKIP_P2P_CHECK:
+            logger.info("Skipping P2P check and trusting the driver's P2P report.")
+            return torch.cuda.can_device_access_peer(rank, i)
+        if not gpu_p2p_access_check(rank, i):
+            return False
+    return True
+
+
+def can_use_custom_all_reduce_with_nvlink(
+    group: torch.distributed.ProcessGroup,
+    device: torch.device,
+    supported_world_size: List[int],
+    cls_name: str,
+) -> Optional[bool]:  # None if fail; otherwise return whether NVLink is available
+    assert (
+        dist.get_backend(group) != dist.Backend.NCCL
+    ), f"{cls_name} should be attached to a non-NCCL group."
+
+    rank = dist.get_rank(group=group)
+    world_size = dist.get_world_size(group=group)
+
+    # No need to initialize custom allreduce for single GPU case.
+    if world_size == 1:
+        return
+
+    # No need to initialize custom allreduce for multi-node case.
+    if not all(in_the_same_node_as(group, source_rank=0)):
+        logger.warning(
+            f"{cls_name} is disabled because this process group" " spans across nodes."
+        )
+        return
+
+    # For not supported world size, we disable custom allreduce.
+    if world_size not in supported_world_size:
+        logger.warning(
+            f"{cls_name} is disabled due to an unsupported world"
+            f" size: {world_size}. Supported world sizes: {supported_world_size}. "
+            "To silence this warning, specify disable_custom_all_reduce=True explicitly.",
+        )
+        return
+
+    cuda_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES", None)
+    if cuda_visible_devices:
+        device_ids = list(map(int, cuda_visible_devices.split(",")))
+    else:
+        device_ids = list(range(torch.cuda.device_count()))
+    physical_device_id = device_ids[device.index]
+    tensor = torch.tensor([physical_device_id], dtype=torch.int, device="cpu")
+    gather_list = [
+        torch.tensor([0], dtype=torch.int, device="cpu") for _ in range(world_size)
+    ]
+    dist.all_gather(gather_list, tensor, group=group)
+    physical_device_ids = [int(t) for t in gather_list]
+    full_nvlink = is_full_nvlink(physical_device_ids, world_size)
+
+    # test nvlink first, this will filter out most of the cases
+    # where custom allreduce is not supported
+    # this checks hardware and driver support for NVLink
+    if False:
+        logger.warning(
+            f"{cls_name} is disabled because it's not supported on"
+            " more than two PCIe-only GPUs. To silence this warning, "
+            "specify disable_custom_all_reduce=True explicitly."
+        )
+        return
+
+    # test P2P capability, this checks software/cudaruntime support
+    # this is expensive to compute at the first time
+    # then we cache the result
+    # On AMD GPU, p2p is always enabled between XGMI connected GPUs
+    if not _is_hip and not can_p2p(rank, world_size):
+        logger.warning(
+            f"{cls_name} is disabled because your platform lacks "
+            "GPU P2P capability or P2P test failed. To silence this "
+            "warning, specify disable_custom_all_reduce=True explicitly."
+        )
+        return
+
+    return full_nvlink
+
+
+if __name__ == "__main__":
+    batch_src, batch_tgt, output_file = pickle.loads(sys.stdin.buffer.read())
+    result = can_actually_p2p(batch_src, batch_tgt)
+    with open(output_file, "wb") as f:
+        f.write(pickle.dumps(result))

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/car_patch/prep_car_patch.sh

@@ -0,0 +1,27 @@
+#!/bin/bash
+# prep_car_patch.sh — build patched custom-AR files at /root/patches/
+# Patch A: should_custom_ar allows any CUDA AR <= max_size (8MB cap keeps
+#          big prefill ARs on NCCL); decode ARs (<=1MB) go custom AR.
+# Patch B: utils gate no longer bails on ws>2 without NVLink (P2P probe
+#          verified all-True on this 8-way PCIe box).
+# V2 stays excluded: can_use_custom_all_reduce_v2 requires full_nvlink is True.
+set -e
+D=/sgl-workspace/sglang/python/sglang/srt/distributed/device_communicators
+mkdir -p /root/patches
+docker cp glm53-nvfp4:$D/custom_all_reduce.py /root/patches/custom_all_reduce.py
+docker cp glm53-nvfp4:$D/custom_all_reduce_utils.py /root/patches/custom_all_reduce_utils.py
+
+ANCHOR='self.original_disabled = False  # Ensure original_disabled == disabled'
+n=$(grep -c "$ANCHOR" /root/patches/custom_all_reduce.py)
+[ "$n" = "1" ] || { echo "ANCHOR_NOT_UNIQUE=$n"; exit 1; }
+
+sed -i 's/if self.world_size == 2 or self.full_nvlink:/if True:/' /root/patches/custom_all_reduce.py
+sed -i 's/if world_size > 2 and not full_nvlink:/if False:/' /root/patches/custom_all_reduce_utils.py
+sed -i "/${ANCHOR}/a\\        logger.warning(f\"SSKJ_CAR_PATCH_ACTIVE ws={self.world_size} full_nvlink={self.full_nvlink}\")" /root/patches/custom_all_reduce.py
+
+python3 -m py_compile /root/patches/custom_all_reduce.py /root/patches/custom_all_reduce_utils.py
+echo "--- verify ---"
+grep -n 'if True:' /root/patches/custom_all_reduce.py
+grep -n 'if False:' /root/patches/custom_all_reduce_utils.py
+grep -n 'SSKJ_CAR_PATCH_ACTIVE' /root/patches/custom_all_reduce.py
+echo PREP_DONE

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/deploy_glm53_605.sh

@@ -0,0 +1,58 @@
+#!/bin/bash
+# deploy_glm53_605.sh — GLM-5.3-NVFP4 TP8 + EAGLE for 174.1.60.5 (team service)
+# Target scenario: cc1-2, 64k/128k input 90% cache hit, single request up to 256k,
+# output-throughput priority. Clean deploy: rm old container, wait VRAM drain, run.
+#
+# Env overrides:
+#   MEMFRAC (0.90)  STEPS (4)  TOPK (1)  DRAFT (5)  CTXLEN (270336)
+#   CHUNK (8192)  MAXPRE (16384)  RESTART (no|yes)  EXTRA ("")
+# Usage:
+#   bash deploy_glm53_605.sh                      # production 4/1/5 @ 0.90
+#   STEPS=5 DRAFT=6 bash deploy_glm53_605.sh      # tuning round
+#   CHUNK=16384 bash deploy_glm53_605.sh          # prefill tuning round
+#   RESTART=yes bash deploy_glm53_605.sh          # production finalize
+set -e
+MEMFRAC=${MEMFRAC:-0.90}
+STEPS=${STEPS:-4}
+TOPK=${TOPK:-1}
+DRAFT=${DRAFT:-5}
+CTXLEN=${CTXLEN:-270336}
+CHUNK=${CHUNK:-8192}
+MAXPRE=${MAXPRE:-16384}
+EXTRA=${EXTRA:-}
+if [ "$RESTART" = "yes" ]; then RP="--restart unless-stopped"; else RP="--restart no"; fi
+
+echo "[deploy] removing old container (if any)"
+# rm -f times out on big GPU containers on this daemon; retry until really gone
+for i in $(seq 1 45); do
+  CID=$(docker ps -a --filter name=glm53-nvfp4 -q)
+  [ -z "$CID" ] && break
+  docker rm -f glm53-nvfp4 >/dev/null 2>&1 || true
+  sleep 2
+done
+
+echo "[deploy] waiting for VRAM drain"
+for i in $(seq 1 45); do
+  used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
+  [ "$used" -lt 2000 ] && break
+  sleep 2
+done
+used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
+echo "[deploy] VRAM now: ${used} MiB total"
+
+echo "[deploy] starting: TP8 EAGLE ${STEPS}/${TOPK}/${DRAFT} memfrac=${MEMFRAC} extra='${EXTRA}'"
+docker run -d --name glm53-nvfp4 --gpus all --shm-size 64g --ipc=host $RP \
+  -p 30000:30000 -v /data/hf_models:/data/hf_models \
+  lmsysorg/sglang:nightly-dev-20260828-daf63171 \
+  python3 -m sglang.launch_server \
+  --model-path /data/hf_models/GLM-5.3-NVFP4 --tp 8 \
+  --mem-fraction-static $MEMFRAC --max-running-requests 16 \
+  --chunked-prefill-size $CHUNK --max-prefill-tokens $MAXPRE \
+  --disable-shared-experts-fusion --moe-runner-backend flashinfer_cutlass --disable-flashinfer-autotune \
+  --speculative-algorithm EAGLE --speculative-num-steps $STEPS --speculative-eagle-topk $TOPK --speculative-num-draft-tokens $DRAFT \
+  --kv-cache-dtype fp8_e4m3 --enable-hierarchical-cache --hicache-ratio 3 \
+  --cuda-graph-max-bs-decode 8 --cuda-graph-bs-decode 1 2 3 4 6 8 --cuda-graph-max-bs-prefill 8 \
+  --context-length $CTXLEN --reasoning-parser glm45 --tool-call-parser glm47 \
+  --host 0.0.0.0 --port 30000 $EXTRA
+
+echo "[deploy] container started; poll: docker logs -f glm53-nvfp4"

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/deploy_glm53_607_exp.sh

@@ -0,0 +1,70 @@
+#!/bin/bash
+# deploy_glm53_607_exp.sh — GLM-5.3-NVFP4 TP8 + EAGLE experiment deploy for 174.1.60.7
+# EXACT copy of deploy_glm53_605.sh (production standard, 8-machine md5 fcd9109b)
+# with only this header changed. Campaign: scenario-1 optimization (i128k/i64k,
+# o512, cc1-4). Do NOT sync this file to other machines.
+#
+# Env overrides:
+#   MEMFRAC (0.90)  STEPS (4)  TOPK (1)  DRAFT (5)  CTXLEN (270336)
+#   CHUNK (8192)  MAXPRE (16384)  RESTART (no|yes)  EXTRA ("")
+#   CAR_PATCH (1)  inject custom-AR 1stage patch (E7b winner, 2026-09-07):
+#     small ARs <=8MB go one-shot kernel instead of NCCL RING_LL on 8-way PCIe.
+#     Patch files at /root/patches/ (build: bash /root/prep_car_patch.sh on a
+#     running stock container). Patch dies with the container; image untouched.
+#     Set CAR_PATCH=0 to deploy stock.
+set -e
+MEMFRAC=${MEMFRAC:-0.90}
+STEPS=${STEPS:-4}
+TOPK=${TOPK:-1}
+DRAFT=${DRAFT:-5}
+CTXLEN=${CTXLEN:-270336}
+CHUNK=${CHUNK:-8192}
+MAXPRE=${MAXPRE:-16384}
+EXTRA=${EXTRA:-}
+if [ "$RESTART" = "yes" ]; then RP="--restart unless-stopped"; else RP="--restart no"; fi
+
+echo "[deploy] removing old container (if any)"
+# rm -f times out on big GPU containers on this daemon; retry until really gone
+for i in $(seq 1 45); do
+  CID=$(docker ps -a --filter name=glm53-nvfp4 -q)
+  [ -z "$CID" ] && break
+  docker rm -f glm53-nvfp4 >/dev/null 2>&1 || true
+  sleep 2
+done
+
+echo "[deploy] waiting for VRAM drain"
+for i in $(seq 1 45); do
+  used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
+  [ "$used" -lt 2000 ] && break
+  sleep 2
+done
+used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
+echo "[deploy] VRAM now: ${used} MiB total"
+
+echo "[deploy] starting: TP8 EAGLE ${STEPS}/${TOPK}/${DRAFT} memfrac=${MEMFRAC} chunk=${CHUNK} extra='${EXTRA}'"
+docker run -d --name glm53-nvfp4 --gpus all --shm-size 64g --ipc=host $RP \
+  -p 30000:30000 -v /data/hf_models:/data/hf_models \
+  lmsysorg/sglang:nightly-dev-20260828-daf63171 \
+  python3 -m sglang.launch_server \
+  --model-path /data/hf_models/GLM-5.3-NVFP4 --tp 8 \
+  --mem-fraction-static $MEMFRAC --max-running-requests 16 \
+  --chunked-prefill-size $CHUNK --max-prefill-tokens $MAXPRE \
+  --disable-shared-experts-fusion --moe-runner-backend flashinfer_cutlass --disable-flashinfer-autotune \
+  --speculative-algorithm EAGLE --speculative-num-steps $STEPS --speculative-eagle-topk $TOPK --speculative-num-draft-tokens $DRAFT \
+  --kv-cache-dtype fp8_e4m3 --enable-hierarchical-cache --hicache-ratio 3 \
+  --cuda-graph-max-bs-decode 8 --cuda-graph-bs-decode 1 2 3 4 6 8 --cuda-graph-max-bs-prefill 8 \
+  --context-length $CTXLEN --reasoning-parser glm45 --tool-call-parser glm47 \
+  --host 0.0.0.0 --port 30000 $EXTRA
+
+echo "[deploy] container started; poll: docker logs -f glm53-nvfp4"
+
+# --- CAR 1stage patch injection (E7b winner) ---
+if [ "${CAR_PATCH:-1}" != "0" ] && [ -f /root/patches/custom_all_reduce.py ]; then
+  echo "[deploy] CAR_PATCH: injecting custom-AR 1stage patch"
+  docker stop -t 20 glm53-nvfp4 >/dev/null 2>&1 || true
+  DPATH=/sgl-workspace/sglang/python/sglang/srt/distributed/device_communicators
+  docker cp /root/patches/custom_all_reduce.py glm53-nvfp4:$DPATH/custom_all_reduce.py
+  docker cp /root/patches/custom_all_reduce_utils.py glm53-nvfp4:$DPATH/custom_all_reduce_utils.py
+  docker start glm53-nvfp4
+  echo "[deploy] CAR_PATCH injected, container restarted; poll health as usual"
+fi

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/deploy_glm53_optimal.sh

@@ -0,0 +1,62 @@
+#!/bin/bash
+# ============================================================
+# GLM-5.3 最优部署方案（6000D 8卡，TP4 PP2 + IndexCache freq=4）
+# 2026-09-07
+#
+#   - 基线配置：TP4 PP2 + cps16k + mem0.85（131.6 tok/s 吞吐基线）
+#   - IndexCache（index_topk_freq=4）：层轴索引复用，省 75% indexer
+#     16K 场景无损失；128K 长上下文并发 1.35-1.47× 提速
+#   - 禁 radix cache；禁投机解码（PP2 与投机框架不兼容，已实测）
+#
+# 用法：bash deploy_glm53_optimal.sh
+#   （60.7 试用版：增加显存排空等待 + 就绪等待加长至 20 分钟）
+# ============================================================
+set -uo pipefail
+
+CONTAINER="glm53-nvfp4"
+IMAGE="lmsysorg/sglang:nightly-dev-20260828-daf63171"
+MODEL="/data/hf_models/GLM-5.3-NVFP4"
+PORT=30000
+TP=4; PP=2; MEM=0.85; MRR=48; CPS=16384
+
+docker rm -f ${CONTAINER} 2>/dev/null || true
+
+# docker rm -f 后显存释放滞后数分钟，不等会把新容器 KV 池压小
+for i in $(seq 1 60); do
+  used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
+  if [ "$used" -lt 1500 ]; then echo "[deploy] drained: ${used} MiB"; break; fi
+  echo "[deploy] drain wait ${i}: ${used} MiB"
+  sleep 10
+done
+
+docker run -d --name ${CONTAINER} --gpus all --shm-size 64g --ipc=host \
+  --restart unless-stopped \
+  -p ${PORT}:${PORT} \
+  -v /data/hf_models:/data/hf_models \
+  ${IMAGE} \
+  python3 -m sglang.launch_server \
+    --model-path ${MODEL} \
+    --tp-size ${TP} --pp-size ${PP} \
+    --mem-fraction-static ${MEM} \
+    --max-running-requests ${MRR} \
+    --disable-radix-cache \
+    --disable-shared-experts-fusion \
+    --moe-runner-backend flashinfer_cutlass \
+    --disable-flashinfer-autotune \
+    --disable-custom-all-reduce \
+    --chunked-prefill-size ${CPS} \
+    --host 0.0.0.0 --port ${PORT} \
+    --json-model-override-args '{"index_topk_freq": 4}'
+
+echo "容器已启动，等待就绪..."
+for i in $(seq 1 120); do
+  code=$(curl -s -o /dev/null -w '%{http_code}' http://localhost:${PORT}/health 2>/dev/null)
+  if [ "$code" = "200" ]; then
+    echo "READY after ~$((i*10))s"
+    docker ps --filter name=${CONTAINER} --format '{{.Names}} {{.Status}}'
+    echo "override args: $(docker inspect ${CONTAINER} --format '{{.Config.Cmd}}' | grep -o 'index_topk_freq[^,}]*')"
+    exit 0
+  fi
+  sleep 10
+done
+echo "TIMEOUT"; exit 1

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/deploy_glm53_optimal_s1.sh

@@ -0,0 +1,59 @@
+#!/bin/bash
+# ============================================================
+# TP4 PP2 + IndexCache（场景一轮：启用 radix cache）
+# 与 deploy_glm53_optimal.sh 唯一差异：去掉 --disable-radix-cache
+# ——场景一前提是 90% 前缀命中，禁 radix 则命中率为 0 无从对比
+# 其余参数原样：TP4 PP2 / mem0.85 / mrr48 / cps16384 / index_topk_freq=4 / 无投机
+# ============================================================
+set -uo pipefail
+
+CONTAINER="glm53-nvfp4"
+IMAGE="lmsysorg/sglang:nightly-dev-20260828-daf63171"
+MODEL="/data/hf_models/GLM-5.3-NVFP4"
+PORT=30000
+TP=4; PP=2; MEM=0.85; MRR=48; CPS=16384
+
+docker rm -f ${CONTAINER} 2>/dev/null || true
+# 补刀循环：docker rm -f 异步滞留，等容器对象彻底消失
+for i in $(seq 1 40); do
+  n=$(docker ps -a --filter name=${CONTAINER} -q | wc -l)
+  if [ "$n" = "0" ]; then echo "[deploy] container object gone"; break; fi
+  sleep 5
+done
+# 排空显存
+for i in $(seq 1 60); do
+  used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
+  if [ "$used" -lt 1500 ]; then echo "[deploy] drained: ${used} MiB"; break; fi
+  echo "[deploy] drain wait ${i}: ${used} MiB"
+  sleep 10
+done
+
+docker run -d --name ${CONTAINER} --gpus all --shm-size 64g --ipc=host \
+  --restart unless-stopped \
+  -p ${PORT}:${PORT} \
+  -v /data/hf_models:/data/hf_models \
+  ${IMAGE} \
+  python3 -m sglang.launch_server \
+    --model-path ${MODEL} \
+    --tp-size ${TP} --pp-size ${PP} \
+    --mem-fraction-static ${MEM} \
+    --max-running-requests ${MRR} \
+    --disable-shared-experts-fusion \
+    --moe-runner-backend flashinfer_cutlass \
+    --disable-flashinfer-autotune \
+    --disable-custom-all-reduce \
+    --chunked-prefill-size ${CPS} \
+    --host 0.0.0.0 --port ${PORT} \
+    --json-model-override-args '{"index_topk_freq": 4}'
+
+echo "容器已启动，等待就绪..."
+for i in $(seq 1 120); do
+  code=$(curl -s -o /dev/null -w '%{http_code}' http://localhost:${PORT}/health 2>/dev/null)
+  if [ "$code" = "200" ]; then
+    echo "READY after ~$((i*10))s"
+    docker ps --filter name=${CONTAINER} --format '{{.Names}} {{.Status}}'
+    exit 0
+  fi
+  sleep 10
+done
+echo "TIMEOUT"; exit 1

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/fetch_pg19_v2.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+"""PG19 corpus fetcher v2: list the GCS bucket (correct path = train/{id}.txt),
+pick the largest books (skipping >20MB multi-volume tomes) until ~80MB of text,
+download in parallel."""
+import os
+import re
+import sys
+from concurrent.futures import ThreadPoolExecutor
+
+import requests
+
+BASE = "https://storage.googleapis.com/deepmind-gutenberg"
+D = "/root/bench_corpus"
+TARGET_TOTAL = 80_000_000
+MAX_BOOK = 20_000_000
+MAX_BOOKS = 80
+
+s = requests.Session()
+
+entries = []
+token = None
+while True:
+    params = {"list-type": "2", "prefix": "train/", "max-keys": "1000"}
+    if token:
+        params["continuation-token"] = token
+    r = s.get(BASE, params=params, timeout=30)
+    txt = r.text
+    for m in re.finditer(r"<Contents><Key>([^<]+)</Key>.*?<Size>(\d+)</Size>", txt, re.DOTALL):
+        entries.append((m.group(1), int(m.group(2)))
+                       )
+    if "<IsTruncated>true</IsTruncated>" not in txt:
+        break
+    mt = re.search(r"<NextContinuationToken>([^<]+)</NextContinuationToken>", txt)
+    if not mt:
+        break
+    token = mt.group(1)
+    print(f"[list] {len(entries)} objects so far", flush=True)
+
+entries.sort(key=lambda x: -x[1])
+selected = []
+total = 0
+for key, size in entries:
+    if size > MAX_BOOK or size < 500_000:
+        continue
+    selected.append((key, size))
+    total += size
+    if total >= TARGET_TOTAL or len(selected) >= MAX_BOOKS:
+        break
+print(f"[select] {len(selected)} books, {total/1e6:.1f} MB", flush=True)
+
+os.makedirs(f"{D}/books", exist_ok=True)
+
+
+def dl(ks):
+    key, size = ks
+    fn = f"{D}/books/{key.split('/')[-1]}"
+    r = requests.get(f"{BASE}/{key}", timeout=300)
+    with open(fn, "wb") as f:
+        f.write(r.content)
+    return key, len(r.content)
+
+
+with ThreadPoolExecutor(max_workers=8) as ex:
+    for key, got in ex.map(dl, selected):
+        print(f"[dl] {key} {got}", flush=True)
+
+real = sum(os.path.getsize(f"{D}/books/{f}") for f in os.listdir(f"{D}/books"))
+print(f"[done] files={len(os.listdir(f'{D}/books'))} total_bytes={real}", flush=True)

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/quality_gate_605.sh

@@ -0,0 +1,71 @@
+#!/bin/bash
+# /data/hf_models/GLM-5.3-NVFP4 质量门: GSM8K 5题 + 中文推理 + tool call
+# 用法: bash quality_gate_flash.sh  (需服务已在 127.0.0.1:30000 就绪)
+PORT=30000
+BASE=http://127.0.0.1:$PORT
+PASS=0; FAIL=0
+
+ask() {  # $1=question $2=expect_number(optional) $3=label
+  local q="$1" expect="$2" label="$3"
+  local out
+  out=$(curl -s -m 300 $BASE/v1/chat/completions -H 'Content-Type: application/json' -d "{
+    \"model\": \"/data/hf_models/GLM-5.3-NVFP4\",
+    \"messages\": [{\"role\": \"user\", \"content\": $(python3 -c "import json,sys; print(json.dumps(sys.argv[1]))" "$q")}],
+    \"temperature\": 0, \"max_tokens\": 2048
+  }")
+  local content finish
+  content=$(echo "$out" | python3 -c "import json,sys; d=json.load(sys.stdin); print(d['choices'][0]['message'].get('content') or '')" 2>/dev/null)
+  finish=$(echo "$out" | python3 -c "import json,sys; d=json.load(sys.stdin); print(d['choices'][0].get('finish_reason'))" 2>/dev/null)
+  if [ -z "$content" ] && [ "$finish" != "tool_calls" ]; then
+    echo "[FAIL] $label: 无输出 finish=$finish"; FAIL=$((FAIL+1)); return
+  fi
+  if [ -n "$expect" ]; then
+    if echo "$content" | grep -qE "(^|[^0-9])$expect([^0-9]|$)"; then
+      echo "[PASS] $label (含 $expect)"; PASS=$((PASS+1))
+    else
+      echo "[FAIL] $label: 期望含 $expect, 实际输出尾部: $(echo "$content" | tail -c 300)"; FAIL=$((FAIL+1))
+    fi
+  else
+    echo "[INFO] $label: $(echo "$content" | head -c 200)"
+  fi
+}
+
+echo "===== GSM8K x5 (temperature=0) ====="
+ask "Natalia sold clips to 48 of her friends in April, and then she sold half as many clips in May. How many clips did Natalia sell altogether in April and May?" "72" "GSM8K-1"
+ask "A robe takes 2 bolts of blue fiber and half that much white fiber. How many bolts in total does it take?" "3" "GSM8K-2"
+ask "Betty has half the money that Linda has. If Linda has 120 dollars, how many dollars does Betty have?" "60" "GSM8K-3"
+ask "A waiter had 9 tables he was waiting on, with 7 women and 3 men at each table. After an hour, all the men left. How many customers remained?" "63" "GSM8K-4"
+ask "Weng earns 12 dollars per hour babysitting. Yesterday she babysat for 50 minutes. How much did she earn?" "10" "GSM8K-5"
+
+echo "===== 中文推理 (鸡兔同笼) ====="
+ask "一个笼子里有鸡和兔，从上面数有35个头，从下面数有94只脚。请问鸡和兔各有几只？请给出推理过程。" "23" "中文推理-鸡数(兔应为12)"
+echo "(兔数检查: 上面输出应同时含 12)"
+
+echo "===== Tool Call ====="
+tc_out=$(curl -s -m 120 $BASE/v1/chat/completions -H 'Content-Type: application/json' -d '{
+  "model": "/data/hf_models/GLM-5.3-NVFP4",
+  "messages": [{"role": "user", "content": "请帮我查一下北京今天的天气"}],
+  "tools": [{"type": "function", "function": {
+    "name": "get_weather",
+    "description": "查询指定城市的天气",
+    "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}}],
+  "tool_choice": "auto", "temperature": 0, "max_tokens": 512
+}')
+tc_calls=$(echo "$tc_out" | python3 -c "
+import json,sys
+d=json.load(sys.stdin)
+msg=d['choices'][0]['message']
+calls=msg.get('tool_calls') or []
+if calls:
+    c=calls[0]
+    print(c['function']['name'], c['function']['arguments'])
+else:
+    print('NO_TOOL_CALL', (msg.get('content') or '')[:150])
+" 2>/dev/null)
+if echo "$tc_calls" | grep -q "get_weather.*北京"; then
+  echo "[PASS] tool call: $tc_calls"; PASS=$((PASS+1))
+else
+  echo "[FAIL] tool call: $tc_calls"; FAIL=$((FAIL+1))
+fi
+
+echo "===== 结果: PASS=$PASS FAIL=$FAIL ====="

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/run_s1_corpus.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+# Scenario 1, real-corpus (PG19) re-test of method A (TP8+EAGLE, live prod config).
+# 8 points: {128k, 64k} x cc{1,2,3,4}, nreq 8, shared-frac 0.9, run-ids 9301-9308.
+# Each run-id maps to a fixed non-overlapping corpus window (see bench_corpus.py).
+#
+# Repo 版与 60.7 原版（md5 5b2534f4c2e2f5760c453abb00dc32a2）唯一差异：
+#   bench 脚本按本目录解析，语料/日志目录可用 CORPUS / LOG_DIR 覆盖（默认 /root，
+#   即服务器原布局，行为不变）。详见实验 README「文件对照」。
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+CORPUS="${CORPUS:-/root/corpus_ids.json}"
+LOG_DIR="${LOG_DIR:-/root/bench_logs}"
+mkdir -p "$LOG_DIR"
+for spec in "131072 1 9301" "131072 2 9302" "131072 3 9303" "131072 4 9304" \
+            "65536 1 9305" "65536 2 9306" "65536 3 9307" "65536 4 9308"; do
+  set -- $spec
+  IL=$1; CC=$2; RID=$3
+  echo "=== input=$IL cc=$CC run=$RID start $(date +%T) ===" >> "$LOG_DIR/r1_runner.log"
+  python3 "${SCRIPT_DIR}/bench_corpus.py" --corpus "$CORPUS" \
+    --input-len $IL --concurrency $CC --num-requests 8 --run-id $RID \
+    --shared-frac 0.9 --output-len 512 > "$LOG_DIR/r1_run${RID}.log" 2>&1
+  echo "=== run=$RID done rc=$? $(date +%T) ===" >> "$LOG_DIR/r1_runner.log"
+done
+echo "ALL_DONE $(date +%T)" >> "$LOG_DIR/r1_runner.log"

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/run_s2_corpus.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+# Scenario 2, real-corpus (PG19) re-test of method A (TP8+EAGLE, high-cc variant:
+# max-running 32, decode graphs bs 4/8/12/16). 3 points, 16k fully-unique prompts.
+# run-ids map to fixed corpus windows (see bench_corpus.py pool C).
+#
+# Repo 版与 60.7 原版（md5 8965a71edef6b1cf406496f0362a5d05）唯一差异：
+#   bench 脚本按本目录解析，语料/日志目录可用 CORPUS / LOG_DIR 覆盖（默认 /root，
+#   即服务器原布局，行为不变）。详见实验 README「文件对照」。
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+CORPUS="${CORPUS:-/root/corpus_ids.json}"
+LOG_DIR="${LOG_DIR:-/root/bench_logs}"
+mkdir -p "$LOG_DIR"
+for spec in "8 16 9311" "16 32 9312" "32 32 9313"; do
+  set -- $spec
+  CC=$1; NR=$2; RID=$3
+  echo "=== cc=$CC nreq=$NR run=$RID start $(date +%T) ===" >> "$LOG_DIR/r2_runner.log"
+  python3 "${SCRIPT_DIR}/bench_corpus.py" --corpus "$CORPUS" \
+    --input-len 16384 --concurrency $CC --num-requests $NR --run-id $RID \
+    --shared-frac 0 --output-len 512 > "$LOG_DIR/r2_run${RID}.log" 2>&1
+  echo "=== run=$RID done rc=$? $(date +%T) ===" >> "$LOG_DIR/r2_runner.log"
+done
+echo "ALL_DONE $(date +%T)" >> "$LOG_DIR/r2_runner.log"

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/summarize_bench.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+"""Compact table from bench_corpus.py r1/r2 logs."""
+import glob
+import json
+import sys
+
+pat = sys.argv[1] if len(sys.argv) > 1 else "/root/bench_logs/r1_run*.log"
+rows = []
+for f in sorted(glob.glob(pat)):
+    txt = open(f).read()
+    if "===== SUMMARY =====" not in txt:
+        rows.append((f, "NO_SUMMARY"))
+        continue
+    d = json.loads(txt.split("===== SUMMARY =====")[1])
+    rows.append((f, (
+        f"run={d['run_id']} il={d['input_len']} cc={d['concurrency']} ok={d['ok']}/{d['num_requests']} "
+        f"wall={d['wall_s']} out_tps={d['output_throughput_tok_s']} "
+        f"in_tps={d['input_throughput_tok_s']} "
+        f"ttft_p50={d['ttft_s']['p50']} ttft_mean={d['ttft_s']['mean']} "
+        f"tpot_p50={d['tpot_s']['p50']} tpot_mean={d['tpot_s']['mean']} "
+        f"e2e_p50={d['e2e_s']['p50']} dec_p50={d['per_req_decode_tok_s']['p50']} "
+        f"accept={d['spec_accept_length_mean']} retr={d['retractions_total']} "
+        f"hit={d['cache_hit_from_logs'].get('hit_rate')}"
+    )))
+for f, r in rows:
+    print(f.split("/")[-1], r)

experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/tokenize_corpus.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+"""Tokenize PG19 books with the served model's tokenizer (run INSIDE container).
+
+Reads /tmp/bench_corpus/books/*.txt (docker cp'd in), writes compact JSON to
+stdout: {"ids": [flat token ids], "books": [[start, end), ...]}
+Progress goes to stderr. Host-side usage:
+  docker exec -i glm53-nvfp4 python3 - < tokenize_corpus.py \
+      > /root/corpus_ids.json 2> /root/tokenize_progress.log
+"""
+import glob
+import json
+import sys
+
+from transformers import AutoTokenizer
+
+MODEL = "/data/hf_models/GLM-5.3-NVFP4"
+
+tok = AutoTokenizer.from_pretrained(MODEL)
+ids_all = []
+books = []
+files = sorted(glob.glob("/tmp/bench_corpus/books/*.txt"))
+for f in files:
+    with open(f, encoding="utf-8", errors="ignore") as fh:
+        text = fh.read()
+    enc = tok(text, add_special_tokens=False)["input_ids"]
+    books.append([len(ids_all), len(ids_all) + len(enc)])
+    ids_all.extend(enc)
+    print(f"[tokenize] {f} -> {len(enc)} tokens", file=sys.stderr)
+
+print(f"[tokenize] total files={len(files)} total_tokens={len(ids_all)}", file=sys.stderr)
+json.dump({"ids": ids_all, "books": books}, sys.stdout, separators=(",", ":"))