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