6000D 双机 DeepSeek-V4-Pro 推理优化计划
-适用环境：
+174.1.51.5 + 174.1.51.7，每台 8 张 RTX PRO 6000 Blackwell Server Edition
当前部署：DeepSeek-V4-Pro，16 张 GPU 组成一个完整实例
当前约束：模型暂时只能使用全部 16 张 GPU，无法额外复制一套模型进行 PD 分离
计划版本：2026-07-31 16:46:05 CST适用环境：
174.1.51.5 + 174.1.51.7，每台 8 张 RTX PRO 6000 Blackwell Server Edition
当前部署：DeepSeek-V4-Pro，16 张 GPU 组成一个完整实例
当前约束：模型暂时只能使用全部 16 张 GPU，无法额外复制一套模型进行 PD 分离
计划版本：2026-08-01 02:45:00 CST
当前执行状态与阶段档案
| DeepSeek-V4-Pro / 双机 Pro6000D / SGLang RDMA 需求建模与并发拐点 | +Phase 2.5 代码已实现；等待正式 Run，以 64K Scout 拟合 bytes/token、带宽平台和 400G 所需 Token TPS | +阶段尚未完成，按门禁暂不创建实验/代码 HTML | +
阶段档案生成门禁：Phase 尚未产出正式阶段结果、完成汇总和汇报确认前， @@ -785,6 +790,21 @@ Phase 2 回答“哪个硬件/Host/通信资源在什么 Case 中升高，以及 Phase 3 只捕获短 TP16 Timeline，回答具体 Kernel、Scheduler gap、Collective、 Rank 同步及计算/通信重叠，不重复 Phase 2 的长时间轻量采样。
+6.5 Phase 2.5：RDMA 需求建模与并发拐点
+
+Phase 2 已证明当前代表负载最高约 83.5 Gbit/s/rail，但单个固定 Case 不能回答
+“提高并发后是否还能继续逼近 400G”。Phase 2.5 保持 TP16/EP2 和服务参数不变，
+先用 64K → 1 的 C=1/4/16/32/64 隔离 Prefill，拟合
+rail_gbps ≈ input_tps × bytes_per_input_token_per_rail × 8；
+再自动选择平台前、拐点和最大稳定并发，对 64K → 1K 重复两次确认。
+
+本阶段必须同时给出每 Rail Mean/P95/Max、双 Rail 单向合计、Input/Output TPS、
+GPU 利用率、Rail 不均衡和 RDMA 错误增量。最终结论不是只报告“最大 C”，而是判断
+模型计算吞吐和 RDMA 链路哪个先进入平台，并形成一套换模型、TP/EP 或输入输出形状后
+可重复使用的带宽需求估算方法。正式结果完成前不创建
+phase2_5_exp.html 或 phase2_5_code.html。
+
7. Phase 3：时间线 Profiling（Nsight Systems 为主）
Nsight Systems、PyTorch Profiler 和 NVTX 是三件不同的东西：Nsight Systems
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/.gitignore b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/.gitignore
new file mode 100644
index 0000000..ecba709
--- /dev/null
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/.gitignore
@@ -0,0 +1,3 @@
+results/
+runtime/
+*.log
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/README.md b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/README.md
new file mode 100644
index 0000000..cabf4a7
--- /dev/null
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/README.md
@@ -0,0 +1,35 @@
+# DeepSeek-V4-Pro 双机 SGLang RDMA 需求建模
+
+Phase 2.5 在不改变 TP16/EP2 服务参数的前提下回答三个问题：
+
+1. 64K 输入下，哪个并发是 Token TPS 与 RDMA 带宽的拐点；
+2. `64K -> 1K` 真实请求最多产生多少双 Rail 流量；
+3. 当前部署会先达到 GPU/模型吞吐平台，还是先打满 400G rail。
+
+唯一入口：
+
+```bash
+cd /data/hzy/sskj/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling
+
+# 只展开命令，不启动服务
+DRY_RUN=1 RUN_ID=dsv4pro-phase2_5-dryrun-$(date +%Y%m%d-%H%M%S) \
+ bash run_rdma_demand_modeling.sh all
+
+# 正式运行
+RUN_ID=dsv4pro-phase2_5-$(date +%Y%m%d-%H%M%S)
+tmux new-session -d -s dsv4pro-phase2_5 \
+ "RUN_ID=${RUN_ID} bash run_rdma_demand_modeling.sh all \
+ 2>&1 | tee /data/hzy/${RUN_ID}.log"
+tmux attach -t dsv4pro-phase2_5
+```
+
+脚本先跑 `64K -> 1` 的 C=1/4/16/32/64 探路点，再自动选择三个代表并发，
+对 `64K -> 1K` 重复两次。服务、benchmark 和 HCA 采集复用已经验证的
+Phase 1/2 实现；本目录只负责工作负载编排与带宽模型拟合。
+
+最终高价值结果：
+
+- `rdma_case_metrics.csv`：每个 Case 的 TPS、每 Rail/双 Rail 带宽和 bytes/token；
+- `rdma_demand_model.json`：拟合参数、并发拐点和理论需求；
+- `rdma_demand_report.md`：可直接用于阶段汇报的结论；
+- `recommendation.env`：Scout 自动选择的 Confirm 并发。
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/config.env b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/config.env
new file mode 100644
index 0000000..62b5a8f
--- /dev/null
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/config.env
@@ -0,0 +1,36 @@
+# DeepSeek-V4-Pro two-node SGLang RDMA demand modeling.
+
+EXPERIMENT="dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling"
+
+PHASE1_DIR="${PHASE1_DIR:-$(cd "$(dirname "${BASH_SOURCE[0]}")/../dsv4pro_pro6000d_2node_sglang_tp16_quick_map" && pwd)}"
+PHASE2_DIR="${PHASE2_DIR:-$(cd "$(dirname "${BASH_SOURCE[0]}")/../dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution" && pwd)}"
+PHASE2_ENTRY="${PHASE2_ENTRY:-${PHASE2_DIR}/run_hardware_contention_attribution.sh}"
+
+# Workload policy. Scout isolates Prefill; confirm preserves the requested 64K -> 1K shape.
+ISL="${ISL:-65536}"
+SCOUT_OSL="${SCOUT_OSL:-1}"
+CONFIRM_OSL="${CONFIRM_OSL:-1024}"
+SCOUT_CONCURRENCIES="${SCOUT_CONCURRENCIES:-1 4 16 32 64}"
+DRY_RUN_CONFIRM_CONCURRENCIES="${DRY_RUN_CONFIRM_CONCURRENCIES:-16 32 64}"
+SCOUT_REPETITIONS="${SCOUT_REPETITIONS:-1}"
+CONFIRM_REPETITIONS="${CONFIRM_REPETITIONS:-2}"
+
+# A point is a plateau candidate when both bandwidth and Input TPS gains stay below this value.
+PLATEAU_GAIN_PCT="${PLATEAU_GAIN_PCT:-5}"
+TARGET_RAIL_GBPS="${TARGET_RAIL_GBPS:-400}"
+PRACTICAL_RAIL_GBPS="${PRACTICAL_RAIL_GBPS:-360}"
+
+# Reuse the verified Phase 2 collectors but skip its one-time communication microbenchmark.
+SAMPLE_INTERVAL_S="${SAMPLE_INTERVAL_S:-1}"
+IDLE_BASELINE_S="${IDLE_BASELINE_S:-5}"
+POST_RUN_COOLDOWN_S="${POST_RUN_COOLDOWN_S:-5}"
+CASE_COOLDOWN_S="${CASE_COOLDOWN_S:-3}"
+COLLECTOR_MAX_SECONDS="${COLLECTOR_MAX_SECONDS:-14400}"
+SCENARIO_TIMEOUT_S="${SCENARIO_TIMEOUT_S:-7200}"
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "${SCRIPT_DIR}/../../.." && pwd)"
+RESULT_BASE="${RESULT_BASE:-${SCRIPT_DIR}/results}"
+RUNTIME_BASE="${RUNTIME_BASE:-${SCRIPT_DIR}/runtime}"
+
+DRY_RUN="${DRY_RUN:-0}"
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/rdma_demand_model.py b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/rdma_demand_model.py
new file mode 100755
index 0000000..cee03db
--- /dev/null
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/rdma_demand_model.py
@@ -0,0 +1,563 @@
+#!/usr/bin/env python3
+"""Fit model-serving RDMA demand from aligned benchmark and HCA counters."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import math
+import statistics
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Iterable
+
+
+def read_csv(path: Path) -> list[dict[str, str]]:
+ if not path.is_file():
+ return []
+ with path.open(encoding="utf-8", newline="") as handle:
+ return list(csv.DictReader(handle))
+
+
+def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ if not rows:
+ path.write_text("", encoding="utf-8")
+ return
+ fields: list[str] = []
+ for row in rows:
+ for field in row:
+ if field not in fields:
+ fields.append(field)
+ with path.open("w", encoding="utf-8", newline="") as handle:
+ writer = csv.DictWriter(handle, fieldnames=fields)
+ writer.writeheader()
+ writer.writerows(rows)
+
+
+def number(value: Any) -> float | None:
+ if value in (None, ""):
+ return None
+ try:
+ parsed = float(value)
+ except (TypeError, ValueError):
+ return None
+ return parsed if math.isfinite(parsed) else None
+
+
+def integer(value: Any) -> int:
+ parsed = number(value)
+ return int(parsed) if parsed is not None else 0
+
+
+def percentile(values: Iterable[float], fraction: float) -> float | None:
+ ordered = sorted(values)
+ if not ordered:
+ return None
+ position = (len(ordered) - 1) * fraction
+ lower = math.floor(position)
+ upper = math.ceil(position)
+ if lower == upper:
+ return ordered[lower]
+ return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower)
+
+
+def mean(values: Iterable[float | None]) -> float | None:
+ present = [value for value in values if value is not None]
+ return statistics.fmean(present) if present else None
+
+
+def fmt(value: Any, digits: int = 2) -> str:
+ parsed = number(value)
+ return "-" if parsed is None else f"{parsed:.{digits}f}"
+
+
+def parse_iso_ns(value: str) -> int:
+ return int(datetime.fromisoformat(value).timestamp() * 1_000_000_000)
+
+
+@dataclass(frozen=True)
+class CaseKey:
+ case_id: str
+ repetition: int
+
+
+def load_bench(result_dir: Path) -> dict[CaseKey, dict[str, str]]:
+ rows = read_csv(result_dir / "bench_summary.csv")
+ result: dict[CaseKey, dict[str, str]] = {}
+ for row in rows:
+ key = CaseKey(row.get("case_id", ""), integer(row.get("repetition")))
+ if key.case_id:
+ result[key] = row
+ return result
+
+
+def load_windows(result_dir: Path) -> dict[CaseKey, dict[str, str]]:
+ rows = read_csv(result_dir / "case_windows.csv")
+ result: dict[CaseKey, dict[str, str]] = {}
+ for row in rows:
+ key = CaseKey(row.get("case_id", ""), integer(row.get("repetition")))
+ if key.case_id:
+ result[key] = row
+ return result
+
+
+def load_case_rdma(
+ result_dir: Path,
+) -> dict[CaseKey, list[dict[str, str]]]:
+ grouped: dict[CaseKey, list[dict[str, str]]] = {}
+ for row in read_csv(result_dir / "case_rdma_summary.csv"):
+ key = CaseKey(row.get("case_id", ""), integer(row.get("repetition")))
+ grouped.setdefault(key, []).append(row)
+ return grouped
+
+
+def interval_rates(
+ rows: list[dict[str, str]], started_ns: int, ended_ns: int
+) -> tuple[list[float], list[float]]:
+ selected = []
+ for row in rows:
+ timestamp = integer(row.get("wall_time_ns"))
+ if started_ns <= timestamp <= ended_ns:
+ selected.append(row)
+ selected.sort(key=lambda row: integer(row.get("wall_time_ns")))
+
+ xmit_rates: list[float] = []
+ rcv_rates: list[float] = []
+ for previous, current in zip(selected, selected[1:]):
+ delta_ns = integer(current.get("wall_time_ns")) - integer(
+ previous.get("wall_time_ns")
+ )
+ if delta_ns <= 0:
+ continue
+ duration_s = delta_ns / 1e9
+ for field, output in (
+ ("port_xmit_data", xmit_rates),
+ ("port_rcv_data", rcv_rates),
+ ):
+ first = number(previous.get(field))
+ last = number(current.get(field))
+ if first is None or last is None or last < first:
+ continue
+ # IB port data counters use 4-octet units.
+ output.append((last - first) * 4 * 8 / duration_s / 1e9)
+ return xmit_rates, rcv_rates
+
+
+def raw_rdma_by_node_hca(
+ result_dir: Path,
+) -> dict[tuple[str, str], list[dict[str, str]]]:
+ grouped: dict[tuple[str, str], list[dict[str, str]]] = {}
+ for node in ("head", "worker"):
+ for row in read_csv(result_dir / node / "rdma.csv"):
+ hca = row.get("hca", "")
+ if hca:
+ grouped.setdefault((node, hca), []).append(row)
+ return grouped
+
+
+def load_gpu_util(result_dir: Path) -> dict[CaseKey, float]:
+ grouped: dict[CaseKey, list[float]] = {}
+ for row in read_csv(result_dir / "case_gpu_node_summary.csv"):
+ key = CaseKey(row.get("case_id", ""), integer(row.get("repetition")))
+ value = number(row.get("gpu_util_pct_mean"))
+ if value is not None:
+ grouped.setdefault(key, []).append(value)
+ return {key: statistics.fmean(values) for key, values in grouped.items()}
+
+
+def aggregate_stage(result_dir: Path, stage_name: str) -> list[dict[str, Any]]:
+ bench = load_bench(result_dir)
+ windows = load_windows(result_dir)
+ case_rdma = load_case_rdma(result_dir)
+ raw_rdma = raw_rdma_by_node_hca(result_dir)
+ gpu_util = load_gpu_util(result_dir)
+ output: list[dict[str, Any]] = []
+
+ for key, bench_row in sorted(
+ bench.items(), key=lambda item: (integer(item[1].get("concurrency")), item[0].repetition)
+ ):
+ window = windows.get(key)
+ rdma_rows = case_rdma.get(key, [])
+ if not window or not rdma_rows:
+ continue
+ started_ns = integer(window.get("started_ns")) or parse_iso_ns(
+ window.get("started_at", "")
+ )
+ ended_ns = integer(window.get("ended_ns")) or parse_iso_ns(
+ window.get("ended_at", "")
+ )
+
+ rail_averages: list[float] = []
+ rail_p95s: list[float] = []
+ rail_maxima: list[float] = []
+ node_averages: dict[str, float] = {}
+ xmit_bytes: list[float] = []
+ error_delta = 0.0
+ for rdma_row in rdma_rows:
+ node = rdma_row.get("node", "")
+ hca = rdma_row.get("hca", "")
+ average = number(rdma_row.get("xmit_gbps"))
+ total_bytes = number(rdma_row.get("xmit_bytes"))
+ if average is not None:
+ rail_averages.append(average)
+ node_averages[node] = node_averages.get(node, 0.0) + average
+ if total_bytes is not None:
+ xmit_bytes.append(total_bytes)
+ rates, _ = interval_rates(raw_rdma.get((node, hca), []), started_ns, ended_ns)
+ p95 = percentile(rates, 0.95)
+ if p95 is not None:
+ rail_p95s.append(p95)
+ if rates:
+ rail_maxima.append(max(rates))
+ for field in (
+ "port_xmit_wait_delta",
+ "port_xmit_discards_delta",
+ "port_rcv_errors_delta",
+ "req_transport_retries_exceeded_delta",
+ "req_rnr_retries_exceeded_delta",
+ "roce_adp_retrans_delta",
+ "roce_adp_retrans_to_delta",
+ ):
+ error_delta += number(rdma_row.get(field)) or 0.0
+
+ total_input_tokens = number(bench_row.get("total_input_tokens"))
+ total_output_tokens = number(bench_row.get("total_output_tokens"))
+ total_tokens = (
+ total_input_tokens + total_output_tokens
+ if total_input_tokens is not None and total_output_tokens is not None
+ else None
+ )
+ mean_xmit_bytes = mean(xmit_bytes)
+ bytes_per_input = (
+ mean_xmit_bytes / total_input_tokens
+ if mean_xmit_bytes is not None and total_input_tokens
+ else None
+ )
+ bytes_per_total = (
+ mean_xmit_bytes / total_tokens
+ if mean_xmit_bytes is not None and total_tokens
+ else None
+ )
+ min_rail = min(rail_averages) if rail_averages else None
+ max_rail = max(rail_averages) if rail_averages else None
+ rail_imbalance = (
+ (max_rail - min_rail) / statistics.fmean(rail_averages) * 100
+ if min_rail is not None and max_rail is not None and rail_averages
+ else None
+ )
+ output.append(
+ {
+ "stage": stage_name,
+ "case_id": key.case_id,
+ "repetition": key.repetition,
+ "status": bench_row.get("status", ""),
+ "isl": integer(bench_row.get("isl")),
+ "osl": integer(bench_row.get("osl")),
+ "concurrency": integer(bench_row.get("concurrency")),
+ "completed": integer(bench_row.get("completed")),
+ "duration_s": number(bench_row.get("duration_s")),
+ "input_tps": number(bench_row.get("input_token_throughput")),
+ "output_tps": number(bench_row.get("output_token_throughput")),
+ "total_tps": number(bench_row.get("total_token_throughput")),
+ "ttft_p95_ms": number(bench_row.get("ttft_p95_ms")),
+ "tpot_p95_ms": number(bench_row.get("tpot_p95_ms")),
+ "gpu_util_mean_pct": gpu_util.get(key),
+ "rail_xmit_avg_gbps": mean(rail_averages),
+ "rail_xmit_p95_gbps": mean(rail_p95s),
+ "rail_xmit_max_gbps": max(rail_maxima) if rail_maxima else None,
+ "dual_rail_xmit_avg_gbps": mean(node_averages.values()),
+ "rail_imbalance_pct": rail_imbalance,
+ "xmit_bytes_per_input_token_per_rail": bytes_per_input,
+ "xmit_bytes_per_total_token_per_rail": bytes_per_total,
+ "rdma_error_delta": error_delta,
+ "measurement_started_at": window.get("started_at", ""),
+ "measurement_ended_at": window.get("ended_at", ""),
+ "source_result_dir": str(result_dir),
+ }
+ )
+ return output
+
+
+def average_by_concurrency(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+ grouped: dict[int, list[dict[str, Any]]] = {}
+ for row in rows:
+ if row.get("status") == "COMPLETED":
+ grouped.setdefault(integer(row.get("concurrency")), []).append(row)
+ output = []
+ metrics = (
+ "input_tps",
+ "output_tps",
+ "total_tps",
+ "ttft_p95_ms",
+ "tpot_p95_ms",
+ "gpu_util_mean_pct",
+ "rail_xmit_avg_gbps",
+ "rail_xmit_p95_gbps",
+ "rail_xmit_max_gbps",
+ "dual_rail_xmit_avg_gbps",
+ "rail_imbalance_pct",
+ "xmit_bytes_per_input_token_per_rail",
+ "xmit_bytes_per_total_token_per_rail",
+ "rdma_error_delta",
+ )
+ for concurrency, group in sorted(grouped.items()):
+ item: dict[str, Any] = {
+ "stage": group[0].get("stage"),
+ "concurrency": concurrency,
+ "repetitions": len(group),
+ }
+ for metric in metrics:
+ item[metric] = mean(number(row.get(metric)) for row in group)
+ output.append(item)
+ return output
+
+
+def fit_hyperbola(rows: list[dict[str, Any]], field: str) -> dict[str, float | None]:
+ points = [
+ (float(integer(row.get("concurrency"))), number(row.get(field)))
+ for row in rows
+ if integer(row.get("concurrency")) > 0 and number(row.get(field)) is not None
+ ]
+ if len(points) < 2:
+ return {"asymptote": None, "half_saturation_c": None, "rmse": None}
+
+ best: tuple[float, float, float] | None = None
+ for exponent_step in range(-20, 121):
+ k = 10 ** (exponent_step / 20)
+ factors = [concurrency / (k + concurrency) for concurrency, _ in points]
+ denominator = sum(factor * factor for factor in factors)
+ if denominator == 0:
+ continue
+ asymptote = sum(
+ factor * value for factor, (_, value) in zip(factors, points)
+ ) / denominator
+ errors = [
+ value - asymptote * factor
+ for factor, (_, value) in zip(factors, points)
+ ]
+ rmse = math.sqrt(sum(error * error for error in errors) / len(errors))
+ if best is None or rmse < best[0]:
+ best = (rmse, asymptote, k)
+ assert best is not None
+ return {
+ "asymptote": best[1],
+ "half_saturation_c": best[2],
+ "rmse": best[0],
+ }
+
+
+def linear_slope(rows: list[dict[str, Any]]) -> float | None:
+ points = [
+ (number(row.get("input_tps")), number(row.get("rail_xmit_avg_gbps")))
+ for row in rows
+ ]
+ present = [(x, y) for x, y in points if x is not None and y is not None]
+ denominator = sum(x * x for x, _ in present)
+ return sum(x * y for x, y in present) / denominator if denominator else None
+
+
+def select_concurrencies(
+ scout: list[dict[str, Any]], plateau_gain_pct: float
+) -> tuple[list[int], int | None, int | None]:
+ ordered = sorted(scout, key=lambda row: integer(row.get("concurrency")))
+ if not ordered:
+ return [], None, None
+ plateau_c = None
+ for previous, current in zip(ordered, ordered[1:]):
+ previous_bw = number(previous.get("rail_xmit_avg_gbps")) or 0
+ current_bw = number(current.get("rail_xmit_avg_gbps")) or 0
+ previous_tps = number(previous.get("input_tps")) or 0
+ current_tps = number(current.get("input_tps")) or 0
+ bw_gain = (current_bw / previous_bw - 1) * 100 if previous_bw else math.inf
+ tps_gain = (current_tps / previous_tps - 1) * 100 if previous_tps else math.inf
+ if bw_gain < plateau_gain_pct and tps_gain < plateau_gain_pct:
+ plateau_c = integer(current.get("concurrency"))
+ break
+
+ best = max(
+ ordered,
+ key=lambda row: number(row.get("rail_xmit_avg_gbps")) or -1,
+ )
+ best_c = integer(best.get("concurrency"))
+ cs = [integer(row.get("concurrency")) for row in ordered]
+ anchor = plateau_c or best_c
+ anchor_index = cs.index(anchor)
+ candidates = {anchor, cs[max(0, anchor_index - 1)], cs[-1]}
+ if len(candidates) < 3:
+ candidates.update(cs[-3:])
+ return sorted(candidates), plateau_c, best_c
+
+
+def build_model(
+ scout_rows: list[dict[str, Any]],
+ confirm_rows: list[dict[str, Any]],
+ plateau_gain_pct: float,
+ target_rail_gbps: float,
+ practical_rail_gbps: float,
+) -> dict[str, Any]:
+ scout = average_by_concurrency(scout_rows)
+ confirm = average_by_concurrency(confirm_rows)
+ selected, plateau_c, best_scout_c = select_concurrencies(
+ scout, plateau_gain_pct
+ )
+ slope = linear_slope(scout)
+ bytes_per_token = slope * 1e9 / 8 if slope is not None else None
+ required_target_tps = target_rail_gbps / slope if slope else None
+ required_practical_tps = practical_rail_gbps / slope if slope else None
+ required_dual_400_tps = (target_rail_gbps / 2) / slope if slope else None
+ bandwidth_fit = fit_hyperbola(scout, "rail_xmit_avg_gbps")
+ input_tps_fit = fit_hyperbola(scout, "input_tps")
+ max_confirm = (
+ max(confirm, key=lambda row: number(row.get("rail_xmit_avg_gbps")) or -1)
+ if confirm
+ else None
+ )
+ asymptote = number(bandwidth_fit.get("asymptote"))
+ if asymptote is None:
+ verdict = "INSUFFICIENT_DATA"
+ elif asymptote < practical_rail_gbps:
+ verdict = "COMPUTE_OR_MODEL_THROUGHPUT_LIMITED_BEFORE_RDMA_SATURATION"
+ else:
+ verdict = "RDMA_SATURATION_POSSIBLE_WITH_MODEL_LOAD"
+
+ return {
+ "generated_at": datetime.now().astimezone().isoformat(),
+ "formula": "rail_gbps ~= input_tps * bytes_per_input_token_per_rail * 8 / 1e9",
+ "plateau_gain_pct": plateau_gain_pct,
+ "target_rail_gbps": target_rail_gbps,
+ "practical_rail_gbps": practical_rail_gbps,
+ "scout": scout,
+ "confirm": confirm,
+ "selected_confirm_concurrencies": selected,
+ "observed_plateau_concurrency": plateau_c,
+ "best_scout_bandwidth_concurrency": best_scout_c,
+ "linear_gbps_per_input_tps": slope,
+ "modeled_bytes_per_input_token_per_rail": bytes_per_token,
+ "required_input_tps_for_400g_per_rail": required_target_tps,
+ "required_input_tps_for_360g_per_rail": required_practical_tps,
+ "required_input_tps_for_400g_dual_rail_aggregate": required_dual_400_tps,
+ "bandwidth_saturation_fit": bandwidth_fit,
+ "input_tps_saturation_fit": input_tps_fit,
+ "best_confirm_point": max_confirm,
+ "verdict": verdict,
+ }
+
+
+def write_recommendation(output_dir: Path, model: dict[str, Any]) -> None:
+ values = model.get("selected_confirm_concurrencies", [])
+ joined = " ".join(str(value) for value in values)
+ (output_dir / "recommendation.env").write_text(
+ f"CONFIRM_CONCURRENCIES='{joined}'\n", encoding="utf-8"
+ )
+
+
+def write_report(output_dir: Path, model: dict[str, Any]) -> None:
+ scout = model.get("scout", [])
+ confirm = model.get("confirm", [])
+ lines = [
+ "# Phase 2.5 RDMA Demand Model",
+ "",
+ f"- Generated: `{model['generated_at']}`",
+ f"- Verdict: `{model['verdict']}`",
+ f"- Model: `rail_gbps ~= input_tps x {fmt(model.get('modeled_bytes_per_input_token_per_rail') / 1e6 if model.get('modeled_bytes_per_input_token_per_rail') else None, 3)} MB/token/rail x 8`",
+ f"- Required Input TPS for 400G per rail: `{fmt(model.get('required_input_tps_for_400g_per_rail'), 0)}`",
+ f"- Required Input TPS for 360G per rail: `{fmt(model.get('required_input_tps_for_360g_per_rail'), 0)}`",
+ f"- Required Input TPS for 400G dual-rail aggregate: `{fmt(model.get('required_input_tps_for_400g_dual_rail_aggregate'), 0)}`",
+ f"- Fitted rail bandwidth ceiling: `{fmt(model.get('bandwidth_saturation_fit', {}).get('asymptote'))} Gbit/s`",
+ f"- Observed plateau concurrency: `{model.get('observed_plateau_concurrency')}`",
+ "",
+ "## Scout: 64K -> 1",
+ "",
+ "| C | Input TPS | Rail avg | Rail P95 | Dual-rail avg | MB/input-token/rail | GPU util |",
+ "|---:|---:|---:|---:|---:|---:|---:|",
+ ]
+ for row in scout:
+ bytes_per_token = number(row.get("xmit_bytes_per_input_token_per_rail"))
+ lines.append(
+ f"| {row['concurrency']} | {fmt(row.get('input_tps'))} | "
+ f"{fmt(row.get('rail_xmit_avg_gbps'))} | {fmt(row.get('rail_xmit_p95_gbps'))} | "
+ f"{fmt(row.get('dual_rail_xmit_avg_gbps'))} | "
+ f"{fmt(bytes_per_token / 1e6 if bytes_per_token else None, 3)} | "
+ f"{fmt(row.get('gpu_util_mean_pct'))} |"
+ )
+ lines.extend(
+ [
+ "",
+ "## Confirm: 64K -> 1K",
+ "",
+ "| C | Reps | Input TPS | Output TPS | Rail avg | Rail P95 | Dual-rail avg | TTFT P95 ms | TPOT P95 ms |",
+ "|---:|---:|---:|---:|---:|---:|---:|---:|---:|",
+ ]
+ )
+ for row in confirm:
+ lines.append(
+ f"| {row['concurrency']} | {row['repetitions']} | {fmt(row.get('input_tps'))} | "
+ f"{fmt(row.get('output_tps'))} | {fmt(row.get('rail_xmit_avg_gbps'))} | "
+ f"{fmt(row.get('rail_xmit_p95_gbps'))} | {fmt(row.get('dual_rail_xmit_avg_gbps'))} | "
+ f"{fmt(row.get('ttft_p95_ms'))} | {fmt(row.get('tpot_p95_ms'))} |"
+ )
+ lines.extend(
+ [
+ "",
+ "## Evidence",
+ "",
+ "- Per-case metrics: `rdma_case_metrics.csv`",
+ "- Machine-readable model: `rdma_demand_model.json`",
+ "- Scout HCA source: `scout/{head,worker}/rdma.csv`",
+ "- Confirm HCA source: `confirm/{head,worker}/rdma.csv`",
+ "- Exact windows: `scout/case_windows.csv`, `confirm/case_windows.csv`",
+ "- Benchmark summaries: `scout/bench_summary.csv`, `confirm/bench_summary.csv`",
+ ]
+ )
+ (output_dir / "rdma_demand_report.md").write_text(
+ "\n".join(lines) + "\n", encoding="utf-8"
+ )
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("mode", choices=("scout", "final"))
+ parser.add_argument("--scout-dir", required=True, type=Path)
+ parser.add_argument("--confirm-dir", type=Path)
+ parser.add_argument("--output-dir", required=True, type=Path)
+ parser.add_argument("--plateau-gain-pct", type=float, default=5.0)
+ parser.add_argument("--target-rail-gbps", type=float, default=400.0)
+ parser.add_argument("--practical-rail-gbps", type=float, default=360.0)
+ return parser.parse_args()
+
+
+def main() -> None:
+ args = parse_args()
+ args.output_dir.mkdir(parents=True, exist_ok=True)
+ scout_rows = aggregate_stage(args.scout_dir, "scout")
+ if not scout_rows:
+ raise SystemExit(f"no aligned scout rows found in {args.scout_dir}")
+ confirm_rows: list[dict[str, Any]] = []
+ if args.mode == "final":
+ if args.confirm_dir is None:
+ raise SystemExit("--confirm-dir is required in final mode")
+ confirm_rows = aggregate_stage(args.confirm_dir, "confirm")
+ if not confirm_rows:
+ raise SystemExit(f"no aligned confirm rows found in {args.confirm_dir}")
+
+ model = build_model(
+ scout_rows,
+ confirm_rows,
+ args.plateau_gain_pct,
+ args.target_rail_gbps,
+ args.practical_rail_gbps,
+ )
+ write_csv(args.output_dir / "rdma_case_metrics.csv", scout_rows + confirm_rows)
+ (args.output_dir / "rdma_demand_model.json").write_text(
+ json.dumps(model, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+ )
+ write_recommendation(args.output_dir, model)
+ write_report(args.output_dir, model)
+ print(json.dumps(model, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+ main()
diff --git a/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/run_rdma_demand_modeling.sh b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/run_rdma_demand_modeling.sh
new file mode 100755
index 0000000..509f327
--- /dev/null
+++ b/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/run_rdma_demand_modeling.sh
@@ -0,0 +1,264 @@
+#!/usr/bin/env bash
+# Measure how model token throughput translates into two-node RDMA bandwidth.
+
+set -Eeuo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck source=/dev/null
+source "${SCRIPT_DIR}/config.env"
+
+ACTION="${1:-all}"
+RUN_ID="${RUN_ID:-dsv4pro-phase2_5-$(date +%Y%m%d-%H%M%S)}"
+RESULT_DIR="${RESULT_BASE}/${RUN_ID}"
+COMMAND_DIR="${RESULT_DIR}/commands"
+SCENARIO_DIR="${RESULT_DIR}/scenarios"
+ANALYZER="${SCRIPT_DIR}/rdma_demand_model.py"
+RUN_LOG_ACTIVE=0
+
+log() {
+ printf '[%s] %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$*"
+}
+
+print_command() {
+ printf '%q ' "$@"
+ printf '\n'
+}
+
+enable_result_logging() {
+ [[ "${DRY_RUN}" == "1" || "${RUN_LOG_ACTIVE}" == "1" ]] && return 0
+ mkdir -p "${RESULT_DIR}"
+ exec > >(tee -a "${RESULT_DIR}/run.log") 2>&1
+ RUN_LOG_ACTIVE=1
+}
+
+normalize_concurrencies() {
+ tr ', ' '\n\n' <<< "$1" |
+ sed '/^[[:space:]]*$/d' |
+ awk '!seen[$0]++' |
+ sort -n |
+ paste -sd' ' -
+}
+
+validate_config() {
+ [[ -x "${PHASE2_ENTRY}" ]] || {
+ log "ERROR: Phase 2 entry is missing: ${PHASE2_ENTRY}"
+ return 1
+ }
+ [[ -f "${ANALYZER}" ]] || {
+ log "ERROR: analyzer is missing: ${ANALYZER}"
+ return 1
+ }
+ local name value
+ for name in ISL SCOUT_OSL CONFIRM_OSL SCOUT_REPETITIONS CONFIRM_REPETITIONS \
+ SAMPLE_INTERVAL_S COLLECTOR_MAX_SECONDS SCENARIO_TIMEOUT_S; do
+ value="${!name}"
+ [[ "${value}" =~ ^[1-9][0-9]*$ ]] || {
+ log "ERROR: ${name} must be a positive integer: ${value}"
+ return 1
+ }
+ done
+ [[ "${PLATEAU_GAIN_PCT}" =~ ^[0-9]+([.][0-9]+)?$ ]] || {
+ log "ERROR: PLATEAU_GAIN_PCT must be numeric"
+ return 1
+ }
+ local concurrency
+ for concurrency in $(normalize_concurrencies "${SCOUT_CONCURRENCIES}"); do
+ [[ "${concurrency}" =~ ^[1-9][0-9]*$ ]] || {
+ log "ERROR: invalid scout concurrency: ${concurrency}"
+ return 1
+ }
+ done
+}
+
+write_scenario_file() {
+ local stage="$1"
+ local osl="$2"
+ local concurrencies="$3"
+ local output="$4"
+ mkdir -p "$(dirname "${output}")"
+ printf '# case_id\tstage\tisl\tosl\tconcurrency\tprompt_multiplier\tmin_prompts\twarmup_requests\tnote\n' > "${output}"
+ local concurrency case_id
+ for concurrency in $(normalize_concurrencies "${concurrencies}"); do
+ case_id="rdma_${stage}_i${ISL}_o${osl}_c${concurrency}"
+ printf '%s\t%s\t%s\t%s\t%s\t1\t1\t0\t%s\n' \
+ "${case_id}" "rdma_${stage}" "${ISL}" "${osl}" "${concurrency}" \
+ "Phase 2.5 ${stage}: correlate model token rate with dual-rail RDMA" \
+ >> "${output}"
+ done
+}
+
+case_ids_for() {
+ local stage="$1"
+ local osl="$2"
+ local concurrencies="$3"
+ local result=() concurrency
+ for concurrency in $(normalize_concurrencies "${concurrencies}"); do
+ result+=("rdma_${stage}_i${ISL}_o${osl}_c${concurrency}")
+ done
+ local joined
+ joined="$(IFS=,; printf '%s' "${result[*]}")"
+ printf '%s\n' "${joined}"
+}
+
+run_phase2_stage() {
+ local stage="$1"
+ local scenario_file="$2"
+ local case_ids="$3"
+ local repetitions="$4"
+ local stage_result="${RESULT_DIR}/${stage}"
+ local command_file="${COMMAND_DIR}/${stage}.cmd.txt"
+ local -a command=(
+ env
+ "DRY_RUN=${DRY_RUN}"
+ "RUN_ID=${stage}"
+ "RESULT_BASE=${RESULT_DIR}"
+ "SCENARIO_FILE=${scenario_file}"
+ "FIXED_CASE_IDS=${case_ids}"
+ "RUN_MIXED_CASE=0"
+ "RUN_COMMUNICATION_BASELINE=0"
+ "NUM_REPETITIONS=${repetitions}"
+ "SAMPLE_INTERVAL_S=${SAMPLE_INTERVAL_S}"
+ "IDLE_BASELINE_S=${IDLE_BASELINE_S}"
+ "POST_RUN_COOLDOWN_S=${POST_RUN_COOLDOWN_S}"
+ "CASE_COOLDOWN_S=${CASE_COOLDOWN_S}"
+ "COLLECTOR_MAX_SECONDS=${COLLECTOR_MAX_SECONDS}"
+ "SCENARIO_TIMEOUT_S=${SCENARIO_TIMEOUT_S}"
+ bash "${PHASE2_ENTRY}" all
+ )
+
+ mkdir -p "${COMMAND_DIR}"
+ print_command "${command[@]}" > "${command_file}"
+ log "START stage=${stage} result=${stage_result}"
+ "${command[@]}"
+ log "DONE stage=${stage} result=${stage_result}"
+}
+
+run_scout() {
+ local scenario_file="${SCENARIO_DIR}/scout.tsv"
+ local concurrencies case_ids
+ concurrencies="$(normalize_concurrencies "${SCOUT_CONCURRENCIES}")"
+ write_scenario_file scout "${SCOUT_OSL}" "${concurrencies}" "${scenario_file}"
+ case_ids="$(case_ids_for scout "${SCOUT_OSL}" "${concurrencies}")"
+ run_phase2_stage scout "${scenario_file}" "${case_ids}" "${SCOUT_REPETITIONS}"
+
+ if [[ "${DRY_RUN}" != "1" ]]; then
+ python3 "${ANALYZER}" scout \
+ --scout-dir "${RESULT_DIR}/scout" \
+ --output-dir "${RESULT_DIR}" \
+ --plateau-gain-pct "${PLATEAU_GAIN_PCT}" \
+ --target-rail-gbps "${TARGET_RAIL_GBPS}" \
+ --practical-rail-gbps "${PRACTICAL_RAIL_GBPS}"
+ fi
+}
+
+confirm_concurrencies() {
+ if [[ "${DRY_RUN}" == "1" ]]; then
+ normalize_concurrencies "${DRY_RUN_CONFIRM_CONCURRENCIES}"
+ return 0
+ fi
+ # shellcheck source=/dev/null
+ source "${RESULT_DIR}/recommendation.env"
+ normalize_concurrencies "${CONFIRM_CONCURRENCIES}"
+}
+
+run_confirm() {
+ local scenario_file="${SCENARIO_DIR}/confirm.tsv"
+ local concurrencies case_ids
+ concurrencies="$(confirm_concurrencies)"
+ write_scenario_file confirm "${CONFIRM_OSL}" "${concurrencies}" "${scenario_file}"
+ case_ids="$(case_ids_for confirm "${CONFIRM_OSL}" "${concurrencies}")"
+ run_phase2_stage confirm "${scenario_file}" "${case_ids}" "${CONFIRM_REPETITIONS}"
+
+ if [[ "${DRY_RUN}" != "1" ]]; then
+ python3 "${ANALYZER}" final \
+ --scout-dir "${RESULT_DIR}/scout" \
+ --confirm-dir "${RESULT_DIR}/confirm" \
+ --output-dir "${RESULT_DIR}" \
+ --plateau-gain-pct "${PLATEAU_GAIN_PCT}" \
+ --target-rail-gbps "${TARGET_RAIL_GBPS}" \
+ --practical-rail-gbps "${PRACTICAL_RAIL_GBPS}"
+ fi
+}
+
+write_manifest() {
+ local commit dirty
+ commit="$(git -C "${REPO_ROOT}" rev-parse HEAD 2>/dev/null || printf unknown)"
+ dirty=0
+ [[ -n "$(git -C "${REPO_ROOT}" status --porcelain 2>/dev/null || true)" ]] && dirty=1
+ mkdir -p "${RESULT_DIR}"
+ cat > "${RESULT_DIR}/run_manifest.txt" <