企业官网建设流程全解析

CANN Ascend C 向量编程入门：基于 TPipe 与 TQue 队列机制实现 Add 向量加法算子

【免费下载链接】cann-samplesCANN高性能实战演进样例与体系化调优知识库项目地址: https://gitcode.com/cann/cann-samples

导读

本文以 CANN 开源仓库 cann-samples 中的 add_tpipe_tque 样例（位于 Samples/0_Introduction/01_simd_cpp_api/01_add/add_tpipe_tque）为对象，系统讲解如何在 Ascend C（SIMD C++ API）编程模型中，使用TPipe与TQue的内存管理和同步机制实现 z = x + y 的向量加法算子。读者学完后，将掌握"多核数据切分 → GM 搬入 UB → 队列入队/出队 → UB 内向量计算 → 结果写回 GM"的完整队列式编程范式，并能够独立完成样例的编译、运行与精度验证。

概述

本样例基于TPipe和TQue的内存和同步管理机制实现 Add 向量加法操作。与静态 Tensor 直接分配 UB 空间的写法不同，队列式编程通过TPipe统一管理 UB 缓冲的生命周期，通过TQue（队列）在"搬入—计算—搬出"各阶段之间传递LocalTensor并隐含阶段间的同步关系，是 Ascend C 中一种基础而重要的编程范式。同一目录下还提供了使用静态 Tensor 实现 Add 的 add 样例，两者在父目录 01_add/README.md 中被明确列为 Ascend C Add 算子的两种实现方式，可以对照学习。

支持的产品及 CANN 软件版本

本样例支持的产品与 CANN 软件版本对应关系如下：

样例规格

• 样例类型（OpType）：Add
• 输入x：形状 [8, 2048]，数据类型 float，数据排布格式 ND
• 输入y：形状 [8, 2048]，数据类型 float，数据排布格式 ND
• 输出z：形状 [8, 2048]，数据类型 float，数据排布格式 ND
• 核函数名：add_custom

计算规格为：对两个形状相同的张量做逐元素相加，计算公式为z = x + y。核函数以 8 个核并行启动，每个核负责处理其中一段连续数据。

目录结构介绍

├── add_tpipe_tque │ ├── scripts │ │ ├── gen_data.py // 输入数据和真值数据生成脚本 │ │ └── verify_result.py // 验证输出数据和真值数据是否一致的验证脚本 │ ├── CMakeLists.txt // 编译工程文件 │ ├── data_utils.h // 数据读入写出函数 │ ├── add_tpipe_tque.asc // Ascend C 样例实现，TPipe 和 TQue 管理内存和同步 & 调用样例 │ └── README.md // 样例说明文档

其中，add_tpipe_tque.asc 是核心源码文件，同时包含核函数实现与 host 侧调用逻辑；data_utils.h 提供二进制文件的读取与写出工具函数。

核心原理：TPipe 与 TQue 队列式编程模型

GM 与 UB：算子的两类数据空间

在 Ascend C 向量编程中，数据主要分布在两级存储上：

• GM（Global Memory）：AI Core 外部的全局内存，容量大但访问速度慢，通过GlobalTensor访问；
• UB（Unified Buffer）：AI Core 内部向量计算专用缓存，容量有限但访问速度快，通过LocalTensor访问。

向量计算单元只能访问 UB 上的数据，因此算子内核必然遵循"搬入—计算—搬出"三段式结构：先用DataCopy将输入从 GM 搬到 UB，在 UB 内完成向量计算，再将结果从 UB 搬回 GM。

TPipe：内存与同步的统一管理者

TPipe（Pipeline）是队列式编程中管理片上内存与流水同步的核心对象。在本样例中，add_custom内部创建了TPipe pipe，并通过pipe.InitBuffer(...)为各队列申请与blockLength对应的 UB 缓冲：

pipe.InitBuffer(inQueueX, 1, blockLength * sizeof(float)); pipe.InitBuffer(inQueueY, 1, blockLength * sizeof(float)); pipe.InitBuffer(outQueueZ, 1, blockLength * sizeof(float));

