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 "aicpu_allreduce.h"
12 : #include <cmath>
13 : #include <algorithm>
14 : #include "common/aicpu_hccl_common.h"
15 :
16 41 : HcclResult AicpuAllreduce::RunAlgorithm(HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataCount,
17 : HcclDataType dataType, u64 strideLen, AivAicpuOpParam * /* nextTask */)
18 : {
19 41 : CHK_PTR_NULL(ctx_);
20 41 : if (CC_EXE_ONE_SHOT_8_STREAM == ctx_->commOpType) {
21 34 : return RunAllReduceReduceBcast(opType, sendBuffer, recvBuffer, dataCount * ctx_->unitSize, dataType);
22 7 : } else if (ctx_->commOpType == CC_EXE_ONE_SHOT_1_STREAM) {
23 1 : return RunAllReduceOneShot1Stream(opType, sendBuffer, recvBuffer, dataCount * ctx_->unitSize, dataType);
24 6 : } else if (ctx_->commOpType == CC_EXE_TWO_SHOT_1_STREAM) {
25 1 : return RunAllReduceTwoShot1Stream(opType, sendBuffer, recvBuffer, dataCount, dataType);
26 5 : } else if (ctx_->commOpType == CC_EXE_ONE_SHOT_HD) {
27 1 : return RunAllReduceOneshotHD(opType, sendBuffer, recvBuffer, dataCount * ctx_->unitSize, dataType);
28 4 : } else if (ctx_->commOpType == CC_EXE_ONE_SHOT_SINGLE_RING) {
29 2 : return RunAllReduceRing(opType, sendBuffer, recvBuffer, dataCount, dataType);
30 : }
31 :
32 2 : if (ctx_->useBufferType == MC2_BUFFER_TYPE_WINDOW_IN) {
33 1 : if (HCCL_SUCCESS == RunAllReduceAlignWin2Win(opType, recvBuffer, dataCount, dataType)) {
34 0 : return HCCL_SUCCESS;
35 : }
36 : } else {
37 1 : if (HCCL_SUCCESS == RunAllReduceAlign(opType, sendBuffer, recvBuffer, dataCount, dataType)) {
38 0 : return HCCL_SUCCESS;
39 : }
40 : }
41 :
42 2 : return RunAllReduce(opType, sendBuffer, recvBuffer, dataCount, dataType);
43 : }
44 :
45 1 : int64_t AicpuAllreduce::RoundUpWithDivisor(u64 value, u64 divisor) const
46 : {
47 1 : if ((value == 0) || (divisor == 0)) {
48 0 : return divisor;
49 : }
50 : // divisor必须大于等于1, 返回value向上取divisor的整数倍的值
51 1 : return ((value + (divisor - 1)) / divisor) * divisor;
52 : }
53 :
54 1 : HcclResult AicpuAllreduce::PrepareSlice(u64 dataCount, HcclDataType dataType, u32 sliceNum,
55 : std::vector<Slice> &dataSlice) const
56 : {
57 1 : Slice temp;
58 1 : u32 unitSize = DataUnitSize(dataType);
59 1 : u64 totalSize = dataCount * unitSize;
60 1 : dataSlice.clear();
61 1 : dataSlice.reserve(sliceNum);
62 1 : if (sliceNum == 0) {
63 0 : HCCL_ERROR("[Prepare][SliceData]data slice prepare, sliceNum is 0.");
64 0 : return HCCL_E_PARA;
65 : }
66 1 : u64 sizePerSlice = (totalSize + sliceNum - 1) / sliceNum; /* 1是为了向上取整 */
67 1 : sizePerSlice = RoundUpWithDivisor(sizePerSlice, HCCL_MIN_SLICE_ALIGN);
68 1 : u64 residueSize = totalSize;
69 1 : u32 i = 0;
70 3 : while (residueSize > 0) {
71 2 : u64 sliceSize = sizePerSlice < residueSize ? sizePerSlice : residueSize;
72 2 : temp.size = sliceSize;
73 2 : temp.offset = totalSize - residueSize;
74 2 : i++;
75 2 : if (sliceSize <= 0) {
76 0 : HCCL_ERROR("[Prepare][SliceData]data_slice_prepare sliceSize[%llu]", sliceSize);
77 0 : return HCCL_E_PARA;
78 : }
79 2 : residueSize -= sliceSize;
80 2 : dataSlice.push_back(temp);
81 : }
82 1 : while (i < sliceNum) {
83 0 : temp.size = 0;
84 0 : temp.offset = totalSize;
85 0 : i++;
86 0 : dataSlice.push_back(temp);
87 : }
88 1 : return HCCL_SUCCESS;
89 : }
90 :
91 3 : void AicpuAllreduce::GetDataSizes16K(std::vector<u64> &dataSizes, u64 allDataSize) const
92 : {
93 3 : u64 num16k = allDataSize / HCCL_COPY_ALIGN;
94 3 : u64 tailSize = allDataSize % HCCL_COPY_ALIGN;
95 :
96 3 : u64 baseNum = num16k / ctx_->rankNum;
97 3 : u64 tailNum = num16k % ctx_->rankNum;
98 :
99 7 : for (u32 i = 0; i < ctx_->rankNum; i++) {
100 4 : dataSizes[i] = baseNum * HCCL_COPY_ALIGN;
101 : }
102 3 : for (u32 i = 0; i < tailNum; i++) {
103 0 : dataSizes[i] += HCCL_COPY_ALIGN;
104 : }
105 3 : dataSizes[ctx_->rankNum - 1] += tailSize;
106 3 : }
107 :
108 1 : HcclResult AicpuAllreduce::RunAllReduceAlignWin2Win(HcclReduceOp opType, void *recvBuffer, u64 dataCount,
109 : HcclDataType dataType) const
110 : {
111 1 : u32 unitSize = ctx_->unitSize;
112 1 : u64 allDataSize = dataCount * unitSize;
113 1 : u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
114 :
115 1 : if (ctx_->rankNum == 0) {
116 0 : return HCCL_E_PARA;
117 : }
118 1 : std::vector<u64> dataSizes(ctx_->rankNum, 0);
119 1 : GetDataSizes16K(dataSizes, allDataSize);
120 :
121 2 : if (dataSizes[0] > ctx_->windowSize || dataSizes[ctx_->rankNum - 1] > ctx_->windowSize ||
122 1 : (ctx_->rankNum - 1) * HCCL_COPY_ALIGN >= allDataSize) {
123 1 : return HCCL_E_PARA;
124 : }
125 :
126 0 : std::vector<u64> dataOffsets(ctx_->rankNum, 0);
127 0 : for (u32 i = 1; i < ctx_->rankNum; i++) {
128 0 : dataOffsets[i] = dataOffsets[i - 1] + dataSizes[i - 1];
129 : }
130 :
131 0 : u64 winOffset = ctx_->winOffset;
132 :
133 : // 1. 前同步
134 0 : TaskOrchestrator::DoPreSync();
135 :
136 : // 2. 跨片SDMA，分批拷贝 + 分批结束同步
137 0 : TaskOrchestrator::IpcCpyWin2Win(dataSizes, winOffset, dataOffsets, opType, dataType);
138 :
139 : // 3. 后同步
140 0 : TaskOrchestrator::DoPostSync();
141 :
142 : // 4. 前同步
143 0 : TaskOrchestrator::DoPreSync();
144 :
145 : // 5. 片内数据 Win拷贝到Rcv
146 0 : TaskOrchestrator::SelfCpyWin2RcvEx1(curOutputPtr, dataSizes[ctx_->rankId], dataOffsets[ctx_->rankId], winOffset,
147 : HCCL_REDUCE_RESERVED, dataType);
148 :
149 : // 6. 跨片SDMA，分批拷贝 + 分批结束同步
150 0 : TaskOrchestrator::IpcCpyWin2RcvEx(curOutputPtr, dataSizes, dataOffsets, winOffset, HCCL_REDUCE_RESERVED, dataType);
151 :
152 : // 7. 后同步
153 0 : TaskOrchestrator::DoPostSync();
154 :
155 0 : TaskOrchestrator::LaunchTasks();
156 :
157 0 : return HCCL_SUCCESS;
158 1 : }
159 :
160 1 : HcclResult AicpuAllreduce::RunAllReduceAlign(HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataCount,
161 : HcclDataType dataType) const
162 : {
163 1 : u32 unitSize = ctx_->unitSize;
164 1 : u64 allDataSize = dataCount * unitSize;
165 :
166 1 : u8 *curInputPtr = static_cast<u8 *>(sendBuffer);
167 1 : u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
168 :
169 1 : if (ctx_->rankNum == 0) {
170 0 : return HCCL_E_PARA;
171 : }
172 1 : std::vector<u64> dataSizes(ctx_->rankNum, 0);
173 1 : GetDataSizes16K(dataSizes, allDataSize);
174 :
175 2 : if (dataSizes[0] > ctx_->windowSize || dataSizes[ctx_->rankNum - 1] > ctx_->windowSize ||
176 1 : (ctx_->rankNum - 1) * HCCL_COPY_ALIGN >= allDataSize) {
177 1 : return HCCL_E_PARA;
178 : }
179 :
180 0 : std::vector<u64> dataOffsets(ctx_->rankNum, 0);
181 0 : for (u32 i = 1; i < ctx_->rankNum; i++) {
182 0 : dataOffsets[i] = dataOffsets[i - 1] + dataSizes[i - 1];
183 : }
184 :
185 : // 1. 片内数据 Snd拷贝到Window
186 0 : TaskOrchestrator::SelfCpySnd2Win(curInputPtr, dataSizes[ctx_->rankId], dataOffsets[ctx_->rankId], 0,
187 : HCCL_REDUCE_RESERVED, dataType);
188 :
189 : // 2. 前同步
190 0 : TaskOrchestrator::DoPreSync();
191 :
192 : // 3. 跨片SDMA，分批拷贝 + 分批结束同步
193 0 : TaskOrchestrator::IpcCpySnd2Win(curInputPtr, dataSizes, dataOffsets, nullptr, opType, dataType);
194 :
195 : // 4. 后同步
196 0 : TaskOrchestrator::DoPostSync();
197 :
198 : // 5. 前同步
199 0 : TaskOrchestrator::DoPreSync();
200 :
201 : // 6. 片内数据 Win拷贝到Rcv
202 0 : TaskOrchestrator::SelfCpyWin2Rcv(curOutputPtr, dataSizes[ctx_->rankId], 0, dataOffsets[ctx_->rankId],
203 : HCCL_REDUCE_RESERVED, dataType);
204 :
205 : // 7. 跨片SDMA，分批拷贝 + 分批结束同步
206 0 : TaskOrchestrator::IpcCpyWin2Rcv(curOutputPtr, dataSizes, nullptr, dataOffsets, HCCL_REDUCE_RESERVED, dataType);
207 :
208 : // 8. 后同步
209 0 : TaskOrchestrator::DoPostSync();
210 :
211 0 : TaskOrchestrator::LaunchTasks();
212 :
213 0 : return HCCL_SUCCESS;
214 1 : }
215 :
216 3 : HcclResult AicpuAllreduce::RunAllReduce(HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataCount,
217 : HcclDataType dataType) const
218 : {
219 3 : u64 windowSize = ctx_->windowSize; // window size default is 200M, maybe need read from cfg/env.
220 3 : u32 unitSize = ctx_->unitSize;
221 3 : u64 maxCountPerLoop = (windowSize / unitSize) * ctx_->rankNum; // 中转内存单次最多能够接受的output count
222 :
223 3 : u8 *curInputPtr = static_cast<u8 *>(sendBuffer);
224 3 : u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
225 3 : u64 inputOffset = 0;
226 3 : u64 outputOffset = 0;
227 3 : u64 countLeft = dataCount;
228 :
229 3 : u64 dataSlice[AC_MAX_RANK_NUM] = {0};
230 3 : u64 sliceSize[AC_MAX_RANK_NUM] = {0};
231 3 : if (ctx_->rankNum <= 0) {
232 1 : return HCCL_E_UNAVAIL;
233 : }
234 :
235 2 : while (countLeft > 0) {
236 0 : curInputPtr += inputOffset;
237 0 : curOutputPtr += outputOffset;
238 0 : u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
239 0 : u64 curSize = curCount * unitSize; // 单位 byte
240 0 : u64 curRankCnt = curCount / ctx_->rankNum;
241 0 : for (u32 i = 0; i < ctx_->rankNum; i++) {
242 0 : dataSlice[i] = i * curRankCnt * unitSize;
243 0 : sliceSize[i] = curRankCnt * unitSize;
244 : }
245 0 : sliceSize[ctx_->rankNum - 1] += (curCount - curRankCnt * ctx_->rankNum) * unitSize;
246 :
247 0 : HCCL_DEBUG("RunAllReducev:curInputPtr[%p], curOutputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr,
248 : curOutputPtr, curCount, curSize);
249 :
250 0 : if (ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN) {
251 0 : RunAllReduceSlice(curOutputPtr, curInputPtr, sliceSize, dataSlice, opType, dataType);
252 : } else {
253 0 : RunAllReduceSliceWin2Win(curOutputPtr, sliceSize, dataSlice, opType, dataType);
254 : }
255 :
256 0 : countLeft -= curCount;
257 0 : inputOffset = curSize;
258 0 : outputOffset = curSize;
259 : }
260 :
261 2 : return HCCL_SUCCESS;
262 : }
263 :
264 1 : void AicpuAllreduce::RunAllReduceSliceWin2Win(u8 *curOutputPtr, u64 *sliceSize, u64 *dataSlice, HcclReduceOp opType,
265 : HcclDataType dataType) const
266 : {
267 1 : u64 winOffset = ctx_->winOffset;
268 : // 1. 前同步
269 1 : TaskOrchestrator::DoPreSync();
270 :
271 : // 2. 跨片SDMA，分批拷贝 + 分批结束同步
272 1 : TaskOrchestrator::IpcCpyWin2Win(sliceSize, dataSlice, opType, winOffset, dataType);
273 :
274 : // 3. 后同步
275 1 : TaskOrchestrator::DoPostSync();
276 :
277 : // 4. 前同步
278 1 : TaskOrchestrator::DoPreSync();
279 :
280 : // 5. 片内数据 Win拷贝到Rcv
281 1 : TaskOrchestrator::SelfCpyWin2RcvEx1(curOutputPtr, sliceSize[ctx_->rankId], dataSlice[ctx_->rankId], winOffset,
282 : HCCL_REDUCE_RESERVED, dataType);
283 :
284 : // 6. 跨片SDMA，分批拷贝 + 分批结束同步
285 1 : TaskOrchestrator::IpcCpyWin2RcvEx(curOutputPtr, sliceSize, dataSlice, winOffset, HCCL_REDUCE_RESERVED, dataType);
286 :
287 : // 7. 后同步
288 1 : TaskOrchestrator::DoPostSync();
289 :
290 1 : TaskOrchestrator::LaunchTasks();
291 1 : }
292 :
293 1 : void AicpuAllreduce::RunAllReduceSlice(u8 *curOutputPtr, u8 *curInputPtr, u64 *sliceSize, u64 *dataSlice,
294 : HcclReduceOp opType, HcclDataType dataType) const
295 : {
296 : // 1. 片内数据 Snd拷贝到Window
297 1 : TaskOrchestrator::SelfCpySnd2Win(curInputPtr, sliceSize[ctx_->rankId], dataSlice[ctx_->rankId], 0,
298 : HCCL_REDUCE_RESERVED, dataType);
299 :
300 : // 2. 前同步
301 1 : TaskOrchestrator::DoPreSync();
302 :
303 : // 3. 跨片SDMA，分批拷贝 + 分批结束同步
304 1 : TaskOrchestrator::IpcCpySnd2Win(curInputPtr, sliceSize, dataSlice, nullptr, opType, dataType);
305 :
306 : // 4. 后同步
307 1 : TaskOrchestrator::DoPostSync();
308 :
309 : // 5. 前同步
310 1 : TaskOrchestrator::DoPreSync();
311 :
312 : // 6. 片内数据 Win拷贝到Rcv
313 1 : TaskOrchestrator::SelfCpyWin2Rcv(curOutputPtr, sliceSize[ctx_->rankId], 0, dataSlice[ctx_->rankId],
314 : HCCL_REDUCE_RESERVED, dataType);
315 :
316 : // 7. 跨片SDMA，分批拷贝 + 分批结束同步
317 1 : TaskOrchestrator::IpcCpyWin2Rcv(curOutputPtr, sliceSize, nullptr, dataSlice, HCCL_REDUCE_RESERVED, dataType);
318 :
319 : // 8. 后同步
320 1 : TaskOrchestrator::DoPostSync();
321 :
322 1 : TaskOrchestrator::LaunchTasks();
323 1 : }
324 :
325 1 : HcclResult AicpuAllreduce::RunAllReduceOneShot4Stream(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
326 : u64 dataSize, HcclDataType dataType) const
327 : {
328 : // 第一轮第一组
329 1 : u32 mainRankId = ctx_->rankId;
330 1 : u32 maxStreamNum = ctx_->rankNum / 2;
331 1 : u32 startRank = 0;
332 1 : u32 endRank = maxStreamNum - 1;
333 :
334 1 : if (mainRankId >= maxStreamNum) {
335 1 : startRank = maxStreamNum;
336 1 : endRank = ctx_->rankNum - 1;
337 : }
338 :
339 : // 第1轮
340 : // 1. 片内数据 拷贝到Window
341 1 : TaskOrchestrator::SelfCpySnd2WinEx(mainRankId, sendBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType,
342 : maxStreamNum);
343 1 : TaskOrchestrator::MainSubPreSync(mainRankId, startRank, endRank, maxStreamNum);
344 :
345 1 : TaskOrchestrator::IpcPreSyncEx(startRank, endRank, maxStreamNum, false);
346 : // 2. 跨片SDMA 片内Send拷贝到对端Window
347 1 : TaskOrchestrator::IpcCpySnd2WinEx(sendBuffer, dataSize, nullptr, nullptr, opType, dataType, startRank, endRank,
348 : maxStreamNum, false);
349 1 : TaskOrchestrator::IpcPostSyncEx(startRank, endRank, maxStreamNum, false);
350 :
351 1 : TaskOrchestrator::MainSubPostSync(mainRankId, startRank, endRank, maxStreamNum);
352 :
353 : // 第2轮
354 1 : u32 remoteRank = (ctx_->rankNum - 1) - mainRankId; // 0-7; 1-6; 2-5; 3-4
355 : // 3. 片内数据 拷贝到recv
356 1 : TaskOrchestrator::SelfCpyWin2RcvEx(mainRankId, recvBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType,
357 : maxStreamNum);
358 1 : TaskOrchestrator::IpcPreSyncEx(remoteRank, remoteRank, maxStreamNum, true);
359 : // 4. 跨片SDMA Window拷贝到对端Recv
360 1 : TaskOrchestrator::IpcCpyWin2RcvEx(recvBuffer, dataSize, nullptr, nullptr, opType, dataType, remoteRank, remoteRank,
361 : maxStreamNum, true);
362 1 : TaskOrchestrator::IpcPostSyncEx(remoteRank, remoteRank, maxStreamNum, true);
363 :
364 : // 5. 下发sqe
365 1 : TaskOrchestrator::LaunchTasksEx(0, maxStreamNum - 1, maxStreamNum);
366 :
367 1 : return HCCL_SUCCESS;
368 : }
369 :
370 34 : HcclResult AicpuAllreduce::RunReduceBcastOnMainSq(u32 mainRankId, u32 maxStreamNum, u32 /* startRank */,
371 : u32 /* endRank */, void *sendBuffer, void *recvBuffer, u64 dataSize, HcclDataType dataType) const
372 : {
373 34 : HCCL_DEBUG("run RunReduceBcastOnMainSq start");
374 34 : if (ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN) {
375 : // 1. reduce
376 34 : TaskOrchestrator::SelfCpySnd2WinEx(mainRankId, sendBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType,
377 : maxStreamNum);
378 : }
379 :
380 : // 2. 前同步
381 34 : TaskOrchestrator::MainSubPreSync();
382 34 : TaskOrchestrator::IpcPreRecordEx(0, maxStreamNum - 1, maxStreamNum, false);
383 : // 3. 后同步
384 34 : TaskOrchestrator::IpcPostWaitEx(0, maxStreamNum - 1, maxStreamNum, true);
385 :
386 : // 4. 前同步
387 34 : TaskOrchestrator::MainSubPreSync();
388 34 : TaskOrchestrator::IpcPreRecordEx(0, maxStreamNum - 1, maxStreamNum, false);
389 : // 5. bcast
390 34 : TaskOrchestrator::SelfCpyWin2RcvEx(mainRankId, recvBuffer, dataSize,
391 34 : ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN ? 0 : ctx_->winOffset, 0, HCCL_REDUCE_RESERVED, dataType,
392 : maxStreamNum);
393 : // 6. 后同步
394 34 : TaskOrchestrator::IpcPostWaitEx(0, maxStreamNum - 1, maxStreamNum, true);
395 :
396 34 : HCCL_DEBUG("run RunReduceBcastOnMainSq end");
397 34 : return HCCL_SUCCESS;
398 : }
399 :
400 0 : HcclResult AicpuAllreduce::RunReduceBcastOnOtherSq(HcclReduceOp opType, u32 mainRankId, u32 maxStreamNum,
401 : void *sendBuffer, void *recvBuffer, u64 dataSize, HcclDataType dataType) const
402 : {
403 0 : HCCL_DEBUG("run RunReduceBcastOnOtherSq start");
404 :
405 : // reduce
406 0 : TaskOrchestrator::IpcPreWaitEx(mainRankId, mainRankId, maxStreamNum, true);
407 0 : if (ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN) {
408 0 : TaskOrchestrator::SelfCpySnd2WinEx(mainRankId, sendBuffer, dataSize, 0, 0, opType, dataType, maxStreamNum);
409 : } else {
410 0 : TaskOrchestrator::IpcCpyWin2WinEx(mainRankId, dataSize, ctx_->winOffset, opType, dataType, maxStreamNum);
411 : }
412 0 : TaskOrchestrator::IpcPostRecordEx(mainRankId, mainRankId, maxStreamNum, true);
413 :
414 : // bcast
415 0 : TaskOrchestrator::IpcPreWaitEx(mainRankId, mainRankId, maxStreamNum, true);
416 0 : TaskOrchestrator::SelfCpyWin2RcvEx(mainRankId, recvBuffer, dataSize,
417 0 : ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN ? 0 : ctx_->winOffset, 0, HCCL_REDUCE_RESERVED, dataType,
418 : maxStreamNum);
419 0 : TaskOrchestrator::IpcPostRecordEx(mainRankId, mainRankId, maxStreamNum, true);
420 0 : HCCL_DEBUG("run RunReduceBcastOnOtherSq end");
421 0 : return HCCL_SUCCESS;
422 : }
423 :
424 34 : HcclResult AicpuAllreduce::RunAllReduceReduceBcast(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
425 : u64 dataSize, HcclDataType dataType) const
426 : {
427 34 : HCCL_DEBUG("run RunAllReduceReduceBcast start");
428 :
429 34 : u32 mainRankId = 0;
430 34 : u32 maxStreamNum = ctx_->rankNum;
431 34 : u32 startRank = 0;
432 34 : u32 endRank = ctx_->rankNum - 1;
433 :
434 34 : if (ctx_->rankId == mainRankId) {
435 34 : RunReduceBcastOnMainSq(mainRankId, maxStreamNum, startRank, endRank, sendBuffer, recvBuffer, dataSize,
436 : dataType);
437 : } else {
438 0 : RunReduceBcastOnOtherSq(opType, mainRankId, maxStreamNum, sendBuffer, recvBuffer, dataSize, dataType);
439 : }
440 :
441 : // 下发sqe
442 34 : TaskOrchestrator::LaunchTasksEx(0, maxStreamNum - 1, maxStreamNum);
443 :
444 34 : HCCL_DEBUG("run RunAllReduceReduceBcast end");
445 34 : return HCCL_SUCCESS;
446 : }
447 :
448 1 : HcclResult AicpuAllreduce::RunAllReduceOneShot1Stream(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
449 : u64 dataSize, HcclDataType dataType) const
450 : {
451 1 : HCCL_INFO("run RunAllReduceOneShot1Stream start");
452 1 : u32 maxStreamNum = ctx_->rankNum;
453 1 : u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
454 1 : u32 startRank = 0;
455 1 : u32 endRank = maxStreamNum - 1;
456 :
457 1 : TaskOrchestrator::SelfCpySnd2WinEx1(sendBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType, maxStreamNum);
458 :
459 1 : TaskOrchestrator::SelfCpySnd2RcvEx(sendBuffer, recvBuffer, 0, 0, dataSize, HCCL_REDUCE_RESERVED, dataType);
460 :
461 1 : TaskOrchestrator::IpcPreSyncEx(startRank, endRank, maxStreamNum, true);
462 :
463 1 : TaskOrchestrator::IpcCpyWin2RcvEx(curOutputPtr, dataSize, nullptr, nullptr, opType, dataType, startRank, endRank,
464 : maxStreamNum, true);
465 :
466 1 : TaskOrchestrator::IpcPostSyncEx(startRank, endRank, maxStreamNum, true);
467 :
468 : // 下发sqe
469 1 : TaskOrchestrator::LaunchTasksEx(0, maxStreamNum - 1, maxStreamNum);
470 :
471 1 : HCCL_INFO("run RunAllReduceOneShot1Stream end");
472 1 : return HCCL_SUCCESS;
473 : }
474 :
475 2 : HcclResult AicpuAllreduce::RunAllReduceTwoShot1Stream(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
476 : u64 dataCount, HcclDataType dataType) const
477 : {
478 2 : HCCL_INFO("run RunAllReduceTwoShot1Stream start");
479 2 : u32 maxStreamNum = ctx_->rankNum;
480 2 : u32 startRank = 0;
481 2 : u32 endRank = maxStreamNum - 1;
482 2 : u8 *curInputPtr = static_cast<u8 *>(sendBuffer);
483 2 : u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
484 2 : u32 unitSize = ctx_->unitSize;
485 2 : u64 inputOffset = 0;
486 2 : u64 outputOffset = 0;
487 2 : u64 windowSize = ctx_->windowSize; // window size default is 200M, maybe need read from cfg/env.
488 2 : u64 maxCountPerLoop = windowSize / unitSize * ctx_->rankNum; // 中转内存单次最多能够接受的output count
489 2 : u64 countLeft = dataCount;
490 :
491 2 : while (countLeft > 0) {
492 0 : curInputPtr += inputOffset;
493 0 : curOutputPtr += outputOffset;
494 0 : u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
495 0 : u64 curSize = curCount * unitSize; // 单位 byte
496 0 : std::vector<Slice> dataSlice;
497 0 : PrepareSlice(curCount, dataType, maxStreamNum, dataSlice);
498 0 : u64 sliceSize = dataSlice[ctx_->rankId].size;
499 :
500 0 : HCCL_INFO("RunAllReducev:curInputPtr[%p], curOutputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr,
501 : curOutputPtr, curCount, curSize);
502 :
503 0 : TaskOrchestrator::SelfCpySnd2WinEx1(curInputPtr, sliceSize, dataSlice[ctx_->rankId].offset, 0,
504 : HCCL_REDUCE_RESERVED, dataType, maxStreamNum);
505 :
506 0 : TaskOrchestrator::IpcPreSyncEx(startRank, endRank, maxStreamNum, true);
507 :
508 0 : TaskOrchestrator::IpcCpySnd2WinSliceEx(curInputPtr, dataSlice, nullptr, opType, dataType, startRank, endRank,
509 : maxStreamNum, true);
510 :
511 0 : TaskOrchestrator::IpcCpyWin2RcvSliceEx(curOutputPtr, dataSlice, nullptr, HCCL_REDUCE_RESERVED, dataType,
512 : startRank, endRank, maxStreamNum, true);
513 :
514 0 : TaskOrchestrator::IpcPostSyncEx(startRank, endRank, maxStreamNum, true);
515 :
516 0 : TaskOrchestrator::SelfCpyWin2Rcv(curOutputPtr, sliceSize, 0, dataSlice[ctx_->rankId].offset,
517 : HCCL_REDUCE_RESERVED, dataType);
518 :
519 : // 下发sqe
520 0 : TaskOrchestrator::LaunchTasksEx(0, maxStreamNum - 1, maxStreamNum);
521 :
522 0 : countLeft -= curCount;
523 0 : inputOffset = curSize;
524 0 : outputOffset = curSize;
525 0 : }
526 :
527 2 : HCCL_INFO("run RunAllReduceTwoShot1Stream end");
528 2 : return HCCL_SUCCESS;
529 : }
530 :
531 : // 计算HD算法给定轮中给定rank的对端的rank号
532 1 : u32 AicpuAllreduce::GetHdPeer(const u32 hdRound, const u32 curRank) const
533 : {
534 : // 将所有的设备分成若干组，相邻的组之间的对位节点相互通信
535 : // 每个组中的设备数量为 2^当前轮次，即：1，2，4，8 ....
536 1 : u32 groupSize = std::pow(2, hdRound);
537 : // 获取当前节点在组内的位置 （同时也是对端节点在组内的位置）
538 1 : u32 rankOffset = curRank % groupSize;
539 : // 获取当前节点在第几组
540 1 : u32 curGroupIdx = curRank / groupSize;
541 : // 偶数号组内的节点与下一组的节点通信，对应的，奇数号组内的节点与上一组的节点通信
542 1 : u32 peerGroupIdx = (curGroupIdx % 2 == 0) ? curGroupIdx + 1 : curGroupIdx - 1;
543 :
544 1 : u32 peerRank = peerGroupIdx * groupSize + rankOffset;
545 1 : return peerRank;
546 : }
547 :
548 : // OneshotHD 算法, 使用了DMA消减
549 : // 只支持卡数大于 2 且为 2 的幂数的场景
550 : // datasize 需小于 ccl buffer 大小
551 1 : HcclResult AicpuAllreduce::RunAllReduceOneshotHD(
552 : HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataSize, HcclDataType dataType) const
553 : {
554 : /* 分3个阶段：
555 : * 第一阶段：将 input 的数据拷贝到 window 上
556 : * 第二阶段：将 input 数据发送到对端 window 进行 reduce 并将 reduce 完的
557 : * window 拷贝到 output
558 : * 第三阶段：不断将对端 window 数据读取到本端 output，
559 : * 如果不是最后一轮，则将 reduce 完的 output 拷贝到 window
560 : */
561 1 : HCCL_INFO("run RunAllReduceOneshotHD start");
562 :
563 1 : u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
564 1 : const u32 curRank = ctx_->rankId;
565 : // 第一阶段：
566 : // 将输入数据拷贝到 Window
567 1 : CHK_RET(TaskOrchestrator::SelfCpySnd2Win(
568 : sendBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType));
569 : // 第二阶段：
570 : // 片内拷贝 片内Send拷贝到对端Window - 卡内双 die 间 allreduce
571 1 : u32 hdRound = 0;
572 1 : u32 peerRank = GetHdPeer(hdRound, curRank);
573 1 : CHK_RET(TaskOrchestrator::IpcPreSyncEx(peerRank, peerRank, ctx_->rankNum, true));
574 :
575 : // 将输入数据写到对端
576 1 : CHK_RET(TaskOrchestrator::IpcCpySnd2WinP2P(
577 : sendBuffer, peerRank, dataSize, 0, 0, opType, dataType));
578 1 : TaskOrchestrator::IpcPostSyncEx(peerRank, peerRank, ctx_->rankNum, true);
579 :
580 : // 片内拷贝 Win拷贝到Rcv
581 1 : CHK_RET(TaskOrchestrator::SelfCpyWin2Rcv(
582 : curOutputPtr, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType));
583 : // 第三阶段：
584 : // 循环 log2(rankNum) - 1 次
585 1 : u32 remainingHdRounds = ctx_->rankNum >> 1;
586 1 : while (remainingHdRounds >>= 1) { // 使用位移代替 log2
587 0 : hdRound++;
588 0 : peerRank = GetHdPeer(hdRound, curRank);
589 : // 跨片拷贝 对端 window 拷贝到 rcv
590 0 : CHK_RET(TaskOrchestrator::IpcPreSyncEx(peerRank, peerRank, ctx_->rankNum, true));
591 0 : CHK_RET(TaskOrchestrator::IpcCpyWin2RcvP2PMainStream(
592 : curOutputPtr, peerRank, dataSize, 0, 0, opType, dataType));
593 0 : CHK_RET(TaskOrchestrator::IpcPostSyncEx(peerRank, peerRank, ctx_->rankNum, true));
594 : // 如果这不是最后一轮，则需要将rcv里的数据同步到 win 里
595 0 : if (remainingHdRounds > 1) {
596 0 : CHK_RET(TaskOrchestrator::SelfCpyRcv2Win(
597 : curOutputPtr, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType));
598 : }
599 : }
600 : // 下发sqe
601 1 : TaskOrchestrator::LaunchTasksEx(0, ctx_->rankNum - 1, ctx_->rankNum);
602 1 : HCCL_INFO("run RunAllReduceOneshotHD end");
603 1 : return HCCL_SUCCESS;
604 : }
605 :
606 : // 将 oriValue 对调整为 alignValue 的倍数
607 2 : u64 AicpuAllreduce::AlignWith(u64 oriValue, u64 alignValue) const
608 : {
609 2 : if (oriValue <= alignValue || alignValue == 0) {
610 2 : return oriValue;
611 : }
612 0 : u64 remain = oriValue % alignValue;
613 0 : return oriValue - remain;
614 : }
615 :
616 : // 按照 cclBuffer 大小将数据切分
617 2 : HcclResult AicpuAllreduce::GetBurstDataCounts(
618 : u64 windowSize, u64 dataCount, std::vector<u64> &burstDataCounts) const
619 : {
620 2 : u64 alignedWindowSize = AlignWith(windowSize, HCCL_COPY_ALIGN);
621 2 : u32 unitSize = ctx_->unitSize;
622 2 : CHK_PRT_RET(unitSize == 0, HCCL_ERROR("UnitSize is 0"), HCCL_E_UNAVAIL);
623 2 : u32 maxDataPerBurst = alignedWindowSize / unitSize;
624 2 : CHK_PRT_RET(maxDataPerBurst == 0, HCCL_ERROR("maxDataPerBurst is 0"), HCCL_E_UNAVAIL);
625 0 : burstDataCounts.insert(burstDataCounts.end(), dataCount / maxDataPerBurst, maxDataPerBurst);
626 0 : u64 tailSize = dataCount % maxDataPerBurst;
627 0 : if (tailSize != 0) {
628 0 : burstDataCounts.push_back(tailSize);
629 : }
630 0 : return HCCL_SUCCESS;
631 : }
632 :
633 1 : std::vector<std::vector<u32>> AicpuAllreduce::GetRingOrders() const
634 : {
635 : // simple ring
636 1 : std::vector<std::vector<u32>> ringOrders;
637 1 : std::vector<u32> ringOrder(ctx_->rankNum);
638 3 : for (u32 i = 0; i < ctx_->rankNum; i++) {
639 2 : ringOrder[i] = i;
640 : }
641 1 : ringOrders.push_back(ringOrder);
642 1 : return ringOrders;
643 1 : }
644 :
645 : // 当前只支持从0开始的rankID
646 1 : HcclResult AicpuAllreduce::reorderRingSlice(const std::vector<u32> &ringOrder, const std::vector<Slice> &ringSlices,
647 : std::vector<Slice> &orderedRingSlices) const
648 : {
649 1 : size_t ringSize = ringOrder.size();
650 1 : CHK_PRT_RET(ringSize == 0, HCCL_ERROR("ringSize is 0"), HCCL_E_UNAVAIL);
651 1 : orderedRingSlices.resize(ringSize);
652 3 : for (size_t rankIdx = 0; rankIdx < ringSize; rankIdx++) {
653 2 : u32 currentRank = ringOrder[rankIdx];
654 2 : u32 previousRankIdx = (rankIdx + ringSize - 1) % ringSize;
655 2 : u32 previousRank = ringOrder[previousRankIdx];
656 2 : Slice currentSlice = ringSlices[previousRank];
657 2 : orderedRingSlices[currentRank] = currentSlice;
658 : }
659 1 : return HCCL_SUCCESS;
660 : }
661 :
662 1 : HcclResult AicpuAllreduce::PrepareRingSlice(const std::vector<std::vector<u32>> &ringOrders, u64 dataCount,
663 : HcclDataType dataType, std::vector<std::vector<Slice>> &orderedAllRingSlice) const
664 : {
665 1 : u32 ringNum = ringOrders.size();
666 1 : u32 sliceNum = ctx_->rankNum * ringNum;
667 1 : std::vector<Slice> dataSlices;
668 1 : PrepareSlice(dataCount, dataType, sliceNum, dataSlices);
669 1 : std::vector<std::vector<Slice>> allRingSlices(ringNum);
670 1 : orderedAllRingSlice.resize(ringNum);
671 :
672 3 : for (size_t i = 0; i < dataSlices.size(); i++) {
673 2 : allRingSlices[i % ringNum].push_back(dataSlices[i]);
674 : }
675 2 : for (size_t i = 0; i < ringNum; i++) {
676 1 : reorderRingSlice(ringOrders[i], allRingSlices[i],
677 1 : orderedAllRingSlice[i]);
678 : }
679 1 : return HCCL_SUCCESS;
680 1 : }
681 :
682 0 : HcclResult AicpuAllreduce::RingIPCPreSync(const u32 stream, const u32 prevRank, const u32 nextRank) const
683 : {
684 : // 通知下游
685 0 : CHK_RET(AicpuDispatcher::SignalRecord(stream, nextRank, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
686 : // 等待上游通知
687 0 : CHK_RET(AicpuDispatcher::SignalWait(stream, prevRank, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
688 0 : return HCCL_SUCCESS;
689 : }
690 :
691 0 : HcclResult AicpuAllreduce::RingIPCPostSync(const u32 stream, const u32 prevRank, const u32 nextRank) const
692 : {
693 : // 回复上游通知
694 0 : CHK_RET(AicpuDispatcher::SignalRecord(stream, prevRank, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
695 : // 等待下游回复，回收 notify
696 0 : CHK_RET(AicpuDispatcher::SignalWait(stream, nextRank, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
697 0 : return HCCL_SUCCESS;
698 : }
699 :
700 0 : HcclResult AicpuAllreduce::GetPrevRankList(const std::vector<u32> &ringOrder,
701 : std::vector<u32> &previousRankList) const
702 : {
703 0 : size_t ringSize = ringOrder.size();
704 0 : for (size_t rankIdx = 0; rankIdx < ringSize; rankIdx++) {
705 0 : u32 currRank = ringOrder[rankIdx];
706 0 : u32 previousRankIdx = (rankIdx + ringSize - 1) % ringSize;
707 0 : previousRankList[currRank] = ringOrder[previousRankIdx];
708 : }
709 0 : return HCCL_SUCCESS;
710 : }
711 :
712 0 : size_t AicpuAllreduce::FindNextRank(const std::vector<u32> &previousRankList, const u32 localRank) const
713 : {
714 0 : return std::find(previousRankList.begin(), previousRankList.end(),
715 0 : localRank) -
716 0 : previousRankList.begin();
717 : }
718 :
719 0 : Slice* AicpuAllreduce::GetNextRingSlice(const std::vector<u32> &previousRankList,
720 : std::vector<Slice> &orderedRingSlices, u32 &curSliceIdx) const
721 : {
722 0 : curSliceIdx = previousRankList[curSliceIdx];
723 0 : Slice* nextSlice = &orderedRingSlices[curSliceIdx];
724 0 : return nextSlice;
725 : }
726 :
727 0 : HcclResult AicpuAllreduce::RunAllReduceRingAlg(
728 : HcclReduceOp opType, void *sendBuffer, void *recvBuffer, std::vector<Slice> &orderedRingSlices,
729 : std::vector<u32> &ringOrder, HcclDataType dataType) const
730 : {
731 0 : size_t ringSize = ringOrder.size();
732 0 : std::vector<u32> previousRankList(ringSize);
733 : // 计算每一个rank在ring环上的前一个rank
734 0 : CHK_RET(GetPrevRankList(ringOrder, previousRankList));
735 0 : const u32 prevRank = previousRankList[ctx_->rankId];
736 0 : const u32 nextRank = FindNextRank(previousRankList, ctx_->rankId);
737 0 : const u32 subStream = prevRank;
738 0 : u32 curSliceIdx = ctx_->rankId;
739 0 : Slice *localSlice = &orderedRingSlices[curSliceIdx];
740 0 : Slice *remoteSlice = nullptr;
741 : // 第一轮：主流准备前2片数据
742 0 : CHK_RET(TaskOrchestrator::SelfCpySnd2Win(
743 : sendBuffer, localSlice->size, localSlice->offset, localSlice->offset, HCCL_REDUCE_RESERVED, dataType));
744 0 : localSlice = GetNextRingSlice(previousRankList, orderedRingSlices, curSliceIdx);
745 0 : CHK_RET(TaskOrchestrator::SelfCpySnd2Win(
746 : sendBuffer, localSlice->size, localSlice->offset, localSlice->offset, HCCL_REDUCE_RESERVED, dataType));
747 0 : for (size_t rankOffset = 0; rankOffset < ringSize - 1; rankOffset++) {
748 0 : remoteSlice = localSlice;
749 : // 从流开始 reduce 操作
750 0 : CHK_RET(TaskOrchestrator::MainSubPreSync(subStream)); // 主流启动从流
751 : // 从流跨片 reduce
752 0 : CHK_RET(RingIPCPreSync(subStream, prevRank, nextRank));
753 0 : CHK_RET(TaskOrchestrator::IpcCpyWin2WinP2P(prevRank, remoteSlice->size, remoteSlice->offset,
754 : remoteSlice->offset, opType, dataType));
755 0 : CHK_RET(RingIPCPostSync(subStream, prevRank, nextRank));
756 0 : if (rankOffset < ringSize - 2) { // ringSize - 2： 最后一轮不进行本地搬运
757 : // 主流继续准备数据
758 0 : localSlice = GetNextRingSlice(previousRankList, orderedRingSlices, curSliceIdx);
759 0 : CHK_RET(TaskOrchestrator::SelfCpySnd2Win(sendBuffer, localSlice->size,
760 : localSlice->offset, localSlice->offset, HCCL_REDUCE_RESERVED, dataType));
761 : }
762 0 : CHK_RET(TaskOrchestrator::MainSubPostSync(subStream)); // 从流通知主流，回收 notify，主流继续执行
763 0 : TaskOrchestrator::LaunchTasksEx(0, ctx_->rankNum - 1, ctx_->rankNum); // 下发sqe
764 : }
765 0 : for (size_t rankOffset = 0; rankOffset < ringSize - 1; rankOffset++) {
766 0 : localSlice = remoteSlice; // 主流搬运上一轮从流准备的数据
767 0 : remoteSlice = GetNextRingSlice(previousRankList, orderedRingSlices, curSliceIdx); // 从流继续往下循环，搬运reduce好的数据
768 0 : CHK_RET(TaskOrchestrator::MainSubPreSync(subStream)); // 主流通知从流回收notify资源
769 0 : TaskOrchestrator::SelfCpyWin2Rcv(recvBuffer, localSlice->size, localSlice->offset,
770 : localSlice->offset, HCCL_REDUCE_RESERVED, dataType);
771 0 : CHK_RET(RingIPCPreSync(subStream, prevRank, nextRank));
772 0 : if (rankOffset < ringSize - 2) { // < RingSize - 2: 不是最后一轮，搬到 window 上，让下游读
773 0 : CHK_RET(TaskOrchestrator::IpcCpyWin2WinP2P(prevRank, remoteSlice->size, remoteSlice->offset,
774 : remoteSlice->offset, HCCL_REDUCE_RESERVED, dataType));
775 : } else { // 最后一轮，dma 消减，直接从对端读入 rcv
776 0 : CHK_RET(TaskOrchestrator::IpcCpyWin2RcvP2P(recvBuffer, prevRank, remoteSlice->size,
777 : remoteSlice->offset, remoteSlice->offset, HCCL_REDUCE_RESERVED, dataType));
778 : }
779 0 : CHK_RET(RingIPCPostSync(subStream, prevRank, nextRank));
780 0 : CHK_RET(TaskOrchestrator::MainSubPostSync(subStream)); // 通知主流继续
781 0 : TaskOrchestrator::LaunchTasksEx(0, ctx_->rankNum - 1, ctx_->rankNum); // 下发sqe
782 : }
783 0 : return HCCL_SUCCESS;
784 0 : }
785 :
786 0 : HcclResult AicpuAllreduce::RunAllReduceRingSingleBurst(
787 : HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataCount,
788 : HcclDataType dataType, std::vector<std::vector<u32>> &ringOrders) const
789 : {
790 0 : std::vector<std::vector<Slice>> orderedRingSlices;
791 0 : PrepareRingSlice(ringOrders, dataCount, dataType, orderedRingSlices);
792 0 : for (size_t i = 0; i < ringOrders.size(); i++) {
793 0 : CHK_RET(RunAllReduceRingAlg(opType, sendBuffer, recvBuffer, orderedRingSlices[i],
794 : ringOrders[i], dataType));
795 : }
796 0 : return HCCL_SUCCESS;
797 0 : }
798 :
799 : // 单 ring 算法, 使用了DMA消减
800 2 : HcclResult AicpuAllreduce::RunAllReduceRing(HcclReduceOp opType, void *sendBuffer,
801 : void *recvBuffer, u64 dataCount, HcclDataType dataType) const
802 : {
803 2 : HCCL_INFO("run RunAllReduceRing start");
804 : // 数据准备阶段
805 2 : std::vector<u64> burstDataCounts;
806 2 : u64 windowSize = ctx_->windowSize;
807 2 : u32 unitSize = ctx_->unitSize;
808 2 : CHK_RET(GetBurstDataCounts(windowSize, dataCount, burstDataCounts));
809 0 : u8 *currSendBuffer = static_cast<u8 *>(sendBuffer);
810 0 : u8 *currRecvBuffer = static_cast<u8 *>(recvBuffer);
811 : u64 burstSize;
812 0 : std::vector<std::vector<u32>> ringOrders = GetRingOrders();
813 0 : for (u64 burstDataCount : burstDataCounts) {
814 0 : burstSize = burstDataCount * unitSize;
815 0 : CHK_RET(RunAllReduceRingSingleBurst(
816 : opType, currSendBuffer, currRecvBuffer, burstDataCount, dataType, ringOrders));
817 0 : currSendBuffer += burstSize;
818 0 : currRecvBuffer += burstSize;
819 : }
820 0 : HCCL_INFO("run RunAllReduceRing end");
821 0 : return HCCL_SUCCESS;
822 2 : }
|
