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_opbase_pipeline.h"
13 :
14 : constexpr u32 STEP_OFFSET_TWO = 2;
15 :
16 : namespace hccl {
17 0 : AllReduceOpbasePipeline::AllReduceOpbasePipeline(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
18 :
19 0 : AllReduceOpbasePipeline::~AllReduceOpbasePipeline() {}
20 :
21 0 : HcclResult AllReduceOpbasePipeline::Prepare(u64 reduceAttrBitMap, [[maybe_unused]] HcomCollOpInfo* opInfo)
22 : {
23 0 : reduceAttr_ = reduceAttrBitMap;
24 0 : return HCCL_SUCCESS;
25 : }
26 :
27 0 : HcclResult AllReduceOpbasePipeline::MainWaitSub()
28 : {
29 0 : u32 subStreamNum = intraRankSize_;
30 0 : for (u32 signalIndex = 0; signalIndex < subStreamNum; signalIndex++) {
31 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, streamNotifyMain_[signalIndex], INVALID_VALUE_STAGE));
32 : }
33 0 : return HCCL_SUCCESS;
34 : }
35 :
36 0 : HcclResult AllReduceOpbasePipeline::SubRecordMain()
37 : {
38 0 : u32 subStreamNum = intraRankSize_;
39 0 : for (u32 streamIndex = 0; streamIndex < subStreamNum; streamIndex++) {
40 0 : CHK_RET(LocalNotify::Post(subStreams_[streamIndex], dispatcher_, streamNotifyMain_[streamIndex], -1));
41 : }
42 0 : return HCCL_SUCCESS;
43 : }
44 :
45 0 : HcclResult AllReduceOpbasePipeline::MainRecordSub()
46 : {
47 0 : u32 subStreamNum = intraRankSize_;
48 0 : for (u32 signalIndex = 0; signalIndex < subStreamNum; signalIndex++) {
49 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, streamNotifySub_[signalIndex], -1));
50 : }
51 0 : return HCCL_SUCCESS;
52 : }
53 :
54 0 : HcclResult AllReduceOpbasePipeline::SubWaitMain()
55 : {
56 0 : u32 subStreamNum = intraRankSize_;
57 0 : for (u32 streamIndex = 0; streamIndex < subStreamNum; streamIndex++) {
58 0 : CHK_RET(LocalNotify::Wait(
59 : subStreams_[streamIndex], dispatcher_, streamNotifySub_[streamIndex], INVALID_VALUE_STAGE));
60 : }
61 0 : return HCCL_SUCCESS;
62 : }
63 :
64 0 : HcclResult AllReduceOpbasePipeline::RunReduceScatterIntraServer(u32 step)
65 : {
66 0 : u32 dmaMemSliceNum = dmaMem_.size();
67 0 : u32 dmaMemSliceId = step % dmaMemSliceNum;
68 0 : for (u32 i = 1; i < intraRankSize_; i++) {
69 0 : u32 remIntraRankId = (intraRankId_ + i) % intraRankSize_;
70 0 : CHK_RET(intraLinks_[remIntraRankId]->TxAck(subStreams_[i - 1]));
71 0 : CHK_RET(intraLinks_[remIntraRankId]->RxAck(subStreams_[i - 1]));
72 0 : void* remoteMemPtr = nullptr;
73 0 : CHK_RET(intraLinks_[remIntraRankId]->GetRemoteMem(
74 : dmaMemSliceId == 1 ? UserMemType::OUTPUT_MEM : UserMemType::INPUT_MEM, &remoteMemPtr));
75 0 : u32 sliceId = ((interRankId_ + step + 1) % interRankSize_) * intraRankSize_ + remIntraRankId;
76 0 : u64 srcOffset = sliceId * memSliceSize_;
77 0 : u64 offset = srcOffset % HCCL_MIN_SLICE_ALIGN_910B;
78 0 : u64 dataSize = memSliceSize_;
79 0 : u64 dataCount = sliceCount_;
80 0 : if (sliceId == (interRankSize_ * intraRankSize_ - 1)) {
81 0 : dataSize = lastSliceSize_;
82 0 : dataCount = lastSliceCount_;
83 : }
84 0 : DeviceMem src = DeviceMem::create(static_cast<u8*>(usrInMem_) + srcOffset, dataSize);
85 0 : DeviceMem dst = DeviceMem::create(static_cast<u8*>(remoteMemPtr) + offset, dataSize);
86 :
87 0 : CHK_RET(HcclReduceAsync(
88 : dispatcher_, src.ptr(), dataCount, dataType_, reductionOp_, subStreams_[i - 1], dst.ptr(),
89 : intraLinks_[remIntraRankId]->GetRemoteRank(), intraLinks_[remIntraRankId]->GetLinkType(),
90 : INLINE_REDUCE_BIT));
91 :
92 0 : CHK_RET(intraLinks_[remIntraRankId]->TxDataSignal(subStreams_[i - 1]));
93 0 : CHK_RET(intraLinks_[remIntraRankId]->RxDataSignal(subStreams_[i - 1]));
94 0 : }
95 0 : return HCCL_SUCCESS;
96 : }
97 :
98 0 : HcclResult AllReduceOpbasePipeline::RunAllGatherIntraServer(u32 step)
99 : {
100 0 : u32 dmaMemSliceNum = dmaMem_.size();
101 0 : u32 dmaMemSliceId = (interRankSize_ + step - 1) % dmaMemSliceNum;
102 0 : for (u32 i = 1; i < intraRankSize_; i++) {
103 0 : u32 remIntraRankId = (intraRankId_ + i) % intraRankSize_;
104 0 : CHK_RET(intraLinks_[remIntraRankId]->TxAck(subStreams_[i - 1]));
105 0 : CHK_RET(intraLinks_[remIntraRankId]->RxAck(subStreams_[i - 1]));
106 0 : void* remoteMemPtr = nullptr;
107 0 : CHK_RET(intraLinks_[remIntraRankId]->GetRemoteMem(
108 : dmaMemSliceId == 1 ? UserMemType::OUTPUT_MEM : UserMemType::INPUT_MEM, &remoteMemPtr));
109 0 : u32 sliceId = ((interRankId_ + step) % interRankSize_) * intraRankSize_ + remIntraRankId;
110 0 : u64 dstOffset = sliceId * memSliceSize_;
111 0 : u64 offset = dstOffset % HCCL_MIN_SLICE_ALIGN_910B;
112 0 : u64 dataSize = memSliceSize_;
113 0 : if (sliceId == (interRankSize_ * intraRankSize_ - 1)) {
114 0 : dataSize = lastSliceSize_;
115 : }
116 0 : DeviceMem src = DeviceMem::create(static_cast<u8*>(remoteMemPtr) + offset, dataSize);
117 0 : DeviceMem dst = DeviceMem::create(static_cast<u8*>(usrOutMem_) + dstOffset, dataSize);
118 0 : CHK_RET(HcclD2DMemcpyAsync(
119 : dispatcher_, dst, src, subStreams_[i - 1], intraLinks_[remIntraRankId]->GetRemoteRank(),
120 : intraLinks_[remIntraRankId]->GetLinkType()));
121 0 : CHK_RET(intraLinks_[remIntraRankId]->TxDataSignal(subStreams_[i - 1]));
122 0 : CHK_RET(intraLinks_[remIntraRankId]->RxDataSignal(subStreams_[i - 1]));
123 0 : }
124 0 : return HCCL_SUCCESS;
125 : }
126 :
127 : HcclResult
128 0 : AllReduceOpbasePipeline::RunReduceScatterInterServer(u32 step, const LINK& prevInterLink, const LINK& nextInterLink)
129 : {
130 0 : CHK_RET(prevInterLink->TxAck(stream_));
131 0 : CHK_RET(nextInterLink->RxAck(stream_));
132 0 : u32 dmaMemSliceNum = dmaMem_.size();
133 0 : u32 rxDMAMemSliceId = (step + 1) % dmaMemSliceNum;
134 0 : u32 txDMAMemSliceId = step % dmaMemSliceNum;
135 0 : UserMemType srcMemType = txDMAMemSliceId == 1 ? UserMemType::OUTPUT_MEM : UserMemType::INPUT_MEM;
136 0 : UserMemType dstMemType = rxDMAMemSliceId == 1 ? UserMemType::OUTPUT_MEM : UserMemType::INPUT_MEM;
137 0 : u32 txSliceId = ((interRankId_ + 1 + step) % interRankSize_) * intraRankSize_ + intraRankId_;
138 0 : u64 sliceMemOffset = memSliceSize_ * txSliceId;
139 0 : u64 offset = sliceMemOffset % HCCL_MIN_SLICE_ALIGN_910B;
140 0 : void* txLocalAddr = static_cast<u8*>(dmaMem_[txDMAMemSliceId].ptr()) + offset;
141 0 : u64 txDataSize = memSliceSize_;
142 0 : if (txSliceId == (interRankSize_ * intraRankSize_ - 1)) {
143 0 : txDataSize = lastSliceSize_;
144 : }
145 0 : DeviceMem srcMem = DeviceMem::create(txLocalAddr, txDataSize);
146 0 : CHK_RET(senderInfo_->run(nextInterLink, offset, srcMem, stream_, dstMemType));
147 0 : HCCL_DEBUG(
148 : "[AllReduceOpbasePipeline][RunReduceScatterInterServer] local rank[%u], localOffset[%llu],"
149 : "tx with slice[%llu]",
150 : rankId_, sliceMemOffset, curSize_);
151 0 : u32 rxSliceId = ((interRankId_ + 2 + step) % interRankSize_) * intraRankSize_ + intraRankId_;
152 0 : u64 rxSliceOffset = memSliceSize_ * rxSliceId;
153 0 : u64 rxOffset = (rxSliceOffset) % HCCL_MIN_SLICE_ALIGN_910B;
154 0 : void* rxLocalAddr = static_cast<u8*>(dmaMem_[rxDMAMemSliceId].ptr()) + rxOffset;
155 0 : u64 rxDataSize = memSliceSize_;
156 0 : if (rxSliceId == (interRankSize_ * intraRankSize_ - 1)) {
157 0 : rxDataSize = lastSliceSize_;
158 : }
159 0 : DeviceMem rxLocalMem = DeviceMem::create(rxLocalAddr, rxDataSize);
160 0 : DstMemType resultMem = DstMemType::RESULT_INPUT_MEM;
161 0 : CHK_RET(reducerInfo_->run(
162 : dispatcher_, prevInterLink, rxOffset, rxLocalMem, rxLocalMem, rxLocalMem, stream_, resultMem, srcMemType));
163 0 : return HCCL_SUCCESS;
164 0 : }
165 :
166 : HcclResult
167 0 : AllReduceOpbasePipeline::RunAllGatherInterServer(u32 step, const LINK& prevInterLink, const LINK& nextInterLink)
168 : {
169 0 : CHK_RET(prevInterLink->TxAck(stream_));
170 0 : CHK_RET(nextInterLink->RxAck(stream_));
171 0 : u32 dmaMemSliceNum = dmaMem_.size();
172 0 : u32 rxDMAMemSliceId = (interRankSize_ + step) % dmaMemSliceNum;
173 0 : u32 txDMAMemSliceId = (interRankSize_ + step - 1) % dmaMemSliceNum;
174 0 : UserMemType srcMemType = txDMAMemSliceId == 1 ? UserMemType::OUTPUT_MEM : UserMemType::INPUT_MEM;
175 0 : UserMemType dstMemType = rxDMAMemSliceId == 1 ? UserMemType::OUTPUT_MEM : UserMemType::INPUT_MEM;
176 0 : u32 txSliceId = ((interRankId_ + step) % interRankSize_) * intraRankSize_ + intraRankId_;
177 0 : u64 sliceMemOffset = memSliceSize_ * txSliceId;
178 0 : u64 offset = sliceMemOffset % HCCL_MIN_SLICE_ALIGN_910B;
179 0 : void* txLocalAddr = static_cast<u8*>(dmaMem_[txDMAMemSliceId].ptr()) + offset;
180 0 : u64 txDataSize = memSliceSize_;
181 0 : if (txSliceId == (interRankSize_ * intraRankSize_ - 1)) {
182 0 : txDataSize = lastSliceSize_;
183 : }
184 0 : DeviceMem srcMem = DeviceMem::create(txLocalAddr, txDataSize);
185 0 : CHK_RET(nextInterLink->TxAsync(
186 : dstMemType, offset, static_cast<u8*>(dmaMem_[txDMAMemSliceId].ptr()) + offset, txDataSize, stream_));
187 :
188 0 : u32 rxSliceId = ((interRankId_ + step + 1) % interRankSize_) * intraRankSize_ + intraRankId_;
189 0 : u64 rxSliceOffset = memSliceSize_ * rxSliceId;
190 0 : u64 rxOffset = rxSliceOffset % HCCL_MIN_SLICE_ALIGN_910B;
191 0 : void* rxLocalAddr = static_cast<u8*>(dmaMem_[rxDMAMemSliceId].ptr()) + rxOffset;
192 0 : u64 rxDataSize = memSliceSize_;
193 0 : if (rxSliceId == (interRankSize_ * intraRankSize_ - 1)) {
194 0 : rxDataSize = lastSliceSize_;
195 : }
196 0 : DeviceMem rxLocalMem = DeviceMem::create(rxLocalAddr, rxDataSize);
197 0 : CHK_RET(prevInterLink->RxAsync(
198 : srcMemType, rxOffset, static_cast<u8*>(dmaMem_[rxDMAMemSliceId].ptr()) + rxOffset, rxDataSize,
199 : stream_)); // wait
200 0 : return HCCL_SUCCESS;
201 0 : }
202 :
203 0 : HcclResult AllReduceOpbasePipeline::CopyToScratchBuffer(u32 step)
204 : {
205 0 : u32 dmaMemSliceNum = dmaMem_.size();
206 0 : u32 dmaMemSliceId = step % dmaMemSliceNum;
207 0 : u32 sliceId = ((interRankId_ + 1 + step) % interRankSize_) * intraRankSize_ + intraRankId_;
208 0 : u64 sliceMemOffset = memSliceSize_ * sliceId;
209 0 : u64 offset = sliceMemOffset % HCCL_MIN_SLICE_ALIGN_910B;
210 : // 把一块切片从userIn 做拷贝到CCLBuffer
211 0 : void* srcAddr = static_cast<u8*>(usrInMem_) + sliceMemOffset;
212 0 : u64 dataSize = memSliceSize_;
213 0 : if (sliceId == (interRankSize_ * intraRankSize_ - 1)) {
214 0 : dataSize = lastSliceSize_;
215 : }
216 0 : DeviceMem locSrc = DeviceMem::create(srcAddr, dataSize);
217 0 : DeviceMem locDst = DeviceMem::create(static_cast<u8*>(dmaMem_[dmaMemSliceId].ptr()) + offset, dataSize);
218 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, locDst, locSrc, stream_));
219 0 : return HCCL_SUCCESS;
220 0 : }
221 :
222 0 : HcclResult AllReduceOpbasePipeline::RunAsync()
223 : {
224 : // inter ring algo
225 0 : u32 prevInterRankId = (interRankId_ + 1) % interRankSize_;
226 0 : u32 nextInterRankId = (interRankId_ - 1 + interRankSize_) % interRankSize_;
227 0 : LINK prevInterLink = interLinks_[prevInterRankId];
228 0 : LINK nextInterLink = interLinks_[nextInterRankId];
229 : // 当前使用2块DMAMem buffer
230 0 : u32 dmaMemSliceNum = dmaMem_.size();
231 0 : for (u32 step = 0; step < interRankSize_; step++) {
232 0 : if (step == 0) {
233 : // 把第一块切片从userIn 做拷贝到CCLBuffer
234 0 : CHK_RET(CopyToScratchBuffer(step));
235 0 : CHK_RET(MainRecordSub());
236 0 : CHK_RET(SubWaitMain());
237 : }
238 : // server内做SDMA的reduce
239 0 : CHK_RET(RunReduceScatterIntraServer(step));
240 0 : CHK_RET(SubRecordMain());
241 0 : CHK_RET(MainWaitSub());
242 0 : if (step < interRankSize_ - 1) {
243 : // 把下一块切片从userIn 做拷贝到CCLBuffer
244 0 : CHK_RET(CopyToScratchBuffer(step + 1));
245 0 : CHK_RET(MainRecordSub());
246 0 : CHK_RET(SubWaitMain());
247 : // server间做RDMA的reduce，可与下一个step的SDMA并发执行
248 0 : CHK_RET(RunReduceScatterInterServer(step, prevInterLink, nextInterLink));
249 : // 确保step[n+2]的SDMA之前step[n]的RDMA已经完成，防止内存踩踏
250 0 : CHK_RET(prevInterLink->PostFinAck(stream_));
251 0 : CHK_RET(nextInterLink->WaitFinAck(stream_));
252 : }
253 : }
254 0 : DeviceMem srcZero = DeviceMem::create(usrInMem_, 0);
255 0 : DeviceMem dstZero = DeviceMem::create(usrOutMem_, 0);
256 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstZero, srcZero, stream_));
257 :
258 0 : for (u32 step = 0; step < interRankSize_; step++) {
259 0 : CHK_RET(MainRecordSub());
260 0 : CHK_RET(SubWaitMain());
261 0 : if (step < interRankSize_ - 1) {
262 0 : CHK_RET(RunAllGatherInterServer(step, prevInterLink, nextInterLink));
263 0 : CHK_RET(prevInterLink->PostFinAck(stream_));
264 0 : CHK_RET(nextInterLink->WaitFinAck(stream_));
265 : // inter的最后一步需要barrier确保数据发完
266 0 : if (step == interRankSize_ - STEP_OFFSET_TWO) {
267 0 : CHK_RET(ExecuteBarrier(prevInterLink, nextInterLink));
268 : }
269 : }
270 0 : CHK_RET(RunAllGatherIntraServer(step));
271 0 : CHK_RET(SubRecordMain());
272 0 : CHK_RET(MainWaitSub());
273 0 : u32 DMAMemSliceId = (interRankSize_ + step - 1) % dmaMemSliceNum;
274 0 : u32 sliceId = ((interRankId_ + step) % interRankSize_) * intraRankSize_ + intraRankId_;
275 0 : u64 sliceMemOffset = memSliceSize_ * sliceId;
276 0 : u64 offset = sliceMemOffset % HCCL_MIN_SLICE_ALIGN_910B;
277 0 : void* txLocalAddr = static_cast<u8*>(dmaMem_[DMAMemSliceId].ptr()) + offset;
278 0 : u64 dataSize = memSliceSize_;
279 0 : if (sliceId == (interRankSize_ * intraRankSize_ - 1)) {
280 0 : dataSize = lastSliceSize_;
281 : }
282 0 : DeviceMem srcMem = DeviceMem::create(txLocalAddr, dataSize);
283 0 : DeviceMem dstMem = DeviceMem::create(static_cast<u8*>(usrOutMem_) + sliceMemOffset, dataSize);
284 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream_));
285 0 : HCCL_INFO("[AllReducePipeline][RunAsync]AllReducePipeline finished groupRankId[%u] ", rankId_);
286 0 : }
287 0 : return HCCL_SUCCESS;
288 0 : }
289 :
290 0 : HcclResult AllReduceOpbasePipeline::Prepare(
291 : const HcomCollOpInfo* opInfo, DeviceMem& cclBufferA, DeviceMem& cclBufferB, const u64 count,
292 : const SubCommInfo& level1CommInfo, const SubCommInfo& level0CommInfo, Stream& mainStream,
293 : std::vector<Stream>& subStream, std::vector<std::shared_ptr<LocalNotify>>& notifyMain,
294 : std::vector<std::shared_ptr<LocalNotify>>& notifySub)
295 : {
296 0 : unitSize_ = SIZE_TABLE[opInfo->dataType];
297 0 : sliceCount_ = count / (level0CommInfo.localRankSize * level1CommInfo.localRankSize);
298 0 : memSliceSize_ = sliceCount_ * unitSize_;
299 0 : lastSliceCount_ = count - sliceCount_ * (level0CommInfo.localRankSize * level1CommInfo.localRankSize - 1);
300 0 : lastSliceSize_ = lastSliceCount_ * unitSize_;
301 :
302 0 : usrInMem_ = opInfo->inputAddr;
303 0 : usrOutMem_ = opInfo->outputAddr;
304 0 : reductionOp_ = opInfo->reduceOp;
305 0 : dataType_ = opInfo->dataType;
306 :
307 : // needed resource
308 : // stream: 1 * mainStream + (n -1) * subStream
309 : // interNotify, streamNotify
310 :
311 : // stream
312 : // mainStream负责locMemCpy、inter执行以及subStream同步控制
313 0 : stream_ = mainStream;
314 : // subStream负责:
315 : // streamId[0:intraRankSize-1]: intraRankSize-1个intra执行
316 0 : subStreams_ = subStream;
317 :
318 0 : intraRankSize_ = level0CommInfo.localRankSize;
319 0 : interRankSize_ = level1CommInfo.localRankSize;
320 0 : intraRankId_ = level0CommInfo.localRank;
321 0 : interRankId_ = level1CommInfo.localRank;
322 0 : rankId_ = intraRankId_ + interRankId_ * intraRankSize_;
323 :
324 : // streamNotify, size: n
325 0 : streamNotifyMain_ = notifyMain;
326 0 : if (streamNotifyMain_.size() < intraRankSize_) {
327 0 : HCCL_ERROR(
328 : "[AllReduceOpbasePipeline][Prepare]rank[%u] streamNotifyMain_ size [%u] error, is smaller than,"
329 : "intraRankSize_[%u]",
330 : rankId_, streamNotifyMain_.size(), intraRankSize_);
331 0 : return HCCL_E_INTERNAL;
332 : }
333 0 : streamNotifySub_ = notifySub;
334 0 : if (streamNotifySub_.size() < intraRankSize_) {
335 0 : HCCL_ERROR(
336 : "[AllReduceOpbasePipeline][Prepare]rank[%u] streamNotifySub_ size [%u] error, is smaller than,"
337 : "intraRankSize_[%u]",
338 : rankId_, streamNotifySub_.size(), intraRankSize_);
339 0 : return HCCL_E_INTERNAL;
340 : }
341 :
342 0 : intraLinks_ = level0CommInfo.links;
343 0 : interLinks_ = level1CommInfo.links;
344 :
345 0 : count_ = count;
346 : // 2级流水,使用2块DMAMem
347 0 : dmaMem_.push_back(cclBufferA);
348 0 : dmaMem_.push_back(cclBufferB);
349 :
350 0 : HCCL_INFO(
351 : "[AllReduceOpbasePipeline][Prepare]streamNum[%u], streamNotifyMainNum[%u], streamNotifySubNum[%u]",
352 : subStreams_.size(), streamNotifyMain_.size(), streamNotifySub_.size());
353 0 : HCCL_INFO(
354 : "[AllReduceOpbasePipeline][Prepare]interLinksNum[%u], intraLinksNum[%u]", interLinks_.size(),
355 : intraLinks_.size());
356 0 : senderInfo_.reset(new (std::nothrow) Sender(dataType_, reductionOp_, reduceAttr_));
357 0 : CHK_SMART_PTR_NULL(senderInfo_);
358 0 : reducerInfo_.reset(new (std::nothrow) Reducer(dataType_, reductionOp_, reduceAttr_));
359 0 : CHK_SMART_PTR_NULL(reducerInfo_);
360 0 : return HCCL_SUCCESS;
361 : }
362 :
363 0 : HcclResult AllReduceOpbasePipeline::GetNslbAdjInfo(
364 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links, AdjInfo& nslbAdjInfo)
365 : {
366 0 : u32 ringNextRank = (rank + 1) % rankSize;
367 0 : LINK nslbNext = links[ringNextRank];
368 0 : CHK_SMART_PTR_NULL(nslbNext);
369 :
370 : // Pipeline 步长合并 等同于 ring
371 0 : NslbDpAdjInfo adjInfoStep = {};
372 0 : adjInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
373 0 : adjInfoStep.phaseId = 1;
374 0 : adjInfoStep.rev = 0;
375 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
376 :
377 : // Pipeline 步长合并 等同于 ring
378 0 : NslbDpAdjInfo nextInfoStep = {};
379 0 : nextInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
380 0 : nextInfoStep.phaseId = nslbAdjInfo.nsAdjInfo[0].phaseId + 1;
381 0 : nextInfoStep.rev = 0;
382 0 : nslbAdjInfo.nsAdjInfo.push_back(nextInfoStep);
383 :
384 0 : nslbAdjInfo.dstRankNum = nslbAdjInfo.nsAdjInfo.size();
385 0 : return HCCL_SUCCESS;
386 0 : }
387 :
388 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_REDUCE_OPBASE_PIPELINE, AllReduceOpbasePipeline);
389 : } // namespace hccl
|