InitBuffer的第一个参数是目标队列，第二个参数为缓冲区个数（本样例为 1，即单缓冲），第三个参数为单个缓冲区的字节大小。由TPipe统一管理这些 UB 空间的生命周期，开发者无需手工管理 UB 地址分配与回收。

TQue：阶段间传递张量的队列机制

TQue是与TPipe配套的队列对象，用于在流水阶段之间传递LocalTensor。本样例声明了三个队列：

AscendC::TQue<AscendC::TPosition::VECIN, 1> inQueueX; AscendC::TQue<AscendC::TPosition::VECIN, 1> inQueueY; AscendC::TQue<AscendC::TPosition::VECOUT, 1> outQueueZ;

模板参数中，TPosition::VECIN/TPosition::VECOUT表示队列对应输入（vector 搬入）或输出（vector 搬出）位置，第二个参数表示队列深度。队列的典型使用流程为：

1. AllocTensor<T>()：从队列申请一个LocalTensor（即从TPipe管理的 UB 缓冲中取出一块）；
2. DataCopy：完成 GM 与 UB 之间的数据搬运；
3. EnQue：将已就绪的LocalTensor入队，作为"生产者"通知后续阶段数据可用；
4. DeQue：后续阶段从队列取出LocalTensor继续处理，作为"消费者"等待数据就绪；
5. FreeTensor：处理完毕后释放该张量，归还给队列/缓冲区。

EnQue与DeQue成对出现，天然构成了阶段间的同步点——DeQue会等待对应EnQue的数据就绪，从而在不显式调用PipeBarrier的情况下完成流水同步。这正是文档中所说的"TPipe 和 TQue 的内存和同步管理机制"的核心体现。

源码级解析：add_custom 核函数

多核数据切分：GetBlockNum 与 GetBlockIdx

核函数入口add_custom接收totalLength（全局输入长度）。在核函数内部，通过GetBlockNum()获取本次启动的核数，按核数均分每个 block 的处理区间；通过GetBlockIdx()获取当前核的编号，据此计算当前核在 GM 中的数据起点，确保各核访问互不重叠的数据分片：

