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_all_reduce_ring_for_910_93_executor.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 :
16 5 : CollAllReduceRingFor91093Executor::CollAllReduceRingFor91093Executor(const HcclDispatcher dispatcher,
17 5 : std::unique_ptr<TopoMatcher> &topoMatcher)
18 5 : : CollAllReduceExecutor(dispatcher, topoMatcher)
19 : {
20 5 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
21 4 : DMAReduceFlag_ = true;
22 : } else {
23 1 : DMAReduceFlag_ = false;
24 : }
25 5 : desc_.deterministic = 1;
26 5 : desc_.level1SupportedAlgos = {
27 : AlgTypeLevel1::ALG_LEVEL1_NHR,
28 : AlgTypeLevel1::ALG_LEVEL1_NB,
29 : AlgTypeLevel1::ALG_LEVEL1_RING,
30 : AlgTypeLevel1::ALG_LEVEL1_AHC,
31 : AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE
32 5 : };
33 5 : desc_.level2SupportedAlgos = {
34 : AlgTypeLevel2::ALG_LEVEL2_NHR,
35 : AlgTypeLevel2::ALG_LEVEL2_NB,
36 : AlgTypeLevel2::ALG_LEVEL2_RING,
37 : AlgTypeLevel2::ALG_LEVEL2_HD
38 5 : };
39 5 : }
40 :
41 5 : HcclResult CollAllReduceRingFor91093Executor::CalcStreamNum(u32& streamNum)
42 : {
43 5 : u32 totalStreamNum = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
44 : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
45 5 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
46 4 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
47 : }
48 5 : streamNum = totalStreamNum - 1;
49 5 : HCCL_INFO("[CollAllReduceRingFor91093Executor][CalcStreamNum] tag[%s] streamNum_[%u].",
50 : tag_.c_str(), streamNum);
51 5 : return HCCL_SUCCESS;
52 : }
53 :
54 5 : HcclResult CollAllReduceRingFor91093Executor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
55 : {
56 5 : TransportMemType inputType = TransportMemType::RESERVED;
57 5 : TransportMemType outputType = TransportMemType::RESERVED;
58 5 : CHK_RET(CalcTransportMemType(inputType, outputType));
59 5 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
60 5 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
61 5 : CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
62 5 : return HCCL_SUCCESS;
63 : }
64 :
65 5 : HcclResult CollAllReduceRingFor91093Executor::CalcTransportMemType(TransportMemType &inputType,
66 : TransportMemType &outputType)
67 : {
68 5 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
69 4 : inputType = TransportMemType::CCL_INPUT;
70 4 : outputType = TransportMemType::CCL_OUTPUT;
71 : } else {
72 1 : inputType = TransportMemType::PARAM_INPUT;
73 1 : outputType = TransportMemType::PARAM_OUTPUT;
74 : }
75 5 : HCCL_INFO("[CollAllReduceRingFor91093Executor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d].",
76 : tag_.c_str(), inputType, outputType);
77 5 : return HCCL_SUCCESS;
78 : }
79 :
80 5 : HcclResult CollAllReduceRingFor91093Executor::CalcLevel0CommInfo(TransportMemType inputType,
81 : TransportMemType outputType,
82 : std::vector<LevelNSubCommTransport>& opTransport)
83 : {
84 5 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
85 5 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
86 5 : return HCCL_SUCCESS;
87 5 : }
88 :
89 16 : bool CollAllReduceRingFor91093Executor::IsSmallData(const u64 totalSize, const u64 curSize)
90 : {
91 : (void)totalSize;
92 16 : bool smallData = IsAllReduceSmallData(curSize);
93 16 : return smallData;
94 : }
95 :
96 16 : bool CollAllReduceRingFor91093Executor::IsHugeData(const u64 curSize)
97 : {
98 16 : bool hugeData = curSize / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE ||
99 : curSize > SDMA_SEND_MAX_SIZE;
100 16 : return hugeData;
101 : }
102 :
103 5 : HcclResult CollAllReduceRingFor91093Executor::CalcLevel2CommInfo(TransportMemType inputType,
104 : TransportMemType outputType,
105 : std::vector<LevelNSubCommTransport>& opTransport)
106 : {
107 5 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
108 5 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
109 0 : HCCL_INFO("[CollAllReduceRingFor91093Executor][CalcLevel2CommInfo] select AHC bypass level2 comm calculate");
110 0 : return HCCL_SUCCESS;
111 : }
112 :
113 5 : CommParaInfo commParaLevel2(COMM_LEVEL2, CommType::COMM_TAG_MAX);
114 5 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
115 1 : commParaLevel2.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
116 1 : HCCL_INFO("[%s]Calc NHRCommInfo", __func__);
117 4 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
118 0 : commParaLevel2.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
119 0 : HCCL_INFO("[%s]Calc NBCommInfo", __func__);
120 4 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_RING) {
121 4 : commParaLevel2.commType = CommType::COMM_TAG_RING_INNER;
122 4 : HCCL_INFO("[%s]Calc RingCommInfo", __func__);
123 : } else {
124 0 : commParaLevel2.commType = CommType::COMM_TAG_HALVING_DOUBLING;
125 0 : HCCL_INFO("[%s]Calc HDCommInfo", __func__);
126 : }
127 5 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel2, opTransport[COMM_LEVEL2], inputType, outputType));
128 5 : return HCCL_SUCCESS;
129 5 : }
130 :
131 0 : HcclResult CollAllReduceRingFor91093Executor::RunIntraSeverReduceScatter(
132 : const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
133 : const u64 count, const HcclDataType &dataType, const HcclReduceOp &reductionOp,
134 : const std::vector<std::vector<Slice>> &multRingsSliceZero, const Stream &stream, s32 profStage,
135 : const u64 baseOffset, const HcomCollOpInfo *opInfo,
136 : const std::vector<std::vector<Slice>> &multRingsUserMemSlice, const bool disableDMAReduce)
137 : {
138 0 : CHK_RET(MultiRingReduceScatter(tag, inputMem, outputMem, count, dataType, reductionOp,
139 : multRingsSliceZero, stream, profStage, baseOffset, opInfo, multRingsUserMemSlice, logicalLevel0plane_));
140 0 : return HCCL_SUCCESS;
141 : }
142 :
143 0 : HcclResult CollAllReduceRingFor91093Executor::RunIntraSeverAllGather(
144 : const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
145 : const u64 count, const HcclDataType &dataType, const std::vector<std::vector<Slice>> &multRingsSliceZero,
146 : const Stream &stream, s32 profStage, const u64 baseOffset, const HcomCollOpInfo *opInfo,
147 : const std::vector<std::vector<Slice>> &multRingsUserMemSlice)
148 : {
149 0 : CHK_RET(MultiRingAllGather(tag, inputMem, outputMem, count, dataType,
150 : multRingsSliceZero, stream, profStage, baseOffset, opInfo, multRingsUserMemSlice, logicalLevel0plane_));
151 0 : return HCCL_SUCCESS;
152 : }
153 :
154 16 : HcclResult CollAllReduceRingFor91093Executor::GetLevelCommInfo()
155 : {
156 16 : logicalLevel0plane_ = COMM_LEVEL0;
157 16 : CHK_RET(CheckCommSize(logicalLevel0plane_, COMM_INDEX_0 + 1));
158 16 : logicalLevel0CommInfo_ = GetSubCommInfo(logicalLevel0plane_, COMM_INDEX_0);
159 16 : u32 commIndex = logicalLevel0CommInfo_.localRank;
160 32 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
161 16 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
162 16 : logicalLevel1plane_ = isSelectAHC ? COMM_LEVEL1_AHC : COMM_LEVEL1;
163 16 : CHK_RET(CheckCommSize(logicalLevel1plane_, commIndex + 1));
164 16 : logicalLevel1CommInfo_ = GetSubCommInfo(logicalLevel1plane_, commIndex);
165 16 : return HCCL_SUCCESS;
166 : }
167 :
168 0 : HcclResult CollAllReduceRingFor91093Executor::PrepareARSLevel1CommInfo(u32 &segmentIdx, u32 &commIndex, u64 &hdSize,
169 : const SubCommInfo &commInfo,
170 : const std::vector<std::vector<Slice>> &multRingsSliceZero,
171 : const std::string &tag, const std::vector<u32>& nicList)
172 : {
173 0 : segmentIdx = logicalLevel0CommInfo_.localRank;
174 0 : commIndex = logicalLevel0CommInfo_.localRank;
175 0 : CHK_PRT_RET(multRingsSliceZero.empty(), HCCL_ERROR("[Prepare][Level1CommInfo]slice map is empty"), HCCL_E_PARA);
176 :
177 0 : if (multRingsSliceZero.size() > 1) {
178 0 : std::vector<u32>::const_iterator iterNic = std::find(nicList.begin(),
179 0 : nicList.end(), logicalLevel0CommInfo_.localRank);
180 0 : if (iterNic != nicList.end()) { // 如果当前rank为通信网口
181 0 : u32 nicIdx = std::distance(nicList.begin(), iterNic);
182 0 : std::unique_lock<std::mutex> lock(nicSendSizeListLock_);
183 0 : auto iter = nicSendSizeList_.find(tag);
184 0 : CHK_PRT_RET(iter == nicSendSizeList_.end(), HCCL_ERROR("[Prepare][Level1CommInfo]find tag[%s] in "\
185 : "nicSendSizeList_ failed", tag.c_str()), HCCL_E_INTERNAL);
186 0 : CHK_PRT_RET(nicIdx >= iter->second.size(), HCCL_ERROR("[Prepare][Level1CommInfo]tag[%s] nicIdx[%u] "\
187 : "invalid, expect less than %zu", tag.c_str(), nicIdx, iter->second.size()), HCCL_E_INTERNAL);
188 0 : hdSize = iter->second[nicIdx]; // 通过nicSendSizeList_得到该网口传输数据量
189 0 : u32 ringRanks = multRingsSliceZero[0].size(); // 获取单个 ring 上设备的数量
190 0 : segmentIdx = ringRanks / nicList.size() * nicIdx; // 通过网口位置得到该网口传输数据的起始位置
191 0 : commIndex = segmentIdx;
192 0 : } else { // 如果当前rank不是通信网口，则不发送数据
193 0 : hdSize = 0;
194 : }
195 0 : } else if (multRingsSliceZero.size() == 1) {
196 0 : segmentIdx = commInfo.localRank;
197 0 : CHK_PRT_RET(segmentIdx >= multRingsSliceZero[0].size(), HCCL_ERROR("[Prepare][Level1CommInfo]index is out of "\
198 : "range. Idx[%u] Slice size[%zu]", segmentIdx, multRingsSliceZero[0].size()), HCCL_E_PARA);
199 0 : hdSize = multRingsSliceZero[0][segmentIdx].size;
200 0 : commIndex = segmentIdx;
201 : } else {
202 0 : return HCCL_E_PARA;
203 : }
204 0 : HCCL_INFO("[CollAllReduceRingFor91093Executor][PrepareARSLevel1CommInfo]userRank[%u] segmentIdx[%u] commIndex[%u] hdSize[%llu]",
205 : topoAttr_.userRank, segmentIdx, commIndex, hdSize);
206 0 : return HCCL_SUCCESS;
207 : }
208 :
209 16 : HcclResult CollAllReduceRingFor91093Executor::KernelRun(const OpParam ¶m, ExecMem &execMem)
210 : {
211 16 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] The CollAllReduceRingFor91093Executor starts", __func__);
212 16 : CHK_RET(ActiveSlaveStreams(param.stream));
213 16 : CHK_RET(GetLevelCommInfo()); // 获取通信域
214 16 : u32 perDataSize = 0;
215 16 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
216 16 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
217 16 : std::vector<std::vector<Slice> > multRingsSliceZero; // 数据基于该rank上环0的偏移
218 16 : u32 sliceNum = logicalLevel0CommInfo_.localRankSize;
219 : // 根据数据量计算每个环上数据的偏移和大小
220 16 : CHK_RET(AlgTemplateBase::PrepareSliceData(execMem.count, perDataSize, sliceNum, 0, dataSegsSlice));
221 :
222 : /* 三步算法step1：外层 - 节点内 reduce-scatter */
223 : // 构造ring algorithm对应的reduce-scatter实例
224 16 : std::vector<u32> mockNicList = topoAttr_.nicList;
225 16 : CHK_RET(GetNicList(mockNicList));
226 16 : u32 level0RankSize = logicalLevel0CommInfo_.localRankSize;
227 16 : bool ARSFlag = topoMatcher_->GetARSFlag();
228 16 : bool ARSDoubleRing = (ARSFlag && (level0RankSize > FACTOR_TWO) && topoAttr_.isARSDoubleRing);
229 :
230 : // 多环数据切分
231 16 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING || ARSDoubleRing) {
232 16 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, mockNicList, logicalLevel0plane_);
233 : } else {
234 0 : multRingsSliceZero.push_back(dataSegsSlice);
235 : }
236 :
237 : // 第一步的reducescatter输出放在CCL buffer上，通过设置nullptr指示不做最后一步的DMA削减动作
238 16 : HcomCollOpInfo reduceScatterOpInfo = {
239 16 : "", execMem.inputPtr, nullptr, execMem.count, param.DataDes.dataType, param.root, param.reduceType
240 16 : };
241 16 : HcomCollOpInfo reduceScatterGraphModeOpInfo = {
242 16 : "", execMem.inputMem.ptr(), nullptr, execMem.count, param.DataDes.dataType, param.root, param.reduceType
243 16 : };
244 16 : HcomCollOpInfo *reduceScatterOpInfoPtr = nullptr;
245 16 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
246 16 : reduceScatterOpInfoPtr = &reduceScatterGraphModeOpInfo;
247 : }
248 16 : if (DMAReduceFlag_) {
249 16 : reduceScatterOpInfoPtr = &reduceScatterOpInfo;
250 : }
251 16 : bool disableDMAReduce = algOpContext_.opRetryHandler.retryEnable &&
252 0 : (algOpContext_.opRetryHandler.inPlaceSupportRetryStatus == InplaceSupportRetryStatus::RETRY_1_ALLOW_NO_DMA_REDUCE_CASE1 ||
253 0 : algOpContext_.opRetryHandler.inPlaceSupportRetryStatus == InplaceSupportRetryStatus::RETRY_1_ALLOW_NO_DMA_REDUCE_CASE2);
254 16 : const std::vector<std::vector<Slice>> multRingsUserMemSliceDefault = std::vector<std::vector<Slice>>(0);
255 16 : CHK_RET(RunIntraSeverReduceScatter(param.tag, execMem.inputMem, execMem.outputMem, execMem.count,
256 : param.DataDes.dataType, param.reduceType, multRingsSliceZero, param.stream,
257 : PROF_STAGE_0, 0, reduceScatterOpInfoPtr, multRingsUserMemSliceDefault, disableDMAReduce));
258 16 : HCCL_INFO("AllReduce double ring stage0 run success.");
259 :
260 32 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
261 16 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
262 :
263 16 : if(ARSFlag && (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
264 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) ){
265 0 : algType_.algoLevel1 = AlgTypeLevel1::ALG_LEVEL1_NHR;
266 : }
267 :
268 : /* 三步算法step2: 内层 - 节点间 allreduce */
269 : u64 hdSize;
270 : u32 segmentIdx;
271 : u32 commIndex;
272 16 : if (ARSFlag) {
273 0 : CHK_RET(PrepareARSLevel1CommInfo(segmentIdx, commIndex, hdSize, logicalLevel0CommInfo_, multRingsSliceZero, param.tag, mockNicList));
274 : } else {
275 16 : CHK_RET(PrepareLevel1CommInfo(segmentIdx, commIndex, hdSize, logicalLevel0CommInfo_, multRingsSliceZero, param.tag));
276 : }
277 16 : if (ARSDoubleRing && reduceScatterOpInfoPtr == nullptr) {
278 0 : DeviceMem srcMem = execMem.inputMem.range(dataSegsSlice[segmentIdx].offset, hdSize);
279 0 : DeviceMem dstMem = execMem.outputMem.range(dataSegsSlice[segmentIdx].offset, hdSize);
280 0 : HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream));
281 0 : }
282 :
283 16 : u64 hdCount = hdSize / perDataSize;
284 16 : if (topoAttr_.superPodNum <= 1 || isSelectAHC) {
285 16 : DeviceMem allreduceInput = execMem.inputMem.range(dataSegsSlice[segmentIdx].offset, hdSize);
286 16 : CHK_SMART_PTR_NULL(allreduceInput);
287 16 : DeviceMem allreduceOutput = execMem.outputMem.range(dataSegsSlice[segmentIdx].offset, hdSize);
288 16 : CHK_SMART_PTR_NULL(allreduceOutput);
289 :
290 16 : u64 reduceAttr = GetReduceAttr(allreduceInput, allreduceOutput, param.DataDes.dataType, param.reduceType);
291 16 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
292 16 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
293 32 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING,
294 16 : dispatcher_);
295 16 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_RING in COMM_LEVEL1", __func__);
296 16 : CHK_SMART_PTR_NULL(level1TempAlg);
297 16 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
298 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
299 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR_V1, dispatcher_);
300 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NHR_V1 in COMM_LEVEL1", __func__);
301 0 : CHK_SMART_PTR_NULL(level1TempAlg);
302 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
303 0 : } else if (isSelectAHC) {
304 : // 获取通信域分组信息
305 0 : std::vector<std::vector<std::vector<u32>>> globalSubGroups;
306 0 : std::map<AHCConcOpType, TemplateType> ahcAlgOption;
307 0 : CHK_RET(topoMatcher_->GetGlobalSubGroups(logicalLevel1plane_, globalSubGroups));
308 0 : topoMatcher_->GetAHCAlgOption(ahcAlgOption);
309 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
310 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_AHC, dispatcher_);
311 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_AHC in COMM_LEVEL1", __func__);
312 : } else {
313 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_AHC_BROKE, dispatcher_);
314 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_AHC_BROKE in COMM_LEVEL1", __func__);
315 : }
316 0 : CHK_SMART_PTR_NULL(level1TempAlg);
317 0 : CHK_RET(level1TempAlg->Prepare(execMem.count, globalSubGroups, ahcAlgOption));
318 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
319 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
320 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB,
321 0 : dispatcher_);
322 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NB in COMM_LEVEL1", __func__);
323 0 : CHK_SMART_PTR_NULL(level1TempAlg);
324 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
325 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
326 0 : u64 curSize = execMem.count * SIZE_TABLE[param.DataDes.dataType]; // 单位 byte
327 0 : HCCL_DEBUG("allreduce ring: curSize[%llu] deviceNumPerAggregation[%u] commLevel0Size[%u]",
328 : curSize, logicalLevel0CommInfo_.localRankSize, logicalLevel0CommInfo_.localRankSize);
329 0 : if (curSize / logicalLevel0CommInfo_.localRankSize <= NHR_ALLREDUCE_SMALL_SIZE) {
330 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR_ONESHOT, dispatcher_);
331 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NHR_ONESHOT in COMM_LEVEL1", __func__);
332 : } else {
333 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
334 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NHR in COMM_LEVEL1", __func__);
335 : }
336 0 : CHK_SMART_PTR_NULL(level1TempAlg);
337 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
338 : } else {
339 0 : HCCL_ERROR("AllReduce ring: algType_[%u] is not supported.", algType_.algoLevel1);
340 0 : return HCCL_E_NOT_SUPPORT;
341 : }
342 16 : CHK_SMART_PTR_NULL(level1TempAlg);
343 16 : u32 rankSize = logicalLevel1CommInfo_.localRankSize;
344 : // 节点间的hd 使用环0来记录
345 80 : CHK_RET(level1TempAlg->Prepare(
346 : allreduceInput, allreduceOutput, allreduceOutput, hdCount,
347 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID,
348 : std::vector<Slice>(0), dataSegsSlice[segmentIdx].offset));
349 16 : CHK_RET(level1TempAlg->RegisterProfiler(
350 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + logicalLevel1CommInfo_.localRank,
351 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
352 16 : CHK_RET(RunTemplate(level1TempAlg, logicalLevel1CommInfo_));
353 :
354 16 : HCCL_INFO("AllReduce double ring stage1 run success");
355 32 : } else {
356 : // 超节点内做reducescatter
357 0 : CHK_RET(CheckCommSize(logicalLevel1plane_, commIndex + 1));
358 0 : u32 level1RankSize = logicalLevel1CommInfo_.localRankSize;
359 0 : u64 level1Offset = dataSegsSlice[segmentIdx].offset;
360 :
361 : // 根据数据量计算每个环上数据的偏移和大小
362 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(hdCount, perDataSize, level1RankSize, 0, dataSegsSlice));
363 0 : DeviceMem reducescatterInput = execMem.inputMem.range(level1Offset, hdSize);
364 0 : CHK_SMART_PTR_NULL(reducescatterInput);
365 0 : DeviceMem reducescatterOutput = execMem.outputMem.range(level1Offset, hdSize);
366 0 : CHK_SMART_PTR_NULL(reducescatterOutput);
367 0 : if (level1RankSize > 1) {
368 0 : u64 reduceAttr = GetReduceAttr(reducescatterInput, reducescatterOutput,
369 0 : param.DataDes.dataType, param.reduceType);
370 0 : std::unique_ptr<AlgTemplateBase> level1RSTempAlg;
371 :
372 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
373 0 : level1RSTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_RING,
374 0 : dispatcher_);
375 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL1", __func__);
376 0 : CHK_SMART_PTR_NULL(level1RSTempAlg);
377 0 : CHK_RET(level1RSTempAlg->Prepare(reduceAttr));
378 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
379 0 : level1RSTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NB,
380 0 : dispatcher_);
381 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_LEVEL1", __func__);
382 0 : CHK_SMART_PTR_NULL(level1RSTempAlg);
383 0 : CHK_RET(level1RSTempAlg->Prepare(reduceAttr));
384 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
385 0 : level1RSTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NHR,
386 0 : dispatcher_);
387 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL1", __func__);
388 0 : CHK_SMART_PTR_NULL(level1RSTempAlg);
389 0 : CHK_RET(level1RSTempAlg->Prepare(reduceAttr, false));
390 : } else {
391 0 : HCCL_ERROR("ReduceScatter ring: algType_[%u] is not supported.", algType_.algoLevel1);
392 0 : return HCCL_E_NOT_SUPPORT;
393 : }
394 0 : CHK_RET(level1RSTempAlg->Prepare(
395 : reducescatterInput, reducescatterInput, reducescatterOutput, hdCount, param.DataDes.dataType,
396 : param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, level1Offset));
397 :
398 0 : CHK_RET(level1RSTempAlg->RegisterProfiler(
399 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + logicalLevel1CommInfo_.localRank,
400 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
401 0 : CHK_RET(RunTemplate(level1RSTempAlg, logicalLevel1CommInfo_));
402 0 : HCCL_INFO("AllReduce double ring [superpod] level1 ReduceScatter run success");
403 0 : }
404 :
405 : // 超节点间做allreduce
406 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
407 0 : u32 rankSize = level2CommInfo.localRankSize;
408 0 : u32 localRank = logicalLevel1CommInfo_.localRank;
409 :
410 : DeviceMem allreduceInput =
411 0 : reducescatterInput.range(dataSegsSlice[localRank].offset, dataSegsSlice[localRank].size);
412 0 : CHK_SMART_PTR_NULL(allreduceInput);
413 : DeviceMem allreduceOutput =
414 0 : reducescatterOutput.range(dataSegsSlice[localRank].offset, dataSegsSlice[localRank].size);
415 0 : CHK_SMART_PTR_NULL(allreduceOutput);
416 :
417 0 : u64 reduceAttr = GetReduceAttr(allreduceInput, allreduceOutput, param.DataDes.dataType, param.reduceType);
418 :
419 0 : std::unique_ptr<AlgTemplateBase> level2ARTempAlg;
420 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
421 0 : level2ARTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB, dispatcher_);
422 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NB in COMM_LEVEL2", __func__);
423 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
424 0 : level2ARTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
425 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NHR in COMM_LEVEL2", __func__);
426 0 : if (algoAttr_.isSupportAtomicWrite) {
427 0 : CHK_SMART_PTR_NULL(level2ARTempAlg);
428 0 : level2ARTempAlg->CloseBarrier();
429 : }
430 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_RING) {
431 0 : level2ARTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
432 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_RING in COMM_LEVEL2", __func__);
433 : } else {
434 0 : level2ARTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
435 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING in COMM_LEVEL2", __func__);
436 : }
437 0 : CHK_SMART_PTR_NULL(level2ARTempAlg);
438 0 : CHK_RET(level2ARTempAlg->Prepare(reduceAttr));
439 :
440 0 : u64 arCount = dataSegsSlice[localRank].size / perDataSize;
441 0 : CHK_RET(level2ARTempAlg->Prepare(
442 : allreduceInput, allreduceOutput, allreduceOutput, arCount,
443 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID,
444 : std::vector<Slice>(0), dataSegsSlice[localRank].offset + level1Offset));
445 0 : CHK_RET(level2ARTempAlg->RegisterProfiler(
446 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank,
447 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
448 0 : CHK_RET(RunTemplate(level2ARTempAlg, level2CommInfo));
449 0 : HCCL_INFO("AllReduce double ring [superpod] level2 AllReduce run success");
450 :
451 : // 超节点内做allgather
452 0 : if (level1RankSize > 1) {
453 0 : std::unique_ptr<AlgTemplateBase> level1AGTempAlg;
454 0 : DeviceMem allgatherInput = execMem.outputMem.range(level1Offset, hdSize);
455 0 : DeviceMem allgatherOutput = execMem.outputMem.range(level1Offset, hdSize);
456 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
457 0 : level1AGTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING,
458 0 : dispatcher_);
459 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL1", __func__);
460 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
461 0 : level1AGTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB,
462 0 : dispatcher_);
463 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL1", __func__);
464 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
465 0 : level1AGTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR,
466 0 : dispatcher_);
467 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL1", __func__);
468 : } else {
469 0 : HCCL_ERROR("AllGather ring: algType_[%u] is not supported.", algType_.algoLevel1);
470 0 : return HCCL_E_NOT_SUPPORT;
471 : }
472 0 : CHK_SMART_PTR_NULL(level1AGTempAlg);
473 0 : CHK_RET(level1AGTempAlg->Prepare(allgatherInput, allgatherOutput, allgatherOutput, arCount,
474 : param.DataDes.dataType, param.stream,
475 : HcclReduceOp::HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice, level1Offset));
476 0 : CHK_RET(level1AGTempAlg->RegisterProfiler(
477 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + logicalLevel1CommInfo_.localRank,
478 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
479 0 : CHK_RET(RunTemplate(level1AGTempAlg, logicalLevel1CommInfo_));
480 0 : HCCL_INFO("AllReduce double ring [superpod] level1 AllGather run success");
481 0 : }
482 0 : }
483 : /* 三步算法step3：外层 - 节点内 allgather */
484 : // 第三步的allgather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
485 16 : HcomCollOpInfo allgatherOpInfo = {
486 16 : "", nullptr, execMem.outputPtr, execMem.count, param.DataDes.dataType, param.root, param.reduceType
487 16 : };
488 16 : HcomCollOpInfo allgatherOpInfoGraphModeOpInfo = {
489 16 : "", nullptr, execMem.outputMem.ptr(), execMem.count, param.DataDes.dataType, param.root, param.reduceType
490 16 : };
491 16 : HcomCollOpInfo *allgatherOpInfoPtr = nullptr;
492 16 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
493 16 : allgatherOpInfoPtr = &allgatherOpInfoGraphModeOpInfo;
494 : }
495 16 : if (DMAReduceFlag_) {
496 16 : allgatherOpInfoPtr = &allgatherOpInfo;
497 : }
498 48 : CHK_RET(RunIntraSeverAllGather(param.tag, execMem.inputMem, execMem.outputMem, hdCount,
499 : param.DataDes.dataType, multRingsSliceZero, param.stream,
500 : PROF_STAGE_2, 0, allgatherOpInfoPtr));
501 16 : HCCL_INFO("AllReduce double ring stage2 run success");
502 16 : return HCCL_SUCCESS;
503 16 : }
504 0 : HcclResult CollAllReduceRingFor91093Executor::Getlevel1CommRank(SubCommInfo& level1CommInfo)
505 : {
506 0 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
507 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
508 0 : if (isSelectAHC) {
509 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
510 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
511 :
512 0 : u32 commIndex = level0CommInfo.localRank;
513 :
514 0 : CommPlane commPlaneLevel1 = isSelectAHC ? COMM_LEVEL1_AHC : COMM_LEVEL1;
515 0 : CHK_RET(CheckCommSize(commPlaneLevel1, commIndex + 1));
516 0 : level1CommInfo = GetSubCommInfo(commPlaneLevel1, commIndex);
517 0 : return HCCL_SUCCESS;
518 0 : }
519 0 : if (CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1) != HCCL_SUCCESS) {
520 0 : return HCCL_E_UNAVAIL;
521 : }
522 0 : level1CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
523 :
524 0 : return HCCL_SUCCESS;
525 : }
526 :
527 0 : HcclResult CollAllReduceRingFor91093Executor::SelectTempAlg(std::unique_ptr<AlgTemplateBase> &level1TempAlg, u32 level1RankSize)
528 : {
529 0 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
530 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
531 0 : if (isSelectAHC) {
532 0 : CommPlane commPlaneLevel1 = COMM_LEVEL1_AHC;
533 : // 获取通信域分组信息
534 0 : std::vector<std::vector<std::vector<u32>>> globalSubGroups;
535 0 : std::map<AHCConcOpType, TemplateType> ahcAlgOption;
536 0 : CHK_RET(topoMatcher_->GetGlobalSubGroups(commPlaneLevel1, globalSubGroups));
537 0 : topoMatcher_->GetAHCAlgOption(ahcAlgOption);
538 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
539 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_AHC, dispatcher_);
540 0 : HCCL_INFO("allreduce ring: using ahc algo inter-server.");
541 : } else {
542 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_AHC_BROKE, dispatcher_);
543 0 : HCCL_INFO("allreduce ring: using ahc-broke algo inter-server.");
544 : }
545 0 : CHK_SMART_PTR_NULL(level1TempAlg);
546 0 : CHK_RET(level1TempAlg->Prepare(NSLBDP_MIN_COUNT, globalSubGroups, ahcAlgOption));
547 0 : return HCCL_SUCCESS;
548 0 : }
549 0 : if (level1RankSize > 1) {
550 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
551 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB, dispatcher_);
552 0 : HCCL_INFO("AllReduce ring: using nonuniform-bruck algo inter-superPod.");
553 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
554 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
555 0 : HCCL_INFO("AllReduce ring: using nonuniform-hierarchical-ring algo inter-superPod.");
556 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_RING) {
557 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
558 0 : HCCL_INFO("AllReduce ring: using ring algo inter-superPod.");
559 : } else {
560 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
561 0 : HCCL_INFO("AllReduce ring: using halving-doubling algo inter-superPod.");
562 : }
563 0 : CHK_SMART_PTR_NULL(level1TempAlg);
564 0 : return HCCL_SUCCESS;
565 : }
566 0 : return HCCL_E_UNAVAIL;
567 : }
568 :
569 16 : HcclResult CollAllReduceRingFor91093Executor::GetNicList(std::vector<u32> &mockNicList)
570 : {
571 16 : mockNicList.clear();
572 16 : if (logicalLevel0plane_ == COMM_LEVEL0_LOGICAL) {
573 0 : mockNicList.reserve(logicalLevel0CommInfo_.localRankSize);
574 0 : for (u32 rankIndex = 0; rankIndex < logicalLevel0CommInfo_.localRankSize; rankIndex++) {
575 0 : mockNicList.push_back(rankIndex);
576 : }
577 : }else{
578 16 : mockNicList = topoAttr_.nicList;
579 : }
580 16 : return HCCL_SUCCESS;
581 : }
582 :
583 : REGISTER_EXEC("AllReduceRingFor91093Executor", AllReduceRingFor91093, CollAllReduceRingFor91093Executor);
584 :
585 : } // namespace hccl
|
