[issue8] Alltoall combine 三模式位级仿真与理论性能上限决策表

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
+```

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)

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