CAM cam_feature 分支四大新通信算子深度分析

分支: remotes/origin/cam_feature | 分析日期: 2026-07-21 目标芯片: Ascend 910_93 (a2e/e2a) / Ascend 910_93 + 910B (detour 系列)

1. 新算子总体定位

1.1 背景问题

在 MoE（Mixture-of-Experts）+ EP（Expert Parallelism）架构中，Attention 侧与 Expert 侧通常运行在不同的 NPU 卡簇上。这两侧之间存在大规模的数据搬运需求：

前向: Attention 输出的 token hidden states 需要按 expert 路由结果分发到对应 Expert 卡 → All-to-Expert (a2e)
反向/回传: Expert 计算完成的 token 结果需要汇总回 Attention 卡 → Expert-to-All (e2a)

此外，当 MoE 集群规模超出单网络拓扑的 fullmesh 互连范围时（例如跨 pod、跨交换机），部分 rank 对之间没有直连链路，通信必须通过中间 rank 绕行 → Detour 机制。

1.2 四个算子的角色

算子	方向	核心功能	场景
a2e	Attention → Expert	将 token hidden states + expert IDs + scales 从 Attention 卡发送到 Expert 卡	MoE 前向 dispatch 的跨簇通信
e2a	Expert → Attention	将 Expert 计算完成的 token 结果发回 Attention 卡	MoE 前向 combine 的跨簇通信
all_to_all_detour	全对全 + 绕行	在 non-fullmesh 拓扑下，通过中间 rank 绕行完成 All-to-All	跨 pod MoE EP 全交换
reduce_scatter_detour	归约散射 + 绕行	在 non-fullmesh 拓扑下，通过绕行完成 Reduce-Scatter	跨 pod 梯度同步 / 归约

1.3 与 master 分支的差异

operator_registry.json 明确显示这四个算子仅存在于 cam_feature 分支的 ascend910_93 列表中。master 分支的 MoE 通信依赖 HCCL 原生 collective（AlltoAll、ReduceScatter）或已存在的 moe_dispatch_* / moe_combine_* 算子，不包含跨 Attention-Expert 簇的 SHMEM 直传算子和 Detour 绕行算子。

2. a2e (All-to-Expert)

源文件: src/cam/comm_operator/ascend_kernels/a2e/

2.1 算子签名

a2e(x, expert_ids, scales, batch_size, hidden_size, topk,
    expert_rank_size, attention_rank_size, rank, group_ep, aiv_num, compute_gate)
→ (expand_x, simulate_expert_ids, simulate_expert_scales, atten_batch_size, x_active_mask)

2.2 核心数据结构

A2ETilingData (a2e_tiling.h):

struct A2ETilingData {
    Mc2InitTiling mc2InitTiling;    // MC2 初始化 tiling
    Mc2CcTiling mc2CcTiling1;       // MC2 通信 tiling
    uint32_t batchSize;
    uint32_t hiddenSize;
    uint32_t topk;
    uint32_t expertRankSize;        // Expert 侧 rank 数
    uint32_t attentionRankSize;     // Attention 侧 rank 数
    uint32_t rank;                  // 当前 rank
    uint32_t computeGate;           // 计算门控（决定发送内容）
};

2.3 执行流程

入口 (op_kernel/a2e.cpp): - 仅 TILING_KEY_IS(21)（BF16）被实际编译 - 实例化 A2e<bfloat16_t, bfloat16_t, false>，调用 init() → process()

init() 阶段 (a2e.h):

获取 HCCL IPC 上下文: cpp epWinContext_ = (__gm__ HcclOpResParam *)AscendC::GetHcclContext<HCCL_GROUP_ID_0>(); 通过 localWindowsIn 和 remoteRes[i].nextDevicePtr->windowsIn 获取每张卡的 IPC 共享内存窗口基址。
Magic 计数器初始化（跨 block 同步令牌）: ```cpp magicTensor_.SetGlobalBuffer((gm int32_t*)((epWinContext_->localWindowsIn)
- IPC_DATA_OFFSET - blockNum * sizeof(int32_t) * INT32_COUNT_PER_BLOCK)); `` 每个 block 通过AtomicAdd在共享内存中顺序写入递增的 blockIdx 值，读取到的值即为该 block 的唯一magic` 序号。这是无锁分布式 barrier 令牌。
共享内存地址映射 (shareAddrs[]): - Attention 卡 (rank >= expertRankSize): 只需映射自己和对应 Expert 卡的地址 - Expert 卡 (rank < expertRankSize): 映射所有 rank 的地址 - 地址公式: localWindowsIn + (magic % 2) * 800MB + rank * 512 - DOUBLE_BUFFER_COUNT = 2 实现 ping-pong 避免上一轮/本轮数据冲突 - OPT_RANK_OFFSET = 512 保证每张卡的 flag 区域不互相覆盖
attnToMoeRatio 计算: cpp attnToMoeRatio = attentionRankSize > expertRankSize ? (attentionRankSize + expertRankSize - 1) / expertRankSize : 1; 表示每个 Expert 卡需要接收来自多少张 Attention 卡的数据。当 attentionRankSize 不能整除 expertRankSize 时，部分 Expert 卡少收一份。

process() 阶段 — 根据 computeGate 分两模式:

模式 A: `computeGate == 0`（纯数据搬运模式）

Attention 卡 (rank >= expertRankSize):
    sendWithMte() → 仅发送 token X

Expert 卡 (rank < expertRankSize):
    for i in 0..attnToMoeRatio:
        recvWithMte(i) → 从 rank + (i+1)*expertRankSize 接收 X

发送端 (sendX): 多核协作 copyGmToGmWithBlocks 将 batchSize * hiddenSize 的 token 数据从本地 GM 拷贝到目标 rank 的 IPC 共享内存，然后写 flag 通知接收端
接收端 (recvWithMte): waitFlagWithScalar 轮询 flag 到位后，从源 rank 的共享内存拷贝到本地 expandX

