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_mix_executor.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 :
16 0 : CollBroadCastMix::CollBroadCastMix(const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher> &topoMatcher)
17 0 : : CollBroadcastExecutor(dispatcher, topoMatcher)
18 : {
19 0 : DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
20 0 : topoAttr_.deviceType == DevType::DEV_TYPE_910_93;
21 0 : }
22 :
23 0 : HcclResult CollBroadCastMix::CalcStreamNum(u32& streamNum)
24 : {
25 0 : u32 totalStreamNum = 0U;
26 :
27 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
28 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
29 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
30 0 : totalStreamNum = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ?
31 : LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
32 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
33 0 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
34 : }
35 : }
36 :
37 0 : streamNum = totalStreamNum - 1;
38 0 : HCCL_INFO("[CollBroadCastMix][CalcStreamNum] tag[%s] streamNum_[%u]", tag_.c_str(), streamNum);
39 0 : return HCCL_SUCCESS;
40 : }
41 :
42 0 : HcclResult CollBroadCastMix::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
43 : {
44 0 : TransportMemType inputType = TransportMemType::RESERVED;
45 0 : TransportMemType outputType = TransportMemType::RESERVED;
46 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
47 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
48 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
49 :
50 : // mix在server间使用NHR通信域，并在多机A+X场景下当未设置使用RDMA时，默认使用RDMA
51 0 : std::vector<SingleSubCommTransport> &commTransportLevel1 = opTransport[COMM_LEVEL1];
52 0 : for (u32 ringIndex = 0; ringIndex < commTransportLevel1.size(); ringIndex++) {
53 0 : for (auto &transportRequest : commTransportLevel1[ringIndex].transportRequests) {
54 0 : transportRequest.isUsedRdma = true;
55 : }
56 : }
57 0 : return HCCL_SUCCESS;
58 : }
59 :
60 0 : HcclResult CollBroadCastMix::CalcLevel0CommInfo(TransportMemType inputType,
61 : TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
62 : {
63 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
64 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
65 0 : commParaLevel0.meshSinglePlane = true;
66 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
67 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
68 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
69 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
70 0 : }
71 0 : return HCCL_SUCCESS;
72 : }
73 :
74 0 : HcclResult CollBroadCastMix::KernelRun(const OpParam ¶m, ExecMem &execMem)
75 : {
76 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollBroadCastMix][KernelRun] The CollBroadCastMix starts.");
77 0 : u32 perDataSize = 0;
78 0 : CHK_RET(SalGetDataTypeSize(param.DataDes.dataType, perDataSize));
79 0 : CHK_PRT_RET(perDataSize == 0,
80 : HCCL_ERROR("[CollBroadCastMix][KernelRun]errNo[0x%01611x] datatype[%d] is invalid",
81 : HCCL_ERROR_CODE(HCCL_E_PARA), param.DataDes.dataType), HCCL_E_PARA);
82 :
83 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
84 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
85 :
86 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
87 : // 按ranksize得到内存切分slice数
88 0 : u32 sliceNum = level0CommInfo.localRankSize;
89 0 : HCCL_DEBUG("[CollBroadCastMix][KernelRun]sliceNum[%u]", sliceNum);
90 : // 将根节点数据切分成sliceNum份
91 0 : CHK_RET(AlgTemplateBase::PrepareSliceData(execMem.count, perDataSize, sliceNum, 0, dataSegsSlice));
92 :
93 0 : std::vector<std::vector<Slice>> mulRingSlice; // 910_93数据基于该rank上环0的偏移
94 :
95 : /* step 1: 节点内 scatter */
96 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
97 0 : u32 ringNum = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) ?
98 : LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
99 :
100 0 : HCCL_DEBUG("[CollBroadCastMix][KernelRun]ringNum[%u]", ringNum);
101 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
102 :
103 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
104 : // 将每slice再切分成2份，按各ring的dev顺序排列
105 0 : mulRingSlice = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, topoAttr_.nicList);
106 0 : CHK_PRT_RET(mulRingSlice.size() != ringNum,
107 : HCCL_ERROR("[CollBroadCastMix][KernelRun] ringNum[%u] !=mulRingSlice size[%zu]",
108 : ringNum, mulRingSlice.size()), HCCL_E_INTERNAL);
109 : } else {
110 0 : mulRingSlice.push_back(dataSegsSlice); // 应该offset全为0，而大小和dataSegsSlice中一样,里面的offset不使用
111 : }
112 :
113 0 : HcomCollOpInfo *scatterOpInfoPtr = nullptr;
114 0 : HcomCollOpInfo scatterOpInfo =
115 0 : {"", execMem.inputPtr, nullptr, param.DataDes.count, param.DataDes.dataType, param.root};
116 :
117 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
118 0 : scatterOpInfoPtr = &scatterOpInfo;
119 0 : CHK_RET(ActiveSlaveStreams(param.stream));
120 0 : CHK_RET(DoubleRingScatter(param.tag, execMem.inputMem, execMem.outputMem, execMem.count,
121 : param.DataDes.dataType, mulRingSlice, param.root, param.stream, scatterOpInfoPtr));
122 : } else {
123 0 : if (DMAReduceFlag_) {
124 0 : scatterOpInfoPtr = &scatterOpInfo;
125 : }
126 0 : CHK_RET(MultiRingScatter(param.tag, execMem.inputMem, execMem.outputMem, execMem.count,
127 : param.DataDes.dataType, mulRingSlice, param.root, param.stream, scatterOpInfoPtr));
128 : }
129 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
130 0 : std::unique_ptr<AlgTemplateBase> level0Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
131 0 : TemplateType::TEMPLATE_SCATTER_MESH, dispatcher_);
132 0 : CHK_SMART_PTR_NULL(level0Executor);
133 0 : CHK_RET(level0Executor->Prepare(level0CommInfo.localRank, level0CommInfo.localRankSize));
134 0 : level0Executor->CloseBarrier();
135 :
136 : /* 节点内执行器 stage0 */
137 0 : u32 rootRank = 0;
138 0 : HcclResult ret = GetRankByUserRank(COMM_LEVEL0, COMM_INDEX_0, param.root, rootRank);
139 :
140 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
141 : HCCL_ERROR("[CollBroadCastMix][KernelRun]errNo[0x%016llx]invalid root[%u] to get userrank",
142 : HCCL_ERROR_CODE(ret), param.root), ret);
143 :
144 0 : CHK_RET(level0Executor->Prepare(execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count,
145 : param.DataDes.dataType, param.stream, HCCL_REDUCE_RESERVED, rootRank, dataSegsSlice));
146 :
147 0 : u32 rankSize = level0CommInfo.localRankSize;
148 0 : CHK_RET(level0Executor->RegisterProfiler((rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) +
149 : level0CommInfo.localRank, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, param.stream));
150 :
151 0 : CHK_RET(RunTemplate(level0Executor, level0CommInfo));
152 0 : }
153 0 : HCCL_INFO("[CollBroadCastMix][KernelRun] level0-scatter run success");
154 :
155 : /* step 2: server间 broadcast */
156 0 : u32 commIndex = level0CommInfo.localRank;
157 0 : u64 level1DataSize = dataSegsSlice[commIndex].size;
158 0 : u64 level1DataCount = level1DataSize / perDataSize;
159 0 : HCCL_DEBUG("[CollBroadCastMix][KernelRun]usrRank[%u] level1 use level1DataCount[%llu]",
160 : topoAttr_.userRank, level1DataCount);
161 :
162 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
163 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
164 0 : HCCL_DEBUG("[CollBroadCastMix][KernelRun]commIdx:%u TagCommInfo[%s].commLevel1.size():%u",
165 : commIndex, param.tag.c_str(), level1CommInfo.localRankSize);
166 :
167 0 : std::unique_ptr<AlgTemplateBase> level1Executor;
168 0 : u64 curSize = execMem.count * perDataSize;
169 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
170 0 : HCCL_DEBUG("[CollBroadCastMix][KernelRun] curSize[%llu] deviceNumPerAggregation[%u] commLevel0Size[%u]",
171 : curSize, topoAttr_.deviceNumPerAggregation, level0CommInfo.localRankSize);
172 0 : if (curSize / topoAttr_.deviceNumPerAggregation <= NHR_BCAST_SMALL_SIZE) {
173 0 : level1Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
174 0 : TemplateType::TEMPLATE_BROADCAST_NHR_ONESHOT, dispatcher_);
175 : } else {
176 0 : level1Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
177 0 : TemplateType::TEMPLATE_BROADCAST_NHR, dispatcher_);
178 : }
179 0 : HCCL_INFO("[CollBroadCastMix][KernelRun]broadcast mix: using nhr algo inter-server.");
180 : } else {
181 0 : HCCL_ERROR("[CollBroadCastMix][KernelRun]broadcast mix: algType[%u] is not supported.", algType_.algoLevel1);
182 0 : return HCCL_E_NOT_SUPPORT;
183 : }
184 0 : CHK_SMART_PTR_NULL(level1Executor);
185 :
186 0 : u32 subUserrankRoot = topoMatcher_->GetSubRootUserRank(topoAttr_.userRank, param.root);
187 0 : CHK_PRT_RET(subUserrankRoot == INVALID_VALUE_RANKID,
188 : HCCL_ERROR("[CollBroadCastMix][KernelRun]subUserrankRoot[%u] is invalid,userRank[%u],root[%u]",
189 : subUserrankRoot, topoAttr_.userRank, param.root), HCCL_E_INTERNAL);
190 :
191 0 : u32 planeRoot = 0;
192 0 : CHK_RET(GetRankByUserRank(COMM_LEVEL1, commIndex, subUserrankRoot, planeRoot));
193 :
194 0 : CHK_RET(level1Executor->Prepare(execMem.inputMem, execMem.inputMem, execMem.outputMem, level1DataCount,
195 : param.DataDes.dataType, param.stream, HCCL_REDUCE_RESERVED, planeRoot, std::vector<Slice>(0),
196 : dataSegsSlice[commIndex].offset));
197 :
198 0 : u32 rankSize = level1CommInfo.localRankSize;
199 0 : CHK_RET(level1Executor->RegisterProfiler((rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank,
200 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, param.stream));
201 :
202 0 : CHK_RET(RunTemplate(level1Executor, level1CommInfo));
203 :
204 0 : HCCL_INFO("[CollBroadCastMix][KernelRun]Broadcast mix stage1 run success");
205 :
206 : /* step 3: 节点内 allgather */
207 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
208 0 : HcomCollOpInfo allgatherOpInfo = {
209 0 : "", nullptr, execMem.outputPtr, param.DataDes.count, param.DataDes.dataType, param.root
210 0 : };
211 0 : HcomCollOpInfo *allgatherOpInfoPtr = DMAReduceFlag_ ? (&allgatherOpInfo) : (nullptr);
212 :
213 0 : CHK_RET(MultiRingAllGather(param.tag, execMem.inputMem, execMem.outputMem, level1DataCount, param.DataDes.dataType,
214 : mulRingSlice, param.stream, PROF_STAGE_2, 0, allgatherOpInfoPtr));
215 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
216 0 : std::unique_ptr<AlgTemplateBase> level0Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
217 0 : TemplateType::TEMPLATE_ALL_GATHER_MESH_ATOMIC, dispatcher_);
218 0 : CHK_SMART_PTR_NULL(level0Executor);
219 0 : CHK_RET(level0Executor->Prepare(algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
220 : topoAttr_.userRank, nullptr, level0CommInfo.localRank, level0CommInfo.localRankSize));
221 :
222 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) { // offline
223 0 : for (u32 streamIndex = 0; streamIndex < algResResp_->slaveStreams.size(); streamIndex++) {
224 0 : CHK_RET(StreamActiveManager::GetInstance(topoAttr_.deviceLogicId).StreamActive(
225 : algResResp_->slaveStreams[streamIndex].ptr(), param.stream.ptr()));
226 : }
227 : }
228 0 : CHK_RET(level0Executor->Prepare(execMem.outputMem, execMem.outputMem, execMem.outputMem, execMem.count,
229 : param.DataDes.dataType, param.stream, HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice));
230 :
231 0 : u32 level0RankSize = level0CommInfo.localRankSize;
232 0 : CHK_RET(level0Executor->RegisterProfiler((0 << PROF_RINGINDEX_OFFSET_OF_PLANEID) +
233 : (level0RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
234 : PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
235 :
236 0 : CHK_RET(RunTemplate(level0Executor, level0CommInfo));
237 0 : }
238 0 : HCCL_INFO("[CollBroadCastMix][KernelRun]Broadcast mix stage2 run success");
239 :
240 0 : return HCCL_SUCCESS;
241 0 : }
242 :
243 0 : HcclResult CollBroadCastMix::DoubleRingScatter(const std::string &tag, DeviceMem inputMem, DeviceMem outputMem,
244 : const u64 count, const HcclDataType dataType, const std::vector<std::vector<Slice> > multRingsSliceZero,
245 : u32 root, Stream stream, const HcomCollOpInfo *opInfo, const u64 baseOffset)
246 : {
247 0 : HCCL_INFO("[CollBroadCastMix][DoubleRingScatter] DoubleRingScatter starts.");
248 0 : HcclResult ret = HCCL_SUCCESS;
249 0 : u32 ringNum = multRingsSliceZero.size();
250 :
251 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
252 :
253 0 : std::vector<std::vector<u32>> ringNics;
254 0 : CHK_RET(GetRingNics(tag, ringNics));
255 :
256 : // 拿到ring环映射关系
257 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
258 0 : SubCommInfo level0CommInfo1 = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_1);
259 :
260 0 : auto nicList = topoAttr_.nicList;
261 : std::vector<std::vector<u32>> multiRingsOrder =
262 0 : GetRingsOrderByTopoType(level0CommInfo.localRankSize, topoType_, nicList);
263 :
264 0 : std::vector<std::vector<u32>> doubleRingsOrders;
265 0 : std::vector<std::vector<Slice>> doubleRingUserMemInputSlices;
266 0 : for (u32 ringIndex = 0; ringIndex < ringNum; ringIndex++) {
267 0 : std::vector<Slice> singleRingSliceZero = multRingsSliceZero[ringIndex];
268 0 : CHK_PRT_RET(singleRingSliceZero.empty(),
269 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]singleRingSliceZero is empty"), HCCL_E_INTERNAL);
270 :
271 : // 生成userMemIn_上对应的slices
272 0 : std::vector<Slice> userMemInputSlices;
273 0 : CHK_RET(
274 : CalUserMemSlices(dataType, opInfo, singleRingSliceZero, ringIndex, multiRingsOrder, userMemInputSlices));
275 0 : doubleRingUserMemInputSlices.push_back(userMemInputSlices);
276 0 : std::vector<u32> rankOrder;
277 0 : CHK_RET(GetRankOrder(multiRingsOrder, ringIndex, rankOrder));
278 0 : doubleRingsOrders.push_back(rankOrder);
279 0 : }
280 0 : std::unique_ptr<AlgTemplateBase> executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
281 0 : TemplateType::TEMPLATE_SCATTER_DOUBLE_RING_DIRECT, dispatcher_);
282 0 : CHK_SMART_PTR_NULL(executor);
283 :
284 0 : CHK_RET(executor ->Prepare(const_cast<HcomCollOpInfo *>(opInfo), topoAttr_.userRank, level0CommInfo1.localRank,
285 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux,
286 : doubleRingsOrders, multRingsSliceZero, doubleRingUserMemInputSlices));
287 :
288 0 : u32 rootRank = 0;
289 0 : ret = GetRankByUserRank(COMM_LEVEL0, COMM_INDEX_0, root, rootRank);
290 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
291 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]invalid root [%u] to get userrank", root), ret);
292 0 : ret = executor->Prepare(inputMem, inputMem, outputMem, count, dataType, stream, HCCL_REDUCE_RESERVED,
293 0 : rootRank, std::vector<Slice>(0), baseOffset);
294 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
295 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]scatter(ring) prepare failed, return[%d]", ret), ret);
296 :
297 0 : u32 rankSize = level0CommInfo.localRankSize;
298 0 : ret = executor->RegisterProfiler((rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
299 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, stream);
300 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
301 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]scatter(ring) register profiler failed,return[%d]", ret), ret);
302 :
303 0 : ret = RunTemplate(executor, level0CommInfo);
304 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
305 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]scatter(ring) run failed, return[%d]", ret), ret);
306 :
307 0 : HCCL_INFO("[CollBroadCastMix] double ring scatter run success");
308 0 : return HCCL_SUCCESS;
309 0 : }
310 :
311 :
312 : REGISTER_EXEC("BroadCastMixExecutor", BroadCastMix, CollBroadCastMix);
313 :
314 : } // namespace hccl
|
