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