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_reduce_scatter_executor.h"
12 : #include <numeric>
13 :
14 : namespace hccl {
15 :
16 48 : CollReduceScatterExecutor::CollReduceScatterExecutor(const HcclDispatcher dispatcher,
17 48 : std::unique_ptr<TopoMatcher> &topoMatcher)
18 48 : : CollCommExecutor(dispatcher, topoMatcher)
19 : {
20 49 : }
21 :
22 14 : HcclResult CollReduceScatterExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
23 : {
24 14 : HcclUs startut = TIME_NOW();
25 14 : ParseParam(param);
26 14 : tag_ = param.tag;
27 14 : algResResp_ = &algRes;
28 14 : const u64 count = param.GetDataCount(topoAttr_.userRank);
29 14 : const HcclDataType dataType = param.GetDataType();
30 14 : HcclResult ret = HCCL_SUCCESS;
31 14 : bool needLaunchAtTheEnd = !is310P3Common_; // 是否需要在Orchestrate()结束时launch任务
32 : // 图模式和单卡场景下不需要Loop
33 14 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
34 12 : ExecMem execMem;
35 12 : execMem.count = count;
36 12 : execMem.scratchMem = algRes.scratchMem;
37 12 : execMem.inputPtr = param.inputPtr;
38 12 : execMem.outputPtr = param.outputPtr;
39 12 : execMem.inputMem = algRes.paramInputMem;
40 12 : execMem.outputMem = algRes.paramOutputMem;
41 12 : ret = KernelRun(param, execMem);
42 12 : if (algOpContext_.opRetryHandler.isPostSync == true) {
43 : // post Sync
44 0 : CHK_RET(RetryPostSync(param, execMem));
45 : }
46 14 : } else if (topoAttr_.userRankSize == 1) {
47 0 : ExecMem execMem;
48 0 : execMem.count = count;
49 0 : execMem.inputPtr = param.inputPtr;
50 0 : execMem.outputPtr = param.outputPtr;
51 0 : execMem.inputMem = algRes.cclInputMem;
52 0 : execMem.outputMem = algRes.cclOutputMem;
53 0 : execMem.scratchMem = algRes.scratchMem;
54 0 : ret = KernelRun(param, execMem);
55 0 : needLaunchAtTheEnd = false;
56 2 : } else if (desc_.isZeroCopy) {
57 : // 在Level0执行KernelRun
58 0 : ExecMem execMem;
59 0 : execMem.count = count;
60 0 : execMem.inputPtr = param.inputPtr;
61 0 : execMem.outputPtr = param.outputPtr;
62 0 : execMem.inputMem = algRes.paramInputMem;
63 0 : execMem.outputMem = algRes.paramOutputMem;
64 0 : execMem.scratchMem = algRes.paramInputMem;
65 0 : ret = KernelRunIntraServerPre(param, execMem);
66 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
67 : HCCL_ERROR("[CollReduceScatterExecutor][Orchestrate]errNo[0x%016llx]ReduceScatter executor KernelRunIntraServerPre failed",
68 : HCCL_ERROR_CODE(ret)), ret);
69 0 : if (algOpContext_.opRetryHandler.isPostSync == true) {
70 : // post Sync
71 0 : CHK_RET(RetryPostSync(param, execMem));
72 : }
73 : // 在Level1和Level2执行RunLoop
74 0 : if (topoAttr_.serverNum > 1) {
75 0 : ret = RunLoop(param, algRes);
76 : } else { // 单机场景，数据直接从UserInput搬到UserOutput
77 0 : u64 totalSize = count * SIZE_TABLE[dataType];
78 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8 *>(algRes.paramInputMem.ptr()) + totalSize * topoAttr_.userRank, totalSize);
79 0 : DeviceMem dstMem = DeviceMem::create(algRes.paramOutputMem.ptr(), totalSize);
80 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
81 0 : }
82 0 : } else {
83 2 : if (algOpContext_.opRetryHandler.isInplacePreSync == true) {
84 : /*当重执行场景，UserInMem > CCLBuffer时，需要在reduce scatter算子前增加一个PreSync函数，提升重执行成功概率*/
85 0 : ExecMem execMem;
86 0 : execMem.count = count;
87 0 : execMem.inputPtr = param.inputPtr;
88 0 : execMem.outputPtr = param.outputPtr;
89 0 : execMem.inputMem = algRes.cclInputMem;
90 0 : execMem.outputMem = algRes.cclOutputMem;
91 0 : execMem.scratchMem = algRes.scratchMem;
92 0 : ret = InplaceOpSync(param, execMem);
93 2 : } else if (isReduceScatterV_) {
94 0 : ret = RunLoopV(param, algRes);
95 0 : needLaunchAtTheEnd = false;
96 : } else {
97 2 : ret = RunLoop(param, algRes);
98 2 : needLaunchAtTheEnd = false;
99 : }
100 : }
101 14 : CHK_PRT_RET(ret != HCCL_SUCCESS,
102 : HCCL_ERROR("[CollReduceScatterExecutor][Orchestrate]errNo[0x%016llx]executor kernel run failed",
103 : HCCL_ERROR_CODE(ret)), ret);
104 :
105 : // Enforce task launch at the end of Orchestrate
106 : // 注意: 不要删除这里的强制launch, 否则会导致aicpu cache功能问题
107 12 : if (needLaunchAtTheEnd) {
108 11 : HCCL_INFO("%s: enforce task launch at the end of Orchestrate", __func__);
109 11 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
110 : }
111 :
112 12 : HCCL_INFO("tag[%s], ReduceScatter executor orchestrate success, take time [%lld]us.",
113 : param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
114 12 : return HCCL_SUCCESS;
115 : }
116 :
117 0 : u64 CollReduceScatterExecutor::CalcLoopMaxCount(const u32 unitSize)
118 : {
119 : // 中转内存单次最多能够接受的output count
120 0 : u64 maxCountPerLoop = inCCLbufferSize_ / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN
121 0 : * HCCL_MIN_SLICE_ALIGN / unitSize;
122 0 : HCCL_INFO("[CollReduceScatterExecutor][CalcLoopMaxCount]using default maxCountPerLoop[%llu] as "
123 : "CCLBuffSize / (userRankSize * unitSize). rsv[%u]", maxCountPerLoop, isReduceScatterV_);
124 0 : return maxCountPerLoop;
125 : }
126 :
127 0 : bool CollReduceScatterExecutor::IsHugeData(const u64 curSize, OpParam *param)
128 : {
129 0 : bool hugeData = (curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE) ||
130 : (curSize > SDMA_SEND_MAX_SIZE);
131 0 : return hugeData;
132 : }
133 :
134 16 : bool CollReduceScatterExecutor::IsSmallData(const u64 totalSize, const u64 curSize)
135 : {
136 16 : HCCL_INFO("[CollReduceScatterExecutor][IsSmallData]opMeta is using the default option: not small data.");
137 16 : return false;
138 : }
139 :
140 2 : HcclResult CollReduceScatterExecutor::RunLoop(OpParam ¶m, AlgResourceResponse &algRes)
141 : {
142 2 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
143 4 : ReduceType reduceType = ((param.reduceType != HCCL_REDUCE_PROD) &&
144 2 : (param.DataDes.dataType != HCCL_DATA_TYPE_INT64)) ?
145 : ReduceType::INLINE_REDUCE : ReduceType::TBE_REDUCE;
146 :
147 2 : u8 *curInputPtr = static_cast<u8 *>(param.inputPtr);
148 2 : u8 *curOutputPtr = static_cast<u8 *>(param.outputPtr);
149 2 : CHK_PTR_NULL(curInputPtr);
150 2 : CHK_PTR_NULL(curOutputPtr);
151 :
152 2 : u64 maxCountPerLoop = CalcLoopMaxCount(unitSize);
153 2 : CHK_PRT_RET(maxCountPerLoop == 0,
154 : HCCL_ERROR("[CollReduceScatterExecutor][RunLoop]maxCountPerLoop is zero."),
155 : HCCL_E_INTERNAL);
156 1 : HCCL_DEBUG("[CollReduceScatterExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
157 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop);
158 : HcclResult ret;
159 1 : for (u64 countLeft = param.DataDes.count, curCount = 0, inputOffset = 0, outputOffset = 0;
160 17 : countLeft > 0; countLeft -= curCount) {
161 16 : curInputPtr += inputOffset;
162 16 : curOutputPtr += outputOffset;
163 : // 判断剩余数据量对应的output size是否大于中转output size
164 16 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
165 16 : u64 curSize = curCount * unitSize; // 单位：字节
166 :
167 16 : HCCL_DEBUG("[CollReduceScatterExecutor][RunLoop]tag[%s], inputOffset[%llu], outputOffset[%llu], " \
168 : "sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%d].",
169 : param.tag.c_str(), inputOffset, outputOffset, curInputPtr, curOutputPtr, curCount, param.DataDes.dataType);
170 :
171 16 : ExecMem execMem;
172 16 : execMem.count = curCount;
173 16 : execMem.inputMem = algRes.cclInputMem;
174 16 : execMem.outputMem = algRes.cclOutputMem;
175 16 : if (scratchMemFlag_) {
176 0 : execMem.scratchMem = algRes.scratchMem;
177 : } else {
178 16 : execMem.scratchMem = algRes.cclOutputMem; // 不需要申请则传入outputmem为scratchmem
179 : }
180 16 : HCCL_DEBUG("[CollReduceScatterExecutor][RunLoop]scratchMem address [%p]", execMem.scratchMem.ptr());
181 :
182 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
183 16 : execMem.inputPtr = curInputPtr;
184 16 : execMem.outputPtr = curOutputPtr;
185 :
186 16 : ret = RunLoopInner(param, reduceType, execMem);
187 16 : CHK_PRT_RET(ret != HCCL_SUCCESS,
188 : HCCL_ERROR("[CollReduceScatterExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s]",
189 : HCCL_ERROR_CODE(ret), param.tag.c_str()), ret);
190 :
191 16 : inputOffset = curSize;
192 16 : outputOffset = curSize;
193 16 : }
194 1 : if (algOpContext_.opRetryHandler.isPostSync == true) {
195 0 : ExecMem execMem;
196 0 : execMem.count = param.DataDes.count;
197 0 : execMem.inputPtr = param.inputPtr;
198 0 : execMem.outputPtr = param.outputPtr;
199 0 : execMem.inputMem = algRes.cclInputMem;
200 0 : execMem.outputMem = algRes.cclOutputMem;
201 0 : execMem.scratchMem = algRes.scratchMem;
202 0 : CHK_RET(RetryPostSync(param, execMem));
203 0 : }
204 1 : return HCCL_SUCCESS;
205 : }
206 :
207 16 : HcclResult CollReduceScatterExecutor::RunLoopInner(OpParam ¶m, const ReduceType &reduceType, ExecMem &execMem)
208 : {
209 16 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
210 16 : u64 curSize = execMem.count * unitSize; // 单位：字节
211 16 : CHK_PRT_RET((execMem.count == 0),
212 : HCCL_ERROR("[CollReduceScatterExecutor][RunLoopInner]In OP_BASE curCount is zero."), HCCL_E_PARA);
213 :
214 : // 不开启dma消减，且通信buffer足够大时，将user in到ccl的拷贝任务合并成一个
215 16 : const bool preloadCopyOpt = IsPreloadCopyOptimizeCondition(param, execMem);
216 :
217 16 : if (!is310P3Common_) {
218 : /* 设置子图复用标志 */
219 16 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
220 16 : bool hugeData = IsHugeData(curSize, ¶m);
221 16 : bool smallData = IsSmallData(param.DataDes.count * unitSize, curSize);
222 16 : bool dataSplit = false;
223 16 : u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
224 16 : auto opMeta = HcclOpMetaInfo::GetOneForReduceScatter(autoSelectedAlgTypeLevel1, param.DataDes.dataType,
225 : reduceType, hugeData, smallData, CopyPattern::BCOPY, dataSplit, deterministic, false, preloadCopyOpt);
226 :
227 16 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
228 : }
229 :
230 16 : if (CCLMemSlice_) {
231 16 : u32 sliceNum = desc_.isZeroCopy ? topoAttr_.serverNum : topoAttr_.userRankSize;
232 16 : execMem.inputMem = execMem.inputMem.range(0, curSize * sliceNum);
233 16 : execMem.outputMem = execMem.outputMem.range(0, curSize);
234 16 : if (scratchMemFlag_) {
235 0 : execMem.scratchMem = execMem.scratchMem.range(0, curSize * topoAttr_.userRankSize);
236 : }
237 : }
238 :
239 : // 执行
240 16 : if (!DMAReduceFlag_) { // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
241 0 : DeviceMem dstMem;
242 0 : DeviceMem srcMem;
243 0 : if (preloadCopyOpt) {
244 : // 中转内存大小足够时，一次性搬完
245 0 : const u64 copySize = param.DataDes.count * unitSize * topoAttr_.userRankSize;
246 0 : dstMem = execMem.inputMem.range(0, copySize);
247 0 : srcMem = DeviceMem::create(static_cast<u8 *>(execMem.inputPtr), copySize);
248 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
249 : } else {
250 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
251 : // 拷贝input上每个slice的数据到中转内存，源端每个slice的size固定为output的size
252 0 : dstMem = execMem.inputMem.range(curSize * i, curSize);
253 0 : srcMem = DeviceMem::create(static_cast<u8 *>(execMem.inputPtr) + param.DataDes.count * unitSize * i,
254 0 : curSize);
255 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
256 : }
257 : }
258 0 : }
259 :
260 16 : HcclResult ret = HCCL_SUCCESS;
261 16 : if (!desc_.isZeroCopy) {
262 16 : ret = KernelRun(param, execMem);
263 : } else {
264 0 : ret = KernelRunInterServer(param, execMem);
265 : }
266 16 : CHK_PRT_RET(ret != HCCL_SUCCESS,
267 : HCCL_ERROR("[CollReduceScatterExecutor][RunLoopInner]errNo[0x%016llx]kernel run error, tag[%s], " \
268 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d], reduce op type[%d], preloadCopyOpt[%d]",
269 : HCCL_ERROR_CODE(ret), param.tag.c_str(), execMem.inputMem.ptr(), execMem.outputMem.ptr(),
270 : execMem.count, param.DataDes.dataType, param.reduceType, preloadCopyOpt),
271 : ret);
272 :
273 16 : if (!DMAReduceFlag_) {
274 : // 如果使用CCL buffer，需要将CCL buffer out中的结果拷贝到user buffer out
275 0 : DeviceMem srcMem = execMem.outputMem.range(0, curSize);
276 0 : DeviceMem dstMem = DeviceMem::create(execMem.outputPtr, curSize);
277 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
278 0 : }
279 16 : HCCL_DEBUG("[CollReduceScatterExecutor][RunLoopInner]inputMem ptr is [%p], outputMem ptr is [%p]",
280 : execMem.inputMem.ptr(), execMem.outputMem.ptr());
281 :
282 16 : if (!is310P3Common_) {
283 16 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
284 : }
285 16 : return ret;
286 : }
287 :
288 0 : HcclResult CollReduceScatterExecutor::RunLoopV(OpParam ¶m, AlgResourceResponse &algRes)
289 : {
290 : // 每轮loop需要重新计算counts和displs
291 0 : const auto *countsPtr = static_cast<const u64*>(param.VDataDes.counts);
292 0 : auto countsLeft = std::vector<u64>(countsPtr, countsPtr + topoAttr_.userRankSize);
293 0 : const auto *displsPtr = static_cast<const u64*>(param.VDataDes.displs);
294 0 : auto displs = std::vector<u64>(displsPtr, displsPtr + topoAttr_.userRankSize);
295 :
296 0 : const HcclDataType dataType = param.VDataDes.dataType;
297 0 : const u32 unitSize = SIZE_TABLE[dataType];
298 :
299 0 : u8 *curInputPtr = static_cast<u8 *>(param.inputPtr);
300 0 : u8 *curOutputPtr = static_cast<u8 *>(param.outputPtr);
301 0 : CHK_PTR_NULL(curInputPtr);
302 :
303 0 : if (UNLIKELY(countsLeft[topoAttr_.userRank] == 0 && curOutputPtr == nullptr)) {
304 : // 若本rank的output count为0，此时允许curOutputPtr传入空指针，为保证后续流程正常执行，赋值为cclout的地址
305 0 : curOutputPtr = static_cast<u8 *>(algRes.cclOutputMem.ptr());
306 0 : HCCL_DEBUG("[CollReduceScatterExecutor][RunLoopV]Since the output count is 0, set curOutputPtr to "
307 : "ccl output[%p]", curOutputPtr);
308 : }
309 0 : CHK_PTR_NULL(curOutputPtr);
310 :
311 0 : ReduceType reduceType = ((param.reduceType != HCCL_REDUCE_PROD) && (dataType != HCCL_DATA_TYPE_INT64)) ?
312 : ReduceType::INLINE_REDUCE : ReduceType::TBE_REDUCE;
313 :
314 : // 计算MaxCountPerLoop
315 0 : const u64 maxCountPerLoop = CalcLoopMaxCount(unitSize);
316 :
317 : HcclResult ret;
318 0 : bool finished = false;
319 0 : while (!finished) {
320 : // 每个块尽可能平分，以均衡利用带宽
321 0 : auto curCounts = std::vector<u64>();
322 0 : auto curDispls = std::vector<u64>();
323 0 : finished = CalcCurCountsAndCurDispls(maxCountPerLoop, countsLeft, displs, curCounts, curDispls, unitSize);
324 : // 打印调测信息
325 0 : PrintCurCountAndCurDispls(curCounts, curDispls);
326 :
327 0 : OpParam curParam = param;
328 0 : curParam.VDataDes.counts = curCounts.data();
329 0 : curParam.VDataDes.displs = curDispls.data();
330 0 : curParam.VDataDes.dataType = dataType;
331 :
332 0 : ExecMem execMem;
333 0 : execMem.count = curCounts[topoAttr_.userRank];
334 0 : execMem.inputPtr = curInputPtr;
335 0 : execMem.outputPtr = curOutputPtr;
336 0 : execMem.inputMem = algRes.cclInputMem;
337 0 : execMem.outputMem = algRes.cclOutputMem;
338 0 : if (scratchMemFlag_) {
339 0 : execMem.scratchMem = algRes.scratchMem;
340 : } else {
341 0 : execMem.scratchMem = algRes.cclOutputMem; // 不需要申请则传入outputmem为scratchmem
342 : }
343 0 : ret = RunLoopInnerV(curParam, reduceType, execMem);
344 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
345 : HCCL_ERROR("[CollReduceScatterExecutor][RunLoopV]errNo[0x%016llx]kernel run error, tag[%s]",
346 : HCCL_ERROR_CODE(ret), curParam.tag.c_str()), ret);
347 :
348 0 : const auto outputSize = curCounts[topoAttr_.userRank] * unitSize;
349 0 : curOutputPtr += outputSize;
350 : // ReduceScatterV curInputPtr不需要偏移，input的偏移由displs计算
351 0 : HCCL_DEBUG("[CollReduceScatterExecutor][RunLoopV]kernel run, finished[%u]", finished);
352 0 : }
353 0 : return HCCL_SUCCESS;
354 0 : }
355 :
356 0 : HcclResult CollReduceScatterExecutor::RunLoopInnerV(OpParam ¶m, const ReduceType &reduceType, ExecMem &execMem)
357 : {
358 0 : const auto *counts = static_cast<u64*>(param.VDataDes.counts);
359 0 : u64 count = counts[topoAttr_.userRank];
360 0 : HcclDataType dataType = param.VDataDes.dataType;
361 0 : u32 unitSize = SIZE_TABLE[dataType];
362 0 : u64 curSize = count * unitSize; // 单位：字节;
363 :
364 : /* 设置子图复用标志 */
365 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
366 0 : bool hugeData = IsHugeData(curSize, ¶m);
367 0 : u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
368 0 : auto opMeta = HcclOpMetaInfo::GetOneForReduceScatterV(autoSelectedAlgTypeLevel1,
369 : dataType, reduceType, hugeData, false, CopyPattern::BCOPY, false, deterministic);
370 :
371 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
372 :
373 0 : if (CCLMemSlice_) {
374 0 : const u64 inputCounts = std::accumulate(counts, counts + topoAttr_.userRankSize, 0ULL);
375 0 : execMem.inputMem = execMem.inputMem.range(0, inputCounts * unitSize);
376 0 : execMem.outputMem = execMem.outputMem.range(0, curSize);
377 0 : if (scratchMemFlag_) {
378 0 : execMem.scratchMem = execMem.scratchMem.range(0, inputCounts * unitSize);
379 : }
380 : }
381 :
382 : // 执行
383 0 : HcclResult ret = KernelRun(param, execMem);
384 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
385 : HCCL_ERROR("[CollReduceScatterExecutor][RunLoopInnerV]errNo[0x%016llx]kernel run error, tag[%s], "
386 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d], reduce op type[%d]", HCCL_ERROR_CODE(ret),
387 : param.tag.c_str(), execMem.inputMem.ptr(), execMem.outputMem.ptr(), execMem.count, dataType, param.reduceType),
388 : ret);
389 :
390 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
391 0 : return ret;
392 : }
393 :
394 0 : bool CollReduceScatterExecutor::CalcCurCountsAndCurDispls(const u64 maxTotalCount, std::vector<u64> &countsLeft,
395 : std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls, u32 unitSize)
396 : {
397 0 : bool finished = false;
398 :
399 0 : curCounts = std::vector<u64>(countsLeft.size(), 0);
400 0 : curDispls = std::vector<u64>(displs.size(), 0);
401 0 : auto allocatableCount = maxTotalCount;
402 :
403 : // 先设置本轮的displacements，等于入参displs
404 0 : std::copy(displs.begin(), displs.end(), curDispls.begin());
405 :
406 : // 分配本轮的counts，如果CCLbuffer空间还没完全利用，则再进行分配
407 0 : while (allocatableCount > 0) {
408 : // 计算现在还有几个rank还有数据需要去通信(countsLeft不为0)
409 : const auto nonZeroCount =
410 0 : std::count_if(countsLeft.begin(), countsLeft.end(), [](const u64 count) { return count != 0; });
411 0 : if (nonZeroCount == 0) {
412 0 : finished = true;
413 0 : break;
414 : }
415 :
416 : // 计算每个rank可以分到多少count
417 0 : auto perRankCount = allocatableCount / nonZeroCount;
418 0 : if (perRankCount == 0) {
419 0 : break;
420 : }
421 :
422 0 : const u64 perRankSize = perRankCount * unitSize;
423 0 : if (perRankSize > HCCL_MIN_SLICE_ALIGN) {
424 0 : perRankCount = perRankSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize; // align for perf
425 0 : } else if ((perRankSize < HCCL_MIN_SLICE_ALIGN) && (allocatableCount != maxTotalCount)) {
426 0 : break;
427 : }
428 :
429 : // 分配好每个rank的counts
430 0 : for (auto i = 0U; i < countsLeft.size(); ++i) {
431 0 : const auto curCount = countsLeft[i] < perRankCount ? countsLeft[i] : perRankCount;
432 0 : allocatableCount -= curCount;
433 0 : curCounts[i] += curCount;
434 0 : countsLeft[i] -= curCount;
435 0 : displs[i] += curCount;
436 : }
437 : }
438 0 : return finished;
439 : }
440 :
441 0 : void CollReduceScatterExecutor::PrintCurCountAndCurDispls(const std::vector<u64> &curCounts,
442 : const std::vector<u64> &curDispls)
443 : {
444 0 : if (HcclCheckLogLevel(DLOG_DEBUG)) {
445 0 : std::ostringstream curLoopInfo;
446 0 : curLoopInfo << "counts[ ";
447 0 : for (auto count : curCounts) {
448 0 : curLoopInfo << count << " ";
449 : }
450 0 : curLoopInfo << "], displs[ ";
451 0 : for (auto displ : curDispls) {
452 0 : curLoopInfo << displ << " ";
453 : }
454 0 : curLoopInfo << "]";
455 0 : HCCL_DEBUG("[CollReduceScatterExecutor][PrintCurCountAndCurDispls] Current loop info: %s",
456 : curLoopInfo.str().c_str());
457 0 : }
458 0 : }
459 :
460 20 : std::vector<std::vector<Slice>> CollReduceScatterExecutor::ReduceScatterRingSlicePrepare(u32 ringNum, u32 sliceNum,
461 : bool useInlineReduce, const DeviceMem& outputMem, std::vector<Slice>& dataSegsSlice, const std::string &tag)
462 : {
463 20 : std::vector<std::vector<Slice>> multiStreamSlice;
464 20 : u64 outputMemSize = outputMem.size();
465 20 : dataSegsSlice.clear();
466 20 : Slice sliceTemp;
467 57 : for (u32 i = 0; i < sliceNum; i++) { // 根据数据量算每个环上数据的偏移和大小
468 37 : sliceTemp.size = outputMemSize;
469 37 : sliceTemp.offset = outputMemSize * i;
470 37 : dataSegsSlice.push_back(sliceTemp);
471 : }
472 20 : bool ARSFlag = topoMatcher_->GetARSFlag();
473 20 : auto nicList = topoAttr_.nicList;
474 20 : if (ARSFlag) {
475 0 : std::vector<u32> mockNicList;
476 0 : for (u32 i = 0; i < sliceNum; i++) {
477 0 : mockNicList.push_back(i);
478 : }
479 0 : nicList = mockNicList;
480 0 : }
481 :
482 : // 再将每个 slice 划分为 ringNum 份
483 20 : if (ringNum == LEVEL0_PLANE_NUM_IN_8PRING) {
484 0 : if (useInlineReduce) {
485 0 : multiStreamSlice = PrepareMultiRingSlice(dataSegsSlice, tag);
486 0 : } else if (outputMem.size() % CCE_REDUCE_ALIGN_SIZE == 0) {
487 0 : multiStreamSlice = PrepareMultiRingSlice(dataSegsSlice, tag);
488 : } else {
489 0 : multiStreamSlice = PrepareMultiRingSlice(dataSegsSlice, tag, true);
490 : }
491 20 : } else if (ringNum == LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE) {
492 : // 双环场景，需要传入正确的 niclist (不涉及网口裁剪)
493 16 : if (useInlineReduce) {
494 16 : multiStreamSlice = PrepareMultiRingSlice(dataSegsSlice, tag, false, nicList);
495 0 : } else if (outputMem.size() % CCE_REDUCE_ALIGN_SIZE == 0) {
496 0 : multiStreamSlice = PrepareMultiRingSlice(dataSegsSlice, tag, false, nicList);
497 : } else {
498 0 : multiStreamSlice = PrepareMultiRingSlice(dataSegsSlice, tag, true, nicList);
499 : }
500 : } else {
501 4 : multiStreamSlice.push_back(dataSegsSlice);
502 : }
503 :
504 20 : return multiStreamSlice;
505 20 : }
506 :
507 0 : HcclResult CollReduceScatterExecutor::PrepareAivBuffers(u32 rankSize, u32 rankId, u32 rankOffset,
508 : DeviceMem &inputMem, DeviceMem &outputMem, std::vector<LINK> &links, void **dataBuffers, void **flagBuffers,
509 : UserMemType dataMemType, UserMemType flagMemType, u32 dataMemOffset, u32 flagMemOffset)
510 : {
511 0 : void *tmpCCLBufferData = nullptr;
512 0 : void *tmpCCLBufferFlag = nullptr;
513 0 : for (u32 i = 0; i < rankSize; i++) {
514 0 : if (i != rankId) {
515 0 : if (links[i + rankOffset] != nullptr) {
516 0 : CHK_RET(links[i + rankOffset]->GetRemoteMem(dataMemType, &(tmpCCLBufferData)));
517 0 : CHK_RET(links[i + rankOffset]->GetRemoteMem(flagMemType, &(tmpCCLBufferFlag)));
518 0 : dataBuffers[i] = static_cast<u8 *>(tmpCCLBufferData) + dataMemOffset;
519 0 : flagBuffers[i] = static_cast<u8 *>(tmpCCLBufferFlag) + flagMemOffset;
520 : }
521 : } else {
522 0 : dataBuffers[i] = static_cast<u8 *>(inputMem.ptr()) + dataMemOffset;
523 0 : flagBuffers[i] = static_cast<u8 *>(outputMem.ptr()) + flagMemOffset;
524 : }
525 : }
526 0 : return HCCL_SUCCESS;
527 : }
528 :
529 0 : std::vector<std::vector<Slice>> CollReduceScatterExecutor::AnyPathReduceScatterRingSlicePrepare(u32 ringNum,
530 : u32 sliceNum, bool useInlineReduce, DeviceMem& outputMem, std::vector<Slice>& dataSegsSlice, const std::string &tag)
531 : {
532 0 : std::vector<std::vector<Slice>> multiStreamSlice;
533 0 : u64 outputMenSize = outputMem.size();
534 0 : dataSegsSlice.clear();
535 0 : Slice sliceTemp;
536 0 : for (u32 i = 0; i < sliceNum; i++) { // 根据数据量算每个环上数据的偏移和大小
537 0 : sliceTemp.size = outputMenSize;
538 0 : sliceTemp.offset = outputMenSize * i;
539 0 : dataSegsSlice.push_back(sliceTemp);
540 : }
541 :
542 : // 再将每个 slice 划分为 ringNum 份
543 0 : if (ringNum == LEVEL0_PLANE_NUM_IN_8PRING) {
544 0 : if (useInlineReduce) {
545 0 : multiStreamSlice = AnyPathPrepareMultiRingSlice(dataSegsSlice, tag);
546 0 : } else if (outputMem.size() % CCE_REDUCE_ALIGN_SIZE == 0) {
547 0 : multiStreamSlice = AnyPathPrepareMultiRingSlice(dataSegsSlice, tag);
548 : } else {
549 0 : multiStreamSlice = AnyPathPrepareMultiRingSlice(dataSegsSlice, tag, true);
550 : }
551 0 : } else if (ringNum == LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE) {
552 : // 双环场景，需要传入正确的 niclist (不涉及网口裁剪)
553 0 : if (useInlineReduce) {
554 0 : multiStreamSlice = AnyPathPrepareMultiRingSlice(dataSegsSlice, tag, false, topoAttr_.nicList);
555 0 : } else if (outputMem.size() % CCE_REDUCE_ALIGN_SIZE == 0) {
556 0 : multiStreamSlice = AnyPathPrepareMultiRingSlice(dataSegsSlice, tag, false, topoAttr_.nicList);
557 : } else {
558 0 : multiStreamSlice = AnyPathPrepareMultiRingSlice(dataSegsSlice, tag, true, topoAttr_.nicList);
559 : }
560 : } else {
561 0 : multiStreamSlice.push_back(dataSegsSlice);
562 : }
563 :
564 0 : return multiStreamSlice;
565 0 : }
566 :
567 0 : HcclResult CollReduceScatterExecutor::RetryPostSync(OpParam& param, ExecMem &execMem)
568 : {
569 0 : if ((algResResp_->slaveStreams).size() == 0) {
570 0 : CHK_RET(PostSyncWithoutSubstream(param, execMem));
571 : } else {
572 0 : PrepareData postSyncPrepareData;
573 0 : postSyncPrepareData.subStreamsPtr = &algResResp_->slaveStreams;
574 0 : postSyncPrepareData.signalPtr = &algResResp_->notifiesMain;
575 0 : postSyncPrepareData.signalAuxPtr = &algResResp_->notifiesAux;
576 0 : postSyncPrepareData.stream = param.stream;
577 0 : CHK_RET(PostSyncWithSubstream(param, execMem, postSyncPrepareData));
578 0 : }
579 0 : return HCCL_SUCCESS;
580 : }
581 :
582 16 : bool CollReduceScatterExecutor::IsPreloadCopyOptimizeCondition(const OpParam ¶m, ExecMem &execMem)
583 : {
584 : // 不开启dma消减，且通信buffer足够大时，将user in到ccl的拷贝任务合并成一个
585 16 : return (!DMAReduceFlag_) && (param.DataDes.count == execMem.count);
586 : }
587 : } // namespace hccl
|
