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 "coll_send_executor.h"
      12              : 
      13              : namespace hccl {
      14              : 
      15            0 : CollSendExecutor::CollSendExecutor(const HcclDispatcher dispatcher,
      16            0 :                                    std::unique_ptr<TopoMatcher> &topoMatcher)
      17            0 :     : CollNativeExecutorBase(dispatcher, topoMatcher)
      18              : {
      19            0 :     DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
      20            0 : }
      21              : 
      22            0 : HcclResult CollSendExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
      23              : {
      24            0 :     HcclUs startut = TIME_NOW();
      25            0 :     tag_ = param.tag;
      26            0 :     algResResp_ = &algRes;
      27              : 
      28            0 :     HcclResult ret = HCCL_SUCCESS;
      29              :     // 图模式场景下不需要Loop
      30            0 :     if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
      31            0 :         DeviceMem InputMem = algRes.paramInputMem;
      32            0 :         ret = RunTemplate(param, InputMem);
      33            0 :     } else {
      34            0 :         ret = RunLoop(param, algRes);
      35              :     }
      36              : 
      37            0 :     CHK_PRT_RET(ret != HCCL_SUCCESS,
      38              :         HCCL_ERROR("[CollSendExecutor][Orchestrate]errNo[0x%016llx]send executor kernel run failed",
      39              :             HCCL_ERROR_CODE(ret)), ret);
      40              : 
      41            0 :     HCCL_INFO("tag[%s] Send Executor orchestrate success, take time [%lld]us.",
      42              :         param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
      43            0 :     return HCCL_SUCCESS;
      44              : }
      45              : 
      46            0 : HcclResult CollSendExecutor::GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo)
      47              : {
      48            0 :     algResResp_ = &algRes;
      49              : 
      50            0 :     SubCommInfo level1CommInfo = GetSubCommInfo(COMM_COMBINE, 0);
      51            0 :     if (level1CommInfo.links.size() == 0) {
      52            0 :         HCCL_ERROR("[CollSendExecutor]links size is 0");
      53              :     }
      54            0 :     u32 localRank= level1CommInfo.localRank;
      55            0 :     u32 localRankSize = level1CommInfo.localRankSize;
      56              : 
      57            0 :     if (localRankSize == 1) {
      58            0 :         return HCCL_SUCCESS;
      59              :     }
      60            0 :     u32 ringNextRank = (localRank + 1) % localRankSize;
      61              : 
      62            0 :     adjInfo.dstRankNum = 1;
      63            0 :     NslbDpAdjInfo adjInfoStep = {0};
      64            0 :     adjInfoStep.dstLocalRankId = ringNextRank;
      65            0 :     adjInfoStep.phaseId = 1;
      66            0 :     adjInfoStep.rev = 0;
      67            0 :     adjInfo.nsAdjInfo.push_back(adjInfoStep);
      68              : 
      69            0 :     HCCL_INFO("[nslbdp]GetAdjInfo localRank[%u], phaseId[%u].",localRank, ringNextRank);
      70            0 :     return HCCL_SUCCESS;
      71            0 : }
      72              : 
      73            0 : HcclResult CollSendExecutor::CalcTransportMemType(TransportMemType &inputType, TransportMemType &outputType)
      74              : {
      75            0 :     if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
      76            0 :         inputType = TransportMemType::CCL_INPUT;
      77            0 :         outputType = TransportMemType::CCL_INPUT;
      78              :     } else {
      79            0 :         inputType = TransportMemType::PARAM_INPUT;
      80            0 :         outputType = TransportMemType::PARAM_OUTPUT;
      81              :     }
      82            0 :     HCCL_INFO("[CollSendExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
      83              :         tag_.c_str(), inputType, outputType);
      84            0 :     return HCCL_SUCCESS;
      85              : }
      86              : 
      87            0 : HcclResult CollSendExecutor::CalcP2PCommInfo(TransportMemType inputType,
      88              :     TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport, u32 dstRank)
      89              : {
      90            0 :     HCCL_INFO("[CollSendExecutor][CalcLevel0CommInfo]tag[%s] start", tag_.c_str());
      91            0 :     CommParaInfo commP2P(COMM_COMBINE, CommType::COMM_TAG_P2P);
      92            0 :     commP2P.peerUserRank = dstRank;
      93            0 :     CHK_RET(CalcCommPlaneInfo(tag_, commP2P, opTransport[COMM_COMBINE], inputType, outputType));
      94            0 :     HCCL_INFO("[CollSendExecutor][CalcLevel0CommInfo]tag[%s] Calc P2PComm finish", tag_.c_str());
      95            0 :     return HCCL_SUCCESS;
      96            0 : }
      97              : 
      98            0 : HcclResult CollSendExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport, u32 dstRank)
      99              : {
     100            0 :     TransportMemType inputType = TransportMemType::RESERVED;
     101            0 :     TransportMemType outputType = TransportMemType::RESERVED;
     102            0 :     CalcTransportMemType(inputType, outputType);
     103            0 :     CHK_RET(CalcP2PCommInfo(inputType, outputType, opTransport, dstRank));
     104            0 :     return HCCL_SUCCESS;
     105              : }
     106              : 
     107            0 : HcclResult CollSendExecutor::CalcResRequest(const OpParam& param, AlgResourceRequest& resourceRequest)
     108              : {
     109            0 :     ParseParam(param);
     110              : 
     111            0 :     u64 scratchMemSize = 0U;
     112            0 :     u32 streamNum = 0U;
     113            0 :     u32 notifyNum = 0U;
     114            0 :     u64 aivBufferRequest = 0U;
     115              :     std::vector<LevelNSubCommTransport> opTransport {
     116            0 :         std::vector<LevelNSubCommTransport>(static_cast<u32>(COMM_LEVEL_RESERVED))
     117            0 :     };
     118              : 
     119            0 :     CHK_RET(CalcCommInfo(opTransport, param.dstRank));
     120              : 
     121            0 :     CHK_RET(BuildResourceRequest(scratchMemSize, streamNum, notifyNum, aivBufferRequest, opTransport, resourceRequest));
     122            0 :     HCCL_INFO("streamNum[%u], notifyNum[%u], sctrachMemSize[%llu], aivBufferRequest[%llu]",
     123              :         resourceRequest.streamNum, resourceRequest.notifyNum, resourceRequest.scratchMemSize,
     124              :         resourceRequest.aivBufferRequest);
     125              :     // 打印建链诉求
     126            0 :     PrintTransportRequest(resourceRequest);
     127            0 :     return HCCL_SUCCESS;
     128            0 : }
     129              : 
     130            0 : HcclResult CollSendExecutor::RunLoop(OpParam &param, AlgResourceResponse &algRes)
     131              : {
     132              :     HcclResult ret;
     133              : 
     134            0 :     u64 commInputSize = algRes.cclInputMem.size();
     135              : 
     136            0 :     u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
     137            0 :     auto meta = HcclOpMetaInfo::GetOneForSend();
     138            0 :     u8 *curInputPtr = static_cast<u8 *>(param.inputPtr);
     139            0 :     CHK_PTR_NULL(curInputPtr);
     140              : 
     141            0 :     u64 inputOffset = 0;
     142            0 :     u64 countLeft = param.DataDes.count;
     143            0 :     while (countLeft > 0) {
     144            0 :         CHK_RET(InitTask(dispatcher_, const_cast<Stream&>(param.stream), meta.isEnableCache, meta.GetCacheKey()));
     145            0 :         curInputPtr += inputOffset;
     146              : 
     147            0 :         HCCL_DEBUG("SendOutPlace:inputOffset[%llu]", inputOffset);
     148            0 :         u64 curCount = ((countLeft * unitSize) > commInputSize) ? (commInputSize / unitSize) : countLeft;
     149            0 :         u64 curSize = curCount * unitSize; // 单位 byte
     150              : 
     151            0 :         HCCL_DEBUG("SendOutPlace:curInputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr, curCount, curSize);
     152              : 
     153              :         // 非A3场景，或A3的SDMA消减场景，send端不消减
     154            0 :         DeviceMem inCommMem(algRes.cclInputMem.ptr(), curSize);
     155            0 :         DeviceMem inMem(curInputPtr, curSize);
     156            0 :         CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inCommMem, inMem, const_cast<Stream&>(param.stream)));
     157            0 :         HCCL_INFO("[CollSendExecutor] inCommMem[%p]size[%lld], inMem[%p]size[%lld]", inCommMem.ptr(), inCommMem.size(), inMem.ptr(), inMem.size());
     158            0 :         ret = RunTemplate(param, inCommMem);
     159            0 :         CHK_PRT_RET(ret != HCCL_SUCCESS,
     160              :             HCCL_ERROR("errNo[0x%016llx] SendOutPlace: send error, tag[%s], ptr[%p], count[%llu], dataType[%d]",
     161              :             HCCL_ERROR_CODE(ret), param.tag.c_str(), curInputPtr, curCount, param.DataDes.dataType),
     162              :             ret);
     163            0 :         HCCL_DEBUG("[CollSendExecutor][RunLoop]copy from user input to ccl input.");
     164              : 
     165            0 :         CHK_PRT_RET((curCount == 0), HCCL_ERROR("In OP_BASE curCount is zero"), HCCL_E_PARA);
     166            0 :         countLeft -= curCount;
     167            0 :         inputOffset = curSize;
     168              : 
     169            0 :         CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
     170            0 :     }
     171            0 :     return HCCL_SUCCESS;
     172              : }
     173              : 
     174            0 : HcclResult CollSendExecutor::RunTemplate(const OpParam &param, DeviceMem &inputMem)
     175              : {
     176            0 :     SubCommInfo commInfo = GetSubCommInfo(COMM_COMBINE, 0);
     177            0 :     if (commInfo.links.size() == 0) {
     178            0 :         HCCL_ERROR("[CollSendExecutor]links size is 0");
     179              :     }
     180            0 :     LINK transportLink = commInfo.links[0];
     181              : 
     182            0 :     SendReceive sendExecutor(dispatcher_, transportLink);
     183            0 :     CHK_RET(sendExecutor.SendPrepare(inputMem, param.dstRank, param.stream));
     184            0 :     CHK_RET(sendExecutor.RegisterProfiler(0, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
     185              : 
     186            0 :     CHK_RET(sendExecutor.SendRunAsync());
     187              : 
     188            0 :     return HCCL_SUCCESS;
     189            0 : }
     190              : 
     191              : REGISTER_EXEC("SendExecutor", Send, CollSendExecutor);
     192              : 
     193              : } // namespace hcclss