CANN opbase 的 AI CPU 任务封装与启动接口解析：aicpu_task 保留接口的完整使用指南

CANN opbase 的 AI CPU 任务封装与启动接口解析：aicpu_task 保留接口的完整使用指南

【免费下载链接】opbase本项目是CANN算子库的基础框架库，为算子提供公共依赖文件和基础调度能力。项目地址: https://gitcode.com/cann/opbase

本文围绕 CANN opbase 算子库中op::internal命名空间下的aicpu_task接口族展开，系统讲解 AI CPU（AI CPU）算子从任务封装、参数组装、任务复用缓存到内核启动的完整链路。该接口族被ADD_TO_LAUNCHER_LIST_AICPU等宏和 AI CPU 算子调度框架内部使用，阅读本文后，你将理解AicpuTask、AicpuTfTask、AicpuCCTask、AicpuTaskSpace的职责划分，掌握任务键（task key）的生成与命中判定逻辑、属性与张量列表的组装方式，以及环境变量ACLNN_CACHE_LIMIT、GE_PROFILING_TO_STD_OUT对任务缓存与耗时打点的影响，从而具备阅读、排查和扩展 AI CPU 算子执行路径的实战能力。

接口定位：AI CPU 任务调度链路的保留接口

aicpu_task中声明的 API 属于op::internal内部命名空间，官方文档明确标注为保留接口（reserved interface），未来可能被修改或废弃，不建议业务代码直接依赖。但从源码看，它们正是 CANN opbase 中 AI CPU 算子"一次封装、多次复用、按需启动"的核心实现：

• aicpu_task.h 中kAicpuKeyBufLen = 1024U定义了任务键缓冲区的固定长度，kDefaultFunctionName = "RunCpuKernel"是 AI CPU 算子的默认入口函数名；
• 任务类的Init/Run虚函数接口（aicpu_task.h）分别完成"参数与二进制装载"和"地址刷新与内核下发"两件事；
• 任务容器AicpuTaskSpace以"算子类型 + 输入元信息 + 属性取值"生成哈希键，实现同构任务的复用。

由于这些接口承担的是框架内部调度职责，本文在讲解时会同时给出接口语义与源码实现证据两条线，方便你在阅读算子内核代码或调试 AI CPU 执行链路时对号入座。

下表完整列出该接口族提供的全部 API（继承自原文档，共 57 项），后续章节将按功能分组深入讲解：

任务实体：AicpuTask 类族及其职责划分

抽象基类 AicpuTask

AicpuTask是 AI CPU 任务的抽象基类，管理任务的封装、初始化、下发与执行相关参数与方法。其完整定义见 AicpuTask 类文档 与 aicpu_task.h。文档中的 API 表列出了"默认实例构造AicpuTask()"，而当前仓库源码中实际提供的是带参构造：

AicpuTask(const std::string& opType, const ge::UnknowShapeOpType unknownType) : opType_(opType), unknownType_(unknownType) {}

该构造将算子类型opType_与动态 Shape 类型unknownType_绑定为任务的固有属性。基类声明了两个纯虚函数，是子类必须实现的契约：

• virtual aclnnStatus Init(const FVector<const aclTensor*>& inputs, const FVector<aclTensor*>& outputs, const AicpuAttrs& attrs) = 0：完成参数缓冲区构建、扩展信息解析与二进制装载；
• virtual aclnnStatus Run(aclOpExecutor* executor, aclrtStream stream) = 0：在指定 stream 上刷新输入输出地址并下发内核。

基类成员中值得关注的是任务键相关字段：

其中inputKey_是任务复用的核心依据：任务创建时会把"线程 ID + 输入张量元信息 + 属性取值"序列化到该缓冲区，后续相同特征的调用通过比对keyLen_与逐字节内容命中同一任务。

两个具体子类：AicpuTfTask 与 AicpuCCTask

AicpuTfTask面向TensorFlow 第三方框架算子（isTf = true时创建），AicpuCCTask面向CANN 自研算子。二者在 aicpu_task.cpp 中的实现差异体现在参数格式与二进制来源上：

AicpuCCTask::GetKernelNameAndSoName（aicpu_task.cpp）展示了算子内核库的选择逻辑：先查询自定义算子注册表（FindAndGetInCustomRegistry），命中则加载自定义内核库并将functionName_设为注册表中的函数名，同时将kernelSoName固定为"custom_sub_repository.so"以保证与 Device 侧一致；未命中则从内置算子信息库加载，其中{"Index", "IndexPut", "LeftShift"}三个 PyTorch 算子（aicpu_task.cpp）使用libpt_kernels.so。

