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 "coll_all_reduce_executor.h"
12 :
13 : namespace hccl {
14 :
15 35 : CollAllReduceExecutor::CollAllReduceExecutor(const HcclDispatcher dispatcher,
16 35 : std::unique_ptr<TopoMatcher> &topoMatcher)
17 35 : : CollCommExecutor(dispatcher, topoMatcher)
18 : {
19 38 : }
20 :
21 32 : HcclResult CollAllReduceExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
22 : {
23 32 : HcclUs startut = TIME_NOW();
24 32 : tag_ = param.tag;
25 32 : algResResp_ = &algRes;
26 32 : HcclResult ret = HCCL_SUCCESS;
27 32 : bool needLaunchAtTheEnd = !is310P3Common_; // 是否需要在Orchestrate()结束时launch任务
28 :
29 : // 图模式和单卡场景下不需要Loop
30 32 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
31 21 : ExecMem execMem;
32 21 : execMem.count = param.DataDes.count;
33 21 : execMem.inputPtr = param.inputPtr;
34 21 : execMem.outputPtr = param.outputPtr;
35 21 : execMem.inputMem = algRes.paramInputMem;
36 21 : execMem.outputMem = algRes.paramOutputMem;
37 21 : execMem.scratchMem = algRes.scratchMem;
38 21 : ret = KernelRun(param, execMem);
39 32 : } else if (topoAttr_.userRankSize == 1) {
40 0 : ExecMem execMem;
41 0 : execMem.count = param.DataDes.count;
42 0 : execMem.inputPtr = param.inputPtr;
43 0 : execMem.outputPtr = param.outputPtr;
44 0 : execMem.inputMem = algRes.cclInputMem;
45 0 : execMem.outputMem = algRes.cclOutputMem;
46 0 : execMem.scratchMem = algRes.scratchMem;
47 0 : ret = KernelRun(param, execMem);
48 0 : needLaunchAtTheEnd = false;
49 11 : } else if ((param.inputPtr == algRes.cclInputMem.ptr()) && (param.outputPtr == algRes.cclOutputMem.ptr())) {
50 1 : ret = AvoidSubgraphLoop(param, algRes);
51 10 : } else if (desc_.isZeroCopy) {
52 0 : ExecMem execMem;
53 0 : execMem.count = param.DataDes.count;
54 0 : execMem.inputMem = algRes.paramInputMem;
55 0 : execMem.outputMem = algRes.paramOutputMem;
56 0 : execMem.scratchMem = algRes.scratchMem;
57 0 : execMem.inputPtr = param.inputPtr;
58 0 : execMem.outputPtr = param.outputPtr;
59 :
60 0 : ret = KernelRunIntraServerPre(param, execMem);
61 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
62 : HCCL_ERROR("[CollAllReduceExecutor][Orchestrate]errNo[0x%016llx]AllReduce executor level0 failed",
63 : HCCL_ERROR_CODE(ret)), ret);
64 :
65 : // 在Level1和Level2执行RunLoop
66 0 : if (topoAttr_.serverNum > 1) {
67 0 : ret = RunLoop(param, algRes);
68 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
69 : HCCL_ERROR("[CollAllReduceExecutor][Orchestrate]errNo[0x%016llx]AllReduce executor runloop failed. RunLoop",
70 : HCCL_ERROR_CODE(ret)), ret);
71 : } else { // 单机场景，数据直接从UserInput搬到UserOutput
72 0 : std::vector<Slice> level0Datalices;
73 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(param.DataDes.count, SIZE_TABLE[param.DataDes.dataType], topoAttr_.deviceNumPerAggregation, 0, level0Datalices));
74 0 : u32 level0Rank = topoAttr_.userRank % topoAttr_.deviceNumPerAggregation;
75 0 : const Slice &slice = level0Datalices[level0Rank];
76 0 : DeviceMem dstMem = DeviceMem::create(static_cast<u8 *>(algRes.paramOutputMem.ptr()) + slice.offset, slice.size);
77 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8 *>(algRes.paramInputMem.ptr()) + slice.offset, slice.size);
78 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
79 0 : HCCL_DEBUG("[CollAllReduceExecutor][Orchestrate]AllReduce RunLoop success");
80 0 : }
81 :
82 0 : ret = KernelRunIntraServerPost(param, execMem);
83 0 : } else {
84 10 : if (algOpContext_.opRetryHandler.isInplacePreSync == true) {
85 : /*当重执行场景，UserInMem > CCLBuffer时，需要在allreduce算子前增加一个PreSync函数，提升重执行成功概率*/
86 0 : ExecMem execMem;
87 0 : execMem.count = param.DataDes.count;
88 0 : execMem.inputPtr = param.inputPtr;
89 0 : execMem.outputPtr = param.outputPtr;
90 0 : execMem.inputMem = algRes.cclInputMem;
91 0 : execMem.outputMem = algRes.cclOutputMem;
92 0 : execMem.scratchMem = algRes.scratchMem;
93 0 : ret = InplaceOpSync(param, execMem);
94 0 : } else {
95 10 : ret = RunLoop(param, algRes);
96 10 : needLaunchAtTheEnd = false;
97 : }
98 : }
99 32 : CHK_PRT_RET(ret != HCCL_SUCCESS,
100 : HCCL_ERROR("[CollAllReduceExecutor][Orchestrate]errNo[0x%016llx]AllReduce executor kernel run failed",
101 : HCCL_ERROR_CODE(ret)), ret);
102 :
103 : // Enforce task launch at the end of Orchestrate
104 : // 注意: 不要删除这里的强制launch, 否则会导致aicpu cache功能问题
105 31 : if (needLaunchAtTheEnd) {
106 22 : HCCL_INFO("%s: enforce task launch at the end of Orchestrate", __func__);
107 22 : CHK_RET(LaunchTaskExtend(dispatcher_,
108 : const_cast<Stream &>(param.stream),
109 : const_cast<std::vector<Stream> &>(algResResp_->slaveStreams)));
110 : }
111 :
112 31 : HCCL_INFO("[CollAllReduceExecutor]tag[%s], AllReduce executor orchestrate success, take time [%lld]us",
113 : param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
114 31 : return HCCL_SUCCESS;
115 : }
116 :
117 7 : u64 CollAllReduceExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
118 : {
119 : // 中转内存单次最多能够接受的output count
120 7 : u64 maxCountPerLoop = cclBuffSize / unitSize;
121 7 : HCCL_WARNING("[CollAllReduceExecutor][CalcLoopMaxCount]" \
122 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize.", maxCountPerLoop);
123 7 : return maxCountPerLoop;
124 : }
125 :
126 0 : bool CollAllReduceExecutor::IsHugeData(const u64 curSize)
127 : {
128 0 : HCCL_WARNING("[CollAllReduceExecutor][IsHugeData]opMeta is using the default option: not huge data.");
129 0 : return false;
130 : }
131 :
132 0 : bool CollAllReduceExecutor::IsSmallData(const u64 totalSize, const u64 curSize)
133 : {
134 0 : HCCL_WARNING("[CollAllReduceExecutor][IsSmallData]opMeta is using the default option: not small data.");
135 0 : return false;
136 : }
137 :
138 23 : HcclResult CollAllReduceExecutor::GetSliceNum(const u64 totalSize, const bool isSmallData, u64& sliceNum, u32 unitSize)
139 : {
140 23 : u64 actualSize = 0;
141 23 : u32 actualRankSize = 0;
142 :
143 23 : if (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_RESERVED ||
144 21 : (topoAttr_.deviceType == DevType::DEV_TYPE_910_93 && !DMAReduceFlag_)) {
145 : // level0算法配null走单层拓扑场景
146 2 : actualSize = totalSize;
147 2 : actualRankSize = topoAttr_.userRankSize;
148 : } else {
149 : // 非单层拓扑场景
150 21 : const u32 localRankSize = topoAttr_.deviceNumPerAggregation;
151 21 : const u32 localRank = topoAttr_.userRank % localRankSize;
152 21 : const u64 tempPerSlice = (totalSize + localRankSize - 1) / localRankSize;
153 21 : const u64 sizePerSlice =
154 21 : ((tempPerSlice + (HCCL_MIN_SLICE_ALIGN - 1)) / HCCL_MIN_SLICE_ALIGN) * HCCL_MIN_SLICE_ALIGN;
155 :
156 21 : if ((localRank + 1) * sizePerSlice < totalSize) {
157 16 : actualSize = sizePerSlice;
158 5 : } else if (localRank * sizePerSlice < totalSize) {
159 5 : actualSize = totalSize - localRank * sizePerSlice;
160 : }
161 :
162 21 : actualRankSize = topoAttr_.userRankSize / localRankSize;
163 : }
164 :
165 23 : if (topoMatcher_->GetDeterministicConfig() == DETERMINISTIC_STRICT) {
166 0 : u64 sizePerBlock = (totalSize / unitSize + topoAttr_.userRankSize - 1) / topoAttr_.userRankSize * unitSize;
167 0 : sizePerBlock = AlgTemplateBase::RoundUpWithDivisor(sizePerBlock, HCCL_MIN_SLICE_ALIGN);
168 0 : sliceNum = isSmallData ? 1 : std::min((totalSize - 1) / sizePerBlock + 1,
169 0 : static_cast<u64>(topoAttr_.userRankSize));
170 23 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
171 0 : if (totalSize > HCCL_MIN_SLICE_ALIGN) {
172 0 : u64 sliceSize = GetSliceSizeOfNB(actualSize, actualRankSize);
173 0 : CHK_PRT_RET(sliceSize == 0,
174 : HCCL_ERROR("[CollAllReduceExecutor][GetSliceNum]sliceSize is zero."),
175 : HCCL_E_PARA);
176 0 : sliceNum = static_cast<u64>(std::ceil(actualSize * 1.0f / sliceSize));
177 : }
178 23 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
179 2 : u64 sliceSize = (actualSize + (actualRankSize - 1)) / actualRankSize;
180 2 : u64 sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSize, HCCL_MIN_SLICE_ALIGN);
181 2 : sliceNum = isSmallData ? 1 : static_cast<u64>(std::ceil(actualSize * 1.0f / sliceSizeAligned));
182 : }
183 23 : return HCCL_SUCCESS;
184 : }
185 :
186 10 : HcclResult CollAllReduceExecutor::RunLoop(OpParam ¶m, AlgResourceResponse &algRes)
187 : {
188 10 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
189 20 : ReduceType reduceType = ((param.reduceType != HCCL_REDUCE_PROD) &&
190 10 : (param.DataDes.dataType != HCCL_DATA_TYPE_INT64)) ?
191 : ReduceType::INLINE_REDUCE : ReduceType::TBE_REDUCE;
192 :
193 10 : u8 *curInputPtr = static_cast<u8 *>(param.inputPtr);
194 10 : u8 *curOutputPtr = static_cast<u8 *>(param.outputPtr);
195 10 : CHK_PTR_NULL(curInputPtr);
196 10 : CHK_PTR_NULL(curOutputPtr);
197 :
198 10 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize); // override
199 10 : if (maxCountPerLoop == 0) {
200 1 : HCCL_ERROR("[CollAllReduceExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
201 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop);
202 1 : return HCCL_E_PARA;
203 : }
204 9 : HCCL_DEBUG("[CollAllReduceExecutor][RunLoop]tag[%s], maxCountPerLoop is [%lu], userRankSize is [%lu].",
205 : param.tag.c_str(), maxCountPerLoop, topoAttr_.userRankSize);
206 :
207 : u64 totalCount;
208 9 : if (desc_.isZeroCopy) { // 对零拷贝场景而言，只在Server间通信切循环
209 0 : std::vector<Slice> level0Datalices;
210 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(param.DataDes.count, unitSize, topoAttr_.deviceNumPerAggregation, 0, level0Datalices));
211 0 : u32 level0Rank = topoAttr_.userRank % topoAttr_.deviceNumPerAggregation;
212 0 : totalCount = level0Datalices[level0Rank].size / unitSize;
213 0 : } else {
214 9 : totalCount = param.DataDes.count;
215 : }
216 :
217 9 : for (u64 countLeft = totalCount, curCount = 0, inputOffset = 0, outputOffset = 0;
218 30 : countLeft > 0; countLeft -= curCount) {
219 21 : curInputPtr += inputOffset;
220 21 : curOutputPtr += outputOffset;
221 : // 判断剩余数据量对应的output size是否大于中转output size
222 21 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
223 21 : u64 curSize = curCount * unitSize; // 单位：字节
224 :
225 21 : HCCL_DEBUG("[CollAllReduceExecutor][RunLoop]tag[%s], inputOffset[%llu], outputOffset[%llu], " \
226 : "sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%d].",
227 : param.tag.c_str(), inputOffset, outputOffset, curInputPtr, curOutputPtr, curCount, param.DataDes.dataType);
228 :
229 21 : ExecMem execMem;
230 21 : execMem.count = curCount;
231 21 : execMem.inputMem = algRes.cclInputMem;
232 21 : execMem.outputMem = algRes.cclOutputMem;
233 21 : execMem.scratchMem = algRes.scratchMem;
234 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
235 21 : execMem.inputPtr = curInputPtr;
236 21 : execMem.outputPtr = curOutputPtr;
237 :
238 21 : CHK_RET(RunLoopInner(param, reduceType, execMem));
239 :
240 21 : inputOffset = curSize;
241 21 : outputOffset = curSize;
242 21 : }
243 9 : return HCCL_SUCCESS;
244 : }
245 :
246 21 : HcclResult CollAllReduceExecutor::RunLoopInner(OpParam ¶m, const ReduceType &reduceType, ExecMem &execMem)
247 : {
248 21 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
249 21 : u64 curSize = execMem.count * unitSize; // 单位：字节
250 21 : HCCL_DEBUG("[CollAllReduceExecutor][RunLoopInner]inputMem[%p][%llu], outputMem[%p][%llu], " \
251 : "intputPtr[%p], outputPtr[%p], curCount[%llu], curSize[%llu]",
252 : execMem.inputMem.ptr(), execMem.inputMem.size(), execMem.outputMem.ptr(), execMem.outputMem.size(),
253 : execMem.inputPtr, execMem.outputPtr, execMem.count, curSize);
254 21 : CHK_PRT_RET((execMem.count == 0),
255 : HCCL_ERROR("[CollAllReduceExecutor][RunLoop]In OP_BASE curCount is zero."), HCCL_E_PARA);
256 :
257 21 : if (!is310P3Common_) {
258 : /* 设置子图复用标志 */
259 21 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
260 21 : bool hugeData = IsHugeData(curSize); // override
261 :
262 21 : if (reduceType == ReduceType::TBE_REDUCE) {
263 : /* TBE reduce 当总count数超过INT32_MAX时，不使能子图复用 */
264 3 : hugeData = hugeData || param.DataDes.count > INT32_MAX;
265 : }
266 :
267 21 : bool smallData = IsSmallData(param.DataDes.count * unitSize, curSize); // override
268 21 : u64 sliceNum = 0;
269 21 : CHK_RET(GetSliceNum(execMem.count * unitSize, smallData, sliceNum, unitSize));
270 21 : bool dataSplit = false;
271 21 : u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
272 21 : CopyPattern copy = DMAReduceFlag_? CopyPattern::ZCOPY : CopyPattern::BCOPY;
273 21 : auto opMeta = HcclOpMetaInfo::GetOneForAllReduce(autoSelectedAlgTypeLevel1,
274 : param.DataDes.dataType, reduceType, smallData, 1, hugeData, copy, sliceNum,
275 : false, true, dataSplit, deterministic);
276 21 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
277 : }
278 :
279 21 : if (CCLMemSlice_) {
280 16 : execMem.inputMem = DeviceMem::create(execMem.inputMem.ptr(), curSize);
281 16 : execMem.outputMem = DeviceMem::create(execMem.outputMem.ptr(), curSize);
282 : }
283 :
284 : // 执行
285 21 : if (!DMAReduceFlag_) {
286 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
287 0 : DeviceMem inMem(execMem.inputPtr, curSize);
288 0 : DeviceMem inCommMem = execMem.inputMem.range(0, curSize);
289 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inCommMem, inMem, param.stream));
290 0 : HCCL_DEBUG("[CollAllReduceExecutor][RunLoop]copy from user in to ccl in.");
291 0 : }
292 21 : HcclResult ret = HCCL_SUCCESS;
293 21 : if (!desc_.isZeroCopy) {
294 21 : ret = KernelRun(param, execMem);
295 : } else {
296 0 : ret = KernelRunInterServer(param, execMem);
297 : }
298 21 : CHK_PRT_RET(ret != HCCL_SUCCESS,
299 : HCCL_ERROR("[CollAllReduceExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], " \
300 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d], reduce op type[%d]",
301 : HCCL_ERROR_CODE(ret), param.tag.c_str(), execMem.inputMem.ptr(), execMem.outputMem.ptr(),
302 : execMem.count, param.DataDes.dataType, param.reduceType),
303 : ret);
304 :
305 21 : if (!DMAReduceFlag_) {
306 : // 如果使用CCL buffer，需要将CCL buffer out中的结果拷贝到user buffer out
307 0 : DeviceMem outCommMem = execMem.outputMem.range(0, curSize);
308 0 : DeviceMem outMem(execMem.outputPtr, curSize);
309 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outMem, outCommMem, param.stream));
310 0 : }
311 :
312 21 : if (!is310P3Common_) {
313 21 : CHK_RET(LaunchTaskExtend(dispatcher_,
314 : const_cast<Stream &>(param.stream),
315 : const_cast<std::vector<Stream> &>(algResResp_->slaveStreams)));
316 : }
317 21 : return ret;
318 : }
319 :
320 1 : HcclResult CollAllReduceExecutor::AvoidSubgraphLoop(OpParam ¶m, AlgResourceResponse &algRes)
321 : {
322 1 : HCCL_DEBUG("[CollAllReduceExecutor][AvoidSubgraphLoop]start.");
323 :
324 1 : u64 unitSize = SIZE_TABLE[param.DataDes.dataType];
325 2 : ReduceType reduceType = ((param.reduceType != HCCL_REDUCE_PROD) &&
326 1 : (param.DataDes.dataType != HCCL_DATA_TYPE_INT64)) ?
327 : ReduceType::INLINE_REDUCE : ReduceType::TBE_REDUCE;
328 1 : auto originalAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
329 1 : bool hugeData =
330 1 : (param.DataDes.count * unitSize) / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE >
331 1 : RDMA_SEND_MAX_SIZE || (param.DataDes.count * unitSize) > SDMA_SEND_MAX_SIZE;
332 1 : u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
333 : auto opMeta =
334 1 : HcclOpMetaInfo::GetOneForAllReduce(originalAlgTypeLevel1, param.DataDes.dataType, reduceType,
335 1 : param.DataDes.count * unitSize <= HCCL_SMALL_COUNT_128_KB, 1, hugeData, CopyPattern::ZCOPY, 1,
336 : false, true, false, deterministic);
337 1 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
338 :
339 1 : DeviceMem src(param.inputPtr, 0);
340 1 : DeviceMem dst(algRes.cclInputMem.ptr(), 0);
341 1 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, param.stream));
342 : /* 入参的正确性由HCCL确保 */
343 1 : ExecMem execMem;
344 1 : execMem.count = param.DataDes.count;
345 1 : execMem.inputMem = algRes.cclInputMem;
346 1 : execMem.outputMem = algRes.cclOutputMem;
347 1 : execMem.scratchMem = algRes.scratchMem;
348 1 : HcclResult ret = KernelRun(param, execMem);
349 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
350 : HCCL_ERROR("[Loop][Allreduce]errNo[0x%016llx] param.reduceTypebase hcclComm AllReduce error, " \
351 : "tag[%s], input_ptr[%p], output_ptr[%p], count[%llu], data_type[%d], op[%d]",
352 : HCCL_ERROR_CODE(ret), param.tag.c_str(), param.inputPtr, param.outputPtr, param.DataDes.count,
353 : param.DataDes.dataType, param.reduceType), ret);
354 1 : CHK_RET(LaunchTask(dispatcher_, param.stream));
355 1 : return HCCL_SUCCESS;
356 1 : }
357 :
358 16 : bool CollAllReduceExecutor::IsAllReduceSmallData(u64 size)
359 : {
360 16 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
361 0 : if (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_RESERVED ||
362 0 : (topoAttr_.deviceType == DevType::DEV_TYPE_910_93 && !DMAReduceFlag_)) { // level0算法配null走单层拓扑场景
363 0 : if (size <= NHR_ALLREDUCE_SMALL_SIZE) {
364 0 : return true;
365 : }
366 : } else {
367 0 : if (size / topoAttr_.deviceNumPerAggregation <= NHR_ALLREDUCE_SMALL_SIZE) {
368 0 : return true;
369 : }
370 : }
371 : }
372 :
373 16 : return false;
374 : }
375 :
376 0 : HcclResult CollAllReduceExecutor::PrepareSliceDataWithAlignSize(u64 totalSize, u32 sliceNum,
377 : u64 piplineOffset, std::vector<Slice>& dataSlice, u64 alignSize)
378 : {
379 0 : Slice temp;
380 0 : dataSlice.clear();
381 0 : dataSlice.reserve(sliceNum);
382 0 : CHK_PRT_RET((sliceNum == 0), HCCL_ERROR("[Prepare][SliceData]data slice prepare, sliceNum is 0"), HCCL_E_PARA);
383 0 : u64 tempPerSlice = (totalSize + sliceNum - 1) / sliceNum; /* 1是为了向上取整 */
384 0 : u64 sizePerSlice = AlgTemplateBase::RoundUpWithDivisor(tempPerSlice, alignSize);
385 0 : HCCL_DEBUG("total_size:%llu sliceNum:%u temp_per_ring:%llu size_per_ring:%llu.", totalSize, sliceNum, tempPerSlice,
386 : sizePerSlice);
387 0 : u64 residueSize = totalSize;
388 0 : u32 i = 0;
389 0 : while (residueSize > 0) {
390 0 : u64 sliceSize = sizePerSlice < residueSize ? sizePerSlice : residueSize;
391 0 : temp.size = sliceSize;
392 0 : temp.offset = totalSize - residueSize + piplineOffset;
393 0 : i++;
394 0 : CHK_PRT_RET((sliceSize <= 0), HCCL_ERROR("[Prepare][SliceData]data_slice_prepare sliceSize[%llu].", sliceSize),
395 : HCCL_E_PARA);
396 0 : residueSize -= sliceSize;
397 0 : dataSlice.push_back(temp);
398 : }
399 0 : HCCL_DEBUG("[%s] PrepareSliceDataWithAlignSize for data_slice_prepare", __func__);
400 0 : while (i < sliceNum) {
401 0 : temp.size = 0;
402 0 : temp.offset = totalSize + piplineOffset;
403 0 : i++;
404 0 : dataSlice.push_back(temp);
405 : }
406 0 : return HCCL_SUCCESS;
407 : }
408 :
409 0 : HcclResult CollAllReduceExecutor::PrepareAivBuffers(u32 rankSize, u32 rankId, u32 rankOffset,
410 : DeviceMem &inputMem, DeviceMem &outputMem, std::vector<LINK> &links, void **dataBuffers, void **flagBuffers,
411 : UserMemType dataMemType, UserMemType flagMemType, u32 dataMemOffset, u32 flagMemOffset)
412 : {
413 0 : void *tmpCCLBufferData = nullptr;
414 0 : void *tmpCCLBufferFlag = nullptr;
415 0 : for (u32 i = 0; i < rankSize; i++) {
416 0 : if (i != rankId) {
417 0 : if (links[i + rankOffset] != nullptr) {
418 0 : CHK_RET(links[i + rankOffset]->GetRemoteMem(dataMemType, &(tmpCCLBufferData)));
419 0 : CHK_RET(links[i + rankOffset]->GetRemoteMem(flagMemType, &(tmpCCLBufferFlag)));
420 0 : dataBuffers[i] = static_cast<u8 *>(tmpCCLBufferData) + dataMemOffset;
421 0 : flagBuffers[i] = static_cast<u8 *>(tmpCCLBufferFlag) + flagMemOffset;
422 : }
423 : } else {
424 0 : dataBuffers[i] = static_cast<u8 *>(inputMem.ptr()) + dataMemOffset;
425 0 : flagBuffers[i] = static_cast<u8 *>(outputMem.ptr()) + flagMemOffset;
426 : }
427 : }
428 0 : return HCCL_SUCCESS;
429 : }
430 :
431 : } // namespace hccl
|
