LCOV - code coverage report
Current view: top level - legacy/ascend950/framework/communicator/hostdpu - task_service.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 83.4 % 217 181
Test Date: 2026-08-18 17:47:01 Functions: 100.0 % 15 15

            Line data    Source code
       1              : /**
       2              :  * Copyright (c) 2025 Huawei Technologies Co., Ltd.
       3              :  * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
       4              :  * CANN Open Software License Agreement Version 2.0 (the "License").
       5              :  * Please refer to the License for details. You may not use this file except in compliance with the License.
       6              :  * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
       7              :  * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
       8              :  * See LICENSE in the root of the software repository for the full text of the License.
       9              :  */
      10              : 
      11              : #include "task_service.h"
      12              : #include <algorithm>
      13              : #include "profiling/dlprof_function_v2.h"
      14              : #include <thread>
      15              : #include <chrono>
      16              : #include <atomic>
      17              : #include "acl/acl_rt.h"
      18              : #include "log.h"
      19              : #include "dpu_kernel_entrance.h"
      20              : #include "env_config/env_config_v2.h"
      21              : 
      22              : namespace Hccl {
      23              : constexpr uint32_t CTRL_HDR_FLAG_LENGTH = 1;
      24              : constexpr uint32_t TASKTYPE_ADDR_LENGTH = 256;
      25              : constexpr uint32_t CTRL_HDR_MSG_ID_LEN = 4;
      26              : constexpr uint32_t TIMEOUT_SIZE_BYTE = 4;      // timeout字段长度为4字节,表示超时时间,单位为秒
      27              : constexpr uint32_t CTRL_HDR_DATA_SIZE_LEN = 8; // size_t 在不同平台上长度不同,取最大值
      28              : constexpr uint32_t CTRL_HDR_DEFAULT_DATA_LEN = 512;
      29              : 
      30              : constexpr uint8_t TASK_UNSET = 0;
      31              : constexpr uint8_t TASK_OK = 1;
      32              : constexpr uint8_t TASK_TERMINATE = 2;
      33              : constexpr uint8_t MEMORY_DEVIDE = 2;
      34              : 
      35           25 : TaskService::TaskService(
      36           25 :     void* deviceMem, int32_t deviceMemSize, void* hostMem, int32_t hostMemSize, std::string commId, uint32_t devId)
      37           25 :     : npu2dpuMem_(deviceMem),
      38           25 :       shmemSize_(deviceMemSize / MEMORY_DEVIDE),
      39           25 :       hostMem_(hostMem),
      40           25 :       hostMemSize_(hostMemSize),
      41           25 :       commId_(commId),
      42           25 :       devId_(devId)
      43              : {
      44           25 :     int32_t controlSize = sizeof(uint8_t) + sizeof(char) * TASKTYPE_ADDR_LENGTH + sizeof(uint32_t) + TIMEOUT_SIZE_BYTE
      45              :                           + CTRL_HDR_DATA_SIZE_LEN;
      46           25 :     if (shmemSize_ < controlSize) {
      47            1 :         leftSize_ = 0;
      48              :     } else {
      49           24 :         leftSize_ = shmemSize_ - controlSize;
      50              :     }
      51           25 :     dpu2npuMem_ = static_cast<uint8_t*>(npu2dpuMem_) + shmemSize_;
      52           25 : }
      53              : 
      54            9 : HcclResult TaskService::TaskRegister(std::string taskType, CallbackTemplate callback)
      55              : {
      56           27 :     HCCL_INFO("[TaskService::%s] taskType[%s]", __func__, taskType.c_str());
      57            9 :     callbacks_.insert({taskType, callback});
      58            9 :     return HCCL_SUCCESS;
      59              : }
      60              : 
      61            2 : HcclResult TaskService::TaskUnRegister(std::string taskType)
      62              : {
      63            6 :     HCCL_INFO("[TaskService::%s] taskType[%s]", __func__, taskType.c_str());
      64            2 :     if (callbacks_.find(taskType) == callbacks_.end()) {
      65            3 :         HCCL_WARNING("[TaskService::%s] TaskType Not Found", __func__);
      66            1 :         return HCCL_E_NOT_FOUND;
      67              :     }
      68            1 :     callbacks_.erase(taskType);
      69            1 :     return HCCL_SUCCESS;
      70              : }
      71              : 
      72            5 : HcclResult TaskService::TaskProfRegister(ProfCallbackTemplate profCallback)
      73              : {
      74            5 :     if (profCallback_ != nullptr) {
      75            1 :         return HCCL_SUCCESS;
      76              :     }
      77            4 :     profCallback_ = profCallback;
      78            4 :     return HCCL_SUCCESS;
      79              : }
      80              : 
      81            4 : HcclResult TaskService::TaskReportRegister(ReportCallbackTemplate reportCallback)
      82              : {
      83            4 :     if (reportCallback_ != nullptr) {
      84            1 :         return HCCL_SUCCESS;
      85              :     }
      86            3 :     reportCallback_ = reportCallback;
      87            3 :     return HCCL_SUCCESS;
      88              : }
      89              : 
      90            9 : HcclResult TaskService::WriteFlag(uint8_t* flagPtr, uint8_t newFlag) const
      91              : {
      92            9 :     errno_t ret = memcpy_s(flagPtr, sizeof(newFlag), &newFlag, sizeof(newFlag));
      93            9 :     if (ret != EOK) {
      94            0 :         HCCL_ERROR("[TaskService::TaskRun] set flag failed: %d", ret);
      95            0 :         return HCCL_E_INTERNAL;
      96              :     }
      97            9 :     return HCCL_SUCCESS;
      98              : }
      99              : 
     100           12 : HcclResult TaskService::ReadFlag(uint8_t* ctrlHdr, uint64_t hdrLen, uint8_t& flag) const
     101              : {
     102           12 :     errno_t ret = memcpy_s(ctrlHdr, hdrLen, npu2dpuMem_, hdrLen);
     103           12 :     if (ret != EOK) {
     104            0 :         HCCL_ERROR("[TaskService::%s] memcpy_s failed on flag, return[%d].", __func__, ret);
     105            0 :         return HCCL_E_INTERNAL;
     106              :     }
     107           12 :     flag = *ctrlHdr;
     108           12 :     return HCCL_SUCCESS;
     109              : }
     110              : 
     111            7 : HcclResult TaskService::ReadTaskType(
     112              :     const uint8_t* ctrlHdr, [[maybe_unused]] uint64_t hdrLen, const uint8_t* srcTaskTypePtr,
     113              :     std::string& taskTypeStr) const
     114              : {
     115            7 :     CHK_PTR_NULL(srcTaskTypePtr);
     116              :     // 读 taskType
     117            7 :     char* taskType = new char[TASKTYPE_ADDR_LENGTH];
     118            7 :     CHK_PTR_NULL(taskType);
     119           14 :     int ret = memcpy_s(
     120            7 :         taskType, (sizeof(char) * TASKTYPE_ADDR_LENGTH), ctrlHdr + CTRL_HDR_FLAG_LENGTH,
     121              :         (sizeof(char) * TASKTYPE_ADDR_LENGTH));
     122            7 :     if (ret != EOK) {
     123            0 :         HCCL_ERROR("[%s] memcpy failed on taskType, return[%d].", __func__, ret);
     124            0 :         delete[] taskType;
     125            0 :         return HCCL_E_INTERNAL;
     126              :     }
     127              :     //  查找 \0
     128            7 :     auto it = std::find(taskType, taskType + TASKTYPE_ADDR_LENGTH, '\0');
     129            7 :     if (it == taskType + TASKTYPE_ADDR_LENGTH) {
     130            0 :         HCCL_ERROR("[TaskService::TaskRun] No Null Character Found Within TaskType Max Length");
     131            0 :         delete[] taskType;
     132            0 :         return HCCL_E_PARA;
     133              :     }
     134            7 :     taskTypeStr = std::string(taskType);
     135            7 :     delete[] taskType;
     136           21 :     HCCL_INFO("[TaskService::TaskRun] read taskType = %s", taskTypeStr.c_str());
     137              : 
     138            7 :     uint32_t msgId{0};
     139            7 :     ret = memcpy_s(&msgId, sizeof(msgId), ctrlHdr + CTRL_HDR_FLAG_LENGTH + TASKTYPE_ADDR_LENGTH, sizeof(msgId));
     140            7 :     if (ret != EOK) {
     141            0 :         HCCL_ERROR("[%s] memcpy failed on msgId, return[%d].", __func__, ret);
     142            0 :         return HCCL_E_INTERNAL;
     143              :     }
     144           21 :     HCCL_INFO("[TaskService::TaskRun] read msgId = %u.", msgId);
     145            7 :     return HCCL_SUCCESS;
     146              : }
     147              : 
     148              : // 共享内存排布:|stop flag[1]|hcclret[2]|dstret[2]|
     149              : // 其中,stop flag为aicpu侧读取是否停止的标志位, hcclret为aicpu背景线程读取是否有错的标志位,
     150              : // dstret为host侧taskexception回调读取是否有错的标志位
     151            7 : HcclResult TaskService::ExecuteTaskexception(int32_t ret)
     152              : {
     153            7 :     if (g_taskExpMemMap.find(commId_) == g_taskExpMemMap.end()) {
     154            3 :         HCCL_ERROR("TaskService::ExecuteTaskexception commId not in g_taskExpMemMap, please check");
     155            1 :         return HCCL_E_NOT_FOUND;
     156              :     }
     157            6 :     void* taskexpShmem = g_taskExpMemMap[commId_][devId_];
     158            6 :     HcclResult hcclRet = static_cast<HcclResult>(ret);
     159            6 :     if (taskexpShmem != nullptr) {
     160            4 :         uint8_t* stopFlagPtr = static_cast<uint8_t*>(taskexpShmem);
     161            4 :         uint8_t* hcclRetPtr = stopFlagPtr + sizeof(uint8_t);
     162            4 :         uint8_t* dstRetPtr = hcclRetPtr + sizeof(uint16_t);
     163            4 :         auto ret = memcpy_s(hcclRetPtr, sizeof(uint16_t), &hcclRet, sizeof(uint16_t)); // aicpu背景线程轮询的标志位
     164            4 :         if (ret != 0) {
     165            0 :             HCCL_ERROR("[TaskService::ExecuteTaskexception] memcpy ret for device failed.");
     166            0 :             return HCCL_E_MEMORY;
     167              :         }
     168            4 :         ret = memcpy_s(dstRetPtr, sizeof(uint16_t), &hcclRet, sizeof(uint16_t)); // host侧taskexception回调执读出错误码
     169            4 :         if (ret != 0) {
     170            0 :             HCCL_ERROR("[TaskService::ExecuteTaskexception] memcpy ret for host failed.");
     171            0 :             return HCCL_E_MEMORY;
     172              :         }
     173              :     }
     174            6 :     return HCCL_SUCCESS;
     175              : }
     176              : 
     177            5 : HcclResult TaskService::ExecuteTaskClean() const
     178              : {
     179            5 :     errno_t ret = memset_s(hostMem_, hostMemSize_, 0, hostMemSize_);
     180            5 :     if (ret != EOK) {
     181            0 :         HCCL_ERROR("memset hostMem failed: %d", ret);
     182            0 :         return HCCL_E_MEMORY;
     183              :     }
     184            5 :     ret = memset_s(npu2dpuMem_, shmemSize_, 0, shmemSize_);
     185            5 :     if (ret != EOK) {
     186            0 :         HCCL_ERROR("memset npu2dpuMem_ failed: %d", ret);
     187            0 :         return HCCL_E_MEMORY;
     188              :     }
     189            5 :     ret = memset_s(dpu2npuMem_, shmemSize_, 0, shmemSize_);
     190            5 :     if (ret != EOK) {
     191            0 :         HCCL_ERROR("memset dpu2npuMem_ failed: %d", ret);
     192            0 :         return HCCL_E_MEMORY;
     193              :     }
     194            5 :     return HCCL_SUCCESS;
     195              : }
     196              : 
     197            7 : HcclResult TaskService::ExecuteTask(uint8_t* ctrlHdr, uint64_t hdrLen, uint8_t* srcPtr, std::string taskTypeStr)
     198              : {
     199            7 :     uint64_t beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
     200            7 :     auto itFunc = callbacks_.find(taskTypeStr);
     201            7 :     if (itFunc == callbacks_.end()) {
     202            3 :         HCCL_ERROR("[TaskService::TaskRun] Callback of taskType[%s] Not Found", taskTypeStr.c_str());
     203            1 :         std::string taskTypeMsg{"map{"};
     204            2 :         for (const auto& pair : callbacks_) {
     205            1 :             taskTypeMsg += pair.first;
     206              :         }
     207            1 :         taskTypeMsg += "}";
     208            3 :         HCCL_ERROR("[TaskService::TaskRun] Callback key : %s", taskTypeMsg.c_str());
     209            1 :         return HCCL_E_NOT_FOUND;
     210            1 :     }
     211              : 
     212              :     // copy data
     213            6 :     uint64_t dataLen = *reinterpret_cast<size_t*>(
     214              :         ctrlHdr + CTRL_HDR_FLAG_LENGTH + TASKTYPE_ADDR_LENGTH + CTRL_HDR_MSG_ID_LEN + TIMEOUT_SIZE_BYTE);
     215            6 :     if (dataLen > static_cast<uint64_t>(leftSize_) || dataLen > static_cast<uint64_t>(hostMemSize_)) {
     216            3 :         HCCL_ERROR(
     217              :             "[TaskService::%s] dataLen[%llu] larger than leftSize[%d] or hostMemSize[%d]", __func__, dataLen, leftSize_,
     218              :             hostMemSize_);
     219            1 :         return HCCL_E_PARA;
     220              :     }
     221            5 :     uint32_t ctrlHdrLen = CTRL_HDR_FLAG_LENGTH + TASKTYPE_ADDR_LENGTH + CTRL_HDR_MSG_ID_LEN + TIMEOUT_SIZE_BYTE
     222              :                           + CTRL_HDR_DATA_SIZE_LEN;
     223              :     /* ctrlHdr提前从deviceMem copy一定长度,如果长度够,直接从ctrlHdr copy,减少一次aclmemcpy耗时 */
     224            5 :     uint8_t* dataPtr = nullptr;
     225            5 :     if (hdrLen < ctrlHdrLen + dataLen) {
     226            0 :         dataPtr = srcPtr + ctrlHdrLen;
     227              :     } else {
     228            5 :         dataPtr = ctrlHdr + ctrlHdrLen;
     229              :     }
     230            5 :     errno_t ret = memcpy_s(hostMem_, leftSize_, dataPtr, dataLen);
     231            5 :     if (ret != EOK) {
     232            0 :         HCCL_ERROR("control data memcpy failed: %d", ret);
     233            0 :         return HCCL_E_MEMORY;
     234              :     }
     235            5 :     auto callbackRet = itFunc->second(reinterpret_cast<uint64_t>(hostMem_), dataLen);
     236            5 :     if (callbackRet != 0) {
     237              :         // dpu任务出错,清理DPUTAG共享内存内容
     238            4 :         CHK_RET(ExecuteTaskClean());
     239              :         // 写DPUTASKEXCEPTION
     240            4 :         CHK_RET(ExecuteTaskexception(callbackRet));
     241            4 :         return HCCL_E_INTERNAL;
     242              :     }
     243            1 :     if (profCallback_ != nullptr) {
     244            0 :         TaskParam taskParam{};
     245            0 :         taskParam.beginTime = beginTime;
     246            0 :         taskParam.taskType = Hccl::TaskParamType::TASK_DPU_KERNEL;
     247            0 :         taskParam.endTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
     248            0 :         taskParam.isMaster = true;
     249            0 :         profCallback_(taskParam, INVALID_U64);
     250            0 :     }
     251            1 :     return HCCL_SUCCESS;
     252              : }
     253              : 
     254            1 : HcclResult TaskService::SynchronizeControlInfo([[maybe_unused]] uint8_t* ctrlHdr, [[maybe_unused]] uint64_t hdrLen)
     255              : {
     256            1 :     CHK_PTR_NULL(npu2dpuMem_);
     257            1 :     CHK_PTR_NULL(dpu2npuMem_);
     258              :     // npu2dpu -> dpu2npu
     259            1 :     int32_t controlDataSize = sizeof(uint8_t) + sizeof(char) * TASKTYPE_ADDR_LENGTH + sizeof(uint32_t);
     260            3 :     HCCL_INFO("[TaskService::TaskRun] Send response: npu2dpu -> dpu2npu memcpy");
     261            1 :     errno_t ret = memcpy_s(dpu2npuMem_, controlDataSize, npu2dpuMem_, controlDataSize);
     262            1 :     if (ret != EOK) {
     263            0 :         HCCL_ERROR("[TaskService::TaskRun] npu2dpu -> dpu2npu memcpy failed: %d", ret);
     264            0 :         return HCCL_E_INTERNAL;
     265              :     }
     266              : 
     267              :     std::atomic_thread_fence(std::memory_order_seq_cst);
     268              : 
     269            1 :     uint8_t newFlag = 1;
     270            3 :     HCCL_INFO("[TaskService::TaskRun] Send response: Set dpu2npu flag -> 1");
     271            1 :     static_cast<std::atomic<uint8_t>*>(dpu2npuMem_)->store(newFlag, std::memory_order_release);
     272            1 :     return HCCL_SUCCESS;
     273              : }
     274              : 
     275            7 : HcclResult TaskService::ProcessTaskOk(uint8_t* ctrlHdr, uint64_t hdrLen, uint8_t* srcFlagPtr, uint8_t* srcTaskTypePtr)
     276              : {
     277            7 :     std::string taskTypeStr;
     278           21 :     HCCL_INFO("[TaskService::TaskRun] flag = %u.", TASK_OK);
     279           21 :     HCCL_INFO("[TaskService::TaskRun] Set npu2dpu flag -> %u.", TASK_UNSET);
     280            7 :     CHK_RET(WriteFlag(srcFlagPtr, TASK_UNSET));
     281            7 :     CHK_RET(ReadTaskType(ctrlHdr, hdrLen, srcTaskTypePtr, taskTypeStr));
     282           25 :     CHK_RET(ExecuteTask(ctrlHdr, hdrLen, srcFlagPtr, taskTypeStr));
     283            1 :     if (reportCallback_ != nullptr) {
     284            0 :         CHK_RET(reportCallback_());
     285              :     }
     286            1 :     CHK_RET(SynchronizeControlInfo(ctrlHdr, hdrLen));
     287            1 :     return HCCL_SUCCESS;
     288            7 : }
     289              : 
     290            2 : HcclResult TaskService::ExecuteExit(uint8_t* srcFlagPtr) const
     291              : {
     292            2 :     CHK_RET(WriteFlag(srcFlagPtr, TASK_TERMINATE_RESPONSE));
     293            6 :     HCCL_INFO("[TaskService::TaskRun] Exiting.");
     294            2 :     return HCCL_SUCCESS;
     295              : }
     296              : 
     297           12 : HcclResult TaskService::TaskRun()
     298              : {
     299            0 :     CHK_PTR_NULL(hostMem_);
     300           12 :     CHK_PTR_NULL(npu2dpuMem_);
     301           12 :     CHK_PTR_NULL(dpu2npuMem_);
     302           36 :     HCCL_INFO(
     303              :         "[TaskService::%s] TaskService{npu2dpuMem:%p; dpu2npuMem:%p; hostMem:%p}", __func__, npu2dpuMem_, dpu2npuMem_,
     304              :         hostMem_);
     305           12 :     if (leftSize_ <= 0) {
     306            3 :         HCCL_ERROR("[TaskService::%s] dataSize[%d] illegal", __func__, leftSize_);
     307            1 :         return HCCL_E_INTERNAL;
     308              :     }
     309           11 :     if (leftSize_ > hostMemSize_) {
     310            3 :         HCCL_ERROR("[TaskService::%s] hostMemSize[%d] less than dataSize[%d]", __func__, hostMemSize_, leftSize_);
     311            1 :         return HCCL_E_INTERNAL;
     312              :     }
     313           10 :     uint8_t flag{0};
     314           10 :     uint8_t* srcFlagPtr = static_cast<uint8_t*>(npu2dpuMem_);
     315           10 :     uint8_t* dstTimeoutPtr
     316           10 :         = static_cast<uint8_t*>(dpu2npuMem_) + sizeof(flag) + sizeof(char) * TASKTYPE_ADDR_LENGTH + CTRL_HDR_MSG_ID_LEN;
     317           10 :     uint8_t* srcTaskTypePtr = srcFlagPtr + sizeof(flag);
     318           10 :     uint64_t hdrLen = CTRL_HDR_FLAG_LENGTH + TASKTYPE_ADDR_LENGTH + CTRL_HDR_MSG_ID_LEN + TIMEOUT_SIZE_BYTE
     319              :                       + CTRL_HDR_DATA_SIZE_LEN + CTRL_HDR_DEFAULT_DATA_LEN;
     320           10 :     uint8_t ctrlHdr[hdrLen];
     321           10 :     u32 timeout = Hccl::EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut();
     322           10 :     timeout = timeout > 1 ? timeout - 1 : timeout; // 执行超时时间减1秒,避免aicpu侧超时
     323           10 :     CHK_SAFETY_FUNC_RET(memcpy_s(dstTimeoutPtr, sizeof(timeout), &timeout, sizeof(timeout)));
     324              : 
     325              :     while (true) {
     326           12 :         CHK_RET(ReadFlag(ctrlHdr, hdrLen, flag));
     327           12 :         switch (flag) {
     328            1 :             case TASK_UNSET:
     329            1 :                 std::this_thread::sleep_for(std::chrono::milliseconds(1));
     330            1 :                 continue;
     331            7 :             case TASK_OK:
     332           25 :                 CHK_RET(ProcessTaskOk(ctrlHdr, hdrLen, srcFlagPtr, srcTaskTypePtr));
     333            1 :                 continue;
     334            2 :             case TASK_TERMINATE:
     335            6 :                 HCCL_INFO("[TaskService::TaskRun] flag = %u.", flag);
     336            6 :                 HCCL_INFO("[TaskService::TaskRun] Set npu2dpu flag -> %u. Task Run END.", TASK_TERMINATE_RESPONSE);
     337            2 :                 CHK_RET(ExecuteExit(srcFlagPtr));
     338            2 :                 return HCCL_SUCCESS;
     339            1 :             case TASK_TERMINATE_RESPONSE:
     340            3 :                 HCCL_INFO("[TaskService::TaskRun] flag = %u. Exiting.", flag);
     341            1 :                 return HCCL_SUCCESS;
     342            1 :             default:
     343            3 :                 HCCL_INFO("[TaskService::TaskRun] flag = %u. Unimplemented flag. Exiting.", flag);
     344            1 :                 return HCCL_SUCCESS;
     345            2 :         }
     346            2 :     }
     347              :     return HCCL_SUCCESS;
     348           12 : }
     349              : } // namespace Hccl
        

Generated by: LCOV version 2.0-1