模式 B: `computeGate == 1`（分核分工模式）

Expert 卡:
    blockIdx == blockNum-1  → sendExpertIds()   // 发送 expert IDs
    blockIdx == blockNum-2  → sendExpertScales() // 发送 expert scales
    其他 block              → sendX()           // 发送 token X

每个 Expert 卡将 expert_ids（路由结果）和 scales（权重）也一并发送给对应的 Attention 卡，使得 Attention 卡可以本地模拟 Expert 分布，用于后续 simulate_expert_ids / simulate_expert_scales 输出。

2.4 数据布局（IPC 共享内存）

shareAddrs[rank] 指向的内存布局:
┌──────────────────────┐ offset 0
│  batchSize (int32)   │  4 bytes
├──────────────────────┤ offset 4
│  expertIds           │  batchSize * topk * sizeof(int32)
├──────────────────────┤ offset 4 + expertIdsSize
│  expertScales        │  batchSize * topk * sizeof(float)
├──────────────────────┤ offset 4 + expertIdsSize + scalesSize
│  X (token data)      │  batchSize * hiddenSize * sizeof(bf16)
└──────────────────────┘

Flag 区域位于 IPC_DATA_OFFSET 之前（前 2MB），使用 mergeMagicWithValue(magic, 0) 编码:

uint64_t mergeMagicWithValue(uint32_t magic, uint32_t count) {
    return (static_cast<uint64_t>(magic) << 32) | static_cast<uint64_t>(count);
}

Flag 偏移量按 (flagIndex * attnToMoeRatio + sendOffset) 索引，每个 Expert 卡可能接收多份 Attention 卡的数据，通过 attnToMoeRatio 区分。

2.5 同步机制

发送端写 flag (通知对端数据就绪):

GlobalTensor<uint64_t> shareFlagGt;
shareFlagGt.SetGlobalBuffer(shareAddrs[rank % expertRankSize]
    + (WAIT_FLAG_OFFSET_3 * attnToMoeRatio + sendOffset) * flagUnitInt64Num);
camCpUB2GM(shareFlagGt, flagLt, 1, sizeof(uint64_t));

接收端轮询 flag:

__aicore__ inline void waitFlagWithScalar(int addr, uint32_t magic) {
    GlobalTensor<uint32_t> flagGt;
    flagGt.SetGlobalBuffer((__gm__ uint32_t *)(shareAddrs[rank] + addr));
    while(1) {
        DATA_FULSH(flagGt, uint32_t);  // cache flush + invalidate
        if (flagGt.GetValue(0) == magic) return;
    }
}

DATA_FULSH 宏确保每次轮询都重新从内存读取（绕过 cache），通过 DataCacheCleanAndInvalid + dsb(DSB_ALL) 实现。

2.6 参数约束

参数	约束
batchSize	1 ~ 1024
hiddenSize	1024 ~ 7168, 256 对齐
topk	1 ~ 8
expertRankSize	> 0 且 ≤ attentionRankSize
aivAlgNum	4 ~ 48
computeGate	0 或 1

2.7 与 dispatch 的关系

a2e 是 MoE dispatch 阶段的跨簇通信原语。传统 dispatch（moe_dispatch_* 算子）在同一 EP group 内部做 token-to-expert 路由。而 a2e 解决的是Attention 卡簇与 Expert 卡簇之间的物理分离场景 — Attention 卡持有 token，Expert 卡持有 expert 权重，中间通过非 HCCL 路径（可能是 PCIe/SIO 互连而非 RDMA）传输。

从 sample 代码可见，Attention 侧发送真实数据，Expert 侧用空 tensor 占位接收 — a2e 是角色感知的，根据 rank < expertRankSize 自动判断当前是 Expert 侧还是 Attention 侧。

3. e2a (Expert-to-All)

源文件: src/cam/comm_operator/ascend_kernels/e2a/

3.1 算子签名

e2a(expand_x, atten_batch_size, batch_size, hidden_size, topk,
    expert_rank_size, attention_rank_size, rank, group_ep, aiv_num)
→ x

3.2 核心数据结构

E2ATilingData (e2a_tiling.h):

struct E2ATilingData {
    Mc2InitTiling mc2InitTiling;
    Mc2CcTiling mc2CcTiling1;
    uint32_t batchSize;
    uint32_t hiddenSize;
    uint32_t topk;
    uint32_t expertRankSize;
    uint32_t attentionRankSize;
    uint32_t rank;
};

注意: e2a 的 tiling 没有 computeGate，因为 e2a 只负责回传 token 数据。

3.3 执行流程

入口: 仅 TILING_KEY_IS(21) (BF16) 编译，实例化 E2a<bfloat16_t>。

process() 阶段 — 极为简洁:

if (rank < expertRankSize) {
    sendWithMte();   // Expert 卡: 发送结果回 Attention 卡
} else {
    recvWithMte();   // Attention 卡: 接收 Expert 结果
}

Expert 端 (`sendWithMte`):

int sendRank = rank + expertRankSize;  // 目标 Attention 卡
for (int i = 0; i < attnToMoeRatio; i++) {
    // 将 expandX 中对应分片拷贝到目标 Attention 卡的 IPC 内存
    copyGmToGmWithBlocks(shareXGt, expandXGt[sendOffset],
        sendBatchSize * hiddenSize, blockNum, blockIdx);
    sendOffset += sendBatchSize * hiddenSize;
    sendRank += expertRankSize;
}
SyncAll();
// 写完所有分片后，一次性通知所有目标 Attention 卡
for (int i = 0; i < attnToMoeRatio; i++) {
    GlobalTensor<uint64_t> shareFlagGt;
    shareFlagGt.SetGlobalBuffer((__gm__ uint64_t *)(shareAddrs[sendRank]));
    camCpUB2GM(shareFlagGt, flagLt, 1, sizeof(uint64_t));
}

