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_parallel_executor.h"
12 : #include <cmath>
13 : #include "log.h"
14 : #include "ins_coll_alg_registry.h"
15 : #include "topo_match_mesh_nhr.h"
16 : #include "topo_match_mesh_nhr_pcie.h"
17 : #include "alg_data_trans_wrapper.h"
18 : #include "ins_temp_reduce_nhr.h"
19 : #include "ins_temp_reduce_mesh_1D.h"
20 : #include "ccu_temp_reduce_nhr_1D_mem2mem.h"
21 : #include "ccu_temp_reduce_mesh_1D_mem2mem.h"
22 :
23 : namespace Hccl {
24 : constexpr u64 MAX_OFFLOAD_SCRATCH_SIZE = 200 * 1024 * 1024; // 200M
25 :
26 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
27 0 : InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::InsReduceParallelExecutor()
28 0 : : InsCollAlgBase()
29 0 : {}
30 :
31 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
32 0 : InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::~InsReduceParallelExecutor()
33 0 : {}
34 :
35 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
36 0 : HcclResult InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcResOffload(
37 : const RankGraph* rankGraph, const u64& dataSize, CollOffloadOpResReq& resReq)
38 : {
39 0 : HCCL_INFO("[InsReduceParallelExecutor] CalcResOffload begins.");
40 : (void)dataSize;
41 0 : u64 scratchMemSize = MAX_OFFLOAD_SCRATCH_SIZE;
42 0 : resReq.requiredScratchMemSize = scratchMemSize; // 200MB
43 : // Topo Match
44 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
45 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
46 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, intraLocalRankSize_, interLocalRankSize_));
47 0 : InsAlgTemplate0 intraTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
48 0 : InsAlgTemplate1 interTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
49 :
50 : // calculate required insQues and prepare queue
51 0 : AlgTempResReq resReqIntra;
52 0 : AlgTempResReq resReqInter;
53 0 : if (enableDetour_) {
54 0 : HCCL_DEBUG("[InsReduceParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
55 0 : CHK_RET(intraTempAlg.CalcResDetour(rankGraph, resReqIntra));
56 : } else {
57 0 : HCCL_DEBUG("[InsReduceParallelExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
58 0 : CHK_RET(intraTempAlg.CalcRes(resReqIntra));
59 : }
60 :
61 0 : CHK_RET(interTempAlg.CalcRes(resReqInter));
62 0 : resReq.requiredSubQueNum = resReqIntra.streamNum + resReqInter.streamNum - 1;
63 0 : return HcclResult::HCCL_SUCCESS;
64 0 : }
65 :
66 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
67 0 : HcclResult InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcRes(
68 : const RankGraph* rankGraph, CollAlgResReq& algResReq)
69 : {
70 0 : HCCL_INFO("[InsReduceParallelExecutor] CalcRes begins.");
71 : // Topo Match
72 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
73 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
74 0 : algResReq.topoInfo.UpdateMultiLevelTopo(virtRanks_, virtRankMap_, vTopo_);
75 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, intraLocalRankSize_, interLocalRankSize_));
76 :
77 : // instantiate a template
78 0 : InsAlgTemplate0 intraTempAlg(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]);
79 0 : InsAlgTemplate1 interTempAlg(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]);
80 :
81 : // calculate required insQues and prepare queue
82 0 : AlgTempResReq resReqIntra;
83 0 : AlgTempResReq resReqInter;
84 0 : if (enableDetour_) {
85 0 : HCCL_DEBUG("[InsReduceParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
86 0 : CHK_RET(intraTempAlg.CalcResDetour(rankGraph, resReqIntra));
87 : } else {
88 0 : HCCL_DEBUG("[InsReduceParallelExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
89 0 : CHK_RET(intraTempAlg.CalcRes(resReqIntra));
90 : }
91 0 : CHK_RET(interTempAlg.CalcRes(resReqInter));
92 :
93 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqIntra.links, algResReq.levelRankPairs));
94 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqInter.links, algResReq.levelRankPairs));
95 0 : algResReq.primQueueNum = resReqIntra.streamNum + resReqInter.streamNum;
96 0 : std::vector<std::tuple<QId, QId, u32>> notifyRequests;
97 :
98 0 : u32 slaveNum = algResReq.primQueueNum - 1;
99 0 : notifyRequests.reserve(slaveNum); // 每个从流需要1个
100 0 : for (QId q = 1; q < algResReq.primQueueNum; q++) {
101 0 : notifyRequests.emplace_back(std::make_tuple(0, q, 0));
102 0 : notifyRequests.emplace_back(std::make_tuple(q, 0, 0));
103 : }
104 :
105 : // nhr算法只有一个stream
106 0 : for (QId q = resReqIntra.streamNum; q < algResReq.primQueueNum; q++) {
107 0 : if (resReqIntra.streamNum == q) {
108 0 : continue;
109 : }
110 0 : notifyRequests.emplace_back(std::make_tuple(resReqIntra.streamNum, q, 0));
111 0 : notifyRequests.emplace_back(std::make_tuple(q, resReqIntra.streamNum, 0));
112 0 : HCCL_DEBUG(
113 : "[InsReduceParallelExecutor] CalcRes notifyRequests:%u->%u. %u->%u", resReqIntra.streamNum, q, q,
114 : resReqIntra.streamNum);
115 : }
116 :
117 0 : algResReq.queueNotifys = notifyRequests;
118 0 : HCCL_DEBUG("[InsReduceParallelExecutor] algResReq.primQueueNum %u", algResReq.primQueueNum);
119 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, algResReq.links));
120 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, algResReq.links));
121 :
122 0 : return HcclResult::HCCL_SUCCESS;
123 0 : }
124 :
125 : // HOST 侧算法入口，将对应的 instruction 添加到指令队列中
126 : // 传入的insQue为一条主流
127 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
128 0 : void InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenTemplateAlgParams0(
129 : const u64 dataOffset, const u64 dataCount, const u64 scratchOffset, TemplateDataParams& tempAlgParams) const
130 : {
131 0 : tempAlgParams.buffInfo.inBuffType = BufferType::INPUT;
132 0 : tempAlgParams.buffInfo.outBuffType = BufferType::OUTPUT;
133 0 : tempAlgParams.buffInfo.scratBuffType = BufferType::SCRATCH;
134 0 : tempAlgParams.buffInfo.inBuffBaseOff = dataOffset;
135 0 : tempAlgParams.buffInfo.outBuffBaseOff = dataOffset;
136 0 : tempAlgParams.buffInfo.scratchBuffBaseOff = scratchOffset;
137 0 : tempAlgParams.sliceSize = dataCount * dataTypeSize_;
138 0 : tempAlgParams.tailSize = tempAlgParams.sliceSize;
139 0 : tempAlgParams.inputSliceStride = 0; // 输入数据仅有 1 个 slice, 不需要 stride
140 0 : tempAlgParams.outputSliceStride = 0;
141 0 : tempAlgParams.repeatNum = 1;
142 0 : return;
143 : }
144 :
145 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
146 0 : void InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenTemplateAlgParams1(
147 : const u64 dataOffset, const u64 dataCount, const u64 scratchOffset, TemplateDataParams& tempAlgParams) const
148 : {
149 0 : tempAlgParams.buffInfo.inBuffType = BufferType::OUTPUT;
150 0 : tempAlgParams.buffInfo.outBuffType = BufferType::OUTPUT;
151 0 : tempAlgParams.buffInfo.scratBuffType = BufferType::SCRATCH;
152 0 : tempAlgParams.buffInfo.inBuffBaseOff = dataOffset;
153 0 : tempAlgParams.buffInfo.outBuffBaseOff = dataOffset;
154 0 : tempAlgParams.buffInfo.scratchBuffBaseOff = scratchOffset;
155 0 : tempAlgParams.sliceSize = dataCount * dataTypeSize_;
156 0 : tempAlgParams.tailSize = tempAlgParams.sliceSize;
157 0 : tempAlgParams.inputSliceStride = 0; // 输入数据仅有 1 个 slice, 不需要 stride
158 0 : tempAlgParams.outputSliceStride = 0;
159 0 : tempAlgParams.repeatNum = 1;
160 0 : return;
161 : }
162 :
163 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
164 0 : void InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GetParallelDataSplitRate(
165 : std::vector<float>& splitDataSize) const
166 : {
167 : // 先做等分，后续根据性能做调整
168 0 : double splitData = 0.5;
169 0 : splitDataSize.push_back(static_cast<float>(splitData));
170 0 : splitDataSize.push_back(static_cast<float>(splitData));
171 0 : return;
172 : }
173 :
174 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
175 0 : HcclResult InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcLocalRoot()
176 : {
177 0 : CHK_PRT_RET(
178 : root_ >= rankSize_, HCCL_ERROR("[CalcLocalRoot] root[%u] is out of rankSize[%u]", root_, rankSize_),
179 : HcclResult::HCCL_E_INTERNAL);
180 :
181 0 : u32 intraLocalRootIdx = root_ % intraLocalRankSize_;
182 0 : intraLocalRoot_ = static_cast<u32>(vTopo_.at(0).at(0).at(intraLocalRootIdx));
183 0 : u32 interLocalRootIdx = root_ / intraLocalRankSize_;
184 0 : interLocalRoot_ = static_cast<u32>(vTopo_.at(1).at(0).at(interLocalRootIdx));
185 :
186 0 : HCCL_INFO(
187 : "[CalcLocalRoot] localRoot: myRank[%d] intraLocalRoot[%u] interLocalRoot[%u]", myRank_, intraLocalRoot_,
188 : interLocalRoot_);
189 0 : return HcclResult::HCCL_SUCCESS;
190 : }
191 :
192 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
193 0 : HcclResult InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
194 : const RankGraph* rankGraph, InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
195 : {
196 0 : AlgTempResReq resReqIntra;
197 0 : AlgTempResReq resReqInter;
198 0 : if (enableDetour_) {
199 0 : HCCL_DEBUG("[%s] Rank[%d], detouring enabled.", __func__, myRank_);
200 0 : CHK_RET(tempAlgIntra.CalcResDetour(rankGraph, resReqIntra));
201 : } else {
202 0 : HCCL_DEBUG("[%s] Rank[%d], detouring disabled.", __func__, myRank_);
203 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
204 : }
205 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
206 :
207 : // 申请算法模板所需资源
208 0 : if (!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
209 0 : HCCL_ERROR("[InsReduceParallelExecutor]resReqIntra.queNum and resReqInter.queNum must larger than 0.");
210 0 : return HcclResult::HCCL_E_INTERNAL;
211 : }
212 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
213 0 : CHK_RET(InitQueue(totalQueueNum, reqQue_));
214 0 : for (u32 i = 0; i < reqQue_.size(); i++) {
215 0 : if (i < resReqIntra.queNum) {
216 0 : intraQue_.push_back(reqQue_[i]);
217 : } else {
218 0 : interQue_.push_back(reqQue_[i]);
219 : }
220 : }
221 0 : syncQueues_.emplace_back(intraQue_[0]);
222 0 : syncQueues_.emplace_back(interQue_[0]);
223 :
224 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, intraLinks_));
225 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, interLinks_));
226 0 : HCCL_INFO(
227 : "[InsReduceParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]", intraLinks_.size(),
228 : interLinks_.size());
229 0 : return HCCL_SUCCESS;
230 0 : }
231 :
232 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
233 0 : HcclResult InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
234 : ConnectedLinkMgr* linkMgr, InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
235 : {
236 0 : AlgTempResReq resReqIntra;
237 0 : AlgTempResReq resReqInter;
238 0 : if (enableDetour_) {
239 0 : HCCL_DEBUG("[%s] Rank[%d], CalcRes with detour enabled", __func__, myRank_);
240 0 : CHK_RET(tempAlgIntra.CalcResDetour(linkMgr, resReqIntra));
241 : } else {
242 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
243 : }
244 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
245 :
246 : // 申请算法模板所需资源
247 0 : if (!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
248 0 : HCCL_ERROR("[InsReduceParallelExecutor]resReqIntra.queNum and resReqInter.queNum must > 0.");
249 0 : return HcclResult::HCCL_E_INTERNAL;
250 : }
251 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
252 0 : CHK_RET(InitQueue(totalQueueNum, reqQue_));
253 0 : for (u32 i = 0; i < reqQue_.size(); i++) {
254 0 : if (i < resReqIntra.queNum) {
255 0 : intraQue_.push_back(reqQue_[i]);
256 : } else {
257 0 : interQue_.push_back(reqQue_[i]);
258 : }
259 : }
260 0 : syncQueues_.emplace_back(intraQue_[0]);
261 0 : syncQueues_.emplace_back(interQue_[0]);
262 :
263 0 : CHK_RET(PrepResLinks(myRank_, resReqIntra.links, linkMgr, intraLinks_));
264 0 : CHK_RET(PrepResLinks(myRank_, resReqInter.links, linkMgr, interLinks_));
265 0 : HCCL_INFO(
266 : "[InsReduceParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]", intraLinks_.size(),
267 : interLinks_.size());
268 0 : return HCCL_SUCCESS;
269 0 : }
270 :
271 : // Aicpu展开
272 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
273 0 : HcclResult InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
274 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
275 : InsQuePtr insQue)
276 : {
277 0 : HCCL_INFO("[InsReduceParallelExecutor] AICPU Orchestrate begins.");
278 : // init and check params
279 0 : CHK_RET(Init(op, params, insQue));
280 : // 所以获取取级通信域的信息
281 0 : vTopo_ = topoInfo.vTopo; // 本通信域内的通信平面
282 0 : virtRankMap_ = topoInfo.virtRankMap; // 本通信域内的 rank 映射表
283 0 : virtRanks_ = topoInfo.virtRanks; // 本通信域内的 rank 集合
284 :
285 : // 计算localRankSize和localRoot
286 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, intraLocalRankSize_, interLocalRankSize_));
287 0 : CHK_RET(CalcLocalRoot());
288 :
289 : // 实例化算法模板类
290 0 : InsAlgTemplate0 tempAlgIntra(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
291 0 : InsAlgTemplate1 tempAlgInter(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
292 :
293 0 : tempAlgIntra.SetDmaMode(dmaMode_);
294 0 : tempAlgIntra.InitReduceInfo(redOp_, dataType_);
295 0 : tempAlgIntra.SetRoot(intraLocalRoot_);
296 0 : tempAlgIntra.SetCollOp(op);
297 :
298 0 : tempAlgInter.SetDmaMode(dmaMode_);
299 0 : tempAlgInter.InitReduceInfo(redOp_, dataType_);
300 0 : tempAlgInter.SetRoot(interLocalRoot_);
301 0 : tempAlgInter.SetCollOp(op);
302 :
303 : // 计算算法模板所需资源
304 0 : CHK_RET(PrepareResForTemplate(linkMgr, tempAlgIntra, tempAlgInter));
305 0 : CHK_RET(GenInsQues(tempAlgIntra, tempAlgInter));
306 0 : HCCL_INFO("[InsReduceParallelExecutor] AICPU Orchestrate success.");
307 0 : return HcclResult::HCCL_SUCCESS;
308 0 : }
309 :
310 : // Host展开
311 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
312 0 : HcclResult InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
313 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
314 : {
315 0 : HCCL_INFO("[InsReduceParallelExecutor] Host Orchestrate begins.");
316 : // init and check params
317 0 : CHK_RET(Init(op, params, insQue));
318 :
319 : // Topo Match
320 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
321 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
322 :
323 : // 计算localRankSize和localRoot
324 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, intraLocalRankSize_, interLocalRankSize_));
325 0 : CHK_RET(CalcLocalRoot());
326 :
327 : // 实例化算法模板类
328 0 : InsAlgTemplate0 tempAlgIntra(myRank_, intraLocalRankSize_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
329 0 : InsAlgTemplate1 tempAlgInter(myRank_, interLocalRankSize_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
330 :
331 0 : tempAlgIntra.SetDmaMode(dmaMode_);
332 0 : tempAlgIntra.SetCollOp(op);
333 0 : tempAlgIntra.InitReduceInfo(redOp_, dataType_);
334 0 : tempAlgIntra.SetRoot(intraLocalRoot_);
335 :
336 0 : tempAlgInter.SetDmaMode(dmaMode_);
337 0 : tempAlgInter.InitReduceInfo(redOp_, dataType_);
338 0 : tempAlgInter.SetCollOp(op);
339 0 : tempAlgInter.SetRoot(interLocalRoot_);
340 :
341 : // 计算算法模板所需资源
342 0 : CHK_RET(PrepareResForTemplate(rankGraph, tempAlgIntra, tempAlgInter));
343 0 : CHK_RET(GenInsQues(tempAlgIntra, tempAlgInter));
344 0 : HCCL_INFO("[InsReduceParallelExecutor] Host Orchestrate success.");
345 0 : return HcclResult::HCCL_SUCCESS;
346 0 : }
347 :
348 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
349 0 : HcclResult InsReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenInsQues(
350 : InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
351 : {
352 0 : std::vector<float> dataSplitSize;
353 0 : GetParallelDataSplitRate(dataSplitSize);
354 0 : u64 alignedSize = 16 * 1024; // 假设需要16K对齐
355 0 : BufferType inBuffType = BufferType::INPUT;
356 0 : BufferType outBuffType = BufferType::OUTPUT;
357 0 : u32 intraScatchteMultipleStage0 = tempAlgIntra.CalcScratchMultiple(inBuffType, outBuffType);
358 0 : u32 interScatchteMultipleStage0 = tempAlgInter.CalcScratchMultiple(inBuffType, outBuffType);
359 0 : u32 intraScatchteMultipleStage1 = tempAlgIntra.CalcScratchMultiple(outBuffType, outBuffType);
360 0 : u32 interScatchteMultipleStage1 = tempAlgInter.CalcScratchMultiple(outBuffType, outBuffType);
361 0 : u32 scratchMultipleIntra = static_cast<u32>(std::max(
362 0 : std::ceil(dataSplitSize[0] * intraScatchteMultipleStage0),
363 0 : std::ceil(dataSplitSize[1] * intraScatchteMultipleStage1)));
364 0 : u32 scratchMultipleInter = static_cast<u32>(std::max(
365 0 : std::ceil(dataSplitSize[1] * interScatchteMultipleStage0),
366 0 : std::ceil(dataSplitSize[0] * interScatchteMultipleStage1)));
367 0 : u32 totalScratchMultiple = scratchMultipleIntra + scratchMultipleInter;
368 0 : u64 scratchMemBlockSize = maxTmpMemSize_;
369 0 : if (totalScratchMultiple > 0) {
370 0 : scratchMemBlockSize = (maxTmpMemSize_ / alignedSize / totalScratchMultiple) * alignedSize;
371 : }
372 0 : u64 intraScratchOffset = 0;
373 0 : u64 interScratchOffset = scratchMultipleIntra * scratchMemBlockSize;
374 :
375 : // dataSplitSize为分数，这里maxCountPerLoop对10取整
376 0 : u64 maxCountPerLoop
377 0 : = (std::min(static_cast<u64>(scratchMemBlockSize), static_cast<u64>(UB_MAX_DATA_SIZE)) / dataTypeSize_ / 10)
378 : * 10;
379 :
380 0 : u32 loopTimes = dataCount_ / maxCountPerLoop + ((dataCount_ % maxCountPerLoop == 0) ? 0 : 1);
381 :
382 0 : TemplateDataParams tempAlgParamsIntra0, tempAlgParamsInter0;
383 0 : TemplateDataParams tempAlgParamsInter1, tempAlgParamsIntra1;
384 0 : TempFuncs tempFuncs;
385 0 : tempFuncs.opMode = opMode_;
386 0 : tempFuncs.enableCounterNotify = false;
387 0 : tempFuncs.isBottom = true;
388 0 : tempFuncs.isForepart = true;
389 0 : for (u32 loopIndex = 0; loopIndex < loopTimes; loopIndex++) {
390 0 : u64 currCount = (loopIndex == loopTimes - 1) ? (dataCount_ - loopIndex * maxCountPerLoop) : maxCountPerLoop;
391 0 : u64 dataCountPerLoopAixs0 = static_cast<u64>(dataSplitSize[0] * currCount);
392 0 : u64 dataCountPerLoopAixs1 = currCount - dataCountPerLoopAixs0;
393 : // 第一步开始前同步
394 :
395 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
396 0 : u64 dataOffset0 = loopIndex * maxCountPerLoop * dataTypeSize_;
397 0 : u64 dataOffset1 = dataOffset0 + dataCountPerLoopAixs0 * dataTypeSize_;
398 : // 数据0的server内的mesh算法
399 0 : GenTemplateAlgParams0(dataOffset0, dataCountPerLoopAixs0, intraScratchOffset, tempAlgParamsIntra0);
400 : // 把每个template需要的queue传进去，比如stars的mesh要传多条queue
401 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra0, intraLinks_, intraQue_));
402 : // 数据1的server间的nhr算法
403 0 : GenTemplateAlgParams0(dataOffset1, dataCountPerLoopAixs1, interScratchOffset, tempAlgParamsInter1);
404 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter1, interLinks_, interQue_));
405 : // 第一步做完后回到主流做尾同步
406 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
407 : // 只有真正root节点的横纵坐标所在的卡，需要做第二步骤，担任过其中一个root节点的，只需要负责发就行了
408 0 : if ((static_cast<u32>(myRank_) != intraLocalRoot_) && (static_cast<u32>(myRank_) != interLocalRoot_)) {
409 0 : continue;
410 : }
411 :
412 : // 第二步开始前同步
413 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
414 0 : if (static_cast<u32>(myRank_) == intraLocalRoot_) {
415 : // 数据0的server间的nhr算法
416 0 : GenTemplateAlgParams1(dataOffset0, dataCountPerLoopAixs0, interScratchOffset, tempAlgParamsInter0);
417 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter0, interLinks_, interQue_));
418 : }
419 0 : if (static_cast<u32>(myRank_) == interLocalRoot_) {
420 : // 数据1的server内的mesh算法
421 0 : GenTemplateAlgParams1(dataOffset1, dataCountPerLoopAixs1, intraScratchOffset, tempAlgParamsIntra1);
422 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra1, intraLinks_, intraQue_));
423 : }
424 : // 尾同步
425 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
426 : }
427 0 : return HcclResult::HCCL_SUCCESS;
428 0 : }
429 :
430 : // 算法注册
431 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
432 : OpType::REDUCE, InsReduceParallelMesh1DNHR, InsReduceParallelExecutor, TopoMatchMeshNHR, InsTempReduceMesh1D,
433 : InsTempReduceNHR);
434 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
435 : OpType::REDUCE, InsReduceParallelMesh1DNHRPcie, InsReduceParallelExecutor, TopoMatchMeshNHRPcie,
436 : InsTempReduceMesh1D, InsTempReduceNHR);
437 : #ifndef CCL_KERNEL_AICPU
438 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
439 : OpType::REDUCE, CcuReduceParallelMesh1DNHR, InsReduceParallelExecutor, TopoMatchMeshNHR, CcuTempReduceMeshMem2Mem1D,
440 : CcuTempReduceNHRMem2Mem1D);
441 : #endif
442 : } // namespace Hccl
|
