CAM 模块测试体系深度分析

分析范围：test/cam/ut/、PyTorch NPU Extension Bridge、集成测试框架、构建系统
分析日期：2026-07-21
项目路径：/Users/backyes/work/claude_workspace/umdk_research/umdk

一、单元测试体系分析

1.1 测试目录结构

test/cam/ut/
├── __init__.py
├── conftest.py              # pytest fixtures：通信域初始化 + 用例同步
├── pytest.ini               # marker 注册
├── run_shell.py             # 覆盖率包装入口
├── util/
│   ├── __init__.py
│   ├── marker.py            # 自定义 marker + 条件跳过
│   └── tool.py              # 分布式工具函数 + 精度校验
├── fused_deep_moe/
│   ├── __init__.py
│   └── test_fused_deep_moe.py
└── moe_dispatch_combine_prefill/
    ├── __init__.py
    └── test_moe_dispatch_combine_prefill.py

关键观察：当前 CAM 的 UT 体系是 纯 Python/pytest 风格，与 URMA 模块的 C++/gtest 体系形成鲜明对比。CAM 没有 C++ 层的 gtest UT，所有测试通过 Python 调用 torch.ops.umdk_cam_op_lib.* 自定义算子完成。

1.2 pytest.ini Marker 体系

[pytest]
markers =
    author : "marks test as author"
    multi_process_test : "marks test as multi_process_test"
    single_process_test : "marks test as single_process_test"
    a2_test : "marks test as A2/910B test"
    a3_test : "marks test as A3/910C test"

Marker	含义	使用场景
`MPTest` (multi_process_test)	多进程测试	所有 CAM 算子测试，通过 torchrun 拉起
`SPTest` (single_process_test)	单进程测试	当前未使用
`A2Test` (a2_test)	华为 910B 芯片测试	当前未使用
`A3Test` (a3_test)	华为 910C 芯片测试	fused_deep_moe、moe_dispatch_combine_prefill
`Author`	用例作者标记	当前未使用

1.3 conftest.py — Fixture 设计

@pytest.fixture(scope="function", autouse=True)
def test_cases_sync():
    if tool.is_torchrun_process():
        dist.barrier()
    yield
    if tool.is_torchrun_process():
        dist.barrier()

@pytest.fixture(scope="session", autouse=True)
def mpt_global_fixture():
    if tool.is_torchrun_process():
        device = torch.device("npu", tool.get_local_rank())
        torch_npu.npu.set_device(device)
        dist.init_process_group(backend="hccl", rank=tool.get_rank(), world_size=tool.get_world_size())
    yield

设计要点： - session 级：mpt_global_fixture 在测试会话开始时通过 HCCL 后端初始化 ProcessGroup，使用 torchrun 注入的环境变量（RANK/WORLD_SIZE/LOCAL_RANK/LOCAL_WORLD_SIZE） - function 级：test_cases_sync 在每个用例前后加 dist.barrier()，保证多进程用例执行同步 - autouse=True：两个 fixture 均自动生效，测试用例无需显式声明

1.4 util/marker.py — 条件跳过机制

def SKIP_ENV_RANKSIZE_UNEQUAL(expect_ranksize, reason='skip if ranksize is not equal to '):
    skip = False
    if expect_ranksize != tool.get_world_size():
        skip = True
        reason += str(expect_ranksize)
    return pytest.mark.skipif(condition=skip, reason=reason)

这是一个运行时条件跳过装饰器：当实际 WORLD_SIZE 不等于期望值时自动 skip。例如 @SKIP_ENV_RANKSIZE_UNEQUAL(4) 表示仅在 4 卡环境执行。

1.5 util/tool.py — 分布式工具函数

函数	功能
`is_torchrun_process()`	判断是否 torchrun 拉起的进程（检查 RANK/WORLD_SIZE/LOCAL_RANK/LOCAL_WORLD_SIZE 四个环境变量）
`get_rank()` / `get_local_rank()` / `get_world_size()` / `get_local_world_size()`	获取分布式信息
`is_run_for_cov()`	检查是否覆盖率模式（ENABLE_COV 环境变量）
`allclose_nparray()`	带容错的 numpy 数组比较：先 `np.allclose`，失败则统计不满足 rtol/atol 的元素比例

精度校验策略（allclose_nparray）：

error = np.abs(data_expect - data_check)
greater = np.greater(error, atol + np.abs(data_check) * rtol)
loss_count = np.count_nonzero(greater)
assert (loss_count / total_count) < rtol

