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