LCOV - code coverage report
Current view: top level - legacy/ascend910/algorithm/impl/coll_executor/coll_all_gather_v - coll_all_gather_v_mesh_executor.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 0.0 % 218 0
Test Date: 2026-08-04 10:52:23 Functions: 0.0 % 17 0

            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_all_gather_v_mesh_executor.h"
      12              : 
      13              : #include <algorithm>
      14              : #include <numeric>
      15              : 
      16              : namespace hccl {
      17            0 : CollAllGatherVMeshExecutor::CollAllGatherVMeshExecutor(const HcclDispatcher dispatcher,
      18            0 :                                                                     std::unique_ptr<TopoMatcher> &topoMatcher)
      19            0 :     : CollAllGatherVExecutor(dispatcher, topoMatcher)
      20              : {
      21            0 :     DMAReduceFlag_ = (topoAttr_.moduleNum <= 1);
      22            0 : }
      23              : 
      24            0 : HcclResult CollAllGatherVMeshExecutor::CalcStreamNum(u32 &streamNum)
      25              : {
      26            0 :     u32 totalStreamNum = 0;
      27            0 :     if (topoAttr_.moduleNum > 1) {
      28            0 :         totalStreamNum = topoAttr_.deviceNumPerAggregation > 1U ? topoAttr_.deviceNumPerAggregation - 1U : 1U;
      29              :     } else {
      30            0 :         totalStreamNum = topoAttr_.deviceNumPerAggregation;
      31              :     }
      32            0 :     streamNum = totalStreamNum - 1U;
      33            0 :     HCCL_INFO("[CollAllGatherVMeshExecutor][CalcStreamNum] tag[%s] streamNum[%u]",
      34              :                 tag_.c_str(), streamNum);
      35            0 :     return HCCL_SUCCESS;
      36              : }
      37              : 
      38            0 : HcclResult CollAllGatherVMeshExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport> &opTransport)
      39              : {
      40            0 :     TransportMemType inputType = TransportMemType::RESERVED;
      41            0 :     TransportMemType outputType = TransportMemType::RESERVED;
      42            0 :     CHK_RET(CalcTransportMemType(inputType, outputType));
      43            0 :     CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
      44            0 :     if (topoAttr_.moduleNum > 1) {
      45            0 :         CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
      46              :     }
      47              : 
      48            0 :     return HCCL_SUCCESS;
      49              : }
      50              : 
      51            0 : HcclResult CollAllGatherVMeshExecutor::CalcTransportMemType(TransportMemType &inputType,
      52              :                                                                     TransportMemType &outputType)
      53              : {
      54            0 :     if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
      55            0 :         inputType = TransportMemType::CCL_INPUT;
      56            0 :         outputType = TransportMemType::CCL_OUTPUT;
      57              :     } else {
      58            0 :         inputType = TransportMemType::PARAM_INPUT;
      59            0 :         outputType = TransportMemType::PARAM_OUTPUT;
      60              :     }
      61            0 :     HCCL_INFO("[CollAllGatherVMeshExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
      62              :                 tag_.c_str(), inputType, outputType);
      63            0 :     return HCCL_SUCCESS;
      64              : }
      65              : 
      66            0 : HcclResult CollAllGatherVMeshExecutor::CalcLevel0CommInfo(TransportMemType inputType,
      67              :                                                                 TransportMemType outputType,
      68              :                                                                 std::vector<LevelNSubCommTransport> &opTransport)
      69              : {
      70            0 :     CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
      71            0 :     CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
      72            0 :     return HCCL_SUCCESS;
      73            0 : }
      74              : 
      75            0 : HcclResult CollAllGatherVMeshExecutor::CalcLevel1CommInfo(TransportMemType inputType,
      76              :     TransportMemType outputType,
      77              :     std::vector<LevelNSubCommTransport>& opTransport)
      78              : {
      79            0 :     HCCL_INFO("[CollAllGatherVMeshExecutor][CalcLevel1CommInfo]tag[%s] start", tag_.c_str());
      80            0 :     CommParaInfo commParaLevel1(COMM_LEVEL1, CommType::COMM_TAG_MAX);
      81            0 :     if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING || (topoAttr_.isDiffDeviceModule && topoAttr_.serverNum == 1)) {
      82            0 :         commParaLevel1.commType = CommType::COMM_TAG_RING_INNER;
      83            0 :         HCCL_INFO("[CollAllGatherVMeshExecutor][CalcLevel1CommInfo]tag[%s] Calc RingCommInfo", tag_.c_str());
      84            0 :     } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
      85            0 :         commParaLevel1.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
      86            0 :         HCCL_INFO("[CollAllGatherVMeshExecutor][CalcLevel1CommInfo]tag[%s] Calc NBCommInfo", tag_.c_str());
      87              :     } else {
      88            0 :         commParaLevel1.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
      89            0 :         HCCL_INFO("[CollAllGatherVMeshExecutor][CalcLevel1CommInfo]tag[%s] Calc NHRCommInfo", tag_.c_str());
      90              :     }
      91            0 :     commParaLevel1.forceRdma = false;
      92            0 :     CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel1, opTransport[commParaLevel1.commPlane], inputType, outputType));
      93              : 
      94            0 :     HCCL_INFO("[CollAllGatherVMeshExecutor][COMM_LEVEL1]tag[%s] Calc CommInfo Finish", tag_.c_str());
      95              : 
      96            0 :     return HCCL_SUCCESS;
      97            0 : }
      98              : 
      99            0 : u64 CollAllGatherVMeshExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
     100              : {
     101              :     u64 maxCountPerLoop;
     102            0 :     if (topoAttr_.moduleNum > 1) {
     103            0 :         maxCountPerLoop = cclBuffSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
     104              :     } else {
     105            0 :         maxCountPerLoop = (cclBuffSize - HCCL_MIN_SLICE_ALIGN_910B) / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
     106              :     }
     107              : 
     108            0 :     return maxCountPerLoop;
     109              : }
     110              : 
     111            0 : bool CollAllGatherVMeshExecutor::IsHugeData(const u64 curSize)
     112              : {
     113            0 :     bool hugeData = curSize * topoAttr_.userRankSize > RDMA_SEND_MAX_SIZE || curSize > SDMA_SEND_MAX_SIZE;
     114            0 :     return hugeData;
     115              : }
     116              : 
     117            0 : HcclResult CollAllGatherVMeshExecutor::RunLevel0(const OpParam &param, ExecMem &execMem,
     118              :                                                         SubCommInfo &level0CommInfo, const SubCommInfo &level1CommInfo)
     119              : {
     120            0 :     HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherVMeshExecutor][KernelRun] userRank[%u] starts.", topoAttr_.userRank);
     121            0 :     u32 perDataSize = SIZE_TABLE[param.VDataDes.dataType];
     122            0 :     const auto counts = static_cast<u64 *>(param.VDataDes.counts);
     123            0 :     u32 serverIndex = level1CommInfo.localRank;
     124            0 :     u32 rankBaseOffset = serverIndex * level0CommInfo.localRankSize;
     125            0 :     u64 countBaseOffset = std::accumulate(counts, counts + rankBaseOffset, 0ULL);
     126              :     // level0的rank 0在整个通信域中的偏移
     127            0 :     u64 baseOffset = countBaseOffset * perDataSize;
     128              :     // allgatherv 计算slice,数据分成ranksize份,每份的起始偏移和大小
     129            0 :     std::vector<Slice> outputSlices;
     130            0 :     u64 outputMemSize = 0;
     131            0 :     for (u32 rank = rankBaseOffset; rank < rankBaseOffset + level0CommInfo.localRankSize; ++rank) {
     132            0 :         Slice userslice;
     133            0 :         userslice.offset = outputMemSize;
     134            0 :         userslice.size = counts[rank] * perDataSize;
     135            0 :         outputSlices.emplace_back(std::move(userslice));
     136            0 :         outputMemSize += userslice.size;
     137              :     }
     138              : 
     139            0 :     u64 inputMemSize = outputSlices[level0CommInfo.localRank].size;
     140            0 :     u64 level0Offset = outputSlices[level0CommInfo.localRank].offset;
     141            0 :     DeviceMem srcMem = execMem.inputMem.range(0, inputMemSize);
     142            0 :     DeviceMem dstMem = execMem.outputMem.range(baseOffset + level0Offset, inputMemSize);
     143            0 :     CHK_SMART_PTR_NULL(dstMem);
     144            0 :     Stream stream = param.stream;
     145              :     // 将数据从input内存拷贝到output内存的对应位置
     146            0 :     HcclResult ret = HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream);
     147            0 :     CHK_PRT_RET(ret != HCCL_SUCCESS,
     148              :                 HCCL_ERROR("[CollAllGatherVMeshExecutor][KernelRun]all gatherV mesh memcpy Failed, Offset[%llu], Size[%llu].",
     149              :                             level0Offset, inputMemSize),
     150              :                 ret);
     151              : 
     152            0 :     CHK_RET(ActiveSlaveStreams(param.stream));
     153              : 
     154              :     //  抽取当前用于多环all gather 的output内存数据
     155            0 :     DeviceMem currentOutputMem = execMem.outputMem.range(baseOffset, outputMemSize);
     156            0 :     CHK_SMART_PTR_NULL(currentOutputMem);
     157              : 
     158            0 :     std::unique_ptr<AlgTemplateBase> level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
     159            0 :         TemplateType::TEMPLATE_ALL_GATHER_MESH_ATOMIC, dispatcher_);
     160            0 :     CHK_SMART_PTR_NULL(level0TempAlg);
     161            0 :     CHK_RET(level0TempAlg->Prepare(algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
     162              :                                     topoAttr_.userRank, nullptr, level0CommInfo.localRank, level0CommInfo.localRankSize));
     163            0 :     CHK_RET(level0TempAlg->Prepare(currentOutputMem, currentOutputMem, execMem.inputMem,
     164              :                                     execMem.count, param.VDataDes.dataType, param.stream, HCCL_REDUCE_RESERVED,
     165              :                                     LEVEL0_BRIDGE_RANK_ID, outputSlices, baseOffset));
     166            0 :     u32 rankSize = level0CommInfo.localRankSize;
     167            0 :     CHK_RET(level0TempAlg->RegisterProfiler((rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
     168              :                                             PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
     169            0 :     CHK_RET(RunTemplate(level0TempAlg, level0CommInfo));
     170            0 :     HCCL_INFO("[CollAllGatherVMeshExecutor][RunLevel0] level 0 for A2 run success");
     171            0 :     return HCCL_SUCCESS;
     172            0 : }
     173              : 
     174            0 : HcclResult CollAllGatherVMeshExecutor::RunLevel1(const OpParam &param, ExecMem &execMem,
     175              :                                                  const SubCommInfo &level0CommInfo, SubCommInfo &level1CommInfo)
     176              : {
     177            0 :     std::unique_ptr<AlgTemplateBase> level1TempAlg;
     178            0 :     if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING || (topoAttr_.isDiffDeviceModule && topoAttr_.serverNum == 1)) {
     179              :         // 1-单server-SDMA
     180            0 :         level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
     181            0 :             TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
     182            0 :         HCCL_INFO("allgatherv mesh: using ring algo inter-server.");
     183            0 :     } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
     184            0 :         level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
     185            0 :             TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
     186            0 :         HCCL_INFO("allgatherv mesh: using nonuniform-bruck algo inter-server.");
     187              :     } else {
     188              :         //使用nhr作为兜底算法
     189            0 :         level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
     190            0 :             TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
     191            0 :         level1TempAlg->CloseBarrier();
     192            0 :         HCCL_INFO("allgatherv mesh: using nhr algo inter-server.");
     193              :     } 
     194            0 :     CHK_SMART_PTR_NULL(level1TempAlg);
     195              : 
     196            0 :     u32 perDataSize = SIZE_TABLE[param.VDataDes.dataType];
     197            0 :     const auto counts = static_cast<u64 *>(param.VDataDes.counts);
     198              :     // allgatherv 计算slice,数据分成level1 ranksize份,每份的起始偏移和大小
     199            0 :     std::vector<Slice> outputSlices;
     200            0 :     u64 outputMemSize = 0;
     201            0 :     for (u32 rankLevel1 = 0; rankLevel1 < level1CommInfo.localRankSize; ++rankLevel1) {
     202            0 :         Slice userslice;
     203              :         // 计算偏移值
     204            0 :         u64 countLevel0 = std::accumulate(counts + rankLevel1 * level0CommInfo.localRankSize,
     205            0 :                                             counts + (rankLevel1 + 1) * level0CommInfo.localRankSize, 0ULL);
     206            0 :         userslice.offset = outputMemSize;
     207            0 :         userslice.size = countLevel0 * perDataSize;
     208            0 :         outputSlices.emplace_back(std::move(userslice));
     209            0 :         outputMemSize += userslice.size;
     210              :     }
     211              : 
     212              :     //  此处虽然带入inputMem作为scratch mem, 但inputMem 不能被使用
     213            0 :     CHK_RET(level1TempAlg->Prepare(execMem.outputMem, execMem.outputMem, execMem.inputMem,
     214              :                                     outputSlices[level1CommInfo.localRank].size / perDataSize,
     215              :                                     param.VDataDes.dataType, param.stream, HcclReduceOp::HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID,
     216              :                                     outputSlices, 0));
     217              : 
     218            0 :     u32 rankSize = level1CommInfo.localRankSize;
     219            0 :     CHK_RET(level1TempAlg->RegisterProfiler((rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank,
     220              :                                             PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
     221              : 
     222            0 :     CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
     223            0 :     HCCL_INFO("[CollAllGatherVMeshExecutor][RunLevel1] level 1 for A2 run success");
     224              : 
     225            0 :     return HCCL_SUCCESS;
     226            0 : }
     227              : 
     228            0 : HcclResult CollAllGatherVMeshExecutor::RunSingleMesh(const OpParam &param, ExecMem &execMem)
     229              : {
     230            0 :     HcclDataType dataType = HCCL_DATA_TYPE_RESERVED;
     231            0 :     dataType = param.VDataDes.dataType;
     232            0 :     const u32 unitSize = SIZE_TABLE[dataType];
     233              : 
     234            0 :     CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
     235            0 :     SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
     236            0 :     u32 rankSize = level0CommInfo.localRankSize;
     237              : 
     238              :     // DMA消减后仅使用ccl out通信,ccl out根据实际使用大小重新申请内存空间
     239            0 :     u64 inputMemSize = execMem.inputMem.size();
     240            0 :     u64 baseOffset = 0;
     241            0 :     DeviceMem curOutputMem = execMem.outputMem.range(baseOffset, inputMemSize);
     242            0 :     CHK_SMART_PTR_NULL(curOutputMem);
     243              : 
     244              :     // allgatherv 计算slice,数据分成ranksize份,每份的起始偏移和大小
     245            0 :     std::vector<Slice> outputSlices;
     246            0 :     const auto counts = static_cast<u64 *>(param.VDataDes.counts);
     247            0 :     const auto displs = static_cast<u64 *>(param.VDataDes.displs);
     248            0 :     for (u32 rank = 0; rank < rankSize; ++rank) {
     249            0 :         Slice userslice;
     250            0 :         userslice.offset = displs[rank] * unitSize;
     251            0 :         userslice.size = counts[rank] * unitSize;
     252            0 :         outputSlices.emplace_back(std::move(userslice));
     253              :     }
     254              : 
     255              :     // DMA消减场景,打包opInfo
     256            0 :     HcomCollOpInfo opInfo = {"", execMem.inputPtr, execMem.outputPtr, execMem.count, dataType,
     257            0 :                                 param.root, param.reduceType};
     258              : 
     259            0 :     std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
     260            0 :         TemplateType::TEMPLATE_ALL_GATHER_MESH_DIRECT, dispatcher_);
     261            0 :     CHK_SMART_PTR_NULL(tempAlg);
     262            0 :     CHK_RET(tempAlg->Prepare(algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
     263              :                                 topoAttr_.userRank, &opInfo, level0CommInfo.localRank, level0CommInfo.localRankSize));
     264              : 
     265            0 :     CHK_RET(tempAlg->Prepare(curOutputMem, curOutputMem, execMem.inputMem, execMem.count,
     266              :                                 dataType, param.stream, HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, outputSlices, baseOffset));
     267              : 
     268            0 :     CHK_RET(tempAlg->RegisterProfiler(
     269              :         (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
     270              :         PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
     271              : 
     272            0 :     CHK_RET(RunTemplate(tempAlg, level0CommInfo));
     273              : 
     274            0 :     HCCL_INFO("[CollAllGatherVMeshExecutor][RunSingleMesh] single mesh for A2 run success");
     275            0 :     return HCCL_SUCCESS;
     276            0 : }
     277              : 
     278            0 : HcclResult CollAllGatherVMeshExecutor::KernelRun(const OpParam &param, ExecMem &execMem)
     279              : {
     280            0 :     HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherVMeshExecutor][KernelRun] userRank[%u] starts.", topoAttr_.userRank);
     281            0 :     if (topoAttr_.moduleNum > 1) {
     282              :         // 获取子通信域信息
     283            0 :         CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
     284            0 :         SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
     285            0 :         CHK_RET(CheckCommSize(COMM_LEVEL1, level0CommInfo.localRank + 1));
     286            0 :         SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0CommInfo.localRank);
     287              : 
     288            0 :         CHK_RET(RunLevel0(param, execMem, level0CommInfo, level1CommInfo));
     289            0 :         CHK_RET(RunLevel1(param, execMem, level0CommInfo, level1CommInfo));
     290            0 :     } else {
     291            0 :         CHK_RET(RunSingleMesh(param, execMem));
     292              :     }
     293              : 
     294            0 :     return HCCL_SUCCESS;
     295              : }
     296              : 
     297            0 : HcclResult CollAllGatherVMeshExecutor::CalcCurCountsAndCurDisplsMultiModule(const u64 maxTotalCount,
     298              :     std::vector<u64> &countsLeft, std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls,
     299              :     bool &finished)
     300              : {
     301            0 :     curCounts = std::vector<u64>(countsLeft.size(), 0);
     302            0 :     curDispls = std::vector<u64>(displs.size(), 0);
     303            0 :     auto allocatableCount = maxTotalCount;
     304              : 
     305            0 :     HCCL_DEBUG("CalcCurCountsAndCurDisplsMultiModule begin");
     306              :     // 先设置本轮的displacements,等于入参displs
     307            0 :     std::copy(displs.begin(), displs.end(), curDispls.begin());
     308              : 
     309              :     // 分配本轮的counts,如果CCLbuffer空间还没完全利用,则再进行分配
     310            0 :     while (allocatableCount > 0) {
     311              :         // 计算现在还有几个rank还有数据需要去通信(countsLeft不为0)
     312              :         const auto nonZeroCount =
     313            0 :             std::count_if(countsLeft.begin(), countsLeft.end(), [](const u64 count) { return count != 0; });
     314            0 :         if (nonZeroCount == 0) {
     315            0 :             finished = true;
     316            0 :             HCCL_INFO("[%s] Calc CurCountsAndCurDispls for multiModule finish", __func__);
     317            0 :             return HCCL_SUCCESS;
     318              :         }
     319              :         // 计算每个rank可以分到多少count
     320            0 :         const auto perRankCount = allocatableCount / nonZeroCount;
     321            0 :         if (perRankCount == 0) {
     322            0 :             break;
     323              :         }
     324            0 :         HCCL_DEBUG("[%s] Calc CurCountsAndCurDispls for perRankCount finish", __func__);
     325            0 :         for (auto i = 0U; i < countsLeft.size(); ++i) {
     326            0 :             const auto curCount = countsLeft[i] < perRankCount ? countsLeft[i] : perRankCount;
     327            0 :             allocatableCount -= curCount;
     328            0 :             curCounts[i] += curCount;
     329            0 :             countsLeft[i] -= curCount;
     330            0 :             displs[i] += curCount;
     331              :         } 
     332              :     }
     333              :     //特殊情况下,allocatableCount 刚好使用完毕时,不仅如此while循环,导致RunLoop额外循环一次
     334              :     const auto nonZeroCount =
     335            0 :         std::count_if(countsLeft.begin(), countsLeft.end(), [](const u64 count) { return count != 0; });
     336            0 :     if (nonZeroCount == 0) {
     337            0 :         finished = true;
     338              :     }
     339            0 :     HCCL_INFO("[%s] Calc CurCountsAndCurDispls for multiModule finish.", __func__);
     340            0 :     return HCCL_SUCCESS;
     341              : }
     342              : 
     343            0 : HcclResult CollAllGatherVMeshExecutor::CalcCurCountsAndCurDisplsSingleModule(const u64 maxTotalCount,
     344              :                                                                         std::vector<u64> &countsLeft, std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls,
     345              :                                                                         bool &finished)
     346              : {
     347            0 :     finished = true;
     348              : 
     349            0 :     curCounts.resize(countsLeft.size(), 0);
     350            0 :     curDispls.resize(displs.size(), 0);
     351              : 
     352              :     // 先设置本轮的displacements,等于入参displs
     353            0 :     std::copy(displs.begin(), displs.end(), curDispls.begin());
     354              : 
     355              :     // 分配好每个rank的counts
     356            0 :     for (auto i = 0U; i < countsLeft.size(); ++i)
     357              :     {
     358            0 :         const auto curCount = countsLeft[i] < maxTotalCount ? countsLeft[i] : maxTotalCount;
     359            0 :         curCounts[i] = curCount;
     360            0 :         countsLeft[i] -= curCount;
     361            0 :         displs[i] += curCount;
     362              : 
     363            0 :         if (countsLeft[i] != 0) {
     364            0 :             finished = false;
     365              :         }
     366              :     }
     367              : 
     368            0 :     return HCCL_SUCCESS;
     369              : }
     370              : 
     371            0 : HcclResult CollAllGatherVMeshExecutor::CalcCurCountsAndCurDispls(const u64 maxTotalCount,
     372              :                                                                         std::vector<u64> &countsLeft, std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls,
     373              :                                                                         bool &finished)
     374              : {
     375            0 :     if (topoAttr_.moduleNum > 1) {
     376            0 :         CHK_RET(CalcCurCountsAndCurDisplsMultiModule(maxTotalCount, countsLeft, displs, curCounts, curDispls, finished));
     377              :     } else {
     378            0 :         CHK_RET(CalcCurCountsAndCurDisplsSingleModule(maxTotalCount, countsLeft, displs, curCounts, curDispls, finished));
     379              :     }
     380            0 :     return HCCL_SUCCESS;
     381              : }
     382              : 
     383              : REGISTER_EXEC("AllGatherVMeshExecutor", AllGatherVMesh, CollAllGatherVMeshExecutor);
     384              : } // namespace hccl
        

Generated by: LCOV version 2.0-1