1.6 run_shell.py — 覆盖率包装

relative_path = os.path.join("..", "..", "..", "scripts", "comm_operator", "tool", "run_with_cov.sh")
script_path = os.path.abspath(os.path.join(script_dir, relative_path))
command = ["bash", script_path] + args
subprocess.run(command)

注意：该文件引用了 scripts/comm_operator/tool/run_with_cov.sh，但当前该文件不存在。这是一个预留的覆盖率收集入口，说明 CAM 规划了基于 shell 的覆盖率收集流程，但尚未落地。

二、fused_deep_moe 测试用例深度分析

2.1 测试策略：参考实现 vs 融合算子对比

测试采用 "Golden Reference vs Fused Op" 的对比策略：

组件	角色	实现方式
`SmallOps`	参考实现（Golden Reference）	手动拆解为 `npu_moe_distribute_dispatch_v2` → `npu_grouped_matmul` → `npu_dequant_swiglu_quant` → `npu_grouped_matmul` → `npu_moe_distribute_combine_v2`
`FusionOp`	被测融合算子	调用 `torch.ops.umdk_cam_op_lib.fused_deep_moe` 单一融合接口

核心断言：

tool.allclose_nparray(small_token_np, fused_token_np, rtol=0.02, atol=2.0)  # token 输出
tool.allclose_nparray(small_share_token_np, fused_share_token_np, rtol=0.02, atol=2.0)  # share 输出
np.testing.assert_array_equal(small_count_np, fused_count_np)  # count 输出精确相等

2.2 测试场景参数化

CASE_4RANK = {
    "totalExpertNum": 16,
    "topk": 8,
    "batchSize": 16,
    "hiddenSize": 7168,
    "intermediateHiddenSize": 2048,
    "dynamicEPLB": False,
    "with_mc2_mask": False,
}

CASE_8RANK = {
    "totalExpertNum": 16,
    "topk": 8,
    "batchSize": 32,
    "hiddenSize": 7168,
    "intermediateHiddenSize": 2048,
    "dynamicEPLB": True,
    "with_mc2_mask": False,
}

注意：CASE_8RANK 已定义但当前测试仅使用 CASE_4RANK。

2.3 测试参数矩阵

维度	覆盖值	说明
执行模式	`GE`（图模式）、`Eager`	通过 `@pytest.mark.parametrize("mode", [("GE"), ('Eager')])`
Rank 数	4	`@SKIP_ENV_RANKSIZE_UNEQUAL(4)`
数据类型	bfloat16	`test_bfloat16 = True`
动态 batch	关闭	`dynamicBS = False`
Share Expert	关闭	`with_share = False`
Smooth Quant	关闭	`with_smooth = False`
MC2 Mask	关闭	`with_mc2_mask = False`

2.4 覆盖率规避策略

if mode == "GE" and tool.is_run_for_cov():
    return

图模式（GE）无法获得算子覆盖率，因此在覆盖率收集模式下自动跳过 GE 测试。

2.5 测试覆盖的算子链路

SmallOps._apply_ops 覆盖的完整 MoE 计算链路：

输入 x
  ↓
npu_moe_distribute_dispatch_v2 (EP 分发)
  ↓
npu_grouped_matmul (GMM1, int8 量化矩阵乘)
  ↓
npu_dequant_swiglu_quant (SwiGLU 激活 + 反量化)
  ↓
npu_grouped_matmul (GMM2, 带 pertoken scale)
  ↓
npu_moe_distribute_combine_v2 (EP 聚合)
  ↓
输出 + share_output + expert_token_nums

2.6 数据生成策略

generate_datas() 函数生成随机测试数据： - 权重：int8 随机值 [-16, 16]，shape 为 [local_expert_num, input_dim, output_dim] - Scale：float32 随机值 [0.0015, 0.0045] - Expert IDs：顺序排列后取模 expert_ids % moe_expert_num - Expert Scales：[0, 1) 均匀分布 - 输入 x：[-5, 5) 均匀分布

三、moe_dispatch_combine_prefill 测试用例深度分析

3.1 测试策略：自包含正确性验证

与 fused_deep_moe 的"参考实现对比"策略不同，dispatch_combine_prefill 采用数学定义验证：

# 参考输出计算
ref_weights = np.where(topk_idx_np == -1, np.float32(0), topk_weights_np).sum(axis=1).reshape(-1, 1)
ref_x = x_np * ref_weights