Attention 端 (`recvWithMte`):

// 轮询等待 flag
while (1) {
    camCpGM2UB(flagLt, flagGt, sizeof(uint64_t));
    if (flagLt.GetValue(0) == mergeMagicWithValue(magic, 0, 0)) break;
}
// flag 到位后，从 IPC 内存拷贝结果到本地
GlobalTensor<T> shareXGt;
shareXGt.SetGlobalBuffer((__gm__ T *)(shareAddrs[rank] + IPC_DATA_OFFSET));
copyGmToGmWithBlocks(xGt, shareXGt, batchSize * hiddenSize, blockNum, blockIdx);

3.4 a2e/e2a 对称性分析

维度	a2e	e2a
方向	Attention → Expert	Expert → Attention
发送方	Attention 卡 (rank ≥ expertRankSize)	Expert 卡 (rank < expertRankSize)
接收方	Expert 卡	Attention 卡
传输内容	X + expertIDs + scales (computeGate=1)	仅 X
发送端分核	computeGate=1 时分核发送 IDs/scales/X	不分核，所有核协作发 X
接收端行为	多核协作接收，按 attnToMoeRatio 拼接	多核协作接收完整 batch
Flag 机制	按消息类型分 flag 偏移	单一 flag 通知
computeGate	有 (0/1 两模式)	无

本质区别: a2e 是多对多分发（一个 Expert 卡接收多个 Attention 卡的 token），e2a 是多对多回传（一个 Attention 卡接收多个 Expert 卡的结果）。a2e 需要更复杂的 flag 系统来区分不同消息类型（batchSize、expertIds、scales、X），e2a 只需一个 flag 通知"数据就绪"。

3.5 与 combine 的关系

e2a 是 MoE combine 阶段的跨簇通信原语。传统 combine（moe_combine_* 算子）在 EP group 内部聚合 expert 输出。e2a 解决 Expert 计算完成后，将结果从 Expert 卡簇回传到 Attention 卡簇的问题。

4. all_to_all_detour

源文件: src/cam/comm_operator/ascend_kernels/all_to_all_detour/

4.1 算子签名

all_to_all_detour(sendData, commRankIds, commArgs, magic) → recvData

sendData: 发送缓冲区，shape [sendCount]
commRankIds: 参与通信的 rank ID 列表，shape [commRankCount]
commArgs: 通信参数结构体（含 peerMems 和 memLen）
magic: 同步令牌

4.2 核心数据结构

All2AllDetourTilingData (all2_all_detour_tiling.h):

BEGIN_TILING_DATA_DEF(All2AllDetourTilingData)
    TILING_DATA_FIELD_DEF(uint32_t, magic);
    TILING_DATA_FIELD_DEF(uint32_t, sendCount);
    TILING_DATA_FIELD_DEF(uint32_t, commRankCount);
END_TILING_DATA_DEF;

4.3 核心算法 (`all2all_small_fullmesh_with_detour.h`)

All2AllSmallFullmeshWithDetour<T> 继承自 Collectives 基类，实现两阶段 detour 通信。

4.3.1 Init 阶段

void Init(KERNELS_ARGS_FUN_ALL2ALL, GM_ADDR commRankIds, uint32_t commRankSize) {
    // 1. 读取 commRankIds，判断当前 rank 是否参与通信
    for (int i = 0; i < commRankSize; ++i)
        this->commRankIds[i] = commRankIdsGM.GetValue(i);

    if (!IsValidCommRank()) return;  // 当前 rank 不在通信列表中则退出

    // 2. 初始化 Collectives 基类（使用 peerMems 而非 HCCL window）
    Collectives::InitByCamCommDetour(KERNELS_ARGS_CALL_MIX);

    // 3. 计算每 rank 数据量和 detour 分片量
    perRankDataNum = GetDataCount(len, commRankSize);  // len / commRankSize
    perRankDataNumForDetour = (perRankDataNum + rankSize - commRankSize)
        / (rankSize - commRankSize + 1);

    // 4. 初始化核心分组
    InitCoreGroup();
}

关键常量: - MAX_CORE_NUM = 48（Ascend 910_93 的 AIV 核数） - rankNumPerCore = (rankSize + 46) / 46 — 每核负责的绕行 rank 数

4.3.2 Process 阶段 — 两阶段流水线

Stage 1: InputToShareSlice()  — 从 input 读取数据，经 detour rank 写入目标 rank 的共享内存
Stage 2: ShareToShareSlice()  — 从共享内存读取数据，写入 output

Stage 1: InputToShareSlice() — 按 blockIdx 分三种情况:

blockIdx == 0: 本地直拷（自己发给自己的数据） cpp CpGM2GMPingPong<T>(perRankDataNum * sizeof(T), inputGt[commRankIdx * perRankDataNum], outputGt[commRankIdx * perRankDataNum], COPYONLY);
blockIdx == 1: 第一块 detour 数据 — 每个目标 comm rank 发送 perRankDataNumForDetour 大小的首块数据，写入对应 detour rank 的共享内存，并写 flag 通知 ```cpp writeGt = shareAddrs[detourRankId] + GetShareMemOffset(rank, dstCommRank, detourRankId) + IPC_DATA_OFFSET; CpGM2GMPingPong(perRankDataNumForDetour * sizeof(T), readGt, writeGt, COPYONLY); // 写 flag CpUB2GM(shareAddrs[detourRankId] + GetShareMemOffset(rank, dstCommRank, detourRankId)
- rank * FLAG_UNIT_INT_NUM, inputUB, sizeof(int64_t)); ```
blockIdx ≥ 2: 后续块 — 按 rankNumPerCore 切分剩余数据，每核处理一部分 detour rank 的拷贝

Stage 2: ShareToShareSlice() — 从共享内存收数据写入 output:

