企业官网建设流程全解析

• 算子库
• 人工智能
• CANN

Real 算子是 CANN ops-math 数学算子库（experimental/math/real）中负责提取复数张量实部的单目算子：对于复数输入输出其实部，对于实数输入则执行恒等拷贝。本文围绕该算子的功能语义、参数与数据类型映射、Host 侧 Tiling 多核分核算法、Kernel 侧基于 GatherMask 的复杂双路径实现，以及 ACLNN 两段式调用接口展开，帮助读者既能在工程中正确调用aclnnReal，也能从源码层面理解其性能优化设计。

产品支持情况

需要说明的是，Real 算子的 AI Core（AICore）实现目前仅在 ascend910b 上注册，这一点在算子定义 real_def.cpp 的this->AICore().AddConfig("ascend910b", aicoreConfig)中有直接体现；而 COMPLEX128 类型走 AICPU 实现，适用于更多平台（详见后文"数据类型映射"）。

功能说明

• 算子功能：提取复数张量的实部（real part）。对于实数类型输入，输出等于输入（恒等操作）。
• 计算公式：

$$ output_i=\text{real}(input_i)=\begin{cases} \text{Re}(input_i), & \text{if } input_i \text{ is complex} \ input_i, & \text{if } input_i \text{ is real} \end{cases} $$

• 与 PyTorch 对应：torch.real()
• 与 NumPy 对应：np.real()

从图模式的算子语义看，real是一个逐元素（elementwise）算子：其 Shape 推导直接复用基类提供的逐元素推导函数InferShape4Elewise（见 real_infershape.cpp），输出 shape 与输入保持一致。

参数说明

在算子定义 real_def.cpp 中，input/output均声明为REQUIRED（必选），Tout属性声明为OPTIONAL（可选）且默认值ge::DT_FLOAT；同时算子声明了DynamicCompileStaticFlag(true)、DynamicFormatFlag(true)、DynamicRankSupportFlag(true)、DynamicShapeSupportFlag(true)，即同时支持动态 shape、动态 rank 与动态格式，这也是它能广泛接入框架侧动态推导链路的基础。

数据类型映射

注：

• COMPLEX128 类型仅支持 AICPU 实现，ascend910b 的 AICore 不支持该类型。
• Tout 值为数据类型枚举值（DT_FLOAT=0, DT_FLOAT16=1, DT_DOUBLE=11），可通过 Tout 属性可选指定输出类型。

Tiling Key 的语义在源码中也有对应定义：RealTilingKey枚举（real_tiling.h）中TILINGKEY_COMPLEX32=1、TILINGKEY_COMPLEX64=2、TILINGKEY_COMPLEX128=3、TILINGKEY_FLOAT16=4、TILINGKEY_FLOAT=5；Host 侧在 real_tiling.cpp 中根据输入 dtype 通过 switch 语句为每种输入类型设置对应的 tilingKey，Kernel 侧再依据 tilingKey 模板实例化不同的输入/输出类型组合。

约束说明

• 输入张量的 shape 必须与输出张量的 shape 相同。
• 支持动态 shape 和动态 rank。
• 输入数据不能为空（在 Tiling 阶段，若totalLength == 0会直接报错返回，见 real_tiling.cpp）。

在 ACLNN 接口层，aclnn_real.cpp 还会额外校验：所有参与张量的维度不超过ACLNN_MAX_SHAPE_RANK（8 维）、self与out的 shape 必须一致、输入输出 dtype 组合必须落在DTYPE_SUPPORT_LIST白名单内（如 COMPLEX64→FLOAT、COMPLEX32→FLOAT16、COMPLEX128→DOUBLE、FLOAT16→FLOAT16、FLOAT→FLOAT）。

实现说明

目录结构

Real 算子遵循 CANN 自定义算子开发的典型分层结构，仓库中的实际布局如下：

experimental/math/real/ ├── op_host/ # Host侧实现 │ ├── real_def.cpp # 算子定义 │ ├── real_infershape.cpp # Shape推导 │ └── real_tiling.cpp # Tiling计算实现 ├── op_kernel/ # Kernel侧实现 │ ├── real.cpp # Kernel入口 │ ├── real_kernel.h # Kernel模板类实现 │ └── real_tiling.h # Tiling数据结构定义 ├── op_api/ # API接口 │ ├── aclnn_real.cpp # ACLNN接口实现 │ ├── aclnn_real.h # ACLNN接口声明 │ ├── real.cpp # 算子API实现（l0op::Real） │ └── real.h # 算子API声明 ├── docs/ │ └── aclnnReal.md # API文档 ├── examples/ │ └── test_aclnn_real.cpp # ACLNN调用示例 └── tests/ut/ # 单元测试 ├── op_api/ │ └── test_aclnn_real.cpp ├── op_host/ │ ├── CMakeLists.txt │ └── test_real_tiling.cpp └── op_kernel/ ├── CMakeLists.txt ├── real_tiling.h ├── test_real.cpp └── real_data/ ├── gen_data.py # 测试数据生成 └── compare_data.py # 结果比对