# 相似度计算（余弦相似度变体）
def calc_diff(x, y):
    x, y = x.astype(np.float64) + 1, y.astype(np.float64) + 1
    denominator = (x * x + y * y).sum()
    sim = 2 * (x * y).sum() / denominator
    return 1 - sim

assert(diff < 5e-5)

3.2 测试场景参数

CASE_16RANK = {
    "num_ranks": 16,
    "num_experts": 16,
    "num_topk": 8,
    "num_tokens": 32,
    "hidden_size": 7168,
}

维度	覆盖值
Rank 数	16
执行模式	Eager（GE 在覆盖率模式下跳过）
量化模式	`use_quant=False` 和 `use_quant=True`

3.3 测试覆盖的算子

# 1. 获取 dispatch layout
output_layout = torch.ops.umdk_cam_op_lib.get_dispatch_layout(topk_idx, num_experts, num_ranks)

# 2. dispatch 阶段
output_dispatch = torch.ops.umdk_cam_op_lib.moe_dispatch_prefill(
    x, topk_idx, topk_weights, num_tokens_per_expert, send_token_idx_small,
    group_ep, rank, num_ranks, use_quant)

# 3. 反量化（如果 use_quant=True）
recv_x = per_token_cast_back(recv_x, dynamic_scales_out) if use_quant else recv_x

# 4. combine 阶段
combine_x = torch.ops.umdk_cam_op_lib.moe_combine_prefill(
    recv_x, topk_idx, topk_weights, expand_idx_out, recv_count,
    group_ep, rank, num_ranks)

3.4 FP8 反量化实现

def per_token_cast_back(x_fp8: torch.Tensor, x_scales: torch.Tensor):
    if x_scales.dtype == torch.int:
        x_scales = x_scales.view(dtype=torch.int8).to(torch.int) << 23
        x_scales = x_scales.view(dtype=torch.float)
    x_fp32 = x_fp8.to(torch.float32).view(x_fp8.size(0), -1, 128)
    x_scales = x_scales.view(x_fp8.size(0), -1, 1)
    return (x_fp32 * x_scales).view(x_fp8.shape).to(torch.bfloat16)

这是 128-element per-tile FP8 反量化：将 int8 的 scale 通过位操作还原为 float，然后按 128 元素粒度做 per-token 缩放。

3.5 数据来源声明

文件头部注明：

Part of codes modified from: MagiAttention
Original Source: https://github.com/SandAI-org/MagiAttention.git

说明该测试用例参考了 SandAI 的 MagiAttention 实现。

四、PyTorch NPU Extension Bridge 实现

4.1 架构概览

┌─────────────────────────────────────────────────────────┐
│                    PyTorch 前端                          │
│              torch.ops.umdk_cam_op_lib.*                 │
└─────────────────────┬───────────────────────────────────┘
                      │
┌─────────────────────▼───────────────────────────────────┐
│              pybind11 绑定层                              │
│  fused_deep_moe.cpp / moe_dispatch_prefill.cpp / ...    │
└─────────────────────┬───────────────────────────────────┘
                      │
┌─────────────────────▼───────────────────────────────────┐
│           pytorch_npu_helper.hpp (EXEC_NPU_CMD)          │
│     at::Tensor → aclTensor 类型转换 + OpApi 调用         │
└─────────────────────┬───────────────────────────────────┘
                      │
┌─────────────────────▼───────────────────────────────────┐
│                    ACL Runtime                           │
│              libopapi.so / libcust_opapi.so              │
└─────────────────────────────────────────────────────────┘

4.2 NPUBridge — Tensor 到 StorageImpl 的桥接

class NPUBridge {
public:
    static NPUStorageImpl *GetNpuStorageImpl(const at::Tensor &tensor);
    static NPUStorageImpl *GetNpuStorageImpl(c10::StorageImpl *storageImpl);
    static NPUStorageImpl *GetNpuStorageImpl(c10::Storage &&storage);
    static NPUStorageDesc &GetNpuStorageImplDesc(const at::Tensor &tensor);
};

实现方式：通过 static_cast 将 PyTorch 的 c10::StorageImpl 强制转换为 CAM 自定义的 NPUStorageImpl。这是侵入式扩展——CAM 在创建 tensor 时用 NPUStorageImpl 替换默认 StorageImpl。

