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_scatter_ring_executor.h"
      12              : 
      13              : namespace hccl {
      14            0 : CollScatterRingExecutor::CollScatterRingExecutor(const HcclDispatcher dispatcher,
      15            0 :                                 std::unique_ptr<TopoMatcher> &topoMatcher)
      16            0 :     : CollScatterExecutor(dispatcher, topoMatcher)
      17              : {
      18            0 : }
      19              : 
      20            0 : HcclResult CollScatterRingExecutor::CalcLevel0CommInfo(TransportMemType inputType,
      21              :     TransportMemType outputType,
      22              :     std::vector<LevelNSubCommTransport>& opTransport)
      23              : {
      24            0 :     HCCL_INFO("[CollScatterRingExecutor][CalcLevel0CommInfo]tag[%s] start", tag_.c_str());
      25            0 :     CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
      26              : 
      27            0 :     CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
      28            0 :     HCCL_INFO("[CollScatterRingExecutor][CalcLevel0CommInfo]tag[%s] Calc RingComm finish", tag_.c_str());
      29            0 :     return HCCL_SUCCESS;
      30            0 : }
      31              : 
      32            0 : HcclResult CollScatterRingExecutor::CalcStreamNum(u32& streamNum)
      33              : {
      34            0 :     u32 totalStreamNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
      35            0 :     if (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_8P_RING) {
      36            0 :         totalStreamNum = LEVEL0_PLANE_NUM_IN_8PRING;
      37              :     }
      38            0 :     streamNum = totalStreamNum - 1;
      39            0 :     HCCL_INFO("[CollScatterRingExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
      40            0 :     return HCCL_SUCCESS;
      41              : }
      42              : 
      43            0 : HcclResult CollScatterRingExecutor::KernelRun(const OpParam &param, ExecMem &execMem)
      44              : {
      45            0 :     HCCL_CONFIG_INFO(HCCL_ALG, "[CollScatterRingExecutor] scatter starts.");
      46            0 :     Stream& stream = const_cast<Stream&>(param.stream);
      47              : 
      48            0 :     u32 perDataSize = SIZE_TABLE[param.DataDes.dataType];
      49              : 
      50            0 :     SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
      51            0 :     u32 level0LocalRank = level0CommInfo.localRank;
      52            0 :     u32 level0LocalRankSize = level0CommInfo.localRankSize;
      53              : 
      54            0 :     u32 commIndex = (topoType_ == TopoType::TOPO_TYPE_8P_RING) ? topoAttr_.devicePhyId : level0LocalRank;
      55              : 
      56            0 :     CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
      57            0 :     SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
      58            0 :     u32 level1LocalRank = level1CommInfo.localRank;
      59            0 :     u32 level1LocalRankSize = level1CommInfo.localRankSize;
      60              : 
      61            0 :     bool bRet = level0LocalRankSize == 0;
      62            0 :     CHK_PRT_RET(bRet, HCCL_ERROR("[CollScatterRingExecutor][KernelRun]tag[%s],comm level0 is empty", tag_.c_str()),
      63              :         HCCL_E_INTERNAL);
      64              : 
      65              :     /* ***********第一步: 节点间scatter ****************************/
      66            0 :     u32 subRoot = INVALID_VALUE_RANKID;
      67            0 :     CHK_RET(topoMatcher_->GetSubRootForScatter(param.root, subRoot));
      68            0 :     CHK_PRT_RET(subRoot == INVALID_VALUE_RANKID,
      69              :         HCCL_ERROR("[CollScatterRingExecutor][KernelRun]GetSubRootForScatter failed, "\
      70              :         "userRank[%u], root[%u], subRoot[%u]", topoAttr_.userRank, param.root, subRoot), HCCL_E_INTERNAL);
      71            0 :     HCCL_DEBUG("[CollScatterRingExecutor][KernelRun]GetSubRootForScatter, userRank[%u], root[%u], subRoot[%u]",
      72              :         topoAttr_.userRank, param.root, subRoot);
      73            0 :     CHK_RET(KernelRunLevel1(execMem.inputMem, execMem.count, param.DataDes.dataType, commIndex,
      74              :         param.root, subRoot, COMM_LEVEL1, stream));
      75              : 
      76              :     /* ***********第二步: 节点内scatter*****************************/
      77            0 :     u32 sliceNum = level0LocalRankSize;
      78            0 :     std::vector<Slice> dataSegsSlice;
      79            0 :     u32 outputOffset = level0LocalRank;
      80            0 :     CHK_RET(PrepareScatterRingSliceData(execMem.count, perDataSize, sliceNum, dataSegsSlice, outputOffset));
      81              : 
      82              :     // 每个server分配的slice大小
      83            0 :     u64 serverSliceSize = execMem.inputMem.size() / level1LocalRankSize;
      84              :     // 每个服务器对应的偏移
      85            0 :     u64 serverSliceOffset = serverSliceSize * level1LocalRank;
      86            0 :     DeviceMem scatterRingInput = execMem.inputMem.range(serverSliceOffset, serverSliceSize);
      87            0 :     CHK_SMART_PTR_NULL(scatterRingInput);
      88            0 :     DeviceMem scatterRingOutput = execMem.inputMem.range(serverSliceOffset, serverSliceSize);
      89            0 :     CHK_SMART_PTR_NULL(scatterRingOutput);
      90              : 
      91            0 :     std::unique_ptr<AlgTemplateBase> level0TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
      92            0 :         TemplateType::TEMPLATE_SCATTER_RING, dispatcher_);
      93            0 :     CHK_SMART_PTR_NULL(level0TempAlg);
      94            0 :     HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_SCATTER_RING in COMM_LEVEL0", __func__);
      95              : 
      96              :     // 偏移需要带入prepare
      97            0 :     u32 rootRankLevel0 = 0;
      98            0 :     CHK_RET(GetRankByUserRank(COMM_LEVEL0, COMM_INDEX_0, subRoot, rootRankLevel0));
      99            0 :     CHK_PRT_RET(rootRankLevel0 == INVALID_VALUE_RANKID,
     100              :         HCCL_ERROR("[CollScatterRingExecutor][KernelRun]rootRankLevel0[%u] is invalid, userRank[%u], subRoot[%u]",
     101              :             rootRankLevel0, topoAttr_.userRank, subRoot),
     102              :         HCCL_E_INTERNAL);
     103              : 
     104            0 :     CHK_RET(level0TempAlg->Prepare(scatterRingInput, scatterRingOutput, execMem.inputMem, execMem.count,
     105              :         param.DataDes.dataType, stream, HCCL_REDUCE_RESERVED, rootRankLevel0, dataSegsSlice, serverSliceOffset));
     106              : 
     107            0 :     HcclResult ret = RunTemplate(level0TempAlg, level0CommInfo);
     108            0 :     CHK_PRT_RET(ret != HCCL_SUCCESS,
     109              :         HCCL_ERROR("[CollScatterRingExecutor][KernelRun]scatter(ring) RunTemplate failed,return[%d]", ret),
     110              :         ret);
     111              : 
     112              :     // 将scratchMem赋值给outputMem
     113            0 :     u8 *scatterRingOutputPtr = static_cast<u8 *>(scatterRingOutput.ptr());
     114            0 :     DeviceMem resultMem(scatterRingOutputPtr + execMem.outputMem.size() * outputOffset, execMem.outputMem.size());
     115            0 :     CHK_RET(HcclD2DMemcpyAsync(dispatcher_, execMem.outputMem, resultMem, stream));
     116            0 :     return HCCL_SUCCESS;
     117            0 : }
     118              : 
     119            0 : HcclResult CollScatterRingExecutor::PrepareScatterRingSliceData(u64 dataCount, u32 unitSize, u32 sliceNum,
     120              :     std::vector<Slice> &dataSlice, u32 &outputOffset)
     121              : {
     122            0 :     CHK_PRT_RET((sliceNum == 0),
     123              :         HCCL_ERROR("[CollScatterRingExecutor][PrepareScatterRingSliceData]sliceNum is zero."), HCCL_E_PARA);
     124              : 
     125              :     // 根据数据量算每个环上数据的偏移和大小
     126            0 :     CHK_RET(PrepareDataSlice(dataCount, unitSize, sliceNum, dataSlice));
     127              : 
     128            0 :     if (topoType_ == TopoType::TOPO_TYPE_8P_RING) {
     129            0 :         std::vector<u32> tmpRing0 = { 0, 1, 2, 6, 5, 4, 7, 3 };
     130            0 :         outputOffset = tmpRing0[outputOffset];
     131            0 :         CHK_RET(ReorderSlice(dataSlice, tmpRing0));
     132            0 :     }
     133            0 :     return HCCL_SUCCESS;
     134              : }
     135              : 
     136              : REGISTER_EXEC("ScatterRingExecutor", ScatterRing, CollScatterRingExecutor);
     137              : 
     138              : }