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_executor.h"
12 :
13 : namespace hccl {
14 0 : CollAllGatherRingExecutor::CollAllGatherRingExecutor(
15 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 0 : : CollAllGatherExecutor(dispatcher, topoMatcher)
17 : {
18 0 : DMAReduceFlag_ = false;
19 0 : }
20 :
21 0 : HcclResult CollAllGatherRingExecutor::CalcStreamNum(u32& streamNum)
22 : {
23 0 : u32 totalStreamNum = 1U;
24 0 : if (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_8P_RING) {
25 0 : totalStreamNum = LEVEL0_PLANE_NUM_IN_8PRING;
26 : }
27 0 : streamNum = totalStreamNum - 1;
28 0 : HCCL_INFO("[CollAllGatherRingExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
29 0 : return HCCL_SUCCESS;
30 : }
31 :
32 0 : HcclResult CollAllGatherRingExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
33 : {
34 0 : TransportMemType inputType = TransportMemType::RESERVED;
35 0 : TransportMemType outputType = TransportMemType::RESERVED;
36 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
37 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
38 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
39 0 : return HCCL_SUCCESS;
40 : }
41 :
42 0 : HcclResult CollAllGatherRingExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
43 : {
44 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
45 0 : inputType = TransportMemType::CCL_INPUT;
46 0 : outputType = TransportMemType::CCL_OUTPUT;
47 : } else {
48 0 : inputType = TransportMemType::PARAM_INPUT;
49 0 : outputType = TransportMemType::PARAM_OUTPUT;
50 : }
51 0 : HCCL_INFO(
52 : "[CollAllGatherRingExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]", tag_.c_str(),
53 : inputType, outputType);
54 0 : return HCCL_SUCCESS;
55 : }
56 :
57 0 : HcclResult CollAllGatherRingExecutor::CalcLevel0CommInfo(
58 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
59 : {
60 0 : HCCL_INFO("[CollAllGatherRingExecutor][CalcLevel0CommInfo]tag[%s] start", tag_.c_str());
61 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
62 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
63 0 : HCCL_INFO("[CollAllGatherRingExecutor][CalcLevel0CommInfo]tag[%s] Calc RingComm finish", tag_.c_str());
64 0 : return HCCL_SUCCESS;
65 0 : }
66 :
67 0 : u64 CollAllGatherRingExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
68 : {
69 0 : u64 maxCountPerLoop = cclBuffSize / (topoAttr_.userRankSize * unitSize);
70 0 : return maxCountPerLoop;
71 : }
72 :
73 0 : HcclResult CollAllGatherRingExecutor::SelectAlgorithmTempAlg(std::unique_ptr<AlgTemplateBase>& level1TempAlg)
74 : {
75 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
76 : level1TempAlg
77 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
78 0 : HCCL_INFO("AllGather ring: using ring algo inter-server.");
79 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
80 : level1TempAlg
81 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
82 0 : HCCL_INFO("AllGather ring: using nhr algo inter-server.");
83 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
84 : level1TempAlg
85 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHRV1, dispatcher_);
86 0 : HCCL_INFO("AllGather ring: using nhr_v1 algo inter-server.");
87 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
88 : level1TempAlg
89 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
90 0 : HCCL_INFO("AllGather ring: using nonuniform-bruck algo inter-server.");
91 : } else {
92 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
93 0 : TemplateType::TEMPLATE_ALL_GATHER_RECURSIVE_HALVING_DOUBLING, dispatcher_);
94 0 : HCCL_INFO("AllGather ring: using halving-doubling algo inter-server.");
95 : }
96 0 : CHK_SMART_PTR_NULL(level1TempAlg);
97 0 : return HCCL_SUCCESS;
98 : }
99 :
100 0 : HcclResult CollAllGatherRingExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
101 : {
102 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherRingExecutor][KernelRun]The AllGatherRingExecutor starts.");
103 0 : u32 perDataSize = 0;
104 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
105 0 : CHK_PRT_RET(
106 : perDataSize == 0,
107 : HCCL_ERROR(
108 : "[CollAllGatherRingExecutor][KernelRun]errNo[0x%016llx] datatype[%d] is invalid",
109 : HCCL_ERROR_CODE(HCCL_E_PARA), param.DataDes.dataType),
110 : HCCL_E_PARA);
111 :
112 : // 获取子通信域的信息
113 0 : u32 ringNum
114 0 : = (topoType_ == TopoType::TOPO_TYPE_8P_RING) ? LEVEL0_PLANE_NUM_IN_8PRING : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
115 :
116 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
117 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
118 0 : u32 commIndex = (ringNum == LEVEL0_PLANE_NUM_IN_8PRING) ? topoAttr_.devicePhyId : level0CommInfo.localRank;
119 0 : u32 level0RankSize = level0CommInfo.localRankSize;
120 :
121 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
122 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
123 0 : u32 serverIndex = level1CommInfo.localRank;
124 :
125 : // 第一步，如果非DMA消减，将数据从input内存拷贝到output内存的对应位置
126 0 : u64 inputMemSize = execMem.inputMem.size();
127 0 : u64 baseOffset = serverIndex * inputMemSize * level0RankSize;
128 0 : u64 level0Offset = commIndex * inputMemSize;
129 0 : DeviceMem dstMem = execMem.outputMem.range(baseOffset + level0Offset, inputMemSize);
130 0 : CHK_SMART_PTR_NULL(dstMem);
131 :
132 0 : HcclResult ret = HcclD2DMemcpyAsync(dispatcher_, dstMem, execMem.inputMem, const_cast<Stream&>(param.stream));
133 0 : CHK_PRT_RET(
134 : ret != HCCL_SUCCESS,
135 : HCCL_ERROR(
136 : "[CollAllGatherRingExecutor][KernelRun]AllGather 8PringHD memcpy Failed, "
137 : "Offset[%llu], Size[%llu]",
138 : baseOffset + level0Offset, inputMemSize),
139 : ret);
140 :
141 : // 第二步，各个AI Server 内 multi ring AllGather
142 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
143 0 : std::vector<std::vector<Slice>> multRingsSliceZero; // 数据基于该rank上环0的偏移
144 0 : u32 sliceNum = level0RankSize;
145 0 : CHK_RET(PrepareAllgatherSlice(sliceNum, inputMemSize, dataSegsSlice));
146 :
147 : // 多环数据切分
148 0 : if (ringNum == LEVEL0_PLANE_NUM_IN_8PRING) {
149 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag);
150 : } else {
151 0 : multRingsSliceZero.push_back(dataSegsSlice);
152 : }
153 0 : CHK_PRT_RET(
154 : multRingsSliceZero.size() != ringNum,
155 : HCCL_ERROR(
156 : "[CollAllGatherRingExecutor][KernelRun]ringNum[%u] != multRingsSliceZero size[%zu]", ringNum,
157 : multRingsSliceZero.size()),
158 : HCCL_E_INTERNAL);
159 :
160 : // 抽取当前用于多环AllGather 的output内存数据
161 0 : DeviceMem currentOutputMem = execMem.outputMem.range(baseOffset, inputMemSize * level0RankSize);
162 0 : CHK_SMART_PTR_NULL(currentOutputMem);
163 :
164 0 : CHK_RET(ActiveSlaveStreams(param.stream));
165 :
166 0 : CHK_RET(MultiRingAllGather(
167 : param.tag, execMem.inputMem, currentOutputMem, execMem.count, param.DataDes.dataType, multRingsSliceZero,
168 : param.stream, PROF_STAGE_1, baseOffset, nullptr));
169 :
170 0 : HCCL_INFO("AllGather 8PringHD level0 run success");
171 :
172 : // 第三步， AI server 间 recursive halving doubling AllGather
173 0 : u64 hdSize = 0;
174 0 : std::vector<u32> nicList = const_cast<std::vector<u32>&>(topoAttr_.nicList);
175 0 : std::vector<u32>::iterator iterNic = std::find(nicList.begin(), nicList.end(), topoAttr_.devicePhyId);
176 0 : if (iterNic != nicList.end()) {
177 0 : hdSize = inputMemSize * level0RankSize;
178 : }
179 0 : u64 hdCount = hdSize / perDataSize;
180 :
181 0 : bool isMultiNic = topoType_ == TopoType::TOPO_TYPE_8P_RING && nicList.size() != DEVICE_EIGHT;
182 0 : bool innRunRet = isMultiNic && (iterNic == nicList.end());
183 0 : if (!innRunRet) { // 满足以下条件, 不做server间通信: 1. 8P ring的拓扑 2. 网口不满配 3. 当前device不出网口
184 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
185 0 : CHK_RET(SelectAlgorithmTempAlg(level1TempAlg));
186 :
187 : // 此处虽然带入inputMem作为scratch mem, 但inputMem 不能被使用
188 0 : CHK_RET(level1TempAlg->Prepare(
189 : execMem.outputMem, execMem.outputMem, execMem.inputMem, hdCount, param.DataDes.dataType, param.stream,
190 : HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, std::vector<Slice>(COMM_INDEX_0), 0));
191 :
192 0 : u32 rankSize = level1CommInfo.localRankSize;
193 0 : CHK_RET(level1TempAlg->RegisterProfiler(
194 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + serverIndex, PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET,
195 : param.stream));
196 :
197 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
198 0 : }
199 0 : HCCL_INFO("AllGather 8PringHD level1 run success");
200 :
201 : // 网口裁剪：AI server 内多网口的allgather
202 0 : if (topoType_ == TopoType::TOPO_TYPE_8P_RING && nicList.size() != DEVICE_EIGHT) {
203 0 : CHK_RET(ActiveSlaveStreams(param.stream)); // 为什么要active两遍
204 :
205 0 : u32 perDataSize = 0;
206 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
207 0 : u64 tempCount = execMem.outputMem.size() / perDataSize;
208 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(tempCount, perDataSize, sliceNum, 0, dataSegsSlice));
209 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, nicList);
210 0 : CHK_PRT_RET(
211 : multRingsSliceZero.size() != ringNum,
212 : HCCL_ERROR(
213 : "[CollAllGatherRingExecutor][KernelRun]"
214 : "ringNum[%u] != multRingsSliceZero size[%zu]",
215 : ringNum, multRingsSliceZero.size()),
216 : HCCL_E_INTERNAL);
217 :
218 0 : CHK_RET(MultiRingAllGather(
219 : param.tag, execMem.outputMem, execMem.outputMem, tempCount / DEVICE_EIGHT, param.DataDes.dataType,
220 : multRingsSliceZero, param.stream, PROF_STAGE_1));
221 :
222 0 : HCCL_INFO("AllGather 8PringHD level1 chunk run success");
223 : }
224 0 : return HCCL_SUCCESS;
225 0 : }
226 0 : HcclResult CollAllGatherRingExecutor::Getlevel1CommRank(SubCommInfo& level1CommInfo)
227 : {
228 0 : if (CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1) != HCCL_SUCCESS) {
229 0 : return HCCL_E_UNAVAIL;
230 : }
231 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
232 0 : u32 ringNum
233 0 : = (topoType_ == TopoType::TOPO_TYPE_8P_RING) ? LEVEL0_PLANE_NUM_IN_8PRING : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
234 0 : u32 commIndex = (ringNum == LEVEL0_PLANE_NUM_IN_8PRING) ? topoAttr_.devicePhyId : level0CommInfo.localRank;
235 :
236 0 : if (CheckCommSize(COMM_LEVEL1, commIndex + 1) != HCCL_SUCCESS) {
237 0 : return HCCL_E_UNAVAIL;
238 : }
239 0 : level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
240 :
241 0 : return HCCL_SUCCESS;
242 0 : }
243 :
244 0 : HcclResult CollAllGatherRingExecutor::SelectTempAlg(std::unique_ptr<AlgTemplateBase>& level1TempAlg, u32 level1RankSize)
245 : {
246 0 : if (level1RankSize > 1) {
247 0 : CHK_RET(SelectAlgorithmTempAlg(level1TempAlg));
248 : }
249 0 : return HCCL_E_UNAVAIL;
250 : }
251 :
252 : REGISTER_EXEC("AllGatherRingExecutor", AllGatherRing, CollAllGatherRingExecutor);
253 :
254 : } // namespace hccl
|