4.3 NPUStorageDesc — NPU 存储元数据

struct NPUStorageDesc {
    c10::SmallVector<int64_t, 5> base_sizes_;
    c10::SmallVector<int64_t, 5> base_strides_;
    c10::SmallVector<int64_t, 5> storage_sizes_;
    int64_t base_offset_ = 0;
    use_byte_size_t base_dtype_ = {};
    aclFormat origin_format_ = ACL_FORMAT_UNDEFINED;
    aclFormat npu_format_ = ACL_FORMAT_ND;
    caffe2::TypeMeta data_type_ = caffe2::TypeMeta::Make<uint8_t>();
};

关键字段： - base_sizes_ / base_strides_：原始 tensor 的 shape/stride - storage_sizes_：实际存储的 storage 维度（考虑 NZ 格式转换） - origin_format_ / npu_format_：ACL 格式（ND、FRACTAL_NZ 等） - data_type_：用于从 StorageImpl 重建 GE tensor

4.4 NPUStorageImpl — 自定义 Storage

struct NPUStorageImpl : public c10::StorageImpl {
    explicit NPUStorageImpl(use_byte_size_t, size_t size_bytes, at::DataPtr data_ptr,
                            at::Allocator *allocator, bool resizable);
    void release_resources() override;
    NPUStorageDesc npu_desc_;
};

工厂函数 make_npu_storage_impl 创建带 NPU 描述的 Storage：

c10::intrusive_ptr<c10::StorageImpl> make_npu_storage_impl(...) {
    if (data_ptr == nullptr) {
        data_ptr = allocator->allocate(size_bytes.as_int_unchecked());
    }
    return c10::make_intrusive<NPUStorageImpl>(...);
}

4.5 pytorch_npu_helper.hpp — 核心转换层

4.5.1 类型映射表

#define AT_ALL_SCALAR_TYPE_AND_ACL_DATATYPE_PAIR(_)                                                                    \
    _(at::ScalarType::Byte, ACL_UINT8)                                                                                 \
    _(at::ScalarType::Char, ACL_INT8)                                                                                  \
    _(at::ScalarType::BFloat16, ACL_BF16)                                                                              \
    ...

完整的 ATen ScalarType → ACL DataType 映射，覆盖 20+ 种数据类型。

4.5.2 FP8/FP4 低精度支持

#ifdef CAM_ACL_HAS_FP8_FP4
static struct LowPrecisionDtypeMapInitializer {
    LowPrecisionDtypeMapInitializer() {
        kATenScalarTypeToAclDataTypeTable[at::ScalarType::Float8_e5m2] = ACL_FLOAT8_E5M2;
        kATenScalarTypeToAclDataTypeTable[at::ScalarType::Float8_e4m3fn] = ACL_FLOAT8_E4M3FN;
        kATenScalarTypeToAclDataTypeTable[at::ScalarType::Float8_e8m0fnu] = ACL_FLOAT8_E8M0;
        kATenScalarTypeToAclDataTypeTable[at::ScalarType::Float4_e2m1fn_x2] = ACL_FLOAT4_E2M1;
    }
} low_precision_dtype_map_initializer;
#endif

关键设计：通过 CAM_ACL_HAS_FP8_FP4 宏在编译期探测 CANN 版本是否支持 FP8/FP4。旧版 CANN 8.5（aclDataType 到 ACL_BF16 截止）也能编译，低精度回退到 view(int8) 方案。

4.5.3 OpApi 动态加载

inline void *GetOpApiFuncAddr(const char *apiName)
{
    static auto custOpApiHandler = GetOpApiLibHandler(GetCustOpApiLibName());  // libcust_opapi.so
    if (custOpApiHandler != nullptr) {
        auto funcAddr = GetOpApiFuncAddrInLib(custOpApiHandler, GetCustOpApiLibName(), apiName);
        if (funcAddr != nullptr) return funcAddr;
    }
    static auto opApiHandler = GetOpApiLibHandler(GetOpApiLibName());  // libopapi.so
    return GetOpApiFuncAddrInLib(opApiHandler, GetOpApiLibName(), apiName);
}

双库回退：优先查找 libcust_opapi.so（自定义算子 API），找不到再回退到 libopapi.so（标准 ACLNN API）。

4.5.4 EXEC_NPU_CMD 宏 — 算子调用统一入口

#define EXEC_NPU_CMD(aclnn_api, ...)

