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