CANN Runtime 示例解析基于 aclrtLaunchHostFunc 的 Stream HostFunc 回调下发与执行顺序控制【免费下载链接】runtime本项目提供CANN运行时组件和维测功能组件。项目地址: https://gitcode.com/cann/runtime导读本文以 CANN runtime 开源仓库中的 1_callback_hostfunc 示例为切入点完整讲解如何在 Stream 上通过aclrtLaunchHostFunc接口下发一个 Host 侧回调函数。该 Host 函数会在当前 Stream 上已下发的任务全部执行完之后被调用并会阻塞之后新添加的任务从而提供一种轻量级的主机侧任务锚点机制——无需显式创建线程即可在设备任务流中插入主机侧逻辑。读完本文你将掌握该示例的编译运行方法、核心 API 调用序列、回调函数的签名约束、遇错即停的 Stream 失效模式配置以及从 AscendCL 到 Runtime 底层实现的关键调用链。示例概述与能力定位样例要解决的问题在异构计算的典型场景中用户经常需要在设备上跑完一批算子任务之后再执行一段主机侧Host逻辑例如记录日志、检查中间结果、推进流水线状态。传统做法是手动创建线程去监听 Stream 完成状态代码繁琐且难以保证与 Stream 任务严格同步。本示例演示的正是 CANN Runtime 提供的替代方案把一个 Host 侧函数作为一个任务挂到 Stream 上。它的执行语义非常明确在已下发的任务全部执行完成之后该 Host 函数才会被调用在它之后添加的任务会被阻塞直到该 Host 函数执行结束。由此HostFunc 回调天然成为 Stream 任务序列中的一个有序节点前后任务之间形成严格的 happens-before 关系可用于设备任务 → 主机处理 → 后续设备任务的编排。从仓库目录结构看回调能力在 example/2_advanced_features/callback 下共有三个递进的样例0_simple_callback演示 Report 回调和 HostFunc 处理线程的注册执行1_callback_hostfunc本文主题演示 Stream 上 HostFunc 回调的下发与执行顺序2_callback_exception演示异常回调和 Runtime 错误信息查询。产品支持情况原文档明确列出了该样例在以下产品上的支持情况编译与运行前请先确认目标环境属于下表范围产品是否支持Ascend 950PR/Ascend 950DT√Atlas A3 训练系列产品/Atlas A3 推理系列产品√Atlas A2 训练系列产品/Atlas A2 推理系列产品√样例文件组成文件作用main.cpp示例主程序初始化、建流、下发算子与 HostFunc 回调、同步、清理资源CMakeLists.txt构建脚本通过ascendc_library编译核函数并链接主程序run.sh一键脚本cmake 配置、编译、安装并运行输出保存到output_msg.txtREADME.md中文说明本文主体README_en.md英文说明环境准备与编译运行编译运行步骤原文档给出的运行步骤如下其中${install_root}需替换为 CANN 安装根目录默认安装在/usr/local/Ascend${git_clone_path}为本仓库的克隆路径# ${install_root} 替换为 CANN 安装根目录默认安装在/usr/local/Ascend目录 source ${install_root}/cann/set_env.sh # 自动识别 SOC_VERSION 和 ASCENDC_CMAKE_DIR source ${git_clone_path}/example/set_sample_env.sh # 编译运行 bash run.sh环境脚本做了什么source ${install_root}/cann/set_env.sh用于加载 CANN 的基础环境变量。第二步source example/set_sample_env.sh则是本仓库为样例提供的自动环境探测脚本它做了三件关键事情详见 example/set_sample_env.sh自动识别SOC_VERSION脚本会调用仓库自带的tools/get_soc_version辅助程序通过 get_soc_version.cpp 编译而成借助aclrtGetSocName这类 ACL 接口探测当前设备的 SoC 版本号再通过正则^[A-Za-z0-9_-]$校验其合法性自动定位ASCENDC_CMAKE_DIR在${cann_path}/arch/tikcpp/ascendc_kernel_cmake、${cann_path}/aarch64-linux/tikcpp/ascendc_kernel_cmake等候选目录中查找ascendc.cmake文件命中即作为核函数编译框架目录导出环境变量最终在当前 Shell 中导出ASCEND_INSTALL_PATH、ASCEND_HOME_PATH、SOC_VERSION、ASCENDC_CMAKE_DIR四个变量供 CMake 构建使用。需要说明的是set_sample_env.sh对环境布局有探测逻辑如include/acl/acl.h与lib64/libacl_rt.so的存在性校验因此要求已安装的 CANN 包布局完整。构建脚本拆解run.sh 的执行流程为source $_ASCEND_INSTALL_PATH/bin/setenv.bash # 加载 CANN 编译环境 cmake -B build -DASCEND_CANN_PACKAGE_PATH${_ASCEND_INSTALL_PATH} # 配置 cmake --build build -j # 编译 cmake --install build # 安装 ./build/main | tee output_msg.txt # 运行并把输出落盘其中set -e保证任何一步失败即退出避免在错误环境下继续执行。最终的运行输出会同时打印到终端并写入output_msg.txt。CMake 配置要点CMakeLists.txt 中有两个值得注意的点通过include(${ASCENDC_CMAKE_DIR}/ascendc.cmake)引入 AscendC 核函数编译框架并用ascendc_library(kernels STATIC ../../../kernel_func/easy_OP.cpp)将核函数源文件编译为静态库kernels主程序main链接kernels与pthread其中pthread正是为回调监听线程机制预留的依赖。核函数本体在 example/kernel_func/easy_OP.cpp 中是一个让输入x自增 1 的简单算子extern C __global__ __aicore__ void EasyOPf(__gm__ uint32_t* x) { int32_t idx block_idx; x[idx] 1; // 部分架构如 NPU 架构号 3510需要显式刷新缓存行 } void EasyOP(uint32_t blockDim, void* stream, uint32_t* x) { EasyOPfblockDim, nullptr, stream(x); }主程序逐步解析main.cpp 完整展示了初始化 → 建流 → 下发算子 → 下发 HostFunc → 同步 → 校验结果 → 清理的标准流程。下面按阶段拆解。1. 回调函数定义namespace { void CallBackFunc(void* arg) { INFO_LOG(Hostfunc callback!!!); } } // namespace回调函数签名必须严格符合aclrtHostFunc类型。在 include/external/acl/acl_rt.h#L391 中其定义如下typedef void (*aclrtHostFunc)(void* args);即入参是一个void*由下发时传入的args透传无返回值。本样例不需要携带用户数据因此下发时传nullptr。2. 初始化与 Device/Context 管理CHECK_ERROR(aclInit(nullptr)); // 初始化 AscendCL 配置 CHECK_ERROR(aclrtSetDevice(deviceId)); // 指定用于运算的 Device CHECK_ERROR(aclrtCreateContext(context, deviceId)); // 创建 Context其中CHECK_ERROR宏来自 example/utils.h它把返回值aclError与ACL_SUCCESS比对失败时打印出错的调用表达式与错误码并返回-1#define CHECK_ERROR(call) \ do { \ aclError __ret (call); \ if (__ret ! ACL_SUCCESS) { \ ERROR_LOG(Operation failed: %s returned error code %d, #call, (int32_t)__ret); \ return -1; \ } \ } while (0)3. 内存申请与数据搬运CHECK_ERROR(aclrtMalloc((void**)numDevice, size, ACL_MEM_MALLOC_HUGE_FIRST)); // 申请 Device 内存 CHECK_ERROR(aclrtMemcpy(numDevice, size, num, size, ACL_MEMCPY_HOST_TO_DEVICE)); // H2D 拷贝aclrtMalloc使用ACL_MEM_MALLOC_HUGE_FIRST策略申请一块 Device 侧uint32_t内存并把主机侧变量num 0的值通过aclrtMemcpy拷贝到 Device 侧。后续核函数正是在这块内存上执行自增。4. 创建 Stream 并设置遇错即停CHECK_ERROR(aclrtCreateStream(stream)); // 创建 Stream CHECK_ERROR(aclrtSetStreamFailureMode(stream, ACL_STOP_ON_FAILURE)); // 设置为遇错即停aclrtSetStreamFailureMode用于设置 Stream 上任务执行遇到错误时的行为默认是遇错继续可切换为遇错即停。在 include/external/acl/acl_rt.h#L49 中#define ACL_STOP_ON_FAILURE 0x00000001U一旦设置为ACL_STOP_ON_FAILUREStream 上某个任务执行出错后后续任务将不再执行。这在调试期非常有用可以保证 HostFunc 回调不会在错误状态下被触发避免掩盖上游故障。该接口的声明位于 include/external/acl/acl_rt.h#L1644。5. 下发算子任务与 HostFunc 回调EasyOP(blockDim, stream, numDevice); // 通过 Stream 下发核函数任务 CHECK_ERROR(aclrtLaunchHostFunc(stream, CallBackFunc, nullptr)); // 下发 Host 侧回调 CHECK_ERROR(aclrtSynchronizeStream(stream)); // 阻塞等待 Stream 上任务完成这里的顺序关系正是本样例的核心语义EasyOP先在 Stream 上排入一个让numDevice自增的算子任务紧接着aclrtLaunchHostFunc把CallBackFunc排到该算子之后。根据 HostFunc 的执行语义回调一定在已下发任务此处即自增算子执行完之后才被调用同时阻塞其后新增的任务。6. 校验结果与资源清理CHECK_ERROR(aclrtMemcpy(num, size, numDevice, size, ACL_MEMCPY_DEVICE_TO_HOST)); // D2H 拷贝 INFO_LOG(After assigning the task through the created stream, the current result is: %d., num); CHECK_ERROR(aclrtFree(numDevice)); // 释放 Device 内存 CHECK_ERROR(aclrtDestroyStreamForce(stream)); // 强制销毁 Stream丢弃所有任务 CHECK_ERROR(aclrtDestroyContext(context)); // 销毁 Context CHECK_ERROR(aclrtResetDeviceForce(deviceId)); // 强制复位 Device回收资源 aclFinalize(); // AscendCL 去初始化 INFO_LOG(Resource cleanup completed.);同步完成后再把 Device 侧的结果拷回主机并打印若执行成功num的值应为 1初始 0 自增 1这从结果上间接验证了算子任务先于同步点完成。收尾阶段使用aclrtDestroyStreamForce与aclrtResetDeviceForce这类强制接口含义是丢弃所有任务、立即回收资源适用于样例这类无需保留任务状态的场景。aclrtLaunchHostFunc 的底层实现链路AscendCL 层的参数校验与转调aclrtLaunchHostFunc在 AscendCL 层的实现位于 src/acl/aclrt_impl/callback.cpp#L181-L188aclError aclrtLaunchHostFuncImpl(aclrtStream stream, aclrtHostFunc fn, void* args) { ACL_PROFILING_REG(acl::AclProfType::AclrtLaunchHostFunc); // 打点供 Profiling 采集 ACL_LOG_INFO(start to execute aclrtLaunchHostFunc.); ACL_REQUIRES_RTS_OK(rtsLaunchHostFunc(static_castrtStream_t(stream), static_castrtCallback_t(fn), args)); ACL_LOG_INFO(successfully execute aclrtLaunchHostFunc); return ACL_SUCCESS; }可以看到AscendCL 层做的主要工作是注册 Profiling 事件ACL_PROFILING_REG事件类型AclrtLaunchHostFunc可在 src/acl/aclrt_impl/toolchain/profiling_manager.cpp 中查证、打印日志然后把用户回调强转为 Runtime 层回调类型rtCallback_t并转调rtsLaunchHostFunc。接口的对外声明与注释在 include/external/acl/acl_rt.h#L5393-L5404注释明确该接口的功能是Enqueues a host function call in a stream。Runtime 层的任务入流在 Runtime 层src/runtime/api/api_c.cc#L3973-L3978 的rtsLaunchHostFunc进一步把调用分发到运行时实例rtError_t rtsLaunchHostFunc(rtStream_t stm, const rtCallback_t callBackFunc, void* const fnData) { auto* const exeStream static_castStream*(stm); const rtError_t error apiInstance-LaunchHostFunc(exeStream, callBackFunc, fnData); ... }LaunchHostFunc是运行时实例apiInstance的虚接口定义于 src/runtime/api/api.hpp#L872-L873且旁边还有LaunchHostFuncV2接受rtHostCpuFunc类型回调的增强版本说明 HostFunc 能力在运行时层面同时保留了基础版与 V2 两条通道。实现上还经过装饰器ApiDecorator见 src/runtime/api/impl/api_decorator.cc与错误码装饰器ApiErrorDecorator见 src/runtime/api/impl/api_error.cc的包装分别承担日志埋点与错误码转换职责。从这条调用链可以提炼出三点工程事实回调线程由 Runtime 内部管理调用方无需也不应自行创建线程aclrtLaunchHostFunc会把 HostFunc 作为 Stream 上的一个任务节点排队其执行线程由 Runtime 内部机制这也是 CMakeLists.txt 需要链接pthread的原因之一统一调度执行顺序由 Stream 的任务队列保证先入队的算子任务执行完HostFunc 任务才会触发之后入队的任务被阻塞形成了严格的先后次序Profiling 原生可见调用会注册AclrtLaunchHostFunc类型的事件便于在 Profiling 数据中观测 HostFunc 的下发时机与执行耗时。运行结果与验证原文档给出的示例输出如下[INFO] Hostfunc callback!!! [INFO] After assigning the task through the created stream, the current result is: ... [INFO] Resource cleanup completed.结合主程序逻辑可以解读这三行输出的时序与含义第一行Hostfunc callback!!!来自CallBackFunc回调体它必然出现在算子任务执行完成之后、aclrtSynchronizeStream返回之前第二行打印的是num的最终值预期为 1证明自增算子在回调之前已真实生效间接验证了 HostFunc 与上游任务的执行顺序第三行Resource cleanup completed.由主程序末尾的清理流程打印标志 Device 内存、Stream、Context 均已回收aclFinalize已调用。关键 API 一览本样例涉及的关键功能点及对应接口整理如下与原文档一致并补充语义说明功能分类接口本样例中的用途初始化aclInit初始化 AscendCL 配置去初始化aclFinalize实现 AscendCL 去初始化Device 管理aclrtSetDevice指定用于运算的 DeviceDevice 管理aclrtResetDeviceForce强制复位当前运算的 Device回收 Device 上的资源Context 管理aclrtCreateContext创建 ContextContext 管理aclrtDestroyContext销毁 ContextStream 管理aclrtCreateStream创建 StreamStream 管理aclrtSynchronizeStream阻塞等待 Stream 上任务的完成Stream 管理aclrtSetStreamFailureMode设置 Stream 执行任务遇错时的行为默认为遇错继续可设置为遇错即停Stream 管理aclrtDestroyStreamForce强制销毁 Stream丢弃所有任务回调控制aclrtLaunchHostFunc直接触发回调无需显式创建线程内存管理aclrtMalloc申请 Device 上的内存内存管理aclrtFree释放 Device 上的内存数据传输aclrtMemcpy通过内存复制的方式实现 H2D / D2H 数据传输延伸阅读与注意事项若希望进一步了解 Report 回调与 HostFunc 处理线程的注册执行可阅读同目录的 0_simple_callback 样例若关注回调失败后的错误处理可参考 2_callback_exception 样例回调函数体内不宜执行过于耗时的阻塞操作因为按执行语义HostFunc 之后入队的任务会被阻塞等待长时间运行的回调会拖慢整个 Stream 的吞吐ACL_STOP_ON_FAILURE模式适用于希望出错即停、避免连带失败的场景默认的遇错继续模式则更贴合需要尽量跑完任务的流水线场景环境变量与安装布局请以本机实际 CANN 安装为准set_sample_env.sh若探测失败例如找不到acl.h或libacl_rt.so请先确认set_env.sh是否已正确加载。已知 issue截至当前仓库版本该样例暂无已知 issue。【免费下载链接】runtime本项目提供CANN运行时组件和维测功能组件。项目地址: https://gitcode.com/cann/runtime创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考