HCCL Window 模式 (Non-SHMEM) 代码深度剖析

日期: 2026-07-21 范围: CAM 库中 moe_dispatch_normal + moe_combine_normal 的完整 3 层栈 (op_api → op_host → op_kernel) 模式: 生产默认模式 (requires_shmem: false)，所有算子当前均使用此路径

1. 整体架构概览

HCCL Window 模式是 CAM 库的生产默认通信路径。与 SHMEM 模式（基于 IPC 共享内存 + 自管理 flag 区）不同，Window 模式直接利用 HCCL 框架分配的 Window 内存窗口（localWindowsIn / localWindowsExp + 远端 windowsIn / windowsExp）进行跨片数据通信。

核心数据结构关系

HcclOpResParam (per-rank, 来自 HCCL 上下文)
├── localWindowsIn     → 本卡 Window 输入区 (combine 读取)
├── localWindowsOut    → 本卡 Window 输出区 (预留)
├── localWindowsExp    → 本卡 Window 扩展区 (dispatch 状态同步)
├── remoteRes[]        → 远端资源指针数组
│   └── HcclRankRelationResV2
│       ├── windowsIn  → 远端 Window 输入区 (本卡 DMA 写入)
│       ├── windowsOut → 远端 Window 输出区
│       └── windowsExp → 远端 Window 扩展区 (flag 同步)
└── winSize            → Window 总大小 (由 HCCL_BUFFSIZE 环境变量控制)

通信模式对比

维度	HCCL Window 模式 (Non-SHMEM)	SHMEM 模式
内存来源	HCCL 框架预分配 Window	IPC 共享内存 + 自管理 flag 区
同步机制	Magic+Value flag (高32位 magic + 低32位 value)	自管理 flag + chunk 协议
双缓冲	Ping-Pong (dataState 0/1)	同样 Ping-Pong
数据区	`localWindowsIn` / `remoteWindowsIn`	IPC data 区
Flag 区	`localWindowsExp` / `remoteWindowsExp`	IPC flag 区 (前 2MB)
当前状态	所有算子默认	预留/实验性

2. Window 内存布局

Window 内存由 HCCL 框架通过 HCCL_BUFFSIZE 环境变量分配（默认 200MB）。CAM 将其划分为多个功能区域：

2.1 整体布局 (单卡视角)

Window Base (localWindowsIn / remoteWindowsIn 指向)
│
├─ [0, 204MB)                    NOTIFY_DISPATCH_BUFF_OFFSET
│   └── notify_dispatch 算子使用的缓冲区
│
├─ [204MB, 204MB + dataArea)     COMBINE_STATE_WIN_OFFSET = 4MB 之后
│   └── combine 数据区 (token 数据)
│       每个 token 占 h512AlignRecvXLen = Ceil(h*2, 512) * 512
│
├─ ... (其他区域)
│
└─ 总计由 totalWinSize 控制

Window Exp Base (localWindowsExp / remoteWindowsExp 指向)
│
├─ [0, 500KB * dataState)        WIN_STATE_OFFSET = 500KB
│   └── 预留区域
│
├─ [500KB, 950KB)                STATE_WIN_OFFSET = 950KB
│   └── 本卡 dataState 状态区 (每 core 512B 对齐)
│
├─ [950KB, 950KB + N*stateOffset)  dispatch flag 区
│   └── 每个 expert 占 stateOffset=32B
│   └── 写入目标: 远端 windowsExp 对应位置
│
├─ [975KB, 975KB + N*512B)       MAGIC_WIN_OFFSET = 975KB
│   └── combine 用 Magic 双缓冲标志 (每 core 512B)
│
└─ 总计由 winExpSize 控制

2.2 关键偏移量定义

// comm_args.h (Moe namespace)
constexpr uint64_t NOTIFY_DISPATCH_BUFF_OFFSET = 204UL * 1024UL * 1024UL;  // 204MB
constexpr int64_t IPC_BUFF_MAX_SIZE = 100 * 1024 * 1024;                   // 100MB
constexpr int64_t IPC_DATA_OFFSET = 2 * 1024 * 1024;                       // 2MB flag + 100MB data

// moe_dispatch_normal.h
constexpr uint64_t WIN_STATE_OFFSET = 500UL * 1024UL;          // 500KB
constexpr uint64_t STATE_WIN_OFFSET = 950UL * 1024UL;          // 950KB
constexpr uint64_t WIN_ADDR_ALIGN = 512UL;                     // 512B 对齐
constexpr uint64_t COMBINE_STATE_WIN_OFFSET = 4UL * 1024UL * 1024UL;  // 4MB

// moe_combine_normal.h
constexpr uint64_t MAGIC_WIN_OFFSET = 975UL * 1024UL;          // 975KB
constexpr uint32_t STATE_OFFSET = 32U;                         // 32B per expert

2.3 数据区 Token 格式 (Dispatch → Combine)

每个 token 在 Window 数据区中的布局：

hOutGMAlignSize = Ceil(h*2 + 44, 512) * 512

┌─────────────────────────────────────────────────────────┐
│  token data (h * sizeof(ExpandXOutType))                │  h*2 bytes (BF16)
│  e.g., h=7168 → 14336 bytes                            │
├─────────────────────────────────────────────────────────┤
│  padding to 32B alignment                              │
├─────────────────────────────────────────────────────────┤
│  dynamicScale (float32) - 仅 DynamicQuant 模式         │  4 bytes
├─────────────────────────────────────────────────────────┤
│  expandIdx[0] = epRankId (int32)                       │  4 bytes
│  expandIdx[1] = tokenIndex (int32)                     │  4 bytes
│  expandIdx[2] = topkIdx (int32)                        │  4 bytes
├─────────────────────────────────────────────────────────┤
│  padding to 512B alignment                             │
└─────────────────────────────────────────────────────────┘
总计 hOutGMAlignSize (通常 14848 → 15360 for h=7168)

3. moe_dispatch_normal 完整执行流程

3.1 文件结构

文件	职责
`op_api/aclnn_moe_dispatch_normal.cpp`	aclnn 接口层，调用 aclnnInner
`op_host/moe_dispatch_normal.cpp`	算子定义 (OpDef)，注册输入/输出/属性
`op_host/moe_dispatch_normal_tiling.cpp`	Tiling 计算，校验参数，计算 Window 大小
`op_kernel/moe_dispatch_normal.cpp`	内核入口，根据 tilingKey 选择模板
`op_kernel/moe_dispatch_normal.h`	核心：完整 kernel 实现

3.2 模板参数

template <typename XType, typename ExpandXOutType, bool DynamicQuant,
          bool IsSmoothScaleExist, bool IsShareExpertRank>
class MoeDispatchNormal;

XType: 输入数据类型 (BF16/FP16)
ExpandXOutType: 输出数据类型 (BF16/FP16/INT8)
DynamicQuant: 是否动态量化 (INT8 输出)
IsSmoothScaleExist: 平滑 scale (未使用)
IsShareExpertRank: 共享 expert rank (未使用)

3.3 TilingKey 路由

// moe_dispatch_normal.cpp (kernel entry)
#define TILINGKEY_NO_QUANT 10000
#define TILINGKEY_QUANT    10002

#if (ORIG_DTYPE_RECV_X == DT_BF16 || ORIG_DTYPE_RECV_X == DT_FLOAT16)
    if (TILING_KEY_IS(TILINGKEY_NO_QUANT)) {
        MoeDispatchNormal<DTYPE_X, DTYPE_RECV_X, false, false, false> op;
        ...
    }
