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