为了兼容旧版本 opp_kernel 与 runtime 包，源码中还维护了两张兼容白名单（aicpu_task.cpp）：tfCompatibleOps（如 ResizeBilinear、Roll、Mean 等 8 个算子）与aicpuCompatibleOps（如 Pad、ReduceSum、Sigmoid 等 10 个算子）。当算子命中白名单或算子信息库不支持新启动接口时，任务会回退到旧流程rtAicpuKernelLaunchExWithArgs，否则优先使用新接口aclrtLaunchKernelWithHostArgs（见AicpuTfTask::Run/AicpuCCTask::Run）。这类兼容细节说明：这些保留接口在真实调度中承担了新旧两代 runtime 启动方式的自动适配。

任务容器与复用机制：AicpuTaskSpace

AicpuTaskSpace管理 AI CPU 任务的复用逻辑，包括任务创建与检索。其构造参数决定了任务容器面向的算子类别：

AicpuTaskSpace(const std::string& opType, const ge::UnknowShapeOpType unknownType = ge::DEPEND_IN_SHAPE, const bool isTf = false)

• opType：算子名；
• unknownType：动态 Shape 类型，默认ge::DEPEND_IN_SHAPE（CANN 自研第一类动态 Shape 算子），若为ge::DEPEND_SHAPE_RANGE则属于需 Device 侧扩展信息回传输出 Shape 的第三类算子；
• isTf：是否第三方框架算子，true时创建AicpuTfTask，false时创建AicpuCCTask。

容器内部使用std::unordered_map<size_t, std::vector<std::unique_ptr<AicpuTask>>>（aicpu_task.h）保存任务，哈希键由GenTaskKey生成。

GetOrCreateTask：查缓存、建任务、写缓存

GetOrCreateTask（aicpu_task_base.cpp）是任务复用的总入口，流程如下：

1. 用CreateTensorList从args的OP_INPUT_ARG参数构建输入张量列表；
2. 调用FindTask尝试命中缓存任务；
3. 未命中则加锁创建新任务：按isTf_实例化AicpuTfTask或AicpuCCTask，调用SetSpace将容器指针写回任务，用CreateAicpuAttrs组装属性表后执行Init；
4. 通过GenTaskKey生成任务键并拷贝到任务的inputKey_/keyLen_，标记isVisit_ = true；
5. 计算 Device 缓存大小并更新executor->workspaceDeviceAicpuMem_；
6. 若当前缓存任务总数低于上限kAicpuCacheLimit则入缓存hashMap_[seed].emplace_back(...)；若缓存已满，则把任务交给SavePendingOverflowTask暂存为一次性任务（one-shot），由AiCpuOneShotKernelLauncher消费。

FindTask：任务键比对与占用过滤

FindTask（aicpu_task_base.cpp）先按当前参数生成任务键，再到哈希表中查找；命中后还要求长度相等、逐字节一致，且该任务未被占用（!task->isVisit_）。命中后同样会累计 Device 缓存大小到executor->workspaceDeviceAicpuMem_，并将任务标记为isVisit_ = true防止并发复用。任务执行完（Run返回后）由AiCpuKernelLauncher在Launch或析构时调用SetVisit(false)归还（aicpu_kernel_launcher.h）。

缓存上限：ACLNN_CACHE_LIMIT 环境变量

任务缓存数量并非无限。ReadAicpuCacheLimit（aicpu_task_base.cpp）读取环境变量ACLNN_CACHE_LIMIT：

• 默认上限：10000 个任务；
• 最大值钳制：10000000 个任务；
• 非数字取值会被忽略并回退默认值（同时打印告警日志）。

超过上限后新任务不再入缓存，而是作为一次性任务立即下发，避免缓存无限增长造成内存膨胀。

引用型输入标记：SetRef / IsRef

SetRef(index, isInput)与IsRef(index, isInput)（aicpu_task_base.cpp）分别用于登记和查询指定下标是否为引用（ref）型输入/输出。注意SetRef只在hasInit_ == false（任务尚未初始化）时生效，容器一旦完成首次初始化，引用索引集合即被冻结。该机制服务于 in-place 类算子（如 IndexPut 中self既是输入又是输出的场景）的地址管理。

内存预算：CalcHostInputDataSize 与 CalcDeviceCacheSize