Tiling 参数说明

Tiling 阶段产出的RealTilingData结构体定义在 real_tiling.h，核心字段含义如下：

• totalUsedCoreNum: 实际使用的总核数
• tailBlockNum: 大核数量（余数 block 数）
• ubPartDataNum: 每次 UB 循环处理的元素数
• smallCoreDataNum: 小核数据量（元素数）
• smallCoreLoopNum: 小核 UB 循环次数
• smallCoreTailDataNum: 小核最后一次循环的元素数
• bigCoreDataNum: 大核数据量（元素数）
• bigCoreLoopNum: 大核 UB 循环次数
• bigCoreTailDataNum: 大核最后一次循环的元素数
• tilingKey: 算子类型标识（1=complex32, 2=complex64, 4=float16, 5=float）
• useNonInplace: 是否使用非 inplace GatherMask 路径（0=inplace, 1=非 inplace）

Tiling 计算入口 real_tiling.cpp 会从输入 shape 取totalLength，校验输入输出 dtype 合法性，然后调用CalcRealTilingParam完成核心参数计算；平台信息（AIV 核数、UB 大小）通过platform_ascendc::PlatformAscendC获取，编译期也可通过RealCompileInfo（含totalCoreNum=30、ubSizePlatForm字段）在TilingPrepare4Real阶段提前预取。最终在PostTiling中写入 Tiling Data，并调用context_->SetBlockDim(totalUsedCoreNum)与context_->SetTilingKey(tilingKey)决定 Kernel 的并发核数与执行分支；同时还会预留一块固定大小的 userWorkspace（RESERVED_WORKSPACE，16MB）。

多核处理策略（大小核分核）

分核逻辑参考 Exp 等逐元素算子的通用做法，核心思路是把数据按 block 粒度均摊到多个 AI Core 上，并通过"大核 +1 block"的方式吸收余数，保证各核负载差不超过 1 个 block：

1. 对齐粒度：

   • Complex 类型：128B 对齐（满足 GatherMask inplace 的 256B 源数据约束）
   • Real 类型：32B 对齐

2. 分核策略：

   • 按输出数据类型字节数将总数据量对齐到对应 block：totalBlocks = Align(totalLength * dataTypeLength, alignSize) / alignSize
   • 若ubPartDataNum >= totalLength：使用 1 核
   • 否则：coreNum = min(totalCoreNum, totalBlocks)
   • everyCoreBlockNum = totalBlocks / coreNum，tailBlockNum = totalBlocks % coreNum

3. 大小核分配：

   • 前tailBlockNum个核为大核，数据量 =(everyCoreBlockNum + 1) * alignSize / dataTypeLength
   • 其余核为小核，数据量 =everyCoreBlockNum * alignSize / dataTypeLength
   • 大小核负载差 ≤ 1 个 block

4. 偏移计算（见 real_kernel.h 中Init的实现）：

   • 大核：globalOffset = blockIdx * bigCoreDataNum
   • 小核：globalOffset = blockIdx * bigCoreDataNum - (bigCoreDataNum - smallCoreDataNum) * (blockIdx - tailBlockNum)（即先按"全大核"假设计算偏移，再回退小核与大核的数据量差）
   • Complex 类型输入：inputOffset = globalOffset * 2（每个元素占 2 个 output 元素空间，源码中以uint64_t计算防止大张量下 offset 溢出）

Complex 双路径策略

GatherMask inplace 要求count * 2 * sizeof(T) % 256 == 0，tiling 据此选择路径（对应useNonInplace标志）：

1. Inplace 路径（useNonInplace=0）：

   • 适用：多核场景 / 单核且 totalLength 满足 256B 对齐
   • UB 分配：inQueue(2x) × 2缓冲 = 4倍系数
   • GatherMask inplace：GatherMask(src, src, mode=1, ...)，配合 pipeline prefetch 优化（在循环中提前下发下一 tile 的 DMA-in，与当前 tile 的 GatherMask 计算重叠，见ProcessComplexTiling）

