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