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_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollAllReduceRingExecutor::CollAllReduceRingExecutor(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
18 : {
19 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
20 0 : && topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
21 0 : DMAReduceFlag_ = true;
22 : } else {
23 0 : DMAReduceFlag_ = false;
24 : }
25 0 : }
26 :
27 0 : HcclResult CollAllReduceRingExecutor::CalcStreamNum(u32& streamNum)
28 : {
29 0 : u32 totalStreamNum = 1U;
30 0 : if (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_8P_RING) {
31 0 : totalStreamNum = LEVEL0_PLANE_NUM_IN_8PRING;
32 0 : } else if (algType_.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_NP_SINGLE_RING) {
33 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
34 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
35 0 : totalStreamNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
36 : } else {
37 0 : totalStreamNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
38 : }
39 : }
40 : }
41 0 : streamNum = totalStreamNum - 1;
42 0 : HCCL_INFO("[CollAllReduceRingExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
43 0 : return HCCL_SUCCESS;
44 : }
45 :
46 0 : HcclResult CollAllReduceRingExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
47 : {
48 0 : TransportMemType inputType = TransportMemType::RESERVED;
49 0 : TransportMemType outputType = TransportMemType::RESERVED;
50 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
51 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
52 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
53 0 : return HCCL_SUCCESS;
54 : }
55 :
56 0 : HcclResult CollAllReduceRingExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
57 : {
58 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
59 0 : inputType = TransportMemType::CCL_INPUT;
60 0 : outputType = TransportMemType::CCL_OUTPUT;
61 : } else {
62 0 : inputType = TransportMemType::PARAM_INPUT;
63 0 : outputType = TransportMemType::PARAM_OUTPUT;
64 : }
65 0 : HCCL_INFO(
66 : "[CollAllReduceRingExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d].", tag_.c_str(),
67 : inputType, outputType);
68 0 : return HCCL_SUCCESS;
69 : }
70 :
71 0 : HcclResult CollAllReduceRingExecutor::CalcLevel0CommInfo(
72 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
73 : {
74 0 : HCCL_INFO("[CollAllReduceRingExecutor][CalcLevel0CommInfo]tag[%s] start.", tag_.c_str());
75 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
76 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
77 0 : HCCL_INFO("[CollAllReduceRingExecutor][CalcLevel0CommInfo]tag[%s] Calc RingComm finish.", tag_.c_str());
78 0 : return HCCL_SUCCESS;
79 0 : }
80 :
81 0 : bool CollAllReduceRingExecutor::IsHugeData(const u64 curSize)
82 : {
83 0 : bool hugeData = curSize / topoAttr_.deviceNumPerAggregation / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE
84 0 : || curSize > SDMA_SEND_MAX_SIZE;
85 0 : return hugeData;
86 : }
87 :
88 0 : bool CollAllReduceRingExecutor::IsSmallData([[maybe_unused]] const u64 totalSize, const u64 curSize)
89 : {
90 0 : bool smallData = DMAReduceFlag_ ? false : IsAllReduceSmallData(curSize);
91 0 : return smallData;
92 : }
93 :
94 0 : HcclResult CollAllReduceRingExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
95 : {
96 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllReduceRingExecutor][Run]The CollAllReduceRingExecutor starts.");
97 0 : u32 perDataSize = 0;
98 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
99 :
100 0 : std::vector<std::vector<Slice>> multRingsSliceZero; // 数据基于该rank上环0的偏移
101 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
102 0 : u32 ringNum
103 0 : = (topoType_ == TopoType::TOPO_TYPE_8P_RING) ? LEVEL0_PLANE_NUM_IN_8PRING : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
104 :
105 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
106 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
107 0 : u32 sliceNum = level0CommInfo.localRankSize;
108 : // 根据数据量计算每个环上数据的偏移和大小
109 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(execMem.count, perDataSize, sliceNum, 0, dataSegsSlice));
110 :
111 : /* 三步算法step1：外层 - 节点内 reduce-scatter */
112 : /* 外层:reduce_scatter */
113 0 : if (ringNum == LEVEL0_PLANE_NUM_IN_8PRING) {
114 : // 构造ring algorithm对应的reduce-scatter实例
115 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, topoAttr_.nicList);
116 0 : CHK_PRT_RET(
117 : multRingsSliceZero.size() != ringNum,
118 : HCCL_ERROR(
119 : "[CollAllReduceRingExecutor]"
120 : "ringNum[%u] !=multRingsSliceZero size[%zu]",
121 : ringNum, multRingsSliceZero.size()),
122 : HCCL_E_INTERNAL);
123 : } else {
124 0 : multRingsSliceZero.push_back(dataSegsSlice); // 应该offset全为0，而大小和dataSegsSlice中一样,里面的offset不使用
125 : }
126 :
127 0 : HcomCollOpInfo* reduceScatterOpInfoPtr = nullptr;
128 : // 第一步的reducescatter输出放在CCL buffer上，通过设置nullptr指示不做最后一步的DMA削减动作
129 0 : HcomCollOpInfo reduceScatterOpInfo
130 0 : = {"", execMem.inputPtr, nullptr, execMem.count, param.DataDes.dataType, param.root, param.reduceType, 0};
131 0 : if (DMAReduceFlag_) {
132 0 : reduceScatterOpInfoPtr = &reduceScatterOpInfo;
133 : }
134 0 : CHK_RET(MultiRingReduceScatter(
135 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.reduceType,
136 : multRingsSliceZero, param.stream, PROF_STAGE_0, 0, reduceScatterOpInfoPtr));
137 :
138 0 : HCCL_INFO("AllReduce ringhd stage0 run success");
139 :
140 : /* 三步算法step2: 内层 - 节点间 allreduce */
141 : u64 hdSize;
142 : u32 segmentIdx;
143 : u32 commIndex;
144 0 : CHK_RET(PrepareLevel1CommInfo(segmentIdx, commIndex, hdSize, level0CommInfo, multRingsSliceZero, param.tag));
145 :
146 0 : u64 hdCount = hdSize / perDataSize;
147 0 : auto nicList = topoAttr_.nicList;
148 0 : bool isMultiNic = topoType_ == TopoType::TOPO_TYPE_8P_RING && nicList.size() != DEVICE_EIGHT;
149 0 : std::vector<u32>::iterator iterNic = std::find(nicList.begin(), nicList.end(), topoAttr_.devicePhyId);
150 0 : bool innRunRet = isMultiNic && (iterNic == nicList.end());
151 0 : HCCL_DEBUG("[CollAllReduceRingExecutor][KernelRun]isMultiNic is %d, innRunRet is %d", isMultiNic, innRunRet);
152 0 : if (!innRunRet) { // 满足以下条件, 不做server间通信: 1. 8P ring的拓扑 2. 网口不满配 3. 当前device不出网口
153 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
154 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
155 :
156 0 : DeviceMem allreduceInput = execMem.inputMem.range(dataSegsSlice[segmentIdx].offset, hdSize);
157 0 : CHK_SMART_PTR_NULL(allreduceInput);
158 0 : DeviceMem allreduceOutput = execMem.outputMem.range(dataSegsSlice[segmentIdx].offset, hdSize);
159 0 : CHK_SMART_PTR_NULL(allreduceOutput);
160 :
161 0 : u64 reduceAttr = GetReduceAttr(allreduceInput, allreduceOutput, param.DataDes.dataType, param.reduceType);
162 :
163 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
164 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
165 : level1TempAlg
166 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
167 0 : HCCL_INFO("AllReduce ring: using ring algo inter-server.");
168 0 : CHK_SMART_PTR_NULL(level1TempAlg);
169 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
170 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
171 0 : u64 curSize = execMem.count * SIZE_TABLE[param.DataDes.dataType]; // 单位 byte
172 0 : HCCL_DEBUG(
173 : "AllReduce ring: curSize[%llu] deviceNumPerAggregation[%u] commLevel0Size[%u]", curSize,
174 : topoAttr_.deviceNumPerAggregation, level0CommInfo.localRankSize);
175 0 : if (curSize / topoAttr_.deviceNumPerAggregation <= NHR_ALLREDUCE_SMALL_SIZE) {
176 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
177 0 : TemplateType::TEMPLATE_ALL_REDUCE_NHR_ONESHOT, dispatcher_);
178 : } else {
179 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
180 0 : TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
181 : }
182 0 : HCCL_INFO("AllReduce ring: using nhr algo inter-server.");
183 0 : CHK_SMART_PTR_NULL(level1TempAlg);
184 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
185 0 : level1TempAlg->CloseBarrier();
186 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
187 : level1TempAlg
188 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR_V1, dispatcher_);
189 0 : HCCL_INFO("AllReduce ring: using nhr_v1 algo inter-server.");
190 0 : CHK_SMART_PTR_NULL(level1TempAlg);
191 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
192 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
193 : level1TempAlg
194 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB, dispatcher_);
195 0 : HCCL_INFO("AllReduce ring: using nonuniform-bruck algo inter-server.");
196 0 : CHK_SMART_PTR_NULL(level1TempAlg);
197 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
198 : } else {
199 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
200 0 : TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
201 0 : HCCL_INFO("AllReduce ring: using Recursive halving-doubling algo inter-server.");
202 0 : CHK_SMART_PTR_NULL(level1TempAlg);
203 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
204 : }
205 :
206 0 : CHK_SMART_PTR_NULL(level1TempAlg);
207 :
208 : // 节点间的hd 使用环0来记录
209 0 : CHK_RET(level1TempAlg->Prepare(
210 : allreduceInput, allreduceOutput, allreduceOutput, hdCount, param.DataDes.dataType, param.stream,
211 : param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), dataSegsSlice[segmentIdx].offset));
212 :
213 0 : CHK_RET(level1TempAlg->RegisterProfiler(
214 : (level1CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_1,
215 : HCCL_EXEC_STEP_NOT_SET, param.stream));
216 :
217 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
218 0 : }
219 0 : HCCL_INFO("AllReduce ringhd stage1 run success");
220 :
221 : /* 三步算法step3：外层 - 节点内 allgather */
222 0 : HcomCollOpInfo* allgatherOpInfoPtr = nullptr;
223 : // 第三步的allgather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
224 0 : HcomCollOpInfo allgatherOpInfo
225 0 : = {"", nullptr, execMem.outputPtr, execMem.count, param.DataDes.dataType, param.root, param.reduceType, 0};
226 0 : if (DMAReduceFlag_) {
227 0 : allgatherOpInfoPtr = &allgatherOpInfo;
228 : }
229 0 : CHK_RET(MultiRingAllGather(
230 : param.tag, execMem.inputMem, execMem.outputMem, hdCount, param.DataDes.dataType, multRingsSliceZero,
231 : param.stream, PROF_STAGE_2, 0, allgatherOpInfoPtr));
232 0 : HCCL_INFO("AllReduce ringhd stage2 run success");
233 0 : return HCCL_SUCCESS;
234 0 : }
235 0 : HcclResult CollAllReduceRingExecutor::Getlevel1CommRank(SubCommInfo& level1CommInfo)
236 : {
237 0 : if (CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1) != HCCL_SUCCESS) {
238 0 : return HCCL_E_UNAVAIL;
239 : }
240 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
241 0 : u32 ringNum
242 0 : = (topoType_ == TopoType::TOPO_TYPE_8P_RING) ? LEVEL0_PLANE_NUM_IN_8PRING : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
243 0 : u32 commIndex = (ringNum == LEVEL0_PLANE_NUM_IN_8PRING) ? topoAttr_.devicePhyId : level0CommInfo.localRank;
244 :
245 0 : if (CheckCommSize(COMM_LEVEL1, commIndex + 1) != HCCL_SUCCESS) {
246 0 : return HCCL_E_UNAVAIL;
247 : }
248 0 : level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
249 :
250 0 : return HCCL_SUCCESS;
251 0 : }
252 :
253 0 : HcclResult CollAllReduceRingExecutor::SelectTempAlg(std::unique_ptr<AlgTemplateBase>& level1TempAlg, u32 level1RankSize)
254 : {
255 0 : if (level1RankSize > 1) {
256 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
257 : level1TempAlg
258 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
259 0 : HCCL_INFO("AllReduce ring: using ring algo inter-server.");
260 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
261 : level1TempAlg
262 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
263 0 : HCCL_INFO("AllReduce ring: using nhr algo inter-server.");
264 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
265 : level1TempAlg
266 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR_V1, dispatcher_);
267 0 : HCCL_INFO("AllReduce ring: using nhr_v1 algo inter-server.");
268 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
269 : // 获取通信域分组信息
270 : level1TempAlg
271 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_AHC, dispatcher_);
272 0 : HCCL_INFO("AllReduce ring: using ahc algo inter-server.");
273 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
274 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
275 0 : TemplateType::TEMPLATE_ALL_REDUCE_AHC_BROKE, dispatcher_);
276 0 : HCCL_INFO("AllReduce ring: using ahc-broke algo inter-server.");
277 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
278 : level1TempAlg
279 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB, dispatcher_);
280 0 : HCCL_INFO("AllReduce ring: using nonuniform-bruck algo inter-server.");
281 : } else {
282 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
283 0 : TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
284 0 : HCCL_INFO("AllReduce ring: using Recursive halving-doubling algo inter-server.");
285 : }
286 0 : CHK_SMART_PTR_NULL(level1TempAlg);
287 0 : return HCCL_SUCCESS;
288 : }
289 0 : return HCCL_E_UNAVAIL;
290 : }
291 : REGISTER_EXEC("AllReduceRingExecutor", AllReduceRing, CollAllReduceRingExecutor);
292 :
293 : } // namespace hccl
|
