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 "all_gather_recursive_hd.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 0 : AllGatherRecursiveHalvingDoubling::AllGatherRecursiveHalvingDoubling(const HcclDispatcher dispatcher)
16 0 : : RecursiveHalvingDoublingBase(dispatcher)
17 : {
18 0 : }
19 :
20 0 : AllGatherRecursiveHalvingDoubling::~AllGatherRecursiveHalvingDoubling()
21 : {
22 0 : }
23 :
24 : // 服务器间allreduce的入口函数
25 0 : HcclResult AllGatherRecursiveHalvingDoubling::RunAsync(const u32 rank, const u32 rankSize,
26 : const std::vector<std::shared_ptr<Transport> > &links)
27 : {
28 0 : CHK_SMART_PTR_NULL(dispatcher_);
29 0 : CHK_PTR_NULL(stream_.ptr());
30 0 : HCCL_INFO("AllGatherRecursiveHalvingDoubling run: rank[%u] totalrank[%u] inputMem[%p] outputMem[%p] count[%llu]",
31 : rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
32 :
33 0 : HcclResult ret = HCCL_SUCCESS;
34 :
35 0 : if (rankSize == 1) {
36 0 : if (inputMem_ != outputMem_) {
37 0 : ret = HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
38 : }
39 0 : return ret;
40 : }
41 :
42 0 : if (links.size() < rankSize) {
43 0 : HCCL_ERROR("[AllGatherRecursiveHalvingDoubling][RunAsync]rank[%u] linksize[%llu] is less than rankSize[%u]",
44 : rank, links.size(), rankSize);
45 0 : return HCCL_E_INTERNAL;
46 : }
47 :
48 0 : ret = CalcPartOneSizeAndBlockSize(rankSize);
49 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllGatherRecursiveHalvingDoubling][RunAsync]Calculate Par1Size[%u] "\
50 : "And BlockSize[%u] Failed! rankSize[%u]", part1Size_, blockSize_, rankSize), ret);
51 :
52 0 : ret = CalculateSlices(dataBytes_, rankSize);
53 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
54 : HCCL_ERROR("[AllGatherRecursiveHalvingDoubling][RunAsync]Calculate slices failed, "\
55 : "dataBytes[%llu], rankSize[%u]", dataBytes_, rankSize), ret);
56 :
57 0 : CHK_RET(GatherInPartOneToEven(rank, links));
58 :
59 0 : CHK_RET(AllGatherInBlock(rank, rankSize, links));
60 :
61 0 : CHK_RET(GatherInPartOneToOdd(rank, links));
62 :
63 0 : HCCL_INFO("AllGatherRecursiveHalvingDoubling finished: rank[%u] finished", rank);
64 0 : return HCCL_SUCCESS;
65 : }
66 :
67 :
68 0 : HcclResult AllGatherRecursiveHalvingDoubling::CalculateSlices(u64 dataBytes, const u32 rankSize) const
69 : {
70 0 : slices_.resize(blockSize_);
71 0 : u64 bytesPerSlice = dataBytes;
72 0 : u64 totalBytes = dataBytes * rankSize;
73 0 : u64 bytesLeft = totalBytes;
74 0 : u32 i = 0;
75 0 : while (bytesLeft > 0 && i < part1Size_ / 2) { // 除2计算part1在做完操作后block内slice数
76 0 : slices_[i].size = 2 * bytesPerSlice < bytesLeft ? 2 * bytesPerSlice : bytesLeft; // 乘2表示slice为part2两倍
77 0 : slices_[i].offset = totalBytes - bytesLeft;
78 0 : bytesLeft -= slices_[i].size;
79 0 : i++;
80 : }
81 :
82 0 : while (bytesLeft > 0) {
83 0 : slices_[i].size = bytesPerSlice < bytesLeft ? bytesPerSlice : bytesLeft;
84 0 : slices_[i].offset = totalBytes - bytesLeft;
85 0 : bytesLeft -= slices_[i].size;
86 0 : i++;
87 : }
88 0 : return HCCL_SUCCESS;
89 : }
90 :
91 0 : HcclResult AllGatherRecursiveHalvingDoubling::GatherInPartOneToEven(u32 rank, const std::vector<LINK> &links)
92 : {
93 0 : if (rank < part1Size_ && rank % 2 == 0) { // 模2判断奇偶性，从下一个rank的output收数据到output
94 0 : u32 peerRank = rank + 1; // 加1计算下一个rank号
95 0 : if (peerRank < links.size()) {
96 0 : CHK_SMART_PTR_NULL(links[peerRank]);
97 :
98 0 : HcclResult ret = links[peerRank]->TxAck(stream_);
99 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
100 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] tx ack from peerank[%u] failed",
101 : rank, peerRank), ret);
102 0 : ret = links[peerRank]->RxAck(stream_);
103 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
104 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] rx ack from peerank[%u] failed",
105 : rank, peerRank), ret);
106 0 : DeviceMem gatherOutputMem = outputMem_.range(dataBytes_ * rank, dataBytes_);
107 : // 接收数据到本端的 output
108 0 : HCCL_DEBUG(
109 : "rank[%u] outputMem[%p] receive from PeerRank[%u] outputMem, Offset[%llu], Size[%llu]",
110 : rank, gatherOutputMem.ptr(), peerRank, baseOffset_ + dataBytes_ * rank,
111 : gatherOutputMem.size());
112 :
113 0 : ret = ExecuteRxSync(links[peerRank], UserMemType::OUTPUT_MEM, dataBytes_ * rank, gatherOutputMem.ptr(),
114 0 : dataBytes_, stream_);
115 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] rx sync from PeerRank[%u] "\
116 : "failed", rank, peerRank), ret);
117 0 : ret = links[peerRank]->RxWaitDone(stream_);
118 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Gather][InPartOneToEven]RxWaitDone failed"), ret);
119 0 : }
120 0 : } else if (rank < part1Size_ && rank % 2 == 1) { // 模2判断奇偶性，向上一个rank发送数据
121 0 : u32 peerRank = rank - 1; // 减1计算上一个rank号
122 : // 发送到对端的output
123 0 : if (peerRank < links.size()) {
124 0 : CHK_SMART_PTR_NULL(links[peerRank]);
125 0 : HcclResult ret = links[peerRank]->TxAck(stream_);
126 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
127 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] tx ack from peerank[%u] failed",
128 : rank, peerRank), ret);
129 0 : ret = links[peerRank]->RxAck(stream_);
130 0 : HCCL_DEBUG("[AllGatherRecursiveHalvingDoubling][GatherInPartOneToEven]peerRank is %u", peerRank);
131 : // 等待对端可以接收数据
132 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
133 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] rx ack from peerank[%u] failed",
134 : rank, peerRank), ret);
135 : // 设置gather的发送内存范围
136 0 : DeviceMem gatherOutputMem = outputMem_.range(dataBytes_ * rank, dataBytes_);
137 : // 发送数据到对端的 output
138 0 : HCCL_DEBUG("rank[%u] outputMem[%p] sends to PeerRank[%u] outputMem, Offset[%llu], Size[%llu]",
139 : rank, gatherOutputMem.ptr(), peerRank, baseOffset_ + dataBytes_ * rank,
140 : gatherOutputMem.size());
141 :
142 0 : ret = ExecuteTxSync(links[peerRank], UserMemType::OUTPUT_MEM, dataBytes_ * rank, gatherOutputMem.ptr(),
143 0 : dataBytes_, stream_);
144 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
145 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] tx sync to PeerRank[%u] failed",
146 : rank, peerRank), ret);
147 0 : ret = links[peerRank]->TxWaitDone(stream_);
148 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Gather][InPartOneToEven]TxWaitDone failed"), ret);
149 0 : }
150 : }
151 0 : return HCCL_SUCCESS;
152 : }
153 :
154 0 : HcclResult AllGatherRecursiveHalvingDoubling::GatherInPartOneToOdd(u32 rank, const std::vector<LINK> &links)
155 : {
156 0 : if (rank < part1Size_ && rank % 2 == 0) { // 模2判断奇偶性，向下一个rank发送数据
157 0 : u32 peerRank = rank + 1; // 加1计算下一个rank号
158 : // 发送到对端的output
159 0 : if (peerRank < links.size()) {
160 0 : CHK_SMART_PTR_NULL(links[peerRank]);
161 0 : HcclResult ret = links[peerRank]->TxAck(stream_);
162 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
163 : HCCL_ERROR("[Gather][InPartOneToOdd]rank[%u] tx ack from peerank[%u] failed.",
164 : rank, peerRank), ret);
165 0 : ret = links[peerRank]->RxAck(stream_);
166 : // 等待对端可以接收数据
167 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
168 : HCCL_ERROR("[Gather][InPartOneToOdd]rank[%u] rx ack from peerank[%u] failed", rank, peerRank), ret);
169 :
170 0 : HCCL_DEBUG("rank[%u] outputMem[%p] sends to PeerRank[%u] outputMem, Offset[%llu], Size[%llu]",
171 : rank, outputMem_.ptr(), peerRank, baseOffset_, outputMem_.size());
172 0 : ret = ExecuteTxSync(links[peerRank], UserMemType::OUTPUT_MEM, baseOffset_, outputMem_.ptr(),
173 0 : outputMem_.size(), stream_);
174 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
175 : HCCL_ERROR("[Gather][InPartOneToOdd]rank[%u] tx sync to PeerRank[%u] failed", rank, peerRank), ret);
176 0 : ret = links[peerRank]->TxWaitDone(stream_);
177 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Gather][InPartOneToOdd]TxWaitDone failed"), ret);
178 : }
179 0 : } else if (rank < part1Size_ && rank % 2 == 1) { // 模2判断奇偶性，从上一个rank的output收数据到output
180 0 : u32 peerRank = rank - 1; // 减1计算上一个rank号
181 0 : if (peerRank < links.size()) {
182 0 : CHK_SMART_PTR_NULL(links[peerRank]);
183 : // 知会对端本人可以接收数据
184 0 : HcclResult ret = links[peerRank]->TxAck(stream_);
185 : // 等待对端可以接收数据
186 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
187 : HCCL_ERROR("[Gather][InPartOneToOdd]rank[%u] tx ack from peerank[%u] failed",
188 : rank, peerRank), ret);
189 0 : ret = links[peerRank]->RxAck(stream_);
190 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
191 : HCCL_ERROR("[Gather][InPartOneToOdd]rank[%u] rx ack from peerank[%u] failed",
192 : rank, peerRank), ret);
193 : // 接收数据到本端的 output
194 0 : HCCL_DEBUG("rank[%u] outputMem[%p] receive from PeerRank[%u] outputMem, Offset[%llu], "\
195 : "Size[%llu]", rank, outputMem_.ptr(), peerRank, baseOffset_, outputMem_.size());
196 0 : ret = ExecuteRxSync(links[peerRank], UserMemType::OUTPUT_MEM, baseOffset_, outputMem_.ptr(),
197 0 : outputMem_.size(), stream_);
198 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
199 : HCCL_ERROR("[Gather][InPartOneToOdd]rank[%u] rx sync from PeerRank[%u] failed", rank, peerRank), ret);
200 0 : ret = links[peerRank]->RxWaitDone(stream_);
201 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Gather][InPartOneToOdd]RxWaitDone failed"), ret);
202 : }
203 : }
204 0 : return HCCL_SUCCESS;
205 : }
206 :
207 0 : HcclResult AllGatherRecursiveHalvingDoubling::AllGatherInBlock(u32 rank, u32 rankSize, const std::vector<LINK> &links)
208 : {
209 0 : u32 rankInBlock = 0;
210 0 : if (rank < part1Size_ && (rank % 2) == 1) { // 模2余1代表当前rank在part1的奇数位置上，不参与block内的计算
211 0 : return HCCL_SUCCESS;
212 0 : } else if (rank < part1Size_ && (rank % 2) == 0) { // 模2余0代表当前rank在part1的偶数位置上，参与block内的计算
213 0 : rankInBlock = rank / 2; // 除2计算出在block内的rank号
214 : } else {
215 0 : rankInBlock = rank - part1Size_ / 2; // rank在part2中，用原始rank减part1除2，计算出在block内的rank号
216 : }
217 :
218 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
219 0 : TemplateType::TEMPLATE_ALL_GATHER_HALVING_DOUBLING, dispatcher_);
220 0 : CHK_SMART_PTR_NULL(tempAlg);
221 0 : CHK_RET(tempAlg->Prepare(blockSize_, UserMemType::OUTPUT_MEM, UserMemType::OUTPUT_MEM));
222 0 : CHK_RET(tempAlg->Prepare(outputMem_, outputMem_, count_, dataType_, stream_,
223 : reductionOp_, root_, slices_, baseOffset_));
224 :
225 0 : CHK_RET(tempAlg->RegisterProfiler(
226 : profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
227 :
228 0 : std::vector<LINK> subLinks;
229 0 : CHK_RET(BuildSubLinks(links, subLinks, rankSize));
230 :
231 0 : CHK_PRT_RET(subLinks.size() == 0,
232 : HCCL_ERROR("[AllGatherRecursiveHalvingDoubling][AllGatherInBlock]rank[%u] BuildSubLinks failed",
233 : rank), HCCL_E_PARA);
234 :
235 0 : CHK_RET(tempAlg->RunAsync(rankInBlock, blockSize_, subLinks));
236 :
237 0 : return HCCL_SUCCESS;
238 0 : }
239 :
240 0 : HcclResult AllGatherRecursiveHalvingDoubling::GetNslbAdjInfo(const u32 rank, const u32 rankSize,
241 : const std::vector<LINK> &links,
242 : AdjInfo& nslbAdjInfo)
243 : {
244 0 : u32 nslbRound = 0;
245 0 : u32 base = 1;
246 0 : const u32 minExponent = 1;
247 0 : HCCL_DEBUG("[AllGatherRecursiveHalvingDoubling]GetNslbAdjInfo begins");
248 0 : while ((base << nslbRound) <= rankSize) {
249 0 : nslbRound++;
250 : }
251 0 : if (nslbRound >= minExponent) {
252 0 : nslbRound = nslbRound - minExponent;
253 : }
254 0 : u32 nslbBlockSize = base << nslbRound;
255 : // 获取第一部分：rank数减block数乘2
256 0 : u32 nslbPart1Size = (rankSize - nslbBlockSize) * NSLBDP_ALL_GATHER_MOLD2;
257 : // 2的次幂场景下处理流程
258 0 : if (nslbPart1Size == 0) {
259 0 : u32 stepNum = 0;
260 0 : while ((rankSize >> (stepNum + 1)) != 0) {
261 0 : stepNum++;
262 : }
263 0 : for (u32 step = 0; step < stepNum; step++) {
264 0 : HCCL_DEBUG("[AllGatherRecursiveHalvingDoubling]current step is %u", step);
265 0 : u32 peerRankBitmask = 1 << (stepNum - step - 1);
266 0 : u32 peerRank = rank ^ peerRankBitmask;
267 0 : NslbDpAdjInfo adjInfoStep = {0};
268 0 : u32 remoteuserRank = links[peerRank]->GetRemoteRank();
269 0 : adjInfoStep.dstLocalRankId = remoteuserRank;
270 0 : adjInfoStep.phaseId = step + 1;
271 0 : adjInfoStep.rev = 0;
272 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
273 0 : HCCL_DEBUG("[AllGatherRecursiveHalvingDoubling]current step %u success", step);
274 : }
275 0 : nslbAdjInfo.dstRankNum = stepNum;
276 0 : return HCCL_SUCCESS;
277 : }
278 : // 非2的次幂场景下，被合并部分的奇数rank处理流程
279 0 : if (rank < nslbPart1Size && rank % NSLBDP_ALL_GATHER_MOLD2 == 1) {
280 0 : u32 peerRank = rank - 1;
281 0 : if (peerRank < links.size()) {
282 0 : NslbDpAdjInfo adjInfoStep = {0};
283 0 : adjInfoStep.dstLocalRankId = links[peerRank]->GetRemoteRank();
284 0 : adjInfoStep.phaseId = 1;
285 0 : adjInfoStep.rev = 0;
286 0 : HCCL_INFO("AllGatherHDR-nslb: peerRank[%u]", peerRank);
287 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
288 0 : nslbAdjInfo.dstRankNum = 1;
289 : }
290 0 : return HCCL_SUCCESS;
291 : }
292 : // 针对合并后映射成2的次幂场景处理
293 0 : u32 rankInBlock = 0;
294 0 : if (rank < nslbPart1Size && (rank % NSLBDP_ALL_GATHER_MOLD2) == 0) {
295 0 : rankInBlock = rank / NSLBDP_ALL_GATHER_MOLD2; // 直接除以2即为本rank的在block内的排序
296 : } else {
297 0 : rankInBlock = rank - nslbPart1Size / NSLBDP_ALL_GATHER_MOLD2; // 通过rank减去part1除2的大小即不处于第一部分的block内rank号
298 : }
299 0 : std::vector<LINK> subLinks;
300 0 : std::vector<LINK>::const_iterator iter = links.begin();
301 0 : subLinks.resize(nslbBlockSize);
302 0 : for (u32 i = 0; i < rankSize; i++) {
303 0 : if (i < nslbPart1Size && (i % NSLBDP_ALL_GATHER_MOLD2) == 1) { // 模2余1代表当前rank在part1的奇数位置上，不参与block内的建链
304 0 : continue;
305 0 : } else if (i < nslbPart1Size && (i % NSLBDP_ALL_GATHER_MOLD2) == 0) { // 模2余0代表当前rank在part1的偶数位置上
306 0 : std::vector<LINK>::const_iterator niter = std::next(iter, i);
307 0 : if (niter != links.end()) {
308 0 : subLinks[i / NSLBDP_ALL_GATHER_MOLD2] = *niter;
309 : }
310 0 : } else {
311 0 : std::vector<LINK>::const_iterator niter = std::next(iter, i);
312 0 : if (niter != links.end()) {
313 0 : subLinks[i - nslbPart1Size / NSLBDP_ALL_GATHER_MOLD2] = *niter;
314 : }
315 : }
316 : }
317 0 : u32 stepNum = 0;
318 0 : while ((rankSize >> (stepNum + 1)) != 0) {
319 0 : stepNum++;
320 : }
321 : // 映射完成后针对以新的通信域进行邻接表获取
322 0 : for (u32 step = 0; step < stepNum; step++) {
323 0 : u32 peerRankBitmask = (1 << step);
324 0 : u32 peerRank = rankInBlock ^ peerRankBitmask;
325 0 : if (subLinks[peerRank] == nullptr) {
326 0 : continue;
327 : }
328 0 : NslbDpAdjInfo adjInfoStep = {0};
329 0 : u32 remoteuserRank = subLinks[peerRank]->GetRemoteRank();
330 0 : HCCL_DEBUG("[AllGatherRecursiveHalvingDoubling][GetNslbAdjInfo]remoteuserRank is %u", remoteuserRank);
331 0 : adjInfoStep.dstLocalRankId = remoteuserRank;
332 0 : adjInfoStep.phaseId = step + 1;
333 0 : adjInfoStep.rev = 0;
334 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
335 : }
336 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.nsAdjInfo.size() ;
337 :
338 0 : if(nslbAdjInfo.nsAdjInfo.size() == 0) {
339 0 : return HCCL_SUCCESS;
340 : }
341 : // 上面处理完成后，紧接着处理合并部分的偶数rank同步到奇数rank增加phaseId
342 0 : if (rank < nslbPart1Size && rank % NSLBDP_ALL_GATHER_MOLD2 == 0) {
343 0 : u32 peerRank = rank + 1;
344 0 : uint16_t phaseSize = nslbAdjInfo.nsAdjInfo.size();
345 0 : if (peerRank < links.size()) {
346 0 : NslbDpAdjInfo adjInfoStep = {0};
347 0 : adjInfoStep.dstLocalRankId = links[peerRank]->GetRemoteRank();
348 0 : adjInfoStep.phaseId = nslbAdjInfo.nsAdjInfo[phaseSize - 1].phaseId + 1;
349 0 : adjInfoStep.rev = 0;
350 0 : HCCL_INFO("Scatter-nslb: peerRank[%u]", peerRank);
351 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
352 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.nsAdjInfo.size();
353 : }
354 0 : return HCCL_SUCCESS;
355 : }
356 0 : return HCCL_SUCCESS;
357 0 : }
358 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_GATHER_RECURSIVE_HALVING_DOUBLING, AllGatherRecursiveHalvingDoubling);
359 : } // namespace hccl
|