这是整个 helper 的核心宏，封装了完整的 ACLNN 算子调用流程：

获取函数地址：GetOpApiFuncAddr 动态加载 aclnn_api 和 aclnn_apiGetWorkspaceSize
初始化大内存：InitHugeMemThreadLocal
类型转换：ConvertTypes(__VA_ARGS__, workspace_size_addr, executor_addr) 将所有参数转为 ACL 类型
获取 Workspace：调用 GetWorkspaceSize 获取所需 workspace 大小
分配 Workspace：通过 at::empty 在 NPU 上分配
执行算子：调用 OpApiFunc，传入 workspace 和 acl_stream
释放资源：ReleaseConvertTypes + UnInitHugeMemThreadLocal + ReleaseHugeMem

4.5.5 ConvertType — Tensor 到 aclTensor 的转换

inline aclTensor *ConvertType(const at::Tensor &at_tensor) {
    // 1. 获取 ACL 数据类型
    aclDataType acl_data_type = kATenScalarTypeToAclDataTypeTable[...];

    // 2. 判断格式（NZ vs ND）
    if (!IsOpInputBaseFormat(at_tensor)) {
        format = NPUBridge::GetNpuStorageImpl(at_tensor)->npu_desc_.npu_format_;
        storageDims = NPUBridge::GetNpuStorageImpl(at_tensor)->npu_desc_.storage_sizes_;
    } else {
        // 根据维度数推断格式：3D→NCL, 4D→NCHW, 5D→NCDHW
    }

    // 3. 创建 aclTensor
    aclCreateTensor(sizes, dimNum, acl_data_type, strides, offset, format, storageDims, ...);
}

4.6 pybind.cpp — 算子注册

PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
    m.def("fused_deep_moe", &FusedDeepMoeImplAutograd, "fused_deep_moe");
    m.def("get_dispatch_layout", &GetDispatchLayoutImplAutograd, "get_dispatch_layout");
    m.def("moe_dispatch_prefill", &MoeDispatchPrefillImplAutograd, "moe_dispatch_prefill");
    m.def("moe_combine_prefill", &MoeCombinePrefillImplAutograd, "moe_combine_prefill");
}

TORCH_LIBRARY(umdk_cam_op_lib, m) {
    m.def("fused_deep_moe(Tensor x, Tensor expert_ids, Tensor[] gmm1_weight, ...) -> Tensor[]");
    m.def("get_dispatch_layout(Tensor topk_idx, int num_experts, int num_ranks) -> (Tensor, Tensor)");
    m.def("moe_dispatch_prefill(...) -> (Tensor, Tensor, Tensor, Tensor, Tensor)");
    m.def("moe_combine_prefill(...) -> Tensor");
}

双层注册： 1. pybind11 层：提供 Python 直接调用入口（torch.ops.umdk_cam_op_lib.fused_deep_moe） 2. TORCH_LIBRARY 层：注册到 PyTorch 的 dispatcher，支持 torch.compile 图模式捕获

五、集成测试框架

5.1 UMDK 集成测试整体架构

test/intergration_test/
├── conftest.py                    # 全局 pytest 配置（随机种子、路径注册）
├── test_framework/
│   ├── app/
│   │   ├── common/                # C 语言公共库（socket、thread pool、log）
│   │   ├── urma/                  # URMA 测试应用层
│   │   ├── dlock/ umq/ ums/ urpc/
│   │   └── cam/                   # ← 当前不存在，CAM 无集成测试
│   ├── common/
│   │   ├── constants.py           # 全局常量（从 /etc/ubus_ci/test_env.yaml 加载）
│   │   ├── host.py                # SSH 远程主机管理
│   │   ├── check_resource.py
│   │   └── utils.py
│   └── ubus_test/
│       └── base_test.py           # 基类：setup_class/teardown、core dump 检测
└── test_suites/
    ├── URMA/                      # URMA 功能测试
    ├── URMA_UT/                   # URMA 单元集成测试
    ├── DLOCK/ UMQ/ UMS/ URPC_LIB/
    └── （无 CAM 测试套件）

5.2 CAM 在集成测试中的位置

关键发现：test_framework/app/cam/ 目录不存在，test_suites/ 下也没有 CAM 相关测试套件。

这意味着： - CAM 当前没有参与 UMDK 的集成测试体系 - CAM 的测试完全依赖单元测试（pytest + 真实 NPU 多卡环境） - 与 URMA 模块（有完整的集成测试套件）形成对比

