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 <cmath>
12 : #include <algorithm>
13 : #include "device_capacity.h"
14 : #include "coll_executor_base.h"
15 : #include "coll_alg_exec_registry.h"
16 : #include "coll_alg_operator.h"
17 :
18 : namespace hccl {
19 : using namespace std;
20 : constexpr float GB2B = 1024 * 1024 * 1024;
21 : constexpr float SECOND2MICROSECOND = 1000000;
22 : constexpr float RHD_FACTOR_TWO = 2.0;
23 : constexpr float RHD_FACTOR_ONE = 1.0;
24 : constexpr float DOUBLE_SUB_HCCLCMD = 2.0; // The hcclCMD can be considered as combination of two hcclCMDs.
25 : constexpr float COPY_TIME_IN_RHD = 1.0;
26 : constexpr double NHR_FACTOR_TWO = 2.0;
27 : constexpr double NHR_FACTOR_THREE = 3.0;
28 : constexpr double NHR_FACTOR_FOUR = 4.0;
29 : constexpr double NHR_SUB_TWO = 2.0;
30 : constexpr float LATENCY = 60; // 静态时延 60 us;
31 : constexpr u64 PIPELINE_MIN_SIZE = 32 * 1024; // 当数据量大于等于32KB时，reduce_scatter和all_gather使能pipeline模式
32 : constexpr u64 PIPELINE_ALLREDUCE_MIN_SIZE = 1024 * 1024; // 当数据量大于等于1MB时，allreduce使能pipeline模式
33 : constexpr u64 PIPELINE_MIN_SIZE_NO_LITE = 2 * 1024 * 1024; // 如不支持RDMALite，当数据量大于等于2MB时，使能pipeline模式
34 : constexpr u64 HCCL_FFTS_CAPACITY = 65535; // FFTS+子图最大容量
35 : constexpr u32 AHC_MIN_SUBGROUP_SPLIT_DIVISOR = 2;
36 : constexpr u32 AHC_LEVEL0_GROUP_SIZE_THRESHOLD = 3;
37 : constexpr u32 SERVER_COUNT_THRESHOLD_FOR_MULTI_DETER_PIPELINE = 2;
38 : constexpr u32 MIN_STRICT_RANK_NUM = 3;
39 :
40 79 : CollAlgOperator::CollAlgOperator(AlgConfigurator* algConfigurator, CCLBufferManager &cclBufferManager,
41 : HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher> &topoMatcher,
42 79 : HcclCMDType opType)
43 79 : : algConfigurator_(algConfigurator), cclBufferManager_(cclBufferManager),
44 559 : dispatcher_(dispatcher), topoMatcher_(topoMatcher), workflowMode_(GetWorkflowMode())
45 : {
46 79 : SetTopoAttr(algConfigurator_);
47 82 : SetAlgoAttr(algConfigurator_);
48 92 : algConfigurator->GetAlgTypeDirect(algType_, opType);
49 82 : algConfigurator->GetAlgoLevel1DefaultSwitch(isAlgoLevel1Default_, opType);
50 78 : algConfigurator->GetTopoType(topoType_);
51 78 : }
52 :
53 0 : HcclResult CollAlgOperator::SelectAlg(const std::string& tag,
54 : const OpParam& param, std::string& algName, std::string& newTag)
55 : {
56 0 : return HCCL_SUCCESS;
57 : }
58 :
59 36 : HcclResult CollAlgOperator::SelectAlg(const std::string& tag,
60 : const OpParam& param, std::string& algName, std::string& newTag, const ResourceLimit &limit)
61 : {
62 36 : return SelectAlg(tag, param, algName, newTag);
63 : }
64 :
65 1 : HcclResult CollAlgOperator::GetAivExecParam(std::string& algName, const OpParam& param,
66 : AlgResourceResponse& algRes, AivSuperKernelArgs &args)
67 : {
68 1 : if (executor_.get() == nullptr) {
69 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
70 0 : CHK_PRT_RET(executor_.get() == nullptr,
71 : HCCL_ERROR("[CollAlgOperator][GetAivExecParam]Fail to find executor for algName[%s]", algName.c_str()),
72 : HCCL_E_PARA);
73 : }
74 1 : return executor_->GetAivExecParam(param, algRes, args);
75 : }
76 :
77 1 : HcclResult CollAlgOperator::CalNumBlocks(std::string& algName, const OpParam& param, u32 &numBlocks, int32_t aivCoreLimit)
78 : {
79 1 : if (executor_.get() == nullptr) {
80 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
81 0 : CHK_PRT_RET(executor_.get() == nullptr,
82 : HCCL_ERROR("[CollAlgOperator][CalNumBlocks]Fail to find executor for algName[%s]", algName.c_str()),
83 : HCCL_E_PARA);
84 0 : CHK_RET(SetExecutorAttr(param));
85 : }
86 :
87 1 : if (aivCoreLimit != 0) {
88 1 : CHK_RET(executor_->SetNumBlocks(aivCoreLimit));
89 : }
90 :
91 1 : if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALL) {
92 0 : return executor_->CalNumBlocks(numBlocks, userRankSize_,
93 0 : param.All2AllDataDes.sendCount * SIZE_TABLE[param.All2AllDataDes.sendType], param.opType);
94 1 : } else if (param.opType == HcclCMDType::HCCL_CMD_ALLREDUCE || param.opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER
95 0 : || param.opType == HcclCMDType::HCCL_CMD_ALLGATHER || param.opType == HcclCMDType::HCCL_CMD_BROADCAST) {
96 1 : return executor_->CalNumBlocks(numBlocks, userRankSize_,
97 1 : param.DataDes.count * SIZE_TABLE[param.DataDes.dataType], param.opType);
98 : } else {
99 0 : return executor_->CalNumBlocks(numBlocks, userRankSize_);
100 : }
101 : return HCCL_SUCCESS;
102 : }
103 :
104 0 : HcclResult CollAlgOperator::GetOpExpansionStr(const OpParam ¶m, const AlgDesc &algDesc, std::string &opExpansionStr)
105 : {
106 0 : if (algDesc.isAivMode) {
107 0 : opExpansionStr = "AIV";
108 0 : } else if (param.aicpuUnfoldMode) {
109 0 : opExpansionStr = "AI_CPU";
110 0 : } else if (static_cast<bool>(topoMatcher_->GetExternalInputHcclEnableFfts())) {
111 0 : opExpansionStr = "HOST";
112 : } else {
113 0 : opExpansionStr = "HOST_TS";
114 : }
115 0 : return HCCL_SUCCESS;
116 : }
117 :
118 77 : HcclResult CollAlgOperator::SelectAlg(const std::string& tag, const OpParam ¶m, const ResourceLimit &limit,
119 : std::string &algName, AlgDesc &algDesc, std::string &newTag)
120 : {
121 77 : bool isOnlyAiv = topoMatcher_->GetIsOnlyAivConfig();
122 218 : bool supportOnlyAiv = (param.opType == HcclCMDType::HCCL_CMD_ALLGATHER ||
123 66 : param.opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER ||
124 31 : param.opType == HcclCMDType::HCCL_CMD_ALLTOALLV ||
125 27 : param.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC ||
126 168 : param.opType == HcclCMDType::HCCL_CMD_ALLTOALL ||
127 26 : param.opType == HcclCMDType::HCCL_CMD_ALLREDUCE);
128 76 : CHK_PRT_RET(isOnlyAiv && !supportOnlyAiv,
129 : HCCL_ERROR("[CollAlgOperator][SelectAlg] opType[%s] currently do not support aivonly",
130 : GetCMDTypeEnumStr(param.opType).c_str()), HCCL_E_NOT_SUPPORT);
131 76 : CHK_PRT_RET(isOnlyAiv && userRankSize_ == 1 && supportOnlyAiv,
132 : HCCL_ERROR("[CollAlgOperator][SelectAlg] aivonly not support, please ensure rankNum is greater than one"),
133 : HCCL_E_NOT_SUPPORT);
134 :
135 : // 兼容老接口
136 76 : if (limit.ifLimit) {
137 76 : CHK_RET(SelectAlg(tag, param, algName, newTag, limit));
138 : } else {
139 0 : CHK_RET(SelectAlg(tag, param, algName, newTag));
140 : }
141 :
142 : // 非AIV算法提前返回, 采用兜底Executor
143 78 : if (algName.empty()) {
144 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec("SendExecutor", dispatcher_, topoMatcher_);
145 0 : CHK_PRT_RET(executor_.get() == nullptr,
146 : HCCL_ERROR("[CollAlgOperator][SelectAlg]Fail to find executor for algName[DefaultExecutor]"),
147 : HCCL_E_PARA);
148 : } else {
149 : // 校验控核
150 78 : if (limit.ifLimit && deviceType_ == DevType::DEV_TYPE_910_93 && topoMatcher_->GetAivModeConfig()) {
151 0 : CHK_RET(SelectAlgFor91093WithCoreLimit(param, limit, algName));
152 : }
153 :
154 : // 从对应executor获取算法描述
155 78 : if (executor_.get() == nullptr) {
156 76 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
157 76 : CHK_PRT_RET(executor_.get() == nullptr,
158 : HCCL_ERROR("[CollAlgOperator][SelectAlg]Fail to find executor for algName[%s]", algName.c_str()),
159 : HCCL_E_PARA);
160 76 : CHK_RET(SetExecutorAttr(param));
161 : }
162 : }
163 :
164 74 : bool isLastSelect = algDesc.isLastSelect;
165 74 : algDesc = executor_->GetAlgDesc();
166 : // 打印维测日志
167 69 : if (UNLIKELY(GetDebugConfig() & HCCL_ALG) && isLastSelect) {
168 : // 获取展开模式，转换成字符串
169 0 : std::string opExpansionStr;
170 0 : CHK_RET(GetOpExpansionStr(param, algDesc, opExpansionStr));
171 : // 尝试获取确定性属性（如果Executor有声明自己是否为确定性）
172 0 : std::string appendStr = "";
173 0 : if (algDesc.deterministic >= 0) {
174 0 : appendStr += "deterministic[" + std::to_string(algDesc.deterministic) + "]";
175 : }
176 : // 打印关键维测内容
177 0 : bool isOpBase = GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
178 0 : HCCL_CONFIG_INFO(HCCL_ALG,
179 : "[%s] newTag[%s] algName[%s] userRank[%u] topoType[%d] algType[%s] "\
180 : "userRankSize[%u] level0Size[%u] moduleNum_[%u] level2Size[%u] ",
181 : __func__, newTag.c_str(), algName.c_str(), userRank_, topoType_, AlgTypeToStr(algDesc.algType).c_str(),
182 : userRankSize_, deviceNumPerAggregation_, moduleNum_, superPodNum_);
183 0 : HCCL_CONFIG_INFO(HCCL_ALG,
184 : "[%s] newTag[%s] "\
185 : "opExpansionMode[%s] isZeroCopy[%u] retryEnable[%u] isOpBase[%u] isCapture[%u] aivCoreLimit[%u] %s.",
186 : __func__, newTag.c_str(),
187 : opExpansionStr.c_str(), algDesc.isZeroCopy, retryEnable_, isOpBase, param.isCapture, limit.aivCoreLimit, appendStr.c_str());
188 0 : }
189 72 : return HCCL_SUCCESS;
190 : }
191 :
192 0 : HcclResult CollAlgOperator::SelectAlgFor91093WithCoreLimit(const OpParam ¶m, const ResourceLimit &limit,
193 : std::string &algName)
194 : {
195 0 : if (executor_.get() == nullptr) {
196 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
197 0 : CHK_PRT_RET(executor_.get() == nullptr,
198 : HCCL_ERROR("[CollAlgOperator][SelectAlgFor91093WithCoreLimit]Fail to find executor for algName[%s]", algName.c_str()),
199 : HCCL_E_PARA);
200 0 : CHK_RET(SetExecutorAttr(param));
201 : }
202 :
203 0 : CHK_RET(SetNumBlocks(limit.aivCoreLimit));
204 :
205 0 : std::string reSelName;
206 0 : switch (param.opType) {
207 0 : case HcclCMDType::HCCL_CMD_ALLREDUCE:
208 0 : reSelName = "AllReduceMeshAivFor91093Executor";
209 0 : break;
210 0 : case HcclCMDType::HCCL_CMD_ALLGATHER:
211 0 : reSelName = "AllGatherMeshAivFor91093Executor";
212 0 : break;
213 0 : case HcclCMDType::HCCL_CMD_REDUCE_SCATTER:
214 0 : reSelName = "ReduceScatterMeshAivFor91093Executor";
215 0 : break;
216 0 : case HcclCMDType::HCCL_CMD_ALLTOALLV:
217 : case HcclCMDType::HCCL_CMD_ALLTOALL:
218 : case HcclCMDType::HCCL_CMD_ALLTOALLVC:
219 0 : reSelName = "AlltoAllMeshAivFor91093Executor";
220 0 : break;
221 0 : default:
222 0 : break;
223 : }
224 :
225 : u32 numBlocks;
226 0 : HcclResult ret = CalNumBlocks(algName, param, numBlocks);
227 0 : if (ret != HCCL_SUCCESS) {
228 0 : CHK_PRT_RET(reSelName.empty() || reSelName == algName,
229 : HCCL_ERROR("[CollAlgOperator][SelectAlgFor91093WithCoreLimit]Fail to check CalNumBlocks for algName[%s]", algName.c_str()),
230 : HCCL_E_PARA);
231 :
232 0 : algName = reSelName;
233 0 : executor_ = nullptr;
234 0 : HCCL_INFO("[CollAlgOperator][SelectAlgFor91093WithCoreLimit]Re select to algName[%s]", reSelName.c_str());
235 : }
236 :
237 0 : return HCCL_SUCCESS;
238 0 : }
239 :
240 77 : HcclResult CollAlgOperator::CalcResRequest(const std::string& algName, const OpParam& param,
241 : AlgResourceRequest& resourceRequest)
242 : {
243 77 : if (executor_.get() == nullptr) {
244 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
245 0 : CHK_PRT_RET(executor_.get() == nullptr,
246 : HCCL_ERROR("[CollAlgOperator][CalcResRequest]Fail to find executor for algName[%s]", algName.c_str()),
247 : HCCL_E_PARA);
248 0 : CHK_RET(SetExecutorAttr(param));
249 : }
250 75 : return executor_->CalcResRequest(param, resourceRequest);
251 : }
252 :
253 24 : HcclResult CollAlgOperator::Orchestrate(const std::string& algName, OpParam& param, AlgResourceResponse& algResource)
254 : {
255 24 : HCCL_INFO("[CollAlgOperator][Orchestrate]algName[%s]", algName.c_str());
256 24 : if (executor_.get() == nullptr) {
257 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
258 0 : CHK_PRT_RET(executor_.get() == nullptr,
259 : HCCL_ERROR("[CollAlgOperator][Orchestrate]Fail to find executor for algName[%s]", algName.c_str()),
260 : HCCL_E_PARA);
261 0 : CHK_RET(SetExecutorAttr(param));
262 : }
263 24 : executor_->SetAivClearEnable(aivClearEnable_);
264 24 : executor_->SetAlgOpContext(algOpContext_);
265 24 : executor_->SetOpCounter(opCounter_);
266 24 : return executor_->Orchestrate(param, algResource);
267 : }
268 :
269 0 : HcclResult CollAlgOperator::GetAdjInfo(const std::string& algName, OpParam& param,
270 : AlgResourceResponse& algResource, AdjInfo& nslbAdjInfo)
271 : {
272 0 : if (executor_.get() == nullptr) {
273 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
274 0 : CHK_PRT_RET(executor_.get() == nullptr,
275 : HCCL_ERROR("[CollAlgOperator][Orchestrate]Fail to find executor for algName[%s]", algName.c_str()),
276 : HCCL_E_PARA);
277 0 : CHK_RET(SetExecutorAttr(param));
278 : }
279 :
280 0 : return executor_->GetAdjInfo(algResource, nslbAdjInfo);
281 : }
282 :
283 25 : HcclResult CollAlgOperator::PrepareCommInfoToDevice(const std::string& algName, AlgResourceResponse& algResource)
284 : {
285 25 : if (executor_.get() == nullptr) {
286 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
287 0 : CHK_PRT_RET(executor_.get() == nullptr,
288 : HCCL_ERROR("[CollAlgOperator][PrepareCommInfoToDevice]Fail to find executor for algName[%s]",
289 : algName.c_str()), HCCL_E_PARA);
290 : }
291 25 : return executor_->PrepareCommInfoToDevice(algResource);
292 : }
293 :
294 0 : HcclResult CollAlgOperator::CalcIncreLinkRequest(const std::string& algName, const OpParam& param,
295 : std::set<u32>& ranksHasLinked, AlgResourceRequest& resourceRequest, bool& needIncreLink)
296 : {
297 0 : if (executor_.get() == nullptr) {
298 0 : executor_ = CollAlgExecRegistry::Instance().GetAlgExec(algName, dispatcher_, topoMatcher_);
299 0 : CHK_PRT_RET(executor_.get() == nullptr,
300 : HCCL_ERROR("[CollAlgOperator][CalcIncreLinkRequest]Fail to find executor for algName[%s]",
301 : algName.c_str()), HCCL_E_PARA);
302 : }
303 0 : return executor_->CalcIncreLinkRequest(param, ranksHasLinked, resourceRequest, needIncreLink);
304 : }
305 :
306 81 : void CollAlgOperator::SetTopoAttr(AlgConfigurator* algConfigurator)
307 : {
308 81 : const HcclTopoAttr& topoAttr = algConfigurator->GetTopoAttr();
309 85 : serverNum_= topoAttr.serverNum;
310 85 : moduleNum_ = topoAttr.moduleNum;
311 85 : superPodNum_ = topoAttr.superPodNum;
312 85 : deviceNumPerServer_ = topoAttr.deviceNumPerServer;
313 85 : deviceNumPerAggregation_ = topoAttr.deviceNumPerAggregation;
314 85 : multiModuleDiffDeviceNumMode_ = topoAttr.multiModuleDiffDeviceNumMode;
315 85 : multiSuperPodDiffServerNumMode_ = topoAttr.multiSuperPodDiffServerNumMode;
316 85 : multiSuperPodDiffDeviceNumMode_ = topoAttr.multiSuperPodDiffDeviceNumMode;
317 85 : isDiffDeviceType_ = topoAttr.isDiffDeviceType;
318 85 : gcdDeviceNumPerAggregation_ = topoAttr.gcdDeviceNumPerAggregation;
319 :
320 85 : meshAggregationRankSize_ = topoAttr.meshAggregationRankSize;
321 85 : isDiffDeviceModule_ = topoAttr.isDiffDeviceModule;
322 85 : isSingleMeshAggregation_ = topoAttr.isSingleMeshAggregation;
323 85 : isAllRankSamePlane_ = topoAttr.isAllRankSamePlane;
324 85 : is310PDuoCard_ = topoAttr.is310PDuoCard;
325 85 : isCommon310P3DUO_ = topoAttr.isCommon310P3DUO;
326 85 : hccsPortNum_ = topoAttr.hccsPortNum;
327 :
328 85 : userRank_ = topoAttr.userRank;
329 85 : realUserRank_ = topoAttr.realUserRank;
330 85 : userRankSize_ = topoAttr.userRankSize;
331 :
332 85 : devicePhyId_ = topoAttr.devicePhyId;
333 85 : deviceLogicId_ = topoAttr.deviceLogicId;
334 85 : deviceType_ = topoAttr.deviceType;
335 :
336 85 : nicList_ = topoAttr.nicList;
337 79 : pairLinkCounter_ = topoAttr.pairLinkCounter;
338 82 : isSupportRdmaLite_ = topoAttr.isSupportRdmaLite;
339 82 : isSupportHccsAndSio_ = topoAttr.isSupportHccsAndSio;
340 82 : useSuperPodMode_ = topoAttr.useSuperPodMode;
341 82 : isARSDoubleRing_ = topoAttr.isARSDoubleRing;
342 82 : return;
343 : }
344 :
345 82 : void CollAlgOperator::SetAlgoAttr(AlgConfigurator* algConfigurator)
346 : {
347 82 : const HcclAlgoAttr& algoAttr = algConfigurator->GetAlgoAttr();
348 82 : isHaveCpuRank_ = algoAttr.isHaveCpuRank;
349 82 : inlineReduceSwitchOn_ = algoAttr.inlineReduceSwitchOn;
350 82 : identifier_ = algoAttr.identifier;
351 92 : return;
352 : }
353 :
354 71 : HcclResult CollAlgOperator::SetExecutorAttr(const OpParam& param)
355 : {
356 71 : CHK_RET(executor_->SetAlgType(algType_));
357 67 : CHK_RET(executor_->SetCCLInBuffer(cclBufferManager_.GetInCCLbufferSize()));
358 :
359 70 : if (param.opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
360 35 : bool isSupportSDMAReduce = false;
361 35 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
362 14 : isSupportSDMAReduce = IsSupportSDMAReduce(param.inputPtr, param.outputPtr, param.DataDes.dataType,
363 14 : param.reduceType);
364 : } else {
365 21 : isSupportSDMAReduce = IsSupportSDMAReduce(cclBufferManager_.GetInCCLbuffer().ptr(),
366 21 : cclBufferManager_.GetOutCCLbuffer().ptr(), param.DataDes.dataType, param.reduceType);
367 : }
368 32 : CHK_RET(executor_->SetIsSupportSDMAReduce(isSupportSDMAReduce));
369 : }
370 69 : return HCCL_SUCCESS;
371 : }
372 :
373 0 : std::string CollAlgOperator::GenerateNewTagByAlgTypeLevel1(std::string tag, std::string algTypeLevel1Tag) const
374 : {
375 0 : if (algTypeLevel1Tag == "") {
376 0 : return tag;
377 : } else {
378 0 : return tag + "_" + algTypeLevel1Tag;
379 : }
380 : }
381 :
382 0 : HcclResult CollAlgOperator::AppendTag(const AlgTypeLevel1 &algTypeLevel1, std::string &tag)
383 : {
384 0 : switch (algTypeLevel1) {
385 0 : case AlgTypeLevel1::ALG_LEVEL1_RING:
386 0 : tag = "ALG_LEVEL1_RING";
387 0 : break;
388 0 : case AlgTypeLevel1::ALG_LEVEL1_HD:
389 0 : tag = "ALG_LEVEL1_HD";
390 0 : break;
391 0 : case AlgTypeLevel1::ALG_LEVEL1_NHR:
392 0 : tag = "ALG_LEVEL1_NHR";
393 0 : break;
394 0 : case AlgTypeLevel1::ALG_LEVEL1_PIPELINE:
395 0 : tag = "ALG_LEVEL1_PIPELINE";
396 0 : break;
397 0 : default:
398 0 : HCCL_WARNING("[CollAlgOperator][AppendTag] The algTypeLevel1 %d is not supported.", algTypeLevel1);
399 0 : break;
400 : }
401 0 : return HCCL_SUCCESS;
402 : }
403 :
404 16 : HcclResult CollAlgOperator::AutoSelectAlgTypeLevel1(HcclCMDType hcclCMDType, u64 countSize, u64 cclBufferSize,
405 : std::string &algTypeLevel1Tag, bool isInlineReduce,
406 : bool isRdmaReduce, bool isAivMode)
407 : {
408 16 : if (isSingleMeshAggregation_) {
409 16 : HCCL_INFO("[AutoSelectAlgTypeLevel1] there are %u server(%u module) in level1, no need to choose algo.",
410 : serverNum_, moduleNum_);
411 20 : return HCCL_SUCCESS;
412 : }
413 :
414 : // auto algo selection process
415 0 : if (isAlgoLevel1Default_) {
416 : // parse algType_ and get algTypeLevel1 and algTypeLevel0
417 0 : auto originalAlgTypeLevel0 = algType_.algoLevel0;
418 : // set algTypeLevel1
419 : AlgTypeLevel1 algTypeLevel1;
420 0 : CHK_RET(
421 : GetDefaultAlgoLevel1V2(
422 : hcclCMDType, countSize, cclBufferSize, algTypeLevel1, isInlineReduce, isRdmaReduce, isAivMode));
423 0 : auto iter = HCCL_ALGO_LEVEL1_NAME_MAP.find(algTypeLevel1);
424 0 : CHK_PRT_RET(iter == HCCL_ALGO_LEVEL1_NAME_MAP.end(),
425 : HCCL_ERROR("[AutoSelectAlgTypeLevel1] level1: algType[%u] is invalid.", algTypeLevel1),
426 : HCCL_E_INTERNAL);
427 0 : HCCL_INFO("[AutoSelectAlgTypeLevel1] there are %u server(%u module) in level1, using %s algo",
428 : serverNum_, moduleNum_, iter->second.c_str());
429 0 : algType_.algoLevel0 = originalAlgTypeLevel0;
430 0 : algType_.algoLevel1 = algTypeLevel1;
431 : // tag 增加所选的算法
432 0 : AppendTag(algTypeLevel1, algTypeLevel1Tag);
433 : }
434 0 : return HCCL_SUCCESS;
435 : }
436 :
437 0 : HcclResult CollAlgOperator::SelectAlgoForComm(HcclCMDType hcclCMDType, float delay, u64 curSize, float bandWidth,
438 : AlgTypeLevel1 &algType)
439 : {
440 : // 从map中查找对应的计算函数
441 0 : auto it = selectFuncMap_.find(hcclCMDType);
442 0 : if (it == selectFuncMap_.end()) {
443 0 : HCCL_ERROR("[Get][AlgTypeLevel1] The hcclCMDType %d is not supported.", hcclCMDType);
444 0 : return HCCL_E_NOT_SUPPORT;
445 : }
446 0 : return (it->second)(delay, curSize, bandWidth, algType);
447 : }
448 :
449 : // 保守估计Pipeline算法所需context数量
450 0 : u32 CollAlgOperator::CalcContextNumForPipeline(HcclCMDType hcclCMDType)
451 : {
452 0 : bool isDeterPipeline = topoMatcher_->GetDeterministicConfig() == DETERMINISTIC_ENABLE
453 0 : && (hcclCMDType == HcclCMDType::HCCL_CMD_ALLREDUCE || hcclCMDType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER);
454 0 : const u32 stepNum = moduleNum_; // 通信步数
455 0 : const u32 hccsContextNumPerStep = 5 * (deviceNumPerAggregation_ - 1); // SDMA跨片每步所需context数
456 0 : const u32 roceContextNumPerStep = 7; // RDMA每步所需context数
457 0 : const u32 copyContextNumPerStep = 1; // SDMA片内每步所需context数
458 0 : const u32 localReduceNumPerStep = isDeterPipeline ? (deviceNumPerAggregation_ - 1) : 0;
459 0 : const u32 contextNumPerStep = hccsContextNumPerStep + roceContextNumPerStep + copyContextNumPerStep
460 0 : + localReduceNumPerStep; // 小计
461 0 : const u32 barrierContextNum = 4; // 通信结束时barrier操作所需context数
462 :
463 0 : switch (hcclCMDType) {
464 0 : case HcclCMDType::HCCL_CMD_ALLREDUCE: // fall-through
465 : case HcclCMDType::HCCL_CMD_REDUCE_SCATTER: // fall-through
466 : case HcclCMDType::HCCL_CMD_ALLGATHER:
467 : case HcclCMDType::HCCL_CMD_ALLGATHER_V:{
468 0 : const u32 copyContextNum = 1; // 通信首尾所需context数量
469 0 : u32 contextNum = stepNum * contextNumPerStep + barrierContextNum + copyContextNum;
470 0 : if (hcclCMDType == HcclCMDType::HCCL_CMD_ALLREDUCE) {
471 0 : contextNum += contextNum;
472 : }
473 0 : if (isDeterPipeline) {
474 0 : contextNum += stepNum - 1; // 最后的local reduce
475 : }
476 0 : return contextNum;
477 : }
478 0 : case HcclCMDType::HCCL_CMD_ALLTOALLV: // fall-through
479 : case HcclCMDType::HCCL_CMD_ALLTOALLVC: // fall-through
480 : case HcclCMDType::HCCL_CMD_ALLTOALL: {
481 0 : const u32 copyContextNum = 1 + moduleNum_; // 通信首尾所需context数量
482 0 : return stepNum * contextNumPerStep + barrierContextNum + copyContextNum;
483 : }
484 0 : default:
485 0 : return 0;
486 : }
487 : }
488 :
489 0 : HcclResult CollAlgOperator::GetDefaultAlgoLevel1V2(HcclCMDType hcclCMDType, u64 curSize, u64 cclBufferSize,
490 : AlgTypeLevel1 &algType, bool isInlineReduce, bool isRdmaReduce, bool isAivMode)
491 : {
492 : // pipeline mode is deployed,where there is multi-sever multi-device(insever) now,
493 : // since RDMA is not reduced by normal serial orchestration of tasks.
494 : // So pipeline mode is more dominant than normal serial orchestration now.
495 0 : auto originalAlgTypeLevel0 = algType_.algoLevel0;
496 0 : bool disdeterniminsticWithInlineReduce = isInlineReduce && isRdmaReduce &&
497 0 : topoMatcher_->GetDeterministicConfig() == DETERMINISTIC_DISABLE;
498 0 : bool deterniminsticWithInlineReduce = isInlineReduce && isRdmaReduce &&
499 0 : topoMatcher_->GetDeterministicConfig() == DETERMINISTIC_ENABLE;
500 :
501 : // 对于不支持Rdma Lite的场景，下发性能较差，RS和AG需要一个很大的数据量（AR的一半）才能掩盖下发时间
502 0 : u64 pipelineMinSize = (isSupportRdmaLite_) ? (PIPELINE_MIN_SIZE) : (PIPELINE_MIN_SIZE_NO_LITE);
503 0 : if (((hcclCMDType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER && disdeterniminsticWithInlineReduce) ||
504 0 : hcclCMDType == HcclCMDType::HCCL_CMD_ALLGATHER || hcclCMDType == HcclCMDType::HCCL_CMD_ALLGATHER_V) &&
505 0 : deviceNumPerAggregation_ != 1 && curSize >= pipelineMinSize && IsAlgTypeLevel0Mesh(originalAlgTypeLevel0) &&
506 0 : CalcContextNumForPipeline(hcclCMDType) <= HCCL_FFTS_CAPACITY) {
507 0 : algType = AlgTypeLevel1::ALG_LEVEL1_PIPELINE;
508 0 : return HCCL_SUCCESS;
509 : }
510 0 : if (hcclCMDType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER && deterniminsticWithInlineReduce &&
511 0 : deviceNumPerAggregation_ > 1 &&
512 0 : curSize >= pipelineMinSize && IsAlgTypeLevel0Mesh(originalAlgTypeLevel0) &&
513 0 : CalcContextNumForPipeline(hcclCMDType) <= HCCL_FFTS_CAPACITY
514 0 : && moduleNum_ > 1 && curSize >= HCCL_SMALL_COUNT_256_KB) {
515 0 : algType = AlgTypeLevel1::ALG_LEVEL1_PIPELINE;
516 0 : return HCCL_SUCCESS;
517 : }
518 :
519 : // 对于不支持Rdma Lite的场景，下发性能较差，AllReduce需要一个较大的数据量才能掩盖下发时间
520 0 : pipelineMinSize = (isSupportRdmaLite_) ? (PIPELINE_ALLREDUCE_MIN_SIZE) : (PIPELINE_MIN_SIZE_NO_LITE);
521 0 : if (hcclCMDType == HcclCMDType::HCCL_CMD_ALLREDUCE) {
522 : // 计算每个slice的大小
523 0 : u64 allreduceCurSize = 0;
524 0 : allreduceCurSize = curSize / (moduleNum_ * deviceNumPerAggregation_);
525 0 : if (disdeterniminsticWithInlineReduce && deviceNumPerAggregation_ != 1 &&
526 0 : allreduceCurSize >= pipelineMinSize && !isAivMode && IsAlgTypeLevel0Mesh(originalAlgTypeLevel0) &&
527 0 : CalcContextNumForPipeline(hcclCMDType) <= HCCL_FFTS_CAPACITY) {
528 0 : algType = AlgTypeLevel1::ALG_LEVEL1_PIPELINE;
529 0 : return HCCL_SUCCESS;
530 : }
531 0 : if (deterniminsticWithInlineReduce &&
532 0 : deviceNumPerAggregation_ > 1 &&
533 0 : allreduceCurSize >= HCCL_SMALL_COUNT_1_MB && !isAivMode && IsAlgTypeLevel0Mesh(originalAlgTypeLevel0) &&
534 0 : CalcContextNumForPipeline(hcclCMDType) <= HCCL_FFTS_CAPACITY) {
535 0 : algType = AlgTypeLevel1::ALG_LEVEL1_PIPELINE;
536 0 : return HCCL_SUCCESS;
537 : }
538 : }
539 0 : u64 dataSizePerLoop = curSize > cclBufferSize ? cclBufferSize : curSize;
540 0 : float delay = LATENCY; // 静态时延 60 us;
541 : float bandWidth;
542 0 : CHK_RET(GetBandWidthPerNPU(1, userRankSize_, deviceNumPerAggregation_, bandWidth)); // 单位：GB/s
543 0 : bandWidth = bandWidth * GB2B; // 单位：B/s
544 0 : CHK_RET(SelectAlgoForComm(hcclCMDType, delay, dataSizePerLoop, bandWidth, algType));
545 0 : return HCCL_SUCCESS;
546 : }
547 :
548 0 : HcclResult CollAlgOperator::SelectAlgoTypeForReduceScatter(float delay, u64 recvCurSize, float bandWidth,
549 : AlgTypeLevel1 &algType)
550 : {
551 0 : auto steps = moduleNum_ - 1;
552 : // theoretical time cost of Ring
553 0 : double ringCost = static_cast<double>(steps) * delay +
554 0 : static_cast<double>(steps) / moduleNum_ * recvCurSize * userRankSize_ / bandWidth *
555 : SECOND2MICROSECOND;
556 :
557 : // theoretical time cost of NHR
558 0 : double nhrCost = ceil(log2(moduleNum_)) * delay +
559 0 : static_cast<double>(moduleNum_ - 1) / moduleNum_ *
560 0 : recvCurSize * userRankSize_ / bandWidth * SECOND2MICROSECOND;
561 :
562 : // compare costs between NHR and Ring, if same cost, Ring > NHR > HD
563 0 : algType = (nhrCost < ringCost) ? AlgTypeLevel1::ALG_LEVEL1_NHR : AlgTypeLevel1::ALG_LEVEL1_RING;
564 0 : double interMinCost = min(nhrCost, ringCost);
565 :
566 : // theoretical time cost of HD/RHD
567 0 : double hdCost = 0.0;
568 0 : if ((moduleNum_ & (moduleNum_ - 1)) == 0) {
569 : // theoretical time cost of HD
570 0 : hdCost = log2(moduleNum_) * delay +
571 0 : static_cast<double>(steps) / moduleNum_ * recvCurSize * userRankSize_ / bandWidth *
572 : SECOND2MICROSECOND;
573 : } else {
574 : // theoretical time cost of Recursive HD,
575 : // the (RHD_FACTOR_TWO * moduleNum_ - RHD_FACTOR_ONE) / moduleNum_ is equal to 1 + (moduleNum_ -1) / moduleNum_
576 0 : hdCost = ceil(log2(moduleNum_)) * delay +
577 0 : static_cast<double>(RHD_FACTOR_TWO * moduleNum_ - RHD_FACTOR_ONE) / moduleNum_ *
578 0 : recvCurSize * userRankSize_ / bandWidth *
579 : SECOND2MICROSECOND;
580 : }
581 :
582 : // compare cost among NHR, HD and Ring
583 0 : algType = (hdCost < interMinCost) ? AlgTypeLevel1::ALG_LEVEL1_HD : algType;
584 0 : return HCCL_SUCCESS;
585 : }
586 :
587 0 : HcclResult CollAlgOperator::SelectAlgoTypeForAllGather(float delay, u64 sendCurSize, float bandWidth,
588 : AlgTypeLevel1 &algType)
589 : {
590 0 : auto steps = moduleNum_ - 1;
591 : // theoretical time cost of Ring
592 0 : double ringCost = static_cast<double>(steps) * delay +
593 0 : static_cast<double>(steps) / moduleNum_ * sendCurSize * userRankSize_ / bandWidth *
594 : SECOND2MICROSECOND;
595 :
596 : // theoretical time cost of NHR
597 0 : double nhrCost = ceil(log2(moduleNum_)) * delay +
598 0 : static_cast<double>(moduleNum_ - 1) / moduleNum_ *
599 0 : sendCurSize * userRankSize_ / bandWidth * SECOND2MICROSECOND;
600 :
601 : // compare costs between NHR and Ring, if same cost, Ring > NHR > HD
602 0 : algType = (nhrCost < ringCost) ? AlgTypeLevel1::ALG_LEVEL1_NHR : AlgTypeLevel1::ALG_LEVEL1_RING;
603 0 : double interMinCost = min(nhrCost, ringCost);
604 :
605 : // theoretical time cost of HD/RHD
606 0 : double hdCost = 0.0;
607 0 : if ((moduleNum_ & (moduleNum_ - 1)) == 0) {
608 : // theoretical time cost of HD
609 0 : hdCost = log2(moduleNum_) * delay +
610 0 : static_cast<double>(steps) / moduleNum_ * sendCurSize * userRankSize_ / bandWidth *
611 : SECOND2MICROSECOND;
612 : } else {
613 : // theoretical time cost of Recursive HD
614 : // the (RHD_FACTOR_TWO * moduleNum_ - RHD_FACTOR_ONE) / moduleNum_ is equal to 1 + (moduleNum_ -1) / moduleNum_
615 0 : hdCost = ceil(log2(moduleNum_)) * delay +
616 0 : static_cast<double>(RHD_FACTOR_TWO * moduleNum_ - RHD_FACTOR_ONE) / moduleNum_ *
617 0 : sendCurSize * userRankSize_ / bandWidth * SECOND2MICROSECOND;
618 : }
619 :
620 : // compare cost among NHR, HD and Ring
621 0 : algType = (hdCost < interMinCost) ? AlgTypeLevel1::ALG_LEVEL1_HD : algType;
622 0 : return HCCL_SUCCESS;
623 : }
624 :
625 0 : HcclResult CollAlgOperator::SelectAlgoTypeForAllGatherV(float delay, u64 sendCurSize, float bandWidth,
626 : AlgTypeLevel1 &algType)
627 : {
628 0 : auto steps = moduleNum_ - 1;
629 : // theoretical time cost of Ring
630 0 : double ringCost = static_cast<double>(steps) * delay +
631 0 : static_cast<double>(steps) / moduleNum_ * sendCurSize * userRankSize_ / bandWidth *
632 : SECOND2MICROSECOND;
633 :
634 0 : HCCL_DEBUG("[%s] CollAlgOperator for SelectAlgoTypeForAllGatherV", __func__);
635 : // theoretical time cost of NHR
636 0 : double nhrCost = ceil(log2(moduleNum_)) * delay +
637 0 : static_cast<double>(moduleNum_ - 1) / moduleNum_ *
638 0 : sendCurSize * userRankSize_ / bandWidth * SECOND2MICROSECOND;
639 :
640 : // compare costs between NHR and Ring, if same cost, Ring > NHR > HD
641 0 : algType = (nhrCost < ringCost) ? AlgTypeLevel1::ALG_LEVEL1_NHR : AlgTypeLevel1::ALG_LEVEL1_RING;
642 :
643 0 : return HCCL_SUCCESS;
644 : }
645 :
646 0 : HcclResult CollAlgOperator::SelectAlgoTypeForGather(float delay, u64 sendCurSize, float bandWidth,
647 : AlgTypeLevel1 &algType)
648 : {
649 0 : auto steps = moduleNum_ - 1;
650 : // theoretical time cost of Ring
651 0 : double ringCost = static_cast<double>(steps) * delay +
652 0 : static_cast<double>(steps) / moduleNum_ * sendCurSize * userRankSize_ / bandWidth *
653 : SECOND2MICROSECOND;
654 0 : double hdCost = 0.0;
655 0 : if ((moduleNum_ & (moduleNum_ - 1)) == 0) {
656 : // theoretical time cost of HD
657 0 : hdCost = log2(moduleNum_) * delay +
658 0 : static_cast<double>(steps) / moduleNum_ * sendCurSize * userRankSize_ / bandWidth *
659 : SECOND2MICROSECOND;
660 : } else {
661 : // theoretical time cost of Recursive HD
662 : // the (RHD_FACTOR_TWO * moduleNum_ - RHD_FACTOR_ONE) / moduleNum_ is equal to 1 + (moduleNum_ -1) / moduleNum_
663 0 : hdCost = ceil(log2(moduleNum_)) * delay +
664 0 : static_cast<double>(RHD_FACTOR_TWO * moduleNum_ - RHD_FACTOR_ONE) / moduleNum_ *
665 0 : sendCurSize * userRankSize_ / bandWidth *
666 : SECOND2MICROSECOND;
667 : }
668 0 : algType = (hdCost < ringCost) ? AlgTypeLevel1::ALG_LEVEL1_HD : AlgTypeLevel1::ALG_LEVEL1_RING;
669 0 : return HCCL_SUCCESS;
670 : }
671 :
672 0 : HcclResult CollAlgOperator::SelectAlgoTypeForAllReduce(float delay, u64 curSize, float bandWidth,
673 : AlgTypeLevel1 &algType)
674 : {
675 0 : auto steps = moduleNum_ - 1;
676 : // theoretical time cost of Ring
677 0 : double ringCost = DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) * delay +
678 0 : DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) / moduleNum_ *
679 0 : curSize / deviceNumPerAggregation_ / bandWidth *
680 : SECOND2MICROSECOND;
681 :
682 : // theoretical time cost of NHR
683 0 : double nhrCost = NHR_FACTOR_TWO * ceil(log2(moduleNum_)) * delay +
684 0 : NHR_FACTOR_TWO * static_cast<double>(moduleNum_ - 1) / moduleNum_ *
685 0 : curSize / deviceNumPerAggregation_ / bandWidth * SECOND2MICROSECOND;
686 :
687 : // compare costs between NHR and Ring, if same cost, Ring > NHR > HD
688 0 : algType = (nhrCost < ringCost) ? AlgTypeLevel1::ALG_LEVEL1_NHR : AlgTypeLevel1::ALG_LEVEL1_RING;
689 0 : double interMinCost = min(nhrCost, ringCost);
690 :
691 : // theoretical time cost of HD/RHD
692 0 : double hdCost = 0.0;
693 0 : if ((moduleNum_ & (moduleNum_ - 1)) == 0) {
694 : // theoretical time cost of HD
695 0 : hdCost = DOUBLE_SUB_HCCLCMD * log2(moduleNum_) * delay +
696 0 : DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) / moduleNum_ *
697 0 : curSize / deviceNumPerAggregation_ / bandWidth *
698 : SECOND2MICROSECOND;
699 : } else {
700 : // theoretical time cost of Recursive HD
701 : // the (RHD_FACTOR_TWO * moduleNum_ - RHD_FACTOR_ONE) / moduleNum_ is equal to 1 + (moduleNum_ -1) / moduleNum_
702 0 : hdCost = DOUBLE_SUB_HCCLCMD * ceil(log2(moduleNum_)) * delay +
703 0 : DOUBLE_SUB_HCCLCMD * static_cast<double>(RHD_FACTOR_TWO * moduleNum_ - RHD_FACTOR_ONE) / moduleNum_ *
704 0 : curSize / deviceNumPerAggregation_ / bandWidth *
705 : SECOND2MICROSECOND;
706 : }
707 :
708 : // compare cost among NHR, HD and Ring
709 0 : algType = (hdCost < interMinCost) ? AlgTypeLevel1::ALG_LEVEL1_HD : algType;
710 0 : return HCCL_SUCCESS;
711 : }
712 :
713 0 : HcclResult CollAlgOperator::SelectAlgoTypeForBroadcast(float delay, u64 curSize, float bandWidth,
714 : AlgTypeLevel1 &algType)
715 : {
716 0 : auto steps = moduleNum_ - 1;
717 : // theoretical time cost of Ring
718 0 : double ringCost = DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) * delay +
719 0 : DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) / moduleNum_ *
720 0 : curSize / deviceNumPerAggregation_ / bandWidth *
721 : SECOND2MICROSECOND;
722 0 : double hdCost = 0.0;
723 0 : if ((moduleNum_ & (moduleNum_ - 1)) == 0) {
724 : // theoretical time cost of HD
725 0 : hdCost = DOUBLE_SUB_HCCLCMD * log2(moduleNum_) * delay +
726 0 : DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) / moduleNum_ *
727 0 : curSize / deviceNumPerAggregation_ / bandWidth
728 0 : * SECOND2MICROSECOND;
729 : } else {
730 : // theoretical time cost of Recursive HD
731 : // rhd-broadcast = scatter + allgather + copy
732 0 : hdCost = (COPY_TIME_IN_RHD + DOUBLE_SUB_HCCLCMD * floor(log2(moduleNum_))) * delay +
733 0 : (COPY_TIME_IN_RHD + DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) / moduleNum_) *
734 0 : curSize / deviceNumPerAggregation_ / bandWidth *
735 : SECOND2MICROSECOND;
736 : }
737 0 : algType = (hdCost < ringCost) ? AlgTypeLevel1::ALG_LEVEL1_HD : AlgTypeLevel1::ALG_LEVEL1_RING;
738 0 : return HCCL_SUCCESS;
739 : }
740 :
741 0 : HcclResult CollAlgOperator::SelectAlgoTypeForReduce(float delay, u64 curSize, float bandWidth,
742 : AlgTypeLevel1 &algType)
743 : {
744 0 : HCCL_DEBUG("[CollAlgOperator]SelectAlgoTypeForReduce start");
745 0 : auto steps = moduleNum_ - 1;
746 : // theoretical time cost of Ring
747 0 : double ringCost = DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) * delay +
748 0 : DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) / moduleNum_ *
749 0 : curSize / deviceNumPerAggregation_ / bandWidth *
750 : SECOND2MICROSECOND;
751 0 : double hdCost = 0.0;
752 0 : if ((moduleNum_ & (moduleNum_ - 1)) == 0) {
753 : // theoretical time cost of HD
754 0 : hdCost = DOUBLE_SUB_HCCLCMD * log2(moduleNum_) * delay +
755 0 : DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) / moduleNum_ *
756 0 : curSize / deviceNumPerAggregation_ / bandWidth *
757 : SECOND2MICROSECOND;
758 : } else {
759 : // theoretical time cost of Recursive HD
760 : // rhd-broadcast = reducescatter + gather + copy
761 0 : hdCost = (COPY_TIME_IN_RHD + DOUBLE_SUB_HCCLCMD * floor(log2(moduleNum_))) * delay +
762 0 : (COPY_TIME_IN_RHD + DOUBLE_SUB_HCCLCMD * static_cast<double>(steps) / moduleNum_) *
763 0 : curSize / deviceNumPerAggregation_ / bandWidth *
764 : SECOND2MICROSECOND;
765 : }
766 0 : algType = (hdCost < ringCost) ? AlgTypeLevel1::ALG_LEVEL1_HD : AlgTypeLevel1::ALG_LEVEL1_RING;
767 0 : return HCCL_SUCCESS;
768 : }
769 :
770 24 : AlgType CollAlgOperator::GetAlgType()
771 : {
772 24 : return algType_;
773 : }
774 :
775 2 : bool CollAlgOperator::Is2U2PInfer()
776 : {
777 0 : return ((deviceNumPerAggregation_ == HCCL_DEVICE_NUM_TWO) && (serverNum_ == 1) &&
778 2 : (deviceType_ == DevType::DEV_TYPE_910B) && (meshAggregationRankSize_ == HCCL_DEVICE_NUM_TWO) &&
779 2 : (pairLinkCounter_[static_cast<u32>(LinkTypeInServer::HCCS_TYPE)] == 0));
780 : }
781 :
782 30 : bool CollAlgOperator::Is910BSingleMesh()
783 : {
784 1 : bool isMeshTopo = topoType_ == TopoType::TOPO_TYPE_NP_MESH || topoType_ == TopoType::TOPO_TYPE_4P_MESH ||
785 31 : topoType_ == TopoType::TOPO_TYPE_2P_MESH || topoType_ == TopoType::TOPO_TYPE_1P_MESH;
786 :
787 : bool isSingleMesh =
788 30 : (deviceType_ == DevType::DEV_TYPE_910B) && (isMeshTopo || Is2U2PInfer()) && (userRankSize_ != 1);
789 30 : return isSingleMesh;
790 : }
791 :
792 1 : bool CollAlgOperator::NeedCreateSingleMeshPlane(const bool isInlineReduce)
793 : {
794 : // 910B 图模式非确定计算，inlineReduce使能，MESH拓扑场景下，创建一个mesh平面
795 1 : bool meshSinglePlane = Is910BSingleMesh() && topoMatcher_->GetDeterministicConfig() == DETERMINISTIC_DISABLE &&
796 1 : isInlineReduce && (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
797 :
798 1 : return meshSinglePlane;
799 : }
800 :
801 26 : bool CollAlgOperator::SingleMeshInlineReduce(void *inputPtr, void *outputPtr, HcclDataType dataType, HcclReduceOp op)
802 : {
803 26 : bool isInlineReduce = IsSupportSDMAReduce(inputPtr, outputPtr, dataType, op);
804 28 : bool singleMeshInlineReduce = Is910BSingleMesh() && isInlineReduce && isSingleMeshAggregation_;
805 28 : return singleMeshInlineReduce;
806 : }
807 :
808 3 : bool CollAlgOperator::IsMultiMeshInlineReduce(void *inputPtr, void *outputPtr, HcclDataType dataType, HcclReduceOp op)
809 : {
810 0 : bool isMeshTopo = topoType_ == TopoType::TOPO_TYPE_NP_MESH || topoType_ == TopoType::TOPO_TYPE_4P_MESH ||
811 3 : topoType_ == TopoType::TOPO_TYPE_2P_MESH || topoType_ == TopoType::TOPO_TYPE_1P_MESH;
812 :
813 3 : bool isInlineReduce = IsSupportSDMAReduce(inputPtr, outputPtr, dataType, op);
814 3 : bool isRdmaReduce = IsSupportRDMAReduce(dataType, op);
815 3 : bool multiMeshInlineReduce = (deviceType_ == DevType::DEV_TYPE_910B) &&
816 6 : isMeshTopo && isInlineReduce && isRdmaReduce && (!isSingleMeshAggregation_);
817 3 : return multiMeshInlineReduce;
818 : }
819 :
820 0 : void CollAlgOperator::SetLegacyHcclImpl(std::unique_ptr<hcclImpl> &impl)
821 : {
822 0 : hcclImpl_ = impl.get();
823 0 : return;
824 : }
825 :
826 8 : HcclResult CollAlgOperator::SetRetryEnable(bool retryEnable)
827 : {
828 8 : retryEnable_ = retryEnable;
829 8 : return HCCL_SUCCESS;
830 : }
831 :
832 7 : HcclResult CollAlgOperator::SetAivClearEnable(bool aivClearEnable)
833 : {
834 7 : aivClearEnable_ = aivClearEnable;
835 7 : return HCCL_SUCCESS;
836 : }
837 :
838 0 : HcclResult CollAlgOperator::SetAlgOpContext(AlgOpContext algOpContext)
839 : {
840 0 : algOpContext_ = algOpContext;
841 0 : return HCCL_SUCCESS;
842 : }
843 :
844 29 : bool CollAlgOperator::SupportRetryWithInplaceCheck(
845 : const HcclCMDType &opType, OpParam ¶m, std::string& algName, u8 &isInplaceStatus,
846 : InplaceSupportRetryStatus &inPlaceSupportRetryStatus)
847 : {
848 : // 不支持inplace的通信算子重执行
849 29 : if (IsHcclOpInplace(opType, param, userRank_, userRankSize_, isInplaceStatus)) {
850 13 : void *commInputPtr = nullptr;
851 13 : u64 commInputSize = 0;
852 21 : CHK_RET(cclBufferManager_.GetInCCLbuffer(commInputPtr, commInputSize));
853 13 : if(!FitRetryConditionforInPlaceOp(opType, param, algName, commInputSize, userRankSize_,
854 13 : retryEnable_, inPlaceSupportRetryStatus)) {
855 8 : HCCL_DEBUG("[CollAlgOperator][OpRetry][AICPU]hccl aicpu can not retry, opType[%s], inputPtr[%p], "
856 : "outputPtr[%p].",
857 : GetCMDTypeEnumStr(opType).c_str(), param.inputPtr, param.outputPtr);
858 8 : return false;
859 : }
860 : }
861 : // true 存在两种情况：
862 : // 1. 非inplace场景
863 : // 2. 是inplace但同时符合retry条件的场景
864 21 : return true;
865 : }
866 :
867 21 : HcclResult CollAlgOperator::GetNumBlocks(u32& numBlocks){
868 21 : CHK_SMART_PTR_NULL(executor_);
869 21 : return executor_->GetNumBlocks(numBlocks);
870 : }
871 :
872 5 : HcclResult CollAlgOperator::SetNumBlocks(const u32& numBlocks){
873 5 : CHK_SMART_PTR_NULL(executor_);
874 5 : return executor_->SetNumBlocks(numBlocks);
875 : }
876 :
877 2 : HcclResult CollAlgOperator::GetCache(HcclCacheInfo& cacheInfo){
878 2 : CHK_SMART_PTR_NULL(executor_);
879 2 : return executor_->GetCache(cacheInfo);
880 : }
881 :
882 24 : HcclResult CollAlgOperator::SetOpCounter(const OpCounterInfo& opCounter)
883 : {
884 24 : opCounter_ = opCounter;
885 24 : return HCCL_SUCCESS;
886 : }
887 :
888 0 : HcclResult CollAlgOperator::SetRmaInfo(void* rmaInfo)
889 : {
890 0 : CHK_SMART_PTR_NULL(executor_);
891 0 : CHK_PTR_NULL(rmaInfo);
892 0 : return executor_->SetRmaInfo(rmaInfo);
893 : }
894 :
895 9 : HcclResult CollAlgOperator::SelectAlgforAHC(u64 dataSize, AHCOpType ahcOpType)
896 : {
897 9 : if (multiModuleDiffDeviceNumMode_) {
898 0 : return HCCL_SUCCESS;
899 : }
900 :
901 9 : bool isAHCWholeConfig = (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_RESERVED &&
902 0 : (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
903 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE));
904 :
905 9 : CommPlane ahcSubGroupLevel = COMM_LEVEL1_AHC;
906 9 : if (isAHCWholeConfig) {
907 0 : if (deviceType_ != DevType::DEV_TYPE_910_93) {
908 0 : ahcSubGroupLevel = COMM_COMBINE;
909 : } else {
910 0 : ahcSubGroupLevel = COMM_COMBINE_ORDER;
911 : }
912 9 : } else if (deviceType_ != DevType::DEV_TYPE_910_93) {
913 0 : HCCL_DEBUG("[AHCAlgSelect] hccl algorithm: 910B not support level1 ahc, return ERROR.");
914 0 : return HCCL_E_PARA;
915 : }
916 :
917 9 : HCCL_INFO("[SelectAlgforAHC] ahcOpType[%u] isAHCWholeConfig[%u] AHClevel[%u] algType_[%u] deviceType_[%u]",
918 : ahcOpType, isAHCWholeConfig, ahcSubGroupLevel, algType_.algoLevel1 , deviceType_);
919 :
920 : AlgTypeLevel1 algTypeLevel1;
921 :
922 9 : std::vector<std::vector<std::vector<u32>>> globalSubGroups;
923 9 : std::map<AHCConcOpType, TemplateType> ahcAlgOption;
924 9 : CHK_RET(topoMatcher_->GetGlobalSubGroups(ahcSubGroupLevel, globalSubGroups));
925 9 : topoMatcher_->GetAHCAlgOption(ahcAlgOption);
926 :
927 9 : AHCAlgSelectParam ahcAlgSelectParam;
928 9 : ahcAlgSelectParam.opType = ahcOpType;
929 9 : ahcAlgSelectParam.dataSize = dataSize;
930 :
931 : //AHC 封装算法选择逻辑
932 9 : CHK_RET(AHCAlgSelect(algTypeLevel1, globalSubGroups, ahcAlgOption, ahcAlgSelectParam));
933 :
934 9 : topoMatcher_->SetAHCAlgOption(ahcAlgOption);
935 :
936 9 : auto iter = HCCL_ALGO_LEVEL1_NAME_MAP.find(algTypeLevel1);
937 9 : CHK_PRT_RET(iter == HCCL_ALGO_LEVEL1_NAME_MAP.end(),
938 : HCCL_ERROR("[AHCAlgSelect] level1: algType_[%u] is invalid.", algTypeLevel1),
939 : HCCL_E_INTERNAL);
940 :
941 : // 支持 AHC 自适应调节为 BROKE 类型
942 9 : if (algType_.algoLevel1 != algTypeLevel1 && algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
943 8 : algType_.algoLevel1 = algTypeLevel1;
944 : }
945 :
946 9 : HCCL_INFO("[AHCAlgSelect] hccl algorithm: there are %u server(%u module) in level1, using %s algo",
947 : serverNum_, moduleNum_, iter->second.c_str());
948 :
949 9 : return HCCL_SUCCESS;
950 9 : }
951 :
952 9 : HcclResult CollAlgOperator::AHCAlgSelect(AlgTypeLevel1 &algType, std::vector<std::vector<std::vector<u32>>> &globalSubGroups,
953 : std::map<AHCConcOpType, TemplateType> &ahcAlgOption, AHCAlgSelectParam &ahcAlgSelectParam)
954 : {
955 : // globalSubGroups 参数检查
956 9 : CHK_RET(CommAHCBaseInfo::CheckGlobalGroups(globalSubGroups));
957 :
958 9 : bool isAHCType = false;
959 9 : u32 minSubGroupSize = globalSubGroups[0][0].size();
960 9 : u32 maxSubGroupSize = globalSubGroups[0][0].size();
961 45 : for (u32 i = 1; i < globalSubGroups[0].size(); ++i) {
962 36 : if (globalSubGroups[0][i].size() < minSubGroupSize) {
963 0 : minSubGroupSize = globalSubGroups[0][i].size();
964 : }
965 36 : if (globalSubGroups[0][i].size() > maxSubGroupSize) {
966 0 : maxSubGroupSize = globalSubGroups[0][i].size();
967 : }
968 : }
969 54 : for (u32 i = 0; i < globalSubGroups[0].size(); ++i) {
970 45 : if (globalSubGroups[0][i].size()!= minSubGroupSize) {
971 0 : isAHCType = true;
972 0 : break;
973 : }
974 : }
975 :
976 : //多平面 reduce scatter 和 all gather 算子，强制写死成BROKE类型
977 9 : if (deviceNumPerServer_ != 1 && ahcAlgSelectParam.opType != AHCOpType::AHC_OP_TYPE_ALLREDUCE) {
978 0 : isAHCType = false;
979 : }
980 :
981 : //add AHC Conc Type logic here, modify init Type depend on the input para
982 9 : CHK_RET(AHCAlgOptionSelect(algType, globalSubGroups, ahcAlgOption, ahcAlgSelectParam));
983 :
984 9 : if (ahcAlgSelectParam.enableAlgAutoSelect == false) { // 关闭算法自适应功能时，默认设置AHC算法
985 0 : algType = AlgTypeLevel1::ALG_LEVEL1_AHC;
986 0 : return HCCL_SUCCESS;
987 : }
988 :
989 9 : if (isAHCType) {
990 0 : algType = AlgTypeLevel1::ALG_LEVEL1_AHC; // 设置为 AHC 类型
991 : } else {
992 9 : algType = AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE; // 设置为 BROKE 类型
993 : }
994 :
995 9 : HCCL_DEBUG("[AHCAlgSelect] end minSubGroupSize = %u maxSubGroupSize = %u isAHCType = %u",
996 : minSubGroupSize, maxSubGroupSize, isAHCType);
997 :
998 9 : return HCCL_SUCCESS;
999 : }
1000 :
1001 9 : HcclResult CollAlgOperator::AHCAlgOptionSelect(const AlgTypeLevel1 &algType, std::vector<std::vector<std::vector<u32>>> &globalSubGroups,
1002 : std::map<AHCConcOpType, TemplateType> &ahcAlgOption, const AHCAlgSelectParam &ahcAlgSelectParam)
1003 : {
1004 : (void) algType;
1005 : (void) ahcAlgSelectParam;
1006 9 : AHCConcOpType ahcConcOpType;
1007 : //一层组间拼接时，分组数大于设定阈值则修改默认算法为NHR
1008 9 : if(globalSubGroups[0].size() <= AHC_LEVEL0_GROUP_SIZE_THRESHOLD ) {
1009 3 : HCCL_DEBUG("[AHCAlgSelect] conc inter select type RING ");
1010 3 : ahcConcOpType = {AHCLevel::AHC_LEVEL_0, ConcType::CONC_INTER, AHCOpType::AHC_OP_TYPE_REDUCE_SCATTER};
1011 3 : ahcAlgOption[ahcConcOpType] = TemplateType::TEMPLATE_REDUCESCATTER_RING;
1012 :
1013 3 : ahcConcOpType = {AHCLevel::AHC_LEVEL_0, ConcType::CONC_INTER, AHCOpType::AHC_OP_TYPE_ALLREDUCE};
1014 3 : ahcAlgOption[ahcConcOpType] = TemplateType::TEMPLATE_ALL_REDUCE_RING;
1015 :
1016 3 : ahcConcOpType = {AHCLevel::AHC_LEVEL_0, ConcType::CONC_INTER, AHCOpType::AHC_OP_TYPE_ALLGATHER};
1017 3 : ahcAlgOption[ahcConcOpType] = TemplateType::TEMPLATE_ALL_GATHER_RING;
1018 : } else {
1019 6 : HCCL_DEBUG("[AHCAlgSelect] conc inter select type NHR ");
1020 6 : ahcConcOpType = {AHCLevel::AHC_LEVEL_0, ConcType::CONC_INTER, AHCOpType::AHC_OP_TYPE_REDUCE_SCATTER};
1021 6 : ahcAlgOption[ahcConcOpType] = TemplateType::TEMPLATE_REDUCESCATTER_NHR;
1022 :
1023 6 : ahcConcOpType = {AHCLevel::AHC_LEVEL_0, ConcType::CONC_INTER, AHCOpType::AHC_OP_TYPE_ALLREDUCE};
1024 6 : ahcAlgOption[ahcConcOpType] = TemplateType::TEMPLATE_ALL_REDUCE_NHR;
1025 :
1026 6 : ahcConcOpType = {AHCLevel::AHC_LEVEL_0, ConcType::CONC_INTER, AHCOpType::AHC_OP_TYPE_ALLGATHER};
1027 6 : ahcAlgOption[ahcConcOpType] = TemplateType::TEMPLATE_ALL_GATHER_NHR;
1028 : }
1029 9 : return HCCL_SUCCESS;
1030 : }
1031 :
1032 0 : u32 CollAlgOperator::CalcOptimalIntraRingsize(u64 count, HcclDataType dataType, HcclCMDType opType)
1033 : {
1034 0 : if (!topoMatcher_->GetARSFlag()) return 0;
1035 :
1036 0 : u32 level0RankSize = topoMatcher_->GetCommPlaneRanks(COMM_LEVEL0)[0].size();
1037 0 : u32 rankSizeInSuperPod = topoMatcher_->GetCommPlaneRanks(COMM_ARS)[0].size();
1038 0 : u32 perDataSize = 0;
1039 0 : CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
1040 : // 不支持 ARS 或环内卡数不是 2 的倍数
1041 0 : u32 level0RingSize = 1;
1042 0 : if (!isARSDoubleRing_ || (level0RankSize % FACTOR_TWO != 0)) {
1043 0 : HCCL_INFO("not Support ARS doubleRing, level0RingSize:[%u], level0RankSize[%u].", level0RingSize, level0RankSize);
1044 0 : return level0RingSize;
1045 : }
1046 : // --- 1. 带宽 & 基本参数 ---
1047 : float bwHCCS, bwHBM, bwSIO;
1048 0 : constexpr u32 level0 = 0;
1049 0 : constexpr u32 level2 = 2;
1050 0 : constexpr u32 level3 = 3;
1051 0 : CHK_RET(GetBandWidthPerNPU(level0, userRankSize_, deviceNumPerAggregation_, bwHCCS));
1052 0 : CHK_RET(GetBandWidthPerNPU(level2, userRankSize_, deviceNumPerAggregation_, bwHBM));
1053 0 : CHK_RET(GetBandWidthPerNPU(level3, userRankSize_, deviceNumPerAggregation_, bwSIO));
1054 0 : float latency = BASE_COMM_LATENCY / MULTIPLIER_MS2US; // ms
1055 : // --- 2. 数据总量 (GB) ---
1056 0 : float baseSizeGB = static_cast<double>(count) * perDataSize / GB2B;
1057 0 : float totalSize = baseSizeGB;
1058 0 : HCCL_INFO("CalcOptimalIntraRingsize: count[%u], totalSize:[%lf]GB, perDataSize[%u].", count, totalSize, perDataSize);
1059 0 : if (opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER || opType == HcclCMDType::HCCL_CMD_ALLGATHER) {
1060 0 : totalSize *= rankSizeInSuperPod;
1061 : }
1062 : // --- 3. 枚举可能的环大小 ---
1063 0 : std::vector<u32> factors;
1064 0 : for (u32 i = 1; i <= rankSizeInSuperPod / i; ++i) {
1065 0 : if (rankSizeInSuperPod % i == 0) {
1066 0 : if (i != rankSizeInSuperPod / i) {
1067 0 : factors.push_back(rankSizeInSuperPod / i);
1068 : }
1069 0 : factors.push_back(i);
1070 : }
1071 : }
1072 0 : std::sort(factors.begin(), factors.end());
1073 : // --- 4. 计算最优带宽 ---
1074 0 : double maxBwARS = 0.0;
1075 0 : for (u32 N1 : factors) {
1076 :
1077 : // 传输时延 (ms)
1078 : double latencyIntra;
1079 0 : if (N1 == FACTOR_TWO) {
1080 0 : latencyIntra = totalSize * MULTIPLIER_S2MS / FACTOR_TWO / bwSIO;
1081 0 : } else if ((N1 % FACTOR_TWO == 0) && (N1 > FACTOR_TWO)) {
1082 0 : latencyIntra = (N1 - 1) * totalSize * MULTIPLIER_S2MS / N1 / bwHCCS / FACTOR_TWO;
1083 : } else {
1084 0 : latencyIntra = (N1 - 1) * totalSize * MULTIPLIER_S2MS / N1 / bwHCCS;
1085 : }
1086 0 : u32 N2 = rankSizeInSuperPod / N1;
1087 : // 静态时延 (ms)
1088 0 : double interStep = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) ? (N2 - 1) : log2(N2);
1089 0 : double latencyStep = (interStep + (N1 - 1)) * latency;
1090 0 : double latencyInter = (N2 - 1) * totalSize * MULTIPLIER_S2MS / N1 / N2 / bwHCCS;
1091 : // HBM 拷贝时延 (ms)
1092 0 : double latencyCopy = totalSize * MULTIPLIER_S2MS / bwHBM;
1093 0 : u8 mul = (HcclCMDType::HCCL_CMD_ALLREDUCE == opType) ? FACTOR_TWO : 1;
1094 0 : double timeCost = mul * (latencyStep + latencyIntra + latencyInter) + latencyCopy;
1095 0 : double bwARS = totalSize / timeCost; //GB/ms
1096 0 : if (bwARS > maxBwARS) {
1097 0 : level0RingSize = N1;
1098 0 : maxBwARS = bwARS;
1099 : }
1100 : }
1101 0 : HCCL_INFO("level0RingSize:[%u], level0RankSize[%u], totalSize:[%lf]GB", level0RingSize, level0RankSize, totalSize);
1102 0 : return level0RingSize;
1103 0 : }
1104 :
1105 20 : bool CollAlgOperator::IsNeedStrictMode(const OpParam& param)
1106 : {
1107 20 : bool isStrictMode = (topoMatcher_->GetDeterministicConfig() == DETERMINISTIC_STRICT)
1108 0 : && (param.DataDes.dataType == HCCL_DATA_TYPE_FP16 || param.DataDes.dataType == HCCL_DATA_TYPE_FP32 ||
1109 0 : param.DataDes.dataType == HCCL_DATA_TYPE_BFP16 || param.DataDes.dataType == HCCL_DATA_TYPE_FP64)
1110 0 : && (param.reduceType == HCCL_REDUCE_SUM || param.reduceType == HCCL_REDUCE_PROD)
1111 20 : && userRankSize_ >= MIN_STRICT_RANK_NUM;
1112 :
1113 20 : return isStrictMode;
1114 : }
1115 :
1116 0 : bool CollAlgOperator::CheckStrictCondition(const OpParam& param) const
1117 : {
1118 0 : CHK_PRT_RET(multiModuleDiffDeviceNumMode_ || multiSuperPodDiffDeviceNumMode_ || multiSuperPodDiffServerNumMode_,
1119 : HCCL_ERROR("[CollAlgOperator][CheckStrictCondition] DETERMINISTIC_STRICT mode not support asymmetrical topo."),
1120 : false);
1121 :
1122 0 : CHK_PRT_RET(param.reduceType == HCCL_REDUCE_PROD,
1123 : HCCL_ERROR("[CollAlgOperator][CheckStrictCondition] DETERMINISTIC_STRICT mode not support PROD."),
1124 : false);
1125 :
1126 0 : CHK_PRT_RET(param.DataDes.dataType == HCCL_DATA_TYPE_FP64,
1127 : HCCL_ERROR("[CollAlgOperator][CheckStrictCondition] DETERMINISTIC_STRICT mode not support FP64."),
1128 : false);
1129 :
1130 0 : CHK_PRT_RET(GetExternalInputInterHccsDisable(),
1131 : HCCL_ERROR("[CollAlgOperator][CheckStrictCondition] DETERMINISTIC_STRICT mode not support HCCS disable."),
1132 : false);
1133 :
1134 0 : return true;
1135 : }
1136 :
1137 : } // namespace hccl
|
