LCOV - code coverage report
Current view: top level - legacy/ascend910/algorithm/base/alg_template/temp_broadcast - broadcast_nb_binary.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 0.0 % 105 0
Test Date: 2026-08-18 17:47:01 Functions: 0.0 % 8 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 "broadcast_nb_binary.h"
      12              : #include "device_capacity.h"
      13              : #include "alg_template_register.h"
      14              : 
      15              : namespace hccl {
      16              : constexpr float LATENCY = 60; // 静态时延 60 us;
      17              : 
      18            0 : BroadcastNBBinary::BroadcastNBBinary(const HcclDispatcher dispatcher) : NBBase(dispatcher) {}
      19              : 
      20            0 : BroadcastNBBinary::~BroadcastNBBinary() {}
      21              : 
      22              : HcclResult
      23            0 : BroadcastNBBinary::RunAsync(const u32 rank, const u32 rankSize, const std::vector<std::shared_ptr<Transport>>& links)
      24              : {
      25            0 :     CHK_SMART_PTR_NULL(dispatcher_);
      26            0 :     CHK_PTR_NULL(stream_.ptr());
      27            0 :     HCCL_INFO("BroadcastNBBinary run: rank[%u] totalrank[%u] count[%llu]", rank, rankSize, count_);
      28              : 
      29            0 :     if (rankSize == 1) {
      30            0 :         return HCCL_SUCCESS;
      31              :     }
      32              : 
      33            0 :     CHK_PRT_RET(
      34              :         links.size() < rankSize,
      35              :         HCCL_ERROR("[BroadcastNBBinary][RunAsync]rank[%u] linksize[%llu] is less than rank size", rank, links.size()),
      36              :         HCCL_E_INTERNAL);
      37              : 
      38            0 :     u32 unitSize = DataUnitSize(dataType_);
      39            0 :     CHK_PRT_RET(
      40              :         unitSize == 0, HCCL_ERROR("[BroadcastNBBinary][RunAsync]rank[%u] unit data size is zero", rank),
      41              :         HCCL_E_INTERNAL);
      42              : 
      43            0 :     HcclResult ret = HCCL_SUCCESS;
      44            0 :     if (inputMem_ != outputMem_) {
      45            0 :         ret = HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
      46            0 :         CHK_PRT_RET(
      47              :             ret != HCCL_SUCCESS,
      48              :             HCCL_ERROR(
      49              :                 "[Run][BroadcastOnRootRank]root rank[%u] memcpy async from input[%p] "
      50              :                 "failed to output[%p]",
      51              :                 rank, inputMem_.ptr(), outputMem_.ptr()),
      52              :             ret);
      53              :     }
      54              : 
      55            0 :     CHK_RET(RunBroadcastNBBinary(rank, rankSize, links));
      56            0 :     HCCL_INFO("BroadcastNBBinary finished: rank[%u] end count[%llu]", rank, count_);
      57            0 :     return HCCL_SUCCESS;
      58              : }
      59              : 
      60            0 : HcclResult BroadcastNBBinary::RunBroadcastNBBinary(
      61              :     const u32 rank, const u32 rankSize, const std::vector<std::shared_ptr<Transport>>& links)
      62              : {
      63            0 :     HcclResult ret = HCCL_SUCCESS;
      64            0 :     if (rank == root_) {
      65            0 :         hasData_ = true;
      66            0 :         CHK_PRT_RET(!inputMem_, HCCL_ERROR("[BroadcastNBBinary][RunAsync]rank[%u] inputmem is null", rank), HCCL_E_PTR);
      67              :     } else {
      68            0 :         CHK_PRT_RET(
      69              :             !outputMem_, HCCL_ERROR("[BroadcastNBBinary][RunAsync]rank[%u] outputmem is null", rank), HCCL_E_PTR);
      70              :     }
      71              : 
      72            0 :     HCCL_DEBUG("root[%u], hasData[%u]", root_, hasData_);
      73              : 
      74            0 :     u64 dataBytes = count_ * DataUnitSize(dataType_);     // 总数据量
      75            0 :     u32 nSteps = CalcCeilLog2(rankSize);                  // 通信步数
      76            0 :     u32 deltaRoot = (rank + rankSize - root_) % rankSize; // 与Root节点的距离
      77              : 
      78            0 :     for (u32 step = 0; step < nSteps; step++) {
      79            0 :         if (deltaRoot < u32(1 << step)) { // 该节点需要发送数据
      80            0 :             if ((step != nSteps - 1 || deltaRoot < (rankSize - (1 << step))) && hasData_) {
      81            0 :                 u32 deltaRank = 1 << step;
      82            0 :                 u32 sendTo = (rank + deltaRank) % rankSize;
      83            0 :                 LINK linkRight = links[sendTo];
      84            0 :                 CHK_SMART_PTR_NULL(linkRight);
      85              : 
      86            0 :                 std::vector<Slice> txSlices;
      87            0 :                 txSlices.resize(1);
      88            0 :                 txSlices[0].offset = baseOffset_;
      89            0 :                 txSlices[0].size = dataBytes;
      90              : 
      91            0 :                 CHK_RET(linkRight->RxAck(stream_));
      92            0 :                 ret = Tx(linkRight, txSlices);
      93            0 :                 CHK_PRT_RET(
      94              :                     ret != HCCL_SUCCESS,
      95              :                     HCCL_ERROR("[Run][Broadcast]rank[%u] step[%u] Right Link tx slices Failed", rank, step), ret);
      96              : 
      97            0 :                 ret = linkRight->TxWaitDone(stream_);
      98            0 :                 CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Run][Broadcast]TxWaitDone failed"), ret);
      99            0 :                 ret = linkRight->GetLinkType() == LinkType::LINK_HCCS ?
     100            0 :                           linkRight->WaitFin(stream_) :
     101            0 :                           linkRight->WaitFinAck(stream_); // P2P和Roce场景都需要同步
     102            0 :                 CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Run][Broadcast]WaitFinAck failed"), ret);
     103            0 :             }
     104            0 :         } else if (deltaRoot < u32(1 << (step + 1)) && rank != root_) { // 该节点需要接收数据
     105            0 :             u32 deltaRank = 1 << step;
     106            0 :             u32 recvFrom = (rank + rankSize - deltaRank) % rankSize;
     107            0 :             LINK linkLeft = links[recvFrom];
     108            0 :             CHK_SMART_PTR_NULL(linkLeft);
     109              : 
     110            0 :             std::vector<Slice> rxSlices;
     111            0 :             rxSlices.resize(1);
     112            0 :             rxSlices[0].offset = baseOffset_;
     113            0 :             rxSlices[0].size = dataBytes;
     114              : 
     115            0 :             CHK_RET(linkLeft->TxAck(stream_));
     116            0 :             ret = Rx(linkLeft, rxSlices);
     117            0 :             CHK_PRT_RET(
     118              :                 ret != HCCL_SUCCESS,
     119              :                 HCCL_ERROR("[Run][Broadcast]rank[%u] step[%u] Right Link rx slices Failed", rank, step), ret);
     120              : 
     121            0 :             ret = linkLeft->RxWaitDone(stream_);
     122            0 :             CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Run][Broadcast]RxWaitDone failed"), ret);
     123            0 :             ret = linkLeft->GetLinkType() == LinkType::LINK_HCCS ?
     124            0 :                       linkLeft->PostFin(stream_) :
     125            0 :                       linkLeft->PostFinAck(stream_); // P2P和Roce场景都需要同步
     126            0 :             CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Run][Broadcast]PostFinAck failed"), ret);
     127            0 :             hasData_ = true;
     128            0 :         }
     129              :     }
     130            0 :     return HCCL_SUCCESS;
     131              : }
     132              : 
     133            0 : HcclResult BroadcastNBBinary::Tx(const LINK& link, const std::vector<Slice>& txSlices)
     134              : {
     135            0 :     std::vector<TxMemoryInfo> txMems;
     136            0 :     for (const Slice& txSlice : txSlices) {
     137            0 :         DeviceMem srcMem = outputMem_.range(txSlice.offset, txSlice.size);
     138            0 :         HCCL_DEBUG("tx srcMem[%p] range[%llu] size[%llu] ", srcMem.ptr(), txSlice.offset, txSlice.size);
     139            0 :         txMems.emplace_back(TxMemoryInfo{UserMemType::OUTPUT_MEM, txSlice.offset, srcMem.ptr(), txSlice.size});
     140            0 :     }
     141              : 
     142            0 :     CHK_RET(link->TxAsync(txMems, stream_));
     143            0 :     return HCCL_SUCCESS;
     144            0 : }
     145              : 
     146            0 : HcclResult BroadcastNBBinary::Rx(const LINK& link, const std::vector<Slice>& rxSlices)
     147              : {
     148            0 :     std::vector<RxMemoryInfo> rxMems;
     149            0 :     for (const Slice& rxSlice : rxSlices) {
     150            0 :         DeviceMem dstMem = outputMem_.range(rxSlice.offset, rxSlice.size);
     151            0 :         HCCL_DEBUG("rx dstMem[%p] range[%llu], size[%llu] ", dstMem.ptr(), rxSlice.offset, rxSlice.size);
     152            0 :         rxMems.emplace_back(RxMemoryInfo{UserMemType::OUTPUT_MEM, rxSlice.offset, dstMem.ptr(), rxSlice.size});
     153            0 :     }
     154              : 
     155            0 :     CHK_RET(link->RxAsync(rxMems, stream_));
     156            0 :     return HCCL_SUCCESS;
     157            0 : }
     158              : 
     159            0 : bool ShouldUseBinaryBroadcastOfNB(
     160              :     const u64 dataSize, const u32 rankSize, const u32 userRankSize, const float deviceNumPerAggregation)
     161              : {
     162              :     // 小数据量和rank size为2时使用Binary broadcast
     163            0 :     HCCL_INFO(
     164              :         "datasize[%llu], ranksize[%u], userRankSize[%u], deviceNumPerAggregation[%f]", dataSize, rankSize, userRankSize,
     165              :         deviceNumPerAggregation);
     166              : 
     167            0 :     constexpr u32 TWO_RANK_SIZE = 2;
     168            0 :     if (rankSize == TWO_RANK_SIZE) {
     169            0 :         return true;
     170              :     }
     171              : 
     172              :     // 通信步数为log_2(rankSize)向上取整
     173            0 :     u32 nSteps = 0;
     174            0 :     for (u32 tmp = rankSize - 1; tmp != 0; tmp >>= 1, ++nSteps) {
     175              :     }
     176              : 
     177              :     float bandWidth; // 网卡出口带宽,用于计算大小包切片策略的阈值
     178            0 :     CHK_RET(GetBandWidthPerNPU(1, userRankSize, static_cast<u32>(deviceNumPerAggregation), bandWidth)); // 单位:GB/s
     179              : 
     180              :     // (公式解释):
     181              :     // bandwidth_ - 网卡出口带宽,单位GB/s
     182              :     // LATENCY - 链路端到端静态时延
     183              :     // * 1000 - 转换为ns
     184            0 :     const float dataSizeBaseNum = bandWidth * LATENCY * 1000;
     185            0 :     constexpr u32 rankSizeOfSmallScale = 4; // 4以下节点数为小规模
     186            0 :     constexpr u32 dataSizeMultiple = 2;     // 通信数据量倍数为2
     187              : 
     188              :     // 计算阈值,用于判断大小包
     189              :     float thresholdDataSize;
     190            0 :     if (rankSize <= rankSizeOfSmallScale) {
     191            0 :         thresholdDataSize = rankSize * dataSizeBaseNum;
     192              :     } else {
     193              :         // 公式:用于计算大包处理的阈值
     194            0 :         thresholdDataSize = (nSteps + nSteps - dataSizeMultiple) / (nSteps - dataSizeMultiple) * dataSizeBaseNum;
     195              :     }
     196              : 
     197            0 :     return dataSize < thresholdDataSize; // 小数据量使用Binary broadcast
     198              : }
     199              : 
     200              : REGISTER_TEMPLATE(TemplateType::TEMPLATE_BROADCAST_NB_BINARY, BroadcastNBBinary);
     201              : } // namespace hccl
        

Generated by: LCOV version 2.0-1