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_gather_ring_for_910_93_executor.h"
12 :
13 : namespace hccl {
14 2 : CollAllGatherRingFor91093Executor::CollAllGatherRingFor91093Executor(const HcclDispatcher dispatcher,
15 2 : std::unique_ptr<TopoMatcher> &topoMatcher)
16 2 : : CollAllGatherExecutor(dispatcher, topoMatcher)
17 : {
18 2 : DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
19 2 : desc_.level1SupportedAlgos = {
20 : AlgTypeLevel1::ALG_LEVEL1_NHR,
21 : AlgTypeLevel1::ALG_LEVEL1_NB,
22 : AlgTypeLevel1::ALG_LEVEL1_RING,
23 : AlgTypeLevel1::ALG_LEVEL1_AHC,
24 : AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE
25 2 : };
26 2 : desc_.level2SupportedAlgos = {
27 : AlgTypeLevel2::ALG_LEVEL2_NHR,
28 : AlgTypeLevel2::ALG_LEVEL2_NB,
29 : AlgTypeLevel2::ALG_LEVEL2_RING
30 2 : };
31 2 : }
32 :
33 2 : HcclResult CollAllGatherRingFor91093Executor::CalcStreamNum(u32& streamNum)
34 : {
35 2 : u32 totalStreamNum = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
36 : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
37 2 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
38 2 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
39 : }
40 :
41 2 : streamNum = totalStreamNum - 1;
42 2 : HCCL_INFO("[CollAllGatherRingFor91093Executor][CalcStreamNum] tag[%s] streamNum_[%u]",
43 : tag_.c_str(), streamNum);
44 2 : return HCCL_SUCCESS;
45 : }
46 :
47 2 : HcclResult CollAllGatherRingFor91093Executor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
48 : {
49 2 : TransportMemType inputType = TransportMemType::RESERVED;
50 2 : TransportMemType outputType = TransportMemType::RESERVED;
51 2 : CHK_RET(CalcTransportMemType(inputType, outputType));
52 2 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
53 2 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
54 2 : CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
55 2 : return HCCL_SUCCESS;
56 : }
57 :
58 2 : HcclResult CollAllGatherRingFor91093Executor::CalcTransportMemType(TransportMemType &inputType,
59 : TransportMemType &outputType)
60 : {
61 2 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
62 2 : inputType = TransportMemType::CCL_INPUT;
63 2 : outputType = TransportMemType::CCL_OUTPUT;
64 : } else {
65 0 : inputType = TransportMemType::PARAM_INPUT;
66 0 : outputType = TransportMemType::PARAM_OUTPUT;
67 : }
68 2 : HCCL_INFO("[CollAllGatherRingFor91093Executor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
69 : tag_.c_str(), inputType, outputType);
70 2 : return HCCL_SUCCESS;
71 : }
72 :
73 2 : HcclResult CollAllGatherRingFor91093Executor::CalcLevel0CommInfo(TransportMemType inputType,
74 : TransportMemType outputType,
75 : std::vector<LevelNSubCommTransport>& opTransport)
76 : {
77 2 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
78 2 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
79 2 : return HCCL_SUCCESS;
80 2 : }
81 :
82 2 : HcclResult CollAllGatherRingFor91093Executor::CalcLevel2CommInfo(TransportMemType inputType, TransportMemType outputType,
83 : std::vector<LevelNSubCommTransport>& opTransport)
84 : {
85 2 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
86 2 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
87 0 : HCCL_INFO("[CollAllGatherRingFor91093Executor][CalcLevel2CommInfo] select AHC bypass level2 comm calculate");
88 0 : return HCCL_SUCCESS;
89 : }
90 :
91 2 : CommParaInfo commParaLevel2(COMM_LEVEL2, CommType::COMM_TAG_MAX);
92 2 : HCCL_DEBUG("[CollAllGatherRingFor91093Executor][CalcLevel2CommInfo]Level2CommInfo start set");
93 2 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
94 0 : commParaLevel2.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
95 0 : HCCL_INFO("[%s]Calc NHRCommInfo.", __func__);
96 2 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
97 0 : commParaLevel2.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
98 0 : HCCL_INFO("[%s]Calc NBCommInfo.", __func__);
99 : } else {
100 2 : commParaLevel2.commType = CommType::COMM_TAG_RING_INNER;
101 2 : HCCL_INFO("[%s]Calc RingCommInfo.", __func__);
102 : }
103 2 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel2, opTransport[COMM_LEVEL2], inputType, outputType));
104 2 : return HCCL_SUCCESS;
105 2 : }
106 :
107 2 : u64 CollAllGatherRingFor91093Executor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
108 : {
109 2 : u64 maxCountPerLoop = cclBuffSize / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN
110 2 : * HCCL_MIN_SLICE_ALIGN / unitSize;
111 2 : return maxCountPerLoop;
112 : }
113 :
114 0 : HcclResult CollAllGatherRingFor91093Executor::RunIntraSeverAllGather(
115 : const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
116 : const u64 count, const HcclDataType &dataType, const std::vector<std::vector<Slice>> &multRingsSliceZero,
117 : const Stream &stream, s32 profStage, const u64 baseOffset, const HcomCollOpInfo *opInfo,
118 : const std::vector<std::vector<Slice>> &multRingsUserMemSlice)
119 : {
120 0 : CHK_RET(MultiRingAllGather(tag, inputMem, outputMem, count, dataType,
121 : multRingsSliceZero, stream, profStage, baseOffset, opInfo, multRingsUserMemSlice, logicalLevel0plane_));
122 0 : return HCCL_SUCCESS;
123 : }
124 :
125 18 : u64 CollAllGatherRingFor91093Executor::CalcDstMemOffset(const OpParam ¶m, u32 perDataSize, u64 inputMemSize) const
126 : {
127 18 : return topoAttr_.userRank * inputMemSize;
128 : }
129 :
130 18 : HcomCollOpInfo CollAllGatherRingFor91093Executor::GetHcomCollOpInfo(const OpParam ¶m, const ExecMem &execMem) const
131 : {
132 18 : HcomCollOpInfo opInfo = {
133 18 : "", execMem.inputPtr, execMem.outputPtr, param.DataDes.count, param.DataDes.dataType, 0, HCCL_REDUCE_RESERVED,
134 18 : param.DataDes.strideCount
135 18 : };
136 18 : if (!DMAReduceFlag_ && (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING)) {
137 0 : opInfo.inputAddr = execMem.inputMem.ptr();
138 0 : opInfo.outputAddr = execMem.outputMem.ptr();
139 : }
140 18 : return opInfo;
141 : }
142 :
143 18 : HcclResult CollAllGatherRingFor91093Executor::PrepareSlicesL0(std::vector<std::vector<Slice>> &multRingsSlice,
144 : const OpParam ¶m, const SubCommInfo &level2CommInfo, const SubCommInfo &level1CommInfo,
145 : const SubCommInfo &level0CommInfo, u32 perDataSize, u64 inputMemSize)
146 : {
147 18 : const u32 level0RankSize = level0CommInfo.localRankSize;
148 18 : const u32 level1RankSize = level1CommInfo.localRankSize;
149 18 : const u32 level2RankSize = level2CommInfo.localRankSize;
150 :
151 18 : std::vector<Slice> dataSegsSlice;
152 18 : CHK_RET(PrepareAllgatherSlice(level0RankSize, inputMemSize, dataSegsSlice));
153 :
154 : // 多环数据切分
155 18 : std::vector<std::vector<Slice>> multRingsSliceZero; // 数据基于该rank上环0的偏移
156 18 : bool ARSFlag = topoMatcher_->GetARSFlag();
157 18 : bool ARSDoubleRing = (ARSFlag && (level0RankSize > FACTOR_TWO) && topoAttr_.isARSDoubleRing);
158 :
159 18 : if (ARSDoubleRing) {
160 0 : std::vector<u32> mockNicList;
161 0 : mockNicList.reserve(level0RankSize);
162 0 : for (u32 rankIndex = 0; rankIndex < level0RankSize; rankIndex++) {
163 0 : mockNicList.push_back(rankIndex);
164 : }
165 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, mockNicList);
166 36 : } else if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING &&
167 18 : !IsSupportUnifiedMarch(param, topoType_, topoAttr_.serverNum, topoAttr_.superPodNum)) {
168 18 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, topoAttr_.nicList);
169 : } else {
170 0 : multRingsSliceZero.push_back(dataSegsSlice);
171 : }
172 54 : for (u32 ringIndex = 0; ringIndex < multRingsSliceZero.size(); ringIndex++) {
173 36 : std::vector<Slice> level2DataSlice;
174 36 : CHK_RET(CalculateLevel2AllgatherSlice(inputMemSize, level0RankSize, level1RankSize, level2RankSize,
175 : multRingsSliceZero, level2DataSlice, ringIndex));
176 36 : multRingsSlice.push_back(level2DataSlice);
177 36 : }
178 :
179 18 : return HCCL_SUCCESS;
180 18 : }
181 :
182 18 : std::vector<Slice> CollAllGatherRingFor91093Executor::PrepareSlicesL1(const OpParam ¶m,
183 : const SubCommInfo &level2CommInfo, const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo,
184 : u32 perDataSize, u64 inputMemSize) const
185 : {
186 18 : const u32 level0RankSize = level0CommInfo.localRankSize;
187 18 : const u32 level0ServerIndex = level0CommInfo.localRank;
188 18 : const u32 level1RankSize = level1CommInfo.localRankSize;
189 18 : const u32 level2RankSize = level2CommInfo.localRankSize;
190 18 : std::vector<Slice> level1DataSegsSlice;
191 54 : for (u32 j = 0; j < level1RankSize; j++) {
192 72 : for (u32 i = 0; i < level2RankSize; i++) {
193 36 : Slice level1Slice;
194 36 : level1Slice.size = inputMemSize;
195 36 : level1Slice.offset = inputMemSize *
196 36 : (i * level1RankSize * level0RankSize + j * level0RankSize + level0ServerIndex);
197 :
198 36 : HCCL_DEBUG("[CollAllGatherRingFor91093Executor][PrepareSlicesL1] rank[%u], level1index[%u], level2index[%u], slices.offset=%llu, slices.size=%llu",
199 : level0CommInfo.localRank, j, i, level1Slice.offset, level1Slice.size);
200 :
201 36 : level1DataSegsSlice.push_back(level1Slice);
202 : }
203 : }
204 18 : return level1DataSegsSlice;
205 0 : }
206 :
207 0 : std::vector<Slice> CollAllGatherRingFor91093Executor::PrepareSlicesL2(const OpParam ¶m,
208 : const SubCommInfo &level2CommInfo, const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo,
209 : u32 perDataSize, u64 inputMemSize) const
210 : {
211 0 : const u32 level0RankSize = level0CommInfo.localRankSize;
212 0 : const u32 level0ServerIndex = level0CommInfo.localRank;
213 0 : const u32 level1RankSize = level1CommInfo.localRankSize;
214 0 : const u32 level1ServerIndex = level1CommInfo.localRank;
215 0 : const u32 level2RankSize = level2CommInfo.localRankSize;
216 0 : std::vector<Slice> level2DataSegsSlice;
217 0 : for (u32 i = 0; i < level2RankSize; i++) {
218 0 : Slice sliceTemp;
219 0 : sliceTemp.size = inputMemSize;
220 0 : sliceTemp.offset = inputMemSize *
221 0 : (i * level1RankSize * level0RankSize + level1ServerIndex * level0RankSize + level0ServerIndex);
222 0 : level2DataSegsSlice.push_back(sliceTemp);
223 : }
224 0 : return level2DataSegsSlice;
225 0 : }
226 :
227 18 : HcclResult CollAllGatherRingFor91093Executor::PrepareUserMemSlices(std::vector<std::vector<Slice>> &userMemSlices,
228 : const std::vector<std::vector<Slice>> &multRingsSlice, const OpParam ¶m, const SubCommInfo &level2CommInfo,
229 : const SubCommInfo &level1CommInfo, const SubCommInfo &level0CommInfo, u32 perDataSize, u64 inputMemSize)
230 : {
231 18 : CHK_PRT_RET(0 < param.DataDes.strideCount && param.DataDes.strideCount < param.DataDes.count,
232 : HCCL_ERROR("[CollAllGatherRingFor91093Executor][KernelRun]strideCount[%llu] is smaller than opCount[%llu]",
233 : param.DataDes.strideCount, param.DataDes.count),
234 : HCCL_E_PARA);
235 18 : HCCL_DEBUG("[CollAllGatherRingFor91093Executor][KernelRun]strideCount[%llu], opCount[%llu]",
236 : param.DataDes.strideCount, param.DataDes.count);
237 :
238 18 : if (!DMAReduceFlag_) {
239 0 : userMemSlices = multRingsSlice;
240 : // 图模式，根据strideCount更新slice的offset
241 0 : if (param.DataDes.strideCount != 0) {
242 0 : CHK_RET(UpdateOffsetBasedOnStrideCount(param, userMemSlices));
243 : }
244 : } else {
245 54 : for (u32 ringIndex = 0; ringIndex < multRingsSlice.size(); ringIndex++) {
246 36 : std::vector<Slice> userMemSlice;
247 180 : for (const auto &cclSlice : multRingsSlice[ringIndex]) {
248 144 : Slice tmpSlice;
249 144 : u64 count = (param.DataDes.strideCount == 0) ? param.DataDes.count : param.DataDes.strideCount;
250 144 : tmpSlice.size = cclSlice.size;
251 288 : tmpSlice.offset = (cclSlice.offset / inputMemSize) * count * perDataSize +
252 144 : multRingsSlice[ringIndex][0].offset;
253 144 : userMemSlice.push_back(tmpSlice);
254 144 : HCCL_DEBUG("rank[%u], ringIndex[%u], tmpSlice.offset=[%llu], size=[%llu]",
255 : topoAttr_.userRank, ringIndex, tmpSlice.offset, tmpSlice.size);
256 : }
257 36 : userMemSlices.push_back(userMemSlice);
258 36 : }
259 : }
260 18 : return HCCL_SUCCESS;
261 : }
262 :
263 18 : HcclResult CollAllGatherRingFor91093Executor::GetLevelCommInfo()
264 : {
265 18 : logicalLevel0plane_ = COMM_LEVEL0;
266 18 : CHK_RET(CheckCommSize(logicalLevel0plane_, COMM_INDEX_0 + 1));
267 18 : logicalLevel0CommInfo_ = GetSubCommInfo(logicalLevel0plane_, COMM_INDEX_0);
268 18 : u32 commIndex = logicalLevel0CommInfo_.localRank;
269 36 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
270 18 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
271 18 : logicalLevel1plane_ = isSelectAHC ? COMM_LEVEL1_AHC : COMM_LEVEL1;
272 18 : CHK_RET(CheckCommSize(logicalLevel1plane_, commIndex + 1));
273 18 : logicalLevel1CommInfo_ = GetSubCommInfo(logicalLevel1plane_, commIndex);
274 18 : return HCCL_SUCCESS;
275 : }
276 :
277 18 : HcclResult CollAllGatherRingFor91093Executor::KernelRun(const OpParam ¶m, ExecMem &execMem)
278 : {
279 18 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] The AllGatherRingExecutor starts, topoType_[%u], agv[%u]",
280 : __func__, topoType_, isAllGatherV_);
281 18 : CHK_RET(GetLevelCommInfo()); // 设置逻辑通信域
282 18 : CHK_RET(ActiveSlaveStreams(param.stream));
283 18 : const HcclDataType dataType = param.GetDataType();
284 18 : u32 perDataSize = 0;
285 18 : CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
286 18 : CHK_PRT_RET(perDataSize == 0,
287 : HCCL_ERROR("[CollAllGatherRingFor91093Executor][KernelRun]errNo[0x%016llx] datatype[%s] is invalid",
288 : HCCL_ERROR_CODE(HCCL_E_PARA), GetDataTypeEnumStr(dataType).c_str()), HCCL_E_PARA);
289 :
290 36 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
291 18 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
292 :
293 18 : u32 level1RankSize = logicalLevel1CommInfo_.localRankSize;
294 :
295 18 : SubCommInfo level2CommInfo;
296 18 : if (isSelectAHC) {
297 0 : level2CommInfo = logicalLevel1CommInfo_;
298 0 : level2CommInfo.localRankSize = 1; // AHC bypass level2
299 : } else {
300 18 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
301 18 : level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
302 : }
303 18 : const u32 level2RankSize = level2CommInfo.localRankSize;
304 :
305 : // 第一步，将数据从input内存拷贝到output内存的对应位置
306 18 : u64 inputMemSize = execMem.inputMem.size();
307 18 : u64 dstMemOffset = CalcDstMemOffset(param, perDataSize, inputMemSize);
308 18 : DeviceMem dstMem = execMem.outputMem.range(dstMemOffset, inputMemSize);
309 18 : CHK_SMART_PTR_NULL(dstMem);
310 :
311 18 : HcomCollOpInfo opInfo = GetHcomCollOpInfo(param, execMem);
312 18 : HcomCollOpInfo *opInfoPtr = (DMAReduceFlag_ || (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING)) ? &opInfo :
313 : nullptr;
314 :
315 : // 图模式opinfo不为空，但需要将数据从ccl input拷贝到ccl output上
316 18 : HcclResult ret = HCCL_SUCCESS;
317 18 : if (!DMAReduceFlag_) {
318 0 : ret = HcclD2DMemcpyAsync(dispatcher_, dstMem, execMem.inputMem, const_cast<Stream&>(param.stream));
319 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
320 : HCCL_ERROR("[CollAllGatherRingFor91093Executor][KernelRun]AllGather double "
321 : "ring memcpy Failed, Offset[%llu], Size[%llu]", dstMemOffset, inputMemSize), ret);
322 : } else {
323 : // 先做server间算法，带有消减拷贝场景数据需要从user input取，拷贝到ccl output上
324 18 : if (level1RankSize > 1 || level2RankSize > 1) {
325 18 : DeviceMem srcMem = DeviceMem::create(static_cast<u8 *>(execMem.inputPtr), inputMemSize);
326 18 : ret = HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream));
327 18 : CHK_PRT_RET(ret != HCCL_SUCCESS,
328 : HCCL_ERROR("[CollAllGatherRingFor91093Executor][KernelRun]AllGather double "
329 : "ring user memcpy Failed, Offset[%llu], Size[%llu]", dstMemOffset, inputMemSize), ret);
330 18 : }
331 : }
332 18 : if (level2RankSize > 1) {
333 0 : std::unique_ptr<AlgTemplateBase> level2AGExecutor;
334 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
335 0 : level2AGExecutor = AlgTemplateRegistry::Instance().GetAlgTemplate(
336 0 : TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
337 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL2", __func__);
338 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
339 0 : level2AGExecutor = AlgTemplateRegistry::Instance().GetAlgTemplate(
340 0 : TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
341 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL2", __func__);
342 : } else {
343 0 : level2AGExecutor = AlgTemplateRegistry::Instance().GetAlgTemplate(
344 0 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
345 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL2", __func__);
346 : }
347 0 : CHK_SMART_PTR_NULL(level2AGExecutor);
348 :
349 0 : std::vector<Slice> level2DataSegsSlice = PrepareSlicesL2(param, level2CommInfo, logicalLevel1CommInfo_, logicalLevel0CommInfo_,
350 0 : perDataSize, inputMemSize);
351 0 : CHK_RET(level2AGExecutor->Prepare(execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count,
352 : dataType, param.stream, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, level2DataSegsSlice, 0));
353 :
354 0 : CHK_RET(level2AGExecutor->RegisterProfiler((
355 : level2RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank,
356 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
357 :
358 0 : CHK_RET(RunTemplate(level2AGExecutor, level2CommInfo));
359 0 : HCCL_INFO("AllGather ring [superpod] level2 AllGather run successtopoType_[%u], agv[%u]",
360 : topoType_, isAllGatherV_);
361 0 : }
362 18 : if (level1RankSize > 1) {
363 : // 计算slice, 不同超节点相同slice
364 18 : std::vector<Slice> level1DataSegsSlice = PrepareSlicesL1(param, level2CommInfo, logicalLevel1CommInfo_, logicalLevel0CommInfo_,
365 18 : perDataSize, inputMemSize);
366 :
367 18 : std::unique_ptr<AlgTemplateBase> level1AGExecutor;
368 18 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
369 36 : level1AGExecutor = AlgTemplateRegistry::Instance().GetAlgTemplate(
370 18 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
371 18 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL1", __func__);
372 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
373 0 : level1AGExecutor = AlgTemplateRegistry::Instance().GetAlgTemplate(
374 0 : TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
375 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL1", __func__);
376 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
377 0 : level1AGExecutor = AlgTemplateRegistry::Instance().GetAlgTemplate(
378 0 : TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
379 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL1", __func__);
380 0 : } else if (isSelectAHC) {
381 : // 获取通信域分组信息
382 0 : std::vector<std::vector<std::vector<u32>>> globalSubGroups;
383 0 : std::map<AHCConcOpType, TemplateType> ahcAlgOption;
384 0 : CHK_RET(topoMatcher_->GetGlobalSubGroups(logicalLevel1plane_, globalSubGroups));
385 0 : topoMatcher_->GetAHCAlgOption(ahcAlgOption);
386 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
387 0 : level1AGExecutor = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_AHC, dispatcher_);
388 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_AHC in COMM_LEVEL1", __func__);
389 : } else {
390 0 : level1AGExecutor = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_AHC_BROKE, dispatcher_);
391 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_AHC_BROKE in COMM_LEVEL1", __func__);
392 : }
393 0 : CHK_SMART_PTR_NULL(level1AGExecutor);
394 0 : CHK_RET(level1AGExecutor->Prepare(execMem.count, globalSubGroups, ahcAlgOption));
395 0 : } else {
396 0 : HCCL_ERROR("AllGather ring: unsupported algtype [%s].", AlgTypeToStr(algType_).c_str());
397 0 : return HCCL_E_NOT_SUPPORT;
398 : }
399 18 : CHK_SMART_PTR_NULL(level1AGExecutor);
400 54 : CHK_RET(level1AGExecutor->Prepare(execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count,
401 : dataType, param.stream, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, level1DataSegsSlice, 0));
402 :
403 18 : CHK_RET(level1AGExecutor->RegisterProfiler((
404 : level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank,
405 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
406 :
407 18 : CHK_RET(RunTemplate(level1AGExecutor, logicalLevel1CommInfo_));
408 18 : HCCL_INFO("AllGather ring [superpod] level1 AllGather run successtopoType_[%u], agv[%u]",
409 : topoType_, isAllGatherV_);
410 18 : }
411 : // 节点内做AllGather ring
412 18 : std::vector<std::vector<Slice>> multRingsSlice;
413 18 : CHK_RET(PrepareSlicesL0(multRingsSlice, param, level2CommInfo, logicalLevel1CommInfo_, logicalLevel0CommInfo_, perDataSize,
414 : inputMemSize));
415 :
416 18 : std::vector<std::vector<Slice>> multRingsUserMemSlice;
417 18 : CHK_RET(PrepareUserMemSlices(multRingsUserMemSlice, multRingsSlice, param, level2CommInfo, logicalLevel1CommInfo_,
418 : logicalLevel0CommInfo_, perDataSize, inputMemSize));
419 :
420 18 : if (DMAReduceFlag_ && (level1RankSize > 1 || level2RankSize > 1)) {
421 : // allgather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
422 18 : opInfo.inputAddr = nullptr;
423 : }
424 18 : CHK_RET(RunIntraSeverAllGather(param.tag, execMem.inputMem, execMem.outputMem, execMem.count, dataType,
425 : multRingsSlice, param.stream, PROF_STAGE_2, 0, opInfoPtr, multRingsUserMemSlice));
426 18 : HCCL_INFO("AllGather ring run success. topoType_[%u], agv[%u]", topoType_, isAllGatherV_);
427 18 : return HCCL_SUCCESS;
428 18 : }
429 :
430 0 : HcclResult CollAllGatherRingFor91093Executor::Getlevel1CommRank(SubCommInfo& level1CommInfo)
431 : {
432 0 : HCCL_INFO("[CollAllGatherRingFor91093Executor][Getlevel1CommRank] Entry Getlevel1CommRank.");
433 0 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
434 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
435 0 : if (isSelectAHC) {
436 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
437 0 : u32 level0ServerIndex = level0CommInfo.localRank;
438 :
439 0 : CommPlane commPlaneLevel1 = COMM_LEVEL1;
440 0 : CHK_RET(CheckCommSize(commPlaneLevel1, level0ServerIndex + 1));
441 0 : level1CommInfo = GetSubCommInfo(commPlaneLevel1, level0ServerIndex);
442 0 : u32 level1RankSize = level1CommInfo.localRankSize;
443 0 : HCCL_INFO("Getlevel1CommRank. level1RankSize[%u]", level1RankSize);
444 0 : return HCCL_SUCCESS;
445 0 : }
446 0 : if (CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1) != HCCL_SUCCESS) {
447 0 : HCCL_INFO("[nslbdp] Getlevel1CommRank size not match.");
448 0 : return HCCL_E_UNAVAIL;
449 : }
450 0 : level1CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
451 :
452 0 : return HCCL_SUCCESS;
453 : }
454 :
455 0 : HcclResult CollAllGatherRingFor91093Executor::SelectTempAlg(std::unique_ptr<AlgTemplateBase> &level1TempAlg, u32 level1RankSize)
456 : {
457 0 : HCCL_INFO("[nslbdp] Entry SelectTempAlg, level1RankSize = [%u].", level1RankSize);
458 0 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
459 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
460 0 : if (isSelectAHC) {
461 0 : CommPlane commPlaneLevel1 = COMM_LEVEL1;
462 : // 获取通信域分组信息
463 0 : std::vector<std::vector<std::vector<u32>>> globalSubGroups;
464 0 : std::map<AHCConcOpType, TemplateType> ahcAlgOption;
465 0 : CHK_RET(topoMatcher_->GetGlobalSubGroups(commPlaneLevel1, globalSubGroups));
466 0 : topoMatcher_->GetAHCAlgOption(ahcAlgOption);
467 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
468 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_AHC, dispatcher_);
469 0 : HCCL_INFO("allgather comm: using ahc algo inter-server.");
470 : } else {
471 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_AHC_BROKE, dispatcher_);
472 0 : HCCL_INFO("allgather comm: using ahc-broke algo inter-server.");
473 : }
474 0 : CHK_SMART_PTR_NULL(level1TempAlg);
475 0 : CHK_RET(level1TempAlg->Prepare(NSLBDP_MIN_COUNT, globalSubGroups, ahcAlgOption));
476 0 : return HCCL_SUCCESS;
477 0 : }
478 0 : if (level1RankSize > 1) {
479 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
480 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
481 0 : TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
482 0 : HCCL_INFO("AllGather ring: using nonuniform-bruck algo inter-superPod.");
483 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
484 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
485 0 : TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
486 0 : HCCL_INFO("AllGather ring: using nonuniform-hierarchical-ring algo inter-superPod.");
487 : } else {
488 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
489 0 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
490 0 : HCCL_INFO("AllGather ring: using ring algo inter-superPod.");
491 : }
492 0 : CHK_SMART_PTR_NULL(level1TempAlg);
493 0 : return HCCL_SUCCESS;
494 : }
495 0 : return HCCL_E_UNAVAIL;
496 : }
497 :
498 : REGISTER_EXEC("AllGatherRingFor91093Executor", AllGatherRingFor91093, CollAllGatherRingFor91093Executor);
499 :
500 : } // namespace hccl
|
