diff --git a/src/code/issue8/README.md b/src/code/issue8/README.md new file mode 100644 index 0000000..253fb99 --- /dev/null +++ b/src/code/issue8/README.md @@ -0,0 +1,28 @@ +# Issue 8 单机可完成部分 + +本目录把 MoE Combine 三种模式拆成两层： + +- `combine_modes.py`：DeepEP 语义对应的 BF16/FP16 位级精度仿真、闭式流量模型； +- `run_analysis.py`：扫描 top-k 的 rank 重复率，生成 JSON、CSV 和模式决策表。 + +运行： + +```bash +python -m pytest src/code/issue8/test_combine_modes.py +python src/code/issue8/run_analysis.py +``` + +默认报告位于 `results/`。其中通信完成时间使用 +`固定时延 + 返回净荷 / 有效带宽` 模型，并明确标记为理论值。真实 DeepEP +combine 时延、NVLink/RDMA 流量和 kernel 路径仍需在多 rank GPU 环境验收。 + +常用参数： + +```bash +python src/code/issue8/run_analysis.py \ + --num-experts 64 --num-topk 8 --num-ranks 8 --hidden 7168 \ + --dtype bf16 --num-tokens 128 \ + --message-token-counts 1,8,32,128,512 \ + --concentrations 0.03,0.1,0.3,1,3,100 \ + --bandwidth-gbytes-s 50 --base-latency-us 4 +``` diff --git a/src/code/issue8/combine_modes.py b/src/code/issue8/combine_modes.py new file mode 100644 index 0000000..ada0567 --- /dev/null +++ b/src/code/issue8/combine_modes.py @@ -0,0 +1,349 @@ +""" +Copyright (c) 2026, TENCENT CORPORATION. All rights reserved. + +See LICENSE.txt for license information + +Content: Issue 8 -- MoE Combine 本地 Reduce 三种模式的精度/流量权衡模型 + +本模块给出 DeepEP V2 direct combine 三种工作模式的 + (1) 网络返回流量的闭式模型与经验统计 + (2) 浮点归约误差的**精确位级仿真** + +误差仿真是硬件无关的: 三种模式的差异完全来自 +"在哪一步把 fp32 中间和舍入回 bf16", 与 GPU 型号/互联无关. +因此本模块的精度结论在任意机器上可复现, 无需 GPU 集群. + +语义依据 (DeepEP 源码): + deep_ep/utils/refs.py::combine / grouped_reduce / ordered_accumulate + - grouped_reduce 在组内用 fp32 累加 (cur_accum_buf 为 float32), + 但在 segment break 处 `.to(data_to_reduce.dtype)` 存回 bf16 <-- 模式 B 的额外舍入 + - ordered_accumulate 沿 num_topk 维按序 fp32 累加, 最后 `.to(data.dtype)` + - 本地分组键为 `topk_idx // num_experts_per_rank` + deep_ep/buffers/elastic.py::get_theoretical_num_sms::get_expected_topk + - 期望命中 rank 数的组合数公式, 本模块复用同一公式做闭式流量模型 +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass +from typing import Dict, Tuple + +import numpy as np + +# --------------------------------------------------------------------------- +# 低精度类型的精确仿真 +# --------------------------------------------------------------------------- + + +def round_to_bf16(x: np.ndarray) -> np.ndarray: + """把 float32 数组按 round-to-nearest-even 舍入到 bfloat16 的可表示值. + + bf16 = float32 的高 16 bit (1 符号 + 8 指数 + 7 尾数). + 返回值仍是 float32 dtype, 但取值集合等于 bf16 可表示集合, 便于后续 fp32 累加. + numpy 无原生 bf16, 这里用位操作实现, 与硬件 RNE 行为一致. + """ + x = np.asarray(x, dtype=np.float32) + u = x.view(np.uint32) + + # NaN 需原样保留: 加 rounding bias 可能把 NaN 变成 Inf + is_nan = np.isnan(x) + + lsb = (u >> np.uint32(16)) & np.uint32(1) # 目标尾数最低位 + bias = np.uint32(0x7FFF) + lsb # RNE: 0x7FFF + lsb + rounded = (u + bias) & np.uint32(0xFFFF0000) + + out = rounded.view(np.float32).copy() + out[is_nan] = np.nan + return out + + +def round_to_fp16(x: np.ndarray) -> np.ndarray: + """舍入到 float16 可表示值, 返回 float32 dtype.""" + return np.asarray(x, dtype=np.float32).astype(np.float16).astype(np.float32) + + +DTYPE_ROUNDERS = { + "bf16": round_to_bf16, + "fp16": round_to_fp16, + "fp32": lambda x: np.asarray(x, dtype=np.float32), +} + +DTYPE_BYTES = {"bf16": 2, "fp16": 2, "fp32": 4} + + +# --------------------------------------------------------------------------- +# 路由生成: 可控"重复率" +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class EPConfig: + """一组 Expert-Parallel 配置.""" + + num_experts: int # E, 全局专家数 + num_topk: int # K, 每 token 选择的专家数 + num_ranks: int # R, EP 域内 rank 数 + hidden: int # H, 隐藏维 + dtype: str = "bf16" # 网络回传的数据类型 + + def __post_init__(self) -> None: + if self.num_experts % self.num_ranks != 0: + raise ValueError("num_experts 必须能被 num_ranks 整除") + if not 1 <= self.num_topk <= self.num_experts: + raise ValueError("num_topk 超出范围") + if self.dtype not in DTYPE_ROUNDERS: + raise ValueError(f"不支持的 dtype: {self.dtype}") + + @property + def experts_per_rank(self) -> int: + return self.num_experts // self.num_ranks + + +def expected_distinct_ranks(cfg: EPConfig) -> float: + """均匀路由下, 一个 token 的 top-K 命中的期望 **不同 rank 数** D. + + 复用 DeepEP `get_theoretical_num_sms` 内部 `get_expected_topk` 的组合数公式: + + E[D] = R * (1 - C(E - E/R, K) / C(E, K)) + + 推导: 对某个固定 rank r, 它**不被命中**当且仅当 K 个专家全部落在 + 其余 (E - E/R) 个专家中, 概率为 C(E-E/R, K)/C(E, K). + 对 R 个 rank 求和并用期望的线性性即得. + """ + e, k, r = cfg.num_experts, cfg.num_topk, cfg.num_ranks + if r == 1: + return 1.0 + miss = math.comb(e - e // r, k) / math.comb(e, k) if e - e // r >= k else 0.0 + return r * (1.0 - miss) + + +def sample_topk_idx( + cfg: EPConfig, + num_tokens: int, + concentration: float, + rng: np.random.Generator, +) -> np.ndarray: + """采样 `[num_tokens, num_topk]` 的专家索引, 通过 `concentration` 控制重复率. + + 每个 token 先从 Dirichlet(concentration * 1_R) 抽一组 rank 偏好权重, + 再在该权重下用 Gumbel-top-k 无放回地抽 K 个不同专家. + + concentration -> +inf : 退化为均匀路由, D 接近 expected_distinct_ranks + concentration -> 0 : top-K 高度集中于少数 rank, D -> 1 (重复率最高) + + 这样可以在同一份代码里连续扫描"重复率", 满足 README 中 + "构造可控的 top-k 分布(如不同的重复率), 以放大三种模式的差异". + """ + r, epr, k = cfg.num_ranks, cfg.experts_per_rank, cfg.num_topk + + if r == 1: + rank_w = np.ones((num_tokens, 1), dtype=np.float64) + else: + rank_w = rng.dirichlet(np.full(r, concentration), size=num_tokens) + + # 把 rank 权重摊到该 rank 的每个专家上 + expert_logp = np.log(np.repeat(rank_w, epr, axis=1) / epr + 1e-300) + + # Gumbel-top-k == 按权重无放回抽样 + gumbel = rng.gumbel(size=(num_tokens, cfg.num_experts)) + return np.argsort(-(expert_logp + gumbel), axis=1)[:, :k].astype(np.int64) + + +def rank_of_expert(topk_idx: np.ndarray, cfg: EPConfig) -> np.ndarray: + """专家索引 -> 所属 rank, 与 DeepEP `topk_idx // num_experts_per_rank` 一致.""" + return topk_idx // cfg.experts_per_rank + + +def duplicate_stats(topk_idx: np.ndarray, cfg: EPConfig) -> Dict[str, float]: + """统计实际重复率. + + duplicate_rate := 1 - D/K, 即"可被本地合并掉的副本占比". + D == K -> 0 (无重复, 本地 reduce 无收益) + D == 1 -> 1-1/K(全部落在同一 rank, 本地 reduce 收益最大) + """ + ranks = rank_of_expert(topk_idx, cfg) + d_per_token = np.array([len(np.unique(row)) for row in ranks], dtype=np.float64) + k = cfg.num_topk + return { + "mean_distinct_ranks": float(d_per_token.mean()), + "duplicate_rate": float(1.0 - d_per_token.mean() / k), + "frac_tokens_single_source": float((d_per_token == 1).mean()), + } + + +# --------------------------------------------------------------------------- +# 三种 combine 模式的位级仿真 +# --------------------------------------------------------------------------- +# +# 记 y[t, i] 为 token t 的第 i 个专家 (i = 0..K-1) 的 fp32 精确输出, +# r(t, i) 为该专家所属 rank, w[t, i] 为 gating 权重. +# +# 模式 A (no-expand / 无需本地 reduce) +# 专家侧 GEMM epilogue 直接在 fp32 里把本 rank 的若干贡献累加完, +# 每个 (token, rank) 只产生 **一个** bf16 值回传. +# partial_r = round(Σ_{i: r(t,i)=r} w·y) <- 输入未被单独舍入 +# result = round(Σ_r partial_r) (fp32 累加) +# 流量 = D 份 +# +# 模式 B (expand + allow_multiple_reduction) +# 专家输出先按 (token, expert) 展开并**物化为 bf16**, 再由一个独立的 +# shared memory 归约 pass 在 fp32 里合并, 但在 segment break 处存回 bf16. +# partial_r = round(Σ_{i: r(t,i)=r} round(w·y)) <- 比 A 多一次输入舍入 +# result = round(Σ_r partial_r) +# 流量 = D 份 (与 A 相同) +# +# 模式 C (expanded send, 不允许本地 reduce) +# 每个副本各自回传, 由 epilogue 在最终位置做 fp32 归约. +# result = round(Σ_i round(w·y)) +# 流量 = K 份 (最多) +# +# 由此可预期: 重复率较高时误差 A < C < B (重复率低时 A 与 C 的差距在 +# 噪声量级, 个别扫描点上 C 可略优于 A), 而流量 A = B < C. +# 即 **B 在流量上不优于 A, 在精度上劣于 A** -- B 的存在价值只在于 +# 上层被迫使用 expand 布局 (如 FP8 dispatch / per-expert scale factor, +# 或 GEMM epilogue 无法融合归约) 的场景. + + +def _weighted(y: np.ndarray, w: np.ndarray) -> np.ndarray: + """按 gating 权重加权, 全程 fp32.""" + return (y.astype(np.float32) * w.astype(np.float32)[:, :, None]).astype(np.float32) + + +def combine_mode_a(y, w, ranks, cfg) -> np.ndarray: + """模式 A: 无需本地 reduce, 专家侧 fp32 融合, 每 (token,rank) 回传一份.""" + rnd = DTYPE_ROUNDERS[cfg.dtype] + wy = _weighted(y, w) + num_tokens = y.shape[0] + out = np.zeros((num_tokens, cfg.hidden), dtype=np.float32) + + for t in range(num_tokens): + acc = np.zeros(cfg.hidden, dtype=np.float32) + for r in np.unique(ranks[t]): + # 本 rank 的贡献在 GEMM epilogue 内以 fp32 累加, 只舍入一次 + partial = np.zeros(cfg.hidden, dtype=np.float32) + for i in np.flatnonzero(ranks[t] == r): + partial += wy[t, i] + acc += rnd(partial) # 回传的是 bf16, 收端 fp32 累加 + out[t] = rnd(acc) + return out + + +def combine_mode_b(y, w, ranks, cfg) -> np.ndarray: + """模式 B: expand + 本地 shared-memory 归约, 输入先物化为 bf16.""" + rnd = DTYPE_ROUNDERS[cfg.dtype] + wy = _weighted(y, w) + num_tokens = y.shape[0] + out = np.zeros((num_tokens, cfg.hidden), dtype=np.float32) + + for t in range(num_tokens): + acc = np.zeros(cfg.hidden, dtype=np.float32) + for r in np.unique(ranks[t]): + partial = np.zeros(cfg.hidden, dtype=np.float32) + # 按 topk 序严格累加, 与 refs.py::grouped_reduce 一致 + for i in np.flatnonzero(ranks[t] == r): + partial += rnd(wy[t, i]) # <- 输入已是 bf16 (expand 布局物化) + acc += rnd(partial) # <- segment break 处存回 bf16 + out[t] = rnd(acc) + return out + + +def combine_mode_c(y, w, ranks, cfg) -> np.ndarray: + """模式 C: 每个副本各自回传, 收端 epilogue 做 fp32 归约.""" + rnd = DTYPE_ROUNDERS[cfg.dtype] + wy = _weighted(y, w) + num_tokens = y.shape[0] + out = np.zeros((num_tokens, cfg.hidden), dtype=np.float32) + + for t in range(num_tokens): + acc = np.zeros(cfg.hidden, dtype=np.float32) + for i in range(cfg.num_topk): + acc += rnd(wy[t, i]) # 每份副本单独舍入后回传 + out[t] = rnd(acc) + return out + + +def combine_reference_fp64(y, w, cfg) -> np.ndarray: + """高精度基线: fp64 全量回传后归约, 作为误差度量的真值.""" + wy = y.astype(np.float64) * w.astype(np.float64)[:, :, None] + return wy.sum(axis=1) + + +COMBINE_MODES = { + "A_no_expand": combine_mode_a, + "B_expand_local_reduce": combine_mode_b, + "C_expanded_send": combine_mode_c, +} + + +# --------------------------------------------------------------------------- +# 流量模型 +# --------------------------------------------------------------------------- + + +def traffic_bytes_per_token(mode: str, mean_distinct_ranks: float, cfg: EPConfig) -> float: + """单 token 的 combine 网络**返回**流量 (字节). + + 返回含本地份额的总流量 (D 份或 K 份净荷); 落在本地 rank 的一份也计入. + (若只计跨 rank 部分, 乘以 (1 - 1/R) 扣除本地份额, 见 traffic_bytes_per_token_remote_only) + """ + nbytes = DTYPE_BYTES[cfg.dtype] * cfg.hidden + if mode == "A_no_expand": + return mean_distinct_ranks * nbytes + if mode == "B_expand_local_reduce": + return mean_distinct_ranks * nbytes + if mode == "C_expanded_send": + return cfg.num_topk * nbytes + raise ValueError(f"未知模式: {mode}") + + +def traffic_bytes_per_token_remote_only( + mode: str, mean_distinct_ranks: float, cfg: EPConfig +) -> float: + """只计真正跨 rank 的部分 (扣除落在本地 rank 的一份).""" + total = traffic_bytes_per_token(mode, mean_distinct_ranks, cfg) + local_frac = 1.0 / cfg.num_ranks if cfg.num_ranks > 1 else 1.0 + return total * (1.0 - local_frac) + + +def analytic_traffic_ratio_b_over_c(cfg: EPConfig) -> float: + """闭式: 模式 B/A 相对模式 C 的流量比 = E[D]/K (均匀路由).""" + return expected_distinct_ranks(cfg) / cfg.num_topk + + +# --------------------------------------------------------------------------- +# 误差度量 +# --------------------------------------------------------------------------- + + +def error_metrics(approx: np.ndarray, exact: np.ndarray) -> Dict[str, float]: + """相对高精度基线的误差指标.""" + a = approx.astype(np.float64) + e = exact.astype(np.float64) + diff = a - e + denom = np.maximum(np.abs(e), 1e-30) + + return { + "max_abs_err": float(np.abs(diff).max()), + "mean_abs_err": float(np.abs(diff).mean()), + "rmse": float(np.sqrt((diff ** 2).mean())), + "max_rel_err": float((np.abs(diff) / denom).max()), + "mean_rel_err": float((np.abs(diff) / denom).mean()), + # 相对 L2, 训练/推理中最常用的整体偏差度量 + "rel_l2": float(np.linalg.norm(diff) / max(np.linalg.norm(e), 1e-30)), + } + + +def make_expert_outputs( + cfg: EPConfig, num_tokens: int, rng: np.random.Generator +) -> Tuple[np.ndarray, np.ndarray]: + """生成专家输出 y 与 gating 权重 w. + + y ~ N(0, 1) 的量级贴近实际 MoE FFN 输出 (已过 LayerNorm 的激活); + w 为 softmax 后归一化的 top-k 权重, 和为 1. + """ + y = rng.standard_normal((num_tokens, cfg.num_topk, cfg.hidden)).astype(np.float32) + logits = rng.standard_normal((num_tokens, cfg.num_topk)).astype(np.float32) + w = np.exp(logits - logits.max(axis=1, keepdims=True)) + w /= w.sum(axis=1, keepdims=True) + return y, w.astype(np.float32) diff --git a/src/code/issue8/results/combine_modes_report.csv b/src/code/issue8/results/combine_modes_report.csv new file mode 100644 index 0000000..3798a72 --- /dev/null +++ b/src/code/issue8/results/combine_modes_report.csv @@ -0,0 +1,19 @@ +concentration,duplicate_rate,mean_distinct_ranks,mode,remote_bytes_per_token,modeled_completion_us,max_abs_err,max_rel_err,rel_l2,selected +0.03,0.791015625,1.671875,A_no_expand,20972.0,57.68832,0.014623181647990258,18.729322809140758,0.0020947050945017242,False +0.03,0.791015625,1.671875,B_expand_local_reduce,20972.0,57.68832,0.021247879104722145,6811.479372779756,0.002663171507585347,True +0.03,0.791015625,1.671875,C_expanded_send,100352.0,260.90112,0.015063938152286127,6811.479372779756,0.0023482991084338403,False +0.1,0.69921875,2.40625,A_no_expand,30184.0,81.27104,0.015006233742157704,139.88551247079022,0.0023059962973855878,False +0.1,0.69921875,2.40625,B_expand_local_reduce,30184.0,81.27104,0.020200299148202205,1740.324120397721,0.002793795232514089,True +0.1,0.69921875,2.40625,C_expanded_send,100352.0,260.90112,0.015264159576808822,1504.519812427196,0.0023487519389937895,False +0.3,0.5947265625,3.2421875,A_no_expand,40670.0,108.1152,0.014273319164415543,2663.063890522244,0.0023637365328187474,False +0.3,0.5947265625,3.2421875,B_expand_local_reduce,40670.0,108.1152,0.015782268199267646,2663.063890522244,0.002836941530286989,True +0.3,0.5947265625,3.2421875,C_expanded_send,100352.0,260.90112,0.011175704506814377,1668.2833939608795,0.0023495517489655646,False +1.0,0.4453125,4.4375,A_no_expand,55664.0,146.49984,0.012882646385472096,1697.8155411114744,0.0023557113431409457,False +1.0,0.4453125,4.4375,B_expand_local_reduce,55664.0,146.49984,0.017827402294474215,1697.8155411114744,0.0027501812253815728,True +1.0,0.4453125,4.4375,C_expanded_send,100352.0,260.90112,0.013979853436700829,1697.8155411114744,0.0023501156496355848,False +3.0,0.36328125,5.09375,A_no_expand,63896.0,167.57376,0.01389796873585869,63728.14419306846,0.002351836543482734,False +3.0,0.36328125,5.09375,B_expand_local_reduce,63896.0,167.57376,0.018520905944202415,63728.14419306846,0.0026986536255635516,True +3.0,0.36328125,5.09375,C_expanded_send,100352.0,260.90112,0.014271232858274363,39829.71512066779,0.002348676372981419,False +100.0,0.3154296875,5.4765625,A_no_expand,68698.0,179.86688,0.015120669336922354,7742.939159190431,0.0023445600358013327,False +100.0,0.3154296875,5.4765625,B_expand_local_reduce,68698.0,179.86688,0.018771920356360106,17031.26615021895,0.0026497314569245713,True +100.0,0.3154296875,5.4765625,C_expanded_send,100352.0,260.90112,0.015120669336922354,6579.648285311867,0.002341408698411671,False diff --git a/src/code/issue8/results/combine_modes_report.json b/src/code/issue8/results/combine_modes_report.json new file mode 100644 index 0000000..84b54b0 --- /dev/null +++ b/src/code/issue8/results/combine_modes_report.json @@ -0,0 +1,621 @@ +{ + "schema_version": 1, + "scope": "single-machine analytical model and bit-exact low-precision simulation", + "configuration": { + "num_experts": 64, + "num_topk": 8, + "num_ranks": 8, + "hidden": 7168, + "dtype": "bf16", + "num_tokens": 128, + "message_token_counts": [ + 1, + 8, + 32, + 128, + 512 + ], + "seed": 20260726 + }, + "time_model": { + "bandwidth_gbytes_s": 50.0, + "base_latency_us": 4.0, + "formula": "T = base_latency + remote_payload_bytes / effective_bandwidth", + "warning": "Modeled values are not DeepEP measurements and require multi-rank validation." + }, + "uniform_routing_closed_form": { + "expected_distinct_ranks": 5.432551281938457, + "local_reduce_to_expanded_traffic_ratio": 0.6790689102423071 + }, + "decision_precision_limit_rel_l2": 0.003, + "scenarios": [ + { + "concentration": 0.03, + "routing": { + "mean_distinct_ranks": 1.671875, + "duplicate_rate": 0.791015625, + "frac_tokens_single_source": 0.4453125 + }, + "modes": { + "A_no_expand": { + "traffic_bytes_per_token": 23968.0, + "remote_traffic_bytes_per_token": 20972.0, + "remote_traffic_bytes_total": 2684416.0, + "modeled_completion_us": 57.68832, + "error": { + "max_abs_err": 0.014623181647990258, + "mean_abs_err": 0.0006604867387258645, + "rmse": 0.0009992996485000945, + "max_rel_err": 18.729322809140758, + "mean_rel_err": 0.003384216033422905, + "rel_l2": 0.0020947050945017242 + } + }, + "B_expand_local_reduce": { + "traffic_bytes_per_token": 23968.0, + "remote_traffic_bytes_per_token": 20972.0, + "remote_traffic_bytes_total": 2684416.0, + "modeled_completion_us": 57.68832, + "error": { + "max_abs_err": 0.021247879104722145, + "mean_abs_err": 0.0008947316632411222, + "rmse": 0.001270492136774298, + "max_rel_err": 6811.479372779756, + "mean_rel_err": 0.01958657697157897, + "rel_l2": 0.002663171507585347 + } + }, + "C_expanded_send": { + "traffic_bytes_per_token": 114688, + "remote_traffic_bytes_per_token": 100352.0, + "remote_traffic_bytes_total": 12845056.0, + "modeled_completion_us": 260.90112, + "error": { + "max_abs_err": 0.015063938152286127, + "mean_abs_err": 0.0008009563078528245, + "rmse": 0.0011202791647333202, + "max_rel_err": 6811.479372779756, + "mean_rel_err": 0.018563369093327755, + "rel_l2": 0.0023482991084338403 + } + } + }, + "decision": { + "mode": "B_expand_local_reduce", + "reason": "Expanded layout is required and local reduction meets the configured precision limit." + }, + "message_size_sweep": [ + { + "tokens": 1, + "modeled_completion_us": { + "A_no_expand": 4.41944, + "B_expand_local_reduce": 4.41944, + "C_expanded_send": 6.00704 + } + }, + { + "tokens": 8, + "modeled_completion_us": { + "A_no_expand": 7.35552, + "B_expand_local_reduce": 7.35552, + "C_expanded_send": 20.05632 + } + }, + { + "tokens": 32, + "modeled_completion_us": { + "A_no_expand": 17.42208, + "B_expand_local_reduce": 17.42208, + "C_expanded_send": 68.22528 + } + }, + { + "tokens": 128, + "modeled_completion_us": { + "A_no_expand": 57.68832, + "B_expand_local_reduce": 57.68832, + "C_expanded_send": 260.90112 + } + }, + { + "tokens": 512, + "modeled_completion_us": { + "A_no_expand": 218.75328, + "B_expand_local_reduce": 218.75328, + "C_expanded_send": 1031.60448 + } + } + ] + }, + { + "concentration": 0.1, + "routing": { + "mean_distinct_ranks": 2.40625, + "duplicate_rate": 0.69921875, + "frac_tokens_single_source": 0.1171875 + }, + "modes": { + "A_no_expand": { + "traffic_bytes_per_token": 34496.0, + "remote_traffic_bytes_per_token": 30184.0, + "remote_traffic_bytes_total": 3863552.0, + "modeled_completion_us": 81.27104, + "error": { + "max_abs_err": 0.015006233742157704, + "mean_abs_err": 0.0007628320043361924, + "rmse": 0.001126138717926693, + "max_rel_err": 139.88551247079022, + "mean_rel_err": 0.00569926033898308, + "rel_l2": 0.0023059962973855878 + } + }, + "B_expand_local_reduce": { + "traffic_bytes_per_token": 34496.0, + "remote_traffic_bytes_per_token": 30184.0, + "remote_traffic_bytes_total": 3863552.0, + "modeled_completion_us": 81.27104, + "error": { + "max_abs_err": 0.020200299148202205, + "mean_abs_err": 0.0009603251253420326, + "rmse": 0.0013643564757064503, + "max_rel_err": 1740.324120397721, + "mean_rel_err": 0.015319257920091917, + "rel_l2": 0.002793795232514089 + } + }, + "C_expanded_send": { + "traffic_bytes_per_token": 114688, + "remote_traffic_bytes_per_token": 100352.0, + "remote_traffic_bytes_total": 12845056.0, + "modeled_completion_us": 260.90112, + "error": { + "max_abs_err": 0.015264159576808822, + "mean_abs_err": 0.0008168594697258431, + "rmse": 0.0011470185361117363, + "max_rel_err": 1504.519812427196, + "mean_rel_err": 0.013255288194215445, + "rel_l2": 0.0023487519389937895 + } + } + }, + "decision": { + "mode": "B_expand_local_reduce", + "reason": "Expanded layout is required and local reduction meets the configured precision limit." + }, + "message_size_sweep": [ + { + "tokens": 1, + "modeled_completion_us": { + "A_no_expand": 4.60368, + "B_expand_local_reduce": 4.60368, + "C_expanded_send": 6.00704 + } + }, + { + "tokens": 8, + "modeled_completion_us": { + "A_no_expand": 8.82944, + "B_expand_local_reduce": 8.82944, + "C_expanded_send": 20.05632 + } + }, + { + "tokens": 32, + "modeled_completion_us": { + "A_no_expand": 23.31776, + "B_expand_local_reduce": 23.31776, + "C_expanded_send": 68.22528 + } + }, + { + "tokens": 128, + "modeled_completion_us": { + "A_no_expand": 81.27104, + "B_expand_local_reduce": 81.27104, + "C_expanded_send": 260.90112 + } + }, + { + "tokens": 512, + "modeled_completion_us": { + "A_no_expand": 313.08416, + "B_expand_local_reduce": 313.08416, + "C_expanded_send": 1031.60448 + } + } + ] + }, + { + "concentration": 0.3, + "routing": { + "mean_distinct_ranks": 3.2421875, + "duplicate_rate": 0.5947265625, + "frac_tokens_single_source": 0.0078125 + }, + "modes": { + "A_no_expand": { + "traffic_bytes_per_token": 46480.0, + "remote_traffic_bytes_per_token": 40670.0, + "remote_traffic_bytes_total": 5205760.0, + "modeled_completion_us": 108.1152, + "error": { + "max_abs_err": 0.014273319164415543, + "mean_abs_err": 0.0007619078414328315, + "rmse": 0.0010945442524782788, + "max_rel_err": 2663.063890522244, + "mean_rel_err": 0.01082043582414753, + "rel_l2": 0.0023637365328187474 + } + }, + "B_expand_local_reduce": { + "traffic_bytes_per_token": 46480.0, + "remote_traffic_bytes_per_token": 40670.0, + "remote_traffic_bytes_total": 5205760.0, + "modeled_completion_us": 108.1152, + "error": { + "max_abs_err": 0.015782268199267646, + "mean_abs_err": 0.0009421611354044202, + "rmse": 0.0013136650398552106, + "max_rel_err": 2663.063890522244, + "mean_rel_err": 0.018047820709736127, + "rel_l2": 0.002836941530286989 + } + }, + "C_expanded_send": { + "traffic_bytes_per_token": 114688, + "remote_traffic_bytes_per_token": 100352.0, + "remote_traffic_bytes_total": 12845056.0, + "modeled_completion_us": 260.90112, + "error": { + "max_abs_err": 0.011175704506814377, + "mean_abs_err": 0.0007879814635449488, + "rmse": 0.0010879758919932658, + "max_rel_err": 1668.2833939608795, + "mean_rel_err": 0.013805149759828005, + "rel_l2": 0.0023495517489655646 + } + } + }, + "decision": { + "mode": "B_expand_local_reduce", + "reason": "Expanded layout is required and local reduction meets the configured precision limit." + }, + "message_size_sweep": [ + { + "tokens": 1, + "modeled_completion_us": { + "A_no_expand": 4.8134, + "B_expand_local_reduce": 4.8134, + "C_expanded_send": 6.00704 + } + }, + { + "tokens": 8, + "modeled_completion_us": { + "A_no_expand": 10.507200000000001, + "B_expand_local_reduce": 10.507200000000001, + "C_expanded_send": 20.05632 + } + }, + { + "tokens": 32, + "modeled_completion_us": { + "A_no_expand": 30.0288, + "B_expand_local_reduce": 30.0288, + "C_expanded_send": 68.22528 + } + }, + { + "tokens": 128, + "modeled_completion_us": { + "A_no_expand": 108.1152, + "B_expand_local_reduce": 108.1152, + "C_expanded_send": 260.90112 + } + }, + { + "tokens": 512, + "modeled_completion_us": { + "A_no_expand": 420.4608, + "B_expand_local_reduce": 420.4608, + "C_expanded_send": 1031.60448 + } + } + ] + }, + { + "concentration": 1.0, + "routing": { + "mean_distinct_ranks": 4.4375, + "duplicate_rate": 0.4453125, + "frac_tokens_single_source": 0.0 + }, + "modes": { + "A_no_expand": { + "traffic_bytes_per_token": 63616.0, + "remote_traffic_bytes_per_token": 55664.0, + "remote_traffic_bytes_total": 7124992.0, + "modeled_completion_us": 146.49984, + "error": { + "max_abs_err": 0.012882646385472096, + "mean_abs_err": 0.0007853210905335799, + "rmse": 0.0011137163984737716, + "max_rel_err": 1697.8155411114744, + "mean_rel_err": 0.01168384908703549, + "rel_l2": 0.0023557113431409457 + } + }, + "B_expand_local_reduce": { + "traffic_bytes_per_token": 63616.0, + "remote_traffic_bytes_per_token": 55664.0, + "remote_traffic_bytes_total": 7124992.0, + "modeled_completion_us": 146.49984, + "error": { + "max_abs_err": 0.017827402294474215, + "mean_abs_err": 0.0009304010861366905, + "rmse": 0.0013002110544657213, + "max_rel_err": 1697.8155411114744, + "mean_rel_err": 0.01714099276864512, + "rel_l2": 0.0027501812253815728 + } + }, + "C_expanded_send": { + "traffic_bytes_per_token": 114688, + "remote_traffic_bytes_per_token": 100352.0, + "remote_traffic_bytes_total": 12845056.0, + "modeled_completion_us": 260.90112, + "error": { + "max_abs_err": 0.013979853436700829, + "mean_abs_err": 0.0007975605339849717, + "rmse": 0.001111070906429129, + "max_rel_err": 1697.8155411114744, + "mean_rel_err": 0.013948519045465466, + "rel_l2": 0.0023501156496355848 + } + } + }, + "decision": { + "mode": "B_expand_local_reduce", + "reason": "Expanded layout is required and local reduction meets the configured precision limit." + }, + "message_size_sweep": [ + { + "tokens": 1, + "modeled_completion_us": { + "A_no_expand": 5.11328, + "B_expand_local_reduce": 5.11328, + "C_expanded_send": 6.00704 + } + }, + { + "tokens": 8, + "modeled_completion_us": { + "A_no_expand": 12.90624, + "B_expand_local_reduce": 12.90624, + "C_expanded_send": 20.05632 + } + }, + { + "tokens": 32, + "modeled_completion_us": { + "A_no_expand": 39.62496, + "B_expand_local_reduce": 39.62496, + "C_expanded_send": 68.22528 + } + }, + { + "tokens": 128, + "modeled_completion_us": { + "A_no_expand": 146.49984, + "B_expand_local_reduce": 146.49984, + "C_expanded_send": 260.90112 + } + }, + { + "tokens": 512, + "modeled_completion_us": { + "A_no_expand": 573.99936, + "B_expand_local_reduce": 573.99936, + "C_expanded_send": 1031.60448 + } + } + ] + }, + { + "concentration": 3.0, + "routing": { + "mean_distinct_ranks": 5.09375, + "duplicate_rate": 0.36328125, + "frac_tokens_single_source": 0.0 + }, + "modes": { + "A_no_expand": { + "traffic_bytes_per_token": 73024.0, + "remote_traffic_bytes_per_token": 63896.0, + "remote_traffic_bytes_total": 8178688.0, + "modeled_completion_us": 167.57376, + "error": { + "max_abs_err": 0.01389796873585869, + "mean_abs_err": 0.0007921864376599693, + "rmse": 0.001120201585741701, + "max_rel_err": 63728.14419306846, + "mean_rel_err": 0.07971652216077245, + "rel_l2": 0.002351836543482734 + } + }, + "B_expand_local_reduce": { + "traffic_bytes_per_token": 73024.0, + "remote_traffic_bytes_per_token": 63896.0, + "remote_traffic_bytes_total": 8178688.0, + "modeled_completion_us": 167.57376, + "error": { + "max_abs_err": 0.018520905944202415, + "mean_abs_err": 0.0009210769776609645, + "rmse": 0.0012853937826168803, + "max_rel_err": 63728.14419306846, + "mean_rel_err": 0.08213816652711273, + "rel_l2": 0.0026986536255635516 + } + }, + "C_expanded_send": { + "traffic_bytes_per_token": 114688, + "remote_traffic_bytes_per_token": 100352.0, + "remote_traffic_bytes_total": 12845056.0, + "modeled_completion_us": 260.90112, + "error": { + "max_abs_err": 0.014271232858274363, + "mean_abs_err": 0.0008020671065987004, + "rmse": 0.001118696367185336, + "max_rel_err": 39829.71512066779, + "mean_rel_err": 0.05409767408787863, + "rel_l2": 0.002348676372981419 + } + } + }, + "decision": { + "mode": "B_expand_local_reduce", + "reason": "Expanded layout is required and local reduction meets the configured precision limit." + }, + "message_size_sweep": [ + { + "tokens": 1, + "modeled_completion_us": { + "A_no_expand": 5.27792, + "B_expand_local_reduce": 5.27792, + "C_expanded_send": 6.00704 + } + }, + { + "tokens": 8, + "modeled_completion_us": { + "A_no_expand": 14.22336, + "B_expand_local_reduce": 14.22336, + "C_expanded_send": 20.05632 + } + }, + { + "tokens": 32, + "modeled_completion_us": { + "A_no_expand": 44.89344, + "B_expand_local_reduce": 44.89344, + "C_expanded_send": 68.22528 + } + }, + { + "tokens": 128, + "modeled_completion_us": { + "A_no_expand": 167.57376, + "B_expand_local_reduce": 167.57376, + "C_expanded_send": 260.90112 + } + }, + { + "tokens": 512, + "modeled_completion_us": { + "A_no_expand": 658.29504, + "B_expand_local_reduce": 658.29504, + "C_expanded_send": 1031.60448 + } + } + ] + }, + { + "concentration": 100.0, + "routing": { + "mean_distinct_ranks": 5.4765625, + "duplicate_rate": 0.3154296875, + "frac_tokens_single_source": 0.0 + }, + "modes": { + "A_no_expand": { + "traffic_bytes_per_token": 78512.0, + "remote_traffic_bytes_per_token": 68698.0, + "remote_traffic_bytes_total": 8793344.0, + "modeled_completion_us": 179.86688, + "error": { + "max_abs_err": 0.015120669336922354, + "mean_abs_err": 0.0008112743179897209, + "rmse": 0.0011599690901076842, + "max_rel_err": 7742.939159190431, + "mean_rel_err": 0.022129422100567443, + "rel_l2": 0.0023445600358013327 + } + }, + "B_expand_local_reduce": { + "traffic_bytes_per_token": 78512.0, + "remote_traffic_bytes_per_token": 68698.0, + "remote_traffic_bytes_total": 8793344.0, + "modeled_completion_us": 179.86688, + "error": { + "max_abs_err": 0.018771920356360106, + "mean_abs_err": 0.0009232002011770004, + "rmse": 0.0013109523919987804, + "max_rel_err": 17031.26615021895, + "mean_rel_err": 0.03638629082824786, + "rel_l2": 0.0026497314569245713 + } + }, + "C_expanded_send": { + "traffic_bytes_per_token": 114688, + "remote_traffic_bytes_per_token": 100352.0, + "remote_traffic_bytes_total": 12845056.0, + "modeled_completion_us": 260.90112, + "error": { + "max_abs_err": 0.015120669336922354, + "mean_abs_err": 0.000818180517381322, + "rmse": 0.001158409968605701, + "max_rel_err": 6579.648285311867, + "mean_rel_err": 0.020169020199928167, + "rel_l2": 0.002341408698411671 + } + } + }, + "decision": { + "mode": "B_expand_local_reduce", + "reason": "Expanded layout is required and local reduction meets the configured precision limit." + }, + "message_size_sweep": [ + { + "tokens": 1, + "modeled_completion_us": { + "A_no_expand": 5.37396, + "B_expand_local_reduce": 5.37396, + "C_expanded_send": 6.00704 + } + }, + { + "tokens": 8, + "modeled_completion_us": { + "A_no_expand": 14.99168, + "B_expand_local_reduce": 14.99168, + "C_expanded_send": 20.05632 + } + }, + { + "tokens": 32, + "modeled_completion_us": { + "A_no_expand": 47.96672, + "B_expand_local_reduce": 47.96672, + "C_expanded_send": 68.22528 + } + }, + { + "tokens": 128, + "modeled_completion_us": { + "A_no_expand": 179.86688, + "B_expand_local_reduce": 179.86688, + "C_expanded_send": 260.90112 + } + }, + { + "tokens": 512, + "modeled_completion_us": { + "A_no_expand": 707.46752, + "B_expand_local_reduce": 707.46752, + "C_expanded_send": 1031.60448 + } + } + ] + } + ] +} diff --git a/src/code/issue8/results/decision_table.md b/src/code/issue8/results/decision_table.md new file mode 100644 index 0000000..c0a9c8f --- /dev/null +++ b/src/code/issue8/results/decision_table.md @@ -0,0 +1,58 @@ +# MoE Combine 模式决策表（单机建模结果） + +配置：E=64，top-k=8，ranks=8，hidden=7168，dtype=bf16，tokens=128。 + +> 下表时延为带宽/固定时延模型，不是 DeepEP 多卡实测结果。 + +精度选择阈值：rel-L2 <= 3.000e-03。 + +| Rank concentration | 重复率 | B 相对 C 流量 | A rel-L2 | B rel-L2 | C rel-L2 | 当前阈值建议 | +|---:|---:|---:|---:|---:|---:|:---| +| 0.03 | 79.10% | 20.90% | 2.095e-03 | 2.663e-03 | 2.348e-03 | B_expand_local_reduce | +| 0.1 | 69.92% | 30.08% | 2.306e-03 | 2.794e-03 | 2.349e-03 | B_expand_local_reduce | +| 0.3 | 59.47% | 40.53% | 2.364e-03 | 2.837e-03 | 2.350e-03 | B_expand_local_reduce | +| 1 | 44.53% | 55.47% | 2.356e-03 | 2.750e-03 | 2.350e-03 | B_expand_local_reduce | +| 3 | 36.33% | 63.67% | 2.352e-03 | 2.699e-03 | 2.349e-03 | B_expand_local_reduce | +| 100 | 31.54% | 68.46% | 2.345e-03 | 2.650e-03 | 2.341e-03 | B_expand_local_reduce | + +## 消息大小扫描（理论完成时间） + +| Rank concentration | Tokens | B time (us) | C time (us) | B 相对 C 节省 | +|---:|---:|---:|---:|---:| +| 0.03 | 1 | 4.419 | 6.007 | 26.43% | +| 0.03 | 8 | 7.356 | 20.056 | 63.33% | +| 0.03 | 32 | 17.422 | 68.225 | 74.46% | +| 0.03 | 128 | 57.688 | 260.901 | 77.89% | +| 0.03 | 512 | 218.753 | 1031.604 | 78.79% | +| 0.1 | 1 | 4.604 | 6.007 | 23.36% | +| 0.1 | 8 | 8.829 | 20.056 | 55.98% | +| 0.1 | 32 | 23.318 | 68.225 | 65.82% | +| 0.1 | 128 | 81.271 | 260.901 | 68.85% | +| 0.1 | 512 | 313.084 | 1031.604 | 69.65% | +| 0.3 | 1 | 4.813 | 6.007 | 19.87% | +| 0.3 | 8 | 10.507 | 20.056 | 47.61% | +| 0.3 | 32 | 30.029 | 68.225 | 55.99% | +| 0.3 | 128 | 108.115 | 260.901 | 58.56% | +| 0.3 | 512 | 420.461 | 1031.604 | 59.24% | +| 1 | 1 | 5.113 | 6.007 | 14.88% | +| 1 | 8 | 12.906 | 20.056 | 35.65% | +| 1 | 32 | 39.625 | 68.225 | 41.92% | +| 1 | 128 | 146.500 | 260.901 | 43.85% | +| 1 | 512 | 573.999 | 1031.604 | 44.36% | +| 3 | 1 | 5.278 | 6.007 | 12.14% | +| 3 | 8 | 14.223 | 20.056 | 29.08% | +| 3 | 32 | 44.893 | 68.225 | 34.20% | +| 3 | 128 | 167.574 | 260.901 | 35.77% | +| 3 | 512 | 658.295 | 1031.604 | 36.19% | +| 100 | 1 | 5.374 | 6.007 | 10.54% | +| 100 | 8 | 14.992 | 20.056 | 25.25% | +| 100 | 32 | 47.967 | 68.225 | 29.69% | +| 100 | 128 | 179.867 | 260.901 | 31.06% | +| 100 | 512 | 707.468 | 1031.604 | 31.42% | + +## 选择规则 + +1. 若每个目标 rank 最多一个副本，选择 A（no-expand），路径最短。 +2. 必须使用 expanded 布局且 B 的 rel-L2 不超过设定阈值时，选择 B，减少返回流量。 +3. 精度优先且 B 超过阈值时，选择 C，以额外网络流量换取更少的本地归并误差。 +4. 最终切换边界必须用 DeepEP 多卡实测时延替换当前带宽模型后再确认。 diff --git a/src/code/issue8/run_analysis.py b/src/code/issue8/run_analysis.py new file mode 100644 index 0000000..13205d0 --- /dev/null +++ b/src/code/issue8/run_analysis.py @@ -0,0 +1,327 @@ +#!/usr/bin/env python3 +"""Generate the single-machine analytical deliverables for Issue 8. + +This script intentionally labels communication time as a model rather than a +measurement. Real DeepEP latency still needs a multi-rank GPU environment. +""" + +from __future__ import annotations + +import argparse +import csv +import json +from pathlib import Path +from typing import Any + +import numpy as np + +from combine_modes import ( + COMBINE_MODES, + EPConfig, + analytic_traffic_ratio_b_over_c, + combine_reference_fp64, + duplicate_stats, + error_metrics, + expected_distinct_ranks, + make_expert_outputs, + rank_of_expert, + sample_topk_idx, + traffic_bytes_per_token, + traffic_bytes_per_token_remote_only, +) + + +def parse_concentrations(value: str) -> list[float]: + try: + result = [float(part) for part in value.split(",")] + except ValueError as exc: + raise argparse.ArgumentTypeError("concentrations must be comma-separated numbers") from exc + if not result or any(item <= 0 for item in result): + raise argparse.ArgumentTypeError("concentrations must be positive") + return result + + +def parse_positive_ints(value: str) -> list[int]: + try: + result = [int(part) for part in value.split(",")] + except ValueError as exc: + raise argparse.ArgumentTypeError("values must be comma-separated integers") from exc + if not result or any(item <= 0 for item in result): + raise argparse.ArgumentTypeError("values must be positive") + return result + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Analyze DeepEP combine modes A/B/C") + parser.add_argument("--num-experts", type=int, default=64) + parser.add_argument("--num-topk", type=int, default=8) + parser.add_argument("--num-ranks", type=int, default=8) + parser.add_argument("--hidden", type=int, default=7168) + parser.add_argument("--dtype", choices=("bf16", "fp16", "fp32"), default="bf16") + parser.add_argument("--num-tokens", type=int, default=128) + parser.add_argument( + "--message-token-counts", + type=parse_positive_ints, + default=[1, 8, 32, 128, 512], + help="token counts used for the modeled message-size sweep", + ) + parser.add_argument("--seed", type=int, default=20260726) + parser.add_argument( + "--concentrations", + type=parse_concentrations, + default=[0.03, 0.1, 0.3, 1.0, 3.0, 100.0], + help="Dirichlet rank concentrations controlling top-k repetition", + ) + parser.add_argument( + "--bandwidth-gbytes-s", + type=float, + default=50.0, + help="effective aggregate return-path bandwidth used only by the time model", + ) + parser.add_argument( + "--base-latency-us", + type=float, + default=4.0, + help="fixed launch/network latency used only by the time model", + ) + parser.add_argument( + "--max-rel-l2", + type=float, + default=3e-3, + help="precision limit when choosing local reduction versus expanded send", + ) + parser.add_argument("--output-dir", type=Path, default=Path("src/code/issue8/results")) + return parser + + +def modeled_completion_us(remote_bytes: float, bandwidth_gbytes_s: float, base_latency_us: float) -> float: + if bandwidth_gbytes_s <= 0 or base_latency_us < 0: + raise ValueError("bandwidth must be positive and latency must be non-negative") + return base_latency_us + remote_bytes / (bandwidth_gbytes_s * 1e9) * 1e6 + + +def choose_mode(row: dict[str, Any], max_rel_l2: float) -> tuple[str, str]: + stats = row["routing"] + modes = row["modes"] + if stats["duplicate_rate"] == 0.0: + return "A_no_expand", "No rank receives multiple top-k copies, so local reduction is unnecessary." + if modes["B_expand_local_reduce"]["error"]["rel_l2"] <= max_rel_l2: + return ( + "B_expand_local_reduce", + "Expanded layout is required and local reduction meets the configured precision limit.", + ) + return ( + "C_expanded_send", + "Local reduction exceeds the configured precision limit; retain all copies for final reduction.", + ) + + +def analyze(args: argparse.Namespace) -> dict[str, Any]: + if args.num_tokens <= 0: + raise ValueError("num_tokens must be positive") + cfg = EPConfig( + num_experts=args.num_experts, + num_topk=args.num_topk, + num_ranks=args.num_ranks, + hidden=args.hidden, + dtype=args.dtype, + ) + + rows: list[dict[str, Any]] = [] + for scenario_index, concentration in enumerate(args.concentrations): + rng = np.random.default_rng(args.seed + scenario_index) + topk_idx = sample_topk_idx(cfg, args.num_tokens, concentration, rng) + ranks = rank_of_expert(topk_idx, cfg) + routing = duplicate_stats(topk_idx, cfg) + expert_outputs, weights = make_expert_outputs(cfg, args.num_tokens, rng) + reference = combine_reference_fp64(expert_outputs, weights, cfg) + + modes: dict[str, Any] = {} + for mode_name, combine in COMBINE_MODES.items(): + output = combine(expert_outputs, weights, ranks, cfg) + bytes_per_token = traffic_bytes_per_token(mode_name, routing["mean_distinct_ranks"], cfg) + remote_per_token = traffic_bytes_per_token_remote_only( + mode_name, routing["mean_distinct_ranks"], cfg + ) + remote_total = remote_per_token * args.num_tokens + modes[mode_name] = { + "traffic_bytes_per_token": bytes_per_token, + "remote_traffic_bytes_per_token": remote_per_token, + "remote_traffic_bytes_total": remote_total, + "modeled_completion_us": modeled_completion_us( + remote_total, args.bandwidth_gbytes_s, args.base_latency_us + ), + "error": error_metrics(output, reference), + } + + row = { + "concentration": concentration, + "routing": routing, + "modes": modes, + } + selected, reason = choose_mode(row, args.max_rel_l2) + row["decision"] = {"mode": selected, "reason": reason} + row["message_size_sweep"] = [ + { + "tokens": token_count, + "modeled_completion_us": { + mode_name: modeled_completion_us( + mode["remote_traffic_bytes_per_token"] * token_count, + args.bandwidth_gbytes_s, + args.base_latency_us, + ) + for mode_name, mode in modes.items() + }, + } + for token_count in args.message_token_counts + ] + rows.append(row) + + return { + "schema_version": 1, + "scope": "single-machine analytical model and bit-exact low-precision simulation", + "configuration": { + "num_experts": cfg.num_experts, + "num_topk": cfg.num_topk, + "num_ranks": cfg.num_ranks, + "hidden": cfg.hidden, + "dtype": cfg.dtype, + "num_tokens": args.num_tokens, + "message_token_counts": args.message_token_counts, + "seed": args.seed, + }, + "time_model": { + "bandwidth_gbytes_s": args.bandwidth_gbytes_s, + "base_latency_us": args.base_latency_us, + "formula": "T = base_latency + remote_payload_bytes / effective_bandwidth", + "warning": "Modeled values are not DeepEP measurements and require multi-rank validation.", + }, + "uniform_routing_closed_form": { + "expected_distinct_ranks": expected_distinct_ranks(cfg), + "local_reduce_to_expanded_traffic_ratio": analytic_traffic_ratio_b_over_c(cfg), + }, + "decision_precision_limit_rel_l2": args.max_rel_l2, + "scenarios": rows, + } + + +def write_csv(report: dict[str, Any], path: Path) -> None: + fieldnames = [ + "concentration", + "duplicate_rate", + "mean_distinct_ranks", + "mode", + "remote_bytes_per_token", + "modeled_completion_us", + "max_abs_err", + "max_rel_err", + "rel_l2", + "selected", + ] + with path.open("w", encoding="utf-8-sig", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fieldnames) + writer.writeheader() + for scenario in report["scenarios"]: + for mode_name, mode in scenario["modes"].items(): + writer.writerow( + { + "concentration": scenario["concentration"], + "duplicate_rate": scenario["routing"]["duplicate_rate"], + "mean_distinct_ranks": scenario["routing"]["mean_distinct_ranks"], + "mode": mode_name, + "remote_bytes_per_token": mode["remote_traffic_bytes_per_token"], + "modeled_completion_us": mode["modeled_completion_us"], + "max_abs_err": mode["error"]["max_abs_err"], + "max_rel_err": mode["error"]["max_rel_err"], + "rel_l2": mode["error"]["rel_l2"], + "selected": mode_name == scenario["decision"]["mode"], + } + ) + + +def write_decision_table(report: dict[str, Any], path: Path) -> None: + cfg = report["configuration"] + lines = [ + "# MoE Combine 模式决策表（单机建模结果）", + "", + ( + f"配置：E={cfg['num_experts']}，top-k={cfg['num_topk']}，ranks={cfg['num_ranks']}，" + f"hidden={cfg['hidden']}，dtype={cfg['dtype']}，tokens={cfg['num_tokens']}。" + ), + "", + "> 下表时延为带宽/固定时延模型，不是 DeepEP 多卡实测结果。", + "", + f"精度选择阈值：rel-L2 <= {report['decision_precision_limit_rel_l2']:.3e}。", + "", + "| Rank concentration | 重复率 | B 相对 C 流量 | A rel-L2 | B rel-L2 | C rel-L2 | 当前阈值建议 |", + "|---:|---:|---:|---:|---:|---:|:---|", + ] + for scenario in report["scenarios"]: + modes = scenario["modes"] + traffic_ratio = ( + modes["B_expand_local_reduce"]["remote_traffic_bytes_per_token"] + / modes["C_expanded_send"]["remote_traffic_bytes_per_token"] + ) + lines.append( + "| {concentration:g} | {duplicate:.2%} | {ratio:.2%} | {a:.3e} | {b:.3e} | " + "{c:.3e} | {selected} |".format( + concentration=scenario["concentration"], + duplicate=scenario["routing"]["duplicate_rate"], + ratio=traffic_ratio, + a=modes["A_no_expand"]["error"]["rel_l2"], + b=modes["B_expand_local_reduce"]["error"]["rel_l2"], + c=modes["C_expanded_send"]["error"]["rel_l2"], + selected=scenario["decision"]["mode"], + ) + ) + lines.extend( + [ + "", + "## 消息大小扫描（理论完成时间）", + "", + "| Rank concentration | Tokens | B time (us) | C time (us) | B 相对 C 节省 |", + "|---:|---:|---:|---:|---:|", + ] + ) + for scenario in report["scenarios"]: + for point in scenario["message_size_sweep"]: + b_time = point["modeled_completion_us"]["B_expand_local_reduce"] + c_time = point["modeled_completion_us"]["C_expanded_send"] + savings = 1.0 - b_time / c_time + lines.append( + f"| {scenario['concentration']:g} | {point['tokens']} | {b_time:.3f} | " + f"{c_time:.3f} | {savings:.2%} |" + ) + lines.extend( + [ + "", + "## 选择规则", + "", + "1. 若每个目标 rank 最多一个副本，选择 A（no-expand），路径最短。", + "2. 必须使用 expanded 布局且 B 的 rel-L2 不超过设定阈值时，选择 B，减少返回流量。", + "3. 精度优先且 B 超过阈值时，选择 C，以额外网络流量换取更少的本地归并误差。", + "4. 最终切换边界必须用 DeepEP 多卡实测时延替换当前带宽模型后再确认。", + "", + ] + ) + path.write_text("\n".join(lines), encoding="utf-8") + + +def main() -> int: + args = build_parser().parse_args() + report = analyze(args) + args.output_dir.mkdir(parents=True, exist_ok=True) + json_path = args.output_dir / "combine_modes_report.json" + csv_path = args.output_dir / "combine_modes_report.csv" + decision_path = args.output_dir / "decision_table.md" + json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + write_csv(report, csv_path) + write_decision_table(report, decision_path) + print(f"Wrote {json_path}") + print(f"Wrote {csv_path}") + print(f"Wrote {decision_path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/code/issue8/test_combine_modes.py b/src/code/issue8/test_combine_modes.py new file mode 100644 index 0000000..01957c4 --- /dev/null +++ b/src/code/issue8/test_combine_modes.py @@ -0,0 +1,35 @@ +import numpy as np + +from combine_modes import ( + EPConfig, + analytic_traffic_ratio_b_over_c, + combine_mode_a, + combine_mode_b, + combine_mode_c, + expected_distinct_ranks, + round_to_bf16, +) + + +def test_bf16_round_to_nearest_even(): + values = np.array([1.0, 1.00390625, 1.0078125], dtype=np.float32) + rounded = round_to_bf16(values) + np.testing.assert_array_equal(rounded, np.array([1.0, 1.0, 1.0078125], dtype=np.float32)) + + +def test_closed_form_traffic_ratio_is_bounded(): + cfg = EPConfig(num_experts=64, num_topk=8, num_ranks=8, hidden=128) + assert 1.0 <= expected_distinct_ranks(cfg) <= cfg.num_topk + assert 0.0 < analytic_traffic_ratio_b_over_c(cfg) <= 1.0 + + +def test_all_modes_match_when_topk_is_one(): + cfg = EPConfig(num_experts=8, num_topk=1, num_ranks=8, hidden=4) + outputs = np.array([[[1.25, -2.0, 0.5, 4.0]]], dtype=np.float32) + weights = np.ones((1, 1), dtype=np.float32) + ranks = np.zeros((1, 1), dtype=np.int64) + + mode_a = combine_mode_a(outputs, weights, ranks, cfg) + np.testing.assert_array_equal(mode_a, combine_mode_b(outputs, weights, ranks, cfg)) + np.testing.assert_array_equal(mode_a, combine_mode_c(outputs, weights, ranks, cfg)) +