LCOV - code coverage report
Current view: top level - coll_communicator_mgr/resource_mgr/local/my_rank/comm_engine_reses/kernel_launch - hccl_kernel_launch_aicpu.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 82.9 % 181 150
Test Date: 2026-07-28 12:11:00 Functions: 100.0 % 7 7

            Line data    Source code
       1              : /**
       2              :  * Copyright (c) 2026 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 <thread>
      12              : #include <sstream>
      13              : #include <string>
      14              : #include <vector>
      15              : 
      16              : #include "hccl/hccl_launch.h"
      17              : #include "acl/acl_rt.h"
      18              : #include "hccl_group.h"
      19              : #include "hccl_res_expt.h"
      20              : #include "hccl_aicpu_interface.h"
      21              : 
      22              : #include "hccl_independent_common.h"
      23              : #include "group_schedule_mgr.h"
      24              : 
      25              : using namespace hccl;
      26              : 
      27              : constexpr uint32_t NUM_ZERO = 0;
      28              : constexpr uint32_t NUM_ONE = 1;
      29              : constexpr uint32_t NUM_TWO = 2;
      30              : constexpr uint32_t NUM_THREE = 3;
      31              : static uint32_t g_KernelLaunchTimeout = UINT16_MAX;
      32              : 
      33            3 : static HcclResult LaunchNotifyWaitToThread(
      34              :     HcclComm comm, aclrtStream unfoldStream, ThreadHandle srcThread, uint32_t dstNotifyIdx)
      35              : {
      36              :     aclrtFuncHandle funcHandle;
      37              :     aclrtArgsHandle argsHandle;
      38            3 :     std::string kernelName = "RunAicpuNotifyWait";
      39              :     // 1. 获取 function handle
      40            3 :     hccl::hcclComm *hcclComm = static_cast<hccl::hcclComm *>(comm);
      41            3 :     auto binKernelHandle = hcclComm->GetBinHandle();
      42            3 :     aclError ret = aclrtBinaryGetFunction(binKernelHandle, kernelName.c_str(), &funcHandle);
      43            3 :     CHK_PRT_RET(ret != ACL_SUCCESS, HCCL_ERROR("[aclrtBinaryGetFunction]errNo[0x%016llx] get func handle failed, "
      44              :         "kernelName:%s", ret, kernelName.c_str()), HCCL_E_RUNTIME);
      45              : 
      46              :     // 2. 初始化 args handle
      47            3 :     ret = aclrtKernelArgsInit(funcHandle, &argsHandle);
      48            3 :     CHK_PRT_RET(ret != ACL_SUCCESS,
      49              :         HCCL_ERROR("[aclrtKernelArgsInit]errNo[0x%016llx] args init failed, kernelName:%s", ret, kernelName.c_str()),
      50              :         HCCL_E_RUNTIME);
      51              : 
      52              :     // 3. 准备参数并 append
      53              :     ThreadNotifyWaitParam param;
      54            3 :     param.thread = srcThread;
      55            3 :     param.notifyIdx = dstNotifyIdx;
      56              :     aclrtParamHandle paraHandle;
      57            3 :     ret = aclrtKernelArgsAppend(argsHandle, &param, sizeof(ThreadNotifyWaitParam), &paraHandle);
      58            3 :     CHK_PRT_RET(ret != ACL_SUCCESS, HCCL_ERROR(
      59              :         "[aclrtKernelArgsAppend]errNo[0x%016llx] args append failed, kernelName:%s", ret, kernelName.c_str()),
      60              :         HCCL_E_RUNTIME);
      61              : 
      62              :     // 4. finalize args
      63            3 :     ret = aclrtKernelArgsFinalize(argsHandle);
      64            3 :     CHK_PRT_RET(ret != ACL_SUCCESS,
      65              :         HCCL_ERROR("[aclrtKernelArgsFinalize]errNo[0x%016llx] args finalize failed, kernelName:%s", ret, kernelName.c_str()),
      66              :         HCCL_E_RUNTIME);
      67              : 
      68              :     // 5. 下发 kernel
      69              :     aclrtLaunchKernelCfg cfg;
      70              :     aclrtLaunchKernelAttr attr;
      71            3 :     attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
      72            3 :     attr.value.timeout = g_KernelLaunchTimeout;
      73            3 :     cfg.numAttrs = 1;
      74            3 :     cfg.attrs = &attr;
      75            3 :     constexpr u32 numBlocks = 1;
      76              : 
      77            3 :     ret = aclrtLaunchKernelWithConfig(funcHandle, numBlocks, unfoldStream, &cfg, argsHandle, nullptr);
      78            3 :     CHK_PRT_RET(ret != ACL_SUCCESS,HCCL_ERROR("[aclrtLaunchKernelWithConfig]errNo[0x%016llx] launch kernel failed, "
      79              :         "kernelName:%s", ret, kernelName.c_str()), HCCL_E_RUNTIME);
      80              : 
      81            3 :     return HCCL_SUCCESS;
      82            3 : }
      83              : 
      84            3 : static HcclResult LaunchP2pExec(HcclComm comm, aclrtStream unfoldStream, const HcclKernelFuncInfo *funcInfo, const void *funcArgs,
      85              :     uint32_t argSize, ThreadHandle sendRecvThread)
      86              : {
      87              :     aclrtFuncHandle funcHandle;
      88              :     aclrtArgsHandle argsHandle;
      89              :     // 1. 获取 function handle
      90            3 :     aclrtBinHandle binKernelHandle = nullptr;
      91            3 :     hccl::hcclComm *hcclComm = static_cast<hccl::hcclComm *>(comm);
      92            3 :     CollComm *collComm = hcclComm->GetCollComm();
      93            3 :     CHK_PTR_NULL(collComm);
      94            2 :     CHK_RET(collComm->GetHcclBinHandle(binKernelHandle));
      95            2 :     aclError ret = aclrtBinaryGetFunction(binKernelHandle, funcInfo->kernelFuncName, &funcHandle);
      96            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
      97              :         HCCL_ERROR("[aclrtBinaryGetFunction]errNo[0x%016llx] get func handle failed, kernelName:%s",
      98              :             ret, funcInfo->kernelFuncName), HCCL_E_RUNTIME);
      99              : 
     100              :     // 2. 初始化 args handle
     101            2 :     ret = aclrtKernelArgsInit(funcHandle, &argsHandle);
     102            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
     103              :         HCCL_ERROR("[aclrtKernelArgsInit]errNo[0x%016llx] args init failed, kernelName:%s",
     104              :             ret, funcInfo->kernelFuncName), HCCL_E_RUNTIME);
     105              : 
     106              :     // 3. 准备参数并 append
     107              :     HcclP2pKernelParam params;
     108            2 :     params.sendRecvThread = sendRecvThread;
     109            2 :     memset_s(params.opParams, P2P_MAX_ARG_SIZE, 0, P2P_MAX_ARG_SIZE);
     110            2 :     memcpy_s(params.opParams, P2P_MAX_ARG_SIZE, funcArgs, argSize);
     111              : 
     112              :     aclrtParamHandle paraHandle;
     113            2 :     ret = aclrtKernelArgsAppend(argsHandle, &params, sizeof(HcclP2pKernelParam), &paraHandle);
     114            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
     115              :         HCCL_ERROR("[aclrtKernelArgsAppend]errNo[0x%016llx] args append failed, kernelName:%s",
     116              :             ret, funcInfo->kernelFuncName), HCCL_E_RUNTIME);
     117              : 
     118              :     // 4. finalize args
     119            2 :     ret = aclrtKernelArgsFinalize(argsHandle);
     120            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
     121              :         HCCL_ERROR("[aclrtKernelArgsFinalize]errNo[0x%016llx] args finalize failed, kernelName:%s",
     122              :             ret, funcInfo->kernelFuncName), HCCL_E_RUNTIME);
     123              : 
     124              :     // 5. 下发 kernel
     125              :     aclrtLaunchKernelCfg cfg;
     126              :     aclrtLaunchKernelAttr attr;
     127            2 :     attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
     128            2 :     attr.value.timeout = g_KernelLaunchTimeout;
     129            2 :     cfg.numAttrs = 1;
     130            2 :     cfg.attrs = &attr;
     131            2 :     constexpr u32 numBlocks = 1;
     132              : 
     133            2 :     ret = aclrtLaunchKernelWithConfig(funcHandle, numBlocks, unfoldStream, &cfg, argsHandle, nullptr);
     134            2 :     CHK_PRT_RET(ret != ACL_SUCCESS, HCCL_ERROR("[aclrtLaunchKernelWithConfig]errNo[0x%016llx] launch kernel failed, "
     135              :             "kernelName:%s", ret, funcInfo->kernelFuncName), HCCL_E_RUNTIME);
     136              : 
     137            2 :     return HCCL_SUCCESS;
     138              : }
     139              : 
     140              : // 放到kernel launch的地方
     141            2 : static HcclResult LaunchNotifyRecordToThread(
     142              :     HcclComm comm, aclrtStream unfoldStream, ThreadHandle srcThread, ThreadHandle dstThread, uint32_t dstNotifyIdx)
     143              : {
     144              :     aclrtFuncHandle funcHandle;
     145              :     aclrtArgsHandle argsHandle;
     146            2 :     std::string kernelName = "RunAicpuNotifyRecord";
     147              : 
     148              :     // 1. 获取 function handle
     149            2 :     hccl::hcclComm *hcclComm = static_cast<hccl::hcclComm *>(comm);
     150            2 :     auto binKernelHandle = hcclComm->GetBinHandle();
     151            2 :     aclError ret = aclrtBinaryGetFunction(binKernelHandle, kernelName.c_str(), &funcHandle);
     152            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
     153              :         HCCL_ERROR(
     154              :             "[aclrtBinaryGetFunction]errNo[0x%016llx] get func handle failed, kernelName:%s", ret, kernelName.c_str()),
     155              :         HCCL_E_RUNTIME);
     156              : 
     157              :     // 2. 初始化 args handle
     158            2 :     ret = aclrtKernelArgsInit(funcHandle, &argsHandle);
     159            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
     160              :         HCCL_ERROR("[aclrtKernelArgsInit]errNo[0x%016llx] args init failed, kernelName:%s", ret, kernelName.c_str()),
     161              :         HCCL_E_RUNTIME);
     162              : 
     163              :     // 3. 准备参数并 append
     164              :     ThreadNotifyRecordParam param;
     165            2 :     param.thread = srcThread;
     166            2 :     param.dstThread = dstThread;
     167            2 :     param.dstNotifyIdx = dstNotifyIdx;
     168              :     aclrtParamHandle paraHandle;
     169            2 :     ret = aclrtKernelArgsAppend(argsHandle, &param, sizeof(ThreadNotifyRecordParam), &paraHandle);
     170            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
     171              :         HCCL_ERROR(
     172              :             "[aclrtKernelArgsAppend]errNo[0x%016llx] args append failed, kernelName:%s", ret, kernelName.c_str()),
     173              :         HCCL_E_RUNTIME);
     174              : 
     175              :     // 4. finalize args
     176            2 :     ret = aclrtKernelArgsFinalize(argsHandle);
     177            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
     178              :         HCCL_ERROR(
     179              :             "[aclrtKernelArgsFinalize]errNo[0x%016llx] args finalize failed, kernelName:%s", ret, kernelName.c_str()),
     180              :         HCCL_E_RUNTIME);
     181              : 
     182              :     // 5. 下发 kernel
     183              :     aclrtLaunchKernelCfg cfg;
     184              :     aclrtLaunchKernelAttr attr;
     185            2 :     attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
     186            2 :     attr.value.timeout = g_KernelLaunchTimeout;
     187            2 :     cfg.numAttrs = 1;
     188            2 :     cfg.attrs = &attr;
     189            2 :     constexpr u32 numBlocks = 1;
     190              : 
     191            2 :     ret = aclrtLaunchKernelWithConfig(funcHandle, numBlocks, unfoldStream, &cfg, argsHandle, nullptr);
     192            2 :     CHK_PRT_RET(ret != ACL_SUCCESS,
     193              :         HCCL_ERROR("[aclrtLaunchKernelWithConfig]errNo[0x%016llx] launch kernel failed, "
     194              :                    "kernelName:%s",
     195              :             ret, kernelName.c_str()),
     196              :         HCCL_E_RUNTIME);
     197              : 
     198            2 :     return HCCL_SUCCESS;
     199            2 : }
     200              : 
     201            1 : static HcclResult AicpuKernelLaunchDirect(HcclComm comm, const HcclKernelFuncInfo *funcInfo,
     202              :     ThreadHandle aicpuThreadHandle, aclrtStream unfoldStream, aclrtStream userStream)
     203              : {
     204            1 :     CHK_PTR_NULL(comm);
     205            1 :     CHK_PTR_NULL(unfoldStream);
     206            1 :     CHK_PTR_NULL(userStream);
     207            1 :     CHK_PTR_NULL(funcInfo);
     208              : 
     209            1 :     void *args = funcInfo->args;
     210            1 :     uint32_t argSize = funcInfo->argSize;
     211            1 :     if (argSize > 0 && args == nullptr) {
     212            0 :         HCCL_ERROR("[AicpuKernelLaunchDirect] args is null but argSize[%u] > 0", argSize);
     213            0 :         return HCCL_E_PTR;
     214              :     }
     215              : 
     216            1 :     ThreadHandle cpuTsThread{0};
     217            1 :     ThreadHandle exportedAicpuTsThread{0};
     218            1 :     ThreadHandle exportedCpuTsThread{0};
     219              :     uint32_t notifyNumOnMainThread;
     220            1 :     CHK_RET(HcclThreadAcquireWithStream(comm, COMM_ENGINE_CPU_TS, userStream, NUM_THREE, &cpuTsThread));
     221            1 :     CHK_RET(HcclThreadExportToCommEngine(comm, NUM_ONE, &cpuTsThread, COMM_ENGINE_AICPU_TS, &exportedAicpuTsThread));
     222            1 :     CHK_RET(HcclThreadExportToCommEngine(comm, NUM_ONE, &aicpuThreadHandle, COMM_ENGINE_CPU_TS, &exportedCpuTsThread));
     223            1 :     CHK_RET(HcclGetNotifyNumInThread(comm, exportedCpuTsThread, COMM_ENGINE_AICPU_TS, &notifyNumOnMainThread));
     224              : 
     225            1 :     CHK_RET(static_cast<HcclResult>(
     226              :         HcommThreadNotifyRecordOnThread(cpuTsThread, exportedCpuTsThread, notifyNumOnMainThread - 1))); // h2d record
     227            1 :     uint64_t beginTime = HcommGetProfilingSysCycleTime(); // AicpuKernel report start
     228            1 :     CHK_RET(LaunchNotifyWaitToThread(comm, unfoldStream, aicpuThreadHandle, notifyNumOnMainThread - 1)); // device wait
     229            1 :     CHK_RET(LaunchP2pExec(comm, unfoldStream, funcInfo, args, argSize, aicpuThreadHandle)); // device run task
     230            0 :     CHK_RET(LaunchNotifyRecordToThread(comm, unfoldStream, aicpuThreadHandle, exportedAicpuTsThread, NUM_ZERO)); // d2h record
     231            0 :     std::string kernelNameCStr(funcInfo->kernelFuncName);
     232            0 :     HcclResult ret = HcclReportAicpuKernel(comm, beginTime, kernelNameCStr.data()); // AicpuKernel report end
     233            0 :     if (ret != HCCL_SUCCESS) {
     234            0 :         HCCL_ERROR("[AicpuKernelLaunchDirect] HcclReportAicpuKernel failed, beginTime %lu, kernelName %s, ret %d ",
     235              :             beginTime, kernelNameCStr.c_str(), ret);
     236            0 :         return ret;
     237              :     }
     238            0 :     CHK_RET(static_cast<HcclResult>(HcommThreadNotifyWaitOnThreadWithDefaultTimeout(cpuTsThread, NUM_ZERO))); // host wait
     239              : 
     240            0 :     return HCCL_SUCCESS;
     241            0 : }
     242              : 
     243            5 : HcclResult HcclAicpuKernelLaunch(HcclComm comm, const HcclOpDesc *opInfo, const HcclKernelFuncInfo *funcInfo,
     244              :     ThreadHandle aicpuThreadHandle, aclrtStream userStream, const HcclKernelLaunchCfg *kernelLaunchCfg)
     245              : {
     246            5 :     CHK_PTR_NULL(comm);
     247            4 :     CHK_PTR_NULL(userStream);
     248            3 :     CHK_PTR_NULL(funcInfo);
     249            2 :     CHK_PTR_NULL(opInfo);
     250            2 :     CHK_PTR_NULL(kernelLaunchCfg);
     251              : 
     252            2 :     uint32_t argSize = funcInfo->argSize;
     253            2 :     void *args = funcInfo->args;
     254              : 
     255            2 :     g_KernelLaunchTimeout = kernelLaunchCfg->timeOut;
     256            2 :     if (argSize > 0 && args == nullptr) {
     257            1 :         HCCL_ERROR("[HcclAicpuKernelLaunch] args is null but argSize[%u] > 0", argSize);
     258            1 :         return HCCL_E_PTR;
     259              :     }
     260              : 
     261            1 :     HCCL_INFO("[HcclAicpuKernelLaunch] opDescType[%u], kernelSo[%s], kernelFuncName[%s], argSize[%u], "
     262              :               "aicpuThreadHandle[%lu], hcclGroupDepth[%d]",
     263              :         opInfo->opDescType, funcInfo->kernelSoName, funcInfo->kernelFuncName, argSize, aicpuThreadHandle, hcclGroupDepth);
     264              : 
     265            1 :     if (hcclGroupDepth > 0) {
     266            0 :         hccl::hcclComm *hcclComm = static_cast<hccl::hcclComm *>(comm);
     267            0 :         CollComm *collComm = hcclComm->GetCollComm();
     268            0 :         CHK_PTR_NULL(collComm);
     269            0 :         if (argSize > P2P_MAX_ARG_SIZE) {
     270            0 :             HCCL_ERROR("[HcclAicpuKernelLaunch] argSize[%u] over P2P_MAX_ARG_SIZE", argSize);
     271            0 :             return HCCL_E_PARA;
     272              :         }
     273            0 :         HCCL_INFO("[HcclAicpuKernelLaunch] group mode, add p2p task hcclGroupDepth[%d]", hcclGroupDepth);
     274              :         HcclP2pTask task;
     275            0 :         task.desc = opInfo->p2p;
     276            0 :         task.stream = opInfo->p2p.unfoldStream;
     277            0 :         memcpy_s(task.funcInfo.kernelSoName, HCCL_KERNEL_SO_NAME_MAX_LEN, funcInfo->kernelSoName,
     278              :             HCCL_KERNEL_SO_NAME_MAX_LEN);
     279            0 :         memcpy_s(task.funcInfo.kernelFuncName, HCCL_KERNEL_FUNC_NAME_MAX_LEN, funcInfo->kernelFuncName,
     280              :             HCCL_KERNEL_FUNC_NAME_MAX_LEN);
     281            0 :         memcpy_s(task.args, P2P_MAX_ARG_SIZE, args, argSize);
     282            0 :         task.argSize = argSize;
     283            0 :         task.usrStream = userStream;
     284            0 :         CHK_RET(collComm->groupScheduleMgr->AppendGroupP2pTask(comm, task, opInfo->p2p));
     285            0 :         return HCCL_SUCCESS;
     286              :     }
     287              : 
     288            1 :     return AicpuKernelLaunchDirect(comm, funcInfo, aicpuThreadHandle, opInfo->p2p.unfoldStream, userStream);
     289              : }
     290              : 
     291            1 : static HcclResult GetStreams(const CollComm *collComm, const std::vector<HcclP2pTask> &sortedSendQue,
     292              :     const std::vector<HcclP2pTask> &sortedRecvQue, aclrtStream &unfoldStream, const aclrtStream &usrStream)
     293              : {
     294            1 :     if (!sortedSendQue.empty()) {
     295            1 :         unfoldStream = sortedSendQue[0].stream;
     296            1 :         CHK_RET(collComm->groupScheduleMgr->SetUsrStream(sortedSendQue[0].usrStream));
     297            0 :     } else if (!sortedRecvQue.empty()) {
     298            0 :         unfoldStream = sortedRecvQue[0].stream;
     299            0 :         CHK_RET(collComm->groupScheduleMgr->SetUsrStream(sortedRecvQue[0].usrStream));
     300              :     } else {
     301            0 :         return HCCL_E_INTERNAL;
     302              :     }
     303            1 :     return HCCL_SUCCESS;
     304              : }
     305              : 
     306            1 : HcclResult groupLaunchA5()
     307              : {
     308            1 :     std::vector<HcclComm> hcclGroupCommListV2 = GetHcclGroupCommList();
     309            1 :     HCCL_INFO("[groupLaunchA5] to the start hcclGroupCommListV2.size[%u]", hcclGroupCommListV2.size());
     310              : 
     311            2 :     for (HcclComm comm : hcclGroupCommListV2) {
     312            1 :         hccl::hcclComm *hcclComm = static_cast<hccl::hcclComm *>(comm);
     313            1 :         CollComm *collComm = hcclComm->GetCollComm();
     314            1 :         CHK_PTR_NULL(collComm);
     315              :         
     316              :         /*新建send/recv流*/
     317              :         ThreadHandle sendRecv[2];
     318            1 :         CHK_RET(HcclThreadAcquire(comm, COMM_ENGINE_AICPU_TS, NUM_TWO, NUM_ONE, sendRecv));
     319            1 :         ThreadHandle aicpuSendThread = sendRecv[0], aicpuRecvThread = sendRecv[1];
     320            1 :         std::vector<HcclP2pTask> sortedSendQue, sortedRecvQue;
     321            1 :         CHK_RET(collComm->groupScheduleMgr->GetP2pTaskSchedule(sortedSendQue, sortedRecvQue));
     322            1 :         aclrtStream unfoldStream = nullptr, usrStream = nullptr;
     323            1 :         ThreadHandle cpuTsThread = 0, exportedAicpuTsThread = 0, exportedCpuTsSendThread = 0, exportedCpuTsRecvThread = 0;
     324            1 :         CHK_RET(GetStreams(collComm, sortedSendQue, sortedRecvQue, unfoldStream, usrStream));
     325              : 
     326            1 :         CHK_RET(collComm->groupScheduleMgr->GetUsrStream(usrStream));
     327            1 :         CHK_RET(HcclThreadAcquireWithStream(comm, COMM_ENGINE_CPU_TS, usrStream, NUM_THREE, &cpuTsThread));
     328            1 :         CHK_RET(HcclThreadExportToCommEngine(comm, NUM_ONE, &cpuTsThread, COMM_ENGINE_AICPU_TS, &exportedAicpuTsThread));
     329            1 :         CHK_RET(HcclThreadExportToCommEngine(comm, NUM_ONE, &aicpuSendThread, COMM_ENGINE_CPU_TS, &exportedCpuTsSendThread));
     330            1 :         CHK_RET(HcclThreadExportToCommEngine(comm, NUM_ONE, &aicpuRecvThread, COMM_ENGINE_CPU_TS, &exportedCpuTsRecvThread));
     331            1 :         CHK_RET(static_cast<HcclResult>(HcommThreadNotifyRecordOnThread(cpuTsThread, exportedCpuTsSendThread, NUM_ZERO)));
     332            1 :         CHK_RET(static_cast<HcclResult>(HcommThreadNotifyRecordOnThread(cpuTsThread, exportedCpuTsRecvThread, NUM_ZERO)));
     333              : 
     334              :         // 下发wait kernel
     335            1 :         CHK_RET(LaunchNotifyWaitToThread(comm, unfoldStream, aicpuSendThread, NUM_ZERO));
     336            1 :         CHK_RET(LaunchNotifyWaitToThread(comm, unfoldStream, aicpuRecvThread, NUM_ZERO));
     337              : 
     338              :         // Send/Recv交替执行以避免死锁
     339            2 :         for (size_t sendIdx = 0, recvIdx = 0; sendIdx < sortedSendQue.size() || recvIdx < sortedRecvQue.size();) {
     340            1 :             if (sendIdx < sortedSendQue.size()) {
     341            1 :                 CHK_RET(LaunchP2pExec(comm, sortedSendQue[sendIdx].stream, &sortedSendQue[sendIdx].funcInfo,
     342              :                     sortedSendQue[sendIdx].args, sortedSendQue[sendIdx].argSize, aicpuSendThread));
     343            1 :                 sendIdx++;
     344              :             }
     345            1 :             if (recvIdx < sortedRecvQue.size()) {
     346            1 :                 CHK_RET(LaunchP2pExec(comm, sortedRecvQue[recvIdx].stream, &sortedRecvQue[recvIdx].funcInfo,
     347              :                     sortedRecvQue[recvIdx].args, sortedRecvQue[recvIdx].argSize, aicpuRecvThread));
     348            1 :                 recvIdx++;
     349              :             }
     350              :         }
     351              : 
     352              :         // 下发record kernel
     353            1 :         CHK_RET(LaunchNotifyRecordToThread(comm, unfoldStream, aicpuSendThread, exportedAicpuTsThread, NUM_ONE));
     354            1 :         CHK_RET(LaunchNotifyRecordToThread(comm, unfoldStream, aicpuRecvThread, exportedAicpuTsThread, NUM_TWO));
     355              : 
     356            1 :         CHK_RET(static_cast<HcclResult>(HcommThreadNotifyWaitOnThreadWithDefaultTimeout(cpuTsThread, NUM_ONE)));
     357            1 :         CHK_RET(static_cast<HcclResult>(HcommThreadNotifyWaitOnThreadWithDefaultTimeout(cpuTsThread, NUM_TWO)));
     358            1 :     }
     359              : 
     360            1 :     SetHcclP2pTaskNums(0);
     361            1 :     ClearHcclGroupCommList();
     362              : 
     363            1 :     return HCCL_SUCCESS;
     364            1 : }
        

Generated by: LCOV version 2.0-1