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 "ins_all_gather_parallel_executor.h"
12 :
13 : #include <cmath>
14 :
15 : #include "log.h"
16 :
17 : #include "ins_coll_alg_registry.h"
18 :
19 : #include "topo_match_mesh_nhr.h"
20 : #include "topo_match_concurr_mesh_nhr.h"
21 : #include "topo_match_mesh_nhr_pcie.h"
22 :
23 : #include "alg_data_trans_wrapper.h"
24 :
25 : #include "ins_temp_all_gather_mesh.h"
26 : #include "ins_temp_all_gather_mesh_2D.h"
27 : #include "ins_temp_all_gather_nhr.h"
28 :
29 : #ifndef CCL_KERNEL_AICPU
30 : #include "ccu_temp_all_gather_nhr_1D_mem2mem.h"
31 : #include "ccu_temp_all_gather_mesh_1D_mem2mem_with_stride.h"
32 : #endif
33 :
34 : namespace Hccl {
35 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
36 0 : InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::InsAllGatherParallelExecutor()
37 0 : : InsCollAlgBase()
38 0 : {}
39 :
40 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
41 0 : InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::~InsAllGatherParallelExecutor()
42 0 : {}
43 :
44 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
45 0 : HcclResult InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcResOffload(
46 : const RankGraph *rankGraph, const u64 &dataSize, CollOffloadOpResReq &resReq)
47 : {
48 : (void)dataSize;
49 :
50 0 : u64 scratchMemSize = 200 * 1024 * 1024;
51 0 : resReq.requiredScratchMemSize = scratchMemSize; // 200MB
52 :
53 : // Topo Match
54 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
55 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
56 :
57 0 : rankSizeLevel0_ = virtRanks_[0].size();
58 0 : rankSizeLevel1_ = virtRanks_[1].size();
59 0 : InsAlgTemplate0 intraTempAlg(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]);
60 0 : InsAlgTemplate1 interTempAlg(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]);
61 0 : std::vector<map<u32, u32>> rank2PathNumMap;
62 0 : HCCL_INFO("[InsAllGatherParallelExecutor] CalcResOffload SetPathNumMap");
63 0 : CHK_RET(SetPathNumMapByRankGraphMultiLevel(rankGraph, virtRanks_, myRank_, rank2PathNumMap));
64 0 : intraTempAlg.setPathNumMap(rank2PathNumMap[0]);
65 0 : interTempAlg.setPathNumMap(rank2PathNumMap[1]);
66 :
67 : // calculate required insQues and prepare queue
68 0 : AlgTempResReq resReqIntra;
69 0 : AlgTempResReq resReqInter;
70 0 : if (enableDetour_) {
71 0 : HCCL_DEBUG("[%s] Rank[%d], CalcResOffload with detouring enabled.", __func__, myRank_);
72 0 : CHK_RET(intraTempAlg.CalcResDetour(rankGraph, resReqIntra));
73 : } else {
74 0 : HCCL_DEBUG("[%s] Rank[%d], CalcResOffload with detouring disabled. rankSizeLevel0[%u] rankSizeLevel1[%u]",
75 : __func__,
76 : myRank_,
77 : rankSizeLevel0_,
78 : rankSizeLevel1_);
79 0 : CHK_RET(intraTempAlg.CalcRes(resReqIntra));
80 : }
81 :
82 0 : CHK_RET(interTempAlg.CalcRes(resReqInter));
83 :
84 0 : resReq.requiredSubQueNum = resReqIntra.streamNum + resReqInter.streamNum - 1;
85 :
86 0 : return HcclResult::HCCL_SUCCESS;
87 0 : }
88 :
89 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
90 0 : HcclResult InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcRes(
91 : const RankGraph *rankGraph, CollAlgResReq &algResReq)
92 : {
93 : // Topo Match
94 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
95 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
96 :
97 0 : algResReq.topoInfo.UpdateMultiLevelTopo(virtRanks_, virtRankMap_, vTopo_);
98 :
99 0 : rankSizeLevel0_ = virtRanks_[0].size();
100 0 : rankSizeLevel1_ = virtRanks_[1].size();
101 :
102 : // instantiate a template
103 0 : InsAlgTemplate0 intraTempAlg(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]);
104 0 : InsAlgTemplate1 interTempAlg(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]);
105 0 : std::vector<map<u32, u32>> rank2PathNumMap;
106 0 : HCCL_INFO("[InsAllGatherParallelExecutor] CalcRes SetPathNumMap");
107 0 : CHK_RET(SetPathNumMapByRankGraphMultiLevel(rankGraph, virtRanks_, myRank_, rank2PathNumMap));
108 0 : intraTempAlg.setPathNumMap(rank2PathNumMap[0]);
109 0 : interTempAlg.setPathNumMap(rank2PathNumMap[1]);
110 :
111 : // calculate required insQues and prepare queue
112 0 : AlgTempResReq resReqIntra;
113 0 : AlgTempResReq resReqInter;
114 0 : if (enableDetour_) {
115 0 : HCCL_DEBUG("[InsAllGatherParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
116 0 : CHK_RET(intraTempAlg.CalcResDetour(rankGraph, resReqIntra));
117 : } else {
118 0 : HCCL_DEBUG("[InsAllGatherParallelExecutor] Rank[%d], CalcRes with detouring disabled."
119 : "rankSizeLevel0[%u] rankSizeLevel1[%u]",
120 : myRank_,
121 : rankSizeLevel0_,
122 : rankSizeLevel1_);
123 0 : CHK_RET(intraTempAlg.CalcRes(resReqIntra));
124 : }
125 0 : CHK_RET(interTempAlg.CalcRes(resReqInter));
126 :
127 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqIntra.links, algResReq.levelRankPairs));
128 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqInter.links, algResReq.levelRankPairs));
129 0 : algResReq.primQueueNum = resReqIntra.streamNum + resReqInter.streamNum;
130 0 : CHK_RET(CalcParallelNotifyReq(algResReq.primQueueNum, resReqIntra.queNum, algResReq.queueNotifys));
131 0 : HCCL_DEBUG("[InsAllGatherParallelExecutor] algResReq.primQueueNum %u", algResReq.primQueueNum);
132 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, algResReq.links));
133 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, algResReq.links));
134 :
135 0 : return HcclResult::HCCL_SUCCESS;
136 0 : }
137 :
138 : // HOST 侧算法入口，将对应的 instruction 添加到指令队列中
139 : // 传入的insQue为一条主流
140 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
141 0 : void InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenTemplateAlgParamsIntra0(
142 : const u64 dataOffset, const u64 dataCountPerLoopAixs0, const u64 scratchOffset,
143 : TemplateDataParams &tempAlgParamsIntra0) const
144 : {
145 0 : tempAlgParamsIntra0.buffInfo.inBuffType = BufferType::INPUT;
146 0 : tempAlgParamsIntra0.buffInfo.outBuffType = BufferType::OUTPUT;
147 0 : tempAlgParamsIntra0.buffInfo.scratBuffType = BufferType::SCRATCH;
148 0 : tempAlgParamsIntra0.buffInfo.inBuffBaseOff = dataOffset;
149 0 : tempAlgParamsIntra0.buffInfo.outBuffBaseOff = rankIdxLevel1_ * rankSizeLevel0_ * dataSize_ + dataOffset;
150 0 : tempAlgParamsIntra0.buffInfo.scratchBuffBaseOff = scratchOffset;
151 0 : tempAlgParamsIntra0.sliceSize = dataCountPerLoopAixs0 * dataTypeSize_;
152 0 : tempAlgParamsIntra0.tailSize = dataCountPerLoopAixs0 * dataTypeSize_;
153 0 : tempAlgParamsIntra0.inputSliceStride = 0;
154 0 : tempAlgParamsIntra0.outputSliceStride = dataSize_;
155 0 : tempAlgParamsIntra0.repeatNum = 1;
156 0 : tempAlgParamsIntra0.inputRepeatStride = 0;
157 0 : tempAlgParamsIntra0.outputRepeatStride = 0;
158 :
159 0 : HCCL_DEBUG(
160 : "[InsAllGatherParallelExecutor][GenTemplateAlgParamsIntra0] rank[%d] inBuffBaseOff[%llu] "
161 : "outBuffBaseOff[%llu] scratchBuffBaseOff[%llu] sliceSize[%llu] outputSliceStride[%llu] rankSizeLevel0[%u] "
162 : "rankSizeLevel1[%u] rankIdxLevel0[%u] rankIdxLevel1[%u]",
163 : myRank_,
164 : tempAlgParamsIntra0.buffInfo.inBuffBaseOff,
165 : tempAlgParamsIntra0.buffInfo.outBuffBaseOff,
166 : tempAlgParamsIntra0.buffInfo.scratchBuffBaseOff,
167 : tempAlgParamsIntra0.sliceSize,
168 : tempAlgParamsIntra0.outputSliceStride,
169 : rankSizeLevel0_,
170 : rankSizeLevel1_,
171 : rankIdxLevel0_,
172 : rankIdxLevel1_);
173 0 : return;
174 : }
175 :
176 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
177 0 : void InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenTemplateAlgParamsInter0(
178 : const u64 dataOffset, const u64 dataCountPerLoopAixs0, const u64 scratchOffset,
179 : TemplateDataParams &tempAlgParamsInter0) const
180 : {
181 0 : tempAlgParamsInter0.buffInfo.inBuffType = BufferType::OUTPUT;
182 0 : tempAlgParamsInter0.buffInfo.outBuffType = BufferType::OUTPUT;
183 0 : tempAlgParamsInter0.buffInfo.scratBuffType = BufferType::SCRATCH;
184 0 : tempAlgParamsInter0.buffInfo.inBuffBaseOff = dataOffset;
185 0 : tempAlgParamsInter0.buffInfo.outBuffBaseOff = dataOffset;
186 0 : tempAlgParamsInter0.buffInfo.scratchBuffBaseOff = scratchOffset;
187 0 : tempAlgParamsInter0.sliceSize = dataCountPerLoopAixs0 * dataTypeSize_;
188 0 : tempAlgParamsInter0.tailSize = dataCountPerLoopAixs0 * dataTypeSize_;
189 0 : tempAlgParamsInter0.inputSliceStride = dataSize_ * rankSizeLevel0_;
190 0 : tempAlgParamsInter0.outputSliceStride = dataSize_ * rankSizeLevel0_;
191 0 : tempAlgParamsInter0.repeatNum = rankSizeLevel0_;
192 0 : tempAlgParamsInter0.inputRepeatStride = dataSize_;
193 0 : tempAlgParamsInter0.outputRepeatStride = dataSize_;
194 0 : HCCL_DEBUG("[InsAllGatherParallelExecutor][GenTemplateAlgParamsInter0] rank[%d] inBuffBaseOff[%llu] "
195 : "outBuffBaseOff[%llu] scratchBuffBaseOff[%llu] sliceSize[%llu] outputSliceStride[%llu] "
196 : "outputRepeatStride[%llu]",
197 : myRank_,
198 : tempAlgParamsInter0.buffInfo.inBuffBaseOff,
199 : tempAlgParamsInter0.buffInfo.outBuffBaseOff,
200 : tempAlgParamsInter0.buffInfo.scratchBuffBaseOff,
201 : tempAlgParamsInter0.sliceSize,
202 : tempAlgParamsInter0.outputSliceStride,
203 : tempAlgParamsInter0.outputRepeatStride);
204 0 : return;
205 : }
206 :
207 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
208 0 : void InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenTemplateAlgParamsInter1(
209 : const u64 dataOffset, const u64 dataCountPerLoopAixs1, const u64 scratchOffset,
210 : TemplateDataParams &tempAlgParamsInter1) const
211 : {
212 0 : tempAlgParamsInter1.buffInfo.inBuffType = BufferType::INPUT;
213 0 : tempAlgParamsInter1.buffInfo.outBuffType = BufferType::OUTPUT;
214 0 : tempAlgParamsInter1.buffInfo.scratBuffType = BufferType::SCRATCH;
215 0 : tempAlgParamsInter1.buffInfo.inBuffBaseOff = dataOffset;
216 0 : tempAlgParamsInter1.buffInfo.outBuffBaseOff = rankIdxLevel0_ * dataSize_ + dataOffset; // for example 0 2 4 | 1 3 5
217 0 : tempAlgParamsInter1.buffInfo.scratchBuffBaseOff = scratchOffset;
218 0 : tempAlgParamsInter1.sliceSize = dataCountPerLoopAixs1 * dataTypeSize_;
219 0 : tempAlgParamsInter1.tailSize = dataCountPerLoopAixs1 * dataTypeSize_;
220 0 : tempAlgParamsInter1.inputSliceStride = 0;
221 0 : tempAlgParamsInter1.outputSliceStride = dataSize_ * rankSizeLevel0_;
222 0 : tempAlgParamsInter1.repeatNum = 1;
223 0 : tempAlgParamsInter1.inputRepeatStride = 0;
224 0 : tempAlgParamsInter1.outputRepeatStride = 0;
225 0 : HCCL_DEBUG("[InsAllGatherParallelExecutor][GenTemplateAlgParamsInter1] rank[%d] inBuffBaseOff[%llu] "
226 : "outBuffBaseOff[%llu] scratchBuffBaseOff[%llu] sliceSize[%llu] outputSliceStride[%llu]",
227 : myRank_,
228 : tempAlgParamsInter1.buffInfo.inBuffBaseOff,
229 : tempAlgParamsInter1.buffInfo.outBuffBaseOff,
230 : tempAlgParamsInter1.buffInfo.scratchBuffBaseOff,
231 : tempAlgParamsInter1.sliceSize,
232 : tempAlgParamsInter1.outputSliceStride);
233 0 : return;
234 : }
235 :
236 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
237 0 : void InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenTemplateAlgParamsIntra1(
238 : const u64 dataOffset, const u64 dataCountPerLoopAixs1, const u64 scratchOffset,
239 : TemplateDataParams &tempAlgParamsIntra1) const
240 : {
241 0 : tempAlgParamsIntra1.buffInfo.inBuffType = BufferType::OUTPUT;
242 0 : tempAlgParamsIntra1.buffInfo.outBuffType = BufferType::OUTPUT;
243 0 : tempAlgParamsIntra1.buffInfo.scratBuffType = BufferType::SCRATCH;
244 0 : tempAlgParamsIntra1.buffInfo.inBuffBaseOff = dataOffset;
245 0 : tempAlgParamsIntra1.buffInfo.outBuffBaseOff = dataOffset;
246 0 : tempAlgParamsIntra1.buffInfo.scratchBuffBaseOff = scratchOffset;
247 0 : tempAlgParamsIntra1.sliceSize = dataCountPerLoopAixs1 * dataTypeSize_;
248 0 : tempAlgParamsIntra1.tailSize = dataCountPerLoopAixs1 * dataTypeSize_;
249 0 : tempAlgParamsIntra1.inputSliceStride = dataSize_;
250 0 : tempAlgParamsIntra1.outputSliceStride = dataSize_;
251 0 : tempAlgParamsIntra1.repeatNum = rankSizeLevel1_;
252 0 : tempAlgParamsIntra1.inputRepeatStride = dataSize_ * rankSizeLevel0_;
253 0 : tempAlgParamsIntra1.outputRepeatStride = dataSize_ * rankSizeLevel0_;
254 0 : HCCL_DEBUG("[InsAllGatherParallelExecutor][GenTemplateAlgParamsIntra1] rank[%d] inBuffBaseOff[%llu] "
255 : "outBuffBaseOff[%llu] scratchBuffBaseOff[%llu] sliceSize[%llu] outputSliceStride[%llu] "
256 : "outputRepeatStride[%llu]",
257 : myRank_,
258 : tempAlgParamsIntra1.buffInfo.inBuffBaseOff,
259 : tempAlgParamsIntra1.buffInfo.outBuffBaseOff,
260 : tempAlgParamsIntra1.buffInfo.scratchBuffBaseOff,
261 : tempAlgParamsIntra1.sliceSize,
262 : tempAlgParamsIntra1.outputSliceStride,
263 : tempAlgParamsIntra1.outputRepeatStride);
264 0 : return;
265 : }
266 :
267 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
268 0 : void InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GetParallelDataSplit(
269 : std::vector<float> &splitDataSize) const
270 : {
271 : // to do 先做等分，后续根据性能做调整
272 0 : double splitData = 0.5;
273 0 : splitDataSize.push_back(splitData);
274 0 : splitDataSize.push_back(splitData);
275 0 : return;
276 : }
277 :
278 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
279 0 : HcclResult InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
280 : const RankGraph *rankGraph, InsAlgTemplate0 &tempAlgIntra, InsAlgTemplate1 &tempAlgInter)
281 : {
282 0 : AlgTempResReq resReqInter;
283 0 : AlgTempResReq resReqIntra;
284 0 : if (enableDetour_) {
285 0 : HCCL_DEBUG("[%s] Rank[%d], CalcRes with detour enabled.", __func__, myRank_);
286 0 : CHK_RET(tempAlgIntra.CalcResDetour(rankGraph, resReqIntra));
287 : } else {
288 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
289 : }
290 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
291 :
292 : // 申请算法模板所需资源
293 0 : if (!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
294 0 : HCCL_ERROR("Intra queNum and Inter queNum must larger than 0.");
295 0 : return HcclResult::HCCL_E_INTERNAL;
296 : }
297 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
298 0 : CHK_RET(InitQueue(totalQueueNum, requiredQue_));
299 0 : for (u32 i = 0; i < requiredQue_.size(); i++) {
300 0 : if (i < resReqIntra.queNum) {
301 0 : intraQue_.push_back(requiredQue_[i]);
302 : } else {
303 0 : interQue_.push_back(requiredQue_[i]);
304 : }
305 : }
306 0 : syncQueues_.emplace_back(intraQue_[0]);
307 0 : syncQueues_.emplace_back(interQue_[0]);
308 :
309 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, intraLinks_));
310 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, interLinks_));
311 0 : HCCL_INFO("[InsAllGatherParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]",
312 : intraLinks_.size(),
313 : interLinks_.size());
314 0 : return HCCL_SUCCESS;
315 0 : }
316 :
317 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
318 0 : HcclResult InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
319 : ConnectedLinkMgr *linkMgr, InsAlgTemplate0 &tempAlgIntra, InsAlgTemplate1 &tempAlgInter)
320 : {
321 0 : AlgTempResReq resReqIntra;
322 0 : AlgTempResReq resReqInter;
323 0 : if (enableDetour_) {
324 0 : HCCL_DEBUG("[%s] Rank[%d], detouring enabled.", __func__, myRank_);
325 0 : CHK_RET(tempAlgIntra.CalcResDetour(linkMgr, resReqIntra));
326 : } else {
327 0 : HCCL_DEBUG("[%s] Rank[%d], detouring disabled.", __func__, myRank_);
328 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
329 : }
330 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
331 :
332 : // 申请算法模板所需资源
333 0 : if (!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
334 0 : HCCL_ERROR("[InsAllGatherParallelExecutor]resReqIntra.queNum and resReqInter.queNum must larger than 0.");
335 0 : return HcclResult::HCCL_E_INTERNAL;
336 : }
337 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
338 0 : CHK_RET(InitQueue(totalQueueNum, requiredQue_));
339 0 : for (u32 q = 0; q < requiredQue_.size(); q++) {
340 0 : if (q < resReqIntra.queNum) {
341 0 : intraQue_.push_back(requiredQue_[q]);
342 : } else {
343 0 : interQue_.push_back(requiredQue_[q]);
344 : }
345 : }
346 0 : syncQueues_.emplace_back(intraQue_[0]);
347 0 : syncQueues_.emplace_back(interQue_[0]);
348 :
349 0 : CHK_RET(PrepResLinks(myRank_, resReqIntra.links, linkMgr, intraLinks_));
350 0 : CHK_RET(PrepResLinks(myRank_, resReqInter.links, linkMgr, interLinks_));
351 0 : HCCL_INFO("[InsAllGatherParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]",
352 : intraLinks_.size(),
353 : interLinks_.size());
354 0 : return HCCL_SUCCESS;
355 0 : }
356 :
357 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
358 0 : HcclResult InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
359 : const AlgTopoInfo &topoInfo, const CollAlgOperator &op, const CollAlgParams ¶ms, ConnectedLinkMgr *linkMgr,
360 : InsQuePtr insQue)
361 : {
362 0 : HCCL_INFO("[InsAllGatherParallelExecutor] Orchestrate begins.");
363 : // init and check params
364 0 : CHK_RET(Init(op, params, insQue));
365 0 : dataType_ = op.dataType;
366 0 : virtRanks_ = topoInfo.virtRanks;
367 0 : vTopo_ = topoInfo.vTopo;
368 0 : virtRankMap_ = topoInfo.virtRankMap;
369 :
370 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
371 0 : rankIdxLevel0_ = myRank_ % virtRanks_[0].size();
372 0 : rankIdxLevel1_ = myRank_ / virtRanks_[0].size();
373 :
374 : // 实例化算法模板类
375 0 : InsAlgTemplate0 tempAlgIntra(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
376 0 : InsAlgTemplate1 tempAlgInter(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
377 :
378 : // 实例化算法模板类
379 :
380 0 : tempAlgIntra.SetDmaMode(dmaMode_);
381 0 : tempAlgIntra.SetCollOp(op); // CCU template需要传递op信息
382 0 : tempAlgIntra.SetDataType(dataType_);
383 :
384 0 : tempAlgInter.SetDmaMode(dmaMode_);
385 0 : tempAlgInter.SetCollOp(op); // CCU template需要传递op信息
386 0 : tempAlgInter.SetDataType(dataType_);
387 :
388 0 : std::vector<std::map<u32, u32>>rank2PathNumMap;
389 0 : CHK_RET(SetPathNumMapByLinkMgrMultiLevel(linkMgr, virtRanks_, myRank_, rank2PathNumMap));
390 0 : tempAlgIntra.setPathNumMap(rank2PathNumMap[0]);
391 0 : tempAlgInter.setPathNumMap(rank2PathNumMap[1]);
392 :
393 : // 计算算法模板所需资源
394 0 : CHK_RET(PrepareResForTemplate(linkMgr, tempAlgIntra, tempAlgInter));
395 :
396 0 : CHK_RET(GenInsQuesHost(tempAlgIntra, tempAlgInter));
397 0 : HCCL_INFO("[InsAllGatherParallelExecutor] Orchestrate success.");
398 :
399 0 : return HcclResult::HCCL_SUCCESS;
400 0 : }
401 :
402 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
403 0 : HcclResult InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
404 : const RankGraph *rankGraph, const CollAlgOperator &op, const CollAlgParams ¶ms, InsQuePtr insQue)
405 : {
406 0 : HCCL_INFO("[InsAllGatherParallelExecutor] Host Orchestrate begins.");
407 : // init and check params
408 0 : CHK_RET(Init(op, params, insQue));
409 :
410 : // Topo Match
411 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
412 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
413 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
414 0 : rankIdxLevel0_ = myRank_ % virtRanks_[0].size();
415 0 : rankIdxLevel1_ = myRank_ / virtRanks_[0].size();
416 0 : HCCL_DEBUG("[InsAllGatherParallelExecutor] my rank is [%d] ranksize is [%u], rankIdxLevel0_ = [%u], rankIdxLevel1_ "
417 : "= [%u] .",
418 : myRank_,
419 : rankSize_,
420 : rankIdxLevel0_,
421 : rankIdxLevel1_);
422 : // 实例化算法模板类
423 0 : InsAlgTemplate0 tempAlgIntra(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
424 0 : InsAlgTemplate1 tempAlgInter(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
425 :
426 0 : tempAlgIntra.SetDmaMode(dmaMode_);
427 0 : tempAlgIntra.SetDataType(dataType_);
428 0 : tempAlgIntra.SetCollOp(op); // CCU template需要传递op信息
429 :
430 0 : tempAlgInter.SetDmaMode(dmaMode_);
431 0 : tempAlgInter.SetDataType(dataType_);
432 0 : tempAlgInter.SetCollOp(op); // CCU template需要传递op信息
433 :
434 : // 计算算法模板所需资源
435 0 : CHK_RET(PrepareResForTemplate(rankGraph, tempAlgIntra, tempAlgInter));
436 :
437 0 : CHK_RET(GenInsQuesHost(tempAlgIntra, tempAlgInter));
438 0 : HCCL_INFO("[InsAllGatherParallelExecutor]Host Orchestrate success.");
439 :
440 0 : return HcclResult::HCCL_SUCCESS;
441 0 : }
442 :
443 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
444 0 : HcclResult InsAllGatherParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenInsQuesHost(
445 : InsAlgTemplate0 &tempAlgIntra, InsAlgTemplate1 &tempAlgInter)
446 : {
447 0 : HCCL_INFO("[InsAllGatherParallelExecutor] AlgTemplate inter server is [%s]", tempAlgIntra.Describe().c_str());
448 0 : HCCL_INFO("[InsAllGatherParallelExecutor] AlgTemplate intra server is [%s]", tempAlgInter.Describe().c_str());
449 0 : std::vector<float> dataSplitSize;
450 0 : GetParallelDataSplit(dataSplitSize);
451 0 : u64 alignedSize = 16 * 1024; // 假设需要16K对齐
452 0 : BufferType inBuffType = BufferType::INPUT;
453 0 : BufferType outBuffType = BufferType::OUTPUT;
454 0 : u32 intraScatchteMultipleStage0 = tempAlgIntra.CalcScratchMultiple(inBuffType, outBuffType);
455 0 : u32 interScatchteMultipleStage0 = tempAlgInter.CalcScratchMultiple(inBuffType, outBuffType);
456 0 : u32 intraScatchteMultipleStage1 = tempAlgIntra.CalcScratchMultiple(outBuffType, outBuffType);
457 0 : u32 interScatchteMultipleStage1 = tempAlgInter.CalcScratchMultiple(outBuffType, outBuffType);
458 0 : u32 scratchMultipleIntra = static_cast<u32>(std::max(std::ceil(dataSplitSize[0] * intraScatchteMultipleStage0),
459 0 : std::ceil(dataSplitSize[1] * intraScatchteMultipleStage1 * rankSizeLevel1_)));
460 0 : u32 scratchMultipleInter = static_cast<u32>(std::max(std::ceil(dataSplitSize[1] * interScatchteMultipleStage0),
461 0 : std::ceil(dataSplitSize[0] * interScatchteMultipleStage1 * rankSizeLevel0_)));
462 0 : u32 totalScratchMultiple = scratchMultipleIntra + scratchMultipleInter;
463 0 : u64 scratchMemBlockSize = maxTmpMemSize_;
464 0 : if (totalScratchMultiple > 0) {
465 0 : scratchMemBlockSize = (maxTmpMemSize_ / alignedSize / totalScratchMultiple) * alignedSize;
466 : }
467 0 : u64 intraScratchOffset = 0;
468 0 : u64 interScratchOffset = scratchMultipleIntra * scratchMemBlockSize;
469 :
470 : // dataSplitSize为分数，这里maxCountPerLoop对10取整
471 0 : u64 maxCountPerLoop =
472 0 : (std::min(static_cast<u64>(scratchMemBlockSize), static_cast<u64>(UB_MAX_DATA_SIZE)) / dataTypeSize_ / 10) * 10;
473 :
474 0 : u32 loopTimes = dataCount_ / maxCountPerLoop + ((dataCount_ % maxCountPerLoop == 0) ? 0 : 1);
475 :
476 0 : TemplateDataParams tempAlgParamsIntra0, tempAlgParamsInter0;
477 0 : TemplateDataParams tempAlgParamsInter1, tempAlgParamsIntra1;
478 0 : TempFuncs tempFuncs;
479 0 : tempFuncs.opMode = opMode_;
480 0 : tempFuncs.enableCounterNotify = false;
481 0 : tempFuncs.isBottom = true;
482 0 : tempFuncs.isForepart = true;
483 0 : for (u32 loopIndex = 0; loopIndex < loopTimes; loopIndex++) {
484 0 : u64 currCount = (loopIndex == loopTimes - 1) ? (dataCount_ - loopIndex * maxCountPerLoop) : maxCountPerLoop;
485 0 : u64 dataCountPerLoopAixs0 = static_cast<u64>(dataSplitSize[0] * currCount);
486 0 : u64 dataCountPerLoopAixs1 = currCount - dataCountPerLoopAixs0;
487 : // 第一步开始前同步
488 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
489 0 : u64 dataOffset0 = loopIndex * maxCountPerLoop * dataTypeSize_;
490 0 : u64 dataOffset1 = dataOffset0 + dataCountPerLoopAixs0 * dataTypeSize_;
491 : // 数据0的server内的mesh算法
492 0 : GenTemplateAlgParamsIntra0(dataOffset0, dataCountPerLoopAixs0, intraScratchOffset, tempAlgParamsIntra0);
493 : // 把每个template需要的queue传进去，比如stars的mesh要传多条queue
494 0 : CHK_RET(tempAlgIntra.GenExtIns(
495 : tempFuncs, tempAlgParamsIntra0, intraLinks_, intraQue_)); // Todo: 这里要把tempFuncs去掉
496 : // 数据1的server间的nhr算法
497 0 : GenTemplateAlgParamsInter1(dataOffset1, dataCountPerLoopAixs1, interScratchOffset, tempAlgParamsInter1);
498 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter1, interLinks_, interQue_));
499 : // 第一步做完后回到主流做尾同步
500 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
501 :
502 : // 第二步开始前同步
503 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
504 : // 数据0的server间的nhr算法
505 0 : GenTemplateAlgParamsInter0(dataOffset0, dataCountPerLoopAixs0, interScratchOffset, tempAlgParamsInter0);
506 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter0, interLinks_, interQue_));
507 : // 数据1的server内的mesh算法
508 0 : GenTemplateAlgParamsIntra1(dataOffset1, dataCountPerLoopAixs1, intraScratchOffset, tempAlgParamsIntra1);
509 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra1, intraLinks_, intraQue_));
510 : // 尾同步
511 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
512 : }
513 0 : return HcclResult::HCCL_SUCCESS;
514 0 : }
515 :
516 : INS_REGISTER_IMPL_BY_TWO_TEMPS(OpType::ALLGATHER, InsAllGatherParallelMesh1DNHR, InsAllGatherParallelExecutor,
517 : TopoMatchMeshNHR, InsTempAllGatherMesh1D, InsTempAllGatherNHR);
518 : INS_REGISTER_IMPL_BY_TWO_TEMPS(OpType::ALLGATHER, InsAllGatherParallelMesh2DNHR, InsAllGatherParallelExecutor,
519 : TopoMatchConcurrMeshNHR, InsTempAllGatherMesh2D, InsTempAllGatherNHR);
520 : INS_REGISTER_IMPL_BY_TWO_TEMPS(OpType::ALLGATHER, InsAllGatherParallelNHRNHR, InsAllGatherParallelExecutor,
521 : TopoMatchMeshNHR, InsTempAllGatherMesh2D, InsTempAllGatherNHR);
522 : INS_REGISTER_IMPL_BY_TWO_TEMPS(OpType::ALLGATHER, InsAllGatherParallelMesh1DNHRPcie, InsAllGatherParallelExecutor,
523 : TopoMatchMeshNHRPcie, InsTempAllGatherMesh1D, InsTempAllGatherNHR);
524 :
525 : // 算法注册
526 : #ifndef CCL_KERNEL_AICPU
527 : INS_REGISTER_IMPL_BY_TWO_TEMPS(OpType::ALLGATHER, CcuAllGatherParallelMesh1DNHR, InsAllGatherParallelExecutor,
528 : TopoMatchMeshNHR, CcuTempAllGatherMesh1DMem2MemWithStride, CcuTempAllGatherNHRMem2Mem1D);
529 : #endif
530 :
531 : } // namespace Hccl
|
