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