blockIdx == 0: 无操作（直拷已在 Stage 1 完成）
blockIdx == 1: 接收每个源 rank 的首块数据 cpp sync.WaitSyncFlag(magic, 1, srcCommRank, rank, 1, GetShareMemOffset(srcCommRank, rank, rank)); CpGM2GMPingPong<T>(perRankDataNumForDetour * sizeof(T), readGt, outputGt[...], COPYONLY);
blockIdx ≥ 2: 接收后续块，通过 detour rank 中转的数据

4.3.3 共享内存地址计算

int64_t GetShareMemOffset(int srcRank, int dstRank, int memRank) {
    return (rankSize * srcRank + dstRank) * memLen
        + memRank * OPT_RANK_OFFSET
        + (magic % PING_PONG_SIZE) * (memLen / 2);
}

(rankSize * srcRank + dstRank) * memLen: 二维矩阵寻址，每对 rank 有独立的内存条带
memRank * OPT_RANK_OFFSET (512): 在条带内按 memRank 偏移，避免多核写同一地址
(magic % 2) * (memLen / 2): ping-pong 双缓冲

4.4 Detour 路由逻辑

GetDstRankId(order) — 获取第 order 个非通信 rank（即 detour rank）的 ID:

int GetDstRankId(int order) {
    int count = 0;
    for (int i = 0; i < rankSize; ++i) {
        if (!IsCommRank(i)) {  // 遍历所有非 commRank 的 rank
            if (count == order) return i;
            ++count;
        }
    }
}

Detour 的含义: 当 commRankSize < rankSize 时，部分 rank 不在通信列表中（例如它们是另一 pod 的 rank，或纯转发节点）。这些 rank 不直接参与 All-to-All 的数据生产/消费，但作为中转站帮助数据绕行到目标。

4.5 数据类型支持

Tiling Key	数据类型
20	float16
21	bfloat16
22	float32
23	int32

支持 5 种 input format 组合（含 FORMAT_NC1HWC0 用于特殊场景）。

4.6 平台适配

// aclnn_all2_all_detour.cpp
if (op::GetCurrentPlatformInfo().GetSocVersion() == op::SocVersion::ASCEND910B) {
    NnopbaseSetHcclServerType(executor, NNOPBASE_HCCL_SERVER_TYPE_AICPU);
} else {
    NnopbaseSetHcclServerType(executor, NNOPBASE_HCCL_SERVER_TYPE_MTE);
}

910B 使用 AICPU 调度，910_93 使用 MTE（Memory Transfer Engine）调度 — 反映两种芯片的 DMA 能力差异。

5. reduce_scatter_detour

源文件: src/cam/comm_operator/ascend_kernels/reduce_scatter_detour/

5.1 算子签名

reduce_scatter_detour(input, commRankIds, commArgs, magic, rank_size, op) → output

input: 输入数据，shape [sendCount]
commRankIds: 参与通信的 rank ID 列表
commArgs: 通信参数（其中 commArgs[0]=rank, commArgs[2]=rankSize）
magic: 同步令牌
rank_size: 通信域大小
op: 归约操作类型（ADD=0, MAX=2 等）

5.2 核心数据结构

ReduceScatterDetourTilingData (reduce_scatter_detour_tiling.h):

struct ReduceScatterDetourTilingData {
    uint32_t sendCount;
    uint32_t op;
    int64_t magic;
    uint32_t commRankCount;
};

5.3 核心算法 (`reduce_scatter_detour.h`)

ReduceScatterWithDetour<T, U> 继承自 Collectives，模板参数: - T: 输入数据类型（如 float16） - U: 累加器类型（如 float），精度更高

5.3.1 Init 阶段

void Init(KERNELS_ARGS_FUN_MIX, GM_ADDR commRankIds, uint32_t commRankSize) {
    // 1. 读取 commRankIds
    // 2. 初始化 Collectives 基类
    detourRankSize = rankSize - commRankSize;  // 绕行 rank 数

    // 3. 计算各阶段核心数
    in2IpcCoreNum = (detourRankSize + 1) * (commRankSize - 1) + 1;
    reduceCoreNum = detourRankSize + 2;
    blockNum = max(in2IpcCoreNum, reduceCoreNum);

    // 4. 计算 detour 分片量
    perDataNumForDetourRank = (len + detourRankSize) / (detourRankSize + 1);
    piplineNum = commRankSize - 1;  // 归约流水线级数
}

核心分配策略: - Input2Ipc: 需要 (detourRankSize + 1) * (commRankSize - 1) + 1 个核 - Reduce: 需要 detourRankSize + 2 个核 - 最终 blockNum = max(两者)，确保两阶段都有足够核

5.3.2 Process 阶段 — 两阶段流水线

Stage 1: Input2Ipc()  — 将输入数据分片写入各 rank 的 IPC 共享内存
Stage 2: Reduce()     — 从共享内存读取各分片，归约后写回输出

Stage 1: `Input2Ipc()` — 数据分发

blockIdx == 0: 本地数据直拷到自己的共享内存 cpp in2IpcDstGtU.SetGlobalBuffer(shareAddrs[rank] + shareMemOffsetInRank + IPC_DATA_OFFSET, len); CpGM2GMPingPong(len * sizeof(U), in2IpcSrcGt, in2IpcDstGtU, COPYONLY); // 写 flag 通知归约阶段 CpUB2GM(shareAddrs[rank] + shareMemOffsetInRank + rank * FLAG_UNIT_INT_NUM, inputUB, sizeof(int64_t));
blockIdx ≥ 1: 分发到其他 rank 的共享内存 ```cpp // 计算目标 comm rank 和 detour rank int targetCommRankIdx = (blockIdx - 1) / (detourRankSize + 1); // 跳过自己 if (targetCommRankIdx >= commRankIdx) targetCommRankIdx++; in2IpcTargetCommRank = commRankIds[targetCommRankIdx];

int offsetIdx = (blockIdx - 1) % (detourRankSize + 1); in2IpcTargetRank = (offsetIdx == 0) ? in2IpcTargetCommRank : GetDetourRankId(offsetIdx - 1);

// 拷贝数据分片到目标 detour rank 的共享内存 in2IpcDstGt.SetGlobalBuffer(shareAddrs[in2IpcTargetRank] + shareMemOffsetInDstRank + IPC_DATA_OFFSET, curDataNumForDetourRank); CpGM2GMPingPong(curDataNumForDetourRank * sizeof(T), in2IpcSrcGt, in2IpcDstGt, COPYONLY); ```