2. 非 Inplace 路径（useNonInplace=1）：

   • 适用：单核且 totalLength 不满足 256B 对齐（如 complex32[4,4]=16元素，16×2×2=64 < 256）
   • UB 分配：inQueue(2x) + outQueue(1x) × 2缓冲 = 6倍系数
   • GatherMask 非 inplace：GatherMask(dst, src, mode=1, mask=count*2, repeatTimes=1)

从 Kernel 侧实现看，RealKernel<S, T>模板类（real_kernel.h）的核心思路非常巧妙：复数在内存中按"实部、虚部交错"存储，因此提取实部等价于从2N个元素中每隔一个取出一个——这正是向量指令 GatherMask 的典型应用场景：

• ExtractRealPart（inplace）通过params.repeatTimes = count * 2 * sizeof(T) / 256一次处理 256B 对齐的数据块；
• ExtractRealPartNonInplace通过mask = count * 2指定抽取的元素个数，将结果写入独立的 outQueue。

而实数输入的恒等分支（ProcessRealIdentity，real_kernel.h）则使用TQueBind绑定输入输出队列做纯拷贝（copy），同样带 prefetch 流水优化。Kernel 入口 real.cpp 按 tilingKey 实例化四组类型组合：

• COMPLEX32_MODE（key=1）：RealKernel<int32_t, half>，即输入按 32bit 复数存储、输出 half（FLOAT16）
• COMPLEX64_MODE（key=2）：RealKernel<int64_t, float>
• FLOAT16_MODE（key=4）：RealKernel<half, half>（恒等拷贝）
• FLOAT_MODE（key=5）：RealKernel<float, float>（恒等拷贝）

模板参数<S, T>中S为输入存储类型、T为输出计算类型，if constexpr (IsSameType<S, T>::value)在编译期就区分了"恒等拷贝"与"复数抽取"两条路径，零运行时开销。

调用说明

ACLNN API 调用

Real 算子对外提供标准的 CANN ACLNN 两段式接口（声明见 aclnn_real.h）：

#include "aclnnop/aclnn_real.h" // 1. 获取workspace大小 aclnnStatus aclnnRealGetWorkspaceSize(const aclTensor* self, aclTensor* out, uint64_t* workspaceSize, aclOpExecutor** executor); // 2. 执行算子 aclnnStatus aclnnReal(void* workspace, uint64_t workspaceSize, aclOpExecutor* executor, aclrtStream stream);

注意：ACLNN API 使用self和out作为参数名，与图模式的input/output不同。

第一段接口aclnnRealGetWorkspaceSize内部完成了四类参数校验（空指针、dtype 支持范围、输入输出 dtype 匹配、shape 一致性，见 aclnn_real.cpp），并构造出完整的计算图：self → l0op::Contiguous（非连续转连续）→ l0op::Real → l0op::ViewCopy（结果写回 out）。其中有两个值得注意的优化分支：

• FLOAT / FLOAT16 输入直接走l0op::ViewCopy透传拷贝，不再进入 Real 计算（aclnn_real.cpp），与"实数输入恒等操作"的语义完全一致；
• self为空 Tensor 时直接返回workspaceSize = 0，跳过后续计算。

L0 层入口l0op::Real（real.cpp）负责推导输出 dtype（COMPLEX64→FLOAT、COMPLEX32→FLOAT16、COMPLEX128→DOUBLE），并依据当前 SoC 版本选择执行后端：ASCEND910B / ASCEND910_93 上支持 AICore 加速（支持 FLOAT、FLOAT16、COMPLEX32、COMPLEX64），其余平台或 COMPLEX128 类型则走 AICPU 实现。

完整调用示例

以下示例来自仓库 examples/test_aclnn_real.cpp，完整演示了"初始化 → 构造 Tensor → 两段式调用 → 同步取结果 → 释放资源"的标准流程：