#elif (ORIG_DTYPE_RECV_X == DT_INT8)
    if (TILING_KEY_IS(TILINGKEY_QUANT)) {
        MoeDispatchNormal<DTYPE_X, DTYPE_RECV_X, true, false, false> op;
        ...
    }
#endif

3.4 Init 阶段详解

__aicore__ inline void MoeDispatchNormal<TemplateTypeFunc>::Init(...)
{
    // 1. 获取 HCCL 上下文 (两个通信域: EP + TP)
    winContext_[COMM_EP_IDX] = GetHcclContext<HCCL_GROUP_ID_0>();  // EP 域
    winContext_[COMM_TP_IDX] = GetHcclContext<1>();                 // TP 域

    // 2. 读取本卡 dataState (Ping-Pong 状态)
    GM_ADDR statusDataSpaceGm = winContext_[COMM_EP_IDX]->localWindowsExp;
    selfDataStatusTensor.SetGlobalBuffer(
        statusDataSpaceGm + STATE_WIN_OFFSET + blockIdx * WIN_ADDR_ALIGN);

    // 3. 获取当前 dataState 并翻转 (0→1, 1→0)
    DataCacheCleanAndInvalid(selfDataStatusTensor);  // 缓存失效，从内存读取
    dataState = selfDataStatusTensor(0);
    selfDataStatusTensor(0) = (dataState == 0) ? 1 : 0;
    DataCacheCleanAndInvalid(selfDataStatusTensor);  // 写回
    PipeBarrier<PIPE_ALL>();

    // 4. 计算 winDataSizeOffset (Ping-Pong 偏移)
    winDataSizeOffset = dataState * (totalWinSize / 2)
                        + globalBatchSize / epRankSize * topK * hSizeAlignCombine;

    // 5. 计算 shareGM 地址 (本卡 Window 数据区基址)
    shareGM = GetWindAddrByRankId(COMM_EP_IDX, epRankId);

    // 6. 初始化 Buffer
    hOutUBAlignSize = Ceil(hScaleIdxSize, UB_ALIGN) * UB_ALIGN;
    tpipe_->InitBuffer(xQueue, BUFFER_NUM, hOutUBAlignSize);  // 2 * 14K = 28K
}

3.5 Process 主流程

__aicore__ inline void Process()
{
    if ASCEND_IS_AIV {
        InputToShare();    // 阶段1: 数据从 GM → Window
        SetStatus();       // 阶段2: 通知远端 "数据已到"
        WaitStatus();      // 阶段3: 等待远端 "数据已到"
        ShareToOutput();   // 阶段4: 数据从 Window → GM 输出
    }
}

3.6 InputToShare — 数据写入 Window

对每个 token (batchSize * topK 个):
  1. 读取 expertId 和 tokenIdx
  2. 计算目标地址: shareGM + hOutGMAlignSize * (dstExpertOffset + curExpertCnt)
  3. 从 GM 读取 token 到 UB (DataCopyPad)
  4. [可选] DynamicQuant: BF16 → FP32 → Abs → ReduceMax → Scale → INT8
  5. FillTriple: 在 token 后附加 (epRankId, tokenIndex, topkIdx)
  6. DataCopyPad: UB → Window (GM)

关键代码路径 (非量化):

for (int32_t tokenIndex = startTokenId; tokenIndex < endTokenId; ++tokenIndex) {
    uint32_t dstExpertId = expertIdsTensor(tokenIndex - startTokenId);
    int32_t curExpertCnt = sendTokenIdxTensor(tokenIndex - startTokenId);
    int32_t dstExpertOffset = sendOffsetTensor(dstExpertId);

    // 计算 Window 目标地址
    GM_ADDR rankGM = shareGM + hOutGMAlignSize * (dstExpertOffset + curExpertCnt);
    dstGT.SetGlobalBuffer((__gm__ ExpandXOutType *)rankGM);

    // GM → UB → Window
    xTmpTensor = xQueue.AllocTensor<ExpandXOutType>();
    DataCopyPad(xTmpTensor, xGT[tokenIndex / topK * h], xCopyParams, tokenCopyPadExtParams);
    xQueue.EnQue(xTmpTensor);
    xTmpTensor = xQueue.DeQue<ExpandXOutType>();

    // 附加三元组 (srcRank, srcTokenIdx, topkIdx)
    FillTriple(xTmpTensor, tokenIndex / topK, tokenIndex % topK);

    // UB → Window (DMA 写入远端或本卡)
    DataCopyPad(dstGT, xTmpTensor, hCommuCopyOutParams);
    xQueue.FreeTensor(xTmpTensor);
}

3.7 SetStatus — 通知远端数据已到

__aicore__ inline void SetStatus()
{
    // 1. 准备 flag 数据 (8 个 float = 1.0f, 即 0x3F800000)
    Duplicate<int32_t>(statusTensor, 0, expNumPerCore * 8);
    Duplicate<int32_t>(statusTensor, 0x3F800000, mask, statusCntAlign / 8, 1, 8);
    // 每个 expert 占 8 个 int32 (32B), 全部填充为 1.0f 的 IEEE754 表示

    SyncAll<true>();  // 全卡同步

    // 2. 将 flag 写入远端 windowsExp 的对应位置
    for (uint32_t i = startExpId; i < endExpId; ++i) {
        uint32_t targetRankId = i / moeExpertNumPerRank;
        uint32_t offset = stateOffset * (epRankId + i % moeExpertNumPerRank * epRankSize);
        GM_ADDR rankGM = GetWindStateAddrByRankId(COMM_EP_IDX, targetRankId) + offset;
        dstStatusGT.SetGlobalBuffer(rankGM);
        DataCopy(dstStatusGT, statusTensor[(i - startExpId) * 8], 8UL);
    }
}

Flag 区布局 (远端 windowsExp):

远端 windowsExp + dataState * WIN_STATE_OFFSET
│
├─ expert 0: [rank0_flag(32B), rank1_flag(32B), ..., rankN_flag(32B)]
├─ expert 1: [rank0_flag(32B), rank1_flag(32B), ..., rankN_flag(32B)]
├─ ...
└─ expert M-1: [...]

每个 flag 占 32B (8 个 float32, 全部为 1.0f)

3.8 WaitStatus — 等待远端数据到达

__aicore__ inline void WaitStatus()
{
    // 1. 准备: 从本卡 windowsExp 读取 flag 区
    GlobalTensor<float> windowInstatusFp32Tensor;
    windowInstatusFp32Tensor.SetGlobalBuffer(
        GetWindStateAddrByRankId(COMM_EP_IDX, epRankId));

    float compareTarget = 1.0f * statusNumPerCore;  // 期望的 flag 总和

    // 2. 轮询直到所有 flag 就绪
    while (sumOfFlag != compareTarget) {
        // 从 GM 读取 flag 到 UB
        DataCopy(statusFp32Tensor,
                 windowInstatusFp32Tensor[startStatusId * stateOffset / sizeof(float)],
                 intriParams);
        SyncFunc<MTE2_V>();

        // ReduceSum: 对所有 flag 求和
        ReduceSum(statusSumOutTensor, statusFp32Tensor, gatherMaskOutTensor,
                  mask, statusNumPerCore, 1);
        SyncFunc<V_S>();
        sumOfFlag = statusSumOutTensor.GetValue(0);
    }

    // 3. 清除 flag (写回 0)
    Duplicate<int32_t>(cleanStateTensor, 0, duplicateMask, Ceil(statusNumPerCore, 8), 1, 8);
    DataCopy(windowInstatusFp32Tensor[...], cleanStateTensor.ReinterpretCast<float>(), intriOutParams);
    SyncAll<true>();
}