Detour 分发逻辑: 每个 comm rank 的数据被切成 detourRankSize + 1 份，其中一份发给目标 rank 本身（offsetIdx == 0），其余通过 detour rank 转发。

Stage 2: `Reduce()` — 归约计算

void Reduce() {
    if (blockIdx == 0) return;  // block 0 不参与归约

    int offsetIdx = (blockIdx - 1) % (detourRankSize + 1);
    int64_t curReduceDataNum = min(len - offsetIdx * perDataNumForDetourRank, perDataNumForDetourRank);

    // 目标 buffer
    reduceDstGt = shareAddrs[rank] + shareMemOffsetInRank + IPC_DATA_OFFSET
        + (blockIdx - 1) * perDataNumForDetourRank;

    for (int i = 0; i < piplineNum; i++) {
        int srcCommRankIdx = (commRankIdx + 1 + i) % commRankSize;
        int srcCommRank = commRankIds[srcCommRankIdx];
        int detourRank = (offsetIdx == 0) ? rank : GetDetourRankId(offsetIdx - 1);

        // 等待源 rank 数据就绪
        if (i == 0) {
            sync.WaitSyncFlag(magic, 1, rank, rank, 1, shareMemOffsetInRank);
        } else {
            sync.WaitOuterFlagDetour(magic, i, rank, rankSize + rank, shareMemOffsetInRank);
        }
        sync.WaitSyncFlag(magic, 1, srcCommRank, detourRank, 1, shareMemOffsetInDstRank);

        // 归约: 将源数据拷贝到目标（CpGM2GMPingPong 支持 ADD/MUL/MAX/MIN）
        CpGM2GMPingPong(curReduceDataNum * sizeof(U), reduceSrcGt, reduceDstGt, atomOp);

        // 设置 outer flag 通知下一级可以开始
        sync.SetOuterFlagDetour(magic, i + 1, rank, rankSize + rank, shareMemOffsetInRank);
    }

    // 等待最后一级完成
    sync.WaitOuterFlagDetour(magic, piplineNum, rank, rankSize + rank, shareMemOffsetInRank);

    // 从共享内存写回输出
    Ipc2OutDstGt = output + (blockIdx - 1) * perDataNumForDetourRank;
    CpGM2GMPingPong(curReduceDataNum * sizeof(T), reduceDstGt, Ipc2OutDstGt, COPYONLY);
}

归约流水线: piplineNum = commRankSize - 1 级，每级从一个源 comm rank 读取数据并归约到本地 buffer。使用 WaitOuterFlagDetour / SetOuterFlagDetour 实现流水线级间同步 — 第 i 级完成后设置 flag，第 i+1 级等待该 flag。

5.4 与标准 Reduce-Scatter 的差异

维度	标准 Reduce-Scatter (HCCL)	reduce_scatter_detour
拓扑要求	需要 fullmesh 或特定拓扑	支持 non-fullmesh，通过 detour 绕行
中间节点	无	使用非 commRank 作为转发节点
归约方式	硬件 tree/ring 归约	软件流水线逐级归约
精度	输入类型直接归约	支持高精度累加器 (T→U, 如 fp16→fp32)
同步	HCCL 内部 barrier	显式 flag 同步 (SyncCollectives)
内存	HCCL 内部 buffer	IPC 共享内存 (peerMems)

5.5 平台与类型

仅支持 Ascend 910B（this->AICore().AddConfig("ascend910b")）
仅支持 FP16 输入（TILING_KEY_FLOAT16 = 20）
累加器为 float（ReduceScatterWithDetour<float16_t, float>）
需要 16MB workspace（SYSTEM_NEED_WORKSPACE = 16 * 1024 * 1024）

6. Detour 机制深度分析

6.1 什么是网络绕行 (Detour)

在超大规模 MoE 集群中，NPU 卡通常按网络拓扑分层组织:

Pod 0 (Fullmesh via HCCS)          Pod 1 (Fullmesh via HCCS)
┌─────────────────────┐            ┌─────────────────────┐
│ Rank 0 ─ Rank 1     │            │ Rank 8 ─ Rank 9     │
│   │        │        │  RDMA/     │   │        │        │
│ Rank 2 ─ Rank 3     │◄──────────►│ Rank 10─ Rank 11    │
│   │        │        │  RoCE      │   │        │        │
│ Rank 4 ─ Rank 5     │  (高延迟)   │ Rank 12─ Rank 13    │
│   │        │        │            │   │        │        │
│ Rank 6 ─ Rank 7     │            │ Rank 14─ Rank 15    │
└─────────────────────┘            └─────────────────────┘

当 Rank 0 需要给 Rank 8 发送数据时，如果跨 pod 带宽不足或链路拥塞，可以通过中间 rank 绕行:

Rank 0 → Rank 3 (同 pod, 高带宽 HCCS) → Rank 11 (跨 pod) → Rank 8

Detour 机制的本质是: 将一次远距离通信拆分为多次短距离通信，利用拓扑局部性降低拥塞、提高有效带宽。

6.2 Detour 在 CAM 中的实现

CAM 的 Detour 机制通过 commRankIds 参数显式指定哪些 rank 参与实际通信:

