LCOV - code coverage report
Current view: top level - legacy/ascend910/algorithm/impl/coll_executor/coll_reduce_scatter - coll_reduce_scatter_comm_executor.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 54.5 % 123 67
Test Date: 2026-08-18 17:47:01 Functions: 70.0 % 10 7

            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_reduce_scatter_comm_executor.h"
      12              : #include "alg_template_register.h"
      13              : 
      14              : namespace hccl {
      15              : 
      16            6 : CollReduceScatterCommExecutor::CollReduceScatterCommExecutor(
      17            6 :     const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
      18            6 :     : CollReduceScatterExecutor(dispatcher, topoMatcher)
      19              : {
      20            6 :     desc_.deterministic = 1;
      21            6 :     DMAReduceFlag_ = false;
      22            6 : }
      23              : 
      24           12 : void CollReduceScatterCommExecutor::ParseParam(const OpParam& param)
      25              : {
      26           12 :     tag_ = param.tag;
      27              : 
      28           12 :     if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
      29            0 :         scratchMemFlag_ = false;
      30              :     } else {
      31           12 :         scratchMemFlag_ = true;
      32              :     }
      33              : 
      34              :     // 记录图模式总数据量
      35           12 :     totalSize_ = topoAttr_.userRankSize * param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
      36           12 :     aicpuUnfoldMode_ = param.aicpuUnfoldMode;
      37           12 : }
      38              : 
      39            6 : HcclResult CollReduceScatterCommExecutor::CalcScratchMemSize(u64& scratchMemSize)
      40              : {
      41            6 :     if (scratchMemFlag_) {
      42            6 :         if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
      43            0 :             scratchMemSize = inCCLbufferSize_;
      44              :         } else {
      45            6 :             scratchMemSize = totalSize_;
      46              :         }
      47              :     } else {
      48            0 :         scratchMemSize = 0U;
      49              :     }
      50              : 
      51            6 :     HCCL_INFO(
      52              :         "[CollReduceScatterCommExecutor][CalcScratchMemSize] tag[%s] scratchMemSize[%llu]", tag_.c_str(),
      53              :         scratchMemSize);
      54            6 :     return HCCL_SUCCESS;
      55              : }
      56              : 
      57            0 : bool CollReduceScatterCommExecutor::IsSmallData(
      58              :     [[maybe_unused]] const u64 totalSize, [[maybe_unused]] const u64 curSize)
      59              : {
      60            0 :     return topoAttr_.deviceType == DevType::DEV_TYPE_910_93;
      61              : }
      62              : 
      63            6 : HcclResult CollReduceScatterCommExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
      64              : {
      65            6 :     TransportMemType inputType = TransportMemType::RESERVED;
      66            6 :     TransportMemType outputType = TransportMemType::RESERVED;
      67            6 :     CHK_RET(CalcTransportMemType(inputType, outputType));
      68            6 :     CHK_RET(CalcCombinedCommInfo(inputType, outputType, opTransport));
      69            6 :     return HCCL_SUCCESS;
      70              : }
      71              : 
      72              : HcclResult
      73            6 : CollReduceScatterCommExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
      74              : {
      75            6 :     if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
      76            0 :         inputType = TransportMemType::CCL_INPUT;
      77            0 :         if (scratchMemFlag_) {
      78            0 :             outputType = TransportMemType::SCRATCH;
      79              :         } else {
      80            0 :             outputType = TransportMemType::CCL_OUTPUT;
      81              :         }
      82              :     } else {
      83            6 :         inputType = TransportMemType::PARAM_INPUT;
      84            6 :         if (scratchMemFlag_) {
      85            6 :             outputType = TransportMemType::SCRATCH;
      86              :         } else {
      87            0 :             outputType = TransportMemType::PARAM_OUTPUT;
      88              :         }
      89              :     }
      90            6 :     HCCL_INFO(
      91              :         "[CollReduceScatterCommExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]", tag_.c_str(),
      92              :         inputType, outputType);
      93            6 :     return HCCL_SUCCESS;
      94              : }
      95              : 
      96            6 : HcclResult CollReduceScatterCommExecutor::CalcCombinedCommInfo(
      97              :     TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
      98              : {
      99            6 :     CommPlane commPlane = COMM_COMBINE_ORDER;
     100              : 
     101            6 :     CommParaInfo commParaInfo(commPlane, CommType::COMM_TAG_MAX);
     102              : 
     103            6 :     if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
     104            0 :         commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
     105            6 :     } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
     106            0 :         commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING_V1;
     107            6 :     } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
     108            0 :         commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
     109            6 :     } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
     110            0 :         commParaInfo.commType = CommType::COMM_TAG_HALVING_DOUBLING;
     111              :     } else {
     112            6 :         commParaInfo.commType = CommType::COMM_TAG_RING_INNER;
     113              :     }
     114            6 :     CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[commPlane], inputType, outputType));
     115              : 
     116            6 :     return HCCL_SUCCESS;
     117            6 : }
     118              : 
     119            0 : u64 CollReduceScatterCommExecutor::CalcLoopMaxCount(const u32 unitSize)
     120              : {
     121              :     // 中转内存单次最多能够接受的output count
     122            0 :     u64 maxCountPerLoop
     123            0 :         = inCCLbufferSize_ / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
     124            0 :     return maxCountPerLoop;
     125              : }
     126              : 
     127            0 : bool CollReduceScatterCommExecutor::IsHugeData(const u64 curSize, [[maybe_unused]] OpParam* param)
     128              : {
     129            0 :     bool hugeData = (curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE)
     130            0 :                     || (curSize > SDMA_SEND_MAX_SIZE);
     131            0 :     return hugeData;
     132              : }
     133              : 
     134            6 : HcclResult CollReduceScatterCommExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
     135              : {
     136            6 :     HCCL_CONFIG_INFO(HCCL_ALG, "[%s] userRank[%u] starts.", __func__, topoAttr_.userRank);
     137            6 :     CommPlane commPlane = COMM_COMBINE_ORDER;
     138              : 
     139            6 :     CHK_RET(CheckCommSize(commPlane, COMM_INDEX_0 + 1));
     140            6 :     SubCommInfo combinedCommInfo = GetSubCommInfo(commPlane, COMM_INDEX_0);
     141              : 
     142            6 :     u64 reduceAttr = GetReduceAttr(execMem.inputMem, execMem.outputMem, param.DataDes.dataType, param.reduceType);
     143              : 
     144              :     // 构造ring algorithm对应的reduce-scatter实例
     145            6 :     std::unique_ptr<AlgTemplateBase> tempAlg;
     146            6 :     if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
     147            0 :         tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NHR, dispatcher_);
     148            0 :         HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_COMBINE_ORDER", __func__);
     149            0 :         CHK_SMART_PTR_NULL(tempAlg);
     150            0 :         CHK_RET(tempAlg->Prepare(reduceAttr, false));
     151            0 :         CHK_RET(tempAlg->Prepare(
     152              :             execMem.inputMem, execMem.outputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType,
     153              :             param.stream, param.reduceType));
     154            0 :         if (topoAttr_.deviceType != DevType::DEV_TYPE_910_93 || algoAttr_.isSupportAtomicWrite) {
     155            0 :             tempAlg->CloseBarrier();
     156              :         }
     157            0 :         CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
     158            6 :     } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
     159              :         tempAlg
     160            0 :             = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NHR_V1, dispatcher_);
     161            0 :         HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR_V1 in COMM_COMBINE_ORDER", __func__);
     162            0 :         CHK_SMART_PTR_NULL(tempAlg);
     163            0 :         CHK_RET(tempAlg->Prepare(reduceAttr));
     164            0 :         CHK_RET(tempAlg->Prepare(
     165              :             execMem.inputMem, execMem.outputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType,
     166              :             param.stream, param.reduceType));
     167            0 :         CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
     168            6 :     } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
     169            0 :         tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NB, dispatcher_);
     170            0 :         HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_COMBINE_ORDER", __func__);
     171            0 :         CHK_SMART_PTR_NULL(tempAlg);
     172            0 :         CHK_RET(tempAlg->Prepare(reduceAttr));
     173            0 :         CHK_RET(tempAlg->Prepare(
     174              :             execMem.inputMem, execMem.outputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType,
     175              :             param.stream, param.reduceType));
     176            0 :         CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
     177            6 :     } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
     178            0 :         tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
     179            0 :             TemplateType::TEMPLATE_REDUCESCATTER_RECURSIVE_HD, dispatcher_);
     180            0 :         HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RECURSIVE_HD in COMM_COMBINE_ORDER", __func__);
     181            0 :         CHK_SMART_PTR_NULL(tempAlg);
     182            0 :         CHK_RET(tempAlg->Prepare(reduceAttr));
     183            0 :         DeviceMem scratchMem = execMem.scratchMem.range(0, execMem.inputMem.size());
     184            0 :         u64 inputDataCount = execMem.inputMem.size() / SIZE_TABLE[param.DataDes.dataType];
     185            0 :         CHK_RET(tempAlg->Prepare(
     186              :             execMem.inputMem, execMem.inputMem, scratchMem, inputDataCount, param.DataDes.dataType, param.stream,
     187              :             param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0)));
     188            0 :         CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
     189            0 :         u64 dataSize = execMem.count * SIZE_TABLE[param.DataDes.dataType];
     190            0 :         DeviceMem srcMem = execMem.inputMem.range(dataSize * topoAttr_.userRank, dataSize);
     191            0 :         DeviceMem dstMem = execMem.outputMem.range(0, dataSize);
     192            0 :         CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
     193            0 :     } else {
     194              :         tempAlg
     195            6 :             = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_RING, dispatcher_);
     196            6 :         HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_COMBINE_ORDER", __func__);
     197            6 :         CHK_SMART_PTR_NULL(tempAlg);
     198            6 :         CHK_RET(tempAlg->Prepare(reduceAttr));
     199           30 :         CHK_RET(tempAlg->Prepare(
     200              :             execMem.inputMem, execMem.inputMem, execMem.scratchMem, execMem.count, param.DataDes.dataType, param.stream,
     201              :             param.reduceType));
     202            6 :         CHK_RET(RunTemplate(tempAlg, combinedCommInfo));
     203              :         // 将cclInBuffer中与userRank_对应的部分拷贝至cclOutBuffer
     204            6 :         u64 dataSize = execMem.count * SIZE_TABLE[param.DataDes.dataType];
     205            6 :         DeviceMem srcMem = execMem.inputMem.range(dataSize * topoAttr_.userRank, dataSize);
     206            6 :         DeviceMem dstMem = execMem.outputMem.range(0, dataSize);
     207            6 :         CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream)));
     208            6 :     }
     209            6 :     return HCCL_SUCCESS;
     210            6 : }
     211              : 
     212              : REGISTER_EXEC("ReduceScatterComm", ReduceScatterComm, CollReduceScatterCommExecutor);
     213              : } // namespace hccl
        

Generated by: LCOV version 2.0-1