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