__global__ __vector__ void add_custom(__gm__ uint8_t* x, __gm__ uint8_t* y, __gm__ uint8_t* z, uint32_t totalLength) { AscendC::TPipe pipe; AscendC::TQue<AscendC::TPosition::VECIN, 1> inQueueX; AscendC::TQue<AscendC::TPosition::VECIN, 1> inQueueY; AscendC::TQue<AscendC::TPosition::VECOUT, 1> outQueueZ; AscendC::GlobalTensor<float> xGm; AscendC::GlobalTensor<float> yGm; AscendC::GlobalTensor<float> zGm; // totalLength 表示全局输入长度；GetBlockNum() 返回本次启动的核数， // 这里按核数均分每个 block 的处理区间。 uint32_t blockLength = totalLength / AscendC::GetBlockNum(); // 根据 block_idx 计算当前核在 GM 中的起始地址，确保各核访问互不重叠的数据分片。 xGm.SetGlobalBuffer((__gm__ float*)x + blockLength * AscendC::GetBlockIdx(), blockLength); yGm.SetGlobalBuffer((__gm__ float*)y + blockLength * AscendC::GetBlockIdx(), blockLength); zGm.SetGlobalBuffer((__gm__ float*)z + blockLength * AscendC::GetBlockIdx(), blockLength); // 为输入和输出队列申请与 blockLength 对应的 UB 缓冲，保持 TPipe/TQue 的编程范式。 pipe.InitBuffer(inQueueX, 1, blockLength * sizeof(float)); pipe.InitBuffer(inQueueY, 1, blockLength * sizeof(float)); pipe.InitBuffer(outQueueZ, 1, blockLength * sizeof(float)); // 使用 DataCopy 将输入从 GM 搬运到 UB，并通过 EnQue 将 LocalTensor 入队， // 供后续计算阶段 DeQue 取用。 AscendC::LocalTensor<float> xLocal = inQueueX.AllocTensor<float>(); AscendC::LocalTensor<float> yLocal = inQueueY.AllocTensor<float>(); AscendC::DataCopy(xLocal, xGm, blockLength); AscendC::DataCopy(yLocal, yGm, blockLength); inQueueX.EnQue(xLocal); inQueueY.EnQue(yLocal); // DeQue 取出输入张量，在 UB 内执行 Add，并将结果 EnQue 到输出队列，供后续写回 GM。 xLocal = inQueueX.DeQue<float>(); yLocal = inQueueY.DeQue<float>(); AscendC::LocalTensor<float> zLocal = outQueueZ.AllocTensor<float>(); AscendC::Add(zLocal, xLocal, yLocal, blockLength); outQueueZ.EnQue<float>(zLocal); inQueueX.FreeTensor(xLocal); inQueueY.FreeTensor(yLocal); // 从输出队列取出结果，并写回当前核负责的 GM 分片。 zLocal = outQueueZ.DeQue<float>(); AscendC::DataCopy(zGm, zLocal, blockLength); outQueueZ.FreeTensor(zLocal); }

处理流程梳理

对照源码，本样例的处理流程为：

1. add_custom作为核入口接收totalLength；
2. 通过GetBlockNum()计算当前 block 的数据长度blockLength，通过GetBlockIdx()计算当前核在 GM 中对应的数据起点；
3. 使用DataCopy把输入数据从 GM 搬到 UB，并通过EnQue将输入LocalTensor放入输入队列；
4. 通过DeQue从输入队列取出输入张量，在 UB 中执行Add，再通过EnQue将结果LocalTensor放入输出队列；
5. 通过DeQue从输出队列取出结果，并使用DataCopy写回当前核负责的 GM 分片。

值得注意的是，输入张量xLocal、yLocal在DeQue取出并完成Add之后通过FreeTensor归还给队列，而输出张量zLocal则是在DeQue取出、DataCopy写回 GM 之后才FreeTensor。从源码结构看，这种"用完即还"的做法保证了同一队列的缓冲可以被复用，正是TPipe/TQue内存管理自动化的体现——开发者不需要显式关心 UB 缓冲的物理地址。

队列说明

本样例使用TPipe和TQue演示基础的队列式编程方式。EnQue用于将已经搬到 UB 的LocalTensor入队，DeQue用于在后续阶段从队列中取出张量继续处理。整条链路"搬入（DataCopy）→ 入队（EnQue）→ 出队（DeQue）→ 计算（Add）→ 入队（EnQue）→ 出队（DeQue）→ 搬出（DataCopy）"构成了一个完整的队列式流水。

核入口与 host 侧调用实现

add_custom核入口负责创建TPipe、TQue和GlobalTensor对象，并按顺序执行搬入、计算、搬出处理链路。Host 侧调用则位于同一文件的main函数中：

int32_t main(int32_t argc, char* argv[]) { // 启动 8 个核并行处理，每个核负责总长度的 1/8。 uint32_t numBlocks = 8; constexpr uint32_t totalLength = 8 * 2048; size_t inputByteSize = totalLength * sizeof(float); size_t outputByteSize = totalLength * sizeof(float); uint8_t* xHost = nullptr; uint8_t* yHost = nullptr; uint8_t* zHost = nullptr; uint8_t* xDevice = nullptr; uint8_t* yDevice = nullptr; uint8_t* zDevice = nullptr; aclInit(nullptr); int32_t deviceId = 0; aclrtSetDevice(deviceId); aclrtStream stream = nullptr; aclrtCreateStream(&stream); aclrtMallocHost((void**)(&xHost), inputByteSize); aclrtMallocHost((void**)(&yHost), inputByteSize); aclrtMallocHost((void**)(&zHost), outputByteSize); aclrtMalloc((void**)&xDevice, inputByteSize, ACL_MEM_MALLOC_HUGE_FIRST); aclrtMalloc((void**)&yDevice, inputByteSize, ACL_MEM_MALLOC_HUGE_FIRST); aclrtMalloc((void**)&zDevice, outputByteSize, ACL_MEM_MALLOC_HUGE_FIRST); ReadFile("./input/input_x.bin", inputByteSize, xHost, inputByteSize); ReadFile("./input/input_y.bin", inputByteSize, yHost, inputByteSize); aclrtMemcpy(xDevice, inputByteSize, xHost, inputByteSize, ACL_MEMCPY_HOST_TO_DEVICE); aclrtMemcpy(yDevice, inputByteSize, yHost, inputByteSize, ACL_MEMCPY_HOST_TO_DEVICE); add_custom<<<numBlocks, 0, stream>>>(xDevice, yDevice, zDevice, totalLength); aclrtSynchronizeStream(stream); aclrtMemcpy(zHost, outputByteSize, zDevice, outputByteSize, ACL_MEMCPY_DEVICE_TO_HOST); WriteFile("./output/output.bin", zHost, outputByteSize); aclrtFree(xDevice); aclrtFree(yDevice); aclrtFree(zDevice); aclrtFreeHost(xHost); aclrtFreeHost(yHost); aclrtFreeHost(zHost); aclrtDestroyStream(stream); aclrtResetDevice(deviceId); aclFinalize(); return 0; }

调用链要点如下：

• ACL 运行环境初始化：依次执行aclInit(nullptr)、aclrtSetDevice(deviceId)、aclrtCreateStream(&stream)，创建 device 与 stream；
• 内存分配与数据准备：通过aclrtMallocHost在 host 侧分配输入输出缓冲，通过aclrtMalloc（ACL_MEM_MALLOC_HUGE_FIRST）在 device 侧分配显存；从./input/input_x.bin、./input/input_y.bin读取输入，再用aclrtMemcpy以ACL_MEMCPY_HOST_TO_DEVICE方向拷贝到 device；
• 内核调用：使用内核调用符<<<>>>调用核函数。add_custom<<<numBlocks, 0, stream>>>(xDevice, yDevice, zDevice, totalLength)中，第一个参数numBlocks = 8指定启动 8 个核并行执行，运行时参数依次传入 Device 侧 x、y、z 张量地址和总数据长度totalLength；
• 结果回读与输出：aclrtSynchronizeStream(stream)等待内核执行完成，aclrtMemcpy以ACL_MEMCPY_DEVICE_TO_HOST方向回拷结果，最后WriteFile将结果写入./output/output.bin；
• 资源释放：依次aclrtFree/aclrtFreeHost释放显存与 host 内存，aclrtDestroyStream销毁流，aclrtResetDevice复位设备，aclFinalize结束 ACL 环境。

数据读写依赖 data_utils.h 中提供的ReadFile与WriteFile两个工具函数：ReadFile通过stat校验文件存在性、S_ISREG校验普通文件类型，并以二进制方式读入指定缓冲；WriteFile以O_RDWR | O_CREAT | O_TRUNC方式创建/截断文件后写入指定字节数，二者均带错误日志与返回值校验，便于快速定位文件读写问题。

数据生成与精度验证

数据生成脚本

scripts/gen_data.py 使用 NumPy 随机生成两个形状为 [8, 2048] 的 float32 输入张量，并同步计算真值：

input_x = np.random.uniform(1, 10, [8, 2048]).astype(np.float32) input_y = np.random.uniform(1, 10, [8, 2048]).astype(np.float32) golden = (input_x + input_y).astype(np.float32) os.makedirs("input", exist_ok=True) os.makedirs("output", exist_ok=True) input_x.tofile("./input/input_x.bin") input_y.tofile("./input/input_y.bin") golden.tofile("./output/golden.bin")

脚本会在样例目录下创建input与output目录，生成input/input_x.bin、input/input_y.bin两份输入数据，以及output/golden.bin真值文件（float32 二进制，与 device 侧内存布局一一对应）。

结果验证脚本

scripts/verify_result.py 将output/output.bin与output/golden.bin分别按 float32 读入并展平，使用np.isclose做逐元素比对，容差设置为：

• 相对误差容限RELATIVE_TOL = 1e-4
• 绝对误差容限ABSOLUTE_TOL = 1e-5
• 错误率容限ERROR_TOL = 1e-4

脚本会打印不一致元素的索引、期望值、实际值与相对偏差（最多打印 100 个），并统计错误比例。当error_ratio <= ERROR_TOL时判定通过并输出test pass!，否则打印[ERROR] result error并以非零码退出。

编译与运行

在样例根目录下执行如下步骤，编译并执行样例。

配置环境变量

请根据当前环境上 CANN 开发套件包的安装方式，配置环境变量：

source ${install_path}/cann/set_env.sh

说明：${install_path}为 CANN 包安装目录，未指定安装目录时默认安装至/usr/local/Ascend下。

样例执行

在样例目录下执行如下命令：

mkdir -p build && cd build; # 创建并进入build目录 cmake -DCMAKE_ASC_ARCHITECTURES=dav-2201 ..;make -j; # 编译工程（默认npu模式） python3 ../scripts/gen_data.py # 生成测试输入数据 ./demo # 执行编译生成的可执行程序，执行样例 python3 ../scripts/verify_result.py output/output.bin output/golden.bin # 验证输出结果是否正确，确认算法逻辑正确

使用 CPU 调试或 NPU 仿真模式时，添加-DCMAKE_ASC_RUN_MODE=cpu或-DCMAKE_ASC_RUN_MODE=sim参数即可：

cmake -DCMAKE_ASC_RUN_MODE=cpu -DCMAKE_ASC_ARCHITECTURES=dav-2201 ..;make -j; # cpu调试模式 cmake -DCMAKE_ASC_RUN_MODE=sim -DCMAKE_ASC_ARCHITECTURES=dav-2201 ..;make -j; # NPU仿真模式

注意：切换编译模式前需清理 cmake 缓存，可在 build 目录下执行rm CMakeCache.txt后重新 cmake。

编译选项说明

从 CMakeLists.txt 可以看到工程的构建细节：项目以project(kernel_samples LANGUAGES ASC CXX)声明 ASC（Ascend C 源码）与 CXX 两种语言，通过find_package(ASC REQUIRED)引入 CANN 的 ASC 编译工具链；add_executable(demo add_tpipe_tque.asc)将.asc文件直接作为可执行程序源码编译，并借助生成器表达式$<$<COMPILE_LANGUAGE:ASC>:--npu-arch=${CMAKE_ASC_ARCHITECTURES}>将dav-2201（或指定的dav-3510）架构参数透传给 ASC 编译器。

执行结果

执行结果如下，说明精度对比成功：

test pass!

与静态 Tensor 实现的对比与延伸

在 01_add 目录 下，Add 算子存在两种实现：基于静态 Tensor 编程的 add 样例 与基于 TQue/TPipe 编程的本样例。对比二者可以直观看到：

• 静态 Tensor 方式需要开发者显式通过LocalMemAllocator在 UB 上分配空间，并手动插入PipeBarrier<PIPE_ALL>()完成搬入—计算—搬出各阶段的流水同步；
• 本样例的 TQue/TPipe 方式则将 UB 缓冲申请（InitBuffer/AllocTensor）与阶段间同步（EnQue/DeQue）内聚到队列机制中，代码结构更贴近"生产—消费"的流水视角，也为后续引入多缓冲（如双缓冲 Ping-Pong）等性能优化范式打下了基础。

对于希望继续深入向量算子性能优化的读者，可以进一步参考仓库中 2_Performance 目录下的系列性能演进样例（如 gelu、softmax、rms_norm 等 story），它们展示了从本类基础范式出发，逐步叠加多核切分、双缓冲、向量函数（VF）融合等优化手段的完整过程。

总结

add_tpipe_tque 样例虽小，却完整覆盖了 Ascend C 队列式编程的全部关键要素：多核数据切分（GetBlockNum/GetBlockIdx）、GM/UB 两级数据流转（GlobalTensor/LocalTensor/DataCopy）、TPipe 统一内存管理（InitBuffer/AllocTensor/FreeTensor）以及 TQue 阶段同步（EnQue/DeQue）。配合 scripts 目录 下的数据生成与精度验证脚本，它同时也是一个可一键编译运行、可量化验证结果正确性的最小可复现工程，非常适合作为 Ascend C 向量算子的入门模板。

【免费下载链接】cann-samplesCANN高性能实战演进样例与体系化调优知识库项目地址: https://gitcode.com/cann/cann-samples