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