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(
73 : "CalResOffload resReqIntra.streamNum [%u], resReqInter.streamNum [%u]", 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 : 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 : 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(
221 : const RankGraph* rankGraph, InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
222 : {
223 0 : AlgTempResReq resReqIntra;
224 0 : AlgTempResReq resReqInter;
225 0 : if (enableDetour_) {
226 0 : HCCL_DEBUG("[InsReduceScatterParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
227 0 : CHK_RET(tempAlgIntra.CalcResDetour(rankGraph, resReqIntra));
228 : } else {
229 0 : HCCL_DEBUG("[InsReduceScatterParallelExecutor] Rank[%d], CalcRes with detour disabled.", myRank_);
230 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
231 : }
232 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
233 :
234 : // 申请算法模板所需资源
235 0 : if (!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
236 0 : HCCL_ERROR("[InsReduceScatterParallelExecutor]resReqIntra.queNum and resReqInter.queNum must larger than 0.");
237 0 : return HcclResult::HCCL_E_INTERNAL;
238 : }
239 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
240 0 : CHK_RET(InitQueue(totalQueueNum, requireQue_));
241 0 : for (u32 i = 0; i < requireQue_.size(); i++) {
242 0 : if (i < resReqIntra.queNum) {
243 0 : intraQue_.push_back(requireQue_[i]);
244 : } else {
245 0 : interQue_.push_back(requireQue_[i]);
246 : }
247 : }
248 0 : syncQueues_.emplace_back(intraQue_[0]);
249 0 : syncQueues_.emplace_back(interQue_[0]);
250 :
251 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, intraLinks_));
252 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, interLinks_));
253 0 : HCCL_INFO(
254 : "[InsReduceScatterParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]", intraLinks_.size(),
255 : interLinks_.size());
256 0 : return HCCL_SUCCESS;
257 0 : }
258 :
259 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
260 0 : HcclResult InsReduceScatterParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
261 : ConnectedLinkMgr* linkMgr, InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
262 : {
263 0 : AlgTempResReq resReqInter;
264 0 : AlgTempResReq resReqIntra;
265 0 : if (enableDetour_) {
266 0 : HCCL_DEBUG("[InsReduceScatterParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
267 0 : CHK_RET(tempAlgIntra.CalcResDetour(linkMgr, resReqIntra));
268 : } else {
269 0 : HCCL_DEBUG("[InsReduceScatterParallelExecutor] Rank[%d], CalcRes with detour disabled.", myRank_);
270 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
271 : }
272 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
273 :
274 : // 申请算法模板所需资源
275 0 : if (!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
276 0 : HCCL_ERROR("[InsReduceScatterParallelExecutor]resReqIntra.queNum and resReqInter.queNum must larger than 0.");
277 0 : return HcclResult::HCCL_E_INTERNAL;
278 : }
279 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
280 0 : CHK_RET(InitQueue(totalQueueNum, requireQue_));
281 0 : u32 intraQueNum = resReqIntra.queNum;
282 0 : for (u32 i = 0; i < requireQue_.size(); i++) {
283 0 : if (i < intraQueNum) {
284 0 : intraQue_.push_back(requireQue_[i]);
285 : } else {
286 0 : interQue_.push_back(requireQue_[i]);
287 : }
288 : }
289 0 : syncQueues_.emplace_back(intraQue_[0]);
290 0 : syncQueues_.emplace_back(interQue_[0]);
291 :
292 0 : CHK_RET(PrepResLinks(myRank_, resReqIntra.links, linkMgr, intraLinks_));
293 0 : CHK_RET(PrepResLinks(myRank_, resReqInter.links, linkMgr, interLinks_));
294 0 : HCCL_INFO(
295 : "[InsReduceScatterParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]", intraLinks_.size(),
296 : interLinks_.size());
297 0 : return HCCL_SUCCESS;
298 0 : }
299 :
300 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
301 0 : HcclResult InsReduceScatterParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
302 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
303 : InsQuePtr insQue)
304 : {
305 : // init and check params
306 0 : CHK_RET(Init(op, params, insQue));
307 :
308 0 : virtRanks_ = topoInfo.virtRanks;
309 0 : vTopo_ = topoInfo.vTopo;
310 0 : virtRankMap_ = topoInfo.virtRankMap;
311 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
312 0 : if (virtRankMap_[0].find(myRank_) != virtRankMap_[0].end()) {
313 0 : rankIdxLevel0_ = virtRankMap_[0][myRank_];
314 : } else {
315 0 : HCCL_ERROR("rank [%d] is not in level 0 topo", myRank_);
316 0 : return HcclResult::HCCL_E_INTERNAL;
317 : }
318 0 : if (virtRankMap_[1].find(myRank_) != virtRankMap_[1].end()) {
319 0 : rankIdxLevel1_ = virtRankMap_[1][myRank_];
320 : } else {
321 0 : HCCL_ERROR("rank [%d] is not in level 1 topo", myRank_);
322 0 : return HcclResult::HCCL_E_INTERNAL;
323 : }
324 :
325 : // 实例化算法模板类
326 0 : InsAlgTemplate0 tempAlgIntra(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
327 0 : InsAlgTemplate1 tempAlgInter(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
328 :
329 : // 实例化算法模板类
330 :
331 0 : tempAlgIntra.SetDmaMode(dmaMode_);
332 0 : tempAlgIntra.SetCollOp(op); // CCU template需要传递op信息
333 0 : tempAlgIntra.InitReduceInfo(redOp_, dataType_);
334 :
335 0 : tempAlgInter.SetDmaMode(dmaMode_);
336 0 : tempAlgInter.SetCollOp(op); // CCU template需要传递op信息
337 0 : tempAlgInter.InitReduceInfo(redOp_, dataType_);
338 :
339 0 : std::vector<std::map<u32, u32>> rank2PathNumMap;
340 0 : SetPathNumMapByLinkMgrMultiLevel(linkMgr, virtRanks_, myRank_, rank2PathNumMap);
341 0 : tempAlgIntra.setPathNumMap(rank2PathNumMap[0]);
342 0 : tempAlgInter.setPathNumMap(rank2PathNumMap[1]);
343 :
344 : // 计算算法模板所需资源
345 0 : CHK_RET(PrepareResForTemplate(linkMgr, tempAlgIntra, tempAlgInter));
346 :
347 0 : CHK_RET(GenInsQuesHost(tempAlgIntra, tempAlgInter));
348 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] Orchestrate success.");
349 :
350 0 : return HcclResult::HCCL_SUCCESS;
351 0 : }
352 :
353 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
354 0 : HcclResult InsReduceScatterParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
355 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
356 : {
357 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] Host Orchestrate begins.");
358 : // init and check params
359 0 : CHK_RET(Init(op, params, insQue));
360 :
361 : // Topo Match
362 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
363 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
364 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
365 :
366 0 : if (virtRankMap_[0].find(myRank_) != virtRankMap_[0].end()) {
367 0 : rankIdxLevel0_ = virtRankMap_[0][myRank_];
368 : } else {
369 0 : HCCL_ERROR("rank [%d] is not in level 0 topo", myRank_);
370 0 : return HcclResult::HCCL_E_INTERNAL;
371 : }
372 0 : if (virtRankMap_[1].find(myRank_) != virtRankMap_[1].end()) {
373 0 : rankIdxLevel1_ = virtRankMap_[1][myRank_];
374 : } else {
375 0 : HCCL_ERROR("rank [%d] is not in level 1 topo", myRank_);
376 0 : return HcclResult::HCCL_E_INTERNAL;
377 : }
378 :
379 : // 实例化算法模板类
380 0 : InsAlgTemplate0 tempAlgIntra(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
381 0 : InsAlgTemplate1 tempAlgInter(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
382 :
383 : // 实例化算法模板类
384 :
385 0 : tempAlgIntra.SetDmaMode(dmaMode_);
386 0 : tempAlgIntra.SetCollOp(op); // CCU template需要传递op信息
387 0 : tempAlgIntra.InitReduceInfo(redOp_, dataType_);
388 :
389 0 : tempAlgInter.SetDmaMode(dmaMode_);
390 0 : tempAlgInter.SetCollOp(op); // CCU template需要传递op信息
391 0 : tempAlgInter.InitReduceInfo(redOp_, dataType_);
392 :
393 : // 计算算法模板所需资源
394 0 : CHK_RET(PrepareResForTemplate(rankGraph, tempAlgIntra, tempAlgInter));
395 :
396 0 : CHK_RET(GenInsQuesHost(tempAlgIntra, tempAlgInter));
397 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] Orchestrate success.");
398 :
399 0 : return HcclResult::HCCL_SUCCESS;
400 0 : }
401 :
402 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
403 0 : HcclResult InsReduceScatterParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenInsQuesHost(
404 : InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
405 : {
406 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] AlgTemplate inter server is [%s]", tempAlgIntra.Describe().c_str());
407 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] AlgTemplate intra server is [%s]", tempAlgInter.Describe().c_str());
408 0 : std::vector<float> dataSplitSize;
409 0 : GetParallelDataSplit(dataSplitSize);
410 0 : u64 alignedSize = 16 * 1024; // 假设需要16K对齐
411 0 : BufferType inBuffType = BufferType::INPUT;
412 0 : BufferType outBuffType = BufferType::OUTPUT;
413 0 : u32 intraScatchteMultipleStage0 = tempAlgIntra.CalcScratchMultiple(inBuffType, outBuffType);
414 0 : u32 interScatchteMultipleStage0 = tempAlgInter.CalcScratchMultiple(inBuffType, outBuffType);
415 0 : u32 intraScatchteMultipleStage1 = tempAlgIntra.CalcScratchMultiple(outBuffType, outBuffType);
416 0 : u32 interScatchteMultipleStage1 = tempAlgInter.CalcScratchMultiple(outBuffType, outBuffType);
417 0 : if (interScatchteMultipleStage0 == 0 || interScatchteMultipleStage1 == 0) {
418 0 : interScatchteMultipleStage0 = rankSizeLevel1_;
419 0 : interScatchteMultipleStage1 = rankSizeLevel1_;
420 : }
421 0 : u32 scratchMultipleIntra0
422 0 : = static_cast<u32>(std::ceil(dataSplitSize[0] * intraScatchteMultipleStage0 * rankSizeLevel1_));
423 0 : u32 scratchMultipleIntra1 = static_cast<u32>(std::ceil(dataSplitSize[1] * intraScatchteMultipleStage1));
424 0 : u32 scratchMultipleInter1
425 0 : = static_cast<u32>(std::ceil(dataSplitSize[1] * interScatchteMultipleStage0 * rankSizeLevel0_));
426 0 : u32 scratchMultipleInter0 = static_cast<u32>(std::ceil(dataSplitSize[0] * interScatchteMultipleStage1));
427 0 : u32 totalScratchMultiple
428 0 : = scratchMultipleIntra0 + scratchMultipleIntra1 + scratchMultipleInter0 + scratchMultipleInter1;
429 0 : u64 scratchMemBlockSize = maxTmpMemSize_;
430 0 : if (totalScratchMultiple > 0) {
431 0 : scratchMemBlockSize = (maxTmpMemSize_ / alignedSize / totalScratchMultiple) * alignedSize;
432 : }
433 0 : u64 intra0ScratchOffset = 0;
434 0 : u64 intra1ScratchOffset = intra0ScratchOffset + scratchMultipleIntra0 * scratchMemBlockSize;
435 0 : u64 inter0ScratchOffset = intra1ScratchOffset + scratchMultipleIntra1 * scratchMemBlockSize;
436 0 : u64 inter1ScratchOffset = inter0ScratchOffset + scratchMultipleInter0 * scratchMemBlockSize;
437 0 : std::vector<u64> scratchOffVec
438 : = {intra0ScratchOffset, intra1ScratchOffset, inter0ScratchOffset, inter1ScratchOffset};
439 :
440 : // dataSplitSize为分数，这里maxCountPerLoop对10取整
441 0 : u64 maxCountPerLoop
442 0 : = (std::min(static_cast<u64>(scratchMemBlockSize), static_cast<u64>(UB_MAX_DATA_SIZE)) / dataTypeSize_ / 10)
443 : * 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(
489 : OpType::REDUCESCATTER, InsReduceScatterParallelMesh1DNHR, InsReduceScatterParallelExecutor, TopoMatchMeshNHR,
490 : InsTempReduceScatterMesh1D, InsTempReduceScatterNHR);
491 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
492 : OpType::REDUCESCATTER, InsReduceScatterParallelMesh2DNHR, InsReduceScatterParallelExecutor, TopoMatchConcurrMeshNHR,
493 : InsTempReduceScatterMesh2D, InsTempReduceScatterNHR);
494 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
495 : OpType::REDUCESCATTER, InsReduceScatterParallelNHRNHR, InsReduceScatterParallelExecutor, TopoMatchConcurrMeshNHR,
496 : InsTempReduceScatterNHR, InsTempReduceScatterNHR);
497 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
498 : OpType::REDUCESCATTER, InsReduceScatterParallelMesh1DNHRPcie, InsReduceScatterParallelExecutor,
499 : TopoMatchMeshNHRPcie, InsTempReduceScatterMesh1D, InsTempReduceScatterNHR);
500 : #ifndef CCL_KERNEL_AICPU
501 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
502 : OpType::REDUCESCATTER, CcuReduceScatterParallelMesh1DNHR, InsReduceScatterParallelExecutor, TopoMatchMeshNHR,
503 : CcuTempReduceScatterMeshMem2Mem1D, CcuTempReduceScatterNHR1DMem2Mem);
504 : #endif
505 : } // namespace Hccl
|