• CalcHostInputDataSize（aicpu_task_base.cpp）：遍历输入张量，仅统计TensorPlacement::kOnHost的输入，按CalcShapeBytes计算数据字节数并以alignBytes对齐后累加；
• CalcDeviceCacheSize（aicpu_task_base.cpp）：在DEPEND_SHAPE_RANGE场景下加上deviceExtMemSize_，并将 Host 输入缓存大小超过 1024 字节的部分计入 Device 预留内存。最终值汇总到executor->workspaceDeviceAicpuMem_，作为工作空间内存申请的依据。

任务键生成：Append1Byte、AppendAttrForKey 与 GenTaskKey

任务复用的核心是把"输入元信息 + 属性取值"序列化成可哈希的字节流。GenTaskKey（aicpu_task_base.cpp）按如下顺序拼装：

1. 线程 ID（GetTid()）；
2. 每个输入张量的GetDataType()与GetViewFormat()各占一个字节；
3. 分隔符'/'（0x2F）；
4. 全部属性参数（OP_ATTR_ARG）经GenKeyByAttrs序列化的字节流；

最终由GenHashBinary计算哈希种子作为容器哈希表键。

字节级基础工具

• Append1Byte(uint8_t *buf, uint8_t src)（aicpu_task.h）：写入一个字节并返回buf + 1，是键拼接的最基本操作；
• 模板版AppendAttrForKey(const V& value, ...)：按sizeof(value)逐字节追加任意简单类型；std::string及其指针版本逐字符追加；std::vector<V>版本逐元素递归追加；aclIntArray/aclFloatArray/aclBoolArray（含 const 与非 const 版本）则通过GetData()[i]逐元素追加。所有追加都受kAicpuKeyBufLen = 1024上限约束，达到上限即停止，防止缓冲区溢出。

哈希种子算法

GenHashBinary（aicpu_task_base.cpp）以 8 字节为单位读取键数据，使用std::hash<uint64_t>结合常量种子kHashSeed = 0x9e3779b9U做混合：

seed ^= hasher(*ptr) + kHashSeed + (seed << 6U) + (seed >> 2U);

尾部不足 8 字节的剩余数据会先拼接进一个uint64_t后参与最后一次混合。注意哈希键是概率性的，FindTask中命中哈希桶后还会做keyLen_与逐字节的精确比对，保证不会因哈希冲突误复用任务。

属性序列化：GenKeyByAttrs 与 CreateAicpuAttrs

• GenKeyByAttrsImpl（aicpu_task_base.cpp）按OpArg的类型标签（OPARG_INT_LIST、OPARG_FLOAT_LIST、OPARG_BOOL、OPARG_INT、OPARG_UINT、OPARG_FLOAT、OPARG_DOUBLE、OPARG_DATATYPE、OPARG_STRING等）分派到对应的AppendAttrForKey重载，用于任务键的生成；
• CreateAicpuAttrsImpl（aicpu_task_base.cpp）按同样类型分派到AddAicpuAttr，用于构造内核可见的属性表AicpuAttrs，二者共用一套 OpArg 类型体系，保证"键生成"与"参数生成"口径一致。

张量列表与属性组装：AppendTensor、CreateTensorList 与 AddAicpuAttr

AI CPU 算子入参既有aclTensor，也有aclScalar、aclIntArray、aclTensorList等非张量形态。AppendTensor的四个重载（aicpu_task.h）统一将它们归一化为张量：

CreateTensorListImpl（aicpu_task.h）根据OpArg::type（OPARG_ACLTENSOR/OPARG_ACLSCALAR/OPARG_INT_LIST/OPARG_ACLTENSOR_LIST）自动分派到上述重载，CreateTensorList则对整个OpArgList执行遍历。这套工具被GetOrCreateTask、SetIoTensors广泛使用，是实现"输入参数统一化为张量列表"的关键。

属性侧，AddAicpuAttr的七个重载（aicpu_task.h）将各类属性值封装为AnyValue存入AicpuAttrs：

• aclIntArray/aclIntArray*→std::vector<int64_t>；
• aclFloatArray/aclFloatArray*→std::vector<float>；
• aclBoolArray/aclBoolArray*→std::vector<bool>；
• 模板版本const V&→ 直接AnyValue::CreateFrom(value)，覆盖标量、字符串等简单类型。

线程 ID：GetTid

GetTid（aicpu_task.h）使用thread_local static缓存syscall(__NR_gettid)的结果，首次调用后不再触发系统调用，性能开销可忽略。线程 ID 被拼入任务键，确保不同线程上的任务不会相互复用（因为GetOrCreateTask的查找与创建存在线程间竞争，键中含线程 ID 可隔离不同线程的任务空间）。

