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