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 "alg_template_register.h"
12 : #include "all_reduce_recursive_hd.h"
13 :
14 : namespace hccl {
15 1 : AllReduceRecursiveHalvingDoubling::AllReduceRecursiveHalvingDoubling(const HcclDispatcher dispatcher)
16 1 : : RecursiveHalvingDoublingBase(dispatcher)
17 1 : {}
18 :
19 2 : AllReduceRecursiveHalvingDoubling::~AllReduceRecursiveHalvingDoubling() {}
20 :
21 1 : HcclResult AllReduceRecursiveHalvingDoubling::Prepare(u64 reduceAttrBitMap, [[maybe_unused]] HcomCollOpInfo* opInfo)
22 : {
23 1 : reduceAttr = reduceAttrBitMap;
24 1 : return HCCL_SUCCESS;
25 : }
26 :
27 : // 算法的主入口
28 : HcclResult
29 0 : AllReduceRecursiveHalvingDoubling::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
30 : {
31 0 : CHK_RET(PrepareRunAsync(rank, rankSize, links));
32 0 : CHK_PRT_RET(
33 : rankSize == 1,
34 : HCCL_INFO(
35 : "[AllReduceRecursiveHalvingDoubling][RunAsync]"
36 : "rankSize[%u], do nothing.",
37 : rankSize),
38 : HCCL_SUCCESS);
39 :
40 0 : CHK_RET(ReduceInPartOne(rank, links));
41 :
42 0 : CHK_RET(ReduceScatterInBlock(rank, rankSize, links));
43 :
44 0 : CHK_RET(AllGatherInBlock(rank, rankSize, links));
45 :
46 0 : CHK_RET(GatherInPartOne(rank, links));
47 :
48 0 : HCCL_INFO("AllReduceRecursiveHalvingDoubling finished: rank[%u] finished", rank);
49 0 : return HCCL_SUCCESS;
50 : }
51 :
52 0 : HcclResult AllReduceRecursiveHalvingDoubling::RunAsyncStaged(
53 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links, RunStage stage)
54 : {
55 0 : CHK_PRT_RET(
56 : rankSize == 1 && stage != RunStage::RUN_PREPARE,
57 : HCCL_INFO(
58 : "[AllReduceRecursiveHalvingDoubling][RunAsyncStaged] rankSize[%u], stage[%d], do nothing.", rankSize,
59 : stage),
60 : HCCL_SUCCESS);
61 0 : switch (stage) {
62 0 : case RunStage::RUN_PREPARE:
63 0 : CHK_RET(PrepareRunAsync(rank, rankSize, links));
64 0 : break;
65 0 : case RunStage::RUN_REDUCE_SCATTER:
66 : // 先执行reducescater
67 0 : CHK_RET(ReduceInPartOne(rank, links));
68 0 : CHK_RET(ReduceScatterInBlock(rank, rankSize, links));
69 0 : break;
70 0 : case RunStage::RUN_ALLGATHER:
71 : // 再执行allgather
72 0 : CHK_RET(AllGatherInBlock(rank, rankSize, links));
73 0 : CHK_RET(GatherInPartOne(rank, links));
74 0 : break;
75 0 : default:
76 0 : HCCL_ERROR("[AllReduceRecursiveHalvingDoubling][RunAsyncStaged]stage[%d]is not support", stage);
77 0 : return HCCL_E_NOT_SUPPORT;
78 : }
79 0 : HCCL_INFO(
80 : "AllReduceRecursiveHalvingDoubling RunAsyncStaged stage[%d] finished: rank[%u] ranksize[%u]", stage, rank,
81 : rankSize);
82 0 : return HCCL_SUCCESS;
83 : }
84 :
85 : HcclResult
86 0 : AllReduceRecursiveHalvingDoubling::PrepareRunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
87 : {
88 0 : CHK_SMART_PTR_NULL(dispatcher_);
89 0 : CHK_PTR_NULL(stream_.ptr());
90 0 : if (!outputMem_ || !inputMem_) {
91 0 : HCCL_ERROR(
92 : "[AllReduceRecursiveHalvingDoubling][RunAsync]rank[%u] run_async inputmem or outputmem is null", rank);
93 0 : return HCCL_E_PTR;
94 : }
95 0 : HCCL_INFO(
96 : "AllReduceRecursiveHalvingDoubling run: rank[%u] totalrank[%u] inputMem[%p] outputMem[%p] count[%llu]", rank,
97 : rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
98 :
99 0 : HcclResult ret = HCCL_SUCCESS;
100 :
101 0 : if (rankSize == 1) {
102 0 : if (inputMem_ != outputMem_) {
103 0 : ret = HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
104 : }
105 0 : return ret;
106 : }
107 :
108 : // 创建reducer & sender
109 0 : senderInfo_.reset(new (std::nothrow) Sender(dataType_, reductionOp_, reduceAttr));
110 0 : CHK_SMART_PTR_NULL(senderInfo_);
111 :
112 0 : reducerInfo_.reset(new (std::nothrow) Reducer(dataType_, reductionOp_, reduceAttr));
113 0 : CHK_SMART_PTR_NULL(reducerInfo_);
114 :
115 0 : bool bRetSize = (links.size() < rankSize);
116 0 : CHK_PRT_RET(
117 : bRetSize,
118 : HCCL_ERROR(
119 : "[AllReduceRecursiveHalvingDoubling][RunAsync]rank[%u] linksize[%llu] is "
120 : "error",
121 : rank, links.size()),
122 : HCCL_E_INTERNAL);
123 :
124 0 : CHK_RET(CalcPartOneSizeAndBlockSize(rankSize));
125 :
126 0 : u32 bytesPerData = SIZE_TABLE[dataType_];
127 0 : u64 dataBytes = count_ * bytesPerData;
128 0 : CHK_RET(CalculateSlices(dataBytes));
129 0 : HCCL_INFO("AllReduceRecursiveHalvingDoubling PrepareRunAsync finished: rank[%u] finished", rank);
130 0 : return HCCL_SUCCESS;
131 : }
132 :
133 0 : HcclResult AllReduceRecursiveHalvingDoubling::ReduceInPartOne(u32 rank, const std::vector<LINK>& links)
134 : {
135 : // 本rank属于第一部分，并且是2的整数倍
136 0 : if (rank < part1Size_ && rank % 2 == 0) { // 1.从下一个rank接收数据到output，2. reduce到本rank的input
137 0 : u32 peerRank = rank + 1;
138 0 : HCCL_DEBUG(
139 : "rank[%u] outputMem receives from PeerRank[%u] inputMem, Offset[%llu], Size[%llu]", rank, peerRank,
140 : baseOffset_, outputMem_.size());
141 :
142 0 : if (peerRank < links.size()) {
143 0 : const LINK& link = links[peerRank];
144 0 : CHK_SMART_PTR_NULL(link);
145 :
146 0 : HcclResult ret = link->TxAck(stream_);
147 0 : CHK_PRT_RET(
148 : ret != HCCL_SUCCESS,
149 : HCCL_ERROR("[Reduce][InPartOneToEven]rank[%u] tx ack from peerank[%u] failed", rank, peerRank), ret);
150 0 : ret = link->RxAck(stream_);
151 0 : CHK_PRT_RET(
152 : ret != HCCL_SUCCESS,
153 : HCCL_ERROR("[Reduce][InPartOneToEven]rank[%u] rx ack from peerank[%u] failed", rank, peerRank), ret);
154 : // 接收数据到本端的 output
155 0 : HCCL_DEBUG("send mem[%p] size[%llu] to peerank[%u]", outputMem_.ptr(), outputMem_.size(), peerRank);
156 0 : ret = link->TxAsync(UserMemType::INPUT_MEM, baseOffset_, outputMem_.ptr(), 0, stream_);
157 0 : CHK_PRT_RET(
158 : ret != HCCL_SUCCESS,
159 : HCCL_ERROR(
160 : "[Reduce][InPartOneToEven]TxAsync: tx async size[%llu] "
161 : "failed",
162 : 0),
163 : ret);
164 0 : CHK_RET(reducerInfo_->run(dispatcher_, link, baseOffset_, inputMem_, inputMem_, outputMem_, stream_));
165 0 : ret = link->RxWaitDone(stream_);
166 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Reduce][InPartOne]RxWaitDone failed"), ret);
167 : }
168 0 : } else if (rank < part1Size_ && rank % 2 == 1) { // 向上一个rank的output发数据 2
169 0 : u32 peerRank = rank - 1;
170 :
171 0 : if (peerRank < links.size()) {
172 0 : const LINK& link = links[peerRank];
173 0 : CHK_SMART_PTR_NULL(link);
174 0 : HcclResult ret = link->TxAck(stream_);
175 0 : CHK_PRT_RET(
176 : ret != HCCL_SUCCESS,
177 : HCCL_ERROR("[Reduce][InPartOneToEven]rank[%u] tx ack from peerank[%u] failed", rank, peerRank), ret);
178 0 : ret = link->RxAck(stream_);
179 0 : CHK_PRT_RET(
180 : ret != HCCL_SUCCESS,
181 : HCCL_ERROR("[Reduce][InPartOneToEven]rank[%u] rx ack from peerank[%u] failed", rank, peerRank), ret);
182 : // 发送到对端的output
183 0 : HCCL_DEBUG(
184 : "rank[%u] sends inputMem[%p] to PeerRank[%u] Offset[%llu], Size[%llu]", rank, inputMem_.ptr(), peerRank,
185 : baseOffset_, inputMem_.size());
186 0 : ret = senderInfo_->run(link, baseOffset_, inputMem_, stream_);
187 0 : CHK_PRT_RET(
188 : ret != HCCL_SUCCESS, HCCL_ERROR("[Reduce][InPartOne]tx sync to peerank[%u] failed", peerRank), ret);
189 0 : ret = link->RxAsync(UserMemType::OUTPUT_MEM, baseOffset_, inputMem_.ptr(), 0, stream_);
190 0 : CHK_PRT_RET(
191 : ret != HCCL_SUCCESS,
192 : HCCL_ERROR("[AlgTemplateBase][ExecuteTxSync]ExecuteTxSync: rx async size[%llu] failed", 0), ret);
193 0 : ret = link->DataReceivedAck(stream_);
194 0 : CHK_PRT_RET(
195 : ret != HCCL_SUCCESS,
196 : HCCL_ERROR("[AlgTemplateBase][ExecuteTxSync]ExecuteTxSync: data received ack failed"), ret);
197 0 : ret = link->TxWaitDone(stream_);
198 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Reduce][InPartOne]TxWaitDone failed"), ret);
199 : }
200 : }
201 :
202 0 : return HCCL_SUCCESS;
203 : }
204 :
205 : HcclResult
206 0 : AllReduceRecursiveHalvingDoubling::ReduceScatterInBlock(u32 rank, u32 rankSize, const std::vector<LINK>& links)
207 : {
208 0 : u32 rankInBlock = 0;
209 0 : if (rank < part1Size_ && (rank % 2) == 1) { // 模2判断奇偶性，本rank处于第一部分，并且为奇数rank
210 0 : return HCCL_SUCCESS;
211 0 : } else if (rank < part1Size_ && (rank % 2) == 0) { // 模2判断奇偶性，本rank 处于第一部分，并且为偶数rank
212 0 : rankInBlock = rank / 2; // 除2计算block内的rank值
213 : } else { // 本rank不属于第一部分
214 0 : rankInBlock = rank - part1Size_ / 2; // 除2计算block内的part1的范围
215 : }
216 : // 直接调用block的reducscatterhd算法
217 : std::unique_ptr<AlgTemplateBase> tempAlg
218 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_HD, dispatcher_);
219 0 : CHK_SMART_PTR_NULL(tempAlg);
220 0 : CHK_RET(tempAlg->Prepare(
221 : inputMem_, outputMem_, outputMem_, count_, dataType_, stream_, reductionOp_, root_, slices_, baseOffset_,
222 : blockSize_, reduceAttr, UserMemType::INPUT_MEM, UserMemType::OUTPUT_MEM));
223 :
224 0 : CHK_RET(tempAlg->RegisterProfiler(profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
225 :
226 : // 重新建立reducscatterscatter需要的链接
227 0 : std::vector<LINK> subLinks;
228 0 : CHK_RET(BuildSubLinks(links, subLinks, rankSize));
229 :
230 0 : CHK_PRT_RET(
231 : subLinks.size() == 0,
232 : HCCL_ERROR(
233 : "[AllReduceRecursiveHalvingDoubling][ReduceScatterInBlock]rank[%u] BuildSubLinks "
234 : "failed",
235 : rank),
236 : HCCL_E_PARA);
237 0 : CHK_RET(tempAlg->RunAsync(rankInBlock, blockSize_, subLinks));
238 0 : return HCCL_SUCCESS;
239 0 : }
240 :
241 0 : HcclResult AllReduceRecursiveHalvingDoubling::AllGatherInBlock(u32 rank, u32 rankSize, const std::vector<LINK>& links)
242 : {
243 0 : u32 rankInBlock = 0;
244 0 : if (rank < part1Size_ && (rank % 2) == 1) { // 模2判断奇偶性，本rank 处于第一部分，并且为奇数rank
245 0 : return HCCL_SUCCESS;
246 0 : } else if (rank < part1Size_ && (rank % 2) == 0) { // 模2判断奇偶性，本rank 处于第一部分，并且为偶数rank
247 0 : rankInBlock = rank / 2; // 在block内的rank为实际rank除以2
248 : } else {
249 0 : rankInBlock = rank - part1Size_ / 2; // 除2计算block内的part1的范围
250 : }
251 : // 直接调用block的allgatherhd算法
252 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
253 0 : TemplateType::TEMPLATE_ALL_GATHER_HALVING_DOUBLING, dispatcher_);
254 0 : CHK_SMART_PTR_NULL(tempAlg);
255 0 : CHK_RET(tempAlg->Prepare(blockSize_, UserMemType::OUTPUT_MEM, UserMemType::OUTPUT_MEM));
256 0 : CHK_RET(tempAlg->Prepare(
257 : outputMem_, outputMem_, count_, dataType_, stream_, reductionOp_, root_, slices_, baseOffset_));
258 :
259 0 : CHK_RET(tempAlg->RegisterProfiler(profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
260 :
261 : // 重新建立allgather需要的链接
262 0 : std::vector<LINK> subLinks;
263 0 : CHK_RET(BuildSubLinks(links, subLinks, rankSize));
264 :
265 0 : CHK_PRT_RET(
266 : subLinks.size() == 0,
267 : HCCL_ERROR(
268 : "[AllReduceRecursiveHalvingDoubling][AllGatherInBlock]rank[%u] build sub "
269 : "links failed",
270 : rank),
271 : HCCL_E_PARA);
272 :
273 0 : CHK_RET(tempAlg->RunAsync(rankInBlock, blockSize_, subLinks));
274 0 : return HCCL_SUCCESS;
275 0 : }
276 :
277 0 : HcclResult AllReduceRecursiveHalvingDoubling::GatherInPartOne(u32 rank, const std::vector<LINK>& links)
278 : {
279 0 : if (rank < part1Size_ && rank % 2 == 0) { // 模2判断奇偶性，本rank 处于第一部分，并且为偶数rank
280 0 : u32 peerRank = rank + 1;
281 : // 发送到对端的output
282 0 : if (peerRank < links.size()) {
283 0 : CHK_SMART_PTR_NULL(links[peerRank]);
284 0 : HcclResult ret = links[peerRank]->TxAck(stream_);
285 0 : CHK_PRT_RET(
286 : ret != HCCL_SUCCESS,
287 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] tx ack from peerank[%u] failed", rank, peerRank), ret);
288 0 : ret = links[peerRank]->RxAck(stream_);
289 0 : CHK_PRT_RET(
290 : ret != HCCL_SUCCESS,
291 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] rx ack from peerank[%u] failed", rank, peerRank), ret);
292 0 : HCCL_DEBUG(
293 : "rank[%u] outputMem[%p] sends to peerrank[%u] outputmem, offset[%llu], size[%llu]", rank,
294 : outputMem_.ptr(), peerRank, baseOffset_, outputMem_.size());
295 0 : ret = ExecuteTxSync(
296 0 : links[peerRank], UserMemType::OUTPUT_MEM, baseOffset_, outputMem_.ptr(), outputMem_.size(), stream_);
297 0 : CHK_PRT_RET(
298 : ret != HCCL_SUCCESS,
299 : HCCL_ERROR(
300 : "[AllReduceRecursiveHalvingDoubling][GatherInPartOne]rank[%u] tx "
301 : "sync to PeerRank[%u] failed",
302 : rank, peerRank),
303 : ret);
304 0 : ret = links[peerRank]->TxWaitDone(stream_);
305 0 : CHK_PRT_RET(
306 : ret != HCCL_SUCCESS,
307 : HCCL_ERROR("[AllReduceRecursiveHalvingDoubling][GatherInPartOne]TxWaitDone failed"), ret);
308 : }
309 0 : } else if (rank < part1Size_ && rank % 2 == 1) { // 模2判断奇偶性，本rank 处于第一部分，并且为奇数rank
310 0 : u32 peerRank = rank - 1;
311 0 : if (peerRank < links.size()) {
312 0 : CHK_SMART_PTR_NULL(links[peerRank]);
313 0 : HcclResult ret = links[peerRank]->TxAck(stream_);
314 0 : CHK_PRT_RET(
315 : ret != HCCL_SUCCESS,
316 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] tx ack from peerank[%u] failed", rank, peerRank), ret);
317 0 : ret = links[peerRank]->RxAck(stream_);
318 0 : CHK_PRT_RET(
319 : ret != HCCL_SUCCESS,
320 : HCCL_ERROR("[Gather][InPartOneToEven]rank[%u] rx ack from peerank[%u] failed", rank, peerRank), ret);
321 : // 等待对端可以接收数据
322 0 : HCCL_DEBUG(
323 : "rank[%u] outputMem[%p] receive from PeerRank[%u] outputMem, Offset[%llu], "
324 : "Size[%llu]",
325 : rank, outputMem_.ptr(), peerRank, baseOffset_, outputMem_.size());
326 0 : ret = ExecuteRxSync(
327 0 : links[peerRank], UserMemType::OUTPUT_MEM, baseOffset_, outputMem_.ptr(), outputMem_.size(), stream_);
328 0 : CHK_PRT_RET(
329 : ret != HCCL_SUCCESS,
330 : HCCL_ERROR(
331 : "[AllReduceRecursiveHalvingDoubling][GatherInPartOne]rank[%u] rx "
332 : "sync from PeerRank[%u] failed",
333 : rank, peerRank),
334 : ret);
335 0 : ret = links[peerRank]->RxWaitDone(stream_);
336 0 : CHK_PRT_RET(
337 : ret != HCCL_SUCCESS,
338 : HCCL_ERROR("[AllReduceRecursiveHalvingDoubling][GatherInPartOne]RxWaitDone failed"), ret);
339 : }
340 : }
341 :
342 0 : return HCCL_SUCCESS;
343 : }
344 :
345 0 : HcclResult AllReduceRecursiveHalvingDoubling::GetCommonNslbAdjInfo(
346 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links, AdjInfo& nslbAdjInfo)
347 : {
348 0 : u32 stepNum = 0;
349 0 : while ((rankSize >> (stepNum + 1)) != 0) {
350 0 : stepNum++;
351 : }
352 : // 执行reducscatter流程
353 0 : for (u32 step = 0; step < stepNum; step++) {
354 0 : u32 peerRankBitmask = 1 << (stepNum - step - 1);
355 0 : u32 peerRank = rank ^ peerRankBitmask;
356 0 : NslbDpAdjInfo adjInfoStep = {};
357 0 : u32 remoteuserRank = links[peerRank]->GetRemoteRank();
358 0 : adjInfoStep.dstLocalRankId = remoteuserRank;
359 0 : adjInfoStep.phaseId = step + 1;
360 0 : adjInfoStep.rev = 0;
361 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
362 : }
363 0 : u32 begin = stepNum;
364 : // 后续执行allgather流程
365 0 : for (u32 step = 0; step < stepNum; step++) {
366 0 : u32 peerRankBitmask = (1 << step);
367 0 : u32 peerRank = rank ^ peerRankBitmask;
368 0 : NslbDpAdjInfo adjInfoStep = {};
369 0 : u32 remoteuserRank = links[peerRank]->GetRemoteRank();
370 0 : adjInfoStep.dstLocalRankId = remoteuserRank;
371 0 : adjInfoStep.phaseId = step + begin + 1;
372 0 : adjInfoStep.rev = 0;
373 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
374 : }
375 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.nsAdjInfo.size();
376 0 : return HCCL_SUCCESS;
377 : }
378 0 : HcclResult AllReduceRecursiveHalvingDoubling::GetOddNslbAdjInfo(
379 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links, AdjInfo& nslbAdjInfo)
380 : {
381 : (void)rankSize;
382 0 : u32 peerRank = rank - 1;
383 0 : if (peerRank < links.size()) {
384 0 : NslbDpAdjInfo adjInfoStep = {};
385 0 : adjInfoStep.dstLocalRankId = links[peerRank]->GetRemoteRank();
386 0 : adjInfoStep.phaseId = 1;
387 0 : adjInfoStep.rev = 0;
388 0 : HCCL_INFO("AllGatherHDR-nslb: peerRank[%u]", peerRank);
389 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
390 0 : nslbAdjInfo.dstRankNum = 1;
391 : }
392 0 : return HCCL_SUCCESS;
393 : }
394 0 : HcclResult AllReduceRecursiveHalvingDoubling::GetNslbAdjInfo(
395 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links, AdjInfo& nslbAdjInfo)
396 : {
397 0 : u32 nslbRound = 0;
398 0 : u32 base = 1;
399 0 : const u32 minExponent = 1;
400 0 : while ((base << nslbRound) <= rankSize) {
401 0 : nslbRound++;
402 : }
403 0 : if (nslbRound >= minExponent) {
404 0 : nslbRound = nslbRound - minExponent;
405 : }
406 0 : u32 nslbBlockSize = base << nslbRound;
407 : // 获取第一部分：rank数减block数乘2
408 0 : u32 nslbPart1Size = (rankSize - nslbBlockSize) * NSLBDP_ALL_REDUCE_MOLD2;
409 : // 2的次幂场景下处理流程
410 0 : if (nslbPart1Size == 0) {
411 0 : GetCommonNslbAdjInfo(rank, rankSize, links, nslbAdjInfo);
412 0 : return HCCL_SUCCESS;
413 : }
414 : // 非2的次幂场景下，被合并部分的奇数rank处理流程
415 0 : if (rank < nslbPart1Size && rank % NSLBDP_ALL_REDUCE_MOLD2 == 1) {
416 0 : GetOddNslbAdjInfo(rank, rankSize, links, nslbAdjInfo);
417 0 : return HCCL_SUCCESS;
418 : }
419 : // 针对合并后映射成2的次幂场景处理
420 0 : u32 rankInBlock = 0;
421 0 : if (rank < nslbPart1Size && (rank % NSLBDP_ALL_REDUCE_MOLD2) == 0) {
422 0 : rankInBlock = rank / NSLBDP_ALL_REDUCE_MOLD2; // 直接除以2即为本rank的在block内的排序
423 : } else {
424 : rankInBlock
425 0 : = rank
426 0 : - nslbPart1Size / NSLBDP_ALL_REDUCE_MOLD2; // 通过rank减去part1除2的大小即不处于第一部分的block内rank号
427 : }
428 0 : std::vector<LINK> subLinks;
429 0 : std::vector<LINK>::const_iterator iter = links.begin();
430 0 : subLinks.resize(nslbBlockSize);
431 0 : for (u32 i = 0; i < rankSize; i++) {
432 0 : if (i < nslbPart1Size
433 0 : && (i % NSLBDP_ALL_REDUCE_MOLD2) == 1) { // 模2余1代表当前rank在part1的奇数位置上，不参与block内的建链
434 0 : continue;
435 0 : } else if (i < nslbPart1Size && (i % NSLBDP_ALL_REDUCE_MOLD2) == 0) { // 模2余0代表当前rank在part1的偶数位置上
436 0 : std::vector<LINK>::const_iterator niter = std::next(iter, i);
437 0 : if (niter != links.end()) {
438 0 : subLinks[i / NSLBDP_ALL_REDUCE_MOLD2] = *niter;
439 : }
440 0 : } else {
441 0 : std::vector<LINK>::const_iterator niter = std::next(iter, i);
442 0 : if (niter != links.end()) {
443 0 : subLinks[i - nslbPart1Size / NSLBDP_ALL_REDUCE_MOLD2] = *niter;
444 : }
445 : }
446 : }
447 0 : u32 stepNum = 0;
448 0 : while ((rankSize >> (stepNum + 1)) != 0) {
449 0 : stepNum++;
450 : }
451 : // 映射完成后针对以新的通信域进行邻接表获取
452 0 : u32 begin = 1;
453 0 : for (u32 step = 0; step < stepNum; step++) {
454 0 : u32 peerRankBitmask = 1 << (stepNum - step - 1);
455 0 : u32 peerRank = rankInBlock ^ peerRankBitmask;
456 0 : if (subLinks[peerRank] == nullptr) {
457 0 : continue;
458 : }
459 0 : NslbDpAdjInfo adjInfoStep = {};
460 0 : u32 remoteuserRank = subLinks[peerRank]->GetRemoteRank();
461 0 : adjInfoStep.dstLocalRankId = remoteuserRank;
462 0 : adjInfoStep.phaseId = step + 1 + begin;
463 0 : adjInfoStep.rev = 0;
464 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
465 : }
466 0 : nslbAdjInfo.dstRankNum = stepNum;
467 :
468 0 : if (nslbAdjInfo.nsAdjInfo.size() == 0) {
469 0 : return HCCL_SUCCESS;
470 : }
471 : // 上面处理完成后，紧接着处理合并部分的偶数rank同步到奇数rank增加phaseId
472 0 : if (rank < nslbPart1Size && rank % NSLBDP_ALL_REDUCE_MOLD2 == 0) {
473 0 : u32 peerRank = rank + 1;
474 0 : uint16_t phaseSize = nslbAdjInfo.nsAdjInfo.size();
475 0 : if (peerRank < links.size()) {
476 0 : NslbDpAdjInfo adjInfoStep = {};
477 0 : adjInfoStep.dstLocalRankId = links[peerRank]->GetRemoteRank();
478 0 : adjInfoStep.phaseId = nslbAdjInfo.nsAdjInfo[phaseSize - 1].phaseId + 1;
479 0 : adjInfoStep.rev = 0;
480 0 : HCCL_INFO("Scatter-nslb: peerRank[%u]", peerRank);
481 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
482 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.nsAdjInfo.size();
483 : }
484 0 : return HCCL_SUCCESS;
485 : }
486 0 : return HCCL_SUCCESS;
487 0 : }
488 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, AllReduceRecursiveHalvingDoubling);
489 : } // namespace hccl
|