启动入口：CreatAicpuKernelLauncher 与 ADD_TO_LAUNCHER_LIST_AICPU

CreatAicpuKernelLauncher：任务 → 启动器 → 执行队列

CreatAicpuKernelLauncher（aicpu_task.cpp）是任务下发对象的创建入口：

1. 校验args非空；
2. 调space.GetOrCreateTask(...)获取（新建或复用）任务；
3. 检查是否有待处理的溢出任务（TakePendingOverflowTask）：有则创建一次性启动器AiCpuOneShotKernelLauncher并executor->AbandonCache(true)丢弃缓存；无则创建可缓存启动器AiCpuKernelLauncher并executor->AbandonCache()；
4. 将启动器加入执行队列executor->AddToKernelLauncherList(launcher)；
5. 调用BuildGraph记录算子图拓扑（输入、输出、工作空间）。

其中AiCpuKernelLauncher（aicpu_kernel_launcher.h）的Launch()依次执行task->SetIoTensors(...)与task->Run(...)；executor->IsRepeatable()为假时，Run返回后立即SetVisit(false)归还任务，否则在析构时归还，以支持执行器的缓存复用语义。

ADD_TO_LAUNCHER_LIST_AICPU 宏

面向算子开发者暴露的统一入口是宏ADD_TO_LAUNCHER_LIST_AICPU（aicpu_task.h）：

ADD_TO_LAUNCHER_LIST_AICPU(KERNEL_NAME, attrNames, opArgs...)

