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