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_broadcast_executor.h"
12 :
13 : namespace hccl {
14 :
15 2 : CollBroadcastExecutor::CollBroadcastExecutor(const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 2 : : CollCommExecutor(dispatcher, topoMatcher)
17 2 : {}
18 :
19 0 : HcclResult CollBroadcastExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
20 : {
21 0 : HcclResult ret = HCCL_SUCCESS;
22 :
23 : // 由于bcast/allgather/reducescatter/reduce/send/recv暂不支持server间ring，需继续使用HD或NHR
24 0 : if (!(algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR)
25 0 : && !(algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1)
26 0 : && !(algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB)) {
27 0 : algType_.algoLevel1 = AlgTypeLevel1::ALG_LEVEL1_HD;
28 0 : HCCL_WARNING("[BroadCastOperator][Broadcast] do not support ring in AlgoLevel1 yet, reset algType_=HD.");
29 : }
30 :
31 0 : tag_ = param.tag;
32 0 : algResResp_ = &algRes;
33 0 : bool needLaunchAtTheEnd = true; // 是否需要在Orchestrate()结束时launch任务
34 : /* ------------执行算法-------------- */
35 0 : HcclUs startut = TIME_NOW();
36 :
37 : // 图模式和单卡场景下不需要Loop
38 0 : ExecMem execMem;
39 0 : execMem.count = param.DataDes.count;
40 0 : execMem.inputPtr = param.inputPtr;
41 0 : execMem.outputPtr = param.inputPtr;
42 0 : HCCL_INFO(
43 : "Orchestrate UserRank[%u], devicePhyId[%u], inputPtr[%p], outputPtr[%p], root[%u]", topoAttr_.userRank,
44 : topoAttr_.devicePhyId, param.inputPtr, param.outputPtr, param.root);
45 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) { // 图模式直接调KernelRun接口
46 0 : HCCL_DEBUG("[CollBroadcastExecutor][Orchestrate]ops kernel broadcast");
47 0 : execMem.inputMem = algRes.paramInputMem;
48 0 : execMem.outputMem = algRes.paramOutputMem;
49 0 : if (scratchMemFlag_) {
50 0 : execMem.scratchMem = algRes.scratchMem;
51 : }
52 0 : ret = KernelRun(param, execMem);
53 0 : } else if (topoAttr_.userRankSize == 1) { // 单卡
54 0 : HCCL_DEBUG("[CollBroadcastExecutor][Orchestrate]1 rank broadcast");
55 0 : return HCCL_SUCCESS;
56 0 : } else if (desc_.isZeroCopy) {
57 0 : execMem.inputMem = algRes.paramInputMem;
58 0 : execMem.outputMem = algRes.paramOutputMem;
59 0 : ret = KernelRunIntraServerPre(param, execMem);
60 0 : CHK_PRT_RET(
61 : ret != HCCL_SUCCESS,
62 : HCCL_ERROR(
63 : "[CollBroadcastExecutor][Orchestrate]errNo[0x%016llx]Broadcast executor level0 failed",
64 : HCCL_ERROR_CODE(ret)),
65 : ret);
66 :
67 : // 在Level1和Level2执行RunLoop
68 0 : if (topoAttr_.serverNum > 1) {
69 0 : ret = RunLoop(param, algRes);
70 0 : CHK_PRT_RET(
71 : ret != HCCL_SUCCESS,
72 : HCCL_ERROR(
73 : "[CollBroadcastExecutor][Orchestrate]errNo[0x%016llx]Broadcast executor runloop failed. RunLoop",
74 : HCCL_ERROR_CODE(ret)),
75 : ret);
76 : } else { // 单机场景，数据直接从UserInput搬到UserOutput
77 0 : std::vector<Slice> level0Datalices;
78 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(
79 : param.DataDes.count, SIZE_TABLE[param.DataDes.dataType], topoAttr_.deviceNumPerAggregation, 0,
80 : level0Datalices));
81 0 : u32 level0Rank = topoAttr_.userRank % topoAttr_.deviceNumPerAggregation;
82 0 : const Slice& slice = level0Datalices[level0Rank];
83 : DeviceMem dstMem
84 0 : = DeviceMem::create(static_cast<u8*>(algRes.paramOutputMem.ptr()) + slice.offset, slice.size);
85 : DeviceMem srcMem
86 0 : = DeviceMem::create(static_cast<u8*>(algRes.paramInputMem.ptr()) + slice.offset, slice.size);
87 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
88 0 : }
89 :
90 0 : ret = KernelRunIntraServerPost(param, execMem);
91 : } else {
92 0 : ret = RunLoop(param, algRes);
93 0 : needLaunchAtTheEnd = false;
94 : }
95 :
96 0 : CHK_PRT_RET(
97 : ret != HCCL_SUCCESS,
98 : HCCL_ERROR(
99 : "[CollBroadcastExecutor][Orchestrate]errNo[0x%016llx]broadcast executor kernel run failed",
100 : HCCL_ERROR_CODE(ret)),
101 : ret);
102 :
103 : // Enforce task launch at the end of Orchestrate
104 : // 注意: 不要删除这里的强制launch, 否则会导致aicpu cache功能问题
105 0 : if (needLaunchAtTheEnd) {
106 0 : HCCL_INFO("%s: enforce task launch at the end of Orchestrate", __func__);
107 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
108 : }
109 :
110 0 : HCCL_INFO(
111 : "tag[%s], Broadcast executor orchestrate success, take time [%lld]us.", param.tag.c_str(),
112 : DURATION_US(TIME_NOW() - startut));
113 0 : return HCCL_SUCCESS;
114 0 : }
115 :
116 0 : HcclResult CollBroadcastExecutor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
117 : {
118 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
119 :
120 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
121 0 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
122 0 : CHK_PTR_NULL(curInputPtr);
123 0 : CHK_PTR_NULL(curOutputPtr);
124 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize);
125 :
126 0 : HCCL_DEBUG(
127 : "[CollBroadcastExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].", param.tag.c_str(),
128 : topoAttr_.userRankSize, maxCountPerLoop);
129 :
130 : u64 totalCount;
131 0 : if (desc_.isZeroCopy) { // 对零拷贝场景而言，只在Server间通信切循环
132 0 : std::vector<Slice> level0Datalices;
133 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(
134 : param.DataDes.count, unitSize, topoAttr_.deviceNumPerAggregation, 0, level0Datalices));
135 0 : u32 level0Rank = topoAttr_.userRank % topoAttr_.deviceNumPerAggregation;
136 0 : totalCount = level0Datalices[level0Rank].size / unitSize;
137 0 : } else {
138 0 : totalCount = param.DataDes.count;
139 : }
140 :
141 0 : for (u64 countLeft = totalCount, curCount = 0, inputOffset = 0; countLeft > 0; countLeft -= curCount) {
142 0 : curInputPtr += inputOffset;
143 : // 判断剩余数据量对应的output size是否大于中转output size
144 0 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
145 0 : u64 curSize = curCount * unitSize; // 单位：字节
146 :
147 0 : ExecMem execMem;
148 0 : execMem.count = curCount;
149 0 : execMem.inputMem = algRes.cclInputMem;
150 0 : execMem.outputMem = algRes.cclInputMem; // broadcast只用一块CCL buffer
151 : // 使用当前Loop偏移到的地址作为当前的inputPtr
152 0 : execMem.inputPtr = curInputPtr;
153 0 : execMem.outputPtr = curInputPtr;
154 :
155 0 : HCCL_DEBUG(
156 : "[CollBroadcastExecutor] RunLoop tag[%s], inputOffset[%llu], "
157 : "curInputPtr[%p], sendCount[%llu], sendSize[%llu], dataType[%s], realUserRank[%u]",
158 : param.tag.c_str(), inputOffset, curInputPtr, curCount, curSize,
159 : GetDataTypeEnumStr(param.DataDes.dataType).c_str(), topoAttr_.realUserRank);
160 :
161 0 : CHK_RET(RunLoopInner(param, execMem));
162 :
163 0 : inputOffset = curSize;
164 0 : }
165 0 : return HCCL_SUCCESS;
166 : }
167 :
168 0 : HcclResult CollBroadcastExecutor::RunLoopInner(OpParam& param, ExecMem& execMem)
169 : {
170 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
171 0 : u64 totalSize = unitSize * param.DataDes.count;
172 0 : bool isRootRank = param.root == topoAttr_.realUserRank ? true : false;
173 0 : u64 curSize = execMem.count * unitSize; // 单位：字节
174 0 : auto inCCLbufferSize = execMem.inputMem.size();
175 0 : u8* curPtr = static_cast<u8*>(execMem.inputPtr);
176 0 : auto originalAlgTypeLevel0 = algType_.algoLevel0;
177 0 : bool isDMATopoOn91093 = originalAlgTypeLevel0 == AlgTypeLevel0::ALG_LEVEL0_NP_SINGLE_RING
178 0 : || originalAlgTypeLevel0 == AlgTypeLevel0::ALG_LEVEL0_NP_DOUBLE_RING;
179 0 : bool isDMAreduceOn91093 = (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
180 0 : && (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) && isDMATopoOn91093)
181 0 : && DMAReduceFlag_;
182 0 : HCCL_DEBUG(
183 : "[CollBroadcastExecutor][RunLoopInner]inputMem[%p], outputMem[%p]"
184 : "intputPtr[%p], curCount[%llu], curSize[%llu]",
185 : execMem.inputMem.ptr(), execMem.outputMem.ptr(), execMem.inputPtr, execMem.count, curSize);
186 0 : CHK_PRT_RET(
187 : (execMem.count == 0), HCCL_ERROR("[CollBroadcastExecutor][RunLoop]In OP_BASE curCount is zero."), HCCL_E_PARA);
188 :
189 0 : bool hugeData
190 0 : = (inCCLbufferSize / topoAttr_.deviceNumPerAggregation > RDMA_SEND_MAX_SIZE) || (curSize > SDMA_SEND_MAX_SIZE);
191 0 : bool isSmallData = IsBroadcastSmallData(curSize, totalSize);
192 0 : u64 sliceNum = 0;
193 0 : CHK_RET(GetSliceNum(curSize, isSmallData, sliceNum));
194 0 : CopyPattern copy = DMAReduceFlag_ ? CopyPattern::ZCOPY : CopyPattern::BCOPY;
195 0 : auto meta = HcclOpMetaInfo::GetOneForBroadcast(isRootRank, param.root, hugeData, isSmallData, sliceNum, copy);
196 0 : CHK_RET(InitTask(dispatcher_, param.stream, meta.isEnableCache, meta.GetCacheKey()));
197 0 : HCCL_INFO(
198 : "RunLoopInner:curPtr[%p], curCount[%llu], curSize[%llu], isSmallData[%u],"
199 : "deviceNumPerAggregation[%u]",
200 : curPtr, execMem.count, curSize, isSmallData, topoAttr_.deviceNumPerAggregation);
201 :
202 : // 执行
203 : HcclResult ret;
204 :
205 : // isDMAreduceOn91093场景
206 0 : if (isDMAreduceOn91093) {
207 0 : if (desc_.isZeroCopy) {
208 0 : ret = KernelRunInterServer(param, execMem);
209 : } else {
210 0 : ret = KernelRun(param, execMem);
211 : }
212 0 : CHK_PRT_RET(
213 : ret != HCCL_SUCCESS,
214 : HCCL_ERROR(
215 : "[CollBroadcastExecutor][RunLoop]errNo[0x%016llx] DMA reduce 91093, tag[%s]", HCCL_ERROR_CODE(ret),
216 : tag_.c_str()),
217 : ret);
218 : } else {
219 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
220 0 : DeviceMem inCommMem = execMem.inputMem.range(0, curSize);
221 0 : DeviceMem inMem(execMem.inputPtr, curSize);
222 0 : if (topoAttr_.userRank == param.root) {
223 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inCommMem, inMem, param.stream));
224 : }
225 0 : HCCL_DEBUG("[CollBroadcastExecutor][RunLoop]copy from user in to ccl in.");
226 :
227 0 : ret = KernelRun(param, execMem);
228 0 : if (topoAttr_.realUserRank != param.root) {
229 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inMem, inCommMem, param.stream));
230 : }
231 :
232 0 : CHK_PRT_RET(
233 : ret != HCCL_SUCCESS,
234 : HCCL_ERROR(
235 : "[CollBroadcastExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], "
236 : "inputMem ptr[%p], count[%llu], dataType[%d]",
237 : HCCL_ERROR_CODE(ret), param.tag.c_str(), execMem.inputMem.ptr(), execMem.count, param.DataDes.dataType),
238 : ret);
239 0 : }
240 :
241 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
242 0 : return ret;
243 : }
244 :
245 0 : u64 CollBroadcastExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
246 : {
247 : // 中转内存单次最多能够接受的output count
248 0 : u64 maxCountPerLoop = cclBuffSize / unitSize;
249 0 : HCCL_WARNING(
250 : "[CollBroadcastExecutor][CalcLoopMaxCount]"
251 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize.",
252 : maxCountPerLoop);
253 0 : return maxCountPerLoop;
254 : }
255 :
256 4 : HcclResult CollBroadcastExecutor::GetSliceNum(const u64 size, const bool isSmallData, u64& sliceNum)
257 : {
258 4 : u64 actualSize = 0;
259 4 : u32 actualRankSize = 0;
260 :
261 4 : if (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_RESERVED) {
262 : // level0算法配null走单层拓扑场景
263 2 : actualSize = size;
264 2 : actualRankSize = topoAttr_.userRankSize;
265 : } else {
266 : // 非单层拓扑场景
267 2 : const u32 localRankSize = topoAttr_.deviceNumPerAggregation;
268 2 : const u32 localRank = topoAttr_.userRank % localRankSize;
269 2 : const u64 tempPerSlice = (size + localRankSize - 1) / localRankSize;
270 2 : const u64 sizePerSlice
271 2 : = ((tempPerSlice + (HCCL_MIN_SLICE_ALIGN - 1)) / HCCL_MIN_SLICE_ALIGN) * HCCL_MIN_SLICE_ALIGN;
272 :
273 2 : if ((localRank + 1) * sizePerSlice < size) {
274 2 : actualSize = sizePerSlice;
275 0 : } else if (localRank * sizePerSlice < size) {
276 0 : actualSize = size - localRank * sizePerSlice;
277 : }
278 :
279 2 : actualRankSize = topoAttr_.userRankSize / localRankSize;
280 : }
281 :
282 4 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
283 4 : u64 sliceSize = (actualSize + (actualRankSize - 1)) / actualRankSize;
284 4 : u64 sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSize, HCCL_MIN_SLICE_ALIGN);
285 4 : sliceNum = isSmallData ? 1 : static_cast<u64>(std::ceil(actualSize * 1.0f / sliceSizeAligned));
286 : }
287 4 : return HCCL_SUCCESS;
288 : }
289 :
290 0 : bool CollBroadcastExecutor::IsBroadcastSmallData(u64 size, u64 totalSize)
291 : {
292 : u64 actualSize;
293 : u64 actualRankSize;
294 :
295 0 : if ((topoAttr_.serverNum == 1) && (topoAttr_.deviceType == DevType::DEV_TYPE_910_93)) {
296 0 : return totalSize <= topoAttr_.userRankSize * HCCL_SMALL_COUNT_2_MB;
297 : }
298 :
299 0 : if (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_RESERVED
300 0 : || (topoAttr_.deviceType == DevType::DEV_TYPE_910_93 && DMAReduceFlag_ == false)) {
301 : // level0算法配null走单层拓扑场景
302 0 : actualSize = size;
303 0 : actualRankSize = topoAttr_.userRankSize;
304 : } else {
305 : // 非单层拓扑场景
306 0 : actualSize = size / topoAttr_.deviceNumPerAggregation;
307 0 : actualRankSize = topoAttr_.userRankSize / topoAttr_.deviceNumPerAggregation;
308 : }
309 :
310 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
311 0 : return actualSize <= NHR_BCAST_SMALL_SIZE;
312 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
313 0 : return ShouldUseBinaryBroadcastOfNB(
314 0 : actualSize, actualRankSize, topoAttr_.userRankSize, topoAttr_.deviceNumPerAggregation);
315 : }
316 0 : return false;
317 : }
318 :
319 0 : HcclResult CollBroadcastExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
320 : {
321 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
322 0 : inputType = TransportMemType::CCL_INPUT;
323 0 : outputType = TransportMemType::CCL_INPUT;
324 : } else {
325 0 : inputType = TransportMemType::PARAM_INPUT;
326 0 : outputType = TransportMemType::PARAM_INPUT;
327 : }
328 0 : HCCL_INFO(
329 : "[CollBroadcastExecutor][CalcTransportMemType] tag[%s] inputType[%d] outputType[%d]", tag_.c_str(), inputType,
330 : outputType);
331 0 : return HCCL_SUCCESS;
332 : }
333 :
334 0 : HcclResult CollBroadcastExecutor::GetRankSliceSize(
335 : HcclDataType dataType, const u64 count, const u32 rankSize, std::vector<Slice>& sliceList)
336 : {
337 0 : if (rankSize <= 0) {
338 0 : HCCL_ERROR(
339 : "[Get][RankSliceSize]errNo[0x%016llx] rankSize[%u] is invalid", HCCL_ERROR_CODE(HCCL_E_PARA), rankSize);
340 0 : return HCCL_E_PARA;
341 : }
342 :
343 0 : u32 perDataSize = 0;
344 0 : CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
345 :
346 0 : u64 align = (count * perDataSize) / rankSize; // 按128字节对齐整除均分
347 0 : if ((count % rankSize) > 0) {
348 0 : align += 1;
349 : }
350 :
351 0 : u64 sliceSize = AlgTemplateBase::RoundUpWithDivisor(align, HCCL_MIN_SLICE_ALIGN);
352 0 : u64 residueSize = count * perDataSize;
353 :
354 0 : for (u32 i = 0; i < rankSize; i++) {
355 0 : Slice slice;
356 0 : slice.size = sliceSize < residueSize ? sliceSize : residueSize;
357 0 : slice.offset = (slice.size == 0) ? 0 : (i * sliceSize);
358 0 : residueSize -= slice.size;
359 :
360 : // 将cout转换为字节数
361 0 : sliceList.push_back(slice);
362 : }
363 :
364 0 : return HCCL_SUCCESS;
365 : }
366 : } // namespace hccl
|
