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(
237 : "[InsReduceScatterParallelExecutor]resReqIntra.queNum and resReqInter.queNum must be 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(
255 : "[InsReduceScatterParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]", 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(
278 : "[InsReduceScatterParallelExecutor]resReqIntra.queNum and resReqInter.queNum must be larger than 0.");
279 0 : return HcclResult::HCCL_E_INTERNAL;
280 : }
281 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
282 0 : CHK_RET(InitQueue(totalQueueNum, requireQue_));
283 0 : u32 intraQueNum = resReqIntra.queNum;
284 0 : for (u32 i = 0; i < requireQue_.size(); i++) {
285 0 : if (i < intraQueNum) {
286 0 : intraQue_.push_back(requireQue_[i]);
287 : } else {
288 0 : interQue_.push_back(requireQue_[i]);
289 : }
290 : }
291 0 : syncQueues_.emplace_back(intraQue_[0]);
292 0 : syncQueues_.emplace_back(interQue_[0]);
293 :
294 0 : CHK_RET(PrepResLinks(myRank_, resReqIntra.links, linkMgr, intraLinks_));
295 0 : CHK_RET(PrepResLinks(myRank_, resReqInter.links, linkMgr, interLinks_));
296 0 : HCCL_INFO(
297 : "[InsReduceScatterParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]", intraLinks_.size(),
298 : interLinks_.size());
299 0 : return HCCL_SUCCESS;
300 0 : }
301 :
302 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
303 0 : HcclResult InsReduceScatterParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
304 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
305 : InsQuePtr insQue)
306 : {
307 : // init and check params
308 0 : CHK_RET(Init(op, params, insQue));
309 :
310 0 : virtRanks_ = topoInfo.virtRanks;
311 0 : vTopo_ = topoInfo.vTopo;
312 0 : virtRankMap_ = topoInfo.virtRankMap;
313 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
314 0 : if (virtRankMap_[0].find(myRank_) != virtRankMap_[0].end()) {
315 0 : rankIdxLevel0_ = virtRankMap_[0][myRank_];
316 : } else {
317 0 : HCCL_ERROR("rank [%d] is not in level 0 topo", myRank_);
318 0 : return HcclResult::HCCL_E_INTERNAL;
319 : }
320 0 : if (virtRankMap_[1].find(myRank_) != virtRankMap_[1].end()) {
321 0 : rankIdxLevel1_ = virtRankMap_[1][myRank_];
322 : } else {
323 0 : HCCL_ERROR("rank [%d] is not in level 1 topo", myRank_);
324 0 : return HcclResult::HCCL_E_INTERNAL;
325 : }
326 :
327 : // 实例化算法模板类
328 0 : InsAlgTemplate0 tempAlgIntra(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
329 0 : InsAlgTemplate1 tempAlgInter(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
330 :
331 : // 实例化算法模板类
332 :
333 0 : tempAlgIntra.SetDmaMode(dmaMode_);
334 0 : tempAlgIntra.SetCollOp(op); // CCU template需要传递op信息
335 0 : tempAlgIntra.InitReduceInfo(redOp_, dataType_);
336 :
337 0 : tempAlgInter.SetDmaMode(dmaMode_);
338 0 : tempAlgInter.SetCollOp(op); // CCU template需要传递op信息
339 0 : tempAlgInter.InitReduceInfo(redOp_, dataType_);
340 :
341 0 : std::vector<std::map<u32, u32>> rank2PathNumMap;
342 0 : SetPathNumMapByLinkMgrMultiLevel(linkMgr, virtRanks_, myRank_, rank2PathNumMap);
343 0 : tempAlgIntra.setPathNumMap(rank2PathNumMap[0]);
344 0 : tempAlgInter.setPathNumMap(rank2PathNumMap[1]);
345 :
346 : // 计算算法模板所需资源
347 0 : CHK_RET(PrepareResForTemplate(linkMgr, tempAlgIntra, tempAlgInter));
348 :
349 0 : CHK_RET(GenInsQuesHost(tempAlgIntra, tempAlgInter));
350 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] Orchestrate success.");
351 :
352 0 : return HcclResult::HCCL_SUCCESS;
353 0 : }
354 :
355 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
356 0 : HcclResult InsReduceScatterParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
357 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
358 : {
359 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] Host Orchestrate begins.");
360 : // init and check params
361 0 : CHK_RET(Init(op, params, insQue));
362 :
363 : // Topo Match
364 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
365 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
366 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
367 :
368 0 : if (virtRankMap_[0].find(myRank_) != virtRankMap_[0].end()) {
369 0 : rankIdxLevel0_ = virtRankMap_[0][myRank_];
370 : } else {
371 0 : HCCL_ERROR("rank [%d] is not in level 0 topo", myRank_);
372 0 : return HcclResult::HCCL_E_INTERNAL;
373 : }
374 0 : if (virtRankMap_[1].find(myRank_) != virtRankMap_[1].end()) {
375 0 : rankIdxLevel1_ = virtRankMap_[1][myRank_];
376 : } else {
377 0 : HCCL_ERROR("rank [%d] is not in level 1 topo", myRank_);
378 0 : return HcclResult::HCCL_E_INTERNAL;
379 : }
380 :
381 : // 实例化算法模板类
382 0 : InsAlgTemplate0 tempAlgIntra(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
383 0 : InsAlgTemplate1 tempAlgInter(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
384 :
385 : // 实例化算法模板类
386 :
387 0 : tempAlgIntra.SetDmaMode(dmaMode_);
388 0 : tempAlgIntra.SetCollOp(op); // CCU template需要传递op信息
389 0 : tempAlgIntra.InitReduceInfo(redOp_, dataType_);
390 :
391 0 : tempAlgInter.SetDmaMode(dmaMode_);
392 0 : tempAlgInter.SetCollOp(op); // CCU template需要传递op信息
393 0 : tempAlgInter.InitReduceInfo(redOp_, dataType_);
394 :
395 : // 计算算法模板所需资源
396 0 : CHK_RET(PrepareResForTemplate(rankGraph, tempAlgIntra, tempAlgInter));
397 :
398 0 : CHK_RET(GenInsQuesHost(tempAlgIntra, tempAlgInter));
399 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] Orchestrate success.");
400 :
401 0 : return HcclResult::HCCL_SUCCESS;
402 0 : }
403 :
404 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
405 0 : HcclResult InsReduceScatterParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenInsQuesHost(
406 : InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
407 : {
408 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] AlgTemplate inter server is [%s]", tempAlgInter.Describe().c_str());
409 0 : HCCL_INFO("[InsReduceScatterParallelExecutor] AlgTemplate intra server is [%s]", tempAlgIntra.Describe().c_str());
410 0 : std::vector<float> dataSplitSize;
411 0 : GetParallelDataSplit(dataSplitSize);
412 0 : u64 alignedSize = 16 * 1024; // 假设需要16K对齐
413 0 : BufferType inBuffType = BufferType::INPUT;
414 0 : BufferType outBuffType = BufferType::OUTPUT;
415 0 : u32 intraScatchteMultipleStage0 = tempAlgIntra.CalcScratchMultiple(inBuffType, outBuffType);
416 0 : u32 interScatchteMultipleStage0 = tempAlgInter.CalcScratchMultiple(inBuffType, outBuffType);
417 0 : u32 intraScatchteMultipleStage1 = tempAlgIntra.CalcScratchMultiple(outBuffType, outBuffType);
418 0 : u32 interScatchteMultipleStage1 = tempAlgInter.CalcScratchMultiple(outBuffType, outBuffType);
419 0 : if (interScatchteMultipleStage0 == 0 || interScatchteMultipleStage1 == 0) {
420 0 : interScatchteMultipleStage0 = rankSizeLevel1_;
421 0 : interScatchteMultipleStage1 = rankSizeLevel1_;
422 : }
423 0 : u32 scratchMultipleIntra0
424 0 : = static_cast<u32>(std::ceil(dataSplitSize[0] * intraScatchteMultipleStage0 * rankSizeLevel1_));
425 0 : u32 scratchMultipleIntra1 = static_cast<u32>(std::ceil(dataSplitSize[1] * intraScatchteMultipleStage1));
426 0 : u32 scratchMultipleInter1
427 0 : = static_cast<u32>(std::ceil(dataSplitSize[1] * interScatchteMultipleStage0 * rankSizeLevel0_));
428 0 : u32 scratchMultipleInter0 = static_cast<u32>(std::ceil(dataSplitSize[0] * interScatchteMultipleStage1));
429 0 : u32 totalScratchMultiple
430 0 : = scratchMultipleIntra0 + scratchMultipleIntra1 + scratchMultipleInter0 + scratchMultipleInter1;
431 0 : u64 scratchMemBlockSize = maxTmpMemSize_;
432 0 : if (totalScratchMultiple > 0) {
433 0 : scratchMemBlockSize = (maxTmpMemSize_ / alignedSize / totalScratchMultiple) * alignedSize;
434 : }
435 0 : u64 intra0ScratchOffset = 0;
436 0 : u64 intra1ScratchOffset = intra0ScratchOffset + scratchMultipleIntra0 * scratchMemBlockSize;
437 0 : u64 inter0ScratchOffset = intra1ScratchOffset + scratchMultipleIntra1 * scratchMemBlockSize;
438 0 : u64 inter1ScratchOffset = inter0ScratchOffset + scratchMultipleInter0 * scratchMemBlockSize;
439 0 : std::vector<u64> scratchOffVec
440 : = {intra0ScratchOffset, intra1ScratchOffset, inter0ScratchOffset, inter1ScratchOffset};
441 :
442 : // dataSplitSize为分数，这里maxCountPerLoop对10取整
443 0 : u64 maxCountPerLoop
444 0 : = (std::min(static_cast<u64>(scratchMemBlockSize), static_cast<u64>(UB_MAX_DATA_SIZE)) / dataTypeSize_ / 10)
445 : * 10;
446 0 : u32 loopTimes = dataCount_ / maxCountPerLoop + ((dataCount_ % maxCountPerLoop == 0) ? 0 : 1);
447 :
448 0 : TempFuncs tempFuncs;
449 0 : tempFuncs.opMode = opMode_;
450 0 : tempFuncs.enableCounterNotify = false;
451 0 : TemplateDataParams tempAlgParamsIntra0;
452 0 : TemplateDataParams tempAlgParamsInter0;
453 0 : TemplateDataParams tempAlgParamsInter1;
454 0 : TemplateDataParams tempAlgParamsIntra1;
455 0 : for (u32 loopIndex = 0; loopIndex < loopTimes; loopIndex++) {
456 0 : u64 currCount = (loopIndex == loopTimes - 1) ? (dataCount_ - loopIndex * maxCountPerLoop) : maxCountPerLoop;
457 0 : u64 dataCountPerLoopAixs0 = static_cast<u64>(dataSplitSize[0] * currCount);
458 0 : u64 dataCountPerLoopAixs1 = currCount - dataCountPerLoopAixs0;
459 : // 第一步开始前同步
460 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
461 0 : u64 dataOffset0 = loopIndex * maxCountPerLoop * dataTypeSize_;
462 0 : u64 dataOffset1 = dataOffset0 + dataCountPerLoopAixs0 * dataTypeSize_;
463 : // 数据0的server内的mesh算法
464 0 : GenTemplateAlgParamsIntra0(dataOffset0, dataCountPerLoopAixs0, scratchOffVec, tempAlgParamsIntra0);
465 : // 把每个template需要的queue传进去，比如stars的mesh要传多条queue
466 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra0, intraLinks_, intraQue_));
467 : // 数据1的server间的nhr算法
468 0 : GenTemplateAlgParamsInter1(dataOffset1, dataCountPerLoopAixs1, scratchOffVec, tempAlgParamsInter1);
469 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter1, interLinks_, interQue_));
470 : // 第一步做完后回到主流做尾同步
471 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
472 :
473 : // 第二步开始前同步
474 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
475 : // 数据0的server间的nhr算法
476 0 : tempFuncs.isBottom = true;
477 0 : GenTemplateAlgParamsInter0(dataOffset0, dataCountPerLoopAixs0, scratchOffVec, tempAlgParamsInter0);
478 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter0, interLinks_, interQue_));
479 0 : tempFuncs.isBottom = false;
480 : // 数据1的server内的mesh算法
481 0 : GenTemplateAlgParamsIntra1(dataOffset1, dataCountPerLoopAixs1, scratchOffVec, tempAlgParamsIntra1);
482 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra1, intraLinks_, intraQue_));
483 : // 尾同步
484 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
485 : }
486 0 : return HcclResult::HCCL_SUCCESS;
487 0 : }
488 :
489 : // 算法注册
490 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
491 : OpType::REDUCESCATTER, InsReduceScatterParallelMesh1DNHR, InsReduceScatterParallelExecutor, TopoMatchMeshNHR,
492 : InsTempReduceScatterMesh1D, InsTempReduceScatterNHR);
493 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
494 : OpType::REDUCESCATTER, InsReduceScatterParallelMesh2DNHR, InsReduceScatterParallelExecutor, TopoMatchConcurrMeshNHR,
495 : InsTempReduceScatterMesh2D, InsTempReduceScatterNHR);
496 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
497 : OpType::REDUCESCATTER, InsReduceScatterParallelNHRNHR, InsReduceScatterParallelExecutor, TopoMatchConcurrMeshNHR,
498 : InsTempReduceScatterNHR, InsTempReduceScatterNHR);
499 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
500 : OpType::REDUCESCATTER, InsReduceScatterParallelMesh1DNHRPcie, InsReduceScatterParallelExecutor,
501 : TopoMatchMeshNHRPcie, InsTempReduceScatterMesh1D, InsTempReduceScatterNHR);
502 : #ifndef CCL_KERNEL_AICPU
503 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
504 : OpType::REDUCESCATTER, CcuReduceScatterParallelMesh1DNHR, InsReduceScatterParallelExecutor, TopoMatchMeshNHR,
505 : CcuTempReduceScatterMeshMem2Mem1D, CcuTempReduceScatterNHR1DMem2Mem);
506 : #endif
507 : } // namespace Hccl
|
