qqtang/sskj
1
3
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Feat] add Phase 2.5 RDMA demand modeling

Browse Source
This commit is contained in:
Zhiyi Hong 2026-08-01 02:40:06 +08:00
parent 119701a417
commit c5fa700c50
8 changed files with 1022 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -1,5 +1,9 @@
# sskj — 多平台大模型推理性能基准测试项目
> **更新（2026-08-01 02:40:00 CST）**
>
> 新增 DeepSeek-V4-Pro 双机 Pro6000D SGLang Phase 2.5 RDMA 需求建模唯一入口。实验保持现有 TP16/EP2 服务参数不变，先以 `64K -> 1` 的 C=1/4/16/32/64 建立 Input TPS 与每 Rail HCA 带宽关系，再自动选择平台前、拐点和最大稳定并发，对 `64K -> 1K` 重复确认。结果将给出每 Token 跨机字节数、400G 所需 Token TPS、并发饱和曲线和“模型计算先饱和还是 RDMA 先饱和”的机器可读结论；正式结果尚未生成，因此暂不创建 Phase 2.5 HTML 档案。
>
> **更新（2026-07-31 19:02:00 CST）**
>
> Phase 3 Nsight smoke 已在 Head/Worker 各生成一份可解析报告；首份时间线确认 NCCL AllReduce、Sparse MLA、CUTLASS GEMM 与 MoE kernel 均可见。根据 smoke 修正正式 capture：Decode 对照和混合负载必须等服务实际进入 `Decode batch` 后才触发；每段等待本节点全部 8 个 rank 完成，而非任一 rank；双节点近同步停止以减少 Gloo 断链噪声。`nsys stats` 改为可写临时 SQLite、保留 Kernel/API/NVTX 高价值汇总并删除临时库，结果与报告数量均 fail-closed。

24
docs/dsv4pro_pro6000d_2node_sglang/推理优化计划.html
View File

@ -400,7 +400,7 @@
<div class="document-header__meta">
<span>节点：174.1.51.5 + 174.1.51.7</span>
<span>资源：16 × RTX PRO 6000 Blackwell</span>
<span>版本：2026-07-31 13:40:03 CST</span>
<span>版本：2026-08-01 02:45:00 CST</span>
</div>
</div>
</header>
@ -415,7 +415,7 @@
<article id="document-content">
<h1>6000D 双机 DeepSeek-V4-Pro 推理优化计划</h1>
<blockquote>
<p>适用环境：<code>174.1.51.5 + 174.1.51.7</code>，每台 8 张 RTX PRO 6000 Blackwell Server Edition<br>当前部署：DeepSeek-V4-Pro，16 张 GPU 组成一个完整实例<br>当前约束：模型暂时只能使用全部 16 张 GPU，无法额外复制一套模型进行 PD 分离<br>计划版本：2026-07-31 16:46:05 CST</p>
<p>适用环境：<code>174.1.51.5 + 174.1.51.7</code>，每台 8 张 RTX PRO 6000 Blackwell Server Edition<br>当前部署：DeepSeek-V4-Pro，16 张 GPU 组成一个完整实例<br>当前约束：模型暂时只能使用全部 16 张 GPU，无法额外复制一套模型进行 PD 分离<br>计划版本：2026-08-01 02:45:00 CST</p>
</blockquote>
<h2>当前执行状态与阶段档案</h2>
<table>
@ -448,6 +448,11 @@
<a href="./phase2_code.html">打开 Phase 2 代码详解</a>
</td>
</tr>
<tr>
<td>DeepSeek-V4-Pro / 双机 Pro6000D / SGLang RDMA 需求建模与并发拐点</td>
<td>Phase 2.5 代码已实现；等待正式 Run，以 64K Scout 拟合 bytes/token、带宽平台和 400G 所需 Token TPS</td>
<td>阶段尚未完成，按门禁暂不创建实验/代码 HTML</td>
</tr>
</tbody></table>
<p>
<strong>阶段档案生成门禁：</strong>Phase 尚未产出正式阶段结果、完成汇总和汇报确认前，
@ -785,6 +790,21 @@ Phase 2 回答“哪个硬件/Host/通信资源在什么 Case 中升高，以及
Phase 3 只捕获短 TP16 Timeline，回答具体 Kernel、Scheduler gap、Collective、
Rank 同步及计算/通信重叠，不重复 Phase 2 的长时间轻量采样。
</p>
<h2>6.5 Phase 2.5：RDMA 需求建模与并发拐点</h2>
<p>
Phase 2 已证明当前代表负载最高约 83.5 Gbit/s/rail，但单个固定 Case 不能回答
“提高并发后是否还能继续逼近 400G”。Phase 2.5 保持 TP16/EP2 和服务参数不变，
先用 <code>64K → 1</code> 的 C=1/4/16/32/64 隔离 Prefill，拟合
<code>rail_gbps ≈ input_tps × bytes_per_input_token_per_rail × 8</code>；
再自动选择平台前、拐点和最大稳定并发，对 <code>64K → 1K</code> 重复两次确认。
</p>
<p>
本阶段必须同时给出每 Rail Mean/P95/Max、双 Rail 单向合计、Input/Output TPS、
GPU 利用率、Rail 不均衡和 RDMA 错误增量。最终结论不是只报告“最大 C”，而是判断
模型计算吞吐和 RDMA 链路哪个先进入平台，并形成一套换模型、TP/EP 或输入输出形状后
可重复使用的带宽需求估算方法。正式结果完成前不创建
<code>phase2_5_exp.html</code> 或 <code>phase2_5_code.html</code>。
</p>
<h2>7. Phase 3：时间线 Profiling（Nsight Systems 为主）</h2>
<p>
Nsight Systems、PyTorch Profiler 和 NVTX 是三件不同的东西：Nsight Systems