// all2all_small_fullmesh_with_detour.h
bool IsValidCommRank() {
    for (int i = 0; i < commRankSize; i++) {
        if (commRankIds[i] == rank) { commRankIdx = i; return true; }
    }
    return false;  // 当前 rank 是 detour rank，不直接参与通信
}

两种角色: 1. CommRank (通信 rank): 实际生产/消费数据的 rank，在 commRankIds 列表中 2. DetourRank (绕行 rank): 不在 commRankIds 中，仅做数据转发

6.3 Detour 数据流 (以 all_to_all_detour 为例)

          Stage 1: InputToShareSlice          Stage 2: ShareToShareSlice
          ─────────────────────────          ──────────────────────────
CommRank A ──► DetourRank X ──► CommRank B    CommRank B ◄── DetourRank X ◄── CommRank A
   │                │               │            │               │              │
   │  1. A 读取本地  │               │            │  4. B 从 X 的   │              │
   │      input 数据 │               │            │     共享内存读   │              │
   │  2. A 写入 X   │               │            │     取数据      │              │
   │     的共享内存  │               │            │  5. B 写入     │              │
   │  3. A 写 flag  │               │            │      output    │              │
   │     通知 X     │               │            │               │              │

6.4 为什么需要 Detour — 拥塞与拓扑不匹配

问题 1: 拓扑不匹配 - HCCL 原语（AlltoAll、ReduceScatter）假设 fullmesh 或特定分层拓扑 - 当实际拓扑是部分连接（如某些 rank 对之间无直连）时，HCCL 可能无法正确路由 - Detour 通过软件显式指定转发路径，绕过拓扑限制

问题 2: 跨 pod 拥塞 - 跨 pod 带宽远低于 pod 内带宽（HCCS vs RoCE） - 直接 All-to-All 会导致跨 pod 链路热点 - Detour 可以将跨 pod 流量分散到多个中间 rank，负载均衡

问题 3: 小规模通信优化 - 算子名中的 "small" 暗示这是针对小数据量场景的优化 - 小数据量下 HCCL 启动开销占比高，使用 SHMEM 直传 + detour 可降低延迟

6.5 Detour 与标准 All-to-All 的复杂度对比

维度	标准 All-to-All	Detour All-to-All
跳数	1 跳 (直连)	2 跳 (经 detour rank)
总通信量	N×(N-1) 条消息	N×(N-1)×2 条消息 (翻倍)
单跳距离	可能很远 (跨 pod)	总是很短 (同 pod)
带宽利用	跨 pod 瓶颈	利用 pod 内高带宽
延迟	单次长延迟	两次短延迟 (可能更优)

适用场景: 当跨 pod 延迟 >> pod 内延迟时，2×短延迟 < 1×长延迟，detour 反而更快。

7. SHMEM 使用

7.1 SHMEM 在四个算子中的使用

算子	SHMEM 类型	获取方式	用途
a2e	HCCL IPC Window	`epWinContext_->localWindowsIn` + `remoteRes[i].windowsIn`	Attention↔Expert 数据传输
e2a	HCCL IPC Window	同上	Expert↔Attention 数据回传
all_to_all_detour	peerMems (CommArgs)	`commArgs->peerMems[rank]`	Detour rank 间数据中转
reduce_scatter_detour	peerMems (CommArgs)	同上	归约中间结果共享

7.2 两种 SHMEM 机制对比

HCCL IPC Window (a2e/e2a): - 由 HCCL 初始化时分配，通过 GetHcclContext 获取 - 结构: HcclOpResParam.localWindowsIn (本地窗口) + remoteRes[i].nextDevicePtr->windowsIn (远端窗口) - 大小: IPC_BUFF_MAX_SIZE_MUL_EXP = 800MB (a2e 中) / IPC_BUFF_MAX_SIZE = 100MB (comm_args.h) - 布局: 前 2MB 为 flag 区 (IPC_DATA_OFFSET = 2MB)，后续为数据区

peerMems (detour 系列): - 由 CommArgs 结构体携带，在算子初始化时由框架分配 - peerMems[rankId] 直接给出每张卡的可访问内存基址 - 通过 InitByCamCommDetour 初始化: cpp peerMemsAddrGm_.SetGlobalBuffer(&(commArgs)->peerMems[0], CAM_MAX_RANK_SIZE); for (int i = 0; i < rankSize; i++) shareAddrs[i] = (GM_ADDR)peerMemsAddrGm_.GetValue(i);

7.3 同步原语

a2e/e2a 的自定义 flag 同步:

// 发送端: 写 flag
LocalTensor<uint64_t> flagLt;
flagLt.SetValue(0, mergeMagicWithValue(magic, 0));
camCpUB2GM(shareFlagGt, flagLt, 1, sizeof(uint64_t));

// 接收端: 轮询 flag
while(1) {
    DATA_FULSH(flagGt, uint32_t);  // cache flush
    if (flagGt.GetValue(0) == magic) return;
}

detour 系列的 SyncCollectives: - SetSyncFlag(magic, value, eventID, rank): 在指定 rank 的共享内存写 flag - WaitSyncFlag(magic, value, eventID, rank): 轮询等待 flag 到位 - SetInnerFlag / WaitInnerFlag: 卡内 block 间同步 - SetOuterFlag / WaitOuterFlag: 跨卡同步 - SetChunkFlag / GetChunkFlag: 分块传输同步

Magic 编码:

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

高 32 位为 magic（防止不同轮次的 flag 误匹配），低 32 位为具体值。

7.4 Ping-Pong 双缓冲

a2e/e2a 和 detour 系列都使用 magic % PING_PONG_SIZE 实现双缓冲:

// a2e
shareAddrs[rank] = localWindowsIn + (magic % 2) * 800MB + rank * 512;

// detour
return (rankSize * srcRank + dstRank) * memLen + memRank * 512
    + (magic % 2) * (memLen / 2);