轮询优化: 使用 ReduceSum 一次性对所有 flag 求和，与 compareTarget 比较。只有当所有 source rank 都写入了 flag 时才退出。

3.9 ShareToOutput — 从 Window 读取到输出

__aicore__ inline void ShareToOutput()
{
    for (uint32_t i = startStatusId; i < endStatusId; ++i) {
        fromRank = i % epRankSize;
        count = recvCountTensor(i) - preCount;
        recvOffset = recvOffsetTensor(i);

        // 计算源 Window 地址
        GM_ADDR recvStart = GetWindAddrByRankId(COMM_EP_IDX, fromRank)
                            + recvOffset * hOutGMAlignSize;

        for (uint32_t j = 0; j < count; ++j) {
            srcTokenGT.SetGlobalBuffer(recvStart + j * hOutGMAlignSize);

            // Window → UB
            xTmpTensor = xQueue.AllocTensor<ExpandXOutType>();
            DataCopyPad(xTmpTensor, srcTokenGT, hCommuCopyOutParams, copyPadExtParams);
            xQueue.EnQue(xTmpTensor);
            xTmpTensor = xQueue.DeQue<ExpandXOutType>();

            // 提取三元组 → expandIdxOut
            DataCopyPad(expandIdxOutGT[(targetOffset + j) * EXPAND_IDX_INFO],
                        xTmpTensorInt[expandIdxStartIdx], dataCopyExandIdxParams);

            // 提取 token data → expandXOut
            dstTokenGT.SetGlobalBuffer(expandXOutGM + (targetOffset + j) * h);
            DataCopyPad(dstTokenGT, xTmpTensor, expandXCopyParams);
            xQueue.FreeTensor(xTmpTensor);
        }
    }
}

4. Window 地址计算

4.1 GetWindAddrByRankId — 数据区地址

__aicore__ inline GM_ADDR GetWindAddrByRankId(uint8_t ctxIdx, const int32_t rankId)
{
    uint32_t curRankId = (ctxIdx == COMM_EP_IDX) ? epRankId : tpRankId;

    if (curRankId == rankId) {
        // 本卡: 使用 localWindowsIn
        return (GM_ADDR)(winContext_[ctxIdx]->localWindowsIn)
               + winDataSizeOffset + COMBINE_STATE_WIN_OFFSET
               + NOTIFY_DISPATCH_BUFF_OFFSET;
    }

    // 远端: 使用 remoteRes 中的 windowsIn
    return (GM_ADDR)(((HcclRankRelationResV2 *)
               (winContext_[ctxIdx]->remoteRes[rankId].nextDevicePtr))->windowsIn)
               + winDataSizeOffset + COMBINE_STATE_WIN_OFFSET
               + NOTIFY_DISPATCH_BUFF_OFFSET;
}

地址公式:

本卡: localWindowsIn + winDataSizeOffset + 4MB + 204MB
远端: remoteRes[rankId].windowsIn + winDataSizeOffset + 4MB + 204MB

4.2 GetWindStateAddrByRankId — 状态区地址

__aicore__ inline GM_ADDR GetWindStateAddrByRankId(uint8_t ctxIdx, const int32_t rankId)
{
    uint32_t curRankId = ctxIdx == COMM_EP_IDX ? epRankId : tpRankId;

    if (curRankId == rankId) {
        // 本卡: 使用 localWindowsExp
        return (GM_ADDR)(winContext_[ctxIdx]->localWindowsExp)
               + dataState * WIN_STATE_OFFSET;
    }

    // 远端: 使用 remoteRes 中的 windowsExp
    return (GM_ADDR)(((HcclRankRelationResV2 *)
               (winContext_[ctxIdx]->remoteRes[rankId].nextDevicePtr))->windowsExp)
               + dataState * WIN_STATE_OFFSET;
}

地址公式:

本卡: localWindowsExp + dataState * 500KB
远端: remoteRes[rankId].windowsExp + dataState * 500KB

4.3 winDataSizeOffset — Ping-Pong 偏移计算

uint64_t hSizeAlignCombine = Ceil(h * sizeof(XType), WIN_ADDR_ALIGN) * WIN_ADDR_ALIGN;
winDataSizeOffset = dataState * (totalWinSize / 2)
                    + globalBatchSize / epRankSize * topK * hSizeAlignCombine;

含义: - dataState * (totalWinSize / 2): Ping-Pong 双缓冲，dataState=0 用前半，dataState=1 用后半 - globalBatchSize / epRankSize * topK * hSizeAlignCombine: 跳过 combine 数据区 (combine 从前往后写，dispatch 从后往前写)

4.4 地址计算全景图

                    ┌──────────────────────────────────────────────┐
                    │              totalWinSize                     │
                    │         (由 HCCL_BUFFSIZE 控制)               │
                    ├──────────────────────────────────────────────┤
                    │  [0, totalWinSize/2)     ← dataState=0      │
                    │  [totalWinSize/2, end)   ← dataState=1      │
                    └──────────────────────────────────────────────┘

dataState=0 时:
  winDataSizeOffset = 0 + bs/epRankSize * topK * hAlign

  Dispatch 数据区起始:
    localWindowsIn + 0 + 4MB + 204MB + bs/epRankSize*topK*hAlign

  Combine 数据区起始:
    localWindowsIn + 0 + 4MB + 204MB

5. Ping-Pong 双缓冲机制

5.1 设计思想

Ping-Pong 双缓冲允许 dispatch 写下一批数据 与 combine 读上一批数据 并行执行，实现通信-计算重叠。

5.2 dataState 翻转逻辑

// Init 阶段
DataCacheCleanAndInvalid(selfDataStatusTensor);  // 缓存失效
dataState = selfDataStatusTensor(0);             // 读取当前状态
selfDataStatusTensor(0) = (dataState == 0) ? 1 : 0;  // 翻转
DataCacheCleanAndInvalid(selfDataStatusTensor);  // 写回

注意: DataCacheCleanAndInvalid 在这里用于确保跨 AIV core 的可见性。所有 core 都读取同一个 selfDataStatusTensor，但只有 blockIdx=0 的 core 实际参与翻转（因为偏移是 blockIdx * WIN_ADDR_ALIGN）。

5.3 数据区划分

Window 数据区 (localWindowsIn + 204MB + 4MB 起始)
│
├─ [0, combineAreaSize)          Combine 数据区 (combine 写，combine 读)
│   每个 token: h512AlignRecvXLen
│
├─ [combineAreaSize, end)        Dispatch 数据区 (dispatch 写，dispatch 读)
│   每个 token: hOutGMAlignSize
│
└─ 总计: totalWinSize / 2 (Ping-Pong 各占一半)

5.4 时序图

时间 ──────────────────────────────────────────────────────────→

Dispatch N-1:
  InputToShare → SetStatus
                 ↓ (flag 写入远端 windowsExp)
Combine N-1:                          ↓
  WaitBuffCopy (轮询 flag) ←──────────┘
  ReadBufferAndWeightedSum

Dispatch N:
  InputToShare (写 dataState=1 区域)  ← 与 Combine N-1 并行!
  SetStatus
                 ↓
Combine N:
  WaitBuffCopy (轮询 flag) ←──────────┘
  ReadBufferAndWeightedSum (读 dataState=0 区域)

6. Magic+Value 同步协议

6.1 编码方式

// sync_collectives.h
constexpr int64_t MAGIC_OFFSET = 32;
constexpr int64_t MAGIC_MASK = ~((1LL << MAGIC_OFFSET) - 1);

__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);
}

64位编码:

┌─────────────────────────────┬─────────────────────────────┐
│      High 32 bits           │       Low 32 bits           │
│      magic (uint32)         │       value (uint32)        │
│   (防 ABA，批次标识)         │       (事件/状态码)          │
└─────────────────────────────┴─────────────────────────────┘

6.2 Magic 翻转机制 (Combine 阶段)

// moe_combine_normal.h — InitMagic()
__aicore__ inline void InitMagic()
{
    // 读取本卡 Magic 标志
    selfMagicTensor.SetGlobalBuffer(
        epWinContext_->localWindowsExp + MAGIC_WIN_OFFSET + coreIdx_ * WIN_512_ALIGN);

    DataCacheCleanAndInvalid(selfMagicTensor);
    magic_ = selfMagicTensor(0);           // 读取当前 magic
    selfMagicTensor(0) = (magic_ == 0) ? 1 : 0;  // 翻转
    DataCacheCleanAndInvalid(selfMagicTensor);
}

Magic 标志布局 (localWindowsExp + 975KB):

localWindowsExp + MAGIC_WIN_OFFSET
│
├─ core 0: [512B]  magic flag (int32, 0 或 1)
├─ core 1: [512B]  magic flag
├─ ...
└─ core N: [512B]  magic flag

6.3 Flag 等待逻辑

// sync_collectives.h — WaitOneRankPartFlag
__aicore__ inline void WaitOneRankPartFlag(__gm__ int64_t* waitAddr,
                                            int64_t flagNum, int64_t checkValue)
{
    do {
        // GM → UB: 读取 flag
        DataCopy(localWait, globalWait[checkedFlagNum * FLAG_UNIT_INT_NUM],
                 (flagNum - checkedFlagNum) * FLAG_UNIT_INT_NUM);
        SetFlag<MTE2_S>(EVENT_ID0);
        WaitFlag<MTE2_S>(EVENT_ID0);

        // 检查每个 flag
        isSync = true;
        for (auto i = 0; i < remainToCheck; ++i) {
            int64_t v = localWait.GetValue(i * FLAG_UNIT_INT_NUM);
            // 检查: magic 位相等 且 value >= checkValue
            if ((v & MAGIC_MASK) != (checkValue & MAGIC_MASK) || v < checkValue) {
                isSync = false;
                checkedFlagNum += i;  // 跳过已确认的 flag
                break;
            }
        }
    } while (!isSync);
}

关键判断条件:

(v & MAGIC_MASK) != (checkValue & MAGIC_MASK) || v < checkValue

MAGIC_MASK = 0xFFFFFFFF00000000 (高32位)
先检查 magic 位是否匹配（防止读到旧一批的 flag）
再检查 value 是否 >= 目标值（支持乱序到达）

6.4 Dispatch 阶段的 Flag 编码

Dispatch 阶段不使用 SyncCollectives 类，而是直接用 float32 = 1.0f 作为 flag:

// SetStatus: 写入 1.0f (IEEE754 = 0x3F800000)
Duplicate<int32_t>(statusTensor, 0x3F800000, mask, statusCntAlign / 8, 1, 8);

// WaitStatus: 轮询求和
float compareTarget = 1.0f * statusNumPerCore;
while (sumOfFlag != compareTarget) {
    DataCopy(statusFp32Tensor, windowInstatusFp32Tensor[...]);
    ReduceSum(statusSumOutTensor, statusFp32Tensor, ...);
    sumOfFlag = statusSumOutTensor.GetValue(0);
}

注意: Dispatch 阶段的 flag 不使用 Magic+Value 编码，而是直接用 float 求和。这是因为 dispatch 的 flag 区在 localWindowsExp + dataState * 500KB，已经通过 dataState 实现了 Ping-Pong 隔离。

6.5 Combine 阶段的 Flag 编码

Combine 阶段使用 float32 = 1.0f 作为 flag (与 dispatch 类似):

// CopyBufferToShareAndSetStatus
Duplicate<uint32_t>(statusTensor, 0x3F800000, FLOAT_NUM_PER_ALIGN);
// 写入远端 windowsIn 对应位置

// WaitBuffCopy
float target = 1.0f * axisK_ * FLOAT_NUM_PER_ALIGN;
while (current != target) {
    DataCopy(stateTensorLocal, stateGMTensor, calCount);
    Sum(tempStateTensorLocal, stateTensorLocal, sumPerKParams);
    current = tempStateTensorLocal(0);
}

7. moe_combine_normal 完整执行流程

7.1 文件结构

文件	职责
`op_api/aclnn_moe_combine_normal.cpp`	aclnn 接口层
`op_host/moe_combine_normal.cpp`	算子定义
`op_host/moe_combine_normal_tiling.cpp`	Tiling 计算
`op_kernel/moe_combine_normal.cpp`	内核入口
`op_kernel/moe_combine_normal.h`	核心：完整 kernel 实现

7.2 模板参数

template <typename RecvXType, typename XType, typename SrcInfoType>
class MoeCombineNormal;

RecvXType: 接收数据类型 (BF16/FP16)
XType: 输出数据类型 (BF16/FP16)
SrcInfoType: 源信息类型 (int32)

7.3 Init 阶段

__aicore__ inline void Init(...)
{
    coreIdx_ = GetBlockIdx();

    // 1. 初始化 Magic (Ping-Pong)
    InitMagic();

    // 2. 初始化 Tiling 数据
    InitTilingData(tilingData);

    // 3. 初始化 Global Buffer
    InitGlobalBuffer(recvX, tokenSrcInfo, epRecvCount, topkWeights, XOut, sendCostStatsOut);

    // 4. 计算 Buffer 长度
    InitBuffLen();

    // 5. 计算 winDataSizeOffset
    PipeBarrier<PIPE_ALL>();
    winDataSizeOffset_ = magic_ * (totalWinSize / 2UL);
    localRankGM_ = GetBufferAddrByRankId(epRankId_);

    // 6. 读取 selfSendCnt_ (本卡需要发送的 token 数)
    DataCacheCleanAndInvalid(epRecvCountGM_[moeExpertNum_ - 1]);
    selfSendCnt_ = epRecvCountGM_(moeExpertNum_ - 1);
}

7.4 Process 主流程

__aicore__ inline void Process()
{
    if ASCEND_IS_AIV {
        CopyBufferToShareAndSetStatus();  // 阶段1: 数据 GM → Window + 通知远端
        ReadBufferFromRemote();           // 阶段2: 等待数据 + 加权求和
    }
}

7.5 CopyBufferToShareAndSetStatus — 数据写入远端 Window

__aicore__ inline void CopyBufferToShareAndSetStatus()
{
    // 1. 按 core 分配 token
    SplitCoreCal(selfSendCnt_, perBlockSendNum, startTokenId, endTokenId);

    // 2. 准备 flag 数据
    Duplicate<uint32_t>(statusTensor, 0x3F800000, FLOAT_NUM_PER_ALIGN);

    // 3. 读取 srcInfo (三元组: srcRankId, srcTokenId, srcTopkId)
    DataCopyPad(srcInfoLocal, tokenSrcInfoGM_[startTokenId * TOKEN_SRC_INFO_LEN], ...);

    // 4. 对每个 token
    for (uint32_t tokenIndex = startTokenId; tokenIndex < endTokenId; tokenIndex++) {
        uint32_t srcRankId = srcInfoLocal(index + RANK_ID_OFFSET_IN_SRC_INFO);
        uint32_t srcTokenId = srcInfoLocal(index + TOKEN_IDX_OFFSET_IN_SRC_INFO);
        uint32_t srcTopkId = srcInfoLocal(index + TOPK_IDX_OFFSET_IN_SRC_INFO);

        // 4a. 复制数据到远端 Window
        CopyBufferToShare(srcRankId, srcTokenId, srcTopkId, tokenIndex);

        // 4b. 设置 flag
        SetStatusBySrcInfo(srcRankId, srcTokenId, srcTopkId);
    }
}