3
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
results/
runtime/
*.log

35
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/README.md Normal file
View File

@ -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 并发。

36
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/config.env Normal file
View File

@ -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}"

563
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/rdma_demand_model.py Executable file
View File

@ -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()

264
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/run_rdma_demand_modeling.sh Executable file
View File

@ -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" <<EOF
run_id=${RUN_ID}
experiment=${EXPERIMENT}
created_at=$(date '+%Y-%m-%dT%H:%M:%S%z')
git_commit=${commit}
git_dirty=${dirty}
isl=${ISL}
scout_osl=${SCOUT_OSL}
confirm_osl=${CONFIRM_OSL}
scout_concurrencies=$(normalize_concurrencies "${SCOUT_CONCURRENCIES}")
scout_repetitions=${SCOUT_REPETITIONS}
confirm_repetitions=${CONFIRM_REPETITIONS}
target_rail_gbps=${TARGET_RAIL_GBPS}
practical_rail_gbps=${PRACTICAL_RAIL_GBPS}
EOF
}
stop_service() {
env \
"RUN_ID=${RUN_ID}-stop" \
"RESULT_BASE=${RESULT_DIR}" \
"RUN_COMMUNICATION_BASELINE=0" \
bash "${PHASE2_ENTRY}" stop || true
}
run_all() {
validate_config
enable_result_logging
write_manifest
trap 'stop_service' INT TERM
run_scout
run_confirm
trap - INT TERM
log "Phase 2.5 complete: ${RESULT_DIR}"
}
main() {
case "${ACTION}" in
audit)
validate_config
log "configuration valid"
;;
scout)
validate_config
enable_result_logging
write_manifest
run_scout
;;
confirm)
validate_config
enable_result_logging
run_confirm
;;
summarize)
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}"
;;
stop)
stop_service
;;
all)
run_all
;;
*)
printf 'Usage: %s {audit|scout|confirm|summarize|stop|all}\n' "$0" >&2
return 2
;;
esac
}
main "$@"

95
experiments/pro6000/dsv4pro_pro6000d_2node_sglang_rdma_demand_modeling/test_rdma_demand_model.py Normal file
View File

@ -0,0 +1,95 @@
#!/usr/bin/env python3
import importlib.util
import sys
import tempfile
import unittest
from pathlib import Path
MODULE_PATH = Path(__file__).with_name("rdma_demand_model.py")
SPEC = importlib.util.spec_from_file_location("rdma_demand_model", MODULE_PATH)
assert SPEC and SPEC.loader
MODEL = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = MODEL
SPEC.loader.exec_module(MODEL)
class RdmaDemandModelTest(unittest.TestCase):
def test_interval_rate_uses_four_octet_counter_units(self):
rows = [
{
"wall_time_ns": "1000000000",
"port_xmit_data": "100",
"port_rcv_data": "200",
},
{
"wall_time_ns": "2000000000",
"port_xmit_data": "100000100",
"port_rcv_data": "100000200",
},
]
xmit, rcv = MODEL.interval_rates(rows, 0, 3_000_000_000)
self.assertAlmostEqual(xmit[0], 3.2)
self.assertAlmostEqual(rcv[0], 3.2)
def test_linear_slope_recovers_bytes_per_token(self):
rows = [
{"input_tps": 1000, "rail_xmit_avg_gbps": 26.8},
{"input_tps": 2000, "rail_xmit_avg_gbps": 53.6},
{"input_tps": 3000, "rail_xmit_avg_gbps": 80.4},
]
slope = MODEL.linear_slope(rows)
self.assertAlmostEqual(slope, 0.0268)
self.assertAlmostEqual(slope * 1e9 / 8, 3_350_000)
def test_plateau_selects_previous_plateau_and_max(self):
rows = [
{"concurrency": 1, "input_tps": 1000, "rail_xmit_avg_gbps": 27},
{"concurrency": 4, "input_tps": 2400, "rail_xmit_avg_gbps": 64},
{"concurrency": 16, "input_tps": 3100, "rail_xmit_avg_gbps": 83},
{"concurrency": 32, "input_tps": 3150, "rail_xmit_avg_gbps": 84},
{"concurrency": 64, "input_tps": 3160, "rail_xmit_avg_gbps": 84.2},
]
selected, plateau, best = MODEL.select_concurrencies(rows, 5.0)
self.assertEqual(plateau, 32)
self.assertEqual(best, 64)
self.assertEqual(selected, [16, 32, 64])
def test_report_and_recommendation_are_created(self):
rows = [
{
"stage": "scout",
"concurrency": 16,
"repetitions": 1,
"input_tps": 3000,
"rail_xmit_avg_gbps": 80,
"rail_xmit_p95_gbps": 90,
"dual_rail_xmit_avg_gbps": 160,
"xmit_bytes_per_input_token_per_rail": 3_333_333,
"gpu_util_mean_pct": 99,
}
]
model = {
"generated_at": "2026-08-01T00:00:00+08:00",
"verdict": "TEST",
"modeled_bytes_per_input_token_per_rail": 3_333_333,
"required_input_tps_for_400g_per_rail": 15000,
"required_input_tps_for_360g_per_rail": 13500,
"required_input_tps_for_400g_dual_rail_aggregate": 7500,
"bandwidth_saturation_fit": {"asymptote": 100},
"observed_plateau_concurrency": 32,
"scout": rows,
"confirm": [],
"selected_confirm_concurrencies": [16, 32, 64],
}
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary)
MODEL.write_recommendation(output, model)
MODEL.write_report(output, model)
self.assertIn("16 32 64", (output / "recommendation.env").read_text())
self.assertIn("15000", (output / "rdma_demand_report.md").read_text())
if __name__ == "__main__":
unittest.main()