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 "reduce_scatter_ring.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 34 : ReduceScatterRing::ReduceScatterRing(const HcclDispatcher dispatcher) : AlgTemplateBase(dispatcher) {}
16 :
17 59 : ReduceScatterRing::~ReduceScatterRing() {}
18 :
19 25 : HcclResult ReduceScatterRing::Prepare(u64 reduceAttrBitMap, HcomCollOpInfo* opInfo)
20 : {
21 : (void)opInfo;
22 25 : reduceAttr_ = reduceAttrBitMap;
23 25 : return HCCL_SUCCESS;
24 : }
25 :
26 0 : HcclResult ReduceScatterRing::RunVectorSourceReducer(
27 : const LINK& link, const std::vector<Slice>& txSlices, const std::vector<Slice>& txSlicetemp)
28 : {
29 : /* 1、对外reduce_scatter，output的大小为每块数据*rank_size。只能发送到对端地址偏移为0开始。
30 : 2、allreduce中使用reduce_scatter，output与Input大小相等，接收和发送偏移相等都为slice.offset */
31 0 : std::vector<SenderMemoryInfo> txMems;
32 0 : for (u32 i = 0; i < txSlices.size(); i++) {
33 0 : DeviceMem srcMem = inputMem_.range(txSlices[i].offset, txSlices[i].size);
34 0 : HCCL_DEBUG(
35 : " send inputmem range[%llu], size[%llu] tx dstmem offset[%llu]", txSlices[i].offset, txSlices[i].size,
36 : txSlicetemp[i].offset);
37 0 : txMems.emplace_back(SenderMemoryInfo{baseOffset_ + txSlicetemp[i].offset, srcMem});
38 0 : }
39 0 : CHK_RET(senderInfo_->run(link, txMems, stream_));
40 :
41 0 : return HCCL_SUCCESS;
42 0 : }
43 :
44 0 : HcclResult ReduceScatterRing::RunVectorDestRducer(
45 : const LINK& link, const std::vector<Slice>& rxSlices, const std::vector<Slice>& rxSlicetemp)
46 : {
47 0 : std::vector<ReducerMemoryInfo> rxReduceMems;
48 0 : for (u32 i = 0; i < rxSlices.size(); i++) {
49 0 : DeviceMem dstMem = inputMem_.range(rxSlices[i].offset, rxSlices[i].size);
50 0 : DeviceMem srcMemTemp = scratchMem_.range(rxSlicetemp[i].offset, rxSlicetemp[i].size);
51 0 : HCCL_DEBUG(
52 : "rcv offset[%llu], size[%llu] ,then reduce with offset[%llu] size[%llu] ", rxSlicetemp[i].offset,
53 : rxSlicetemp[i].size, rxSlices[i].offset, rxSlices[i].size);
54 0 : rxReduceMems.emplace_back(ReducerMemoryInfo{baseOffset_ + rxSlices[i].offset, dstMem, dstMem, srcMemTemp});
55 0 : }
56 0 : CHK_RET(reducerInfo_->run(dispatcher_, link, rxReduceMems, stream_));
57 :
58 0 : return HCCL_SUCCESS;
59 0 : }
60 :
61 0 : HcclResult ReduceScatterRing::RunSourceReducer(const LINK& link, const Slice& txSlice, const Slice& txSlicetemp)
62 : {
63 : /* 1、对外reduce_scatter，output的大小为每块数据*rank_size。只能发送到对端地址偏移为0开始。
64 : 2、allreduce中使用reduce_scatter，output与Input大小相等，接收和发送偏移相等都为slice.offset */
65 0 : DeviceMem srcMem = inputMem_.range(txSlice.offset, txSlice.size);
66 0 : HCCL_DEBUG(
67 : " send inputmem range[%llu], size[%llu] tx dstmem offset[%llu]", txSlice.offset, txSlice.size,
68 : txSlicetemp.offset);
69 0 : CHK_RET(senderInfo_->run(link, baseOffset_ + txSlicetemp.offset, srcMem, stream_));
70 :
71 0 : return HCCL_SUCCESS;
72 0 : }
73 :
74 0 : HcclResult ReduceScatterRing::RunDestRducer(const LINK& link, const Slice& rxSlice, const Slice& rxSlicetemp)
75 : {
76 0 : DeviceMem dstMem = inputMem_.range(rxSlice.offset, rxSlice.size);
77 0 : DeviceMem srcMemTemp = scratchMem_.range(rxSlicetemp.offset, rxSlicetemp.size);
78 0 : HCCL_DEBUG(
79 : "rcv offset[%llu], size[%llu] ,then reduce with "
80 : "offset[%llu] size[%llu] ",
81 : rxSlicetemp.offset, rxSlicetemp.size, rxSlice.offset, rxSlice.size);
82 0 : CHK_RET(reducerInfo_->run(dispatcher_, link, baseOffset_ + rxSlice.offset, dstMem, dstMem, srcMemTemp, stream_));
83 :
84 0 : return HCCL_SUCCESS;
85 0 : }
86 :
87 0 : HcclResult ReduceScatterRing::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
88 : {
89 : // 判断stream, dispatcher是否为空
90 0 : CHK_SMART_PTR_NULL(dispatcher_);
91 0 : CHK_PTR_NULL(stream_.ptr());
92 0 : if (!outputMem_ || !inputMem_) {
93 0 : HCCL_ERROR("[ReduceScatterRing][RunAsync]rank[%u] run_async inputmem or outputmem is null", rank);
94 0 : return HCCL_E_PTR;
95 : }
96 0 : HCCL_INFO(
97 : "ReduceScatterRing run: rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", rank, rankSize,
98 : inputMem_.ptr(), outputMem_.ptr(), count_);
99 :
100 : // 判断rank_size == 1
101 0 : if (rankSize == 1) {
102 0 : if (inputMem_ != outputMem_) {
103 0 : return HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
104 : }
105 0 : HCCL_DEBUG("[ReduceScatterRing]rankSize is 1, RunAsync success");
106 0 : return HCCL_SUCCESS;
107 : }
108 :
109 : // 创建reducer & sender
110 0 : senderInfo_.reset(new (std::nothrow) Sender(dataType_, reductionOp_, reduceAttr_));
111 0 : CHK_SMART_PTR_NULL(senderInfo_);
112 :
113 0 : reducerInfo_.reset(new (std::nothrow) Reducer(dataType_, reductionOp_, reduceAttr_));
114 0 : CHK_SMART_PTR_NULL(reducerInfo_);
115 :
116 0 : if (links.size() < rankSize) {
117 0 : HCCL_ERROR(
118 : "[ReduceScatterRing][RunAsync]rank[%u] link size[%llu] is less than rank size[%u]", rank, links.size(),
119 : rankSize);
120 0 : return HCCL_E_INTERNAL;
121 : }
122 :
123 0 : u32 ringPrevRank = (rank + rankSize - 1) % rankSize;
124 0 : linkLeft_ = links[ringPrevRank];
125 0 : CHK_SMART_PTR_NULL(linkLeft_);
126 :
127 0 : u32 ringNextRank = (rank + 1) % rankSize;
128 0 : HCCL_DEBUG("[ReduceScatterRing][RunAsync]ringPrevRank is %u, ringNextRank is %u", ringPrevRank, ringNextRank);
129 0 : linkRight_ = links[ringNextRank];
130 0 : CHK_SMART_PTR_NULL(linkRight_);
131 :
132 0 : u32 unitSize = DataUnitSize(dataType_);
133 0 : if (unitSize == 0) {
134 0 : HCCL_ERROR("[ReduceScatterRing][RunAsync]rank[%u] unit data size is zero", rank);
135 0 : return HCCL_E_INTERNAL;
136 : }
137 :
138 0 : std::vector<Slice> outputSlices(slices_);
139 :
140 0 : if (slices_.size() == 0) {
141 0 : slices_.resize(rankSize);
142 0 : outputSlices.resize(rankSize);
143 :
144 : // 生成std::vector<Slice> slices_
145 0 : u64 sliceSize = count_ * unitSize;
146 :
147 0 : for (u32 i = 0; i < rankSize; i++) {
148 0 : slices_[i].size = sliceSize;
149 0 : slices_[i].offset = (i * sliceSize);
150 :
151 0 : outputSlices[i].size = sliceSize;
152 0 : outputSlices[i].offset = (inputMem_.size() > outputMem_.size()) ? 0 : (i * sliceSize);
153 0 : HCCL_DEBUG(
154 : "rank[%u], slices[%u].offset=[%llu], slices[%u].size=[%llu] outputSlices[%u].offset=[%llu], "
155 : "outputSlices[%u].size=[%llu] ",
156 : rank, i, slices_[i].offset, i, slices_[i].size, i, outputSlices[i].offset, i, outputSlices[i].size);
157 : }
158 : }
159 :
160 : // 运行reduce-scatter, ring算法
161 : // 单环场景下 nicRankList_ 长度默认为 8。
162 : // 多环场景下 nicRankList_ 长度为网口数量。此时若 rankSize != nicRankList_ 则为网口裁剪场景
163 0 : if (rankSize != HCCL_NIC_MAX_NUM || nicRankList_.size() == HCCL_NIC_MAX_NUM) {
164 : // 非网口裁剪场景:
165 0 : CHK_RET(RunReduceScatter(rank, rankSize, slices_, outputSlices));
166 : } else {
167 : // 网口裁剪场景：当前仅在 910A 8P_RING (4环)，且网口不满配情况下使用
168 0 : CHK_RET(ReduceScatterSlicesPrep(rankSize, nicRankList_.size()));
169 :
170 0 : CHK_RET(RunReduceScatterChunk(rank, rankSize, slices_, outputSlices));
171 : }
172 :
173 0 : if (barrierSwitchOn_) {
174 : // 执行barrier，保证数据收发完成
175 0 : CHK_RET(ExecuteBarrier(linkLeft_, linkRight_));
176 : }
177 :
178 0 : HCCL_INFO("ReduceScatterRing finished: rank[%u]", rank);
179 0 : return HCCL_SUCCESS;
180 0 : }
181 :
182 0 : HcclResult ReduceScatterRing::RunReduceScatter(
183 : const u32 rank, const u32 rankSize, const std::vector<Slice>& inputSlices, const std::vector<Slice>& outputSlices)
184 : {
185 0 : bool bRetSize = (inputSlices.size() < rankSize);
186 0 : CHK_PRT_RET(
187 : bRetSize,
188 : HCCL_ERROR(
189 : "[Run][ReduceScatter]rank[%u] inputslice size[%llu] is less than rank size[%u]", rank, inputSlices.size(),
190 : rankSize),
191 : HCCL_E_INTERNAL);
192 :
193 0 : bRetSize = (outputSlices.size() < rankSize);
194 0 : HCCL_DEBUG("[Run][ReduceScatter]ReduceScatterRing for bRetSize is %d", bRetSize);
195 0 : CHK_PRT_RET(
196 : bRetSize,
197 : HCCL_ERROR(
198 : "[Run][ReduceScatter]rank[%u] outputslice size[%llu] is less than rank size[%u]", rank, outputSlices.size(),
199 : rankSize),
200 : HCCL_E_INTERNAL);
201 :
202 0 : HcclResult ret = HCCL_SUCCESS;
203 :
204 0 : u32 sliceSize = inputSlices.size() / rankSize;
205 :
206 : // 获取rx_slice, 首先向本rank前2个rank处发ack消息
207 0 : u32 rxSliceIndex = (rank + rankSize - 2) % rankSize;
208 0 : HCCL_DEBUG(
209 : "rank[%u] outputslice[%u] size[%llu] will tx_ack to left", rank, rxSliceIndex, outputSlices[rxSliceIndex].size);
210 :
211 0 : CHK_RET(linkLeft_->TxAck(stream_)); // NotifyRecord
212 :
213 : // reduce源操作, 获取tx_slice，从本rank前一rank开始接收ack
214 0 : u32 txSliceIndex = (rank + rankSize - 1) % rankSize;
215 0 : HCCL_DEBUG("rank[%u] inputSlices[%u] will rx_ack", rank, txSliceIndex);
216 :
217 0 : CHK_RET(linkRight_->RxAck(stream_)); // NotifyWait
218 :
219 0 : std::vector<Slice> txInputSegsSlice;
220 0 : std::vector<Slice> txOutputSegsSlice;
221 0 : for (u32 j = 0; j < sliceSize; j++) {
222 0 : txInputSegsSlice.push_back(inputSlices[txSliceIndex * sliceSize + j]);
223 0 : txOutputSegsSlice.push_back(outputSlices[txSliceIndex * sliceSize + j]);
224 : }
225 0 : ret = RunVectorSourceReducer(linkRight_, txInputSegsSlice, txOutputSegsSlice); // NotifyRecord
226 0 : CHK_PRT_RET(
227 : ret != HCCL_SUCCESS,
228 : HCCL_ERROR("[Run][ReduceScatter]rank[%u] txSliceIndex[%u] Reducer src run failed.", rank, txSliceIndex), ret);
229 :
230 : // 本rank既当reduce源, 也当reduce操作的目的
231 0 : for (u32 i = 0; i < (rankSize - 2); i++) { // 中间rank_size - 2次传输
232 : // reduce目的操作
233 0 : HCCL_DEBUG(
234 : "rank[%u] round[%u] use sliceindex[%u] inputSlicesoffset[%llu] "
235 : "size[%llu] outputSlices offset[%llu] rcv and reduce",
236 : rank, i, rxSliceIndex, inputSlices[rxSliceIndex].offset, inputSlices[rxSliceIndex].size,
237 : outputSlices[rxSliceIndex].offset);
238 :
239 0 : std::vector<Slice> rxInputSegsSlice;
240 0 : std::vector<Slice> rxOutputSegsSlice;
241 0 : HCCL_DEBUG("[ReduceScatterRing]RunReduceScatter for sliceSize is %u", sliceSize);
242 0 : for (u32 j = 0; j < sliceSize; j++) {
243 0 : rxInputSegsSlice.push_back(inputSlices[rxSliceIndex * sliceSize + j]);
244 0 : rxOutputSegsSlice.push_back(outputSlices[rxSliceIndex * sliceSize + j]);
245 : }
246 0 : ret = RunVectorDestRducer(linkLeft_, rxInputSegsSlice, rxOutputSegsSlice);
247 :
248 0 : CHK_PRT_RET(
249 : ret != HCCL_SUCCESS,
250 : HCCL_ERROR(
251 : "[Run][ReduceScatter]rank[%u] round[%u] rxSlice[%u] Reducer dst run failed", rank, i, rxSliceIndex),
252 : ret);
253 :
254 0 : ret = linkLeft_->RxWaitDone(stream_);
255 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Run][ReduceScatter]RxWaitDone failed"), ret);
256 0 : ret = linkRight_->TxWaitDone(stream_);
257 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Run][ReduceScatter]TxWaitDone failed"), ret);
258 :
259 : // 获取rx_slice
260 0 : rxSliceIndex = (rxSliceIndex + rankSize - 1) % rankSize;
261 :
262 0 : HCCL_DEBUG("rank[%u] round[%u] outputslice[%u] will tx_ack ", rank, i, rxSliceIndex);
263 0 : ret = linkLeft_->TxAck(stream_);
264 0 : CHK_PRT_RET(
265 : ret != HCCL_SUCCESS, HCCL_ERROR("[Run][ReduceScatter]rank[%u] round[%u] tx ack failed", rank, i), ret);
266 :
267 : // reduce源操作, 获取tx_slice
268 0 : txSliceIndex = (txSliceIndex + rankSize - 1) % rankSize;
269 0 : HCCL_DEBUG("rank[%u] round[%u] use sliceindex[%u] rx ack ", rank, i, txSliceIndex);
270 0 : ret = linkRight_->RxAck(stream_);
271 0 : CHK_PRT_RET(
272 : ret != HCCL_SUCCESS, HCCL_ERROR("[Run][ReduceScatter]rank[%u] round[%u] rx ack run failed", rank, i), ret);
273 :
274 0 : std::vector<Slice> txInputSegsSlice;
275 0 : std::vector<Slice> txOutputSegsSlice;
276 0 : for (u32 j = 0; j < sliceSize; j++) {
277 0 : txInputSegsSlice.push_back(inputSlices[txSliceIndex * sliceSize + j]);
278 0 : txOutputSegsSlice.push_back(outputSlices[txSliceIndex * sliceSize + j]);
279 : }
280 0 : ret = RunVectorSourceReducer(linkRight_, txInputSegsSlice, txOutputSegsSlice);
281 0 : CHK_PRT_RET(
282 : ret != HCCL_SUCCESS, HCCL_ERROR("[Run][ReduceScatter]rank[%u] round[%u] Reducer src run failed", rank, i),
283 : ret);
284 0 : }
285 :
286 : /* * 末尾传输, 本rank只当reduce目的, 根据单buffer还是双buffer来决定如何搬移
287 : 当前简化处理, 只考虑单buffer的场景, 双buffer则在run_async中多拷贝一次 */
288 0 : std::vector<ReducerMemoryInfo> rxReduceMems;
289 0 : for (u32 i = 0; i < sliceSize; i++) {
290 : DeviceMem dstMem
291 0 : = outputMem_.range(outputSlices[rank * sliceSize + i].offset, outputSlices[rank * sliceSize + i].size);
292 : // reduce目的操作
293 : DeviceMem srcMem
294 0 : = inputMem_.range(inputSlices[rank * sliceSize + i].offset, inputSlices[rank * sliceSize + i].size);
295 : DeviceMem scratchMem
296 0 : = scratchMem_.range(outputSlices[rank * sliceSize + i].offset, outputSlices[rank * sliceSize + i].size);
297 0 : HCCL_DEBUG(
298 : "final reduce rank[%u] reduce with inputMem_ offset[%llu] to ouput_mem_ offset[%llu] size[%llu]", rank,
299 : inputSlices[rank * sliceSize + i].offset, outputSlices[rank * sliceSize + i].offset,
300 : outputSlices[rank * sliceSize + i].size);
301 0 : rxReduceMems.emplace_back(
302 0 : ReducerMemoryInfo{baseOffset_ + inputSlices[rank * sliceSize + i].offset, srcMem, dstMem, scratchMem});
303 0 : }
304 0 : CHK_RET(reducerInfo_->run(dispatcher_, linkLeft_, rxReduceMems, stream_));
305 :
306 0 : ret = linkLeft_->RxWaitDone(stream_);
307 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Run][ReduceScatter]RxWaitDone failed"), ret);
308 0 : ret = linkRight_->TxWaitDone(stream_);
309 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Run][ReduceScatter]TxWaitDone failed"), ret);
310 :
311 0 : return HCCL_SUCCESS;
312 0 : }
313 :
314 0 : HcclResult ReduceScatterRing::RunReduceScatterChunk(
315 : const u32 rank, const u32 rankSize, const std::vector<Slice>& inputSlices, const std::vector<Slice>& outputSlices)
316 : {
317 0 : bool bRetSize = (inputSlices.size() < rankSize);
318 0 : CHK_PRT_RET(
319 : bRetSize,
320 : HCCL_ERROR(
321 : "[Run][ReduceScatterChunk]rank[%u] inputslice size[%llu] is less than rank "
322 : "size[%u]",
323 : rank, outputSlices.size(), rankSize),
324 : HCCL_E_INTERNAL);
325 :
326 0 : bRetSize = (outputSlices.size() < rankSize);
327 0 : CHK_PRT_RET(
328 : bRetSize,
329 : HCCL_ERROR(
330 : "[Run][ReduceScatterChunk]rank[%u] outputslice size[%llu] is less than rank "
331 : "size[%u]",
332 : rank, outputSlices.size(), rankSize),
333 : HCCL_E_INTERNAL);
334 :
335 : HcclResult ret;
336 0 : u32 sendSliceLen = rankSliceLists_[rank].size();
337 0 : u32 chunkSize = HCCL_NIC_MAX_NUM / nicRankList_.size();
338 0 : if (sendSliceLen >= chunkSize) {
339 0 : CHK_RET(HeadReduceScatterChunk(rank, rankSize, inputSlices, outputSlices));
340 0 : for (u32 midRankIdx = 1; midRankIdx < sendSliceLen - 1; midRankIdx++) {
341 0 : ret = MidReduceScatterChunk(rank, rankSize, midRankIdx, inputSlices, outputSlices);
342 0 : CHK_PRT_RET(
343 : ret != HCCL_SUCCESS,
344 : HCCL_ERROR(
345 : "[Run][ReduceScatterChunk]rank[%u] run mid[%u] ReduceScatter chunk failed", rank, midRankIdx),
346 : HCCL_E_INTERNAL);
347 : }
348 : }
349 0 : CHK_RET(TailReduceScatterChunk(rank, rankSize, sendSliceLen - 1, inputSlices, outputSlices));
350 0 : std::vector<u32>::iterator iterNic = std::find(nicRankList_.begin(), nicRankList_.end(), rank);
351 0 : if (iterNic != nicRankList_.end()) {
352 0 : u32 nicIdx = distance(nicRankList_.begin(), iterNic);
353 0 : u32 rxSliceIndex = nicIdx * chunkSize;
354 0 : if (outputSlices[rxSliceIndex].size > 0) {
355 0 : ret = RunDestRducer(linkLeft_, inputSlices[rxSliceIndex], outputSlices[rxSliceIndex]);
356 0 : CHK_PRT_RET(
357 : ret != HCCL_SUCCESS,
358 : HCCL_ERROR(
359 : "[Run][ReduceScatterChunk]rank[%u] rxSlice[%u] Reducer dst run "
360 : "failed",
361 : rank, rxSliceIndex),
362 : ret);
363 : }
364 0 : for (u32 chunkIdx = 1; chunkIdx < chunkSize; chunkIdx++) {
365 0 : rxSliceIndex = nicIdx * chunkSize + chunkIdx;
366 0 : if (outputSlices[rxSliceIndex].size > 0) {
367 0 : CHK_RET(linkLeft_->TxAck(stream_));
368 :
369 0 : ret = RunDestRducer(linkLeft_, inputSlices[rxSliceIndex], outputSlices[rxSliceIndex]);
370 0 : CHK_PRT_RET(
371 : ret != HCCL_SUCCESS,
372 : HCCL_ERROR(
373 : "[Run][ReduceScatterChunk]rank[%u] rxSlice[%u] Reducer dst "
374 : "run failed",
375 : rank, rxSliceIndex),
376 : ret);
377 : }
378 : }
379 : }
380 0 : return HCCL_SUCCESS;
381 : }
382 :
383 0 : HcclResult ReduceScatterRing::HeadReduceScatterChunk(
384 : u32 rank, u32 rankSize, const std::vector<Slice>& inputSlices, const std::vector<Slice>& outputSlices)
385 : {
386 0 : bool bRetSize = (inputSlices.size() < rankSize);
387 0 : CHK_PRT_RET(
388 : bRetSize,
389 : HCCL_ERROR(
390 : "[ReduceScatterRing][HeadReduceScatterChunk]rank[%u] inputslice size[%llu] is "
391 : "less than rank size[%u]",
392 : rank, outputSlices.size(), rankSize),
393 : HCCL_E_INTERNAL);
394 :
395 0 : bRetSize = (outputSlices.size() < rankSize);
396 0 : CHK_PRT_RET(
397 : bRetSize,
398 : HCCL_ERROR(
399 : "[ReduceScatterRing][HeadReduceScatterChunk]rank[%u] outputslice size[%llu] is "
400 : "less than rank size[%u]",
401 : rank, outputSlices.size(), rankSize),
402 : HCCL_E_INTERNAL);
403 :
404 0 : u32 rxSliceIndex = rankSliceLists_[rank][0];
405 0 : u32 txSliceIndex = rxSliceIndex;
406 0 : std::vector<u32> preRankSlices(rankSliceLists_[(rank - 1 + rankSize) % rankSize]);
407 0 : std::vector<u32>::iterator iterSlice = std::find(preRankSlices.begin(), preRankSlices.end(), rxSliceIndex);
408 0 : if (iterSlice != preRankSlices.end()) {
409 0 : if (outputSlices[rxSliceIndex].size > 0) {
410 0 : CHK_RET(linkLeft_->TxAck(stream_));
411 :
412 0 : HcclResult ret = RunDestRducer(linkLeft_, inputSlices[rxSliceIndex], outputSlices[rxSliceIndex]);
413 0 : CHK_PRT_RET(
414 : ret != HCCL_SUCCESS,
415 : HCCL_ERROR(
416 : "[ReduceScatterRing][HeadReduceScatterChunk]rank[%u] rxSlice[%u] "
417 : "Reducer dst run failed",
418 : rank, rxSliceIndex),
419 : ret);
420 : }
421 : }
422 0 : iterSlice = std::find(preRankSlices.begin(), preRankSlices.end(), rankSliceLists_[rank][1]);
423 0 : if (iterSlice != preRankSlices.end()) {
424 0 : CHK_RET(MidReduceScatterChunk(rank, rankSize, 0, inputSlices, outputSlices));
425 : } else {
426 0 : if (inputSlices[txSliceIndex].size > 0) {
427 0 : CHK_RET(linkRight_->RxAck(stream_));
428 :
429 0 : CHK_RET(RunSourceReducer(linkRight_, inputSlices[txSliceIndex], outputSlices[txSliceIndex]));
430 : }
431 : }
432 0 : return HCCL_SUCCESS;
433 0 : }
434 :
435 0 : HcclResult ReduceScatterRing::MidReduceScatterChunk(
436 : u32 rank, u32 rankSize, u32 sliceIdx, const std::vector<Slice>& inputSlices, const std::vector<Slice>& outputSlices)
437 : {
438 0 : bool bRetSize = (inputSlices.size() < rankSize);
439 0 : CHK_PRT_RET(
440 : bRetSize,
441 : HCCL_ERROR(
442 : "[ReduceScatterRing][MidReduceScatterChunk]rank[%u] inputslice size[%llu] is "
443 : "less than rank size[%u]",
444 : rank, outputSlices.size(), rankSize),
445 : HCCL_E_INTERNAL);
446 :
447 0 : bRetSize = (outputSlices.size() < rankSize);
448 0 : CHK_PRT_RET(
449 : bRetSize,
450 : HCCL_ERROR(
451 : "[ReduceScatterRing][MidReduceScatterChunk]rank[%u] outputslice size[%llu] is "
452 : "less than rank size[%u]",
453 : rank, outputSlices.size(), rankSize),
454 : HCCL_E_INTERNAL);
455 :
456 : HcclResult ret;
457 0 : u32 rxSliceIndex = rankSliceLists_[rank][sliceIdx + 1];
458 0 : u32 txSliceIndex = rankSliceLists_[rank][sliceIdx];
459 0 : std::vector<u32> preRankSlices(rankSliceLists_[(rank - 1 + rankSize) % rankSize]);
460 0 : std::vector<u32>::iterator iterSlice = std::find(preRankSlices.begin(), preRankSlices.end(), rxSliceIndex);
461 0 : if (iterSlice != preRankSlices.end()) {
462 0 : if (outputSlices[rxSliceIndex].size > 0) {
463 0 : CHK_RET(linkLeft_->TxAck(stream_));
464 : }
465 0 : if (inputSlices[txSliceIndex].size > 0) {
466 0 : CHK_RET(linkRight_->RxAck(stream_));
467 :
468 0 : CHK_RET(RunSourceReducer(linkRight_, inputSlices[txSliceIndex], outputSlices[txSliceIndex]));
469 : }
470 0 : if (outputSlices[rxSliceIndex].size > 0) {
471 0 : ret = RunDestRducer(linkLeft_, inputSlices[rxSliceIndex], outputSlices[rxSliceIndex]);
472 0 : CHK_PRT_RET(
473 : ret != HCCL_SUCCESS,
474 : HCCL_ERROR(
475 : "[ReduceScatterRing][MidReduceScatterChunk]rank[%u] rxSlice[%u] "
476 : "Reducer dst run failed",
477 : rank, rxSliceIndex),
478 : ret);
479 : }
480 : } else {
481 0 : if (inputSlices[txSliceIndex].size > 0) {
482 0 : CHK_RET(linkRight_->RxAck(stream_));
483 :
484 0 : CHK_RET(RunSourceReducer(linkRight_, inputSlices[txSliceIndex], outputSlices[txSliceIndex]));
485 : }
486 : }
487 0 : return HCCL_SUCCESS;
488 0 : }
489 :
490 0 : HcclResult ReduceScatterRing::TailReduceScatterChunk(
491 : u32 rank, u32 rankSize, u32 sliceIdx, const std::vector<Slice>& inputSlices, const std::vector<Slice>& outputSlices)
492 : {
493 0 : bool bRetSize = (inputSlices.size() < rankSize);
494 0 : CHK_PRT_RET(
495 : bRetSize,
496 : HCCL_ERROR(
497 : "[ReduceScatterRing][TailReduceScatterChunk]rank[%u] inputslice size[%llu] is "
498 : "less than rank size[%u]",
499 : rank, outputSlices.size(), rankSize),
500 : HCCL_E_INTERNAL);
501 :
502 0 : bRetSize = (outputSlices.size() < rankSize);
503 0 : CHK_PRT_RET(
504 : bRetSize,
505 : HCCL_ERROR(
506 : "[ReduceScatterRing][TailReduceScatterChunk]rank[%u] outputslice size[%llu] is "
507 : "less than rank size[%u]",
508 : rank, outputSlices.size(), rankSize),
509 : HCCL_E_INTERNAL);
510 :
511 0 : std::vector<u32> preRankSlices(rankSliceLists_[(rank - 1 + rankSize) % rankSize]);
512 0 : std::vector<u32>::iterator iterNic = std::find(nicRankList_.begin(), nicRankList_.end(), rank);
513 0 : if (iterNic != nicRankList_.end()) {
514 0 : u32 nicIdx = distance(nicRankList_.begin(), iterNic);
515 0 : u32 rxSliceIndex = HCCL_NIC_MAX_NUM / nicRankList_.size() * nicIdx;
516 0 : if (outputSlices[rxSliceIndex].size > 0) {
517 0 : CHK_RET(linkLeft_->TxAck(stream_));
518 : }
519 : }
520 0 : if (rankSliceLists_[rank].size() > 0) {
521 0 : u32 txSliceIndex = rankSliceLists_[rank][sliceIdx];
522 0 : if (inputSlices[txSliceIndex].size > 0) {
523 0 : CHK_RET(linkRight_->RxAck(stream_));
524 :
525 0 : CHK_RET(RunSourceReducer(linkRight_, inputSlices[txSliceIndex], outputSlices[txSliceIndex]));
526 : }
527 : }
528 0 : return HCCL_SUCCESS;
529 0 : }
530 :
531 0 : HcclResult ReduceScatterRing::ReduceScatterSlicesPrep(u32 rankSize, u32 nicSize)
532 : {
533 0 : u32 chunkSize = HCCL_NIC_MAX_NUM / nicSize;
534 0 : for (u32 rankIdx = 0; rankIdx < rankSize; rankIdx++) {
535 0 : std::vector<u32> sliceList;
536 0 : for (u32 nicDis = 1; nicDis < rankSize; nicDis++) {
537 0 : std::vector<u32>::iterator iterNic;
538 0 : u32 nicIdx = (rankIdx + rankSize - nicDis) % rankSize;
539 0 : iterNic = std::find(nicRankList_.begin(), nicRankList_.end(), nicIdx);
540 0 : if (iterNic != nicRankList_.end()) {
541 0 : u32 nicListIdx = distance(nicRankList_.begin(), iterNic);
542 0 : for (u32 chunkIdx = 0; chunkIdx < chunkSize; chunkIdx++) {
543 0 : sliceList.push_back(chunkSize * nicListIdx + chunkIdx);
544 : }
545 : }
546 : }
547 0 : rankSliceLists_.push_back(sliceList);
548 0 : }
549 0 : HCCL_DEBUG("[ReduceScatterRing]ReduceScatterSlicesPrep success");
550 0 : return HCCL_SUCCESS;
551 : }
552 :
553 0 : HcclResult ReduceScatterRing::GetNslbAdjInfo(
554 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links, AdjInfo& nslbAdjInfo)
555 : {
556 0 : u32 ringNextRank = (rank + 1) % rankSize;
557 0 : LINK nslbNext = links[ringNextRank];
558 0 : if (nslbAdjInfo.nsAdjInfo.size() > 0) {
559 0 : nslbAdjInfo.nsAdjInfo[0].dstLocalRankId = (rank + 1) % rankSize;
560 : } else {
561 0 : NslbDpAdjInfo adjInfoStep = {};
562 0 : nslbAdjInfo.dstRankNum = 1;
563 0 : adjInfoStep.dstLocalRankId = nslbNext->GetRemoteRank();
564 0 : adjInfoStep.phaseId = 1;
565 0 : adjInfoStep.rev = 0;
566 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
567 : }
568 0 : return HCCL_SUCCESS;
569 0 : }
570 :
571 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_REDUCESCATTER_RING, ReduceScatterRing);
572 : } // namespace hccl
|