5.3 URMA 集成测试框架参考

URMA 的集成测试框架设计值得 CAM 借鉴：

BaseTest 基类（ubus_test/base_test.py）： - setup_class：解析测试环境 YAML、记录 core dump 状态 - teardown_class：进程清理、core dump 检测 - parse_host：通过 SSH 连接多台测试主机 - 支持多主机（host_list）、多设备（test_nic1_dev）

测试环境配置（/etc/ubus_ci/test_env.yaml）：

host_info:
  host1:
    user: root
    manage_nic: { name: xxx, ip: x.x.x.x }
    test_nic1: { name: xxx, ip: x.x.x.x, eid: xxx }
  host2: ...

conftest.py 全局配置：

def pytest_configure(config):
    random_seed = config.getoption("--random_seed")
    random.seed(int(random_seed))
    const.CASE_PATH = ...
    const.TMOUT = timeout

六、回归测试机制

6.1 当前回归测试现状

机制	CAM 状态	说明
CI Pipeline	❌ 未发现	无 Jenkinsfile / .github/workflows / .gitlab-ci.yml
自动化触发	❌	无 pre-commit / post-merge 自动触发证据
覆盖率门禁	❌	`run_with_cov.sh` 不存在
多芯片覆盖	⚠️ 部分	A3（910C）有 marker，A2（910B）有 marker 但无测试使用
多 Rank 覆盖	⚠️ 部分	4-Rank 和 16-Rank 有测试，但 8-Rank 场景已定义未使用

6.2 条件跳过作为回归防护

@SKIP_ENV_RANKSIZE_UNEQUAL(4)  # 仅 4 卡环境执行
@SKIP_ENV_RANKSIZE_UNEQUAL(16) # 仅 16 卡环境执行

这种机制确保： - 测试不会在错误的硬件配置下执行（避免误报） - 但也不会在错误的配置下失败（skip 而非 fail）

6.3 与 URMA 回归测试的对比

URMA 有完善的回归测试基础设施（test/urma/script/urma_ut_common.sh）：

能力	URMA	CAM
分阶段构建	✅ `common/core/cmd_tlv/bond/uvs/fuzz`	❌
代码覆盖率	✅ lcov + genhtml + 90% 门禁	❌
ASAN 内存检测	✅ `-fsanitize=address`	❌
Fuzz 测试	✅ `urma_api_fuzz.c` + API 覆盖率检查	❌
增量覆盖率	✅ `check_diff_coverage.py --threshold 90`	❌
JUnit 报告	✅ `--output-junit`	❌

七、测试基础设施

7.1 构建系统

7.1.1 顶层构建流程

build/cam/build.sh (入口)
  └── build/cam/comm_operator/build.sh
        ├── compile_ascend_proj.sh  → 编译 .run 包（op_host + op_kernel）
        ├── build_pybind.sh         → 编译 .whl 包（pybind11 扩展）
        └── BuildTest (未实现)      → UT 构建入口

7.1.2 compile_ascend_proj.sh — 算子编译

BuildTest 函数：在 build.sh 中被调用（if [ $ENABLE_UT_BUILD -eq 1 ]; then BuildTest; fi），但在整个代码库中未找到定义。这是一个未实现的预留接口。

编译流程： 1. select_ops.py 解析 operator_registry.json，根据 SOC/SHMEM/quant 筛选算子 2. msopgen gen 生成 CMakePresets.json 和项目骨架 3. copy_ops_include 仅复制选中算子的源码（include-list 模式） 4. cmake -S . -B build_out --preset=default 配置 5. cmake --build build_out --target binary -j${CAM_BUILD_JOBS:-8} 编译 6. 生成 .run 包

7.1.3 operator_registry.json — 算子注册表

{
  "soc_versions": {
    "ascend910_93": [
      "dispatch_layout", "fused_deep_moe", "moe_combine_normal",
      "moe_dispatch_normal", "notify_dispatch"
    ]
  },
  "operator_meta": {
    "fused_deep_moe": { "requires_shmem": false, "family": "fused_deep_moe", "variant": "base" }
  }
}

设计亮点： - SOC 版本 → 算子列表的映射 - requires_shmem 标记（SHMEM 未安装时自动过滤） - family + variant 处理互斥变体（base vs w4a8 共享文件名，不可同时编译）

7.1.4 select_ops.py — 算子选择逻辑

