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