CopyBufferToShare 详解:

__aicore__ inline void CopyBufferToShare(uint32_t srcRankId, uint32_t srcTokenId,
                                          uint32_t srcTopkId, uint32_t tkIndex)
{
    uint32_t tokenOffset = tkIndex * axisH_;

    // 计算远端 Window 目标地址
    GM_ADDR dstGM = GetBufferAddrByRankId(srcRankId)
                    + (srcTokenId * axisK_ + srcTopkId) * h512AlignRecvXLen_;
    GlobalTensor<XType> dstWindow;
    dstWindow.SetGlobalBuffer(dstGM);

    // GM → UB → Window
    LocalTensor<RecvXType> localCopyTensor;
    localCopyTensor = localCopyQueue_.AllocTensor<RecvXType>();
    DataCopyPad(localCopyTensor, recvXGM_[tokenOffset], xOutCopyParams, copyPadExtParams);
    localCopyQueue_.EnQue(localCopyTensor);
    localCopyTensor = localCopyQueue_.DeQue<RecvXType>();
    DataCopyPad(dstWindow, localCopyTensor, xOutCopyParams);
    localCopyQueue_.FreeTensor(localCopyTensor);
}

SetStatusBySrcInfo 详解:

__aicore__ inline void SetStatusBySrcInfo(uint32_t srcRankId, uint32_t srcTokenId,
                                           uint32_t srcTopkId)
{
    LocalTensor<uint32_t> statusTensor = stateBuf_.AllocTensor<uint32_t>();

    // 计算远端 flag 地址
    GM_ADDR stateGM = GetStateAddrByRankId(srcRankId)
                      + (srcTokenId * axisK_ + srcTopkId) * UB_32_ALIGN;
    GlobalTensor<uint32_t> stateGMTensor;
    stateGMTensor.SetGlobalBuffer(stateGM);

    // UB → GM (写入 1.0f)
    DataCopy<uint32_t>(stateGMTensor, statusTensor, FLOAT_NUM_PER_ALIGN);
}

7.6 ReadBufferFromRemote — 等待数据 + 加权求和

__aicore__ inline void ReadBufferFromRemote()
{
    // 1. 按 core 分配 token
    SplitCoreCal(axisBS_, tokenPerBlock, startTokenIndex, endTokenIndex);

    // 2. 初始化 Buffer
    tpipe_->InitBuffer(xOutBuf_, h32AlignRecvXLen_);
    tpipe_->InitBuffer(tokenFloatBuf_, h32AlignFloatLen_);
    tpipe_->InitBuffer(weightedMulBuf_, h256AlignFloatLen_);
    tpipe_->InitBuffer(sumFloatBuf_, h32AlignFloatLen_);
    tpipe_->InitBuffer(weightedSumQueue_, DOUBLE_BUFFER, h32AlignRecvXLen_);
    tpipe_->InitBuffer(stateBuf_, axisK_ * UB_32_ALIGN);
    tpipe_->InitBuffer(tempStateBuf_, axisK_ * UB_32_ALIGN);
    tpipe_->InitBuffer(topkWeightsBuf_, tokenPerBlock * axisK_ * sizeof(float));

    // 3. 读取 topK weights
    DataCopyPad(topkWeightsLocal, topkWeightsGM_[startTokenIndex * axisK_], ...);

    // 4. 对每个 token
    for (uint32_t tokenIndex = startTokenIndex; tokenIndex < endTokenIndex; tokenIndex++) {
        WaitBuffCopy(tokenIndex);              // 等待所有 topK 数据到达
        ReadBufferAndWeightedSum(tokenIndex);  // 加权求和
    }
}

7.7 WaitBuffCopy — 等待 topK 数据全部到达

__aicore__ inline void WaitBuffCopy(uint32_t tokenIndex)
{
    uint32_t calCount = axisK_ * FLOAT_NUM_PER_ALIGN;

    // 本卡 Window 中该 token 对应的 flag 区
    GM_ADDR stateGM = GetStateAddrByRankId(epRankId_) + tokenIndex * axisK_ * UB_32_ALIGN;
    GlobalTensor<float> stateGMTensor;
    stateGMTensor.SetGlobalBuffer(stateGM);

    float target = 1.0f * axisK_ * FLOAT_NUM_PER_ALIGN;

    // 轮询求和
    while (current != target) {
        SyncFunc<S_MTE2>();
        DataCopy<float>(stateTensorLocal, stateGMTensor, calCount);
        SyncFunc<MTE2_V>();
        Sum(tempStateTensorLocal, stateTensorLocal, sumPerKParams);
        SyncFunc<V_S>();
        current = tempStateTensorLocal(0);
    }

    // 清除 flag
    SyncFunc<S_V>();
    Duplicate<float>(tempStateTensorLocal, 0.0f, calCount);
    SyncFunc<V_MTE3>();
    DataCopy<float>(stateGMTensor, tempStateTensorLocal, calCount);
}

7.8 ReadBufferAndWeightedSum — 加权求和

__aicore__ inline void ReadBufferAndWeightedSum(uint32_t tokenIndex, uint32_t startTokenIndex)
{
    LocalTensor<float> tokenFloatLocal = tokenFloatBuf_.Get<float>();
    LocalTensor<float> weightedMulBufLocal = weightedMulBuf_.Get<float>();
    LocalTensor<float> sumFloatBufLocal = sumFloatBuf_.Get<float>();
    LocalTensor<float> topkWeightsLocal = topkWeightsBuf_.Get<float>();

    // 初始化累加器 (FP32)
    Duplicate(sumFloatBufLocal, 0.0f, axisH_);

    for (uint32_t topkId = 0; topkId < axisK_; topkId++) {
        float scale = topkWeightsLocal.GetValue((tokenIndex - startTokenIndex) * axisK_ + topkId);

        // 从本卡 Window 读取 token
        GM_ADDR localTokenAddr = localRankGM_ + (tokenIndex * axisK_ + topkId) * h512AlignRecvXLen_;
        GlobalTensor<XType> localTokenTensor;
        localTokenTensor.SetGlobalBuffer(localTokenAddr);

        LocalTensor<XType> tmpToken = weightedSumQueue_.AllocTensor<XType>();
        DataCopyPad(tmpToken, localTokenTensor, xOutCopyParams, copyPadExtParams);
        weightedSumQueue_.EnQue(tmpToken);
        tmpToken = weightedSumQueue_.DeQue<XType>();

        // BF16 → FP32
        Cast(tokenFloatLocal, tmpToken, CAST_NONE, axisH_);
        PipeBarrier<PIPE_V>();

        // 乘 scale
        Muls(weightedMulBufLocal, tokenFloatLocal, scale, axisH_);
        PipeBarrier<PIPE_V>();

        // 累加
        Add(sumFloatBufLocal, sumFloatBufLocal, weightedMulBufLocal, axisH_);
        weightedSumQueue_.FreeTensor(tmpToken);
    }

    PipeBarrier<PIPE_V>();

    // FP32 → BF16 输出
    LocalTensor<XType> xOutLocal = xOutBuf_.Get<XType>();
    Cast(xOutLocal, sumFloatBufLocal, CAST_RINT, axisH_);
    SyncFunc<V_MTE3>();
    DataCopyPad(xOutGlobal_[tokenIndex * axisH_], xOutLocal, xOutCopyParams);
}