#include "aclnnop/aclnn_real.h" #include "acl/acl.h" #include <iostream> #include <vector> #define CHECK_RET(cond, return_expr) \ do { \ if (!(cond)) { \ return_expr; \ } \ } while (0) int Init(int32_t deviceId, aclrtStream* stream) { auto ret = aclInit(nullptr); CHECK_RET(ret == ACL_SUCCESS, return ret); ret = aclrtSetDevice(deviceId); CHECK_RET(ret == ACL_SUCCESS, return ret); ret = aclrtCreateStream(stream); CHECK_RET(ret == ACL_SUCCESS, return ret); return 0; } int main() { // 1. device/stream初始化 int32_t deviceId = 0; aclrtStream stream; auto ret = Init(deviceId, &stream); CHECK_RET(ret == ACL_SUCCESS, return ret); // 2. 构造输入与输出（FLOAT 输入，shape [4,2]，恒等输出） std::vector<int64_t> selfShape = {4, 2}; std::vector<int64_t> outShape = {4, 2}; void* selfDeviceAddr = nullptr; void* outDeviceAddr = nullptr; aclTensor* self = nullptr; aclTensor* out = nullptr; std::vector<float> selfHostData = {1, -1, -1, -2, 2, -2, -3, 3}; std::vector<float> outHostData = {1, -1, -1, -2, 2, -2, -3, 3}; // （此处通过 CreateAclTensor 辅助函数完成 aclrtMalloc + aclrtMemcpy + aclCreateTensor， // 完整代码见 examples/test_aclnn_real.cpp） // 3. 两段式调用：先获取 workspace 大小，再执行 uint64_t workspaceSize = 0; aclOpExecutor* executor; ret = aclnnRealGetWorkspaceSize(self, out, &workspaceSize, &executor); CHECK_RET(ret == ACL_SUCCESS, return ret); void* workspaceAddr = nullptr; if (workspaceSize > 0) { ret = aclrtMalloc(&workspaceAddr, workspaceSize, ACL_MEM_MALLOC_HUGE_FIRST); CHECK_RET(ret == ACL_SUCCESS, return ret); } ret = aclnnReal(workspaceAddr, workspaceSize, executor, stream); CHECK_RET(ret == ACL_SUCCESS, return ret); // 4. 同步等待任务执行结束 ret = aclrtSynchronizeStream(stream); CHECK_RET(ret == ACL_SUCCESS, return ret); // 5. 将 device 侧结果拷回 host 侧并打印 // PrintOutResult(outShape, &outDeviceAddr); // 6. 释放 aclTensor 与 device 资源 aclDestroyTensor(self); aclDestroyTensor(out); aclrtFree(selfDeviceAddr); aclrtFree(outDeviceAddr); if (workspaceSize > 0) { aclrtFree(workspaceAddr); } aclrtDestroyStream(stream); aclrtResetDevice(deviceId); aclFinalize(); return 0; }

示例中selfShape = {4, 2}、selfHostData = {1, -1, -1, -2, 2, -2, -3, 3}，由于输入为 FLOAT 类型，走恒等路径后输出与输入完全一致；若将输入改为 COMPLEX64，则需要把实部、虚部交错排列，输出将只保留实部数据（偶数下标元素）。API 的详细参数表与输入输出类型对应关系可进一步查阅 docs/aclnnReal.md。

约束补充（ACLNN 接口层）

• self 与 out 的 shape 必须相同。
• 支持动态 shape 和动态 rank。
• 输入数据不能为空。
• 确定性计算：aclnnReal 默认确定性实现。

测试与验证

Real 算子提供了三层单元测试（见 tests/ut 目录）：

• op_kernel 测试（test_real.cpp）：配合real_data/gen_data.py与real_data/compare_data.py生成输入数据并比对 Kernel 输出，覆盖复数抽取与实数恒等两类场景；
• op_host 测试（test_real_tiling.cpp）：校验不同 shape / dtype 组合下 Tiling 参数的合理性，重点覆盖单核/多核、大小核分核与 256B 对齐约束的边界条件；
• op_api 测试（test_aclnn_real.cpp）：验证两段式 ACLNN 接口在 Device 上的实际执行结果。

这类"数据生成脚本 + Host Tiling 单测 + Kernel 仿真 + API 端到端"的组合，正是 CANN 算子开发中推荐的完整验证链路。

总结

Real 算子是理解 CANN ops-math 库"单目逐元素算子"工程范式的极佳样例：语义上它只做一件简单的事（提取复数实部 / 实数恒等），但工程实现上融合了动态 shape 推导、基于 dtype 的 tilingKey 分派、大小核负载均衡的多核分核算法，以及巧用 GatherMask 向量指令在 inplace / 非 inplace 两条路径之间自适应切换的 Kernel 设计。无论是希望快速在工程中接入复数实部计算，还是想借鉴其 Tiling 与 Kernel 优化思路来开发自己的算子，都可以从 experimental/math/real 这份实现与文档中找到完整参考。

• 算子库
• 人工智能
• CANN