def resolve(registry, soc_list, shmem, quant, user_ops):
    # 1. 确定候选集（全量 or 用户指定）
    # 2. fused_deep_moe base/w4a8 互斥规则
    # 3. SHMEM 过滤
    # 4. 保持 registry 顺序（确定性）

7.2 CMake 构建配置

7.2.1 op_kernel/CMakeLists.txt

npu_op_kernel_options("" ALL OPTIONS ${KERNEL_COMPILE_OPTS})
npu_op_kernel_sources(ascendc_kernels KERNEL_DIR ./)
npu_op_kernel_library(ascendc_kernels SRC_BASE ${CMAKE_CURRENT_SOURCE_DIR}/ TILING_LIBRARY cust_optiling)
npu_op_package_add(${package_name} LIBRARY ascendc_kernels)

编译选项： - SHMEM 包含路径：-I$ENV{SHMEM_HOME_PATH}/shmem/include - CATLASS 包含路径：-I$ENV{CATLASS_HOME_PATH}/include - Debug 模式：-g -O0 - 生产模式：无额外 flag

7.2.2 op_host/CMakeLists.txt

CANN 版本探测：

set(CANN83_PATH "${CANN_ROOT}/include/platform/platform_infos_def.h")
set(CANN82_PATH "${CANN_ROOT}/include/experiment/platform/platform/platform_infos_def.h")

5 步构建流程： 1. npu_op_code_gen：生成 ACLNN stubs、op_proto.cc、fallback 2. npu_op_library(cust_opapi ACLNN ...)：ACLLN 单算子调用库 3. npu_op_library(cust_op_proto GRAPH ...)：算子原型库（图编译） 4. npu_op_library(cust_optiling TILING ...)：Tiling 库 5. npu_op_package_add：注册到 run 包

7.3 测试链接方式

CAM 测试不链接 C++ 测试库（gtest/gmock），而是： 1. 通过 import umdk_cam_op_lib 加载 pybind 编译的 .so 2. 通过 torch.ops.umdk_cam_op_lib.* 调用算子 3. 通过 torch.distributed + HCCL 实现多进程通信

八、测试覆盖率与质量保证策略

8.1 当前覆盖率机制

层次	机制	状态
Python 算子覆盖率	`tool.is_run_for_cov()` 跳过 GE 模式	✅ 已实现
C++ 代码覆盖率	`run_with_cov.sh`（不存在）	❌ 未实现
算子级覆盖率	无	❌
分支覆盖率	无	❌

8.2 精度质量保证

8.2.1 容差设计

测试	rtol	atol	说明
fused_deep_moe token output	0.02	2.0	宽松容差（int8 量化累积误差）
fused_deep_moe share output	0.02	2.0	同上
fused_deep_moe count output	精确相等	-	`assert_array_equal`
dispatch_combine_prefill	5e-5（余弦相似度）	-	更严格（FP8 反量化后）

8.2.2 容差选择的合理性

rtol=0.02, atol=2.0：对于 int8 量化 MoE（hidden_size=7168），2.0 的绝对误差容忍度对应约 0.03% 的相对误差（假设值域 [-5,5]），是合理的量化精度验证
5e-5 余弦相似度：对于 dispatch/combine（纯通信 + 加权），要求几乎无损

8.3 测试质量评估

8.3.1 优势

真实硬件验证：测试在真实 NPU 多卡环境运行，非 mock/simulator
端到端对比：fused_deep_moe 有完整的 Golden Reference 实现
多模式覆盖：Eager + GE 双模式
条件跳过：通过 marker 实现环境自适应

8.3.2 不足

不足	影响	建议
无 C++ 单元测试	算子内部逻辑（kernel、tiling）无独立验证	参考 URMA 引入 gtest
无 Fuzz 测试	边界条件覆盖不足	引入 fuzz 生成随机 shape/config
无 ASAN 检测	内存问题无法在测试阶段发现	测试构建启用 `-fsanitize=address`
无代码覆盖率门禁	无法衡量测试完整性	接入 lcov + diff coverage
无集成测试	算子间交互、端到端训练无验证	添加 CAM 集成测试套件
无 CI 配置	回归测试无法自动触发	添加 Jenkinsfile / GitHub Actions
CASE_8RANK 已定义未使用	8 卡场景无覆盖	启用或删除
无动态 batch / share / smooth 测试	关键功能分支未覆盖	扩展参数化矩阵
无异常输入测试	鲁棒性未知	添加边界值、空输入、超范围值测试