关键设计: 使用 FP32 累加 避免 BF16 累加误差。每个 token 先 Cast 到 FP32，乘 weight，累加，最后 Cast 回 BF16。

7.9 Combine 地址计算

// 状态区地址 (flag)
__aicore__ GM_ADDR GetStateAddrByRankId(const int32_t rankId)
{
    GM_ADDR bufferAddr;
    if (epRankId_ == rankId) {
        bufferAddr = epWinContext_->localWindowsIn;
    } else {
        bufferAddr = ((HcclRankRelationResV2 *)
                      epWinContext_->remoteRes[rankId].nextDevicePtr)->windowsIn;
    }
    return bufferAddr + winDataSizeOffset_ + NOTIFY_DISPATCH_BUFF_OFFSET;
}

// 数据区地址 (token)
__aicore__ GM_ADDR GetBufferAddrByRankId(const int32_t rankId)
{
    return GetStateAddrByRankId(rankId) + COMBINE_STATE_WIN_OFFSET;
}

地址公式:

状态区: windowsIn + winDataSizeOffset + 204MB
数据区: windowsIn + winDataSizeOffset + 204MB + 4MB

8. Tiling 策略

8.1 Tiling 数据结构

// moe_dispatch_normal_tiling.h
struct MoeDispatchNormalInfo {
    uint32_t epWorldSize;   // EP 通信域大小
    uint32_t tpWorldSize;   // TP 通信域大小
    uint32_t epRankId;      // 本卡 EP Rank ID
    uint32_t tpRankId;      // 本卡 TP Rank ID
    uint32_t moeExpertNum;  // MoE expert 总数
    uint32_t quantMode;     // 量化模式 (0=无, 2=动态)
    uint32_t globalBs;      // 全局 batch size
    uint32_t bs;            // 本地 batch size
    uint32_t k;             // topK
    uint32_t h;             // hidden size
    uint32_t aivNum;        // AIV core 数量
    bool isQuant;           // 是否量化
    bool isEnableDiagnose;  // 是否启用诊断
    uint64_t totalUbSize;   // UB 总大小
    uint64_t totalWinSize;  // Window 总大小
};

8.2 Window 大小校验

// moe_dispatch_normal_tiling.cpp
uint64_t maxWindowSize = Mc2TilingUtils::GetMaxWindowSize();  // 从 HCCL_BUFFSIZE 读取

// 计算每个 token 需要的空间
uint64_t tokenActualLen = Ceil(h * 2, 32) * 32 + 44;  // h*2 + scale(32B) + triple(12B)
uint64_t tokenNeedSizeDispatch = Ceil(tokenActualLen, 512) * 512;
uint64_t tokenNeedSizeCombine = Ceil(h * 2, 512) * 512;

// 总需求 (双缓冲)
uint64_t actualSize = (maxBs * k * (tokenNeedSizeCombine + tokenNeedSizeDispatch)
                       + COMBINE_STATE_WIN_OFFSET + NOTIFY_DISPATCH_WIN_OFFSET)
                      * DOUBLE_DATA_BUFFER;

// 校验
if (actualSize > maxWindowSize) {
    // 报错: HCCL_BUFFSIZE too SMALL
}

8.3 分片策略

Dispatch 阶段:

// 按 core 分配 token
sendTokenNum = expertIdsCnt / blockNum;  // expertIdsCnt = batchSize * topK
remainTokenNum = expertIdsCnt % blockNum;
startTokenId = sendTokenNum * blockIdx;
if (blockIdx < remainTokenNum) {
    sendTokenNum += 1;
    startTokenId += blockIdx;
} else {
    startTokenId += remainTokenNum;
}

Combine 阶段:

// SplitCoreCal: 按 core 分配 token
perCoreNum = totalNum / aivNum_;
remainderRankNum = totalNum % aivNum_;
startIdx = perCoreNum * coreIdx_;
if (coreIdx_ < remainderRankNum) {
    perCoreNum++;
    startIdx += coreIdx_;
} else {
    startIdx += remainderRankNum;
}

8.4 状态区分配

// 每个 core 负责的 expert 数量
statusNumPerCore = moeExpertNum / blockNum;
remainStatus = moeExpertNum % blockNum;
startStatusId = statusNumPerCore * blockIdx;
if (blockIdx < remainStatus) {
    statusNumPerCore += 1;
    startStatusId += blockIdx;
} else {
    startStatusId += remainStatus;
}
endStatusId = startStatusId + statusNumPerCore;

8.5 Workspace 计算

// moe_dispatch_normal_tiling.cpp
workSpaces[0] = SYSTEM_NEED_WORKSPACE + WORKSPACE_ELEMENT_OFFSET * aivNum * aivNum;
// SYSTEM_NEED_WORKSPACE = 16MB
// WORKSPACE_ELEMENT_OFFSET = 512
// 例如 aivNum=48: 16MB + 512 * 48 * 48 = 16MB + 11.25MB = 27.25MB

// moe_combine_normal_tiling.cpp
workspace[0] = SYSTEM_NEED_WORKSPACE;  // 16MB

8.6 HCCL 通信配置

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

// Combine: AlltoAll (EP) + ReduceScatter (TP)
std::string algConfigAllToAllStr = "AlltoAll=level0:fullmesh;level1:pairwise";
std::string algConfigReduceScatterStr = "ReduceScatter=level0:ring";

9. 常量速查表

9.1 Window 布局常量

常量	值	来源文件	用途
`NOTIFY_DISPATCH_BUFF_OFFSET`	204MB (213909504)	comm_args.h	notify_dispatch 缓冲区偏移
`IPC_BUFF_MAX_SIZE`	100MB	comm_args.h	IPC 缓冲区最大大小
`IPC_DATA_OFFSET`	2MB	comm_args.h	IPC flag 区大小
`WIN_STATE_OFFSET`	500KB	moe_dispatch_normal.h	dataState 间偏移
`STATE_WIN_OFFSET`	950KB	moe_dispatch_normal.h	本卡状态区偏移
`WIN_ADDR_ALIGN`	512B	moe_dispatch_normal.h	Window 地址对齐
`COMBINE_STATE_WIN_OFFSET`	4MB	moe_dispatch_normal.h	combine 数据区偏移
`MAGIC_WIN_OFFSET`	975KB (998400)	moe_combine_normal.h	combine magic 标志偏移
`STATE_OFFSET`	32B	moe_dispatch_normal.h	每个 expert flag 占用

9.2 对齐常量

常量	值	用途
`UB_ALIGN`	32B	UB 地址对齐
`UB_32_ALIGN`	32B	UB 32字节对齐
`MUL_256_ALIGN`	256B	256B 对齐 (用于 FP32 累加)
`WIN_512_ALIGN`	512B	Window 地址对齐
`FLOAT_NUM_PER_ALIGN`	8	每 32B 对齐的 float 数量

9.3 缓冲区常量

常量	值	用途
`BUFFER_NUM`	2	Ping-Pong 双缓冲
`DOUBLE_BUFFER`	2	双缓冲
`EXPAND_IDX_INFO`	3	三元组长度 (rank, token, topk)
`TOKEN_SRC_INFO_LEN`	3	源信息长度

9.4 通信常量

常量	值	用途
`COMM_NUM`	2	通信域数量 (EP + TP)
`COMM_EP_IDX`	0	EP 通信域索引
`COMM_TP_IDX`	1	TP 通信域索引
`CAM_MAX_RANK_SIZE`	384	最大支持 rank 数
`MAX_EP_WORLD_SIZE`	384	最大 EP world size
`MAX_TP_WORLD_SIZE`	2	最大 TP world size