作用: 避免上一轮通信的数据被本轮覆盖。奇数轮使用内存前半，偶数轮使用后半。

8. 与现有算子链的关系

8.1 MoE EP 通信流水线全景

┌─────────────────────────────────────────────────────────────────────────┐
│                    MoE Expert Parallelism 前向流水线                      │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  Attention 卡簇 (rank ≥ expertRankSize)                                  │
│  ┌──────────────────────────────────────────────┐                       │
│  │  1. Attention 计算                            │                       │
│  │  2. Gate/Routing → expert_ids, scales         │                       │
│  │  3. a2e() ─────────────发送 token + 路由 ─────┼────► Expert 卡簇     │
│  │  4. e2a() ◄────────────接收 Expert 结果 ──────┼───── (rank < epSize) │
│  │  5. 后续 Attention 层                         │                       │
│  └──────────────────────────────────────────────┘                       │
│                                                                         │
│  Expert 卡簇 (rank < expertRankSize)                                    │
│  ┌──────────────────────────────────────────────┐                       │
│  │  3. a2e() ◄─────────────接收 token + 路由 ───┼───── Attention 卡簇   │
│  │  4. moe_dispatch_* (EP 内 AlltoAll)           │                       │
│  │  5. Expert FFN 计算                           │                       │
│  │  6. moe_combine_* (EP 内 AlltoAll)            │                       │
│  │  7. e2a() ─────────────发送结果 ─────────────┼────► Attention 卡簇   │
│  └──────────────────────────────────────────────┘                       │
│                                                                         │
│  跨 Pod 场景 (rankSize > commRankSize)                                   │
│  ┌──────────────────────────────────────────────┐                       │
│  │  all_to_all_detour() — 经 detour rank 绕行    │                       │
│  │  reduce_scatter_detour() — 经 detour rank 归约 │                       │
│  └──────────────────────────────────────────────┘                       │
└─────────────────────────────────────────────────────────────────────────┘

8.2 新算子与现有算子的协作关系

现有算子 (master)	新算子 (cam_feature)	关系
`moe_dispatch_normal_zb`	a2e	a2e 是 dispatch 的前置跨簇传输，将 token 从 Attention 卡送到 Expert 卡后，再由 moe_dispatch 做 EP 内分发
`moe_combine_normal_zb`	e2a	e2a 是 combine 的后置跨簇传输，Expert 计算结果经 moe_combine 聚合后，由 e2a 送回 Attention 卡
`moe_dispatch_layout_zb`	a2e	layout 算子管理 token-to-expert 映射，a2e 负责按映射传输
HCCL AlltoAll	all_to_all_detour	detour 是 AlltoAll 在 non-fullmesh 拓扑下的替代实现
HCCL ReduceScatter	reduce_scatter_detour	detour 是 ReduceScatter 在 non-fullmesh 拓扑下的替代实现
`fused_deep_moe`	a2e/e2a	fused_deep_moe 融合 EP 内 dispatch+FFN+combine，a2e/e2a 处理跨簇部分

8.3 端到端数据流 (以 sample 为例)

# Attention 侧 (rank >= expert_rank_size)
# 1. 发送阶段
a2e_output = a2e(x, expert_ids, scales, ...)
#    → expand_x (空), simulate_expert_ids, simulate_expert_scales, atten_batch_size, x_active_mask
# 2. 接收阶段
e2a_output = e2a(expand_x=x, atten_batch_size=atten_batch_size, ...)
#    → x (Expert 计算结果)

# Expert 侧 (rank < expert_rank_size)
# 1. 接收阶段
a2e_output = a2e(dummy_x, dummy_ids, dummy_scales, ...)
#    → expand_x (真实 token), simulate_expert_ids, simulate_expert_scales, atten_batch_size, x_active_mask
# 2. 发送阶段
e2a_output = e2a(expand_x=expand_x, atten_batch_size=atten_batch_size, ...)
#    → x (空)

8.4 设计意图总结

解耦跨簇通信与 EP 内通信: a2e/e2a 处理 Attention↔Expert 跨簇传输，moe_dispatch/moe_combine 处理 EP 内传输，各司其职
SHMEM 直传替代 HCCL collective: 对于小数据量或特定拓扑，SHMEM 直传延迟更低
Detour 解决拓扑限制: 当 HCCL 原生 collective 无法适配非 fullmesh 拓扑时，detour 提供软件可控的绕行方案
精度提升: reduce_scatter_detour 支持 fp16 输入 + fp32 累加，避免低精度归约的精度损失
平台差异化: a2e/e2a 面向 910_93 (AI 集群)，detour 系列同时支持 910_93 和 910B (推理/混合场景)

附录 A: 关键常量汇总

常量	值	来源	含义
`CAM_MAX_RANK_SIZE`	384	comm_args.h	最大支持 rank 数
`IPC_BUFF_MAX_SIZE`	100MB	comm_args.h	IPC 数据区大小
`IPC_BUFF_MAX_SIZE_MUL_EXP`	800MB	a2e.h	a2e 扩展 IPC 区
`IPC_DATA_OFFSET`	2MB	comm_args.h	flag 区大小
`PING_PONG_SIZE`	2	comm_args.h	双缓冲
`OPT_RANK_OFFSET`	512	a2e.h	rank 间内存偏移
`MAX_CORE_NUM`	48	all2all_small_fullmesh_with_detour.h	最大 AIV 核数
`FLAG_UNIT_INT_NUM`	4	sync_collectives.h	flag 单元 int64 数
`MAGIC_OFFSET`	32	sync_collectives.h	magic 高位偏移
`UB_SINGLE_DMA_SIZE_MAX`	190KB	comm_args.h	单次 DMA 最大量
`UB_SINGLE_TOTAL_SIZE_MAX`	192KB	comm_args.h	UB 总量

附录 B: 文件索引