宏内部通过GetOpArgContext(opArgs)收集算子入参，再调用CreatAicpuKernelLauncher(KERNEL_NAME##OpTypeId(), space, executor, attrNames, opArgCtx)。根据宏使用文档，其典型用法如下：

// 为 IndexPut 算子创建执行任务：accumulate 为属性名及属性参数， // selfRef、values、masks、indices 为输入，out 为输出 ADD_TO_LAUNCHER_LIST_AICPU(IndexPut, OP_ATTR_NAMES({"accumulate"}), OP_INPUT(selfRef, values, masks, indices), OP_OUTPUT(out), OP_ATTR(accumulate));

使用注意事项：

• 若算子需要INFER_SHAPE，必须先调用 INFER_SHAPE 宏 完成形状推导；
• 宏展开后会调用 OP_ATTR_NAMES、OP_INPUT(x...)、OP_OUTPUT(x...)、OP_ATTR(x...)等配套宏；
• 任务真正执行发生在第二阶段 APIaclnn_Xxx_被调用时，即"创建任务入队"与"执行"解耦。

op::internal::AiCpuKernelLauncher还会通过LauncherRepeatableChecker（CheckRepeatable）判断同一执行器是否可在多组输入间复用，这也是SetVisit/IsRepeatable配合实现"任务级复用"的完整闭环。

任务生命周期：Init、Run 与 SetIoTensors

Init：装载与建参

AicpuTfTask::Init（aicpu_task.cpp）与AicpuCCTask::Init（aicpu_task.cpp）都执行以下关键步骤：

1. 计算性能分析 ID（MsprofGetHashId）与汇总条目 ID（GenSummaryItemId）；
2. 创建AicpuExtInfoHandler并生成扩展信息缓冲（TF 或 CC 格式），DEPEND_SHAPE_RANGE场景下记录deviceExtMemSize_；
3. 创建对应参数处理器（AicpuTfArgsHandler/AicpuCCArgsHandler），生成并构建任务参数（GenTfArgs/GenCCArgs+BuildTfArgs/BuildCCArgs），其中 CC 版本会带上内核函数名与内核库名；
4. 将扩展信息解析到 Host 参数缓冲（extInfoHandle_->Parse）；
5. 从 JSON 算子信息库装载二进制句柄，并根据白名单与IsSupportedNewLaunch()决定使用新/旧启动接口。

Run：地址刷新与内核下发

AicpuCCTask::Run（aicpu_task.cpp）的时序可作为典型代表：

1. argsHandle_->UpdateDeviceExtInfoAddr(extInfoHandle_->deviceExtInfo_)同步 Device 侧扩展信息地址；
2. 若本次 stream 与上次不同，调用extInfoHandle_->UpdateKernelId()刷新内核 ID；
3. argsHandle_->UpdateIoAddr(...)刷新输入输出地址；
4. 在OpDfxGuard保护下执行内核下发：新接口走aclrtLaunchKernelWithHostArgs（自定义算子先aclrtRegisterCpuFunc注册函数句柄），旧接口走rtAicpuKernelLaunchExWithArgs；
5. 对DEPEND_SHAPE_RANGE算子，调用extInfoHandle_->UpdateOutputShapeFromExtInfo(outputs_, stream)把 Device 侧计算出的输出 Shape 回拷到 Host；
6. 在开启性能开关时上报附加信息（ReportAdditionInfo，任务类型MSPROF_GE_TASK_TYPE_AI_CPU）；
7. 最后调用PrintAicpuAllTimeStampInfo(opType)输出各阶段耗时（若已启用打点）。

SetIoTensors：执行前的输入输出刷新

AicpuTask::SetIoTensors（aicpu_task_base.cpp）清空并重建任务的inputs_/outputs_列表，然后调用extInfoHandle_->UpdateInputAndOutputShape(...)更新输入输出 Shape 与地址。由于 AI CPU 任务会被多次复用，每次真正下发前都必须用本次调用的实际张量刷新任务内部状态，这正是SetIoTensors在AiCpuKernelLauncher::Launch中被首先调用的原因。

资源管理与清理：aclnnAicpuFinalize 与 Clear

aclnnAicpuFinalize（aicpu_task_base.cpp）负责 AI CPU 模块的反初始化：在全局锁gAicpuTaskSpaceMutex_保护下遍历全局容器集合gAicpuTaskSpaceSet，对每个容器调用space->Clear()（即hashMap_.clear()）释放全部缓存任务，然后清空容器集合。源码注释特别强调：Clear()内部不加锁，以避免 Finalize 与任务 Init 并发调用时"容器成员锁与全局锁"互相死锁。

容器创建后由SaveAicpuTaskSpace（aicpu_task_base.cpp）注册到全局集合，保证 Finalize 时能遍历到所有已创建容器。因此，进程退出前的资源回收依赖开发者（或框架收尾逻辑）正确调用aclnnAicpuFinalize()。

调试与性能分析：PrintAicpuAllTimeStampInfo 与环境变量

PrintAicpuAllTimeStampInfo(opType)（aicpu_task_base.cpp）在gAicpuTimeStamp.isEnable为真时，按算子执行阶段打印耗时（微秒）：

打点开关由环境变量GE_PROFILING_TO_STD_OUT=1控制（EnableAicpuTimeStamp，aicpu_task_base.cpp），输出样例形如：

test case opType: IndexPut, index: 0 FindTask : 0.015000 ...

这套打点机制配合RecordAicpuTime插桩点，可用于定位 AI CPU 算子执行链路中各环节的性能瓶颈，是调优分布式/高并发 AI CPU 算子时的实用工具。

总结与使用建议

aicpu_task接口族是 CANN opbase 中 AI CPU 算子调度链路的内部引擎：AicpuTaskSpace负责"按算子类型 + 输入元信息 + 属性值"复用任务，AicpuTask（含AicpuTfTask/AicpuCCTask）负责参数封装与内核下发，AppendTensor/AppendAttrForKey/AddAicpuAttr等工具函数统一了张量列表与属性的序列化口径，CreatAicpuKernelLauncher与ADD_TO_LAUNCHER_LIST_AICPU则把任务接入aclOpExecutor的执行队列，配合SetIoTensors/SetVisit/SetRef实现"入队即封装、执行即刷新"的复用闭环。

在使用与学习上给出三点建议：

1. 保持对保留接口的敬畏：这些 API 属于op::internal且官方标注为保留状态，算子开发者应通过ADD_TO_LAUNCHER_LIST_AICPU等宏间接使用，避免直接耦合内部实现；
2. 关注缓存相关配置：ACLNN_CACHE_LIMIT（默认 10000、上限 10000000）直接决定任务缓存规模，高频动态 Shape 场景下需结合CalcDeviceCacheSize的工作空间预算评估内存占用；
3. 善用调试开关：设置GE_PROFILING_TO_STD_OUT=1即可在不改代码的情况下观察 AI CPU 算子各阶段的耗时分布，快速定位 FindTask 或 Launch 阶段的异常。

若需进一步深入，可继续阅读同目录下的 AicpuTask 类文档、AicpuTaskSpace 类文档、aicpu_args_handler 与 aicpu_ext_info_handle，并对照源码 aicpu_task.h、aicpu_task.cpp、aicpu_task_base.cpp 及测试样例 test_op_cache.cpp 验证本文所述的各条链路。

【免费下载链接】opbase本项目是CANN算子库的基础框架库，为算子提供公共依赖文件和基础调度能力。项目地址: https://gitcode.com/cann/opbase