9.5 同步常量

常量	值	用途
`MAGIC_OFFSET`	32	magic 在高 32 位
`MAGIC_MASK`	`0xFFFFFFFF00000000`	magic 位掩码
`FLAG_UNIT_INT_NUM`	4	每个 flag 占 int64 数量
`SYNC_UNIT_SIZE`	32B	每个同步单元大小
`CYCLE_TO_TIME`	50	系统周期→时间转换

9.6 限制常量

常量	值	用途
`MOE_EXPERT_MAX_NUM`	512	最大 expert 数量
`K_MAX`	16	最大 topK
`H_MIN`	1024	最小 hidden size
`H_MAX`	7168	最大 hidden size
`BS_UPPER_BOUND`	8000	最大 batch size
`DISPATCH_STATUS_MAX_SUPPORT_NUM`	1280	最大 dispatch 状态数

10. 与 SHMEM 模式的代码级对比

10.1 内存管理

维度	HCCL Window 模式	SHMEM 模式
内存获取	`GetHcclContext<HCCL_GROUP_ID_0>()` 获取预分配 Window	IPC 共享内存 + 自管理 flag 区
数据区	`localWindowsIn` / `remoteWindowsIn`	IPC data 区 (100MB)
Flag 区	`localWindowsExp` / `remoteWindowsExp`	IPC flag 区 (2MB)
地址计算	`GetWindAddrByRankId` + 固定偏移	IPC 基址 + chunk 偏移

10.2 同步机制

维度	HCCL Window 模式	SHMEM 模式
Flag 编码	float32 = 1.0f (dispatch/combine)	Magic+Value (int64)
等待方式	ReduceSum/Sum 求和比较	`WaitOneRankPartFlag` 逐 flag 检查
清除方式	Duplicate 0 + DataCopy	SetFlag(0)
防 ABA	dataState Ping-Pong 隔离	Magic 翻转 (高 32 位)

10.3 双缓冲实现

维度	HCCL Window 模式	SHMEM 模式
缓冲切换	dataState (0/1) 存储在 `localWindowsExp + STATE_WIN_OFFSET`	magic 存储在 flag 区
偏移计算	`dataState * (totalWinSize / 2)`	`magic * (totalSize / 2)`
翻转时机	Init 阶段读取并翻转	Init 阶段读取并翻转

10.4 地址计算差异

HCCL Window 模式:

// 数据区
localWindowsIn + winDataSizeOffset + COMBINE_STATE_WIN_OFFSET + NOTIFY_DISPATCH_BUFF_OFFSET
// 状态区
localWindowsExp + dataState * WIN_STATE_OFFSET

SHMEM 模式 (推测):

// 数据区
ipcDataAddr + magic * (totalSize / 2) + tokenOffset
// Flag 区
ipcFlagAddr + magic * (flagSize / 2) + flagOffset

10.5 代码路径差异

代码路径	HCCL Window 模式	SHMEM 模式
上下文获取	`GetHcclContext<HCCL_GROUP_ID_0>()`	IPC 共享内存初始化
远端访问	`remoteRes[rankId].nextDevicePtr`	IPC 映射
Flag 写入	DataCopy 到远端 windowsExp	SetFlag 到 IPC flag 区
Flag 等待	ReduceSum 求和	WaitOneRankPartFlag 逐 flag 检查
同步原语	SyncFunc (HardEvent)	SyncCollectives 类

10.6 HCCL Window 独有的代码路径

双通信域管理: winContext_[COMM_EP_IDX] + winContext_[COMM_TP_IDX]
MC2 Tiling: Mc2InitTiling + Mc2CcTiling 配置
HCCL 算法配置: AlltoAll=level0:fullmesh;level1:pairwise
Window 大小校验: Mc2TilingUtils::GetMaxWindowSize() 从环境变量读取
dataState 状态机: selfDataStatusTensor 存储在 localWindowsExp + STATE_WIN_OFFSET
Magic 标志: MAGIC_WIN_OFFSET 用于 combine 阶段

11. 关键执行流程时序图

11.1 Dispatch → Combine 全流程

Rank 0                                    Rank 1
──────                                    ──────
Dispatch:
  InputToShare()
    ├─ 读取 token[0..N]
    ├─ 量化 (可选)
    ├─ FillTriple (srcRank=0, tokenIdx, topkIdx)
    └─ DataCopyPad → Rank1.windowsIn
         │
         │ DMA 跨片写入
         ▼
  SetStatus()                              SetStatus()
    ├─ 准备 flag (1.0f)                      ├─ 准备 flag (1.0f)
    └─ DataCopy → Rank1.windowsExp            └─ DataCopy → Rank0.windowsExp
         │                                         │
         ▼                                         ▼
  WaitStatus()                             WaitStatus()
    ├─ 轮询 windowsExp                         ├─ 轮询 windowsExp
    │  (ReduceSum == target?)                  │  (ReduceSum == target?)
    │  ← Rank1 已写入 flag                     │  ← Rank0 已写入 flag
    ├─ 清除 flag                               ├─ 清除 flag
    └─ SyncAll<true>()                         └─ SyncAll<true>()
         │                                         │
         ▼                                         ▼
  ShareToOutput()                          ShareToOutput()
    ├─ 从 Rank1.windowsIn 读取 token             ├─ 从 Rank0.windowsIn 读取 token
    ├─ 提取三元组 → expandIdxOut                ├─ 提取三元组 → expandIdxOut
    └─ DataCopyPad → expandXOut                 └─ DataCopyPad → expandXOut
         │                                         │
         ▼                                         ▼
Combine:                                   Combine:
  CopyBufferToShareAndSetStatus()          CopyBufferToShareAndSetStatus()
    ├─ 读取 recvX[tokenIdx]                      ├─ 读取 recvX[tokenIdx]
    ├─ DataCopyPad → Rank1.windowsIn             ├─ DataCopyPad → Rank0.windowsIn
    └─ SetStatusBySrcInfo → Rank1.windowsIn      └─ SetStatusBySrcInfo → Rank0.windowsIn
         │                                         │
         ▼                                         ▼
  ReadBufferFromRemote()                   ReadBufferFromRemote()
    ├─ WaitBuffCopy(tokenIdx)                    ├─ WaitBuffCopy(tokenIdx)
    │  (Sum == target?)                          │  (Sum == target?)
    │  ← Rank1 已写入 flag                       │  ← Rank0 已写入 flag
    ├─ 清除 flag                                 ├─ 清除 flag
    ├─ ReadBufferAndWeightedSum                  ├─ ReadBufferAndWeightedSum
    │  ├─ for topk in 0..K:                      │  ├─ for topk in 0..K:
    │  │  ├─ 从 localWindowsIn 读取 token         │  │  ├─ 从 localWindowsIn 读取 token
    │  │  ├─ Cast BF16→FP32                      │  │  ├─ Cast BF16→FP32
    │  │  ├─ Muls (scale)                        │  │  ├─ Muls (scale)
    │  │  └─ Add (累加)                          │  │  └─ Add (累加)
    │  ├─ Cast FP32→BF16                         │  ├─ Cast FP32→BF16
    │  └─ DataCopyPad → xOut                     │  └─ DataCopyPad → xOut
    └─ done                                      └─ done

11.2 Ping-Pong 时序

时间 ──────────────────────────────────────────────────────────────────→

                    ┌─── Iteration N ─────────────────────┐
                    │                                      │
