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 CSIMD 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 软件版本对应关系如下产品CANN 软件版本Ascend 950PR / Ascend 950DT CANN 9.1.0Atlas A3 训练系列产品 / Atlas A3 推理系列产品 CANN 9.0.0Atlas A2 训练系列产品 / Atlas A2 推理系列产品 CANN 9.0.0样例规格样例类型OpTypeAdd输入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 向量编程中数据主要分布在两级存储上GMGlobal MemoryAI Core 外部的全局内存容量大但访问速度慢通过GlobalTensor访问UBUnified BufferAI Core 内部向量计算专用缓存容量有限但访问速度快通过LocalTensor访问。向量计算单元只能访问 UB 上的数据因此算子内核必然遵循搬入—计算—搬出三段式结构先用DataCopy将输入从 GM 搬到 UB在 UB 内完成向量计算再将结果从 UB 搬回 GM。TPipe内存与同步的统一管理者TPipePipeline是队列式编程中管理片上内存与流水同步的核心对象。在本样例中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::TQueAscendC::TPosition::VECIN, 1 inQueueX; AscendC::TQueAscendC::TPosition::VECIN, 1 inQueueY; AscendC::TQueAscendC::TPosition::VECOUT, 1 outQueueZ;模板参数中TPosition::VECIN/TPosition::VECOUT表示队列对应输入vector 搬入或输出vector 搬出位置第二个参数表示队列深度。队列的典型使用流程为AllocTensorT()从队列申请一个LocalTensor即从TPipe管理的 UB 缓冲中取出一块DataCopy完成 GM 与 UB 之间的数据搬运EnQue将已就绪的LocalTensor入队作为生产者通知后续阶段数据可用DeQue后续阶段从队列取出LocalTensor继续处理作为消费者等待数据就绪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::TQueAscendC::TPosition::VECIN, 1 inQueueX; AscendC::TQueAscendC::TPosition::VECIN, 1 inQueueY; AscendC::TQueAscendC::TPosition::VECOUT, 1 outQueueZ; AscendC::GlobalTensorfloat xGm; AscendC::GlobalTensorfloat yGm; AscendC::GlobalTensorfloat 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::LocalTensorfloat xLocal inQueueX.AllocTensorfloat(); AscendC::LocalTensorfloat yLocal inQueueY.AllocTensorfloat(); AscendC::DataCopy(xLocal, xGm, blockLength); AscendC::DataCopy(yLocal, yGm, blockLength); inQueueX.EnQue(xLocal); inQueueY.EnQue(yLocal); // DeQue 取出输入张量在 UB 内执行 Add并将结果 EnQue 到输出队列供后续写回 GM。 xLocal inQueueX.DeQuefloat(); yLocal inQueueY.DeQuefloat(); AscendC::LocalTensorfloat zLocal outQueueZ.AllocTensorfloat(); AscendC::Add(zLocal, xLocal, yLocal, blockLength); outQueueZ.EnQuefloat(zLocal); inQueueX.FreeTensor(xLocal); inQueueY.FreeTensor(yLocal); // 从输出队列取出结果并写回当前核负责的 GM 分片。 zLocal outQueueZ.DeQuefloat(); AscendC::DataCopy(zGm, zLocal, blockLength); outQueueZ.FreeTensor(zLocal); }处理流程梳理对照源码本样例的处理流程为add_custom作为核入口接收totalLength通过GetBlockNum()计算当前 block 的数据长度blockLength通过GetBlockIdx()计算当前核在 GM 中对应的数据起点使用DataCopy把输入数据从 GM 搬到 UB并通过EnQue将输入LocalTensor放入输入队列通过DeQue从输入队列取出输入张量在 UB 中执行Add再通过EnQue将结果LocalTensor放入输出队列通过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_customnumBlocks, 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 侧分配输入输出缓冲通过aclrtMallocACL_MEM_MALLOC_HUGE_FIRST在 device 侧分配显存从./input/input_x.bin、./input/input_y.bin读取输入再用aclrtMemcpy以ACL_MEMCPY_HOST_TO_DEVICE方向拷贝到 device内核调用使用内核调用符调用核函数。add_customnumBlocks, 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_okTrue) os.makedirs(output, exist_okTrue) 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_ARCHITECTURESdav-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_MODEcpu或-DCMAKE_ASC_RUN_MODEsim参数即可cmake -DCMAKE_ASC_RUN_MODEcpu -DCMAKE_ASC_ARCHITECTURESdav-2201 ..;make -j; # cpu调试模式 cmake -DCMAKE_ASC_RUN_MODEsim -DCMAKE_ASC_ARCHITECTURESdav-2201 ..;make -j; # NPU仿真模式注意切换编译模式前需清理 cmake 缓存可在 build 目录下执行rm CMakeCache.txt后重新 cmake。编译选项说明选项可选值说明CMAKE_ASC_RUN_MODEnpu默认、cpu、sim运行模式NPU 运行、CPU 调试、NPU 仿真CMAKE_ASC_ARCHITECTURESdav-2201默认、dav-3510NPU 架构dav-2201 对应 Atlas A2 训练系列产品/Atlas A2 推理系列产品和 Atlas A3 训练系列产品/Atlas A3 推理系列产品dav-3510 对应 Ascend 950PR/Ascend 950DT从 CMakeLists.txt 可以看到工程的构建细节项目以project(kernel_samples LANGUAGES ASC CXX)声明 ASCAscend 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 上分配空间并手动插入PipeBarrierPIPE_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
CANN Ascend C 向量编程入门：基于 TPipe 与 TQue 队列机制实现 Add 向量加法算子
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 CSIMD 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 软件版本对应关系如下产品CANN 软件版本Ascend 950PR / Ascend 950DT CANN 9.1.0Atlas A3 训练系列产品 / Atlas A3 推理系列产品 CANN 9.0.0Atlas A2 训练系列产品 / Atlas A2 推理系列产品 CANN 9.0.0样例规格样例类型OpTypeAdd输入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 向量编程中数据主要分布在两级存储上GMGlobal MemoryAI Core 外部的全局内存容量大但访问速度慢通过GlobalTensor访问UBUnified BufferAI Core 内部向量计算专用缓存容量有限但访问速度快通过LocalTensor访问。向量计算单元只能访问 UB 上的数据因此算子内核必然遵循搬入—计算—搬出三段式结构先用DataCopy将输入从 GM 搬到 UB在 UB 内完成向量计算再将结果从 UB 搬回 GM。TPipe内存与同步的统一管理者TPipePipeline是队列式编程中管理片上内存与流水同步的核心对象。在本样例中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::TQueAscendC::TPosition::VECIN, 1 inQueueX; AscendC::TQueAscendC::TPosition::VECIN, 1 inQueueY; AscendC::TQueAscendC::TPosition::VECOUT, 1 outQueueZ;模板参数中TPosition::VECIN/TPosition::VECOUT表示队列对应输入vector 搬入或输出vector 搬出位置第二个参数表示队列深度。队列的典型使用流程为AllocTensorT()从队列申请一个LocalTensor即从TPipe管理的 UB 缓冲中取出一块DataCopy完成 GM 与 UB 之间的数据搬运EnQue将已就绪的LocalTensor入队作为生产者通知后续阶段数据可用DeQue后续阶段从队列取出LocalTensor继续处理作为消费者等待数据就绪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::TQueAscendC::TPosition::VECIN, 1 inQueueX; AscendC::TQueAscendC::TPosition::VECIN, 1 inQueueY; AscendC::TQueAscendC::TPosition::VECOUT, 1 outQueueZ; AscendC::GlobalTensorfloat xGm; AscendC::GlobalTensorfloat yGm; AscendC::GlobalTensorfloat 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::LocalTensorfloat xLocal inQueueX.AllocTensorfloat(); AscendC::LocalTensorfloat yLocal inQueueY.AllocTensorfloat(); AscendC::DataCopy(xLocal, xGm, blockLength); AscendC::DataCopy(yLocal, yGm, blockLength); inQueueX.EnQue(xLocal); inQueueY.EnQue(yLocal); // DeQue 取出输入张量在 UB 内执行 Add并将结果 EnQue 到输出队列供后续写回 GM。 xLocal inQueueX.DeQuefloat(); yLocal inQueueY.DeQuefloat(); AscendC::LocalTensorfloat zLocal outQueueZ.AllocTensorfloat(); AscendC::Add(zLocal, xLocal, yLocal, blockLength); outQueueZ.EnQuefloat(zLocal); inQueueX.FreeTensor(xLocal); inQueueY.FreeTensor(yLocal); // 从输出队列取出结果并写回当前核负责的 GM 分片。 zLocal outQueueZ.DeQuefloat(); AscendC::DataCopy(zGm, zLocal, blockLength); outQueueZ.FreeTensor(zLocal); }处理流程梳理对照源码本样例的处理流程为add_custom作为核入口接收totalLength通过GetBlockNum()计算当前 block 的数据长度blockLength通过GetBlockIdx()计算当前核在 GM 中对应的数据起点使用DataCopy把输入数据从 GM 搬到 UB并通过EnQue将输入LocalTensor放入输入队列通过DeQue从输入队列取出输入张量在 UB 中执行Add再通过EnQue将结果LocalTensor放入输出队列通过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_customnumBlocks, 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 侧分配输入输出缓冲通过aclrtMallocACL_MEM_MALLOC_HUGE_FIRST在 device 侧分配显存从./input/input_x.bin、./input/input_y.bin读取输入再用aclrtMemcpy以ACL_MEMCPY_HOST_TO_DEVICE方向拷贝到 device内核调用使用内核调用符调用核函数。add_customnumBlocks, 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_okTrue) os.makedirs(output, exist_okTrue) 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_ARCHITECTURESdav-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_MODEcpu或-DCMAKE_ASC_RUN_MODEsim参数即可cmake -DCMAKE_ASC_RUN_MODEcpu -DCMAKE_ASC_ARCHITECTURESdav-2201 ..;make -j; # cpu调试模式 cmake -DCMAKE_ASC_RUN_MODEsim -DCMAKE_ASC_ARCHITECTURESdav-2201 ..;make -j; # NPU仿真模式注意切换编译模式前需清理 cmake 缓存可在 build 目录下执行rm CMakeCache.txt后重新 cmake。编译选项说明选项可选值说明CMAKE_ASC_RUN_MODEnpu默认、cpu、sim运行模式NPU 运行、CPU 调试、NPU 仿真CMAKE_ASC_ARCHITECTURESdav-2201默认、dav-3510NPU 架构dav-2201 对应 Atlas A2 训练系列产品/Atlas A2 推理系列产品和 Atlas A3 训练系列产品/Atlas A3 推理系列产品dav-3510 对应 Ascend 950PR/Ascend 950DT从 CMakeLists.txt 可以看到工程的构建细节项目以project(kernel_samples LANGUAGES ASC CXX)声明 ASCAscend 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 上分配空间并手动插入PipeBarrierPIPE_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
