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