Dispatch:           │  InputToShare → SetStatus             │
                    │  (写 dataState=0 区域)                 │
                    │                 ↓                     │
Combine:            │                 │  WaitBuffCopy        │
                    │                 │  (轮询 flag)          │
                    │                 ↓                     │
                    │  ReadBufferAndWeightedSum             │
                    │  (读 dataState=0 区域)                 │
                    │                                      │
────────────────────────────────────────────────────────────────────
                    │  ┌─── Iteration N+1 ─────────────────┐
                    │  │                                    │
                    │  │  InputToShare → SetStatus           │
                    │  │  (写 dataState=1 区域)               │
                    │  │               ↓                   │
                    │  │               │  WaitBuffCopy      │
                    │  │               │  (轮询 flag)        │
                    │  │               ↓                   │
                    │  │  ReadBufferAndWeightedSum           │
                    │  │  (读 dataState=1 区域)               │
                    │  │                                    │
────────────────────────────────────────────────────────────────────

附录 A: 关键函数索引

moe_dispatch_normal.h

函数	行号	职责
`Init`	173-261	初始化上下文、dataState、Buffer
`InputToShare`	346-410	GM → Window 数据写入
`SetStatus`	412-439	写入 flag 到远端 windowsExp
`WaitStatus`	441-543	轮询 flag 直到全部就绪
`ShareToOutput`	545-591	Window → GM 输出
`Process`	593-601	主流程编排
`GetWindAddrByRankId`	67-76	数据区地址计算
`GetWindStateAddrByRankId`	78-87	状态区地址计算
`FillTriple`	334-344	附加三元组 (rank, token, topk)
`QuantProcess`	295-332	动态量化 (BF16 → INT8)

moe_combine_normal.h

函数	行号	职责
`Init`	202-221	初始化 Magic、Tiling、Buffer
`InitMagic`	144-156	读取并翻转 Magic 标志
`CopyBufferToShareAndSetStatus`	224-283	GM → Window + flag 写入
`CopyBufferToShare`	286-303	数据 DMA 到远端 Window
`SetStatusBySrcInfo`	307-314	写入 flag 到远端
`WaitBuffCopy`	317-340	轮询 flag 直到 topK 全部到达
`ReadBufferAndWeightedSum`	343-377	加权求和 (FP32 累加)
`ReadBufferFromRemote`	379-412	主循环: 等待 + 求和
`Process`	414-420	主流程编排
`GetStateAddrByRankId`	67-76	状态区地址计算
`GetBufferAddrByRankId`	78-81	数据区地址计算

sync_collectives.h

函数	行号	职责
`MergeMagicWithValue`	303-307	Magic+Value 编码
`SetSyncFlag`	46-67	设置同步 flag
`WaitSyncFlag`	87-108	等待同步 flag
`WaitOneRankPartFlag`	320-347	轮询部分 flag
`CheckOneRankPartFlag`	356-376	检查部分 flag (一次性)

附录 B: 调试与诊断

B.1 Diagnose 模式

两个算子都支持 isEnableDiagnose 标志，用于收集通信耗时统计:

// Dispatch: waitRecvCostStats
if (isEnableDiagnose) {
    int32_t durationTime = (GetSystemCycle() - systemCycleStart) / CYCLE_TO_TIME;  // us
    // 按 srcRank 累加耗时
    waitRecvCostStatsTensor.SetValue(srcRank, preTime + durationTime);
}

// Combine: sendCostStats
if (isEnableDiagnose) {
    int32_t durationTime = (GetSystemCycle() - sendStartCycle) / CYCLE_TO_TIME;  // us
    sendCostStatsTensor.SetValue(srcRank, preTime + durationTime);
}

B.2 关键日志

// Tiling 阶段
OPS_LOG_D(nodeName, "windowSize = %lu", maxWindowSize);
OPS_LOG_D(nodeName, "blockDim=%u, aivNum=%u, ubSize=%lu", blockDim, aivNum, ubSize);
OPS_LOG_E(nodeName, "HCCL_BUFFSIZE is too SMALL, ... NEEDED_HCCL_BUFFSIZE=%luMB, HCCL_BUFFSIZE=%luMB.",
          actualSize / MB_SIZE + 1UL, maxWindowSize / MB_SIZE);

总结: HCCL Window 模式通过 HCCL 框架预分配的 Window 内存窗口，实现了高效的跨片通信。其核心设计包括: (1) Ping-Pong 双缓冲实现通信-计算重叠; (2) Magic+Value 编码实现无锁同步; (3) 三元组嵌入 (srcRank, srcTokenIdx, topkIdx) 避免额外通信; (4) FP32 累加避免精度损失。该模式是 CAM 库的生产默认路径，所有算子当前均使用此模式。

← SHMEM 模式 — 用户态 RDMA 📑 索引 notify_dispatch + dispatch_layout →

非 SHMEM — HCCL Window — UMDK/CAM

HCCL Window 模式 (Non-SHMEM) 代码深度剖析

目录

1. 整体架构概览

核心数据结构关系

通信模式对比

2. Window 内存布局

2.1 整体布局 (单卡视角)

2.2 关键偏移量定义

2.3 数据区 Token 格式 (Dispatch → Combine)

3. moe_dispatch_normal 完整执行流程

3.1 文件结构

3.2 模板参数

3.3 TilingKey 路由

3.4 Init 阶段详解

3.5 Process 主流程

3.6 InputToShare — 数据写入 Window

3.7 SetStatus — 通知远端数据已到

3.8 WaitStatus — 等待远端数据到达

3.9 ShareToOutput — 从 Window 读取到输出

4. Window 地址计算

4.1 GetWindAddrByRankId — 数据区地址

4.2 GetWindStateAddrByRankId — 状态区地址

4.3 winDataSizeOffset — Ping-Pong 偏移计算

4.4 地址计算全景图

5. Ping-Pong 双缓冲机制

5.1 设计思想

5.2 dataState 翻转逻辑

5.3 数据区划分

5.4 时序图

6. Magic+Value 同步协议

6.1 编码方式

6.2 Magic 翻转机制 (Combine 阶段)

6.3 Flag 等待逻辑

6.4 Dispatch 阶段的 Flag 编码

6.5 Combine 阶段的 Flag 编码

7. moe_combine_normal 完整执行流程

7.1 文件结构

7.2 模板参数

7.3 Init 阶段

7.4 Process 主流程

7.5 CopyBufferToShareAndSetStatus — 数据写入远端 Window

7.6 ReadBufferFromRemote — 等待数据 + 加权求和

7.7 WaitBuffCopy — 等待 topK 数据全部到达

7.8 ReadBufferAndWeightedSum — 加权求和

7.9 Combine 地址计算

8. Tiling 策略

8.1 Tiling 数据结构

8.2 Window 大小校验

8.3 分片策略

8.4 状态区分配

8.5 Workspace 计算

8.6 HCCL 通信配置

9. 常量速查表

9.1 Window 布局常量

9.2 对齐常量

9.3 缓冲区常量

9.4 通信常量

9.5 同步常量

9.6 限制常量

10. 与 SHMEM 模式的代码级对比

10.1 内存管理

10.2 同步机制

10.3 双缓冲实现

10.4 地址计算差异

10.5 代码路径差异

10.6 HCCL Window 独有的代码路径

11. 关键执行流程时序图

11.1 Dispatch → Combine 全流程

11.2 Ping-Pong 时序

附录 A: 关键函数索引

moe_dispatch_normal.h

moe_combine_normal.h

sync_collectives.h

附录 B: 调试与诊断

B.1 Diagnose 模式

B.2 关键日志