8.4 与 URMA 测试体系的对比总结

维度	URMA	CAM
测试语言	C++ (gtest) + Python	Python (pytest)
测试层级	UT + Fuzz + 集成	仅 UT
代码覆盖率	lcov + 90% 门禁	无
内存安全	ASAN	无
构建系统	CMake + ctest	shell + pytest
报告格式	JUnit XML + HTML	控制台输出
分阶段执行	6 个 phase	无
多芯片支持	无芯片限制	A2/A3 marker（部分使用）
多 Rank 支持	单进程为主	多进程（torchrun）

九、关键发现与建议

9.1 架构级发现

CAM 采用"Python 测试 + C++ 算子"的混合模式：测试逻辑完全在 Python 层，通过 pybind 调用 C++ 算子。这种模式开发效率高，但无法覆盖 C++ 内部逻辑。
NPUBridge 是侵入式设计：通过 static_cast 将 c10::StorageImpl 转为 NPUStorageImpl，依赖 CAM 自己创建 tensor。如果 PyTorch 内部实现变化（如 StorageImpl 布局调整），可能导致隐蔽 bug。
pytorch_npu_helper.hpp 是完整的 ACL 抽象层：EXEC_NPU_CMD 宏封装了完整的算子调用生命周期（workspace 管理、大内存初始化、类型转换、错误处理），设计成熟。
FP8/FP4 支持有版本兼容设计：通过 CAM_ACL_HAS_FP8_FP4 宏实现新旧 CANN 的兼容编译。

9.2 测试策略建议

短期： - 启用 CASE_8RANK 测试 - 添加 with_share=True、with_smooth=True、with_mc2_mask=True 的参数化组合 - 实现 run_with_cov.sh 覆盖率收集
中期： - 为 C++ 核心逻辑（NPUBridge、StorageImpl、类型转换）添加 gtest - 添加 Fuzz 测试覆盖随机 shape/config - 接入 CI Pipeline（Jenkinsfile）
长期： - 构建 CAM 集成测试套件（参考 URMA 框架） - 添加端到端训练验证（小模型 + 真实数据） - 建立性能回归测试（吞吐量/延迟基线）

附录：关键文件路径索引

文件	路径
conftest.py	`test/cam/ut/conftest.py`
pytest.ini	`test/cam/ut/pytest.ini`
marker.py	`test/cam/ut/util/marker.py`
tool.py	`test/cam/ut/util/tool.py`
fused_deep_moe 测试	`test/cam/ut/fused_deep_moe/test_fused_deep_moe.py`
dispatch_combine 测试	`test/cam/ut/moe_dispatch_combine_prefill/test_moe_dispatch_combine_prefill.py`
NPUBridge.h	`src/cam/comm_operator/pybind/pytorch_extension/NPUBridge.h`
NPUBridge.cpp	`src/cam/comm_operator/pybind/pytorch_extension/NPUBridge.cpp`
NPUStorageImpl.h	`src/cam/comm_operator/pybind/pytorch_extension/NPUStorageImpl.h`
NPUStorageImpl.cpp	`src/cam/comm_operator/pybind/pytorch_extension/NPUStorageImpl.cpp`
pytorch_npu_helper.hpp	`src/cam/comm_operator/pybind/pytorch_extension/pytorch_npu_helper.hpp`
pybind.cpp	`src/cam/comm_operator/pybind/pybind.cpp`
compile_ascend_proj.sh	`build/cam/comm_operator/compile_ascend_proj.sh`
build.sh	`build/cam/comm_operator/build.sh`
operator_registry.json	`src/cam/comm_operator/ascend_kernels/operator_registry.json`
select_ops.py	`build/cam/comm_operator/select_ops.py`
op_kernel/CMakeLists.txt	`src/cam/comm_operator/ascend_kernels/cmake_files/op_kernel/CMakeLists.txt`
op_host/CMakeLists.txt	`src/cam/comm_operator/ascend_kernels/cmake_files/op_host/CMakeLists.txt`
URMA UT CMakeLists	`test/urma/CMakeLists.txt`
URMA UT 脚本	`test/urma/script/urma_ut_common.sh`
集成测试框架	`test/intergration_test/test_framework/`

← FusedDeepMoE 融合算子 📑 索引 SHMEM 模式 — 用户态 RDMA →

测试体系与回归测试 — UMDK/CAM