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 "log.h"
12 : #include "ins_coll_alg_registry.h"
13 : #include "topo_match_mesh_nhr.h"
14 : #include "topo_match_mesh_nhr_pcie.h"
15 : #include "alg_data_trans_wrapper.h"
16 :
17 : #include "ins_temp_broadcast_mesh_1D_two_shot.h"
18 : #include "ins_temp_broadcast_nhr.h"
19 : #include "ccu_temp_broadcast_mesh_1D_mem2mem.h"
20 : #include "ccu_temp_broadcast_nhr_1D_mem2mem.h"
21 : #include "ins_broadcast_parallel_executor.h"
22 :
23 :
24 : namespace Hccl {
25 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
26 0 : InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::InsBroadcastParallelExecutor()
27 0 : : InsCollAlgBase()
28 : {
29 0 : }
30 :
31 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
32 0 : InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::~InsBroadcastParallelExecutor()
33 : {
34 0 : }
35 :
36 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
37 0 : HcclResult InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcRes(
38 : const RankGraph *rankGraph, CollAlgResReq &algResReq)
39 : {
40 0 : HCCL_INFO("[InsBroadcastParallelExecutor] CalcRes start, rank[%d]", myRank_);
41 :
42 : // Topo Match
43 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
44 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
45 0 : algResReq.topoInfo.UpdateMultiLevelTopo(virtRanks_, virtRankMap_, vTopo_);
46 :
47 : // 计算localRankSize
48 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, intraLocalRankSize_, interLocalRankSize_));
49 :
50 : // 实例化算法模板类
51 0 : InsAlgTemplate0 intraTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
52 0 : InsAlgTemplate1 interTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
53 :
54 : // 计算和准备Queue资源
55 0 : AlgTempResReq resReqIntra;
56 0 : AlgTempResReq resReqInter;
57 0 : if (enableDetour_) {
58 0 : HCCL_DEBUG("[InsBroadcastParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
59 0 : CHK_RET(intraTempAlg.CalcResDetour(rankGraph, resReqIntra));
60 : } else {
61 0 : HCCL_DEBUG("[InsBroadcastParallelExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
62 0 : CHK_RET(intraTempAlg.CalcRes(resReqIntra));
63 : }
64 0 : CHK_RET(interTempAlg.CalcRes(resReqInter));
65 :
66 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqIntra.links, algResReq.levelRankPairs));
67 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqInter.links, algResReq.levelRankPairs));
68 0 : algResReq.primQueueNum = resReqIntra.streamNum + resReqInter.streamNum;
69 :
70 0 : CHK_RET(CalcParallelNotifyReq(algResReq.primQueueNum, resReqIntra.queNum, algResReq.queueNotifys));
71 :
72 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, algResReq.links));
73 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, algResReq.links));
74 :
75 0 : HCCL_INFO("[InsBroadcastParallelExecutor] CalcRes end, rank[%d], required total que num [%u], que notify num [%u]",
76 : myRank_, algResReq.primQueueNum, algResReq.queueNotifys.size());
77 :
78 0 : return HcclResult::HCCL_SUCCESS;
79 0 : }
80 :
81 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
82 0 : HcclResult InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcResOffload(
83 : const RankGraph *rankGraph, const u64 &dataSize, CollOffloadOpResReq &resReq)
84 : {
85 0 : HCCL_INFO("[InsBroadcastParallelExecutor] CalcResOffload start, rank[%d]", myRank_);
86 :
87 : (void)dataSize;
88 0 : u64 scratchMemSize = 200 * 1024 * 1024;
89 0 : resReq.requiredScratchMemSize = scratchMemSize; // 200MB
90 : // Topo Match
91 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
92 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
93 :
94 : // 计算localRankSize
95 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, intraLocalRankSize_, interLocalRankSize_));
96 :
97 : // 实例化算法模板类
98 0 : InsAlgTemplate0 intraTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
99 0 : InsAlgTemplate1 interTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
100 :
101 : // 计算和准备Queue资源
102 0 : AlgTempResReq resReqIntra;
103 0 : AlgTempResReq resReqInter;
104 0 : if (enableDetour_) {
105 0 : HCCL_DEBUG("[InsBroadcastParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
106 0 : CHK_RET(intraTempAlg.CalcResDetour(rankGraph, resReqIntra));
107 : } else {
108 0 : HCCL_DEBUG("[InsBroadcastParallelExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
109 0 : CHK_RET(intraTempAlg.CalcRes(resReqIntra));
110 : }
111 0 : CHK_RET(interTempAlg.CalcRes(resReqInter));
112 :
113 0 : resReq.requiredSubQueNum = resReqIntra.streamNum + resReqInter.streamNum - 1;
114 :
115 0 : HCCL_INFO("[InsBroadcastParallelExecutor] CalcResOffload end, rank[%d], required sub que num is [%u]",
116 : myRank_, resReq.requiredSubQueNum);
117 :
118 0 : return HcclResult::HCCL_SUCCESS;
119 0 : }
120 :
121 : // Host展开
122 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
123 0 : HcclResult InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
124 : const RankGraph *rankGraph, const CollAlgOperator &op, const CollAlgParams ¶ms, InsQuePtr insQue)
125 : {
126 0 : HCCL_INFO("[InsBroadcastParallelExecutor] Host orchestrate begins.");
127 :
128 : // 初始化参数
129 0 : CHK_RET(Init(op, params, insQue));
130 :
131 : // 获取算法Topo信息
132 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
133 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
134 :
135 : // 计算localRankSize和localRoot
136 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, intraLocalRankSize_, interLocalRankSize_));
137 0 : CHK_RET(CalcLocalRoot());
138 :
139 : // 实例化算法模板类
140 0 : InsAlgTemplate0 tempAlgIntra(myRank_, intraLocalRankSize_, vTopo_.at(0), virtRankMap_.at(0)); //server内算法
141 0 : InsAlgTemplate1 tempAlgInter(myRank_, interLocalRankSize_, vTopo_.at(1), virtRankMap_.at(1)); //server间算法
142 :
143 : // 传入Template参数
144 0 : tempAlgIntra.SetDmaMode(dmaMode_);
145 0 : tempAlgIntra.SetCollOp(op);
146 0 : tempAlgIntra.SetDataType(dataType_);
147 0 : tempAlgIntra.SetRoot(intraLocalRoot_);
148 :
149 0 : tempAlgInter.SetDmaMode(dmaMode_);
150 0 : tempAlgInter.SetCollOp(op);
151 0 : tempAlgInter.SetDataType(dataType_);
152 0 : tempAlgInter.SetRoot(interLocalRoot_);
153 :
154 : // 计算算法模板所需资源
155 0 : CHK_RET(PrepareResForTemplate(rankGraph, tempAlgIntra, tempAlgInter));
156 :
157 : // 算法展开
158 0 : CHK_RET(GenInsQues(tempAlgIntra, tempAlgInter));
159 :
160 0 : HCCL_INFO("[InsBroadcastParallelExecutor] Host orchestrate success.");
161 0 : return HcclResult::HCCL_SUCCESS;
162 0 : }
163 :
164 : // Aicpu展开
165 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
166 0 : HcclResult InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
167 : const AlgTopoInfo &topoInfo, const CollAlgOperator &op, const CollAlgParams ¶ms, ConnectedLinkMgr *linkMgr,
168 : InsQuePtr insQue)
169 : {
170 0 : HCCL_INFO("[InsBroadcastParallelExecutor] Aicpu orchestrate begins.");
171 :
172 : // 初始化参数
173 0 : CHK_RET(Init(op, params, insQue));
174 :
175 : // 获取算法Topo信息
176 0 : vTopo_ = topoInfo.vTopo; // 本通信域内的通信平面
177 0 : virtRanks_ = topoInfo.virtRanks; // 本通信域内的 rank 集合
178 0 : virtRankMap_ = topoInfo.virtRankMap; // 本通信域内的 rank 映射表
179 :
180 : // 计算localRankSize和localRoot
181 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, intraLocalRankSize_, interLocalRankSize_));
182 0 : CHK_RET(CalcLocalRoot());
183 :
184 : // 实例化算法模板类
185 0 : InsAlgTemplate0 tempAlgIntra(myRank_, intraLocalRankSize_, vTopo_.at(0), virtRankMap_.at(0)); //server内算法
186 0 : InsAlgTemplate1 tempAlgInter(myRank_, interLocalRankSize_, vTopo_.at(1), virtRankMap_.at(1)); //server间算法
187 :
188 : // 传入Template参数
189 0 : tempAlgIntra.SetDmaMode(dmaMode_);
190 0 : tempAlgIntra.SetCollOp(op);
191 0 : tempAlgIntra.SetDataType(dataType_);
192 0 : tempAlgIntra.SetRoot(intraLocalRoot_);
193 :
194 0 : tempAlgInter.SetDmaMode(dmaMode_);
195 0 : tempAlgInter.SetCollOp(op);
196 0 : tempAlgInter.SetDataType(dataType_);
197 0 : tempAlgInter.SetRoot(interLocalRoot_);
198 :
199 : // 计算算法模板所需资源
200 0 : CHK_RET(PrepareResForTemplate(linkMgr, tempAlgIntra, tempAlgInter));
201 :
202 : // 算法展开
203 0 : CHK_RET(GenInsQues(tempAlgIntra, tempAlgInter));
204 :
205 0 : HCCL_INFO("[InsBroadcastParallelExecutor] Aicpu orchestrate success.");
206 0 : return HcclResult::HCCL_SUCCESS;
207 0 : }
208 :
209 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
210 0 : void InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GetParallelDataSplit(
211 : std::vector<float> &splitDataSize) const
212 : {
213 : // to do 先做等分，后续根据性能做调整
214 0 : double splitData = 0.5;
215 0 : splitDataSize.push_back(splitData);
216 0 : splitDataSize.push_back(splitData);
217 0 : return;
218 : }
219 :
220 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
221 0 : HcclResult InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcLocalRoot()
222 : {
223 0 : CHK_PRT_RET(root_ >= rankSize_,
224 : HCCL_ERROR("[CalcLocalRoot] root[%u] is out of rankSize[%u]", root_, rankSize_),
225 : HcclResult::HCCL_E_INTERNAL);
226 :
227 0 : u32 intraLocalRootIdx = root_ % intraLocalRankSize_;
228 0 : intraLocalRoot_ = static_cast<u32>(vTopo_.at(0).at(0).at(intraLocalRootIdx));
229 0 : u32 interLocalRootIdx = root_ / intraLocalRankSize_;
230 0 : interLocalRoot_ = static_cast<u32>(vTopo_.at(1).at(0).at(interLocalRootIdx));
231 :
232 0 : HCCL_INFO("[CalcLocalRoot] localRoot: myRank[%d] intraLocalRoot[%u] interLocalRoot[%u]",
233 : myRank_, intraLocalRoot_, interLocalRoot_);
234 0 : return HcclResult::HCCL_SUCCESS;
235 : }
236 :
237 : // Host
238 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
239 0 : HcclResult InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
240 : const RankGraph *rankGraph, InsAlgTemplate0 &tempAlgIntra, InsAlgTemplate1 &tempAlgInter)
241 : {
242 0 : AlgTempResReq resReqIntra;
243 0 : AlgTempResReq resReqInter;
244 0 : if (enableDetour_) {
245 0 : HCCL_DEBUG("[%s] Rank[%d], CalcRes with detouring enabled", __func__, myRank_);
246 0 : CHK_RET(tempAlgIntra.CalcResDetour(rankGraph, resReqIntra));
247 : } else {
248 0 : HCCL_DEBUG("[%s] Rank[%d], CalcRes with detouring disabled", __func__, myRank_);
249 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
250 : }
251 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
252 :
253 : // 申请算法模板所需资源
254 0 : if(!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
255 0 : HCCL_ERROR("[InsBroadcastParallelExecutor]resReqIntra.queNum and resReqInter.queNum must > 0.");
256 0 : return HcclResult::HCCL_E_INTERNAL;
257 : }
258 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
259 0 : CHK_RET(InitQueue(totalQueueNum, requiredQue_));
260 0 : for(u32 i = 0 ; i < requiredQue_.size(); i++) {
261 0 : if (i < resReqIntra.queNum) {
262 0 : intraQue_.push_back(requiredQue_.at(i));
263 : } else {
264 0 : interQue_.push_back(requiredQue_.at(i));
265 : }
266 : }
267 : // 每个算法的第0条流用于同步
268 0 : syncQueues_.emplace_back(intraQue_.at(0));
269 0 : syncQueues_.emplace_back(interQue_.at(0));
270 :
271 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, intraLinks_));
272 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, interLinks_));
273 0 : HCCL_INFO("[InsBroadcastParallelExecutor] intraLinks size[%zu], interLinks size[%zu]",
274 : intraLinks_.size(), interLinks_.size());
275 :
276 0 : return HcclResult::HCCL_SUCCESS;
277 0 : }
278 :
279 : // Aicpu
280 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
281 0 : HcclResult InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
282 : ConnectedLinkMgr *linkMgr, InsAlgTemplate0 &tempAlgIntra, InsAlgTemplate1 &tempAlgInter)
283 : {
284 0 : AlgTempResReq resReqIntra;
285 0 : AlgTempResReq resReqInter;
286 0 : if (enableDetour_) {
287 0 : HCCL_DEBUG("[%s] Rank[%d], with detouring enabled", __func__, myRank_);
288 0 : CHK_RET(tempAlgIntra.CalcResDetour(linkMgr, resReqIntra));
289 : } else {
290 0 : HCCL_DEBUG("[%s] Rank[%d], with detouring disabled", __func__, myRank_);
291 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
292 : }
293 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
294 0 : u32 intraQueNum = resReqIntra.queNum;
295 0 : u32 interQueNum = resReqInter.queNum;
296 : // 申请算法模板所需资源
297 0 : if(!(intraQueNum > 0 && interQueNum > 0)) {
298 0 : HCCL_ERROR("[InsBroadcastParallelExecutor]resReqIntra.queNum and resReqInter.queNum must larger than 0.");
299 0 : return HcclResult::HCCL_E_INTERNAL;
300 : }
301 0 : u32 totalQueueNum = intraQueNum + interQueNum;
302 0 : CHK_RET(InitQueue(totalQueueNum, requiredQue_));
303 0 : for(u32 i = 0 ; i < requiredQue_.size(); i++) {
304 0 : if (i < intraQueNum) {
305 0 : intraQue_.push_back(requiredQue_.at(i));
306 : } else {
307 0 : interQue_.push_back(requiredQue_.at(i));
308 : }
309 : }
310 : // 每个算法的第0条流用于同步
311 0 : syncQueues_.emplace_back(intraQue_.at(0));
312 0 : syncQueues_.emplace_back(interQue_.at(0));
313 :
314 0 : CHK_RET(PrepResLinks(myRank_, resReqIntra.links, linkMgr, intraLinks_));
315 0 : CHK_RET(PrepResLinks(myRank_, resReqInter.links, linkMgr, interLinks_));
316 0 : HCCL_INFO("[InsBroadcastParallelExecutor] intraLinks size[%zu], interLinks size[%zu]",
317 : intraLinks_.size(), interLinks_.size());
318 :
319 0 : return HcclResult::HCCL_SUCCESS;
320 0 : }
321 :
322 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
323 0 : void InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenDataParams(
324 : const u64 dataOffset, const u64 sliceCount, const u64 scratchOffsetCount,
325 : TemplateDataParams &dataParams) const
326 : {
327 0 : dataParams.buffInfo.inBuffType = BufferType::INPUT;
328 0 : dataParams.buffInfo.outBuffType = BufferType::INPUT;
329 0 : dataParams.buffInfo.scratBuffType = BufferType::SCRATCH;
330 0 : dataParams.buffInfo.inBuffBaseOff = dataOffset;
331 0 : dataParams.buffInfo.outBuffBaseOff = dataOffset;
332 0 : dataParams.buffInfo.scratchBuffBaseOff = scratchOffsetCount * dataTypeSize_;
333 0 : dataParams.sliceSize = sliceCount * dataTypeSize_;
334 :
335 0 : dataParams.inputSliceStride = 0;
336 0 : dataParams.outputSliceStride = 0;
337 0 : dataParams.repeatNum = 1;
338 0 : dataParams.inputRepeatStride = 0;
339 0 : dataParams.outputRepeatStride = 0;
340 0 : return;
341 : }
342 :
343 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
344 0 : HcclResult InsBroadcastParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenInsQues(
345 : InsAlgTemplate0 &tempAlgIntra, InsAlgTemplate1 &tempAlgInter)
346 : {
347 0 : HCCL_INFO("[InsBroadcastParallelExecutor] AlgTemplate intra server is [%s]", tempAlgIntra.Describe().c_str());
348 0 : HCCL_INFO("[InsBroadcastParallelExecutor] AlgTemplate inter server is [%s]", tempAlgInter.Describe().c_str());
349 :
350 0 : std::vector<float> dataSplitSize;
351 0 : GetParallelDataSplit(dataSplitSize);
352 :
353 0 : u32 multipleIntra = tempAlgIntra.CalcScratchMultiple(BufferType::INPUT, BufferType::INPUT);
354 0 : u32 multipleInter = tempAlgInter.CalcScratchMultiple(BufferType::INPUT, BufferType::INPUT);
355 :
356 : // 按照intraData0+interData1，以及intraData1+interData0两种方式分别计算，取multiple最大需求
357 0 : float multiple0 = dataSplitSize.at(0) * float(multipleIntra) + dataSplitSize.at(1) * float(multipleInter);
358 0 : float multiple1 = dataSplitSize.at(1) * float(multipleIntra) + dataSplitSize.at(0) * float(multipleInter);
359 0 : float multiple = std::max(multiple0, multiple1);
360 :
361 : // 数据切分
362 0 : u64 sliceCount = std::min(static_cast<u64>(UB_MAX_DATA_SIZE) / dataTypeSize_, dataCount_);
363 0 : if (multiple > 0 && maxTmpMemSize_ > 0) {
364 0 : u64 scratchCount = maxTmpMemSize_ / dataTypeSize_; // 按照count来切分
365 0 : sliceCount = static_cast<u64>(float(scratchCount) / multiple); // 向下取整，防止Scratch溢出
366 : }
367 0 : u64 sliceCountPart0 = static_cast<u64>(float(sliceCount) * dataSplitSize.at(0));
368 0 : u64 sliceCountPart1 = sliceCount - sliceCountPart0;
369 :
370 0 : if(sliceCount == 0){
371 0 : HCCL_WARNING("The divisor cannot be zero.");
372 0 : return HcclResult::HCCL_SUCCESS;
373 : }
374 : // 计算循环次数
375 0 : u32 loopTimes = dataCount_ / sliceCount + ((dataCount_ % sliceCount == 0) ? 0 : 1);
376 : // 计算尾块
377 0 : u64 finalSliceCount = dataCount_ - (loopTimes - 1) * sliceCount;
378 0 : u64 finalSliceCountPart0 = static_cast<u64>(float(finalSliceCount) * dataSplitSize.at(0));
379 0 : u64 finalSliceCountPart1 = finalSliceCount - finalSliceCountPart0;
380 : // 计算Scratch偏移，数据尾块必然小于常规块，不用额外计算尾块时的Scratch偏移
381 0 : u64 scratchOffsetCountIntraStage0 = 0;
382 0 : u64 scratchOffsetCountInterStage0 = sliceCountPart0 * multipleIntra;
383 0 : u64 scratchOffsetCountInterStage1 = 0;
384 0 : u64 scratchOffsetCountIntraStage1 = sliceCountPart0 * multipleInter;
385 :
386 0 : TemplateDataParams tempAlgParamsIntra0;
387 0 : TemplateDataParams tempAlgParamsInter0;
388 0 : TemplateDataParams tempAlgParamsInter1;
389 0 : TemplateDataParams tempAlgParamsIntra1;
390 0 : TempFuncs tempFuncs;
391 0 : tempFuncs.opMode = opMode_;
392 0 : tempFuncs.enableCounterNotify = false;
393 :
394 0 : for (u32 loopIndex = 0; loopIndex < loopTimes; loopIndex++) {
395 0 : u64 currCountPart0 = (loopIndex == loopTimes - 1) ? finalSliceCountPart0 : sliceCountPart0;
396 0 : u64 currCountPart1 = (loopIndex == loopTimes - 1) ? finalSliceCountPart1 : sliceCountPart1;
397 0 : u64 dataOffset0 = loopIndex * sliceCount * dataTypeSize_;
398 0 : u64 dataOffset1 = dataOffset0 + currCountPart0 * dataTypeSize_;
399 :
400 : // 第一步开始前同步
401 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
402 : // server内topo包含root_的rank进行展开，其它rank不展开
403 0 : if (intraLocalRoot_ == root_ && currCountPart0 > 0) {
404 : //数据0的server内的mesh算法
405 0 : GenDataParams(dataOffset0, currCountPart0, scratchOffsetCountIntraStage0, tempAlgParamsIntra0);
406 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra0, intraLinks_, intraQue_));
407 : }
408 : // server间topo包含root_的rank进行展开，其它rank不展开
409 0 : if (interLocalRoot_ == root_ && currCountPart1 > 0) {
410 : //数据1的server间的nhr算法
411 0 : GenDataParams(dataOffset1, currCountPart1, scratchOffsetCountInterStage0, tempAlgParamsInter1);
412 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter1, interLinks_, interQue_));
413 : }
414 : // 第一步做完后回到主流做尾同步
415 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
416 :
417 : // 第二步开始前同步
418 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
419 0 : if (currCountPart0 > 0) {
420 : // 数据0的server间的nhr算法
421 0 : GenDataParams(dataOffset0, currCountPart0, scratchOffsetCountInterStage1, tempAlgParamsInter0);
422 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter0, interLinks_, interQue_));
423 : }
424 0 : if (currCountPart1 > 0) {
425 : // 数据1的server内的mesh算法
426 0 : GenDataParams(dataOffset1, currCountPart1, scratchOffsetCountIntraStage1, tempAlgParamsIntra1);
427 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra1, intraLinks_, intraQue_));
428 : }
429 : // 尾同步
430 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
431 : }
432 :
433 0 : return HcclResult::HCCL_SUCCESS;
434 0 : }
435 :
436 : // 算法注册
437 : INS_REGISTER_IMPL_BY_TWO_TEMPS(OpType::BROADCAST, InsBroadcastParallelMesh1DNHR, InsBroadcastParallelExecutor,
438 : TopoMatchMeshNHR, InsTempBroadcastMesh1DTwoShot, InsTempBroadcastNHR);
439 : INS_REGISTER_IMPL_BY_TWO_TEMPS(OpType::BROADCAST, InsBroadcastParallelMesh1DNHRPcie, InsBroadcastParallelExecutor,
440 : TopoMatchMeshNHRPcie, InsTempBroadcastMesh1DTwoShot, InsTempBroadcastNHR);
441 :
442 : #ifndef CCL_KERNEL_AICPU
443 : INS_REGISTER_IMPL_BY_TWO_TEMPS(OpType::BROADCAST, CcuBroadcastParallelMesh1DNHR, InsBroadcastParallelExecutor,
444 : TopoMatchMeshNHR, CcuTempBroadcastMesh1DMem2Mem, CcuTempBroadcastNHRMem2Mem1D);
445 : #endif
446 : } // namespace Hccl
|
