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, param.root};
122 :
123 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
124 0 : scatterOpInfoPtr = &scatterOpInfo;
125 0 : CHK_RET(ActiveSlaveStreams(param.stream));
126 0 : CHK_RET(DoubleRingScatter(
127 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, mulRingSlice,
128 : param.root, param.stream, scatterOpInfoPtr));
129 : } else {
130 0 : if (DMAReduceFlag_) {
131 0 : scatterOpInfoPtr = &scatterOpInfo;
132 : }
133 0 : CHK_RET(MultiRingScatter(
134 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, mulRingSlice,
135 : param.root, param.stream, scatterOpInfoPtr));
136 : }
137 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
138 : std::unique_ptr<AlgTemplateBase> level0Executor
139 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_SCATTER_MESH, dispatcher_);
140 0 : CHK_SMART_PTR_NULL(level0Executor);
141 0 : CHK_RET(level0Executor->Prepare(level0CommInfo.localRank, level0CommInfo.localRankSize));
142 0 : level0Executor->CloseBarrier();
143 :
144 : /* 节点内执行器 stage0 */
145 0 : u32 rootRank = 0;
146 0 : HcclResult ret = GetRankByUserRank(COMM_LEVEL0, COMM_INDEX_0, param.root, rootRank);
147 :
148 0 : CHK_PRT_RET(
149 : ret != HCCL_SUCCESS,
150 : HCCL_ERROR(
151 : "[CollBroadCastMix][KernelRun]errNo[0x%016llx]invalid root[%u] to get userrank", HCCL_ERROR_CODE(ret),
152 : param.root),
153 : ret);
154 :
155 0 : CHK_RET(level0Executor->Prepare(
156 : execMem.inputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, param.stream,
157 : HCCL_REDUCE_RESERVED, rootRank, dataSegsSlice));
158 :
159 0 : u32 rankSize = level0CommInfo.localRankSize;
160 0 : CHK_RET(level0Executor->RegisterProfiler(
161 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_0,
162 : HCCL_EXEC_STEP_NOT_SET, param.stream));
163 :
164 0 : CHK_RET(RunTemplate(level0Executor, level0CommInfo));
165 0 : }
166 0 : HCCL_INFO("[CollBroadCastMix][KernelRun] level0-scatter run success");
167 :
168 : /* step 2: server间 broadcast */
169 0 : u32 commIndex = level0CommInfo.localRank;
170 0 : u64 level1DataSize = dataSegsSlice[commIndex].size;
171 0 : u64 level1DataCount = level1DataSize / perDataSize;
172 0 : HCCL_DEBUG(
173 : "[CollBroadCastMix][KernelRun]usrRank[%u] level1 use level1DataCount[%llu]", topoAttr_.userRank,
174 : level1DataCount);
175 :
176 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
177 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
178 0 : HCCL_DEBUG(
179 : "[CollBroadCastMix][KernelRun]commIdx:%u TagCommInfo[%s].commLevel1.size():%u", commIndex, param.tag.c_str(),
180 : level1CommInfo.localRankSize);
181 :
182 0 : std::unique_ptr<AlgTemplateBase> level1Executor;
183 0 : u64 curSize = execMem.count * perDataSize;
184 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
185 0 : HCCL_DEBUG(
186 : "[CollBroadCastMix][KernelRun] curSize[%llu] deviceNumPerAggregation[%u] commLevel0Size[%u]", curSize,
187 : topoAttr_.deviceNumPerAggregation, level0CommInfo.localRankSize);
188 0 : if (curSize / topoAttr_.deviceNumPerAggregation <= NHR_BCAST_SMALL_SIZE) {
189 0 : level1Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
190 0 : TemplateType::TEMPLATE_BROADCAST_NHR_ONESHOT, dispatcher_);
191 : } else {
192 : level1Executor
193 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_BROADCAST_NHR, dispatcher_);
194 : }
195 0 : HCCL_INFO("[CollBroadCastMix][KernelRun]broadcast mix: using nhr algo inter-server.");
196 : } else {
197 0 : HCCL_ERROR("[CollBroadCastMix][KernelRun]broadcast mix: algType[%u] is not supported.", algType_.algoLevel1);
198 0 : return HCCL_E_NOT_SUPPORT;
199 : }
200 0 : CHK_SMART_PTR_NULL(level1Executor);
201 :
202 0 : u32 subUserrankRoot = topoMatcher_->GetSubRootUserRank(topoAttr_.userRank, param.root);
203 0 : CHK_PRT_RET(
204 : subUserrankRoot == INVALID_VALUE_RANKID,
205 : HCCL_ERROR(
206 : "[CollBroadCastMix][KernelRun]subUserrankRoot[%u] is invalid,userRank[%u],root[%u]", subUserrankRoot,
207 : topoAttr_.userRank, param.root),
208 : HCCL_E_INTERNAL);
209 :
210 0 : u32 planeRoot = 0;
211 0 : CHK_RET(GetRankByUserRank(COMM_LEVEL1, commIndex, subUserrankRoot, planeRoot));
212 :
213 0 : CHK_RET(level1Executor->Prepare(
214 : execMem.inputMem, execMem.inputMem, execMem.outputMem, level1DataCount, param.DataDes.dataType, param.stream,
215 : HCCL_REDUCE_RESERVED, planeRoot, std::vector<Slice>(0), dataSegsSlice[commIndex].offset));
216 :
217 0 : u32 rankSize = level1CommInfo.localRankSize;
218 0 : CHK_RET(level1Executor->RegisterProfiler(
219 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo.localRank, PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET,
220 : param.stream));
221 :
222 0 : CHK_RET(RunTemplate(level1Executor, level1CommInfo));
223 :
224 0 : HCCL_INFO("[CollBroadCastMix][KernelRun]Broadcast mix stage1 run success");
225 :
226 : /* step 3: 节点内 allgather */
227 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
228 0 : HcomCollOpInfo allgatherOpInfo
229 0 : = {"", nullptr, execMem.outputPtr, param.DataDes.count, param.DataDes.dataType, param.root};
230 0 : HcomCollOpInfo* allgatherOpInfoPtr = DMAReduceFlag_ ? (&allgatherOpInfo) : (nullptr);
231 :
232 0 : CHK_RET(MultiRingAllGather(
233 : param.tag, execMem.inputMem, execMem.outputMem, level1DataCount, param.DataDes.dataType, mulRingSlice,
234 : param.stream, PROF_STAGE_2, 0, allgatherOpInfoPtr));
235 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
236 0 : std::unique_ptr<AlgTemplateBase> level0Executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
237 0 : TemplateType::TEMPLATE_ALL_GATHER_MESH_ATOMIC, dispatcher_);
238 0 : CHK_SMART_PTR_NULL(level0Executor);
239 0 : CHK_RET(level0Executor->Prepare(
240 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux, topoAttr_.userRank, nullptr,
241 : level0CommInfo.localRank, level0CommInfo.localRankSize));
242 :
243 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) { // offline
244 0 : for (u32 streamIndex = 0; streamIndex < algResResp_->slaveStreams.size(); streamIndex++) {
245 0 : CHK_RET(StreamActiveManager::GetInstance(topoAttr_.deviceLogicId)
246 : .StreamActive(algResResp_->slaveStreams[streamIndex].ptr(), param.stream.ptr()));
247 : }
248 : }
249 0 : CHK_RET(level0Executor->Prepare(
250 : execMem.outputMem, execMem.outputMem, execMem.outputMem, execMem.count, param.DataDes.dataType,
251 : param.stream, HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, dataSegsSlice));
252 :
253 0 : u32 level0RankSize = level0CommInfo.localRankSize;
254 0 : CHK_RET(level0Executor->RegisterProfiler(
255 : (0 << PROF_RINGINDEX_OFFSET_OF_PLANEID) + (level0RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID)
256 : + level0CommInfo.localRank,
257 : PROF_STAGE_2, HCCL_EXEC_STEP_NOT_SET, param.stream));
258 :
259 0 : CHK_RET(RunTemplate(level0Executor, level0CommInfo));
260 0 : }
261 0 : HCCL_INFO("[CollBroadCastMix][KernelRun]Broadcast mix stage2 run success");
262 :
263 0 : return HCCL_SUCCESS;
264 0 : }
265 :
266 0 : HcclResult CollBroadCastMix::DoubleRingScatter(
267 : const std::string& tag, DeviceMem inputMem, DeviceMem outputMem, const u64 count, const HcclDataType dataType,
268 : const std::vector<std::vector<Slice>> multRingsSliceZero, u32 root, Stream stream, const HcomCollOpInfo* opInfo,
269 : const u64 baseOffset)
270 : {
271 0 : HCCL_INFO("[CollBroadCastMix][DoubleRingScatter] DoubleRingScatter starts.");
272 0 : HcclResult ret = HCCL_SUCCESS;
273 0 : u32 ringNum = multRingsSliceZero.size();
274 :
275 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
276 :
277 0 : std::vector<std::vector<u32>> ringNics;
278 0 : CHK_RET(GetRingNics(tag, ringNics));
279 :
280 : // 拿到ring环映射关系
281 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
282 0 : SubCommInfo level0CommInfo1 = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_1);
283 :
284 0 : auto nicList = topoAttr_.nicList;
285 : std::vector<std::vector<u32>> multiRingsOrder
286 0 : = GetRingsOrderByTopoType(level0CommInfo.localRankSize, topoType_, nicList);
287 :
288 0 : std::vector<std::vector<u32>> doubleRingsOrders;
289 0 : std::vector<std::vector<Slice>> doubleRingUserMemInputSlices;
290 0 : for (u32 ringIndex = 0; ringIndex < ringNum; ringIndex++) {
291 0 : std::vector<Slice> singleRingSliceZero = multRingsSliceZero[ringIndex];
292 0 : CHK_PRT_RET(
293 : singleRingSliceZero.empty(),
294 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]singleRingSliceZero is empty"), HCCL_E_INTERNAL);
295 :
296 : // 生成userMemIn_上对应的slices
297 0 : std::vector<Slice> userMemInputSlices;
298 0 : CHK_RET(
299 : CalUserMemSlices(dataType, opInfo, singleRingSliceZero, ringIndex, multiRingsOrder, userMemInputSlices));
300 0 : doubleRingUserMemInputSlices.push_back(userMemInputSlices);
301 0 : std::vector<u32> rankOrder;
302 0 : CHK_RET(GetRankOrder(multiRingsOrder, ringIndex, rankOrder));
303 0 : doubleRingsOrders.push_back(rankOrder);
304 0 : }
305 0 : std::unique_ptr<AlgTemplateBase> executor = AlgTemplateRegistry::Instance().GetAlgTemplate(
306 0 : TemplateType::TEMPLATE_SCATTER_DOUBLE_RING_DIRECT, dispatcher_);
307 0 : CHK_SMART_PTR_NULL(executor);
308 :
309 0 : CHK_RET(executor->Prepare(
310 : const_cast<HcomCollOpInfo*>(opInfo), topoAttr_.userRank, level0CommInfo1.localRank, algResResp_->slaveStreams,
311 : algResResp_->notifiesMain, algResResp_->notifiesAux, doubleRingsOrders, multRingsSliceZero,
312 : doubleRingUserMemInputSlices));
313 :
314 0 : u32 rootRank = 0;
315 0 : ret = GetRankByUserRank(COMM_LEVEL0, COMM_INDEX_0, root, rootRank);
316 0 : CHK_PRT_RET(
317 : ret != HCCL_SUCCESS, HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]invalid root [%u] to get userrank", root),
318 : ret);
319 0 : ret = executor->Prepare(
320 0 : inputMem, inputMem, outputMem, count, dataType, stream, HCCL_REDUCE_RESERVED, rootRank, std::vector<Slice>(0),
321 : baseOffset);
322 0 : CHK_PRT_RET(
323 : ret != HCCL_SUCCESS,
324 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]scatter(ring) prepare failed, return[%d]", ret), ret);
325 :
326 0 : u32 rankSize = level0CommInfo.localRankSize;
327 0 : ret = executor->RegisterProfiler(
328 0 : (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET,
329 : stream);
330 0 : CHK_PRT_RET(
331 : ret != HCCL_SUCCESS,
332 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]scatter(ring) register profiler failed,return[%d]", ret), ret);
333 :
334 0 : ret = RunTemplate(executor, level0CommInfo);
335 0 : CHK_PRT_RET(
336 : ret != HCCL_SUCCESS,
337 : HCCL_ERROR("[CollBroadCastMix][DoubleRingScatter]scatter(ring) run failed, return[%d]", ret), ret);
338 :
339 0 : HCCL_INFO("[CollBroadCastMix] double ring scatter run success");
340 0 : return HCCL_SUCCESS;
341 0 : }
342 :
343 : REGISTER_EXEC("BroadCastMixExecutor", BroadCastMix, CollBroadCastMix);
344 :
345 : } // namespace hccl
|
