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_zerocopy_executor.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 :
16 : constexpr u32 TWO_RING = 2;
17 :
18 0 : CollAllReduceRingZerocopyExecutor::CollAllReduceRingZerocopyExecutor(const HcclDispatcher dispatcher,
19 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
20 0 : : CollAllReduceExecutor(dispatcher, topoMatcher)
21 : {
22 0 : DMAReduceFlag_ = true; // 设为true，以禁用RunLoop中的本地拷贝
23 0 : desc_.isZeroCopy = true;
24 0 : desc_.deterministic = 1;
25 0 : desc_.level1SupportedAlgos = {
26 : AlgTypeLevel1::ALG_LEVEL1_NHR,
27 : AlgTypeLevel1::ALG_LEVEL1_NB,
28 : AlgTypeLevel1::ALG_LEVEL1_RING,
29 : AlgTypeLevel1::ALG_LEVEL1_AHC,
30 : AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE
31 0 : };
32 0 : desc_.level2SupportedAlgos = {
33 : AlgTypeLevel2::ALG_LEVEL2_NHR,
34 : AlgTypeLevel2::ALG_LEVEL2_NB,
35 : AlgTypeLevel2::ALG_LEVEL2_RING,
36 : AlgTypeLevel2::ALG_LEVEL2_HD
37 0 : };
38 0 : }
39 :
40 0 : HcclResult CollAllReduceRingZerocopyExecutor::CalcStreamNum(u32& streamNum)
41 : {
42 0 : u32 totalStreamNum = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
43 : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
44 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
45 0 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
46 : }
47 0 : streamNum = totalStreamNum - 1;
48 0 : HCCL_INFO("[CollAllReduceRingZerocopyExecutor][CalcStreamNum] tag[%s] streamNum_[%u]",
49 : tag_.c_str(), streamNum);
50 0 : return HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult CollAllReduceRingZerocopyExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
54 : {
55 0 : TransportMemType inputType = TransportMemType::RESERVED;
56 0 : TransportMemType outputType = TransportMemType::RESERVED;
57 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
58 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
59 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
60 0 : CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
61 0 : return HCCL_SUCCESS;
62 : }
63 :
64 0 : HcclResult CollAllReduceRingZerocopyExecutor::CalcTransportMemType(TransportMemType &inputType,
65 : TransportMemType &outputType)
66 : {
67 0 : inputType = TransportMemType::CCL_INPUT;
68 0 : outputType = TransportMemType::CCL_OUTPUT;
69 0 : HCCL_INFO("[CollAllReduceRingZerocopyExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
70 : tag_.c_str(), inputType, outputType);
71 0 : return HCCL_SUCCESS;
72 : }
73 :
74 0 : HcclResult CollAllReduceRingZerocopyExecutor::CalcLevel0CommInfo(TransportMemType inputType,
75 : TransportMemType outputType,
76 : std::vector<LevelNSubCommTransport>& opTransport)
77 : {
78 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
79 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
80 0 : LevelNSubCommTransport &commTransportLevel0 = opTransport[COMM_LEVEL0];
81 0 : for (u32 subCommIndex = 0; subCommIndex < commTransportLevel0.size(); subCommIndex++) {
82 0 : commTransportLevel0[subCommIndex].isZeroCopy = true;
83 : }
84 0 : return HCCL_SUCCESS;
85 0 : }
86 :
87 0 : HcclResult CollAllReduceRingZerocopyExecutor::DoubleRingReduceScatter(const std::string &tag,
88 : DeviceMem inputMem, DeviceMem outputMem, const u64 count, const HcclDataType dataType,
89 : const HcclReduceOp reductionOp, const std::vector<std::vector<Slice>> multRingsSliceZero, Stream stream,
90 : s32 profStage, const u64 baseOffset, const HcomCollOpInfo *opInfo,
91 : const std::vector<std::vector<Slice>> multRingsUserMemSlice)
92 : {
93 : (void) tag;
94 0 : HCCL_INFO("[CollAllReduceRingZerocopyExecutor][DoubleRingReduceScatter] DoubleRingReduceScatter starts");
95 :
96 0 : u64 reduceAttr = GetReduceAttr(inputMem, outputMem, dataType, reductionOp);
97 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
98 0 : SubCommInfo level0RingCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
99 :
100 : // 适配AlignedDoubleRing的入参
101 0 : u32 ringSize = multRingsSliceZero[0].size();
102 0 : std::vector<std::vector<u32>> rankOrders(TWO_RING, std::vector<u32>(ringSize));
103 0 : for (u32 i = 0; i < ringSize; i++) {
104 0 : rankOrders[0][i] = i;
105 0 : rankOrders[1][i] = (i == 0) ? 0 : (ringSize - i);
106 : }
107 :
108 : // 执行算法编排
109 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
110 0 : TemplateType::TEMPLATE_REDUCESCATTER_DB_RING, dispatcher_);
111 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_DB_RING in COMM_LEVEL0", __func__);
112 0 : CHK_SMART_PTR_NULL(tempAlg);
113 0 : CHK_RET(tempAlg->Prepare(inputMem, inputMem, outputMem, count, dataType, stream,
114 : multRingsSliceZero, reductionOp, LEVEL0_BRIDGE_RANK_ID, baseOffset, false,
115 : reduceAttr, opInfo, topoAttr_.userRank, algResResp_->slaveStreams, algResResp_->notifiesMain,
116 : algResResp_->notifiesAux, rankOrders, multRingsUserMemSlice));
117 0 : u32 ringIndexOp = COMM_INDEX_0;
118 0 : u32 rankSize = level0RingCommInfo.localRankSize;
119 0 : CHK_RET(tempAlg->RegisterProfiler(((ringIndexOp + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID) +
120 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0RingCommInfo.localRank, profStage,
121 : HCCL_EXEC_STEP_NOT_SET, stream));
122 0 : CHK_RET(RunTemplate(tempAlg, level0RingCommInfo));
123 :
124 0 : return HCCL_SUCCESS;
125 0 : }
126 :
127 0 : HcclResult CollAllReduceRingZerocopyExecutor::KernelRunIntraServerPre(const OpParam ¶m, ExecMem &execMem)
128 : {
129 0 : HCCL_CONFIG_INFO(HCCL_ALG,
130 : "[CollAllReduceRingZerocopyExecutor][Run]The CollAllReduceRingZerocopyExecutor starts");
131 0 : bool isAHCAlgo = algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
132 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE;
133 0 : CHK_RET(GetCommRankInfoNormal(level0Rank_, level0RankSize_, level1Rank_, level1RankSize_,
134 : level2Rank_, level2RankSize_, isAHCAlgo));
135 :
136 0 : u32 perDataSize = 0;
137 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
138 :
139 0 : std::vector<Slice> level0Datalices;
140 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(param.DataDes.count, perDataSize, level0RankSize_, 0, level0Datalices));
141 :
142 0 : HcomCollOpInfo reduceScatterOpInfo = {
143 0 : "", execMem.inputMem.ptr(), nullptr, execMem.count, param.DataDes.dataType, param.root, param.reduceType
144 0 : };
145 :
146 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING) {
147 : // 构造slice数据
148 0 : level0MultiRingDataSlices_ = {level0Datalices};
149 : // 执行算法编排
150 0 : CHK_RET(MultiRingReduceScatter(param.tag, execMem.inputMem, execMem.outputMem, execMem.count,
151 : param.DataDes.dataType, param.reduceType,
152 : level0MultiRingDataSlices_, param.stream, PROF_STAGE_0, 0, &reduceScatterOpInfo, level0MultiRingDataSlices_));
153 : } else {
154 0 : CHK_PRT_RET(topoType_ != TopoType::TOPO_TYPE_NP_DOUBLE_RING,
155 : HCCL_ERROR("[%s] unknown topoType: %u", __func__, topoType_), HCCL_E_NOT_SUPPORT);
156 : // 构造slice数据（适配AlignedDoubleRing算法）
157 0 : level0MultiRingDataSlices_.resize(TWO_RING);
158 0 : level0MultiRingDataSlices_[0].resize(level0RankSize_);
159 0 : level0MultiRingDataSlices_[1].resize(level0RankSize_);
160 0 : for (u32 i = 0; i < level0Datalices.size(); i++) {
161 0 : level0MultiRingDataSlices_[0][i].offset = level0Datalices[i].offset;
162 0 : level0MultiRingDataSlices_[0][i].size = level0Datalices[i].size / perDataSize / TWO_RING * perDataSize;
163 0 : u32 j = (i == 0) ? 0 : (level0RankSize_ - i);
164 0 : level0MultiRingDataSlices_[1][j].offset = level0MultiRingDataSlices_[0][i].offset + level0MultiRingDataSlices_[0][i].size;
165 0 : level0MultiRingDataSlices_[1][j].size = level0Datalices[i].size - level0MultiRingDataSlices_[0][i].size;
166 : }
167 : // 执行算法编排
168 0 : CHK_RET(DoubleRingReduceScatter(param.tag, execMem.inputMem, execMem.outputMem, execMem.count,
169 : param.DataDes.dataType, param.reduceType, level0MultiRingDataSlices_, param.stream, PROF_STAGE_0,
170 : 0, &reduceScatterOpInfo, level0MultiRingDataSlices_));
171 : }
172 :
173 0 : HCCL_INFO("AllReduce double ring stage0 run success");
174 0 : return HCCL_SUCCESS;
175 0 : }
176 :
177 :
178 0 : HcclResult CollAllReduceRingZerocopyExecutor::KernelRunInterServer(const OpParam ¶m, ExecMem &execMem)
179 : {
180 0 : u32 perDataSize = 0;
181 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
182 0 : Stream stream = param.stream;
183 :
184 : // copy data from user_in -> ccl_in
185 0 : std::vector<Slice> level0Datalices;
186 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(param.DataDes.count, perDataSize, level0RankSize_, 0, level0Datalices));
187 0 : u64 level1DataSize = execMem.count * perDataSize;
188 0 : DeviceMem dstMem = execMem.inputMem.range(0, level1DataSize);
189 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8 *>(execMem.inputPtr) + level0Datalices[level0Rank_].offset,
190 0 : level1DataSize);
191 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream));
192 0 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
193 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
194 0 : if (topoAttr_.superPodNum <= 1 || isSelectAHC) {
195 0 : CHK_RET(KernelRunInterServerAllReduceSingleSuperpod(param, execMem, level1DataSize));
196 0 : } else {
197 0 : CHK_RET(KernelRunInterServerAllReduceMultiSuperpod(param, execMem, level1DataSize));
198 : }
199 :
200 : // copy results from ccl_out -> user_out
201 0 : srcMem = execMem.outputMem.range(0, level1DataSize);
202 0 : dstMem = DeviceMem::create(static_cast<u8 *>(execMem.outputPtr) + level0Datalices[level0Rank_].offset,
203 0 : level1DataSize);
204 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream));
205 :
206 0 : HCCL_INFO("AllReduce double ring stage2 run success");
207 0 : return HCCL_SUCCESS;
208 0 : }
209 :
210 : // 单超节点场景，Level1直接执行AllReduce编排
211 0 : HcclResult CollAllReduceRingZerocopyExecutor::KernelRunInterServerAllReduceSingleSuperpod(const OpParam ¶m, const ExecMem &execMem, const u64 level1DataSize)
212 : {
213 : // 获取Level1通信域
214 0 : bool isSelectAHC = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
215 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE);
216 0 : CommPlane commPlaneLevel1 = isSelectAHC ? COMM_LEVEL1_AHC : COMM_LEVEL1;
217 0 : CHK_RET(CheckCommSize(commPlaneLevel1, level0Rank_ + 1));
218 0 : SubCommInfo level1CommInfo = GetSubCommInfo(commPlaneLevel1, level0Rank_);
219 :
220 : // 获取Template
221 0 : u32 perDataSize = 0;
222 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
223 0 : u64 level1DataCount = level1DataSize / perDataSize;
224 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
225 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
226 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING,
227 0 : dispatcher_);
228 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_RING in COMM_LEVEL1", __func__);
229 0 : CHK_SMART_PTR_NULL(level1TempAlg);
230 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC || algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
231 : // 获取通信域分组信息
232 0 : std::vector<std::vector<std::vector<u32>>> globalSubGroups;
233 0 : std::map<AHCConcOpType, TemplateType> ahcAlgOption;
234 0 : CHK_RET(topoMatcher_->GetGlobalSubGroups(commPlaneLevel1, globalSubGroups));
235 0 : topoMatcher_->GetAHCAlgOption(ahcAlgOption);
236 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
237 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_AHC, dispatcher_);
238 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_AHC in COMM_LEVEL1", __func__);
239 : } else {
240 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_AHC_BROKE, dispatcher_);
241 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_AHC_BROKE in COMM_LEVEL1", __func__);
242 : }
243 0 : CHK_SMART_PTR_NULL(level1TempAlg);
244 0 : CHK_RET(level1TempAlg->Prepare(level1DataCount, globalSubGroups, ahcAlgOption));
245 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
246 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB,
247 0 : dispatcher_);
248 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NB in COMM_LEVEL1", __func__);
249 0 : CHK_SMART_PTR_NULL(level1TempAlg);
250 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
251 0 : HCCL_DEBUG("AllReduce ring: level1DataSize[%llu] deviceNumPerAggregation[%u] commLevel0Size[%u]",
252 : level1DataSize, topoAttr_.deviceNumPerAggregation, level0RankSize_);
253 0 : if (level1DataSize / topoAttr_.deviceNumPerAggregation <= NHR_ALLREDUCE_SMALL_SIZE) {
254 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR_ONESHOT, dispatcher_);
255 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NHR_ONESHOT in COMM_LEVEL1", __func__);
256 : } else {
257 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
258 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NHR in COMM_LEVEL1", __func__);
259 : }
260 0 : CHK_SMART_PTR_NULL(level1TempAlg);
261 : } else {
262 0 : HCCL_ERROR("AllReduce ring: unsupported level1 algtype [%s]", AlgTypeToStr(algType_).c_str());
263 0 : return HCCL_E_NOT_SUPPORT;
264 : }
265 :
266 : // 执行算法编排
267 0 : DeviceMem allreduceInput = execMem.inputMem.range(0, level1DataSize);
268 0 : DeviceMem allreduceOutput = execMem.outputMem.range(0, level1DataSize);
269 0 : u64 reduceAttr = GetReduceAttr(allreduceInput, allreduceOutput, param.DataDes.dataType, param.reduceType);
270 0 : CHK_RET(level1TempAlg->Prepare(reduceAttr));
271 0 : CHK_RET(level1TempAlg->Prepare(allreduceInput, allreduceOutput, allreduceOutput, level1DataCount,
272 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, std::vector<Slice>(0), 0));
273 0 : CHK_RET(level1TempAlg->RegisterProfiler(
274 : (level1CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank,
275 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
276 0 : CHK_RET(RunTemplate(level1TempAlg, level1CommInfo));
277 :
278 0 : return HCCL_SUCCESS;
279 0 : }
280 :
281 : // 单超节点场景，Level1先ReduceScatter，Level2再AllReduce，最后Level1再做AllGather
282 0 : HcclResult CollAllReduceRingZerocopyExecutor::KernelRunInterServerAllReduceMultiSuperpod(const OpParam ¶m, const ExecMem &execMem, const u64 level1DataSize)
283 : {
284 0 : u32 perDataSize = 0;
285 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
286 :
287 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, level0Rank_ + 1));
288 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0Rank_);
289 :
290 : // 根据数据量计算level1的数据切分
291 0 : u64 level1DataCount = level1DataSize / perDataSize;
292 0 : std::vector<Slice> level1DataSlices;
293 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(level1DataCount, perDataSize, level1RankSize_, 0, level1DataSlices));
294 0 : u64 level2DataCount = level1DataSlices[level1Rank_].size / perDataSize;
295 0 : DeviceMem level1InputMem = execMem.inputMem.range(0, level1DataSize);
296 0 : DeviceMem level1OutputMem = execMem.outputMem.range(0, level1DataSize);
297 : DeviceMem level2InputMem = level1InputMem.range(
298 0 : level1DataSlices[level1Rank_].offset, level1DataSlices[level1Rank_].size);
299 : DeviceMem level2OutputMem = level1OutputMem.range(
300 0 : level1DataSlices[level1Rank_].offset, level1DataSlices[level1Rank_].size);
301 :
302 : // Step1：超节点内、节点间做ReduceScatter
303 0 : if (level1RankSize_ > 1) {
304 : // 获取Template
305 0 : u64 reduceAttr = GetReduceAttr(level1InputMem, level1OutputMem, param.DataDes.dataType, param.reduceType);
306 0 : std::unique_ptr<AlgTemplateBase> level1RSTempAlg;
307 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
308 0 : level1RSTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_RING,
309 0 : dispatcher_);
310 0 : CHK_SMART_PTR_NULL(level1RSTempAlg);
311 0 : CHK_RET(level1RSTempAlg->Prepare(reduceAttr));
312 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_RING in COMM_LEVEL1", __func__);
313 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
314 0 : level1RSTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NB,
315 0 : dispatcher_);
316 0 : CHK_SMART_PTR_NULL(level1RSTempAlg);
317 0 : CHK_RET(level1RSTempAlg->Prepare(reduceAttr));
318 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NB in COMM_LEVEL1", __func__);
319 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
320 0 : level1RSTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_REDUCESCATTER_NHR,
321 0 : dispatcher_);
322 0 : CHK_SMART_PTR_NULL(level1RSTempAlg);
323 0 : CHK_RET(level1RSTempAlg->Prepare(reduceAttr, false));
324 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_REDUCESCATTER_NHR in COMM_LEVEL1", __func__);
325 : } else {
326 0 : HCCL_ERROR("AllReduce ring: unsupported level1 algtype [%s]", AlgTypeToStr(algType_).c_str());
327 0 : return HCCL_E_NOT_SUPPORT;
328 : }
329 :
330 : // 执行算法编排
331 0 : CHK_RET(level1RSTempAlg->Prepare(
332 : level1InputMem, level1InputMem, level1OutputMem, level1DataCount, param.DataDes.dataType,
333 : param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID, level1DataSlices, 0));
334 0 : CHK_RET(level1RSTempAlg->RegisterProfiler(
335 : (level1RankSize_ << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1Rank_,
336 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
337 0 : CHK_RET(RunTemplate(level1RSTempAlg, level1CommInfo));
338 0 : HCCL_INFO("AllReduce double ring [superpod] level1 ReduceScatter run success");
339 0 : }
340 :
341 : // Step2：超节点间做allreduce
342 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
343 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
344 : // ==> 获取Template
345 0 : std::unique_ptr<AlgTemplateBase> level2ARTempAlg;
346 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
347 0 : level2ARTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NB, dispatcher_);
348 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NB in COMM_LEVEL2", __func__);
349 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
350 0 : level2ARTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_NHR, dispatcher_);
351 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_NHR in COMM_LEVEL2", __func__);
352 0 : if (algoAttr_.isSupportAtomicWrite) {
353 0 : CHK_SMART_PTR_NULL(level2ARTempAlg);
354 0 : level2ARTempAlg->CloseBarrier();
355 : }
356 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_RING) {
357 0 : level2ARTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RING, dispatcher_);
358 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_RING in COMM_LEVEL2", __func__);
359 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_HD) {
360 0 : level2ARTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING, dispatcher_);
361 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_REDUCE_RECURSIVE_HALVING_DOUBLING in COMM_LEVEL2", __func__);
362 : } else {
363 0 : HCCL_ERROR("AllReduce ring: unsupported level2 algtype [%s]", AlgTypeToStr(algType_).c_str());
364 0 : return HCCL_E_NOT_SUPPORT;
365 : }
366 0 : CHK_SMART_PTR_NULL(level2ARTempAlg);
367 : // ==> 执行算法编排
368 0 : u64 reduceAttr = GetReduceAttr(level2InputMem, level2OutputMem, param.DataDes.dataType, param.reduceType);
369 0 : CHK_RET(level2ARTempAlg->Prepare(reduceAttr));
370 0 : CHK_RET(level2ARTempAlg->Prepare(
371 : level2InputMem, level2OutputMem, level2OutputMem, level2DataCount,
372 : param.DataDes.dataType, param.stream, param.reduceType, LEVEL0_BRIDGE_RANK_ID,
373 : std::vector<Slice>(0), level1DataSlices[level1Rank_].offset));
374 0 : CHK_RET(level2ARTempAlg->RegisterProfiler(
375 : (level2RankSize_ << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2Rank_,
376 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
377 0 : CHK_RET(RunTemplate(level2ARTempAlg, level2CommInfo));
378 0 : HCCL_INFO("AllReduce double ring [superpod] level2 AllReduce run success");
379 :
380 : // Step3：超节点内、节点间做allgather
381 0 : if (level1RankSize_ > 1) {
382 : // 获取Template
383 0 : std::unique_ptr<AlgTemplateBase> level1AGTempAlg;
384 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
385 0 : level1AGTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING,
386 0 : dispatcher_);
387 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL1", __func__);
388 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
389 0 : level1AGTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB,
390 0 : dispatcher_);
391 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL1", __func__);
392 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
393 0 : level1AGTempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR,
394 0 : dispatcher_);
395 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL1", __func__);
396 : } else {
397 0 : HCCL_ERROR("AllReduce ring: algType_[%u] is not supported", algType_.algoLevel1);
398 0 : return HCCL_E_NOT_SUPPORT;
399 : }
400 0 : CHK_SMART_PTR_NULL(level1AGTempAlg);
401 : // 执行算法编排
402 0 : CHK_RET(level1AGTempAlg->Prepare(level1OutputMem, level1OutputMem, level1OutputMem, level1DataCount,
403 : param.DataDes.dataType, param.stream,
404 : HcclReduceOp::HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, level1DataSlices, 0));
405 0 : CHK_RET(level1AGTempAlg->RegisterProfiler(
406 : (level1RankSize_ << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1Rank_,
407 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
408 0 : CHK_RET(RunTemplate(level1AGTempAlg, level1CommInfo));
409 0 : HCCL_INFO("AllReduce double ring [superpod] level1 AllGather run success");
410 0 : }
411 :
412 0 : return HCCL_SUCCESS;
413 0 : }
414 :
415 0 : HcclResult CollAllReduceRingZerocopyExecutor::DoubleRingAllGather(
416 : const std::string &tag, DeviceMem inputMem, DeviceMem outputMem,
417 : const u64 count, const HcclDataType dataType, const std::vector<std::vector<Slice> > multRingsSliceZero,
418 : Stream stream, s32 profStage, const u64 baseOffset, HcomCollOpInfo *opInfo,
419 : const std::vector<std::vector<Slice>> multRingsUserMemSlice)
420 : {
421 : (void) tag;
422 0 : HCCL_INFO("[CollAllReduceRingZerocopyExecutor][DoubleRingAllGather] DoubleRingAllGather starts");
423 :
424 : // 拿到ring环映射关系
425 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
426 0 : SubCommInfo level0RingCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
427 :
428 : // 适配AlignedDoubleRing的入参
429 0 : u32 ringSize = multRingsSliceZero[0].size();
430 0 : std::vector<std::vector<u32>> rankOrders(TWO_RING, std::vector<u32>(ringSize));
431 0 : for (u32 i = 0; i < ringSize; i++) {
432 0 : rankOrders[1][i] = (i == 0) ? 0 : (ringSize - i);
433 0 : rankOrders[0][i] = i;
434 : }
435 :
436 : // 执行算法编排
437 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALIGNED_ALL_GATHER_DOUBLE_RING in COMM_LEVEL0", __func__);
438 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
439 0 : TemplateType::TEMPLATE_ALIGNED_ALL_GATHER_DOUBLE_RING, dispatcher_);
440 0 : CHK_SMART_PTR_NULL(tempAlg);
441 0 : CHK_RET(tempAlg->Prepare(opInfo, topoAttr_.userRank, algResResp_->slaveStreams, algResResp_->notifiesMain,
442 : algResResp_->notifiesAux, rankOrders, multRingsUserMemSlice));
443 0 : CHK_RET(tempAlg->Prepare(outputMem, outputMem, inputMem, count, dataType, stream, multRingsSliceZero,
444 : HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, baseOffset));
445 0 : CHK_RET(tempAlg->RegisterProfiler(((COMM_INDEX_0 + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID) +
446 : (level0RankSize_ << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0Rank_, profStage,
447 : HCCL_EXEC_STEP_NOT_SET, stream));
448 0 : CHK_RET(RunTemplate(tempAlg, level0RingCommInfo));
449 :
450 0 : return HCCL_SUCCESS;
451 0 : }
452 :
453 0 : HcclResult CollAllReduceRingZerocopyExecutor::KernelRunIntraServerPost(const OpParam ¶m, ExecMem &execMem)
454 : {
455 0 : HcomCollOpInfo allgatherOpInfo = {
456 0 : "", nullptr, execMem.outputMem.ptr(), execMem.count, param.DataDes.dataType, param.root, param.reduceType
457 0 : };
458 :
459 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING) {
460 0 : CHK_RET(MultiRingAllGather(param.tag, execMem.inputMem, execMem.outputMem, execMem.count,
461 : param.DataDes.dataType, level0MultiRingDataSlices_, param.stream, PROF_STAGE_2,
462 : 0, &allgatherOpInfo, level0MultiRingDataSlices_));
463 : } else {
464 0 : CHK_PRT_RET(topoType_ != TopoType::TOPO_TYPE_NP_DOUBLE_RING,
465 : HCCL_ERROR("[%s] unknown topoType: %u", __func__, topoType_), HCCL_E_NOT_SUPPORT);
466 0 : CHK_RET(DoubleRingAllGather(param.tag, execMem.inputMem, execMem.outputMem, execMem.count,
467 : param.DataDes.dataType, level0MultiRingDataSlices_, param.stream, PROF_STAGE_2,
468 : 0, &allgatherOpInfo, level0MultiRingDataSlices_));
469 : }
470 0 : return HCCL_SUCCESS;
471 : }
472 :
473 : REGISTER_EXEC("AllReduceRingZerocopyExecutor", AllReduceRingZerocopy, CollAllReduceRingZerocopyExecutor);
474 :
475 : } // namespace hccl
|
