CAM EP Communication Kernels 深度分析报告

分析范围: src/cam/comm_operator/ascend_kernels/ 下的 4 个核心 MoE EP 通信算子 目标硬件: Ascend 910_93 (A3) 分析日期: 2026-07-21 分析层级: op_api → op_host → op_kernel → pybind 全栈

1. 架构总览与设计哲学

1.1 CAM 的定位

CAM (Communication Acceleration for Matrix) 是华为 Ascend NPU 面向 MoE (Mixture-of-Experts) Expert Parallelism 场景的通信加速库。其核心设计目标是：

消除 All-to-All 瓶颈: MoE 中 token 需要跨卡路由到对应 expert 所在的 rank，传统 All-to-All 通信开销巨大
零拷贝 (IPC): 利用 HCCL Window 机制实现卡间直接内存访问，避免显式收发
计算通信重叠: 通过状态位同步 + 双缓冲 (ping-pong) 实现流水线隐藏延迟

1.2 算子调用链

Python 前端
    │
    ▼
Pybind 层 (TORCH_LIBRARY + EXEC_NPU_CMD)
    │
    ▼
aclnn API 层 (op_api/aclnn_*.cpp) ─── 调用 aclnnInner* 内部实现
    │
    ▼
Tiling 层 (op_host/*_tiling.cpp) ─── 计算分片参数、校验合法性、设置 HCCL 配置
    │
    ▼
Kernel 层 (op_kernel/*.cpp) ─── 设备端执行，操作 HCCL Window + IPC

1.3 核心数据流

                    ┌─────────────────────────────────────────────────────────┐
                    │                    MoE EP Pipeline                      │
                    │                                                         │
  TopkIdx           │   Notify     MoeDispatch     Expert      MoeCombine     │
   (tokens×k)       │   Dispatch     Normal        Compute      Normal        │
     │              │      │           │             │            │            │
     ▼              │      ▼           ▼             ▼            ▼            │
 ┌────────┐         │  ┌────────┐  ┌────────┐  ┌────────┐  ┌────────┐       │
 │Dispatch│────────▶│  │SendData│  │ X+Idx  │  │Expert  │  │Weighted│       │
 │Layout  │         │  │Offset  │  │→Window │  │Forward │  │  Sum   │       │
 └────────┘         │  └────────┘  └────────┘  └────────┘  └────────┘       │
                    │                                                         │
                    └─────────────────────────────────────────────────────────┘

2. 共享基础设施分析

2.1 CommArgs (`utils/op_kernel/comm_args.h`)

这是整个通信层的核心参数结构体：

struct CommArgs {
    int rank;              // 全局 rank ID
    int localRank;         // 本地 rank (fullmesh 内部编号)
    int rankSize;          // 全局 rank 数量 (最大 384)
    int localRankSize;     // 本地 fullmesh 互联 rank 数
    GM_ADDR peerMems[CAM_MAX_RANK_SIZE];  // 初始化时获取的各卡 Buffer 地址
    int64_t sendCountMatrix[CAM_MAX_RANK_SIZE * CAM_MAX_RANK_SIZE];  // All-to-All 计数矩阵
    int64_t sendCounts[CAM_MAX_RANK_SIZE];
    int64_t recvCounts[CAM_MAX_RANK_SIZE];
    // ...
};

关键常量：

常量	值	含义
`CAM_MAX_RANK_SIZE`	384	最大支持 384 张 NPU 卡
`IPC_BUFF_MAX_SIZE`	100 MB	IPC 数据区最大 100MB
`IPC_DATA_OFFSET`	2 MB	IPC 前 2MB 作为 flag 区
`UB_SINGLE_DMA_SIZE_MAX`	190 KB	单次 DMA 最大数据量
`PING_PONG_SIZE`	2	双缓冲深度

2.2 SyncCollectives (`utils/op_kernel/sync_collectives.h`)

这是卡间同步原语的核心实现，使用 HCCL Window 上的 flag 实现跨卡同步：

同步标志结构：

constexpr int64_t FLAG_UNIT_INT_NUM = 4;      // 每个 flag 占 4 个 int64
constexpr int64_t SYNC_UNIT_SIZE = 4 * sizeof(int64_t);  // 32 Bytes
constexpr int64_t MAGIC_OFFSET = 32;          // magic 占高 32 位

核心机制 — Magic + Value 编码：

__aicore__ inline int64_t MergeMagicWithValue(int32_t magic, int32_t value) {
    return (static_cast<int64_t>(static_cast<uint32_t>(magic)) << MAGIC_OFFSET) 
           | static_cast<int64_t>(value);
}

设计洞察: 高 32 位 magic 用于区分不同轮次（避免 ABA 问题），低 32 位 value 表示具体事件。这种无锁设计允许跨卡原子操作。

内存布局 (每个 rank 的 shareAddr)：

┌──────────────────────────────────────────────────────┐
│ 段 A: 卡内同步 flags (blockNum × FLAG_UNIT_INT_NUM)   │
├──────────────────────────────────────────────────────┤
│ 段 B: 卡间同步 flags (blockNum × FLAG_UNIT_INT_NUM)   │
├──────────────────────────────────────────────────────┤
│ Chunk Flags: IPC 传输完成标记 (per destRank)            │
└──────────────────────────────────────────────────────┘

WaitOneRankPartFlag 实现 — 核心等待循环：

do {
    DataCopy(localWait, globalWait[...], ...);  // GM→UB 拷贝
    WaitFlag<MTE2_S>();                          // 等待 DMA 完成
    for (i = 0; i < remainToCheck; ++i) {
        v = localWait.GetValue(i * FLAG_UNIT_INT_NUM);
        if ((v & MAGIC_MASK) != (checkValue & MAGIC_MASK) || v < checkValue) {
            isSync = false;
            checkedFlagNum += i;  // 跳过已确认的 flag
            break;
        }
    }
} while (!isSync);

优化点: 增量检查（checkedFlagNum）避免重复检查已完成的 flag，减少 UB 读取开销。

2.3 DataCopy 工具 (`utils/op_kernel/data_copy.h`)

提供 GM↔UB、UB↔UB 的模板化拷贝函数： - CpUB2GM: UB → Global Memory (含原子操作支持) - CpGM2UB: Global Memory → UB - CopyUB2UB: UB 内部拷贝 - SetAtomicOpType<T>(op): 设置原子操作类型（ADD/MUL/MAX/MIN）

2.4 Mc2TilingUtils (`utils/op_host/mc2_tiling_utils.h`)

关键功能：读取 HCCL_BUFFSIZE 环境变量，计算 Window 可用大小：

uint64_t maxWindowSize = defaultWindowSize * 1024UL * 1024UL;  // 默认 200MB

2.5 TilingArgs (`utils/op_host/tiling_args.h`)

定义关键偏移常量：

constexpr uint64_t COMBINE_STATE_WIN_OFFSET = 4UL * 1024UL * 1024UL;    // Combine 状态区偏移
constexpr uint64_t NOTIFY_DISPATCH_WIN_OFFSET = 204UL * 1024UL * 1024UL; // NotifyDispatch 偏移

3. dispatch_layout 算子

3.1 功能定位

dispatch_layout (aclnnDispatchLayout) 是 MoE 路由的前置分析算子，纯计算、无通信。它根据 topkIdx 计算：

每个 rank 需要接收的 token 数 (numTokensPerRank)
每个 expert 需要处理的 token 数 (numTokensPerExpert)
每个 token 是否在某个 rank 内 (isTokenInRank)
用于 notify 的发送数据 (notifySendData)
小 sendTokenIdx (sendTokenIdxSmall)

3.2 op_api 层

aclnnStatus aclnnDispatchLayout(void *workspace, uint64_t workspaceSize, 
                                 aclOpExecutor *executor, aclrtStream stream) {
    if (NnopbaseSetHcclServerType) {
        NnopbaseSetHcclServerType(executor, NNOPBASE_HCCL_SERVER_TYPE_MTE);  // 关键: 使用 MTE 模式
    }
    return aclnnInnerDispatchLayout(workspace, workspaceSize, executor, stream);
}

关键设计: 强制设置 HCCL Server Type 为 MTE (Memory Transfer Engine)，绕过 AICPU 中转。

3.3 op_host 层 (Tiling 逻辑)

输入/输出定义： - Input: topkIdx (int64, [numTokens, numTopk]) - Attrs: num_tokens, num_ranks, num_experts, num_topk, local_ranksize - Outputs: numTokensPerRank, numTokensPerExpert, isTokenInRank, notifySendData, sendTokenIdxSmall

核心参数校验： - numRanks ∈ (0, 384] - numExperts ∈ (0, 512] - numTopk ∈ (0, 16] - numTokens ∈ (0, 8000] - 约束: numExperts % numRanks == 0（expert 必须均匀分布）

Workspace 计算：

workSpaces[0] = SYSTEM_NEED_WORKSPACE + KERNEL_USE_WORKSPACE;  // 16MB + 1MB

Tiling Key 固定为 23。

3.4 op_kernel 层 (核心算法)

并行策略: 按 token 维度切分到多个 AIV core：

aivNum_ = numTokens_ <= maxAivNum ? numTokens_ : maxAivNum;
// 每个 core 处理 tempTokens_ 个 token

核心算法 — Process()：

Phase 1: 统计 (每个 core 独立)
  for each token i in [startToken, endToken):
    for each topk j:
      expert_idx = topkIdx[i][j]
      numTokensPerExpert[expert_idx]++
      rank_id = expert_idx / experts_per_rank
      if !seenRank[rank_id]:
        numTokensPerRank[rank_id]++
        isTokenInRank[i][rank_id] = 1
        seenRank[rank_id] = 1

Phase 2: 全局聚合 (原子加)
  SetAtomicAdd<T>()
  for each other core i:
    DataCopyPad(notifySendData[i * numExperts], numTokensPerExpert)
  SetAtomicNone()

Phase 3: SyncAll 全卡同步
  SyncAll<true>()

Phase 4: 计算 sendTokenIdxSmall
  for each token i:
    for each topk j:
      expert_idx = topkIdx[i][j]
      sendTokenIdxSmall[i][j] = aggregatedNumTokensPerExpert[expert_idx]++

算法洞察: Phase 2 使用 SetAtomicAdd 而非显式 reduce，利用硬件原子操作实现跨 core 聚合，避免额外的同步开销。

UB 内存规划：

topkIdx32AlignIntLen_ = Ceil(tempTokens_ * numTopk_ * sizeof(int64_t), 32) * 32;
numTokensPerRank32AlignIntLen_ = Ceil(numRanks_ * sizeof(T), 32) * 32;
numTokensPerExpert32AlignIntLen_ = Ceil(numExperts_ * sizeof(T), 32) * 32;
isTokenInRank32AlignIntLen_ = Ceil(tempTokens_ * numRanks_ * sizeof(T), 32) * 32;

4. notify_dispatch 算子

4.1 功能定位

notify_dispatch (aclnnNotifyDispatch) 是元数据 All-to-All 通信算子。它在实际 token 数据传输前，先让每张卡知道：

其他卡要发送给自己多少 token (recvCount)
从哪个 offset 开始存放 (recvOffset)
本卡要给其他卡发送的 offset (sendDataOffset)
全局最大 batch size (maxBs)
每个本地 expert 接收的 token 总数 (recvTokensPerExpert)

4.2 op_api 层

参数列表 (极复杂，反映通信元数据交换的本质)：

aclnnNotifyDispatchGetWorkspaceSize(
    sendData, tokenPerExpertData, sendCount, numTokens,
    commGroup, rankSize, rankId, localRankSize, localRankId,
    sendDataOffset, recvData, totalRecvTokens, recvCount, recvOffset,
    maxBs, recvTokensPerExpert, ...
)

4.3 op_host 层

HCCL 通信配置：

uint32_t opType1 = OP_TYPE_ALL_TO_ALL;  // 8 = AlltoAll
std::string algConfigAllToAllStr = "AlltoAll=level0:fullmesh;level1:pairwise";

通信拓扑: level0 使用 fullmesh（适合机内全互联），level1 使用 pairwise（适合跨机逐步传输）。这是经典的两级 All-to-All 策略。

Tiling Key 选择（基于数据类型）: - Float16 → 20 - BFloat16 → 21 - Float → 22 - Int → 23 - Ascend910B 额外 +100

4.4 op_kernel 层 (核心算法)

NotifyDispatch 是 CAM 中最复杂的算子，它实现了自管理的 All-to-All（不依赖 HCCL 集合通信）：

四阶段流水线：

Process():
  if (blockIdx < 1):
    AssembleSendData()         // Phase 0: Core 0 组装发送数据
  SyncAll<true>()
  if (blockIdx < coreNumPerStageX):
    InputToShareSlice()        // Phase 1: 将自己的数据写入其他卡的 Window
  if (blockIdx < coreNumPerStageY):
    ShareToShareSlice()        // Phase 2: 从其他卡的 Window 读取数据
  SyncAll<true>()
  ReorderOutput()              // Phase 3a: 重排接收到的数据
  BuildTotalRecvTokens()       // Phase 3b: 构建 totalRecvTokens (blockIdx==0)
  BuildRecvCount()             // Phase 3c: 构建 recvCount (blockIdx==1)
  BuildRecvOffset()            // Phase 3d: 构建 recvOffset (blockIdx==2)
  BuildMaxBs()                 // Phase 3e: 构建 maxBs (blockIdx==3)
  BuildRecvTokenPerExp()       // Phase 3f: 构建 recvTokensPerExpert (blockIdx==4)

AssembleSendData 详解：

// 每个 expert 的 sendPerGroup = 3 个字段:
// [numTokensExpert, prefixSum, numTokens]
for (int i = 0; i < numExperts; ++i) {
    int numTokensExpert = tokenPerExpertUB[i];
    sendDataUB[i * sendPerGroup] = numTokensExpert;       // 该 expert 有多少 token
    sendDataUB[i * sendPerGroup + 1] = prefixSum;          // 偏移量
    sendDataUB[i * sendPerGroup + 2] = numTokens;          // 总 token 数
    sendDataOffsetUB[i] = prefixSum;
    prefixSum += numTokensExpert;
}

InputToShareSlice — 核心 IPC 写入：

// 每个 core 负责对 copyLen 个 rank 进行写入
CpGM2GMPingPong<T>(copyLen * perRankDataNum * sizeof(T), 
                    readGt, writeGt[copyOffset * perRankDataNum], COPYONLY);
// 写入完成后设置 flag 通知对端
uint64_t v = MergeMagicWithValue(magic, 1);
CpUB2GM(shareAddrs[i] + rank * FLAG_UNIT_INT_NUM, inputUB, sizeof(uint64_t));

设计亮点: 使用 Ping-Pong 双缓冲 GM→GM 拷贝，UB_HEAD_OFFSET (96B) 和 UB_MID_OFFSET (96B + 95KB + 32B) 交替使用，实现流水隐藏。

ShareToShareSlice — 读取对端数据：

// 等待本卡数据已被写入对端
WaitSyncFlag(magic, 1, copyOffset, rank, copyLen);
// 从对端 Window 读取属于本卡的数据
for (int i = 0; i < copyLen; i++) {
    CpGM2GMPingPong<T>(perRankDataNum * sizeof(T), 
                        readGt1[i][rank * perRankDataNum],
                        recvDataOutputGt[checkRank[i] * perRankDataNum], COPYONLY);
}

CpGM2GMPingPong — 核心 DMA 引擎：

// 双缓冲交替:
// 偶数轮: 用 inputUB[0] + outputUB[0]
// 奇数轮: 用 inputUB[1] + outputUB[1]
for (int64_t i = 0; dataSizeRemain > 0; i++) {
    event_t eventId = (i & 1) ? EVENT_ID0 : EVENT_ID1;
    WaitFlag<MTE3_MTE2>(eventId);        // 等待上一轮写完成
    CpGM2UB(inputUB[i&1], input, size);  // GM→UB
    CpUB2GM(output, outputUB[i&1], size); // UB→GM
    SetFlag<MTE3_MTE2>(eventId);         // 通知下一轮可开始
    dataSizeRemain -= size;
}

4.5 同步机制

Magic 值获取：

__aicore__ inline uint64_t NotifyDispatch<T>::GetMagicValue(void) {
    // 从 localWindowsExp + STATE_WIN_OFFSET + blockIdx * UB_ALIGN_SIZE 读取
    magic = selfDataStatusTensor(blockIdx * UB_ALIGN_SIZE);
    if (magic <= 0) magic = 1;
    selfDataStatusTensor(blockIdx * UB_ALIGN_SIZE) = magic + 1;  // 自增防 ABA
    return magic;
}

Window 地址计算：

shareAddrs[rank] = GetWindAddrByRankId(rank, ctxIdx) 
                   + (magic % PING_PONG_SIZE) * (IPC_BUFF_MAX_SIZE + IPC_DATA_OFFSET);

Ping-Pong 双 Window: 交替使用两个 102MB+2MB 的 buffer，避免读写冲突。

5. moe_dispatch_normal 算子

5.1 功能定位

moe_dispatch_normal (aclnnMoeDispatchNormal) 是 token 数据实际分发算子。它将 hidden states 通过 HCCL Window 路由到对应 expert 所在的 rank。

5.2 op_api 层

aclnnMoeDispatchNormalGetWorkspaceSize(
    x, topkIdx, sendOffset, sendTokenIdx,    // 输入
    recvOffset, recvCount,                    // 从 notify 获取的接收信息
    groupEp, epWorldSize, epRankId,           // EP 通信域
    groupTpOptional, tpWorldSize, tpRankId,   // 可选 TP 通信域
    moeExpertNum, quantMode, globalBs,        // 配置参数
    recvX, recvXScales, assistInfoForCombine, // 输出
    waitRecvCostStats,                        // 可选诊断
    ...
)

5.3 op_host 层 (Tiling 逻辑)

输入/输出多态（根据 quantMode 变化）：

Index	Input/Output	quantMode=0	quantMode=2
0	x	BF16	BF16
1	topk_idx	INT32	INT32
0	recv_x	BF16	INT8
1	x_scales	FLOAT	FLOAT

HCCL 通信配置：

// EP 域: All-to-All (fullmesh + pairwise)
std::string algConfigAllToAllStr = "AlltoAll=level0:fullmesh;level1:pairwise";
// TP 域: AllGather (ring)
std::string algConfigAllGatherStr = "AllGather=level0:ring";

Window 大小校验：

tokenActualLen = Ceil(h * 2, 32) * 32 + 44;  // h_align_32B + scale(32B) + expandIdx(3*4B)
tokenNeedSizeDispatch = Ceil(tokenActualLen, 512) * 512;
tokenNeedSizeCombine = Ceil(h * 2, 512) * 512;
actualSize = (maxBs * k * (tokenNeedSizeCombine + tokenNeedSizeDispatch) 
              + 4MB + 204MB) * 2;
// 校验 actualSize <= HCCL_BUFFSIZE

Tiling Key 计算：

tilingKey = 10000 + quantMode + (tpWorldSize == 2 ? 100 : 0);
// 10000: 无量化, 10002: 动态量化, 10100: TP=2 无量化

Workspace：

workSpaces[0] = 16MB + 512 * aivNum * aivNum;  // 用于诊断统计

5.4 op_kernel 层 (核心算法)

模板参数化：

template <typename XType, typename ExpandXOutType, 
          bool dynamicQuant, bool IsSmoothScaleExist, bool IsShareExpertRank>
class MoeDispatchNormal { ... };

四阶段流水线 (Process)：

InputToShare()    →  将 token 数据写入目标 rank 的 Window
SetStatus()       →  设置目标 rank 的状态位 (通知数据已到达)
WaitStatus()      →  等待所有源 rank 的数据到达
ShareToOutput()   →  从 Window 读取数据到本地输出

InputToShare 详解：

// 每个 core 负责 [startTokenId, endTokenId) 的 token
for (tokenIndex = startTokenId; tokenIndex < endTokenId; ++tokenIndex) {
    dstExpertId = expertIdsTensor(tokenIndex - startTokenId);
    curExpertCnt = sendTokenIdxTensor(tokenIndex - startTokenId);
    dstExpertOffset = sendOffsetTensor(dstExpertId);
    // 计算目标 Window 地址
    rankGM = shareGM + hOutGMAlignSize * (dstExpertOffset + curExpertCnt);

    if (DynamicQuant) {
        // 量化路径: BF16 → FP32 → Abs → ReduceMax → Scale → INT8
        QuantProcess();
        FillTriple(xOutTensor, tokenIndex / topK, tokenIndex % topK);
        DataCopyPad(dstGT, xOutTensor, hCommuCopyOutParams);
    } else {
        // 非量化路径: 直接拷贝 + 填充三元组
        FillTriple(xTmpTensor, tokenIndex / topK, tokenIndex % topK);
        DataCopyPad(dstGT, xTmpTensor, hCommuCopyOutParams);
    }
}

FillTriple — 元数据嵌入：

__aicore__ inline void FillTriple(LocalTensor<ExpandXOutType> &xOutTensor, 
                                   uint32_t tokenIndex, uint32_t k) {
    LocalTensor<int32_t> xOutTint32 = xOutTensor.ReinterpretCast<int32_t>();
    xOutTint32(expandIdxStartIdx) = epRankId;      // 源 rank ID
    xOutTint32(expandIdxStartIdx + 1) = tokenIndex; // 源 token index
    xOutTint32(expandIdxStartIdx + 2) = k;          // topk index
}

设计洞察: 在每个 token 的 hidden state 后面附加三元组 (srcRank, srcTokenIdx, topkIdx)，这样 combine 阶段无需额外通信即可知道数据来源。

SetStatus — 状态位同步：

// 将 status 写入目标 rank 的 Window 状态区
for (uint32_t i = startExpId; i < endExpId; ++i) {
    targetRankId = i / moeExpertNumPerRank;
    offset = stateOffset * (epRankId + i % moeExpertNumPerRank * epRankSize);
    rankGM = GetWindStateAddrByRankId(COMM_EP_IDX, targetRankId) + offset;
    DataCopy(dstStatusGT, statusTensor[(i - startExpId) * 8], 8UL);
}

WaitStatus — 等待数据到达：

while (sumOfFlag != compareTarget) {
    DataCopy(statusFp32Tensor, windowInstatusFp32Tensor[startStatusId * stateOffset / 4], ...);
    ReduceSum(statusSumOutTensor, statusFp32Tensor, gatherMaskOutTensor, ...);
    sumOfFlag = statusSumOutTensor.GetValue(0);
}
// 清理状态
Duplicate<int32_t>(cleanStateTensor, 0, ...);
DataCopy(windowInstatusFp32Tensor[...], cleanStateTensor.ReinterpretCast<float>(), ...);

性能洞察: 使用 ReduceSum 在 Vector 单元上快速求和所有状态位，当 sum == statusNumPerCore 时说明所有数据已到达。

ShareToOutput — 从 Window 读取：

for (uint32_t i = startStatusId; i < endStatusId; ++i) {
    fromRank = i % epRankSize;
    count = recvCountTensor(i) - preCount;
    for (uint32_t j = 0; j < count; ++j++) {
        // 从 Window 读取 token + expandIdx
        DataCopyPad(xTmpTensor, srcTokenGT, hCommuCopyOutParams);
        // 提取 expandIdx (三元组)
        DataCopyPad(expandIdxOutGT[(targetOffset + j) * 3], xTmpTensorInt[expandIdxStartIdx], ...);
        // 如果是量化模式, 提取 scale
        if (DynamicQuant) {
            DataCopyPad(dynamicScalesOutGT[targetOffset + j], xOutFp32Tensor[hUBAlignSize / 4], ...);
        }
        // 拷贝 token 到输出
        DataCopyPad(dstTokenGT, xTmpTensor, expandXCopyParams);
    }
}

5.5 量化机制 (QuantProcess)

__aicore__ inline void QuantProcess() {
    Cast(floatLocalTemp, xInTensor, CAST_NONE, h);        // BF16 → FP32
    Abs(floatLocalAbsTemp, floatLocalTemp, h);              // |x|
    ReduceMaxInplace(floatLocalAbsTemp, h);                 // max(|x|)
    dynamicScale = 127.0 / (floatLocalAbsTemp.GetValue(0) + 1e-12f);
    Muls(floatLocalTemp, floatLocalTemp, dynamicScale, h);  // x * scale
    Cast(int32LocalTemp, floatLocalTemp, CAST_RINT, h);    // FP32 → INT32 (四舍五入)
    Cast(halfLocalTemp, int32LocalTemp, CAST_ROUND, h);    // INT32 → FP16 (dequant)
    Cast(xOutTensor, halfLocalTemp, CAST_TRUNC, h);        // FP16 → INT8
    // 存储 scale 供 combine 反量化
    floatLocalTemp.SetValue(hUBAlignSize / sizeof(float), 1.0 / dynamicScale);
}

ReduceMaxInplace 优化：

// 两级 reduce: 先两两比较 (stride=8), 再 WholeReduceMax
if (repsFp32 > 1) {
    Max(srcLocal, srcLocal[64], srcLocal, 64, repsFp32 - 1, {1,1,1,0,8,0});
}
WholeReduceMax(srcLocal, srcLocal, mask, 1, 8, 1, 8);

6. moe_combine_normal 算子

6.1 功能定位

moe_combine_normal (aclnnMoeCombineNormal) 是 token 加权聚合算子。Expert 计算完成后，将结果通过 Window 发回源 rank，并按 topk_weight 加权求和。

6.2 op_api 层

aclnnMoeCombineNormalGetWorkspaceSize(
    recvX, tokenSrcInfo, epRecvCounts, recvTopkWeights, tpRecvCountsOptional,
    epGroupName, epWorldSize, epRankId,
    tpGroupNameOptional, tpWorldSize, tpRankId,
    moeExpertNum, globalBs, out, sendCostStats, ...
)

6.3 op_host 层 (Tiling 逻辑)

输入/输出： - recv_x: [totalRecvTokens, h] (BF16/FP16) - token_src_info: [totalRecvTokens * 3] (INT32) — expandIdx - ep_recv_counts: [moeExpertNum] (INT32) - recv_topk_weights: [bs, k] (FLOAT) - x: [bs, h] (BF16/FP16)

HCCL 通信配置：

// EP: All-to-All (fullmesh + pairwise)
// TP: ReduceScatter (ring) ← 注意这里是 ReduceScatter 不是 AllGather
std::string algConfigReduceScatterStr = "ReduceScatter=level0:ring";

Tiling 约束：

// combine 不支持多 expert + TP 同时启用
OPS_ERR_IF((localMoeExpertNum > 1) && (tpWorldSize > 1), 
    "Cannot support multi-moeExpert in a rank when tpWorldSize > 1");

6.4 op_kernel 层 (核心算法)

两阶段流水线：

CopyBufferToShareAndSetStatus()  →  将 expert 输出写入源 rank 的 Window，设置状态
ReadBufferFromRemote()           →  等待数据到达，读取并按 weight 加权求和

CopyBufferToShareAndSetStatus 详解：

// 每个 core 负责 [startTokenId, endTokenId) 的 token
for (tokenIndex = startTokenId; tokenIndex < endTokenId; tokenIndex++) {
    srcRankId = srcInfoLocal(index + 0);  // 源 rank
    srcTokenId = srcInfoLocal(index + 1);  // 源 token
    srcTopkId = srcInfoLocal(index + 2);   // topk index

    // 拷贝数据到源 rank 的 Window
    CopyBufferToShare(srcRankId, srcTokenId, srcTopkId, tokenIndex);
    // 设置状态位
    SetStatusBySrcInfo(srcRankId, srcTokenId, srcTopkId);
}

CopyBufferToShare — Window 写入：

GM_ADDR dstGM = GetBufferAddrByRankId(srcRankId) 
              + (srcTokenId * axisK_ + srcTopkId) * h512AlignRecvXLen_;
// 512 字节对齐存储，每个 slot 存一个 topk 结果
DataCopyPad(dstWindow, localCopyTensor, xOutCopyParams);

SetStatusBySrcInfo — 状态更新：

GM_ADDR stateGM = GetStateAddrByRankId(srcRankId) 
                + (srcTokenId * axisK_ + srcTopkId) * UB_32_ALIGN;
// 写入 1.0 (0x3F800000) 表示该 slot 数据已就绪
DataCopy<uint32_t>(stateGMTensor, statusTensor, FLOAT_NUM_PER_ALIGN);

WaitBuffCopy — 等待数据到达：

target = axisK_ * FLOAT_NUM_PER_ALIGN;  // 所有 topk 数据都到达
while (current != target) {
    DataCopy<float>(stateTensorLocal, stateGMTensor, calCount);
    Sum(tempStateTensorLocal, stateTensorLocal, sumPerKParams);
    current = tempStateTensorLocal(0);
}
// 清零状态
Duplicate<float>(tempStateTensorLocal, 0.0, calCount);
DataCopy<float>(stateGMTensor, tempStateTensorLocal, calCount);

ReadBufferAndWeightedSum — 加权求和：

Duplicate(sumFloatBufLocal, 0, axisH_);  // 初始化累加器
for (topkId = 0; topkId < axisK_; topkId++) {
    scale = topkWeightsLocal.GetValue((tokenIndex - startTokenIndex) * axisK_ + topkId);
    // 从本卡 Window 读取该 topk 对应的结果
    localTokenAddr = localRankGM_ + (tokenIndex * axisK_ + topkId) * h512AlignRecvXLen_;
    DataCopyPad(tmpToken, localTokenTensor, xOutCopyParams);
    // 类型转换 + 乘权重 + 累加
    Cast(tokenFloatLocal, tmpToken, CAST_NONE, axisH_);     // BF16 → FP32
    Muls(weightedMulBufLocal, tokenFloatLocal, scale, axisH_); // * weight
    Add(sumFloatBufLocal, sumFloatBufLocal, weightedMulBufLocal, axisH_); // sum
}
// 最终结果转回 BF16
Cast(xOutLocal, sumFloatBufLocal, CAST_RINT, axisH_);
DataCopyPad(xOutGlobal_[tokenIndex * axisH_], xOutLocal, xOutCopyParams);

数值精度洞察: 累加使用 FP32 精度（避免 BF16 累加误差），最终输出时转回 BF16/FP16。这是混合精度计算的标准做法。

6.5 Magic 双缓冲机制

__aicore__ inline void MoeCombineNormal::InitMagic() {
    selfMagicTensor.SetGlobalBuffer(
        (__gm__ int32_t *)((GM_ADDR)epWinContext_->localWindowsExp 
                            + MAGIC_WIN_OFFSET + coreIdx_ * WIN_512_ALIGN));
    DataCacheCleanAndInvalid<int32_t, ...>(selfMagicTensor);
    magic_ = selfMagicTensor(0);
    selfMagicTensor(0) = ((magic_ == 0) ? 1 : 0);  // 0/1 翻转
}

// 使用 magic 选择 Window 的哪一半
winDataSizeOffset_ = magic_ * (totalWinSize / 2UL);

7. Pybind 绑定层

7.1 注册机制

使用 PyTorch 的 Dispatcher + TORCH_LIBRARY 双重注册：

// 1. pybind11 模块注册 (Python 直接调用)
PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
    m.def("get_dispatch_layout", &GetDispatchLayoutImplAutograd, ...);
    m.def("moe_dispatch_prefill", &MoeDispatchPrefillImplAutograd, ...);
    m.def("moe_combine_prefill", &MoeCombinePrefillImplAutograd, ...);
}

// 2. TORCH_LIBRARY Schema 注册 (供 GE 图引擎使用)
TORCH_LIBRARY(umdk_cam_op_lib, m) {
    m.def("get_dispatch_layout(Tensor topk_idx, int num_experts, int num_ranks) -> (Tensor, Tensor)");
    m.def("moe_dispatch_prefill(Tensor x, Tensor topk_idx, ...) -> (Tensor, Tensor, Tensor, Tensor, Tensor)");
    m.def("moe_combine_prefill(Tensor x, Tensor topk_idx, ...) -> Tensor");
}

7.2 三层调用架构

Python 前端
    │
    ▼
AutogradPrivateUse1  (支持反向传播)
    │  MoeDispatchPrefill::apply() → forward() / backward()
    ▼
PrivateUse1 (NPU 实际执行)
    │  MoeDispatchPrefillImplNpu() → EXEC_NPU_CMD
    ▼
aclnn API (CANN 运行时)

7.3 get_dispatch_layout 绑定

std::tuple<at::Tensor, at::Tensor> GetDispatchLayoutImplNpu(
    const at::Tensor &topkIdx, int64_t numExperts, int64_t numRanks) {
    // 类型转换: topkIdx → int64
    at::Tensor topkIdxInt64 = topkIdx.scalar_type() == at::kLong ? topkIdx : topkIdx.to(at::kLong);

    // 分配输出 tensor
    auto numTokensPerExpert = at::zeros({numExperts}, at::dtype(at::kInt).device(device));
    auto sendTokenIdxSmall = at::zeros({numTokens, numTopk}, at::dtype(at::kInt).device(device));
    auto notifySendData = at::zeros({notifySendDataSize}, ...);

    // 调用 aclnn API
    EXEC_NPU_CMD(aclnnDispatchLayout, topkIdxInt64, numTokens, numRanks, 
                 numExperts, numTopk, localRanksize,
                 numTokensPerRank, numTokensPerExpert, isTokenInRank, 
                 notifySendData, sendTokenIdxSmall);

    return std::make_tuple(numTokensPerExpert, sendTokenIdxSmall);
}

7.4 moe_dispatch_prefill 绑定

这是最复杂的 pybind 函数，它组合了 notify_dispatch + moe_dispatch_normal 两个算子：

std::tuple<at::Tensor, at::Tensor, at::Tensor, at::Tensor, at::Tensor> 
MoeDispatchPrefillImplNpu(...) {
    // Step 1: 准备 notify 的输入
    auto sendData = torch::empty({numExperts * 3}, ...);  // sendPerGroup=3
    auto sendDataOffset = torch::empty({numExperts}, ...);

    // Step 2: 调用 notify_dispatch (All-to-All 交换元数据)
    EXEC_NPU_CMD(aclnnNotifyDispatch, sendData, numTokensPerExpert, sendCount, numTokens,
                 groupEpPtr, numRanks, rank, localRankSize, localRankId,
                 sendDataOffset, recvData, totalRecvToken, recvCount, recvOffset, 
                 maxBs, recvTokensPerExpert);

    // Step 3: 计算 globalBs 和 numRecvTokens
    int64_t gBs = maxBs.item<int>() * numRanks;
    int64_t trt = totalRecvToken.item<int>();
    int numRecvTokens = (trt == 0) ? 1 : trt;

    // Step 4: 分配 dispatch 输出
    auto expandxOut = useQuant ? torch::empty({numRecvTokens, hidden}, at::kChar)
                               : torch::empty({numRecvTokens, hidden}, x.options());
    auto dynamicScalesOut = torch::empty({numRecvTokens}, at::kFloat);
    auto expandIdxOut = torch::empty({numRecvTokens * 3}, at::kInt);

    // Step 5: 调用 moe_dispatch_normal (实际 token 传输)
    EXEC_NPU_CMD(aclnnMoeDispatchNormal, x, expertIds, sendDataOffset, sendTokenIdxSmall,
                 recvOffset, recvCount, groupEpPtr, numRanks, rank,
                 groupEpPtr, tpSize, tpRank, numExperts, quantMode, gBs,
                 expandxOut, dynamicScalesOut, expandIdxOut, dispatchWaitRecvCostStatsOut);

    return {expandxOut, dynamicScalesOut, expandIdxOut, recvCount, recvTokensPerExpert};
}

7.5 moe_combine_prefill 绑定

at::Tensor MoeCombinePrefillImplNpu(...) {
    // 类型转换
    at::Tensor topkIdxInt32 = topkIdx.scalar_type() == at::kInt ? topkIdx : topkIdx.to(at::kInt);
    at::Tensor expertScales = topkWeights.scalar_type() == at::kFloat ? topkWeights : topkWeights.to(at::kFloat);

    // 分配输出
    auto combinedX = torch::empty({expertScales.size(0), hidden}, x.options());

    // 调用 combine
    EXEC_NPU_CMD(aclnnMoeCombineNormal, x, tokenSrcInfo, epSendCounts, expertScales,
                 tpSendCounts, groupEpPtr, numRanks, rank, groupEpPtr, tpWorldSize, tpRankId,
                 moeExpertNumber, globalBs, combinedX, combineSendCostStatsOut);

    return combinedX;
}

7.6 异常处理 (torch_bind_exception.h)

class TorchBindException : public std::exception {
    // 捕获文件名、行号、断言条件
    message = "Failed: " + name + " error " + file + ":" + line + " error message: " + error;
};

#define TORCH_BIND_ASSERT(cond) \
    if (not(cond)) { throw TorchBindException("Assertion", __FILE__, __LINE__, #cond); }

8. 通信模式全局分析

8.1 通信模式总结

算子	通信类型	HCCL 操作	数据量
dispatch_layout	无	无	纯本地计算
notify_dispatch	All-to-All	自管理 IPC	极小 (元数据)
moe_dispatch_normal	间接写入	Window IPC	大 (hidden states)
moe_combine_normal	间接写入	Window IPC	大 (expert 输出)

8.2 自管理 All-to-All 的优势

CAM 的 notify_dispatch 不直接使用 HCCL 的 All-to-All API，而是自己管理 IPC：

精确控制: 直接操作 Window 地址，避免 HCCL 额外开销
流水线: Ping-Pong 双缓冲实现计算通信重叠
同步优化: Magic+Value 编码实现无锁同步
内存效率: 原地更新，无需额外 buffer

8.3 通信拓扑

Level 0 (机内): FullMesh
  ┌───┐  ┌───┐  ┌───┐  ┌───┐
  │ 0 │──│ 1 │──│ 2 │──│ 3 │
  └───┘  └───┘  └───┘  └───┘
    │      │      │      │
  ┌───┐  ┌───┐  ┌───┐  ┌───┐
  │ 4 │──│ 5 │──│ 6 │──│ 7 │
  └───┘  └───┘  └───┘  └───┘

Level 1 (跨机): Pairwise
  Machine 0 ←──→ Machine 1 ←──→ Machine 2 ...

9. 量化机制深度分析

9.1 量化模式

enum { NO_SCALES = 0, DYNAMIC_SCALES = 2 };

quantMode=0: 无量化，BF16/FP16 直通
quantMode=2: 动态 INT8 量化（per-token 动态 scale）

9.2 量化流程 (Dispatch 阶段)

BF16 Token → FP32 → |x| → max(|x|) → scale = 127/max → x*scale → INT8
                                                         ↓
                                                  存储 1/scale 供反量化

9.3 反量化流程 (Combine 阶段)

// 如果 combine 需要反量化 (当前代码中 combine 直接操作 recvX)
Cast(tokenFloatLocal, tmpToken, CAST_NONE, axisH_);  // INT8 → FP32 (如果量化)
Muls(weightedMulBufLocal, tokenFloatLocal, scale, axisH_);

9.4 量化对内存的影响

非量化模式 (per token):

[h * 2B (BF16)] + [32B (scale)] + [12B (expandIdx)] = h*2 + 44 Bytes

量化模式 (per token):

[h * 1B (INT8)] + [32B (scale)] + [12B (expandIdx)] = h + 44 Bytes

带宽节省: 当 h=7168 时，非量化 14380B → 量化 7212B，节省约 50% Window 带宽。

10. 关键性能优化点

10.1 双缓冲 (Ping-Pong)

NotifyDispatch 的 CpGM2GMPingPong：

__ubuf__ U *inputUB[2] = {UB_HEAD_OFFSET, UB_MID_OFFSET};
for (int64_t i = 0; dataSizeRemain > 0; i++) {
    event_t eventId = (i & 1) ? EVENT_ID0 : EVENT_ID1;
    WaitFlag<MTE3_MTE2>(eventId);     // 等待写完成
    CpGM2UB(inputUB[i&1], input, size);
    CpUB2GM(output, outputUB[i&1], size);
    SetFlag<MTE3_MTE2>(eventId);      // 通知可开始下一轮
}

Combine 的 Magic 双 Window：

winDataSizeOffset_ = magic_ * (totalWinSize / 2UL);  // 交替使用两半

10.2 地址对齐

所有 Window 访问都严格对齐到 512B (WIN_ADDR_ALIGN)：

hOutGMAlignSize = Ceil(hScaleIdxSize, WIN_ADDR_ALIGN) * WIN_ADDR_ALIGN;
h512AlignRecvXLen_ = Ceil(hRecvXTypeLen_, WIN_512_ALIGN) * WIN_512_ALIGN;

原因: HCCL Window 的跨卡访问需要 512B 对齐以达到最佳 DMA 性能。

10.3 增量 Flag 检查

checkedFlagNum += i;  // 跳过已确认的 flag

避免重复检查已完成的 flag，减少 GM→UB 读取。

10.4 状态位批量 ReduceSum

// WaitStatus: 使用 ReduceSum 快速判断所有数据是否到达
ReduceSum(statusSumOutTensor, statusFp32Tensor, gatherMaskOutTensor, ...);
sumOfFlag = statusSumOutTensor.GetValue(0);

10.5 原子操作聚合

// dispatch_layout 中使用原子加实现跨 core 聚合
AscendC::SetAtomicAdd<T>();
for (int i = coreIdx_ + 1; i < aivNum_; ++i) {
    DataCopyPad(tempExpertGM_[i * numExperts], numTokensPerExpertTensor, ...);
}
AscendC::SetAtomicNone();

10.6 诊断模式 (可选)

if (isEnableDiagnose) {
    int32_t durationTime = (GetSystemCycle() - systemCycleStart) / CYCLE_TO_TIME;  // us
    waitRecvCostStatsTensor.SetValue(srcRank, preTime + durationTime);
}

设计亮点: 诊断输出是可选的 (OPTIONAL tensor)，不影响正常路径性能。

11. HCCL Window 内存布局

11.1 整体布局

┌──────────────────────────────────────────────────────────────────────────┐
│                        HCCL Window (默认 200MB)                          │
├──────────────────────────────────────────────────────────────────────────┤
│ 0x000000 │ NotifyDispatch 数据区 (IPC Data, 100MB)                       │
│          │   ├── Flag 区 (2MB)                                           │
│          │   └── Data 区 (100MB)                                         │
├──────────┼───────────────────────────────────────────────────────────────┤
│ 0xCC0000 │ Combine 状态区 (4MB) — COMBINE_STATE_WIN_OFFSET                │
├──────────┼───────────────────────────────────────────────────────────────┤
│ 0x1900000│ Dispatch 数据区 — NOTIFY_DISPATCH_WIN_OFFSET (204MB)           │
│          │   ├── Token Data (ping)                                       │
│          │   └── Token Data (pong)                                       │
├──────────┼───────────────────────────────────────────────────────────────┤
│          │ Combine 数据区                                                 │
│          │   ├── Token Data (ping)                                       │
│          │   └── Token Data (pong)                                       │
├──────────┼───────────────────────────────────────────────────────────────┤
│          │ 状态/Flag 区                                                   │
│          │   ├── DataState (ping/pong 标志)                               │
│          │   ├── Magic (combine 双缓冲标志)                                │
│          │   └── Per-token Status (数据到达标志)                           │
└──────────┴───────────────────────────────────────────────────────────────┘

11.2 Token 存储格式 (Dispatch)

每个 token slot (512B 对齐):
┌──────────────────────────────────────────────────────┐
│ [0, h*2)        │ BF16 hidden states (或 INT8 量化)    │
│ [h*2, h*2+32)   │ Scale (FP32, 量化模式)              │
│ [h*2+32, h*2+44)│ ExpandIdx: [srcRank, srcToken, topk] │
│ [h*2+44, 512n)  │ Padding                             │
└──────────────────────────────────────────────────────┘

11.3 Token 存储格式 (Combine)

每个 token slot (512B 对齐):
┌──────────────────────────────────────────────────────┐
│ [0, h*2)        │ Expert 输出 (BF16)                  │
│ [h*2, 512n)     │ Padding                             │
└──────────────────────────────────────────────────────┘

每个 slot 对应一个 status flag (32B 对齐):
┌──────────────────────────────────────────────────────┐
│ [0, 32)         │ 1.0 (数据就绪) / 0.0 (未就绪)       │
└──────────────────────────────────────────────────────┘

附录 A: 关键常量速查表

常量	值	来源文件	含义
CAM_MAX_RANK_SIZE	384	comm_args.h	最大 rank 数
IPC_BUFF_MAX_SIZE	100 MB	comm_args.h	IPC 数据区大小
IPC_DATA_OFFSET	2 MB	comm_args.h	IPC flag 区大小
UB_SINGLE_DMA_SIZE_MAX	190 KB	comm_args.h	单次 DMA 上限
UB_SINGLE_PING_PONG_ADD_SIZE_MAX	95 KB	comm_args.h	Ping-Pong 单缓冲
COMBINE_STATE_WIN_OFFSET	4 MB	tiling_args.h	Combine 状态偏移
NOTIFY_DISPATCH_WIN_OFFSET	204 MB	tiling_args.h	Dispatch 数据偏移
WIN_STATE_OFFSET	500 KB/950 KB	moe_dispatch_normal.h	状态区偏移
MAGIC_WIN_OFFSET	975 KB	moe_combine_normal.h	Magic 标志偏移
WIN_ADDR_ALIGN	512 B	多处	Window 地址对齐
FLAG_UNIT_INT_NUM	4	sync_collectives.h	每个 flag 占 4 个 int64
STATE_OFFSET	32	moe_dispatch_normal.h	状态位间隔

附录 B: 文件索引

src/cam/comm_operator/
├── ascend_kernels/
│   ├── AddCustom.json                          # 算子定义 (AddCustom 占位)
│   ├── operator_registry.json                  # 算子注册表
│   ├── dispatch_layout/
│   │   ├── op_api/aclnn_dispatch_layout.{h,cpp}
│   │   ├── op_host/dispatch_layout.cpp         # 算子定义
│   │   ├── op_host/dispatch_layout_tiling.cpp  # Tiling 逻辑
│   │   └── op_kernel/
│   │       ├── dispatch_layout.{h,cpp}         # Kernel 实现
│   │       └── dispatch_layout_tiling.h        # Tiling 数据结构
│   ├── notify_dispatch/
│   │   ├── op_api/aclnn_notify_dispatch.{h,cpp}
│   │   ├── op_host/notify_dispatch.cpp
│   │   ├── op_host/notify_dispatch_tiling.cpp
│   │   └── op_kernel/notify_dispatch.{h,cpp}
│   ├── moe_dispatch_normal/
│   │   ├── op_api/aclnn_moe_dispatch_normal.{h,cpp}
│   │   ├── op_host/moe_dispatch_normal.cpp
│   │   ├── op_host/moe_dispatch_normal_tiling.cpp
│   │   └── op_kernel/moe_dispatch_normal.{h,cpp}
│   ├── moe_combine_normal/
│   │   ├── op_api/aclnn_moe_combine_normal.{h,cpp}
│   │   ├── op_host/moe_combine_normal.cpp
│   │   ├── op_host/moe_combine_normal_tiling.cpp
│   │   └── op_kernel/moe_combine_normal.{h,cpp}
│   └── utils/
│       ├── op_host/
│       │   ├── tiling_args.h                   # 关键偏移常量
│       │   ├── mc2_tiling_utils.h              # Window 大小计算
│       │   ├── dfx_base.h                      # 日志基础设施
│       │   ├── ops_log.h / ops_error.h / error_log.h
│       └── op_kernel/
│           ├── comm_args.h                     # CommArgs 结构体
│           ├── sync_collectives.h              # 跨卡同步原语
│           ├── data_copy.h                     # GM↔UB 拷贝
│           └── moe_distribute_base.h           # HCCL 拓扑结构
└── pybind/
    ├── pybind.cpp                              # pybind11 模块 + TORCH_LIBRARY
    ├── functions.h                             # 函数声明
    ├── get_dispatch_layout.cpp                 # dispatch_layout 绑定
    ├── moe_dispatch_prefill.cpp                # dispatch 绑定 (notify + dispatch)
    ├── moe_combine_prefill.cpp                 # combine 绑定
    └── torch_bind_exception.h                 # 异常处理

附录 C: 关键洞察总结

C.1 设计范式

Window-based IPC: 所有跨卡数据交换通过 HCCL Window 直接内存访问，避免显式 Send/Recv
Flag-based Synchronization: 使用 Magic+Value 编码的 flag 实现无锁同步
Two-phase Pipeline: 先交换元数据 (notify)，再传输实际数据 (dispatch)
Self-managed All-to-All: notify_dispatch 自己管理 IPC，不依赖 HCCL 集合通信

C.2 性能关键路径

Token 路由延迟 = notify_dispatch (All-to-All 元数据) 
               + dispatch_normal (Window 写入 + 状态同步)
               + expert_compute
               + combine_normal (Window 读取 + 加权求和)

C.3 与竞品对比洞察

特性	CAM (华为)	NCCL (NVIDIA)	差异
All-to-All	自管理 IPC	NCCL 原生	CAM 更精细控制
同步	Magic+Flag	NCCL 内部	CAM 显式可见
量化	内置 INT8	需外部实现	CAM 端到端
双缓冲	Magic 翻转	无	CAM 避免冲突

C.4 潜在优化方向

量化 Combine: 当前 combine 阶段未利用 INT8 反量化，可在 Window 读取时直接反量化
TP 支持: 当前 TP 仅支持 ring AllGather/ReduceScatter，可探索 2D-HMesh
动态 Topk: 当前 topk 固定，可探索动态 topk 路由
Sparse Expert: 当前 expert 均匀分布，可探索负载均衡路由

← 编译系统与部署 📑 索引 FusedDeepMoE 融合算子 →

EP 通信算子深度分析 — UMDK/CAM

CAM EP Communication Kernels 深度分析报告

目录

1. 架构总览与设计哲学

1.1 CAM 的定位

1.2 算子调用链

1.3 核心数据流

2. 共享基础设施分析

2.1 CommArgs (utils/op_kernel/comm_args.h)

2.2 SyncCollectives (utils/op_kernel/sync_collectives.h)

2.3 DataCopy 工具 (utils/op_kernel/data_copy.h)

2.4 Mc2TilingUtils (utils/op_host/mc2_tiling_utils.h)

2.5 TilingArgs (utils/op_host/tiling_args.h)

3. dispatch_layout 算子

3.1 功能定位

3.2 op_api 层

3.3 op_host 层 (Tiling 逻辑)

3.4 op_kernel 层 (核心算法)

4. notify_dispatch 算子

4.1 功能定位

4.2 op_api 层

4.3 op_host 层

4.4 op_kernel 层 (核心算法)

4.5 同步机制

5. moe_dispatch_normal 算子

5.1 功能定位

5.2 op_api 层

5.3 op_host 层 (Tiling 逻辑)

5.4 op_kernel 层 (核心算法)

5.5 量化机制 (QuantProcess)

6. moe_combine_normal 算子

6.1 功能定位

6.2 op_api 层

6.3 op_host 层 (Tiling 逻辑)

6.4 op_kernel 层 (核心算法)

6.5 Magic 双缓冲机制

7. Pybind 绑定层

7.1 注册机制

7.2 三层调用架构

7.3 get_dispatch_layout 绑定

7.4 moe_dispatch_prefill 绑定

7.5 moe_combine_prefill 绑定

7.6 异常处理 (torch_bind_exception.h)

8. 通信模式全局分析

8.1 通信模式总结

8.2 自管理 All-to-All 的优势

8.3 通信拓扑

9. 量化机制深度分析

9.1 量化模式

9.2 量化流程 (Dispatch 阶段)

9.3 反量化流程 (Combine 阶段)

9.4 量化对内存的影响

10. 关键性能优化点

10.1 双缓冲 (Ping-Pong)

10.2 地址对齐

10.3 增量 Flag 检查

10.4 状态位批量 ReduceSum

10.5 原子操作聚合

10.6 诊断模式 (可选)

11. HCCL Window 内存布局

11.1 整体布局

11.2 Token 存储格式 (Dispatch)

11.3 Token 存储格式 (Combine)

附录 A: 关键常量速查表

附录 B: 文件索引

附录 C: 关键洞察总结

C.1 设计范式

C.2 性能关键路径

C.3 与竞品对比洞察

C.4 潜在优化方向

2.1 CommArgs (`utils/op_kernel/comm_args.h`)

2.2 SyncCollectives (`utils/op_kernel/sync_collectives.h`)

2.3 DataCopy 工具 (`utils/op_kernel/data_copy.h`)

2.4 Mc2TilingUtils (`utils/op_host/mc2_tiling_utils.h`)

2.5 TilingArgs (`utils/op_host/tiling_args.h`)