src/cam/comm_operator/ascend_kernels/
├── a2e/
│   ├── op_kernel/a2e.cpp          # 设备入口
│   ├── op_kernel/a2e.h            # A2e 模板类（核心实现）
│   ├── op_kernel/a2e_tiling.h     # Tiling 数据结构
│   ├── op_host/a2e.cpp            # Tiling 函数 + 算子定义
│   └── op_api/aclnn_a2e.cpp/.h    # aclnn API 封装
├── e2a/
│   ├── op_kernel/e2a.cpp          # 设备入口
│   ├── op_kernel/e2a.h            # E2a 模板类（核心实现）
│   ├── op_kernel/e2a_tiling.h     # Tiling 数据结构
│   ├── op_host/e2a.cpp            # Tiling 函数 + 算子定义
│   └── op_api/aclnn_e2a.cpp/.h    # aclnn API 封装
├── all_to_all_detour/
│   ├── op_kernel/all2_all_detour.cpp                    # 设备入口
│   ├── op_kernel/all2all_small_fullmesh_with_detour.h   # 核心算法
│   ├── op_host/all2_all_detour.cpp                      # Tiling 函数 + 算子定义
│   ├── op_host/all2_all_detour_tiling.h                 # Tiling 数据结构
│   └── op_api/aclnn_all2_all_detour.cpp/.h              # aclnn API 封装
├── reduce_scatter_detour/
│   ├── op_kernel/reduce_scatter_detour.cpp              # 设备入口
│   ├── op_kernel/reduce_scatter_detour.h                # 核心算法
│   ├── op_kernel/reduce_scatter_detour_tiling.h         # Tiling 数据结构
│   ├── op_host/reduce_scatter_detour.cpp                # Tiling 函数 + 算子定义
│   ├── op_host/reduce_scatter_detour_tiling.cpp         # Tiling 实现
│   └── op_api/aclnn_reduce_scatter_detour.cpp/.h        # aclnn API 封装
├── utils/op_kernel/
│   ├── collectives.h              # Detour 基类（CpGM2GMPingPong, InitByCamCommDetour）
│   ├── sync_collectives.h         # Flag 同步原语
│   ├── comm_args.h                # CommArgs 结构体 + 全局常量
│   ├── moe_distribute_base.h      # HCCL 拓扑/资源结构体
│   └── data_copy.h                # GM-GM/GM-UB 拷贝辅助
├── pybind/
│   ├── a2e.cpp                    # PyTorch 算子注册
│   └── e2a.cpp                    # PyTorch 算子注册
└── operator_registry.json         # 算子注册表

附录 C: 关键函数调用链

a2e 调用链

a2e() [op_kernel/a2e.cpp]
 └─ A2e<bf16,bf16,false>::init()
     ├─ GetHcclContext<HCCL_GROUP_ID_0>()     // 获取 IPC 窗口
     ├─ AtomicAdd → magicTensor_              // 分布式 magic 分配
     └─ shareAddrs[] 计算                      // 每 rank IPC 地址映射
 └─ A2e::process()
     ├─ [Attention 卡] sendWithMte()
     │   ├─ sendX()        → copyGmToGmWithBlocks + flag
     │   ├─ sendExpertIds() → CpGM2GMPingPong + flag
     │   └─ sendExpertScales() → CpGM2GMPingPong + flag
     └─ [Expert 卡] recvWithMte()
         ├─ waitFlagWithScalar() → DATA_FULSH 轮询
         ├─ recvExpertIdsWithMte() → camCpGM2UB + camCpUB2GM
         └─ copyGmToGmWithBlocks() → expandX

all_to_all_detour 调用链

all2_all_detour() [op_kernel/all2_all_detour.cpp]
 └─ All2AllSmallFullmeshWithDetour<T>::Init()
     ├─ IsValidCommRank()                      // 检查是否参与通信
     └─ Collectives::InitByCamCommDetour()     // 初始化 peerMems 映射
 └─ All2AllSmallFullmeshWithDetour<T>::Process()
     ├─ [blockIdx < coreNumPerStageX] InputToShareSlice()
     │   ├─ blockIdx==0: 本地直拷
     │   ├─ blockIdx==1: 首块 detour 数据 → 写 flag
     │   └─ blockIdx≥2: 后续块 detour 数据 → 写 flag
     └─ [blockIdx < coreNumPerStageY] ShareToShareSlice()
         ├─ blockIdx==0: 无操作
         ├─ blockIdx==1: WaitSyncFlag → 收首块 → output
         └─ blockIdx≥2: WaitSyncFlag → 收后续块 → output

reduce_scatter_detour 调用链

reduce_scatter_detour() [op_kernel/reduce_scatter_detour.cpp]
 └─ ReduceScatterWithDetour<fp16,fp32>::Init()
     ├─ IsValidCommRank()
     ├─ Collectives::InitByCamCommDetour()
     └─ 计算 in2IpcCoreNum / reduceCoreNum / blockNum
 └─ ReduceScatterWithDetour<T,U>::Process()
     ├─ [blockIdx < in2IpcCoreNum] Input2Ipc()
     │   ├─ blockIdx==0: 本地数据 → 共享内存 + flag
     │   └─ blockIdx≥1: 分片 → detour rank 共享内存 + flag
     └─ [blockIdx < reduceCoreNum] Reduce()
         ├─ blockIdx==0: 无操作
         └─ blockIdx≥1: 流水线归约
             ├─ WaitSyncFlag(本地/远端数据就绪)
             ├─ CpGM2GMPingPong(..., atomOp)  // 归约
             ├─ SetOuterFlagDetour(通知下一级)
             └─ WaitOuterFlagDetour(最后一级) → Ipc2Out

报告完

← FusedDeepMoE 三变体 📑 索引 ZB 工具 + 编译系统 + HFA →

新通信算子 — A2E/E2A/Detour — UMDK/CAM