[Feat] add DSV4-Pro two-node SGLang hardware attribution
This commit is contained in:
Zhiyi Hong
parent
451782457d
commit
daa4221128
@ -1,5 +1,9 @@
|
|||||||
# sskj — 多平台大模型推理性能基准测试项目
|
# sskj — 多平台大模型推理性能基准测试项目
|
||||||
|
|
||||||
|
> **更新（2026-07-31 11:57:13 CST）**
|
||||||
|
>
|
||||||
|
> 实现 DeepSeek-V4-Pro 双机 Pro6000D SGLang TP16 的 Phase 2 硬件与资源竞争归因。新增唯一入口 `run_hardware_contention_attribution.sh`，内部复用 Phase 1 的双机服务与 benchmark，不要求用户手工启动 Phase 1；默认重放长/并发 Prefill、普通/持续/长上下文 Decode 和混合 Prefill/Decode A/B。Head 与 Worker 在同一诊断窗口采集 GPU、DCGM、CPU、进程、NUMA、`eth0/eth3` 和 `mlx5_0/mlx5_3` RDMA 数据，并保存 Case marker、完整命令、Manifest 和结构化汇总。正式执行只需运行 Phase 2 的 `all` 入口。
|
||||||
|
>
|
||||||
> **更新（2026-07-31 10:45:24 CST）**
|
> **更新（2026-07-31 10:45:24 CST）**
|
||||||
>
|
>
|
||||||
> 为 DeepSeek-V4-Pro 双机 Pro6000D SGLang TP16 的 Phase 2 硬件归因同步宿主机监控环境：`174.1.51.5` 与 `174.1.51.7` 均安装 `sysstat 12.5.2`，`pidstat`、`mpstat`、`sar -n DEV,EDEV` 的 1 秒实时采样验证通过。两节点的 sysstat service/timer 保持 `disabled/inactive`，正式实验由编排脚本显式启停采集，避免后台周期任务干扰 Case。Phase 2 必须同时保存 Head/Worker 的 sysstat、GPU、SGLang 与 RDMA 时间序列；`docker top` 用于 PID 映射，`perf` 用于 CPU 硬件事件，`numastat` 用于 NUMA 亲和，`mlx5_0/mlx5_3` HCA 计数器用于 RoCE 数据面，均不能由 sysstat 完整替代。
|
> 为 DeepSeek-V4-Pro 双机 Pro6000D SGLang TP16 的 Phase 2 硬件归因同步宿主机监控环境：`174.1.51.5` 与 `174.1.51.7` 均安装 `sysstat 12.5.2`，`pidstat`、`mpstat`、`sar -n DEV,EDEV` 的 1 秒实时采样验证通过。两节点的 sysstat service/timer 保持 `disabled/inactive`，正式实验由编排脚本显式启停采集，避免后台周期任务干扰 Case。Phase 2 必须同时保存 Head/Worker 的 sysstat、GPU、SGLang 与 RDMA 时间序列；`docker top` 用于 PID 映射，`perf` 用于 CPU 硬件事件，`numastat` 用于 NUMA 亲和，`mlx5_0/mlx5_3` HCA 计数器用于 RoCE 数据面，均不能由 sysstat 完整替代。
|
||||||
@ -81,6 +85,7 @@
|
|||||||
| `experiments/pro6000/dsv4_pro6000_vllm_tp_dp_matrix/` | RTX 6000D + vLLM |
|
| `experiments/pro6000/dsv4_pro6000_vllm_tp_dp_matrix/` | RTX 6000D + vLLM |
|
||||||
| `experiments/pro6000/dsv4_pro6000_sglang_tp_dp_matrix/` | RTX 6000D + SGLang |
|
| `experiments/pro6000/dsv4_pro6000_sglang_tp_dp_matrix/` | RTX 6000D + SGLang |
|
||||||
| `experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/` | 双机 RTX 6000D + DeepSeek-V4-Pro，SGLang TP16 快速性能地图与混合干扰 A/B |
|
| `experiments/pro6000/dsv4pro_pro6000d_2node_sglang_tp16_quick_map/` | 双机 RTX 6000D + DeepSeek-V4-Pro，SGLang TP16 快速性能地图与混合干扰 A/B |
|
||||||
|
| `experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/` | 双机 RTX 6000D + DeepSeek-V4-Pro，SGLang TP16 硬件与资源竞争归因 |
|
||||||
|
|
||||||
TP/DP matrix 目录内：`run_bench.sh` 跑固定并发矩阵；`run_adaptive_concurrency.sh` 从 C=1 指数倍增搜饱和点；`run_adaptive_concurrency_add16.sh` 从 C=16 线性 +16 步进、带 TTFT SLO 停止与回退（当前主力用法，见 `experiments/ADAPTIVE_CONCURRENCY_USAGE.md`）。专项 quick-map 的入口以对应目录 README 为准。
|
TP/DP matrix 目录内：`run_bench.sh` 跑固定并发矩阵；`run_adaptive_concurrency.sh` 从 C=1 指数倍增搜饱和点；`run_adaptive_concurrency_add16.sh` 从 C=16 线性 +16 步进、带 TTFT SLO 停止与回退（当前主力用法，见 `experiments/ADAPTIVE_CONCURRENCY_USAGE.md`）。专项 quick-map 的入口以对应目录 README 为准。
|
||||||
|
|
||||||
|
|||||||
@ -609,10 +609,9 @@ wait "${background_pid}"</code></pre>
|
|||||||
</tbody>
|
</tbody>
|
||||||
</table>
|
</table>
|
||||||
<p class="decision">
|
<p class="decision">
|
||||||
Phase 2 优先采集三类代表负载：<code>128K → 1，C=1</code> 的纯长 Prefill、
|
Phase 2 在上述长 Prefill、并发 Prefill和混合 A/B 之外，还会采集普通 Decode、
|
||||||
<code>32K → 1，C=16</code> 的并发 Prefill，以及
|
长输出 Decode 与长上下文 Decode。目标是区分计算、显存带宽、调度排队、
|
||||||
<code>1K → 1K，C=32</code> 在有无 128K 注入时的混合 A/B。
|
跨机通信、KV 增长和节点不均衡。
|
||||||
目标是区分计算、显存带宽、调度排队、跨机通信和节点不均衡。
|
|
||||||
</p>
|
</p>
|
||||||
<p>
|
<p>
|
||||||
完整产物：
|
完整产物：
|
||||||
@ -647,8 +646,8 @@ tmux attach -t dsv4pro-phase1-full</code></pre>
|
|||||||
|
|
||||||
<p>
|
<p>
|
||||||
下一阶段：
|
下一阶段：
|
||||||
<a class="back" href="./phase2_dsv4pro_pro6000d_2node_sglang_prefill_hardware_attribution.html">
|
<a class="back" href="./phase2_dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution.html">
|
||||||
Phase 2：Prefill 硬件指标归因
|
Phase 2：硬件与资源竞争归因
|
||||||
</a>
|
</a>
|
||||||
</p>
|
</p>
|
||||||
<p><a class="back" href="./推理优化计划.html">返回推理优化主计划</a></p>
|
<p><a class="back" href="./推理优化计划.html">返回推理优化主计划</a></p>
|
||||||
|
|||||||
@ -3,7 +3,7 @@
|
|||||||
<head>
|
<head>
|
||||||
<meta charset="utf-8">
|
<meta charset="utf-8">
|
||||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||||
<title>Phase 2：DeepSeek-V4-Pro 双机 Pro6000D SGLang Prefill 硬件归因</title>
|
<title>Phase 2：DeepSeek-V4-Pro 双机 Pro6000D SGLang 硬件与资源竞争归因</title>
|
||||||
<style>
|
<style>
|
||||||
:root {
|
:root {
|
||||||
color-scheme: light;
|
color-scheme: light;
|
||||||
@ -131,11 +131,11 @@
|
|||||||
<header>
|
<header>
|
||||||
<div class="header-inner">
|
<div class="header-inner">
|
||||||
<p class="eyebrow">Design, Implementation & Result Record</p>
|
<p class="eyebrow">Design, Implementation & Result Record</p>
|
||||||
<h1>Phase 2：DeepSeek-V4-Pro 双机 Pro6000D SGLang Prefill 硬件归因</h1>
|
<h1>Phase 2：DeepSeek-V4-Pro 双机 Pro6000D SGLang 硬件与资源竞争归因</h1>
|
||||||
<div class="meta">
|
<div class="meta">
|
||||||
<span>节点：174.1.51.5 + 174.1.51.7</span>
|
<span>节点：174.1.51.5 + 174.1.51.7</span>
|
||||||
<span>拓扑：SGLang TP16 / EP2</span>
|
<span>拓扑：SGLang TP16 / EP2</span>
|
||||||
<span>更新：2026-07-30 22:55:37 CST</span>
|
<span>更新：2026-07-31 12:26:00 CST</span>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
</header>
|
</header>
|
||||||
@ -144,9 +144,9 @@
|
|||||||
<a class="back" href="./推理优化计划.html">返回推理优化主计划</a>
|
<a class="back" href="./推理优化计划.html">返回推理优化主计划</a>
|
||||||
|
|
||||||
<p class="status">
|
<p class="status">
|
||||||
<strong>当前状态：Phase 1 前置条件已通过，等待阶段汇报确认后开始实现。</strong>
|
<strong>当前状态：Phase 2 代码已实现，静态检查、5 项单元测试与本地 Dry-run 已通过。</strong>
|
||||||
Phase 2 代码尚未创建。本页已经根据最终 Phase 1 结果选择诊断 Case；
|
尚未启动双机模型服务或正式采集 GPU 数据。真机运行后，本页只保留成功 Run 的
|
||||||
后续代码改动、静态验证、真机运行和结果判断会同步写入本页。
|
Run ID、命令、结果与结论。
|
||||||
</p>
|
</p>
|
||||||
|
|
||||||
<h2>1. Phase 1 交接结果</h2>
|
<h2>1. Phase 1 交接结果</h2>
|
||||||
@ -157,13 +157,17 @@
|
|||||||
<tbody>
|
<tbody>
|
||||||
<tr><td>128K → 1，C=1</td><td>Input TPS 2,710.16；TTFT P95 48.344 s</td><td>纯长 Prefill 的计算、显存与通信归因</td></tr>
|
<tr><td>128K → 1，C=1</td><td>Input TPS 2,710.16；TTFT P95 48.344 s</td><td>纯长 Prefill 的计算、显存与通信归因</td></tr>
|
||||||
<tr><td>32K → 1，C=16</td><td>Input TPS 3,112.77；TTFT P95 162.087 s</td><td>并发 Prefill 的排队、Chunk 调度与节点均衡</td></tr>
|
<tr><td>32K → 1，C=16</td><td>Input TPS 3,112.77；TTFT P95 162.087 s</td><td>并发 Prefill 的排队、Chunk 调度与节点均衡</td></tr>
|
||||||
|
<tr><td>1K → 1K，C=32</td><td>Output TPS 461.68；TPOT P95 63.31 ms</td><td>普通 Decode 的 GPU、CPU 与通信基线</td></tr>
|
||||||
|
<tr><td>1K → 4K，C=16</td><td>Output TPS 310.02；TPOT P95 50.33 ms</td><td>持续 Decode、KV 增长和稳态资源占用</td></tr>
|
||||||
|
<tr><td>128K → 1K，C=1</td><td>TTFT P95 49.326 s；TPOT P95 32.24 ms</td><td>分离长 Prefill 与长上下文 Decode 成本</td></tr>
|
||||||
<tr><td>1K → 1K，C=32 + 128K 注入</td><td>Output TPS -24.08%；TPOT P95 +66.55%</td><td>Prefill 干扰 Decode 时的硬件资源竞争</td></tr>
|
<tr><td>1K → 1K，C=32 + 128K 注入</td><td>Output TPS -24.08%；TPOT P95 +66.55%</td><td>Prefill 干扰 Decode 时的硬件资源竞争</td></tr>
|
||||||
</tbody>
|
</tbody>
|
||||||
</table>
|
</table>
|
||||||
<p>
|
<p>
|
||||||
最终基线已由 Head 与 Worker 日志证明使用
|
最终基线已由 Head 与 Worker 日志证明使用
|
||||||
<code>mlx5_0/mlx5_3</code> 双 Rail <code>NET/IB + GDRDMA</code>，
|
<code>mlx5_0/mlx5_3</code> 双 Rail <code>NET/IB + GDRDMA</code>，
|
||||||
正式测量请求为冷 Prefix，12/12 结果成功。Phase 2 保持相同服务配置和请求口径。
|
正式测量请求为冷 Prefix。正式矩阵 12/12、长 Decode 补测 2/2 均成功。
|
||||||
|
Phase 2 保持相同服务配置和请求口径。
|
||||||
</p>
|
</p>
|
||||||
|
|
||||||
<h2>2. 本阶段的边界</h2>
|
<h2>2. 本阶段的边界</h2>
|
||||||
@ -172,7 +176,7 @@
|
|||||||
<li>保留模型、镜像、TP16、EP2、显存比例和已验证的双 Rail NCCL 配置。</li>
|
<li>保留模型、镜像、TP16、EP2、显存比例和已验证的双 Rail NCCL 配置。</li>
|
||||||
<li>不启用 Nsight Systems、PyTorch Profiler、NCCL DEBUG 或投机解码。</li>
|
<li>不启用 Nsight Systems、PyTorch Profiler、NCCL DEBUG 或投机解码。</li>
|
||||||
<li>不调参，不尝试优化；先获得足以区分瓶颈类别的硬件证据。</li>
|
<li>不调参，不尝试优化；先获得足以区分瓶颈类别的硬件证据。</li>
|
||||||
<li>只重放本页选出的三个代表负载，不重复 Phase 1 全矩阵。</li>
|
<li>只重放五个固定代表负载和一组混合 A/B，不重复 Phase 1 全矩阵。</li>
|
||||||
<li>采集器从请求开始前启动，到请求结束后停止，不能中途补采后声称完整。</li>
|
<li>采集器从请求开始前启动，到请求结束后停止，不能中途补采后声称完整。</li>
|
||||||
</ul>
|
</ul>
|
||||||
|
|
||||||
@ -196,16 +200,60 @@
|
|||||||
<li>保存两节点静态快照：GPU/NIC/NUMA 拓扑、驱动、CUDA、镜像与服务命令。</li>
|
<li>保存两节点静态快照：GPU/NIC/NUMA 拓扑、驱动、CUDA、镜像与服务命令。</li>
|
||||||
<li>复用 Phase 1 已验证的 <code>run_quick_map.sh start</code> 启动同配置双机服务。</li>
|
<li>复用 Phase 1 已验证的 <code>run_quick_map.sh start</code> 启动同配置双机服务。</li>
|
||||||
<li>在 Head 和 Worker 同时启动 GPU、CPU、网卡与 RDMA 采样，先记录 15 秒空闲基线。</li>
|
<li>在 Head 和 Worker 同时启动 GPU、CPU、网卡与 RDMA 采样，先记录 15 秒空闲基线。</li>
|
||||||
<li>依次重放 <code>128K → 1, C=1</code> 与 <code>32K → 1, C=16</code>，请求参数和 Seed 与 Phase 1 一致。</li>
|
<li>依次重放 <code>128K → 1, C=1</code>、<code>32K → 1, C=16</code> 与 <code>1K → 1K, C=32</code>。</li>
|
||||||
|
<li>重放 <code>1K → 4K, C=16</code> 和 <code>128K → 1K, C=1</code>，观察持续与长上下文 Decode。</li>
|
||||||
<li>重放 <code>1K → 1K, C=32</code> Control 与 128K Prefill 注入 Treatment，保留相同注入时序。</li>
|
<li>重放 <code>1K → 1K, C=32</code> Control 与 128K Prefill 注入 Treatment，保留相同注入时序。</li>
|
||||||
<li>请求结束后继续采样 15 秒，再停止采集器和服务。</li>
|
<li>请求结束后继续采样 15 秒，再停止采集器和服务。</li>
|
||||||
<li>按时间戳将请求、GPU、CPU 和双 Rail 指标对齐，生成摘要与判定。</li>
|
<li>按时间戳将请求、GPU、CPU 和双 Rail 指标对齐，生成摘要与判定。</li>
|
||||||
</ol>
|
</ol>
|
||||||
<pre><code>idle 15s │ 128K C1 │ 32K C16 │ Decode Control │ Decode + Prefill │ cooldown 15s
|
<pre><code>idle 15s
|
||||||
Head 与 Worker 的所有采集器覆盖完整诊断窗口</code></pre>
|
│ 128K→1 C1 │ 32K→1 C16 │ 1K→1K C32
|
||||||
|
│ 1K→4K C16 │ 128K→1K C1
|
||||||
|
│ Decode Control │ Decode + Prefill
|
||||||
|
cooldown 15s
|
||||||
|
|
||||||
|
Head 与 Worker 的所有采集器覆盖完整诊断窗口。</code></pre>
|
||||||
<p>
|
<p>
|
||||||
Phase 1 中服务加载约 5 分 30 秒，三个诊断负载合计为分钟级；
|
Phase 1 中服务加载约 5 分 30 秒；五个固定负载加混合 A/B、静态快照、
|
||||||
连同静态快照、采样和清理，目标仍控制在 30 分钟内。
|
采样和清理，目标仍控制在约 30 分钟内。
|
||||||
|
</p>
|
||||||
|
|
||||||
|
<h3>4.1 你只需要运行的入口</h3>
|
||||||
|
<p class="decision">
|
||||||
|
<strong>操作规则：只执行 Phase 2 的 <code>all</code>。</strong>
|
||||||
|
不要手工执行 Phase 1 的 <code>start</code> 或 <code>stop</code>。
|
||||||
|
Phase 2 会在内部复用它们，并负责异常退出时的采集器、Head、Worker 清理。
|
||||||
|
</p>
|
||||||
|
<pre><code class="language-bash"># [仅在 174.1.51.5 执行]
|
||||||
|
cd /data/hzy/sskj/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution
|
||||||
|
|
||||||
|
# 第一次先展开全部命令，不启动服务、不占用 GPU、不发送请求
|
||||||
|
DRY_RUN=1 RUN_ID=dsv4pro-phase2-dryrun-$(date +%Y%m%d-%H%M%S) \
|
||||||
|
bash run_hardware_contention_attribution.sh all
|
||||||
|
|
||||||
|
# 正式实验：仍然只有同一个 all 入口，tmux 只负责断线后继续运行
|
||||||
|
RUN_ID=dsv4pro-phase2-$(date +%Y%m%d-%H%M%S)
|
||||||
|
tmux new-session -d -s dsv4pro-phase2 \
|
||||||
|
"RUN_ID=${RUN_ID} bash run_hardware_contention_attribution.sh all \
|
||||||
|
2>&1 | tee /data/hzy/${RUN_ID}.log"
|
||||||
|
|
||||||
|
tmux attach -t dsv4pro-phase2</code></pre>
|
||||||
|
<p><strong><code>all</code> 内部执行顺序：</strong></p>
|
||||||
|
<pre><code>配置与工具门禁
|
||||||
|
→ Phase 1 start：启动同配置 TP16 服务
|
||||||
|
→ 两节点静态快照
|
||||||
|
→ 启动两节点采集器并记录 15 秒 idle
|
||||||
|
→ 五个固定 Case
|
||||||
|
→ 混合 Prefill/Decode A/B
|
||||||
|
→ 15 秒 cooldown
|
||||||
|
→ 停止采集器并保存后快照
|
||||||
|
→ Phase 1 stop：停止 Head/Worker
|
||||||
|
→ 生成按 Case 对齐的 CSV、JSON 与 report.md</code></pre>
|
||||||
|
<p>
|
||||||
|
Phase 1 的作用是提供已经验证过的双机 Docker 服务和 Benchmark 实现，
|
||||||
|
不是第二个用户入口。实际展开的服务、Benchmark 和采集命令都会写入
|
||||||
|
<code>results/<RUN_ID>/service/</code>、<code>commands/</code> 和
|
||||||
|
<code>head|worker/collector_commands/</code>，不依赖跨文档猜测。
|
||||||
</p>
|
</p>
|
||||||
|
|
||||||
<h2>5. 采集指标</h2>
|
<h2>5. 采集指标</h2>
|
||||||
@ -223,24 +271,24 @@
|
|||||||
</table>
|
</table>
|
||||||
|
|
||||||
<h2>6. 精简代码设计</h2>
|
<h2>6. 精简代码设计</h2>
|
||||||
<p>计划新增目录：</p>
|
<p>已新增目录：</p>
|
||||||
<pre><code>/data/hzy/sskj/experiments/pro6000/
|
<pre><code>/data/hzy/sskj/experiments/pro6000/
|
||||||
dsv4pro_pro6000d_2node_sglang_prefill_hardware_attribution/</code></pre>
|
dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution/</code></pre>
|
||||||
<table>
|
<table>
|
||||||
<thead>
|
<thead>
|
||||||
<tr><th>文件</th><th>计划职责</th><th>当前状态</th></tr>
|
<tr><th>文件</th><th>职责</th><th>当前状态</th></tr>
|
||||||
</thead>
|
</thead>
|
||||||
<tbody>
|
<tbody>
|
||||||
<tr><td><code>run_prefill_hardware_attribution.sh</code></td><td>唯一 Shell 入口；服务启停、双节点采集器、单 Case、Trap 清理</td><td class="pending">待实现</td></tr>
|
<tr><td><code>run_hardware_contention_attribution.sh</code></td><td>唯一 Shell 入口；服务启停、双节点采集器、Case 编排、健康检查和 Trap 清理</td><td class="pass">已实现</td></tr>
|
||||||
<tr><td><code>config.env</code></td><td>Phase 1 入口路径、三个诊断 Case、采样间隔、结果路径</td><td class="pending">待实现</td></tr>
|
<tr><td><code>config.env</code></td><td>Phase 1 相对路径、节点、五个固定 Case、混合 A/B 与采样策略</td><td class="pass">已实现</td></tr>
|
||||||
<tr><td><code>hardware_attribution.py</code></td><td>结构化解析、时间对齐、统计摘要与报告生成</td><td class="pending">待实现</td></tr>
|
<tr><td><code>hardware_contention_attribution.py</code></td><td>Manifest、标记、Bench 校验、按 Case 时间窗切片和硬件摘要</td><td class="pass">已实现</td></tr>
|
||||||
<tr><td><code>tests/test_hardware_attribution.py</code></td><td>计数器差分、单位换算、统计与缺失工具回退测试</td><td class="pending">待实现</td></tr>
|
<tr><td><code>tests/test_hardware_contention_attribution.py</code></td><td>GPU 统计、RDMA 单位、Marker、嵌套结果与时间窗测试</td><td class="pass">5/5 通过</td></tr>
|
||||||
<tr><td><code>README.md</code></td><td>入口命令、环境变量和结果目录说明</td><td class="pending">待实现</td></tr>
|
<tr><td><code>README.md</code></td><td>唯一入口、范围和结果目录说明</td><td class="pass">已实现</td></tr>
|
||||||
</tbody>
|
</tbody>
|
||||||
</table>
|
</table>
|
||||||
<p class="decision">
|
<p class="decision">
|
||||||
Phase 2 不复制双机 Docker 启停实现。唯一入口通过环境变量调用 Phase 1 的
|
Phase 2 不复制双机 Docker 启停实现。唯一入口在内部调用 Phase 1 的
|
||||||
<code>run_quick_map.sh start/stop</code>，只新增硬件采集和三个代表负载的编排。
|
<code>run_quick_map.sh start/fixed/mixed/stop</code>，只新增硬件采集、时间对齐和代表负载编排。
|
||||||
顶层仍只保留一个 Shell 文件，不创建额外 tmux、launch、start 或 stop 脚本。
|
顶层仍只保留一个 Shell 文件，不创建额外 tmux、launch、start 或 stop 脚本。
|
||||||
</p>
|
</p>
|
||||||
|
|
||||||
@ -258,28 +306,36 @@ dsv4pro_pro6000d_2node_sglang_prefill_hardware_attribution/</code></pre>
|
|||||||
head_server.log
|
head_server.log
|
||||||
worker_server.log
|
worker_server.log
|
||||||
head/
|
head/
|
||||||
gpu.csv
|
gpu_samples.csv
|
||||||
cpu_mpstat.log
|
dcgm_dmon.log
|
||||||
cpu_pidstat.log
|
mpstat.log
|
||||||
net_sar.log
|
pidstat.log
|
||||||
|
sar_net.log
|
||||||
|
perf_stat.log
|
||||||
|
docker_top.log
|
||||||
|
numastat.log
|
||||||
rdma.csv
|
rdma.csv
|
||||||
static/
|
static_before.log
|
||||||
|
static_after.log
|
||||||
|
collector_commands/
|
||||||
worker/
|
worker/
|
||||||
gpu.csv
|
...
|
||||||
cpu_mpstat.log
|
collector_status.csv
|
||||||
cpu_pidstat.log
|
markers.csv
|
||||||
net_sar.log
|
bench_summary.csv
|
||||||
rdma.csv
|
gpu_summary.csv
|
||||||
static/
|
rdma_summary.csv
|
||||||
|
case_windows.csv
|
||||||
|
case_gpu_summary.csv
|
||||||
|
case_rdma_summary.csv
|
||||||
summary.json
|
summary.json
|
||||||
summary.csv
|
|
||||||
report.md</code></pre>
|
report.md</code></pre>
|
||||||
|
|
||||||
<h2>8. 验收条件</h2>
|
<h2>8. 验收条件</h2>
|
||||||
<ul>
|
<ul>
|
||||||
<li>Bench 的 ISL、OSL、并发、Seed、缓存状态与 Phase 1 对应 Case 一致。</li>
|
<li>Bench 的 ISL、OSL、并发、Seed、缓存状态与 Phase 1 对应 Case 一致。</li>
|
||||||
<li>两节点采集器均覆盖请求开始前 15 秒到结束后 15 秒。</li>
|
<li>两节点采集器均覆盖请求开始前 15 秒到结束后 15 秒。</li>
|
||||||
<li>每份时间序列有节点名、墙钟时间和单调时钟，能与服务 Chunk 日志对齐。</li>
|
<li>GPU/DCGM/RDMA 时间序列带节点名和墙钟时间，能按 Phase 1 的实际 Case 测量窗口切片。</li>
|
||||||
<li>采集器不可用时记录 <code>UNAVAILABLE</code> 和原因，不静默跳过。</li>
|
<li>采集器不可用时记录 <code>UNAVAILABLE</code> 和原因，不静默跳过。</li>
|
||||||
<li>异常退出仍会停止采集器、Head/Worker 容器并确认 GPU 释放。</li>
|
<li>异常退出仍会停止采集器、Head/Worker 容器并确认 GPU 释放。</li>
|
||||||
<li>报告至少能缩小到“计算/显存、CPU 调度、网络通信、频率节流、节点不均衡”中的一个或两个方向。</li>
|
<li>报告至少能缩小到“计算/显存、CPU 调度、网络通信、频率节流、节点不均衡”中的一个或两个方向。</li>
|
||||||
@ -294,13 +350,16 @@ dsv4pro_pro6000d_2node_sglang_prefill_hardware_attribution/</code></pre>
|
|||||||
<tbody>
|
<tbody>
|
||||||
<tr><td>2026-07-30 15:46 CST</td><td>创建 Phase 2 设计与档案</td><td>采用单入口和独立轻量采样，不复制双机服务启动逻辑</td></tr>
|
<tr><td>2026-07-30 15:46 CST</td><td>创建 Phase 2 设计与档案</td><td>采用单入口和独立轻量采样，不复制双机服务启动逻辑</td></tr>
|
||||||
<tr><td>2026-07-30 22:55:37 CST</td><td>完成 Phase 1 阶段交接</td><td>双 Rail 门禁和 12/12 正式结果通过；选定纯 Prefill、并发 Prefill、混合干扰三个诊断负载</td></tr>
|
<tr><td>2026-07-30 22:55:37 CST</td><td>完成 Phase 1 阶段交接</td><td>双 Rail 门禁和 12/12 正式结果通过；选定纯 Prefill、并发 Prefill、混合干扰三个诊断负载</td></tr>
|
||||||
|
<tr><td>2026-07-31 12:26:00 CST</td><td>完成 Phase 2 代码</td><td>扩展为五个固定负载和混合 A/B；实现双节点 GPU/DCGM/CPU/NUMA/网络/RDMA 采集、时间对齐和异常清理</td></tr>
|
||||||
|
<tr><td>2026-07-31 12:26:00 CST</td><td>本地验证</td><td><code>bash -n</code>、Python 编译、5 项单元测试与全流程 Dry-run 通过；未占用 GPU</td></tr>
|
||||||
</tbody>
|
</tbody>
|
||||||
</table>
|
</table>
|
||||||
|
|
||||||
<h2>10. 真机结果</h2>
|
<h2>10. 真机结果</h2>
|
||||||
<p class="pending">
|
<p class="pending">
|
||||||
尚未运行。按阶段门约定，先完成 Phase 1 汇报；用户确认进入 Phase 2 后，
|
尚未运行正式双机诊断。代码和本地 Dry-run 已完成；下一步是在
|
||||||
再实现代码、完成静态验证和 Dry-run，并在启动真机诊断前说明具体改动。
|
<code>174.1.51.5</code> 做服务器端静态检查与 Dry-run，随后由同一
|
||||||
|
<code>all</code> 入口启动正式 Run。正式完成后本节将替换为成功 Run 的结果和瓶颈判断。
|
||||||
</p>
|
</p>
|
||||||
|
|
||||||
<p><a class="back" href="./phase1_dsv4pro_pro6000d_2node_sglang_quick_map.html">返回 Phase 1 实施记录</a></p>
|
<p><a class="back" href="./phase1_dsv4pro_pro6000d_2node_sglang_quick_map.html">返回 Phase 1 实施记录</a></p>
|
||||||
@ -438,9 +438,9 @@
|
|||||||
<td><a href="./phase1_dsv4pro_pro6000d_2node_sglang_quick_map.html">打开实施记录</a></td>
|
<td><a href="./phase1_dsv4pro_pro6000d_2node_sglang_quick_map.html">打开实施记录</a></td>
|
||||||
</tr>
|
</tr>
|
||||||
<tr>
|
<tr>
|
||||||
<td>DeepSeek-V4-Pro / 双机 Pro6000D / SGLang Prefill 硬件指标归因</td>
|
<td>DeepSeek-V4-Pro / 双机 Pro6000D / SGLang 硬件与资源竞争归因</td>
|
||||||
<td>Phase 1 已交接，三个诊断负载已选定；等待阶段确认后开始代码实现</td>
|
<td>代码、单元测试与本地 Dry-run 已完成；等待服务器端 Dry-run 与正式运行</td>
|
||||||
<td><a href="./phase2_dsv4pro_pro6000d_2node_sglang_prefill_hardware_attribution.html">打开 Phase 2 档案</a></td>
|
<td><a href="./phase2_dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution.html">打开 Phase 2 档案</a></td>
|
||||||
</tr>
|
</tr>
|
||||||
</tbody></table>
|
</tbody></table>
|
||||||
<p>
|
<p>
|
||||||
@ -724,11 +724,11 @@ TPOT P95 增加 66.55%。Phase 2 将围绕这两个现象采集硬件时间序
|
|||||||
<h2>6. Phase 2：同步采集轻量硬件指标</h2>
|
<h2>6. Phase 2：同步采集轻量硬件指标</h2>
|
||||||
<p>
|
<p>
|
||||||
本阶段的设计、代码改动与结果同步维护在
|
本阶段的设计、代码改动与结果同步维护在
|
||||||
<a href="./phase2_dsv4pro_pro6000d_2node_sglang_prefill_hardware_attribution.html">
|
<a href="./phase2_dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution.html">
|
||||||
Phase 2：Prefill 硬件指标归因档案</a>。本阶段只重放三类代表负载：
|
Phase 2：硬件与资源竞争归因档案</a>。本阶段重放长 Prefill、并发 Prefill、
|
||||||
<code>128K → 1, C=1</code>、<code>32K → 1, C=16</code>，以及
|
普通 Decode、长输出 Decode、长上下文 Decode，以及
|
||||||
<code>1K → 1K, C=32</code> 的混合 A/B。目标是在约 30 分钟内区分 GPU、
|
<code>1K → 1K, C=32</code> 的混合 A/B。目标是在约 30 分钟内区分 GPU、
|
||||||
CPU、双 Rail、调度排队、频率节流和节点不均衡。
|
CPU、NUMA、双 Rail、调度排队、频率节流和节点不均衡。
|
||||||
</p>
|
</p>
|
||||||
<h3>6.1 GPU</h3>
|
<h3>6.1 GPU</h3>
|
||||||
<p>测试期间持续记录：</p>
|
<p>测试期间持续记录：</p>
|
||||||
@ -805,7 +805,7 @@ Phase 1 已证明 <code>NCCL_CROSS_NIC=1</code> 可以稳定完成端到端 Run
|
|||||||
<li>Phase 1 保留一份不启用 Profiler 的端到端基线，避免 TPS 和时延被诊断工具污染。</li>
|
<li>Phase 1 保留一份不启用 Profiler 的端到端基线，避免 TPS 和时延被诊断工具污染。</li>
|
||||||
<li>GPU、CPU 和网络的轻量采样可以伴随后续基线运行，但必须从 Case 开始前启动，并使用统一时间戳与 Case ID 对齐。</li>
|
<li>GPU、CPU 和网络的轻量采样可以伴随后续基线运行，但必须从 Case 开始前启动，并使用统一时间戳与 Case ID 对齐。</li>
|
||||||
<li>Phase 1 已完整结束，其无 Profiler 结果作为后续 A/B 的请求层基线。</li>
|
<li>Phase 1 已完整结束，其无 Profiler 结果作为后续 A/B 的请求层基线。</li>
|
||||||
<li>Phase 2 重放三个代表负载并完整采集轻量指标；得到瓶颈方向后，再决定 Phase 3 捕获哪段短时间线。</li>
|
<li>Phase 2 重放五个固定代表负载和一组混合 A/B，并完整采集轻量指标；得到瓶颈方向后，再决定 Phase 3 捕获哪段短时间线。</li>
|
||||||
<li>联合诊断 Run 的吞吐和时延只用于解释时间线；正式性能变化仍与 Phase 1 的无 Profiler 结果比较。</li>
|
<li>联合诊断 Run 的吞吐和时延只用于解释时间线；正式性能变化仍与 Phase 1 的无 Profiler 结果比较。</li>
|
||||||
</ul>
|
</ul>
|
||||||
<h2>7. Phase 3：时间线 Profiling（Nsight Systems 为主）</h2>
|
<h2>7. Phase 3：时间线 Profiling（Nsight Systems 为主）</h2>
|
||||||
|
|||||||
@ -0,0 +1,94 @@
|
|||||||
|
# DeepSeek-V4-Pro / Pro6000D / Two-Node / SGLang Hardware Attribution
|
||||||
|
|
||||||
|
Phase 2 replays a small set of Phase 1 workloads while collecting aligned GPU,
|
||||||
|
CPU, NUMA, SGLang, Ethernet and RDMA evidence on both nodes.
|
||||||
|
|
||||||
|
## Run
|
||||||
|
|
||||||
|
The user runs only this entry on `174.1.51.5`:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
cd /data/hzy/sskj/experiments/pro6000/dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution
|
||||||
|
|
||||||
|
RUN_ID="dsv4pro-phase2-$(date +%Y%m%d-%H%M%S)"
|
||||||
|
tmux new-session -d -s dsv4pro-phase2 \
|
||||||
|
"RUN_ID=${RUN_ID} bash run_hardware_contention_attribution.sh all \
|
||||||
|
2>&1 | tee /data/hzy/${RUN_ID}.log"
|
||||||
|
```
|
||||||
|
|
||||||
|
`all` internally calls the Phase 1 entry to start and stop the two-node TP16
|
||||||
|
service. Do not manually run Phase 1 `start` or `stop` in parallel.
|
||||||
|
|
||||||
|
Before using GPUs:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
bash -n run_hardware_contention_attribution.sh
|
||||||
|
python3 -m unittest discover -s tests -v
|
||||||
|
DRY_RUN=1 RUN_ID=dsv4pro-phase2-dryrun \
|
||||||
|
bash run_hardware_contention_attribution.sh all
|
||||||
|
```
|
||||||
|
|
||||||
|
## Scope
|
||||||
|
|
||||||
|
The default run covers:
|
||||||
|
|
||||||
|
| Workload | Purpose |
|
||||||
|
|---|---|
|
||||||
|
| `128K -> 1, C=1` | Long Prefill latency |
|
||||||
|
| `32K -> 1, C=16` | Concurrent Prefill throughput |
|
||||||
|
| `1K -> 1K, C=32` | Ordinary Decode |
|
||||||
|
| `1K -> 4K, C=16` | Sustained Decode and KV growth |
|
||||||
|
| `128K -> 1K, C=1` | Long-context Decode |
|
||||||
|
| `1K -> 1K, C=32` plus `128K -> 1` | Prefill/Decode contention |
|
||||||
|
|
||||||
|
The service and benchmark implementation remains owned by Phase 1. Phase 2
|
||||||
|
adds only the monitoring window, case markers and hardware summaries.
|
||||||
|
|
||||||
|
## Files
|
||||||
|
|
||||||
|
| File | Purpose |
|
||||||
|
|---|---|
|
||||||
|
| `run_hardware_contention_attribution.sh` | Only shell entry; service, collectors, cases, cleanup |
|
||||||
|
| `config.env` | Phase 1 path, node addresses, cases and sampling policy |
|
||||||
|
| `hardware_contention_attribution.py` | Markers, manifest, validation and summaries |
|
||||||
|
| `tests/test_hardware_contention_attribution.py` | Parser and unit-conversion tests |
|
||||||
|
|
||||||
|
## Output
|
||||||
|
|
||||||
|
```text
|
||||||
|
results/<RUN_ID>/
|
||||||
|
manifest.json
|
||||||
|
run.log
|
||||||
|
markers.csv
|
||||||
|
bench/<case-or-suite>/
|
||||||
|
service/
|
||||||
|
head_server_cmd.txt
|
||||||
|
worker_server_cmd.txt
|
||||||
|
head_server.log
|
||||||
|
worker_server.log
|
||||||
|
head/
|
||||||
|
gpu_samples.csv
|
||||||
|
dcgm_dmon.log
|
||||||
|
mpstat.log
|
||||||
|
pidstat.log
|
||||||
|
sar_net.log
|
||||||
|
perf_stat.log
|
||||||
|
docker_top.log
|
||||||
|
rdma.csv
|
||||||
|
static_before.log
|
||||||
|
static_after.log
|
||||||
|
collector_commands/
|
||||||
|
worker/
|
||||||
|
...
|
||||||
|
bench_summary.csv
|
||||||
|
gpu_summary.csv
|
||||||
|
rdma_summary.csv
|
||||||
|
case_windows.csv
|
||||||
|
case_gpu_summary.csv
|
||||||
|
case_rdma_summary.csv
|
||||||
|
summary.json
|
||||||
|
report.md
|
||||||
|
```
|
||||||
|
|
||||||
|
Raw collectors are diagnostic evidence. Final performance comparisons must
|
||||||
|
still use a no-profiler benchmark such as Phase 1.
|
||||||
@ -0,0 +1,40 @@
|
|||||||
|
# DeepSeek-V4-Pro two-node SGLang hardware and contention attribution.
|
||||||
|
|
||||||
|
EXPERIMENT="dsv4pro_pro6000d_2node_sglang_hardware_contention_attribution"
|
||||||
|
|
||||||
|
# Reuse the verified Phase 1 service and benchmark implementation.
|
||||||
|
PHASE1_DIR="${PHASE1_DIR:-$(cd "$(dirname "${BASH_SOURCE[0]}")/../dsv4pro_pro6000d_2node_sglang_tp16_quick_map" && pwd)}"
|
||||||
|
PHASE1_ENTRY="${PHASE1_ENTRY:-${PHASE1_DIR}/run_quick_map.sh}"
|
||||||
|
PHASE1_EXPERIMENT="${PHASE1_EXPERIMENT:-dsv4pro_pro6000d_2node_sglang_tp16_quick_map}"
|
||||||
|
|
||||||
|
# The two compute-network addresses. The script runs on HEAD_NODE and reaches
|
||||||
|
# WORKER_NODE with the SSH setup already used by Phase 1.
|
||||||
|
HEAD_NODE="${HEAD_NODE:-10.101.0.11}"
|
||||||
|
WORKER_NODE="${WORKER_NODE:-10.101.0.13}"
|
||||||
|
SGLANG_PORT="${SGLANG_PORT:-30002}"
|
||||||
|
HEAD_CONTAINER="${HEAD_CONTAINER:-${PHASE1_EXPERIMENT}_head}"
|
||||||
|
WORKER_CONTAINER="${WORKER_CONTAINER:-${PHASE1_EXPERIMENT}_worker}"
|
||||||
|
|
||||||
|
# Five isolated representative loads plus the existing mixed Prefill/Decode A/B.
|
||||||
|
FIXED_CASE_IDS="${FIXED_CASE_IDS:-long_prefill_latency_128k_c1,mid_prefill_throughput_32k_c16,decode_throughput_1k_to_1k_c32,long_output_decode_1k_to_4k_c16,long_context_decode_128k_to_1k_c1}"
|
||||||
|
RUN_MIXED_CASE="${RUN_MIXED_CASE:-1}"
|
||||||
|
|
||||||
|
# Monitoring policy.
|
||||||
|
SAMPLE_INTERVAL_S="${SAMPLE_INTERVAL_S:-1}"
|
||||||
|
IDLE_BASELINE_S="${IDLE_BASELINE_S:-15}"
|
||||||
|
POST_RUN_COOLDOWN_S="${POST_RUN_COOLDOWN_S:-15}"
|
||||||
|
CASE_COOLDOWN_S="${CASE_COOLDOWN_S:-5}"
|
||||||
|
COLLECTOR_MAX_SECONDS="${COLLECTOR_MAX_SECONDS:-7200}"
|
||||||
|
DCGM_FIELD_IDS="${DCGM_FIELD_IDS:-1001,1002,1003,1004,1005,1009,1010}"
|
||||||
|
PERF_EVENTS="${PERF_EVENTS:-cycles,instructions,cache-misses,context-switches,cpu-migrations,page-faults}"
|
||||||
|
RDMA_HCAS="${RDMA_HCAS:-mlx5_0 mlx5_3}"
|
||||||
|
NUMASTAT_INTERVAL_S="${NUMASTAT_INTERVAL_S:-5}"
|
||||||
|
CLOCK_SKEW_TOLERANCE_S="${CLOCK_SKEW_TOLERANCE_S:-2}"
|
||||||
|
|
||||||
|
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}"
|
||||||
|
ALLOW_PARTIAL_COLLECTORS="${ALLOW_PARTIAL_COLLECTORS:-1}"
|
||||||
@ -0,0 +1,574 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Metadata and lightweight summaries for Phase 2 hardware attribution."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import csv
|
||||||
|
import json
|
||||||
|
import math
|
||||||
|
import statistics
|
||||||
|
import time
|
||||||
|
from datetime import datetime
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any, Iterable
|
||||||
|
|
||||||
|
|
||||||
|
GPU_FIELDS = [
|
||||||
|
"wall_time_ns",
|
||||||
|
"node",
|
||||||
|
"gpu",
|
||||||
|
"device_timestamp",
|
||||||
|
"gpu_util_pct",
|
||||||
|
"memory_util_pct",
|
||||||
|
"memory_used_mib",
|
||||||
|
"memory_total_mib",
|
||||||
|
"power_w",
|
||||||
|
"temperature_c",
|
||||||
|
"sm_clock_mhz",
|
||||||
|
"memory_clock_mhz",
|
||||||
|
"pstate",
|
||||||
|
]
|
||||||
|
|
||||||
|
RDMA_FIELDS = [
|
||||||
|
"wall_time_ns",
|
||||||
|
"node",
|
||||||
|
"hca",
|
||||||
|
"port_xmit_data",
|
||||||
|
"port_rcv_data",
|
||||||
|
"port_xmit_packets",
|
||||||
|
"port_rcv_packets",
|
||||||
|
"port_xmit_wait",
|
||||||
|
"port_xmit_discards",
|
||||||
|
"port_rcv_errors",
|
||||||
|
"req_transport_retries_exceeded",
|
||||||
|
"req_rnr_retries_exceeded",
|
||||||
|
"roce_adp_retrans",
|
||||||
|
"roce_adp_retrans_to",
|
||||||
|
"np_ecn_marked_roce_packets",
|
||||||
|
"np_cnp_sent",
|
||||||
|
"rp_cnp_handled",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def now_iso() -> str:
|
||||||
|
return datetime.now().astimezone().isoformat(timespec="seconds")
|
||||||
|
|
||||||
|
|
||||||
|
def write_json(path: Path, value: dict[str, Any]) -> None:
|
||||||
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
path.write_text(
|
||||||
|
json.dumps(value, indent=2, ensure_ascii=False) + "\n",
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def parse_number(value: Any) -> float | None:
|
||||||
|
if value is None:
|
||||||
|
return None
|
||||||
|
text = str(value).strip()
|
||||||
|
if not text or text.upper() in {"N/A", "NA", "-"}:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
parsed = float(text)
|
||||||
|
except ValueError:
|
||||||
|
return None
|
||||||
|
return parsed if math.isfinite(parsed) else None
|
||||||
|
|
||||||
|
|
||||||
|
def percentile(values: Iterable[float], quantile: float) -> float | None:
|
||||||
|
ordered = sorted(values)
|
||||||
|
if not ordered:
|
||||||
|
return None
|
||||||
|
if len(ordered) == 1:
|
||||||
|
return ordered[0]
|
||||||
|
rank = (len(ordered) - 1) * quantile
|
||||||
|
lower = math.floor(rank)
|
||||||
|
upper = math.ceil(rank)
|
||||||
|
if lower == upper:
|
||||||
|
return ordered[lower]
|
||||||
|
weight = rank - lower
|
||||||
|
return ordered[lower] * (1 - weight) + ordered[upper] * weight
|
||||||
|
|
||||||
|
|
||||||
|
def read_csv_rows(path: Path) -> list[dict[str, str]]:
|
||||||
|
if not path.exists() or path.stat().st_size == 0:
|
||||||
|
return []
|
||||||
|
with path.open(newline="", encoding="utf-8", errors="replace") as handle:
|
||||||
|
return list(csv.DictReader(handle))
|
||||||
|
|
||||||
|
|
||||||
|
def summarize_gpu_rows(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
|
||||||
|
grouped: dict[tuple[str, str], list[dict[str, str]]] = {}
|
||||||
|
for row in rows:
|
||||||
|
node = row.get("node", "")
|
||||||
|
gpu = row.get("gpu", "")
|
||||||
|
if not node or not gpu:
|
||||||
|
continue
|
||||||
|
grouped.setdefault((node, gpu), []).append(row)
|
||||||
|
|
||||||
|
result: list[dict[str, Any]] = []
|
||||||
|
for (node, gpu), rows in sorted(grouped.items()):
|
||||||
|
item: dict[str, Any] = {
|
||||||
|
"node": node,
|
||||||
|
"gpu": gpu,
|
||||||
|
"samples": len(rows),
|
||||||
|
}
|
||||||
|
for field in (
|
||||||
|
"gpu_util_pct",
|
||||||
|
"memory_util_pct",
|
||||||
|
"memory_used_mib",
|
||||||
|
"power_w",
|
||||||
|
"temperature_c",
|
||||||
|
"sm_clock_mhz",
|
||||||
|
"memory_clock_mhz",
|
||||||
|
):
|
||||||
|
values = [
|
||||||
|
value
|
||||||
|
for value in (parse_number(row.get(field)) for row in rows)
|
||||||
|
if value is not None
|
||||||
|
]
|
||||||
|
item[f"{field}_mean"] = statistics.fmean(values) if values else None
|
||||||
|
item[f"{field}_p95"] = percentile(values, 0.95)
|
||||||
|
item[f"{field}_max"] = max(values) if values else None
|
||||||
|
result.append(item)
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def summarize_gpu(path: Path) -> list[dict[str, Any]]:
|
||||||
|
return summarize_gpu_rows(read_csv_rows(path))
|
||||||
|
|
||||||
|
|
||||||
|
def nonnegative_delta(first: float | None, last: float | None) -> float | None:
|
||||||
|
if first is None or last is None or last < first:
|
||||||
|
return None
|
||||||
|
return last - first
|
||||||
|
|
||||||
|
|
||||||
|
def summarize_rdma_rows(rows: list[dict[str, str]]) -> list[dict[str, Any]]:
|
||||||
|
grouped: dict[tuple[str, str], list[dict[str, str]]] = {}
|
||||||
|
for row in rows:
|
||||||
|
node = row.get("node", "")
|
||||||
|
hca = row.get("hca", "")
|
||||||
|
if not node or not hca:
|
||||||
|
continue
|
||||||
|
grouped.setdefault((node, hca), []).append(row)
|
||||||
|
|
||||||
|
result: list[dict[str, Any]] = []
|
||||||
|
for (node, hca), rows in sorted(grouped.items()):
|
||||||
|
rows.sort(key=lambda item: int(parse_number(item.get("wall_time_ns")) or 0))
|
||||||
|
first = rows[0]
|
||||||
|
last = rows[-1]
|
||||||
|
first_ns = parse_number(first.get("wall_time_ns"))
|
||||||
|
last_ns = parse_number(last.get("wall_time_ns"))
|
||||||
|
duration_s = (
|
||||||
|
(last_ns - first_ns) / 1_000_000_000
|
||||||
|
if first_ns is not None and last_ns is not None and last_ns > first_ns
|
||||||
|
else None
|
||||||
|
)
|
||||||
|
|
||||||
|
xmit_words = nonnegative_delta(
|
||||||
|
parse_number(first.get("port_xmit_data")),
|
||||||
|
parse_number(last.get("port_xmit_data")),
|
||||||
|
)
|
||||||
|
rcv_words = nonnegative_delta(
|
||||||
|
parse_number(first.get("port_rcv_data")),
|
||||||
|
parse_number(last.get("port_rcv_data")),
|
||||||
|
)
|
||||||
|
# IB port_{xmit,rcv}_data counters are measured in 4-octet units.
|
||||||
|
xmit_bytes = xmit_words * 4 if xmit_words is not None else None
|
||||||
|
rcv_bytes = rcv_words * 4 if rcv_words is not None else None
|
||||||
|
|
||||||
|
item: dict[str, Any] = {
|
||||||
|
"node": node,
|
||||||
|
"hca": hca,
|
||||||
|
"samples": len(rows),
|
||||||
|
"duration_s": duration_s,
|
||||||
|
"xmit_bytes": xmit_bytes,
|
||||||
|
"rcv_bytes": rcv_bytes,
|
||||||
|
"xmit_gbps": (
|
||||||
|
xmit_bytes * 8 / duration_s / 1e9
|
||||||
|
if xmit_bytes is not None and duration_s
|
||||||
|
else None
|
||||||
|
),
|
||||||
|
"rcv_gbps": (
|
||||||
|
rcv_bytes * 8 / duration_s / 1e9
|
||||||
|
if rcv_bytes is not None and duration_s
|
||||||
|
else None
|
||||||
|
),
|
||||||
|
}
|
||||||
|
for field in RDMA_FIELDS[7:]:
|
||||||
|
item[f"{field}_delta"] = nonnegative_delta(
|
||||||
|
parse_number(first.get(field)),
|
||||||
|
parse_number(last.get(field)),
|
||||||
|
)
|
||||||
|
result.append(item)
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def summarize_rdma(path: Path) -> list[dict[str, Any]]:
|
||||||
|
return summarize_rdma_rows(read_csv_rows(path))
|
||||||
|
|
||||||
|
|
||||||
|
def load_bench_rows(result_dir: Path) -> list[dict[str, str]]:
|
||||||
|
rows: list[dict[str, str]] = []
|
||||||
|
for summary_path in sorted((result_dir / "bench").glob("*/summary.csv")):
|
||||||
|
for row in read_csv_rows(summary_path):
|
||||||
|
row = dict(row)
|
||||||
|
row["phase2_bench_run"] = summary_path.parent.name
|
||||||
|
rows.append(row)
|
||||||
|
return rows
|
||||||
|
|
||||||
|
|
||||||
|
def iso_to_ns(value: str) -> int | None:
|
||||||
|
if not value:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
parsed = datetime.fromisoformat(value)
|
||||||
|
except ValueError:
|
||||||
|
return None
|
||||||
|
return int(parsed.timestamp() * 1_000_000_000)
|
||||||
|
|
||||||
|
|
||||||
|
def load_case_windows(result_dir: Path) -> list[dict[str, Any]]:
|
||||||
|
windows: list[dict[str, Any]] = []
|
||||||
|
for meta_path in sorted((result_dir / "bench").glob("*/cases/*/rep*/meta.json")):
|
||||||
|
try:
|
||||||
|
meta = json.loads(meta_path.read_text(encoding="utf-8"))
|
||||||
|
except (OSError, json.JSONDecodeError):
|
||||||
|
continue
|
||||||
|
started_ns = iso_to_ns(str(meta.get("started_at", "")))
|
||||||
|
ended_ns = iso_to_ns(str(meta.get("ended_at", "")))
|
||||||
|
if started_ns is None or ended_ns is None or ended_ns < started_ns:
|
||||||
|
continue
|
||||||
|
windows.append(
|
||||||
|
{
|
||||||
|
"phase2_bench_run": meta_path.parents[3].name,
|
||||||
|
"case_id": str(meta.get("case_id", "")),
|
||||||
|
"role": str(meta.get("role", "")),
|
||||||
|
"stage": str(meta.get("stage", "")),
|
||||||
|
"repetition": int(meta.get("repetition", 0) or 0),
|
||||||
|
"status": str(meta.get("status", "")),
|
||||||
|
"started_at": str(meta.get("started_at", "")),
|
||||||
|
"ended_at": str(meta.get("ended_at", "")),
|
||||||
|
"started_ns": started_ns,
|
||||||
|
"ended_ns": ended_ns,
|
||||||
|
"duration_s": (ended_ns - started_ns) / 1_000_000_000,
|
||||||
|
"meta_path": str(meta_path),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return windows
|
||||||
|
|
||||||
|
|
||||||
|
def rows_in_window(
|
||||||
|
rows: list[dict[str, str]],
|
||||||
|
started_ns: int,
|
||||||
|
ended_ns: int,
|
||||||
|
) -> list[dict[str, str]]:
|
||||||
|
result = []
|
||||||
|
for row in rows:
|
||||||
|
timestamp = parse_number(row.get("wall_time_ns"))
|
||||||
|
if timestamp is not None and started_ns <= timestamp <= ended_ns:
|
||||||
|
result.append(row)
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def summarize_case_hardware(
|
||||||
|
result_dir: Path,
|
||||||
|
windows: list[dict[str, Any]],
|
||||||
|
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||||||
|
gpu_by_node = {
|
||||||
|
node: read_csv_rows(result_dir / node / "gpu_samples.csv")
|
||||||
|
for node in ("head", "worker")
|
||||||
|
}
|
||||||
|
rdma_by_node = {
|
||||||
|
node: read_csv_rows(result_dir / node / "rdma.csv")
|
||||||
|
for node in ("head", "worker")
|
||||||
|
}
|
||||||
|
case_gpu: list[dict[str, Any]] = []
|
||||||
|
case_rdma: list[dict[str, Any]] = []
|
||||||
|
identity_fields = (
|
||||||
|
"phase2_bench_run",
|
||||||
|
"case_id",
|
||||||
|
"role",
|
||||||
|
"stage",
|
||||||
|
"repetition",
|
||||||
|
"status",
|
||||||
|
"started_at",
|
||||||
|
"ended_at",
|
||||||
|
"duration_s",
|
||||||
|
)
|
||||||
|
for window in windows:
|
||||||
|
identity = {field: window[field] for field in identity_fields}
|
||||||
|
for node in ("head", "worker"):
|
||||||
|
gpu_rows = rows_in_window(
|
||||||
|
gpu_by_node[node],
|
||||||
|
window["started_ns"],
|
||||||
|
window["ended_ns"],
|
||||||
|
)
|
||||||
|
for row in summarize_gpu_rows(gpu_rows):
|
||||||
|
case_gpu.append({**identity, **row})
|
||||||
|
|
||||||
|
rdma_rows = rows_in_window(
|
||||||
|
rdma_by_node[node],
|
||||||
|
window["started_ns"],
|
||||||
|
window["ended_ns"],
|
||||||
|
)
|
||||||
|
for row in summarize_rdma_rows(rdma_rows):
|
||||||
|
case_rdma.append({**identity, **row})
|
||||||
|
return case_gpu, case_rdma
|
||||||
|
|
||||||
|
|
||||||
|
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", newline="", encoding="utf-8") as handle:
|
||||||
|
writer = csv.DictWriter(handle, fieldnames=fields)
|
||||||
|
writer.writeheader()
|
||||||
|
writer.writerows(rows)
|
||||||
|
|
||||||
|
|
||||||
|
def append_marker(path: Path, node: str, event: str, case_id: str, status: str) -> None:
|
||||||
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
exists = path.exists() and path.stat().st_size > 0
|
||||||
|
with path.open("a", newline="", encoding="utf-8") as handle:
|
||||||
|
writer = csv.DictWriter(
|
||||||
|
handle,
|
||||||
|
fieldnames=[
|
||||||
|
"wall_time_ns",
|
||||||
|
"recorded_at",
|
||||||
|
"node",
|
||||||
|
"event",
|
||||||
|
"case_id",
|
||||||
|
"status",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
if not exists:
|
||||||
|
writer.writeheader()
|
||||||
|
writer.writerow(
|
||||||
|
{
|
||||||
|
"wall_time_ns": time.time_ns(),
|
||||||
|
"recorded_at": now_iso(),
|
||||||
|
"node": node,
|
||||||
|
"event": event,
|
||||||
|
"case_id": case_id,
|
||||||
|
"status": status,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def create_manifest(args: argparse.Namespace) -> None:
|
||||||
|
write_json(
|
||||||
|
Path(args.path),
|
||||||
|
{
|
||||||
|
"run_id": args.run_id,
|
||||||
|
"status": "RUNNING" if not args.dry_run else "DRY_RUN",
|
||||||
|
"created_at": now_iso(),
|
||||||
|
"completed_at": None,
|
||||||
|
"experiment": args.experiment,
|
||||||
|
"git_commit": args.git_commit,
|
||||||
|
"git_dirty": bool(args.git_dirty),
|
||||||
|
"phase1_entry": args.phase1_entry,
|
||||||
|
"head_node": args.head_node,
|
||||||
|
"worker_node": args.worker_node,
|
||||||
|
"fixed_case_ids": args.fixed_case_ids,
|
||||||
|
"run_mixed_case": bool(args.run_mixed_case),
|
||||||
|
"sample_interval_s": args.sample_interval_s,
|
||||||
|
"numastat_interval_s": args.numastat_interval_s,
|
||||||
|
"clock_skew_tolerance_s": args.clock_skew_tolerance_s,
|
||||||
|
"idle_baseline_s": args.idle_baseline_s,
|
||||||
|
"post_run_cooldown_s": args.post_run_cooldown_s,
|
||||||
|
"dry_run": bool(args.dry_run),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def finish_manifest(path: Path, status: str) -> None:
|
||||||
|
value = json.loads(path.read_text(encoding="utf-8"))
|
||||||
|
value["status"] = status
|
||||||
|
value["completed_at"] = now_iso()
|
||||||
|
write_json(path, value)
|
||||||
|
|
||||||
|
|
||||||
|
def check_bench(summary_path: Path) -> bool:
|
||||||
|
rows = read_csv_rows(summary_path)
|
||||||
|
return bool(rows) and all(row.get("status") == "COMPLETED" for row in rows)
|
||||||
|
|
||||||
|
|
||||||
|
def summarize(result_dir: Path) -> dict[str, Any]:
|
||||||
|
gpu_rows: list[dict[str, Any]] = []
|
||||||
|
rdma_rows: list[dict[str, Any]] = []
|
||||||
|
for node in ("head", "worker"):
|
||||||
|
gpu_rows.extend(summarize_gpu(result_dir / node / "gpu_samples.csv"))
|
||||||
|
rdma_rows.extend(summarize_rdma(result_dir / node / "rdma.csv"))
|
||||||
|
bench_rows = load_bench_rows(result_dir)
|
||||||
|
case_windows = load_case_windows(result_dir)
|
||||||
|
case_gpu_rows, case_rdma_rows = summarize_case_hardware(
|
||||||
|
result_dir,
|
||||||
|
case_windows,
|
||||||
|
)
|
||||||
|
|
||||||
|
write_csv(result_dir / "gpu_summary.csv", gpu_rows)
|
||||||
|
write_csv(result_dir / "rdma_summary.csv", rdma_rows)
|
||||||
|
write_csv(result_dir / "bench_summary.csv", bench_rows)
|
||||||
|
write_csv(result_dir / "case_windows.csv", case_windows)
|
||||||
|
write_csv(result_dir / "case_gpu_summary.csv", case_gpu_rows)
|
||||||
|
write_csv(result_dir / "case_rdma_summary.csv", case_rdma_rows)
|
||||||
|
|
||||||
|
failed_bench = [row for row in bench_rows if row.get("status") != "COMPLETED"]
|
||||||
|
collector_status_rows = read_csv_rows(result_dir / "collector_status.csv")
|
||||||
|
collector_status_counts: dict[str, int] = {}
|
||||||
|
for row in collector_status_rows:
|
||||||
|
status = row.get("status", "") or "UNKNOWN"
|
||||||
|
collector_status_counts[status] = collector_status_counts.get(status, 0) + 1
|
||||||
|
collector_files = {}
|
||||||
|
for node in ("head", "worker"):
|
||||||
|
node_dir = result_dir / node
|
||||||
|
collector_files[node] = {
|
||||||
|
path.name: path.stat().st_size
|
||||||
|
for path in sorted(node_dir.glob("*"))
|
||||||
|
if path.is_file()
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = {
|
||||||
|
"generated_at": now_iso(),
|
||||||
|
"bench_rows": len(bench_rows),
|
||||||
|
"failed_bench_rows": len(failed_bench),
|
||||||
|
"gpu_summary_rows": len(gpu_rows),
|
||||||
|
"rdma_summary_rows": len(rdma_rows),
|
||||||
|
"case_windows": len(case_windows),
|
||||||
|
"case_gpu_summary_rows": len(case_gpu_rows),
|
||||||
|
"case_rdma_summary_rows": len(case_rdma_rows),
|
||||||
|
"collector_status_counts": collector_status_counts,
|
||||||
|
"collector_files": collector_files,
|
||||||
|
}
|
||||||
|
write_json(result_dir / "summary.json", summary)
|
||||||
|
|
||||||
|
report = [
|
||||||
|
"# Phase 2 Hardware Attribution",
|
||||||
|
"",
|
||||||
|
f"- Generated: `{summary['generated_at']}`",
|
||||||
|
f"- Bench rows: `{len(bench_rows)}`",
|
||||||
|
f"- Failed bench rows: `{len(failed_bench)}`",
|
||||||
|
f"- GPU summary rows: `{len(gpu_rows)}`",
|
||||||
|
f"- RDMA summary rows: `{len(rdma_rows)}`",
|
||||||
|
f"- Case windows: `{len(case_windows)}`",
|
||||||
|
f"- Collector status counts: `{json.dumps(collector_status_counts, sort_keys=True)}`",
|
||||||
|
"",
|
||||||
|
"## Bench",
|
||||||
|
"",
|
||||||
|
"| Run | Case | Status | Input TPS | Output TPS | TTFT P95 (ms) | TPOT P95 (ms) |",
|
||||||
|
"|---|---|---|---:|---:|---:|---:|",
|
||||||
|
]
|
||||||
|
for row in bench_rows:
|
||||||
|
report.append(
|
||||||
|
"| {phase2_bench_run} | {case_id} | {status} | {input_token_throughput} | "
|
||||||
|
"{output_token_throughput} | {ttft_p95_ms} | {tpot_p95_ms} |".format(
|
||||||
|
**{
|
||||||
|
"phase2_bench_run": row.get("phase2_bench_run", ""),
|
||||||
|
"case_id": row.get("case_id", ""),
|
||||||
|
"status": row.get("status", ""),
|
||||||
|
"input_token_throughput": row.get("input_token_throughput", ""),
|
||||||
|
"output_token_throughput": row.get("output_token_throughput", ""),
|
||||||
|
"ttft_p95_ms": row.get("ttft_p95_ms", ""),
|
||||||
|
"tpot_p95_ms": row.get("tpot_p95_ms", ""),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
)
|
||||||
|
report.extend(
|
||||||
|
[
|
||||||
|
"",
|
||||||
|
"## Machine-readable summaries",
|
||||||
|
"",
|
||||||
|
"- `gpu_summary.csv`",
|
||||||
|
"- `rdma_summary.csv`",
|
||||||
|
"- `bench_summary.csv`",
|
||||||
|
"- `case_windows.csv`",
|
||||||
|
"- `case_gpu_summary.csv`",
|
||||||
|
"- `case_rdma_summary.csv`",
|
||||||
|
"- `summary.json`",
|
||||||
|
"",
|
||||||
|
"The final bottleneck conclusion is written only after aligning these files "
|
||||||
|
"with `markers.csv`, raw DCGM/sysstat logs, and SGLang service logs.",
|
||||||
|
"",
|
||||||
|
]
|
||||||
|
)
|
||||||
|
(result_dir / "report.md").write_text("\n".join(report), encoding="utf-8")
|
||||||
|
return summary
|
||||||
|
|
||||||
|
|
||||||
|
def build_parser() -> argparse.ArgumentParser:
|
||||||
|
parser = argparse.ArgumentParser()
|
||||||
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||||
|
|
||||||
|
marker_parser = subparsers.add_parser("marker")
|
||||||
|
marker_parser.add_argument("--path", required=True)
|
||||||
|
marker_parser.add_argument("--node", required=True)
|
||||||
|
marker_parser.add_argument("--event", required=True)
|
||||||
|
marker_parser.add_argument("--case-id", default="")
|
||||||
|
marker_parser.add_argument("--status", default="")
|
||||||
|
|
||||||
|
manifest_parser = subparsers.add_parser("manifest")
|
||||||
|
manifest_parser.add_argument("--path", required=True)
|
||||||
|
manifest_parser.add_argument("--run-id", required=True)
|
||||||
|
manifest_parser.add_argument("--experiment", required=True)
|
||||||
|
manifest_parser.add_argument("--git-commit", required=True)
|
||||||
|
manifest_parser.add_argument("--git-dirty", type=int, required=True)
|
||||||
|
manifest_parser.add_argument("--phase1-entry", required=True)
|
||||||
|
manifest_parser.add_argument("--head-node", required=True)
|
||||||
|
manifest_parser.add_argument("--worker-node", required=True)
|
||||||
|
manifest_parser.add_argument("--fixed-case-ids", required=True)
|
||||||
|
manifest_parser.add_argument("--run-mixed-case", type=int, required=True)
|
||||||
|
manifest_parser.add_argument("--sample-interval-s", type=int, required=True)
|
||||||
|
manifest_parser.add_argument("--numastat-interval-s", type=int, required=True)
|
||||||
|
manifest_parser.add_argument("--clock-skew-tolerance-s", type=int, required=True)
|
||||||
|
manifest_parser.add_argument("--idle-baseline-s", type=int, required=True)
|
||||||
|
manifest_parser.add_argument("--post-run-cooldown-s", type=int, required=True)
|
||||||
|
manifest_parser.add_argument("--dry-run", type=int, required=True)
|
||||||
|
|
||||||
|
finish_parser = subparsers.add_parser("finish")
|
||||||
|
finish_parser.add_argument("--path", required=True)
|
||||||
|
finish_parser.add_argument("--status", required=True)
|
||||||
|
|
||||||
|
check_parser = subparsers.add_parser("check-bench")
|
||||||
|
check_parser.add_argument("--summary", required=True)
|
||||||
|
|
||||||
|
summarize_parser = subparsers.add_parser("summarize")
|
||||||
|
summarize_parser.add_argument("result_dir")
|
||||||
|
return parser
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
args = build_parser().parse_args()
|
||||||
|
if args.command == "marker":
|
||||||
|
append_marker(
|
||||||
|
Path(args.path),
|
||||||
|
args.node,
|
||||||
|
args.event,
|
||||||
|
args.case_id,
|
||||||
|
args.status,
|
||||||
|
)
|
||||||
|
return 0
|
||||||
|
if args.command == "manifest":
|
||||||
|
create_manifest(args)
|
||||||
|
return 0
|
||||||
|
if args.command == "finish":
|
||||||
|
finish_manifest(Path(args.path), args.status)
|
||||||
|
return 0
|
||||||
|
if args.command == "check-bench":
|
||||||
|
return 0 if check_bench(Path(args.summary)) else 1
|
||||||
|
if args.command == "summarize":
|
||||||
|
summarize(Path(args.result_dir))
|
||||||
|
return 0
|
||||||
|
raise AssertionError(args.command)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
@ -0,0 +1,689 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# Collect aligned two-node hardware evidence around representative SGLang loads.
|
||||||
|
|
||||||
|
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-$(date +%Y%m%d-%H%M%S)}"
|
||||||
|
RESULT_DIR="${RESULT_BASE}/${RUN_ID}"
|
||||||
|
RESULT_TOOL="${SCRIPT_DIR}/hardware_contention_attribution.py"
|
||||||
|
MARKERS_PATH="${RESULT_DIR}/markers.csv"
|
||||||
|
SERVICE_DIR="${RESULT_DIR}/service"
|
||||||
|
COMMAND_DIR="${RESULT_DIR}/commands"
|
||||||
|
COLLECTOR_TAG="phase2-collector-${RUN_ID}"
|
||||||
|
|
||||||
|
declare -a COLLECTOR_PIDS=()
|
||||||
|
declare -a COLLECTOR_LABELS=()
|
||||||
|
SERVICE_STARTED=0
|
||||||
|
COLLECTORS_STARTED=0
|
||||||
|
RUN_LOG_ACTIVE=0
|
||||||
|
|
||||||
|
log() {
|
||||||
|
printf '[%s] %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$*"
|
||||||
|
}
|
||||||
|
|
||||||
|
print_command() {
|
||||||
|
printf '%q ' "$@"
|
||||||
|
printf '\n'
|
||||||
|
}
|
||||||
|
|
||||||
|
node_is_local() {
|
||||||
|
local node="$1"
|
||||||
|
local local_ips
|
||||||
|
[[ "${node}" == "localhost" || "${node}" == "127.0.0.1" ]] && return 0
|
||||||
|
local_ips="$(hostname -I 2>/dev/null || true)"
|
||||||
|
[[ " ${local_ips} " == *" ${node} "* ]]
|
||||||
|
}
|
||||||
|
|
||||||
|
run_on_node() {
|
||||||
|
local node="$1"
|
||||||
|
local command="$2"
|
||||||
|
if node_is_local "${node}"; then
|
||||||
|
bash -lc "${command}"
|
||||||
|
else
|
||||||
|
ssh -o BatchMode=yes -o StrictHostKeyChecking=no "${node}" "${command}"
|
||||||
|
fi
|
||||||
|
}
|
||||||
|
|
||||||
|
write_command_file() {
|
||||||
|
local path="$1"
|
||||||
|
shift
|
||||||
|
mkdir -p "$(dirname "${path}")"
|
||||||
|
print_command "$@" > "${path}"
|
||||||
|
}
|
||||||
|
|
||||||
|
enable_result_logging() {
|
||||||
|
[[ "${RUN_LOG_ACTIVE}" == "1" || "${DRY_RUN}" == "1" ]] && return 0
|
||||||
|
mkdir -p "${RESULT_DIR}"
|
||||||
|
exec > >(tee -a "${RESULT_DIR}/run.log") 2>&1
|
||||||
|
RUN_LOG_ACTIVE=1
|
||||||
|
}
|
||||||
|
|
||||||
|
validate_config() {
|
||||||
|
[[ -x "${PHASE1_ENTRY}" ]] || {
|
||||||
|
log "ERROR: Phase 1 entry is missing or not executable: ${PHASE1_ENTRY}"
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
[[ -x "${RESULT_TOOL}" ]] || {
|
||||||
|
log "ERROR: result tool is missing or not executable: ${RESULT_TOOL}"
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
case "${RUN_MIXED_CASE}" in
|
||||||
|
0|1) ;;
|
||||||
|
*)
|
||||||
|
log "ERROR: RUN_MIXED_CASE must be 0 or 1"
|
||||||
|
return 1
|
||||||
|
;;
|
||||||
|
esac
|
||||||
|
case "${ALLOW_PARTIAL_COLLECTORS}" in
|
||||||
|
0|1) ;;
|
||||||
|
*)
|
||||||
|
log "ERROR: ALLOW_PARTIAL_COLLECTORS must be 0 or 1"
|
||||||
|
return 1
|
||||||
|
;;
|
||||||
|
esac
|
||||||
|
[[ "${SAMPLE_INTERVAL_S}" =~ ^[1-9][0-9]*$ ]] || {
|
||||||
|
log "ERROR: SAMPLE_INTERVAL_S must be a positive integer"
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
[[ "${NUMASTAT_INTERVAL_S}" =~ ^[1-9][0-9]*$ ]] || {
|
||||||
|
log "ERROR: NUMASTAT_INTERVAL_S must be a positive integer"
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
[[ "${CLOCK_SKEW_TOLERANCE_S}" =~ ^[0-9]+$ ]] || {
|
||||||
|
log "ERROR: CLOCK_SKEW_TOLERANCE_S must be a non-negative integer"
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
[[ -n "${FIXED_CASE_IDS//[[:space:],]/}" ]] || {
|
||||||
|
log "ERROR: FIXED_CASE_IDS is empty"
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
preflight_node_tools() {
|
||||||
|
local node="$1"
|
||||||
|
local required="nvidia-smi docker mpstat pidstat sar numastat perf"
|
||||||
|
local tool
|
||||||
|
for tool in ${required}; do
|
||||||
|
if ! run_on_node "${node}" "command -v '${tool}' >/dev/null"; then
|
||||||
|
log "ERROR: ${node} is missing required tool: ${tool}"
|
||||||
|
return 1
|
||||||
|
fi
|
||||||
|
done
|
||||||
|
if ! run_on_node "${node}" "command -v dcgmi >/dev/null"; then
|
||||||
|
if [[ "${ALLOW_PARTIAL_COLLECTORS}" == "1" ]]; then
|
||||||
|
log "WARN: ${node} has no dcgmi; DCGM collector will be unavailable"
|
||||||
|
else
|
||||||
|
log "ERROR: ${node} is missing required tool: dcgmi"
|
||||||
|
return 1
|
||||||
|
fi
|
||||||
|
fi
|
||||||
|
local hca
|
||||||
|
for hca in ${RDMA_HCAS}; do
|
||||||
|
if ! run_on_node "${node}" \
|
||||||
|
"test -d '/sys/class/infiniband/${hca}/ports/1'"; then
|
||||||
|
log "ERROR: ${node} is missing RDMA HCA counters for ${hca}"
|
||||||
|
return 1
|
||||||
|
fi
|
||||||
|
done
|
||||||
|
}
|
||||||
|
|
||||||
|
preflight_clock_sync() {
|
||||||
|
local head_epoch worker_epoch delta
|
||||||
|
head_epoch="$(run_on_node "${HEAD_NODE}" "date +%s")"
|
||||||
|
worker_epoch="$(run_on_node "${WORKER_NODE}" "date +%s")"
|
||||||
|
[[ "${head_epoch}" =~ ^[0-9]+$ && "${worker_epoch}" =~ ^[0-9]+$ ]] || {
|
||||||
|
log "ERROR: failed to read node clocks: head=${head_epoch} worker=${worker_epoch}"
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
delta=$(( head_epoch - worker_epoch ))
|
||||||
|
(( delta < 0 )) && delta=$(( -delta ))
|
||||||
|
if (( delta > CLOCK_SKEW_TOLERANCE_S )); then
|
||||||
|
log "ERROR: node clock skew ${delta}s exceeds ${CLOCK_SKEW_TOLERANCE_S}s"
|
||||||
|
return 1
|
||||||
|
fi
|
||||||
|
log "node clock skew check passed: delta=${delta}s"
|
||||||
|
}
|
||||||
|
|
||||||
|
write_manifest() {
|
||||||
|
local git_commit git_dirty
|
||||||
|
git_commit="$(git -C "${REPO_ROOT}" rev-parse HEAD 2>/dev/null || printf unknown)"
|
||||||
|
git_dirty=0
|
||||||
|
[[ -n "$(git -C "${REPO_ROOT}" status --porcelain 2>/dev/null || true)" ]] \
|
||||||
|
&& git_dirty=1
|
||||||
|
python3 "${RESULT_TOOL}" manifest \
|
||||||
|
--path "${RESULT_DIR}/manifest.json" \
|
||||||
|
--run-id "${RUN_ID}" \
|
||||||
|
--experiment "${EXPERIMENT}" \
|
||||||
|
--git-commit "${git_commit}" \
|
||||||
|
--git-dirty "${git_dirty}" \
|
||||||
|
--phase1-entry "${PHASE1_ENTRY}" \
|
||||||
|
--head-node "${HEAD_NODE}" \
|
||||||
|
--worker-node "${WORKER_NODE}" \
|
||||||
|
--fixed-case-ids "${FIXED_CASE_IDS}" \
|
||||||
|
--run-mixed-case "${RUN_MIXED_CASE}" \
|
||||||
|
--sample-interval-s "${SAMPLE_INTERVAL_S}" \
|
||||||
|
--numastat-interval-s "${NUMASTAT_INTERVAL_S}" \
|
||||||
|
--clock-skew-tolerance-s "${CLOCK_SKEW_TOLERANCE_S}" \
|
||||||
|
--idle-baseline-s "${IDLE_BASELINE_S}" \
|
||||||
|
--post-run-cooldown-s "${POST_RUN_COOLDOWN_S}" \
|
||||||
|
--dry-run "${DRY_RUN}"
|
||||||
|
}
|
||||||
|
|
||||||
|
mark_event() {
|
||||||
|
local event="$1"
|
||||||
|
local case_id="${2:-}"
|
||||||
|
local status="${3:-}"
|
||||||
|
local role
|
||||||
|
for role in orchestrator head worker; do
|
||||||
|
local path="${MARKERS_PATH}"
|
||||||
|
[[ "${role}" == "head" ]] && path="${RESULT_DIR}/head/markers.csv"
|
||||||
|
[[ "${role}" == "worker" ]] && path="${RESULT_DIR}/worker/markers.csv"
|
||||||
|
python3 "${RESULT_TOOL}" marker \
|
||||||
|
--path "${path}" \
|
||||||
|
--node "${role}" \
|
||||||
|
--event "${event}" \
|
||||||
|
--case-id "${case_id}" \
|
||||||
|
--status "${status}"
|
||||||
|
done
|
||||||
|
}
|
||||||
|
|
||||||
|
run_phase1_action() {
|
||||||
|
local action="$1"
|
||||||
|
local phase_run_id="${2:-${RUN_ID}}"
|
||||||
|
local case_ids="${3:-}"
|
||||||
|
local -a command=(
|
||||||
|
env
|
||||||
|
"DRY_RUN=${DRY_RUN}"
|
||||||
|
"RUN_ID=${phase_run_id}"
|
||||||
|
"RESULT_BASE=${RESULT_DIR}/bench"
|
||||||
|
"SERVER_ARTIFACT_DIR=${SERVICE_DIR}"
|
||||||
|
"CASE_IDS=${case_ids}"
|
||||||
|
"CASE_COOLDOWN_S=0"
|
||||||
|
bash "${PHASE1_ENTRY}" "${action}"
|
||||||
|
)
|
||||||
|
write_command_file \
|
||||||
|
"${COMMAND_DIR}/phase1_${action}_${phase_run_id}.cmd.txt" \
|
||||||
|
"${command[@]}"
|
||||||
|
"${command[@]}"
|
||||||
|
}
|
||||||
|
|
||||||
|
start_service() {
|
||||||
|
mark_event service_start
|
||||||
|
# Set this before launch so the EXIT trap also cleans a partially started pair.
|
||||||
|
SERVICE_STARTED=1
|
||||||
|
run_phase1_action start service ""
|
||||||
|
mark_event service_healthy
|
||||||
|
}
|
||||||
|
|
||||||
|
stop_service() {
|
||||||
|
[[ "${SERVICE_STARTED}" == "1" || "${ACTION}" == "stop" ]] || return 0
|
||||||
|
run_phase1_action stop service "" || true
|
||||||
|
SERVICE_STARTED=0
|
||||||
|
mark_event service_stopped
|
||||||
|
}
|
||||||
|
|
||||||
|
capture_command() {
|
||||||
|
local role="$1"
|
||||||
|
local node="$2"
|
||||||
|
local label="$3"
|
||||||
|
local command="$4"
|
||||||
|
local node_dir="${RESULT_DIR}/${role}"
|
||||||
|
local output="${node_dir}/${label}.log"
|
||||||
|
local command_file="${node_dir}/collector_commands/${label}.cmd.txt"
|
||||||
|
mkdir -p "$(dirname "${command_file}")"
|
||||||
|
printf 'node=%q\ncommand=%q\n' "${node}" "${command}" > "${command_file}"
|
||||||
|
if [[ "${DRY_RUN}" == "1" ]]; then
|
||||||
|
log "[DRY] capture role=${role} node=${node} label=${label}: ${command}"
|
||||||
|
return 0
|
||||||
|
fi
|
||||||
|
run_on_node "${node}" "${command}" > "${output}" 2>&1
|
||||||
|
}
|
||||||
|
|
||||||
|
static_snapshot_command() {
|
||||||
|
local container="$1"
|
||||||
|
printf '%s' "
|
||||||
|
set +e
|
||||||
|
date --iso-8601=ns
|
||||||
|
hostname
|
||||||
|
uname -a
|
||||||
|
nvidia-smi
|
||||||
|
nvidia-smi topo -m
|
||||||
|
nvidia-smi --query-gpu=index,uuid,pci.bus_id,driver_version,memory.total,memory.used,pstate,temperature.gpu,power.draw,clocks.sm,clocks.mem --format=csv
|
||||||
|
lscpu
|
||||||
|
numactl --hardware
|
||||||
|
numastat -m
|
||||||
|
ip -details link show eth0
|
||||||
|
ip -details link show eth3
|
||||||
|
ethtool eth0
|
||||||
|
ethtool eth3
|
||||||
|
ethtool -S eth0
|
||||||
|
ethtool -S eth3
|
||||||
|
ibdev2netdev
|
||||||
|
ibstat
|
||||||
|
rdma link show
|
||||||
|
docker inspect '${container}'
|
||||||
|
docker top '${container}' -eo pid,ppid,psr,pcpu,pmem,stat,comm,args
|
||||||
|
"
|
||||||
|
}
|
||||||
|
|
||||||
|
capture_static_snapshots() {
|
||||||
|
local suffix="$1"
|
||||||
|
capture_command \
|
||||||
|
head "${HEAD_NODE}" "static_${suffix}" \
|
||||||
|
"$(static_snapshot_command "${HEAD_CONTAINER}")"
|
||||||
|
capture_command \
|
||||||
|
worker "${WORKER_NODE}" "static_${suffix}" \
|
||||||
|
"$(static_snapshot_command "${WORKER_CONTAINER}")"
|
||||||
|
}
|
||||||
|
|
||||||
|
start_stream_collector() {
|
||||||
|
local role="$1"
|
||||||
|
local node="$2"
|
||||||
|
local label="$3"
|
||||||
|
local command="$4"
|
||||||
|
local node_dir="${RESULT_DIR}/${role}"
|
||||||
|
local output="${node_dir}/${label}"
|
||||||
|
local command_file="${node_dir}/collector_commands/${label}.cmd.txt"
|
||||||
|
local wrapped quoted
|
||||||
|
printf -v quoted '%q' "${command}"
|
||||||
|
wrapped="timeout --signal=TERM --kill-after=5s ${COLLECTOR_MAX_SECONDS}s bash -lc ${quoted} ${COLLECTOR_TAG}-${role}-${label}"
|
||||||
|
|
||||||
|
mkdir -p "$(dirname "${command_file}")"
|
||||||
|
printf 'node=%q\ncommand=%q\n' "${node}" "${wrapped}" > "${command_file}"
|
||||||
|
if [[ "${DRY_RUN}" == "1" ]]; then
|
||||||
|
log "[DRY] collector role=${role} node=${node} label=${label}: ${wrapped}"
|
||||||
|
printf '%s,%s,%s,%s,%s\n' \
|
||||||
|
"$(date +%s%N)" "${role}" "${label}" DRY_RUN "" \
|
||||||
|
>> "${RESULT_DIR}/collector_status.csv"
|
||||||
|
return 0
|
||||||
|
fi
|
||||||
|
|
||||||
|
if node_is_local "${node}"; then
|
||||||
|
bash -lc "${wrapped}" > "${output}" 2>&1 &
|
||||||
|
else
|
||||||
|
ssh -o BatchMode=yes -o StrictHostKeyChecking=no \
|
||||||
|
"${node}" "${wrapped}" > "${output}" 2>&1 &
|
||||||
|
fi
|
||||||
|
COLLECTOR_PIDS+=("$!")
|
||||||
|
COLLECTOR_LABELS+=("${role}/${label}")
|
||||||
|
printf '%s,%s,%s,%s,%s\n' \
|
||||||
|
"$(date +%s%N)" "${role}" "${label}" STARTED "$!" \
|
||||||
|
>> "${RESULT_DIR}/collector_status.csv"
|
||||||
|
}
|
||||||
|
|
||||||
|
gpu_sampler_command() {
|
||||||
|
local role="$1"
|
||||||
|
printf '%s' "
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
printf '%s\n' 'wall_time_ns,node,gpu,device_timestamp,gpu_util_pct,memory_util_pct,memory_used_mib,memory_total_mib,power_w,temperature_c,sm_clock_mhz,memory_clock_mhz,pstate'
|
||||||
|
while :; do
|
||||||
|
wall_time_ns=\$(date +%s%N)
|
||||||
|
nvidia-smi --query-gpu=index,timestamp,utilization.gpu,utilization.memory,memory.used,memory.total,power.draw,temperature.gpu,clocks.sm,clocks.mem,pstate --format=csv,noheader,nounits |
|
||||||
|
sed \"s/^/\${wall_time_ns},${role},/\"
|
||||||
|
sleep '${SAMPLE_INTERVAL_S}'
|
||||||
|
done
|
||||||
|
"
|
||||||
|
}
|
||||||
|
|
||||||
|
rdma_sampler_command() {
|
||||||
|
local role="$1"
|
||||||
|
printf '%s' "
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
read_counter() {
|
||||||
|
if [[ -r \"\$1\" ]]; then
|
||||||
|
tr -d '\\n' < \"\$1\"
|
||||||
|
fi
|
||||||
|
}
|
||||||
|
printf '%s\n' 'wall_time_ns,node,hca,port_xmit_data,port_rcv_data,port_xmit_packets,port_rcv_packets,port_xmit_wait,port_xmit_discards,port_rcv_errors,req_transport_retries_exceeded,req_rnr_retries_exceeded,roce_adp_retrans,roce_adp_retrans_to,np_ecn_marked_roce_packets,np_cnp_sent,rp_cnp_handled'
|
||||||
|
while :; do
|
||||||
|
wall_time_ns=\$(date +%s%N)
|
||||||
|
for hca in ${RDMA_HCAS}; do
|
||||||
|
base=\"/sys/class/infiniband/\${hca}/ports/1\"
|
||||||
|
printf '%s,%s,%s' \"\${wall_time_ns}\" '${role}' \"\${hca}\"
|
||||||
|
for counter in port_xmit_data port_rcv_data port_xmit_packets port_rcv_packets port_xmit_wait port_xmit_discards port_rcv_errors req_transport_retries_exceeded req_rnr_retries_exceeded roce_adp_retrans roce_adp_retrans_to np_ecn_marked_roce_packets np_cnp_sent rp_cnp_handled; do
|
||||||
|
path=\"\${base}/counters/\${counter}\"
|
||||||
|
[[ -r \"\${path}\" ]] || path=\"\${base}/hw_counters/\${counter}\"
|
||||||
|
printf ',%s' \"\$(read_counter \"\${path}\")\"
|
||||||
|
done
|
||||||
|
printf '\\n'
|
||||||
|
done
|
||||||
|
sleep '${SAMPLE_INTERVAL_S}'
|
||||||
|
done
|
||||||
|
"
|
||||||
|
}
|
||||||
|
|
||||||
|
docker_top_command() {
|
||||||
|
local container="$1"
|
||||||
|
printf '%s' "
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
while :; do
|
||||||
|
printf 'wall_time_ns=%s\\n' \"\$(date +%s%N)\"
|
||||||
|
docker top '${container}' -eo pid,ppid,psr,pcpu,pmem,stat,comm,args
|
||||||
|
sleep '${SAMPLE_INTERVAL_S}'
|
||||||
|
done
|
||||||
|
"
|
||||||
|
}
|
||||||
|
|
||||||
|
container_pid_preamble() {
|
||||||
|
local container="$1"
|
||||||
|
printf '%s' "
|
||||||
|
PIDS=\$(docker top '${container}' -eo pid 2>/dev/null |
|
||||||
|
awk 'NR > 1 {print \$1}' |
|
||||||
|
paste -sd, -)
|
||||||
|
if [[ -z \"\${PIDS}\" ]]; then
|
||||||
|
printf 'container has no visible processes: %s\\n' '${container}' >&2
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
"
|
||||||
|
}
|
||||||
|
|
||||||
|
numastat_command() {
|
||||||
|
local container="$1"
|
||||||
|
printf '%s' "
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
while :; do
|
||||||
|
printf 'wall_time_ns=%s\\n' \"\$(date +%s%N)\"
|
||||||
|
PIDS=\$(docker top '${container}' -eo pid 2>/dev/null |
|
||||||
|
awk 'NR > 1 {print \$1}')
|
||||||
|
if [[ -z \"\${PIDS}\" ]]; then
|
||||||
|
printf 'container has no visible processes: %s\\n' '${container}' >&2
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
for pid in \${PIDS}; do
|
||||||
|
numastat -p \"\${pid}\"
|
||||||
|
done
|
||||||
|
sleep '${NUMASTAT_INTERVAL_S}'
|
||||||
|
done
|
||||||
|
"
|
||||||
|
}
|
||||||
|
|
||||||
|
start_node_collectors() {
|
||||||
|
local role="$1"
|
||||||
|
local node="$2"
|
||||||
|
local container="$3"
|
||||||
|
local dcgm_command="
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
printf '%s\\n' 'wall_time_ns,node,dcgm_output'
|
||||||
|
dcgmi dmon -e '${DCGM_FIELD_IDS}' -d '$(( SAMPLE_INTERVAL_S * 1000 ))' |
|
||||||
|
while IFS= read -r line; do
|
||||||
|
printf '%s,%s,%s\\n' \"\$(date +%s%N)\" '${role}' \"\${line}\"
|
||||||
|
done
|
||||||
|
"
|
||||||
|
local mpstat_command="
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
mpstat -P ALL '${SAMPLE_INTERVAL_S}'
|
||||||
|
"
|
||||||
|
local pidstat_command
|
||||||
|
pidstat_command="$(container_pid_preamble "${container}")"
|
||||||
|
pidstat_command+="
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
pidstat -durwt -p \"\${PIDS}\" '${SAMPLE_INTERVAL_S}'
|
||||||
|
"
|
||||||
|
local sar_command="
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
sar -n DEV,EDEV '${SAMPLE_INTERVAL_S}'
|
||||||
|
"
|
||||||
|
local perf_command
|
||||||
|
perf_command="$(container_pid_preamble "${container}")"
|
||||||
|
perf_command+="
|
||||||
|
trap 'exit 0' HUP TERM PIPE
|
||||||
|
perf stat -p \"\${PIDS}\" -I 1000 -e '${PERF_EVENTS}'
|
||||||
|
"
|
||||||
|
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" gpu_samples.csv "$(gpu_sampler_command "${role}")"
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" rdma.csv "$(rdma_sampler_command "${role}")"
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" dcgm_dmon.log "${dcgm_command}"
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" mpstat.log "${mpstat_command}"
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" pidstat.log "${pidstat_command}"
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" sar_net.log "${sar_command}"
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" perf_stat.log "${perf_command}"
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" docker_top.log "$(docker_top_command "${container}")"
|
||||||
|
start_stream_collector \
|
||||||
|
"${role}" "${node}" numastat.log "$(numastat_command "${container}")"
|
||||||
|
}
|
||||||
|
|
||||||
|
start_collectors() {
|
||||||
|
mark_event collectors_start
|
||||||
|
start_node_collectors head "${HEAD_NODE}" "${HEAD_CONTAINER}"
|
||||||
|
start_node_collectors worker "${WORKER_NODE}" "${WORKER_CONTAINER}"
|
||||||
|
COLLECTORS_STARTED=1
|
||||||
|
if [[ "${DRY_RUN}" != "1" ]]; then
|
||||||
|
sleep 2
|
||||||
|
check_collectors
|
||||||
|
fi
|
||||||
|
}
|
||||||
|
|
||||||
|
check_collectors() {
|
||||||
|
local failed=0 index pid label
|
||||||
|
for index in "${!COLLECTOR_PIDS[@]}"; do
|
||||||
|
pid="${COLLECTOR_PIDS[$index]}"
|
||||||
|
label="${COLLECTOR_LABELS[$index]}"
|
||||||
|
if ! kill -0 "${pid}" 2>/dev/null; then
|
||||||
|
log "WARN: collector exited early: ${label}"
|
||||||
|
printf '%s,%s,%s,%s,%s\n' \
|
||||||
|
"$(date +%s%N)" "${label%%/*}" "${label#*/}" EXITED_EARLY "${pid}" \
|
||||||
|
>> "${RESULT_DIR}/collector_status.csv"
|
||||||
|
((failed+=1))
|
||||||
|
fi
|
||||||
|
done
|
||||||
|
if (( failed > 0 )) && [[ "${ALLOW_PARTIAL_COLLECTORS}" != "1" ]]; then
|
||||||
|
log "ERROR: ${failed} collectors exited early"
|
||||||
|
return 1
|
||||||
|
fi
|
||||||
|
}
|
||||||
|
|
||||||
|
stop_collectors() {
|
||||||
|
[[ "${COLLECTORS_STARTED}" == "1" ]] || return 0
|
||||||
|
if [[ "${DRY_RUN}" != "1" ]]; then
|
||||||
|
local index pid label collector_state
|
||||||
|
for index in "${!COLLECTOR_PIDS[@]}"; do
|
||||||
|
pid="${COLLECTOR_PIDS[$index]}"
|
||||||
|
label="${COLLECTOR_LABELS[$index]}"
|
||||||
|
collector_state="EXITED_BEFORE_STOP"
|
||||||
|
if kill -0 "${pid}" 2>/dev/null; then
|
||||||
|
collector_state="STOPPED"
|
||||||
|
kill -TERM "${pid}" 2>/dev/null || true
|
||||||
|
fi
|
||||||
|
wait "${pid}" 2>/dev/null || true
|
||||||
|
printf '%s,%s,%s,%s,%s\n' \
|
||||||
|
"$(date +%s%N)" "${label%%/*}" "${label#*/}" "${collector_state}" \
|
||||||
|
"${pid}" >> "${RESULT_DIR}/collector_status.csv"
|
||||||
|
done
|
||||||
|
local role node pattern
|
||||||
|
for role in head worker; do
|
||||||
|
node="${HEAD_NODE}"
|
||||||
|
[[ "${role}" == "worker" ]] && node="${WORKER_NODE}"
|
||||||
|
pattern="[${COLLECTOR_TAG:0:1}]${COLLECTOR_TAG:1}-${role}-"
|
||||||
|
run_on_node "${node}" \
|
||||||
|
"pkill -TERM -f '${pattern}' >/dev/null 2>&1 || true" || true
|
||||||
|
done
|
||||||
|
fi
|
||||||
|
COLLECTOR_PIDS=()
|
||||||
|
COLLECTOR_LABELS=()
|
||||||
|
COLLECTORS_STARTED=0
|
||||||
|
mark_event collectors_stopped
|
||||||
|
}
|
||||||
|
|
||||||
|
sleep_if_real() {
|
||||||
|
local seconds="$1"
|
||||||
|
if [[ "${DRY_RUN}" == "1" ]]; then
|
||||||
|
log "[DRY] sleep ${seconds}s"
|
||||||
|
else
|
||||||
|
sleep "${seconds}"
|
||||||
|
fi
|
||||||
|
}
|
||||||
|
|
||||||
|
run_fixed_case() {
|
||||||
|
local case_id="$1"
|
||||||
|
local phase_run_id="fixed_${case_id}"
|
||||||
|
local summary_path="${RESULT_DIR}/bench/${phase_run_id}/summary.csv"
|
||||||
|
local rc=0
|
||||||
|
|
||||||
|
mark_event case_start "${case_id}"
|
||||||
|
log "START Phase 2 fixed case=${case_id}"
|
||||||
|
if run_phase1_action fixed "${phase_run_id}" "${case_id}"; then
|
||||||
|
:
|
||||||
|
else
|
||||||
|
rc=$?
|
||||||
|
fi
|
||||||
|
if [[ "${DRY_RUN}" != "1" ]] \
|
||||||
|
&& ! python3 "${RESULT_TOOL}" check-bench --summary "${summary_path}"; then
|
||||||
|
rc=1
|
||||||
|
fi
|
||||||
|
if (( rc == 0 )); then
|
||||||
|
mark_event case_end "${case_id}" COMPLETED
|
||||||
|
log "DONE Phase 2 fixed case=${case_id}"
|
||||||
|
else
|
||||||
|
mark_event case_end "${case_id}" FAILED
|
||||||
|
log "FAIL Phase 2 fixed case=${case_id}"
|
||||||
|
fi
|
||||||
|
sleep_if_real "${CASE_COOLDOWN_S}"
|
||||||
|
return "${rc}"
|
||||||
|
}
|
||||||
|
|
||||||
|
run_mixed_case() {
|
||||||
|
local phase_run_id="mixed_prefill_decode_interference"
|
||||||
|
local summary_path="${RESULT_DIR}/bench/${phase_run_id}/summary.csv"
|
||||||
|
local rc=0
|
||||||
|
|
||||||
|
mark_event case_start "${phase_run_id}"
|
||||||
|
log "START Phase 2 mixed Prefill/Decode A/B"
|
||||||
|
if run_phase1_action mixed "${phase_run_id}" ""; then
|
||||||
|
:
|
||||||
|
else
|
||||||
|
rc=$?
|
||||||
|
fi
|
||||||
|
if [[ "${DRY_RUN}" != "1" ]] \
|
||||||
|
&& ! python3 "${RESULT_TOOL}" check-bench --summary "${summary_path}"; then
|
||||||
|
rc=1
|
||||||
|
fi
|
||||||
|
if (( rc == 0 )); then
|
||||||
|
mark_event case_end "${phase_run_id}" COMPLETED
|
||||||
|
log "DONE Phase 2 mixed Prefill/Decode A/B"
|
||||||
|
else
|
||||||
|
mark_event case_end "${phase_run_id}" FAILED
|
||||||
|
log "FAIL Phase 2 mixed Prefill/Decode A/B"
|
||||||
|
fi
|
||||||
|
return "${rc}"
|
||||||
|
}
|
||||||
|
|
||||||
|
summarize_results() {
|
||||||
|
python3 "${RESULT_TOOL}" summarize "${RESULT_DIR}"
|
||||||
|
}
|
||||||
|
|
||||||
|
finish_manifest() {
|
||||||
|
local status="$1"
|
||||||
|
python3 "${RESULT_TOOL}" finish \
|
||||||
|
--path "${RESULT_DIR}/manifest.json" \
|
||||||
|
--status "${status}"
|
||||||
|
}
|
||||||
|
|
||||||
|
cleanup() {
|
||||||
|
local rc=$?
|
||||||
|
stop_collectors || true
|
||||||
|
stop_service || true
|
||||||
|
if (( rc != 0 )) && [[ -f "${RESULT_DIR}/manifest.json" ]]; then
|
||||||
|
finish_manifest ABORTED || true
|
||||||
|
fi
|
||||||
|
return "${rc}"
|
||||||
|
}
|
||||||
|
|
||||||
|
run_all() {
|
||||||
|
validate_config
|
||||||
|
enable_result_logging
|
||||||
|
mkdir -p \
|
||||||
|
"${RESULT_DIR}/head/collector_commands" \
|
||||||
|
"${RESULT_DIR}/worker/collector_commands" \
|
||||||
|
"${SERVICE_DIR}" \
|
||||||
|
"${COMMAND_DIR}"
|
||||||
|
printf '%s\n' 'wall_time_ns,node,collector,status,pid' \
|
||||||
|
> "${RESULT_DIR}/collector_status.csv"
|
||||||
|
write_manifest
|
||||||
|
|
||||||
|
if [[ "${DRY_RUN}" != "1" ]]; then
|
||||||
|
preflight_node_tools "${HEAD_NODE}"
|
||||||
|
preflight_node_tools "${WORKER_NODE}"
|
||||||
|
preflight_clock_sync
|
||||||
|
fi
|
||||||
|
|
||||||
|
trap cleanup EXIT INT TERM
|
||||||
|
start_service
|
||||||
|
capture_static_snapshots before
|
||||||
|
start_collectors
|
||||||
|
mark_event idle_start
|
||||||
|
sleep_if_real "${IDLE_BASELINE_S}"
|
||||||
|
mark_event idle_end
|
||||||
|
|
||||||
|
local failures=0
|
||||||
|
local service_available=1
|
||||||
|
local normalized="${FIXED_CASE_IDS//,/ }"
|
||||||
|
local case_id
|
||||||
|
for case_id in ${normalized}; do
|
||||||
|
run_fixed_case "${case_id}" || ((failures+=1))
|
||||||
|
if [[ "${DRY_RUN}" != "1" ]] \
|
||||||
|
&& ! curl --fail --silent --max-time 5 \
|
||||||
|
"http://${HEAD_NODE}:${SGLANG_PORT}/health" >/dev/null; then
|
||||||
|
log "ERROR: service became unhealthy after case=${case_id}"
|
||||||
|
((failures+=1))
|
||||||
|
service_available=0
|
||||||
|
break
|
||||||
|
fi
|
||||||
|
done
|
||||||
|
|
||||||
|
if [[ "${RUN_MIXED_CASE}" == "1" && "${service_available}" == "1" ]]; then
|
||||||
|
run_mixed_case || ((failures+=1))
|
||||||
|
elif [[ "${RUN_MIXED_CASE}" == "1" ]]; then
|
||||||
|
log "SKIP mixed Prefill/Decode A/B because the service is unhealthy"
|
||||||
|
fi
|
||||||
|
|
||||||
|
mark_event cooldown_start
|
||||||
|
sleep_if_real "${POST_RUN_COOLDOWN_S}"
|
||||||
|
mark_event cooldown_end
|
||||||
|
stop_collectors
|
||||||
|
capture_static_snapshots after
|
||||||
|
stop_service
|
||||||
|
summarize_results
|
||||||
|
|
||||||
|
local status="COMPLETED"
|
||||||
|
(( failures > 0 )) && status="COMPLETED_WITH_FAILURES"
|
||||||
|
[[ "${DRY_RUN}" == "1" ]] && status="DRY_RUN"
|
||||||
|
finish_manifest "${status}"
|
||||||
|
trap - EXIT INT TERM
|
||||||
|
log "Phase 2 complete: status=${status} result=${RESULT_DIR}"
|
||||||
|
(( failures == 0 ))
|
||||||
|
}
|
||||||
|
|
||||||
|
main() {
|
||||||
|
case "${ACTION}" in
|
||||||
|
all)
|
||||||
|
run_all
|
||||||
|
;;
|
||||||
|
summarize)
|
||||||
|
enable_result_logging
|
||||||
|
summarize_results
|
||||||
|
;;
|
||||||
|
stop)
|
||||||
|
enable_result_logging
|
||||||
|
stop_service
|
||||||
|
;;
|
||||||
|
*)
|
||||||
|
printf 'Usage: %s {all|summarize|stop}\n' "$0" >&2
|
||||||
|
return 2
|
||||||
|
;;
|
||||||
|
esac
|
||||||
|
}
|
||||||
|
|
||||||
|
main "$@"
|
||||||
@ -0,0 +1,197 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import csv
|
||||||
|
import json
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
EXPERIMENT_DIR = Path(__file__).resolve().parents[1]
|
||||||
|
sys.path.insert(0, str(EXPERIMENT_DIR))
|
||||||
|
|
||||||
|
import hardware_contention_attribution as attribution # noqa: E402
|
||||||
|
|
||||||
|
|
||||||
|
class HardwareContentionAttributionTest(unittest.TestCase):
|
||||||
|
def test_gpu_summary_ignores_na_and_computes_percentiles(self) -> None:
|
||||||
|
with tempfile.TemporaryDirectory() as temporary:
|
||||||
|
path = Path(temporary) / "gpu_samples.csv"
|
||||||
|
self._write_csv(
|
||||||
|
path,
|
||||||
|
attribution.GPU_FIELDS,
|
||||||
|
[
|
||||||
|
{
|
||||||
|
"wall_time_ns": "1",
|
||||||
|
"node": "head",
|
||||||
|
"gpu": "0",
|
||||||
|
"gpu_util_pct": "10",
|
||||||
|
"power_w": "100",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"wall_time_ns": "2",
|
||||||
|
"node": "head",
|
||||||
|
"gpu": "0",
|
||||||
|
"gpu_util_pct": "N/A",
|
||||||
|
"power_w": "200",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"wall_time_ns": "3",
|
||||||
|
"node": "head",
|
||||||
|
"gpu": "0",
|
||||||
|
"gpu_util_pct": "90",
|
||||||
|
"power_w": "300",
|
||||||
|
},
|
||||||
|
],
|
||||||
|
)
|
||||||
|
rows = attribution.summarize_gpu(path)
|
||||||
|
self.assertEqual(len(rows), 1)
|
||||||
|
self.assertEqual(rows[0]["samples"], 3)
|
||||||
|
self.assertEqual(rows[0]["gpu_util_pct_mean"], 50)
|
||||||
|
self.assertAlmostEqual(rows[0]["power_w_p95"], 290)
|
||||||
|
|
||||||
|
def test_rdma_summary_converts_four_octet_units(self) -> None:
|
||||||
|
with tempfile.TemporaryDirectory() as temporary:
|
||||||
|
path = Path(temporary) / "rdma.csv"
|
||||||
|
self._write_csv(
|
||||||
|
path,
|
||||||
|
attribution.RDMA_FIELDS,
|
||||||
|
[
|
||||||
|
{
|
||||||
|
"wall_time_ns": "1000000000",
|
||||||
|
"node": "worker",
|
||||||
|
"hca": "mlx5_0",
|
||||||
|
"port_xmit_data": "100",
|
||||||
|
"port_rcv_data": "200",
|
||||||
|
"port_xmit_wait": "2",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"wall_time_ns": "3000000000",
|
||||||
|
"node": "worker",
|
||||||
|
"hca": "mlx5_0",
|
||||||
|
"port_xmit_data": "1100",
|
||||||
|
"port_rcv_data": "2200",
|
||||||
|
"port_xmit_wait": "7",
|
||||||
|
},
|
||||||
|
],
|
||||||
|
)
|
||||||
|
rows = attribution.summarize_rdma(path)
|
||||||
|
self.assertEqual(rows[0]["xmit_bytes"], 4000)
|
||||||
|
self.assertEqual(rows[0]["rcv_bytes"], 8000)
|
||||||
|
self.assertEqual(rows[0]["port_xmit_wait_delta"], 5)
|
||||||
|
self.assertAlmostEqual(rows[0]["xmit_gbps"], 0.000016)
|
||||||
|
|
||||||
|
def test_marker_appends_header_once(self) -> None:
|
||||||
|
with tempfile.TemporaryDirectory() as temporary:
|
||||||
|
path = Path(temporary) / "markers.csv"
|
||||||
|
attribution.append_marker(path, "head", "case_start", "prefill", "")
|
||||||
|
attribution.append_marker(path, "head", "case_end", "prefill", "COMPLETED")
|
||||||
|
rows = attribution.read_csv_rows(path)
|
||||||
|
self.assertEqual(len(rows), 2)
|
||||||
|
self.assertEqual(rows[1]["status"], "COMPLETED")
|
||||||
|
|
||||||
|
def test_summarize_collects_nested_phase1_bench_rows(self) -> None:
|
||||||
|
with tempfile.TemporaryDirectory() as temporary:
|
||||||
|
result_dir = Path(temporary)
|
||||||
|
summary_path = result_dir / "bench" / "prefill" / "summary.csv"
|
||||||
|
self._write_csv(
|
||||||
|
summary_path,
|
||||||
|
["case_id", "status", "input_token_throughput", "output_token_throughput"],
|
||||||
|
[
|
||||||
|
{
|
||||||
|
"case_id": "long_prefill_latency_128k_c1",
|
||||||
|
"status": "COMPLETED",
|
||||||
|
"input_token_throughput": "2700",
|
||||||
|
"output_token_throughput": "0.02",
|
||||||
|
}
|
||||||
|
],
|
||||||
|
)
|
||||||
|
summary = attribution.summarize(result_dir)
|
||||||
|
self.assertEqual(summary["bench_rows"], 1)
|
||||||
|
self.assertEqual(summary["failed_bench_rows"], 0)
|
||||||
|
rows = attribution.read_csv_rows(result_dir / "bench_summary.csv")
|
||||||
|
self.assertEqual(rows[0]["phase2_bench_run"], "prefill")
|
||||||
|
self.assertTrue((result_dir / "report.md").exists())
|
||||||
|
|
||||||
|
def test_case_hardware_is_cut_by_phase1_meta_window(self) -> None:
|
||||||
|
with tempfile.TemporaryDirectory() as temporary:
|
||||||
|
result_dir = Path(temporary)
|
||||||
|
meta_path = (
|
||||||
|
result_dir
|
||||||
|
/ "bench"
|
||||||
|
/ "prefill"
|
||||||
|
/ "cases"
|
||||||
|
/ "long_prefill_latency_128k_c1"
|
||||||
|
/ "rep1"
|
||||||
|
/ "meta.json"
|
||||||
|
)
|
||||||
|
meta_path.parent.mkdir(parents=True)
|
||||||
|
meta_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"case_id": "long_prefill_latency_128k_c1",
|
||||||
|
"role": "",
|
||||||
|
"stage": "prefill_latency",
|
||||||
|
"repetition": 1,
|
||||||
|
"status": "COMPLETED",
|
||||||
|
"started_at": "2026-07-31T00:00:01+00:00",
|
||||||
|
"ended_at": "2026-07-31T00:00:03+00:00",
|
||||||
|
}
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
start_ns = attribution.iso_to_ns("2026-07-31T00:00:00+00:00")
|
||||||
|
self.assertIsNotNone(start_ns)
|
||||||
|
assert start_ns is not None
|
||||||
|
self._write_csv(
|
||||||
|
result_dir / "head" / "gpu_samples.csv",
|
||||||
|
attribution.GPU_FIELDS,
|
||||||
|
[
|
||||||
|
{
|
||||||
|
"wall_time_ns": str(start_ns),
|
||||||
|
"node": "head",
|
||||||
|
"gpu": "0",
|
||||||
|
"gpu_util_pct": "5",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"wall_time_ns": str(start_ns + 2_000_000_000),
|
||||||
|
"node": "head",
|
||||||
|
"gpu": "0",
|
||||||
|
"gpu_util_pct": "80",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"wall_time_ns": str(start_ns + 4_000_000_000),
|
||||||
|
"node": "head",
|
||||||
|
"gpu": "0",
|
||||||
|
"gpu_util_pct": "10",
|
||||||
|
},
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
windows = attribution.load_case_windows(result_dir)
|
||||||
|
gpu_rows, _ = attribution.summarize_case_hardware(result_dir, windows)
|
||||||
|
self.assertEqual(len(windows), 1)
|
||||||
|
self.assertEqual(len(gpu_rows), 1)
|
||||||
|
self.assertEqual(gpu_rows[0]["gpu_util_pct_mean"], 80)
|
||||||
|
self.assertEqual(
|
||||||
|
gpu_rows[0]["case_id"],
|
||||||
|
"long_prefill_latency_128k_c1",
|
||||||
|
)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _write_csv(
|
||||||
|
path: Path,
|
||||||
|
fields: list[str],
|
||||||
|
rows: list[dict[str, str]],
|
||||||
|
) -> None:
|
||||||
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
with path.open("w", newline="", encoding="utf-8") as handle:
|
||||||
|
writer = csv.DictWriter(handle, fieldnames=fields)
|
||||||
|
writer.writeheader()
|
||||||
|
for row in rows:
|
||||||
|
writer.writerow(row)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Loading…
x
Reference in New Issue
Block a user