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_all_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_concurr_mesh_nhr.h"
17 : #include "topo_match_mesh_nhr_pcie.h"
18 : #include "alg_data_trans_wrapper.h"
19 : #include "ins_temp_all_reduce_nhr.h"
20 : #include "ins_temp_all_reduce_mesh_1D_two_shot.h"
21 : #include "ins_temp_all_reduce_mesh_2D_two_shot.h"
22 : #include "ccu_temp_all_reduce_nhr_1D_mem2mem.h"
23 : #include "ccu_temp_all_reduce_mesh_1D_mem2mem.h"
24 :
25 : namespace Hccl {
26 : constexpr u64 MAX_OFFLOAD_SCRATCH_SIZE = 200 * 1024 * 1024; // 200M
27 :
28 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
29 0 : InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::InsAllReduceParallelExecutor()
30 0 : : InsCollAlgBase()
31 0 : {}
32 :
33 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
34 0 : InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::~InsAllReduceParallelExecutor()
35 0 : {}
36 :
37 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
38 0 : HcclResult InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcResOffload(
39 : const RankGraph* rankGraph, const u64& dataSize, CollOffloadOpResReq& resReq)
40 : {
41 0 : HCCL_INFO("[InsAllReduceParallelExecutor] CalcResOffload begins.");
42 : (void)dataSize;
43 0 : uint64_t tempSize = 2;
44 0 : u64 scratchMemSize = MAX_OFFLOAD_SCRATCH_SIZE;
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("[InsV2AllGatherSoleExecutor] 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("[InsAllReduceParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
64 0 : CHK_RET(intraTempAlg.CalcResDetour(rankGraph, resReqIntra));
65 : } else {
66 0 : HCCL_DEBUG("[InsAllReduceParallelExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
67 0 : CHK_RET(intraTempAlg.CalcRes(resReqIntra));
68 : }
69 :
70 0 : CHK_RET(interTempAlg.CalcRes(resReqInter));
71 :
72 : // 算法从流数量 = Σ(temp的que数量 + temp的从流数量 * temp调用次数) - 算法主流数量
73 0 : resReq.requiredSubQueNum = resReqIntra.queNum + (resReqIntra.streamNum - resReqIntra.queNum) * tempSize
74 0 : + resReqInter.queNum + (resReqInter.streamNum - resReqInter.queNum) * tempSize - 1;
75 0 : HCCL_INFO("[InsAllReduceParallelExecutor::CalcResOffload]requiredSubQueNum = %llu", resReq.requiredSubQueNum);
76 :
77 0 : return HcclResult::HCCL_SUCCESS;
78 0 : }
79 :
80 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
81 0 : HcclResult InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcRes(
82 : const RankGraph* rankGraph, CollAlgResReq& algResReq)
83 : {
84 0 : HCCL_INFO("[InsAllReduceParallelExecutor] CalcRes begins.");
85 : // Topo Match
86 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
87 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
88 0 : algResReq.topoInfo.UpdateMultiLevelTopo(virtRanks_, virtRankMap_, vTopo_);
89 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
90 :
91 : // instantiate a template
92 0 : InsAlgTemplate0 intraTempAlg(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]);
93 0 : InsAlgTemplate1 interTempAlg(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]);
94 :
95 0 : std::vector<map<u32, u32>> rank2PathNumMap;
96 0 : HCCL_INFO("[InsAllReduceParallelExecutor] CalcResOffload SetPathNumMap");
97 0 : CHK_RET(SetPathNumMapByRankGraphMultiLevel(rankGraph, virtRanks_, myRank_, rank2PathNumMap));
98 0 : intraTempAlg.setPathNumMap(rank2PathNumMap[0]);
99 0 : interTempAlg.setPathNumMap(rank2PathNumMap[1]);
100 :
101 : // calculate required insQues and prepare queue
102 0 : AlgTempResReq resReqIntra;
103 0 : AlgTempResReq resReqInter;
104 0 : if (enableDetour_) {
105 0 : HCCL_DEBUG("[InsAllReduceParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
106 0 : CHK_RET(intraTempAlg.CalcResDetour(rankGraph, resReqIntra));
107 : } else {
108 0 : HCCL_DEBUG("[InsAllReduceParallelExecutor] 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 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqIntra.links, algResReq.levelRankPairs));
114 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, resReqInter.links, algResReq.levelRankPairs));
115 0 : algResReq.primQueueNum = resReqIntra.queNum + resReqInter.queNum;
116 0 : CHK_RET(CalcParallelNotifyReq(algResReq.primQueueNum, resReqIntra.queNum, algResReq.queueNotifys));
117 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, algResReq.links));
118 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, algResReq.links));
119 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, algResReq.links));
120 0 : return HcclResult::HCCL_SUCCESS;
121 0 : }
122 :
123 : // HOST 侧算法入口，将对应的 instruction 添加到指令队列中
124 : // 传入的insQue为一条主流
125 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
126 0 : void InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenTemplateAlgParams0(
127 : const u64 dataOffset, const u64 dataCount, const u64 scratchOffset, TemplateDataParams& tempAlgParams) const
128 : {
129 0 : tempAlgParams.buffInfo.inBuffType = BufferType::INPUT;
130 0 : tempAlgParams.buffInfo.outBuffType = BufferType::OUTPUT;
131 0 : tempAlgParams.buffInfo.scratBuffType = BufferType::SCRATCH;
132 0 : tempAlgParams.buffInfo.inBuffBaseOff = dataOffset;
133 0 : tempAlgParams.buffInfo.outBuffBaseOff = dataOffset;
134 0 : tempAlgParams.buffInfo.scratchBuffBaseOff = scratchOffset;
135 0 : tempAlgParams.sliceSize = dataCount * dataTypeSize_;
136 0 : tempAlgParams.tailSize = tempAlgParams.sliceSize;
137 0 : tempAlgParams.inputSliceStride = 0; // 输入数据仅有 1 个 slice, 不需要 stride
138 0 : tempAlgParams.outputSliceStride = 0;
139 :
140 0 : return;
141 : }
142 :
143 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
144 0 : void InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenTemplateAlgParams1(
145 : const u64 dataOffset, const u64 dataCount, const u64 scratchOffset, TemplateDataParams& tempAlgParams) const
146 : {
147 0 : tempAlgParams.buffInfo.inBuffType = BufferType::OUTPUT;
148 0 : tempAlgParams.buffInfo.outBuffType = BufferType::OUTPUT;
149 0 : tempAlgParams.buffInfo.scratBuffType = BufferType::SCRATCH;
150 0 : tempAlgParams.buffInfo.inBuffBaseOff = dataOffset;
151 0 : tempAlgParams.buffInfo.outBuffBaseOff = dataOffset;
152 0 : tempAlgParams.buffInfo.scratchBuffBaseOff = scratchOffset;
153 0 : tempAlgParams.sliceSize = dataCount * dataTypeSize_;
154 0 : tempAlgParams.tailSize = tempAlgParams.sliceSize;
155 0 : tempAlgParams.inputSliceStride = 0; // 输入数据仅有 1 个 slice, 不需要 stride
156 0 : tempAlgParams.outputSliceStride = 0;
157 0 : return;
158 : }
159 :
160 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
161 0 : void InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GetParallelDataSplitRate(
162 : std::vector<float>& splitDataSize) const
163 : {
164 : // to do 先做等分，后续根据性能做调整
165 0 : double splitData = 0.5;
166 0 : splitDataSize.push_back(splitData);
167 0 : splitDataSize.push_back(splitData);
168 0 : return;
169 : }
170 :
171 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
172 0 : HcclResult InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
173 : const RankGraph* rankGraph, InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
174 : {
175 0 : AlgTempResReq resReqInter;
176 0 : AlgTempResReq resReqIntra;
177 0 : if (enableDetour_) {
178 0 : HCCL_DEBUG("[%s] Rank[%d], detouring enabled.", __func__, myRank_);
179 0 : CHK_RET(tempAlgIntra.CalcResDetour(rankGraph, resReqIntra));
180 : } else {
181 0 : HCCL_DEBUG("[%s] Rank[%d], detouring disabled.", __func__, myRank_);
182 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
183 : }
184 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
185 :
186 : // 申请算法模板所需资源
187 0 : if (!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
188 0 : HCCL_ERROR("[InsAllReduceParallelExecutor]resReqIntra.queNum and resReqInter.queNum must larger than 0.");
189 0 : return HcclResult::HCCL_E_INTERNAL;
190 : }
191 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
192 0 : CHK_RET(InitQueue(totalQueueNum, requiredQue_));
193 0 : for (u32 qIdx = 0; qIdx < requiredQue_.size(); qIdx++) {
194 0 : if (qIdx < resReqIntra.queNum) {
195 0 : intraQue_.push_back(requiredQue_[qIdx]);
196 : } else {
197 0 : interQue_.push_back(requiredQue_[qIdx]);
198 : }
199 : }
200 0 : syncQueues_.emplace_back(intraQue_[0]);
201 0 : syncQueues_.emplace_back(interQue_[0]);
202 :
203 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqIntra.links, intraLinks_));
204 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, resReqInter.links, interLinks_));
205 0 : HCCL_INFO(
206 : "[InsAllReduceParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]", intraLinks_.size(),
207 : interLinks_.size());
208 0 : return HCCL_SUCCESS;
209 0 : }
210 :
211 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
212 0 : HcclResult InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::PrepareResForTemplate(
213 : ConnectedLinkMgr* linkMgr, InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
214 : {
215 0 : AlgTempResReq resReqIntra;
216 0 : AlgTempResReq resReqInter;
217 0 : if (enableDetour_) {
218 0 : HCCL_DEBUG("[InsAllReduceParallelExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
219 0 : CHK_RET(tempAlgIntra.CalcResDetour(linkMgr, resReqIntra));
220 : } else {
221 0 : HCCL_DEBUG("[InsAllReduceParallelExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
222 0 : CHK_RET(tempAlgIntra.CalcRes(resReqIntra));
223 : }
224 0 : CHK_RET(tempAlgInter.CalcRes(resReqInter));
225 :
226 : // 申请算法模板所需资源
227 0 : if (!(resReqIntra.queNum > 0 && resReqInter.queNum > 0)) {
228 0 : HCCL_ERROR("[InsAllReduceParallelExecutor] Intra queNum and Inter queNum must larger than 0.");
229 0 : return HcclResult::HCCL_E_INTERNAL;
230 : }
231 0 : u32 totalQueueNum = resReqIntra.queNum + resReqInter.queNum;
232 0 : CHK_RET(InitQueue(totalQueueNum, requiredQue_));
233 0 : for (u32 i = 0; i < requiredQue_.size(); i++) {
234 0 : if (i < resReqIntra.queNum) {
235 0 : intraQue_.push_back(requiredQue_[i]);
236 : } else {
237 0 : interQue_.push_back(requiredQue_[i]);
238 : }
239 : }
240 0 : syncQueues_.emplace_back(intraQue_[0]);
241 0 : syncQueues_.emplace_back(interQue_[0]);
242 :
243 0 : CHK_RET(PrepResLinks(myRank_, resReqIntra.links, linkMgr, intraLinks_));
244 0 : CHK_RET(PrepResLinks(myRank_, resReqInter.links, linkMgr, interLinks_));
245 0 : HCCL_INFO(
246 : "[InsAllReduceParallelExecutor] intraLinks_ size[%zu], interLinks_ size[%zu]", intraLinks_.size(),
247 : interLinks_.size());
248 0 : return HCCL_SUCCESS;
249 0 : }
250 :
251 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
252 0 : HcclResult InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::CalcSendDataSize(
253 : u64& memBlockSize, float& SplitRate, u32& multipleIntra, u32& multipleInter)
254 : {
255 0 : std::vector<float> dataSplitSize;
256 0 : GetParallelDataSplitRate(dataSplitSize);
257 0 : uint64_t templateNum = 2;
258 0 : if (multipleIntra == 0 && multipleInter == 0) {
259 0 : memBlockSize = UB_MAX_DATA_SIZE + UB_MAX_DATA_SIZE;
260 0 : } else if ((multipleIntra == 0 && multipleInter > 0) || (multipleInter == 0 && multipleIntra > 0)) {
261 : // 因为数据要交替在两个template中执行，因此最终要以数据处理量小的template为准
262 0 : if (multipleIntra > 0) {
263 0 : memBlockSize = std::min(static_cast<u64>(UB_MAX_DATA_SIZE), maxTmpMemSize_ / multipleIntra) * templateNum;
264 0 : Intra0ScratchSize = maxTmpMemSize_;
265 0 : Intra1ScratchSize = maxTmpMemSize_;
266 : } else {
267 0 : memBlockSize = std::min(static_cast<u64>(UB_MAX_DATA_SIZE), maxTmpMemSize_ / multipleInter) * templateNum;
268 0 : Inter0ScratchSize = maxTmpMemSize_;
269 0 : Inter1ScratchSize = maxTmpMemSize_;
270 : }
271 : } else { // multipleIntra >0 && multipleInter >0, 理论上dataSplitSize[0]=0.5时，scratch buffer利用率最大
272 0 : SplitRate = dataSplitSize[0];
273 0 : u32 subMultiple0 = static_cast<u32>(std::ceil(SplitRate * multipleIntra + (1 - SplitRate) * multipleInter));
274 0 : u32 subMultiple1 = static_cast<u32>(std::ceil((1 - SplitRate) * multipleIntra + SplitRate * multipleInter));
275 0 : u64 totalScratchMultiple = std::max(subMultiple0, subMultiple1);
276 0 : memBlockSize = std::min(static_cast<u64>(UB_MAX_DATA_SIZE), maxTmpMemSize_ / totalScratchMultiple);
277 :
278 0 : interScratchOffset0 = static_cast<u64>(memBlockSize * SplitRate * multipleIntra);
279 0 : interScratchOffset1 = static_cast<u64>(memBlockSize * (1 - SplitRate) * multipleIntra);
280 0 : Intra0ScratchSize = interScratchOffset0;
281 0 : Inter0ScratchSize = interScratchOffset1;
282 0 : Intra1ScratchSize = interScratchOffset1;
283 0 : Inter1ScratchSize = interScratchOffset0;
284 : }
285 0 : return HCCL_SUCCESS;
286 0 : }
287 :
288 : /*
289 : *@Desc: AICPU算法编排
290 : */
291 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
292 0 : HcclResult InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
293 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
294 : InsQuePtr insQue)
295 : {
296 0 : HCCL_INFO("[InsAllReduceParallelExecutor] AICPU Orchestrate begins.");
297 : // init and check params
298 0 : CHK_RET(Init(op, params, insQue));
299 : // 所以获取取级通信域的信息
300 0 : vTopo_ = topoInfo.vTopo; // 本通信域内的通信平面
301 0 : virtRankMap_ = topoInfo.virtRankMap; // 本通信域内的 rank 映射表
302 0 : virtRanks_ = topoInfo.virtRanks; // 本通信域内的 rank 集合
303 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
304 :
305 : // 实例化算法模板类
306 0 : InsAlgTemplate0 tempAlgIntra(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
307 0 : InsAlgTemplate1 tempAlgInter(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
308 :
309 0 : tempAlgInter.SetDmaMode(dmaMode_);
310 0 : tempAlgInter.InitReduceInfo(redOp_, dataType_);
311 0 : tempAlgInter.SetCollOp(op);
312 :
313 0 : tempAlgIntra.SetDmaMode(dmaMode_);
314 0 : tempAlgIntra.InitReduceInfo(redOp_, dataType_);
315 0 : tempAlgIntra.SetCollOp(op);
316 :
317 0 : std::vector<std::map<u32, u32>> rank2PathNumMap;
318 0 : SetPathNumMapByLinkMgrMultiLevel(linkMgr, virtRanks_, myRank_, rank2PathNumMap);
319 0 : tempAlgIntra.setPathNumMap(rank2PathNumMap[0]);
320 0 : tempAlgInter.setPathNumMap(rank2PathNumMap[1]);
321 :
322 : // 计算算法模板所需资源
323 0 : CHK_RET(PrepareResForTemplate(linkMgr, tempAlgIntra, tempAlgInter));
324 0 : CHK_RET(GenInsQues(tempAlgIntra, tempAlgInter));
325 0 : HCCL_INFO("[InsAllReduceParallelExecutor] Orchestrate success.");
326 :
327 0 : return HcclResult::HCCL_SUCCESS;
328 0 : }
329 :
330 : /*
331 : *@Desc: Host算法编排
332 : */
333 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
334 0 : HcclResult InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::Orchestrate(
335 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
336 : {
337 0 : HCCL_INFO("[InsAllReduceParallelExecutor] Host Orchestrate begins.");
338 : // init and check params
339 0 : CHK_RET(Init(op, params, insQue));
340 :
341 : // Topo Match
342 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
343 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
344 :
345 0 : CHK_RET(CalcLocalRankSize(myRank_, virtRanks_, rankSizeLevel0_, rankSizeLevel1_));
346 :
347 : // 实例化算法模板类
348 0 : InsAlgTemplate0 tempAlgIntra(myRank_, rankSizeLevel0_, vTopo_[0], virtRankMap_[0]); // server内算法，比如mesh
349 0 : InsAlgTemplate1 tempAlgInter(myRank_, rankSizeLevel1_, vTopo_[1], virtRankMap_[1]); // server间算法，比如nhr
350 :
351 0 : tempAlgIntra.InitReduceInfo(redOp_, dataType_);
352 0 : tempAlgIntra.SetDmaMode(dmaMode_);
353 0 : tempAlgIntra.SetCollOp(op);
354 :
355 0 : tempAlgInter.SetDmaMode(dmaMode_);
356 0 : tempAlgInter.SetCollOp(op); // CCU template需要传递op信息
357 0 : tempAlgInter.InitReduceInfo(redOp_, dataType_);
358 :
359 : // 计算算法模板所需资源
360 0 : CHK_RET(PrepareResForTemplate(rankGraph, tempAlgIntra, tempAlgInter));
361 :
362 0 : CHK_RET(GenInsQues(tempAlgIntra, tempAlgInter));
363 0 : HCCL_INFO("[InsAllReduceParallelExecutor] Orchestrate success.");
364 :
365 0 : return HcclResult::HCCL_SUCCESS;
366 0 : }
367 :
368 : /*
369 : @Desc: 本方法主要实现的是跨框算法实现，如下图，框内和框间分别用不同的算法实现
370 : /-------------------\ /-------------------\
371 : | /----\ /----\ | | /----\ /----\ |
372 : | |card| |card| | | |card| |card| |
373 : | \----/ \----/ | | \----/ \----/ |
374 : | | | |
375 : | Machine 1 | | Machine 2 |
376 : \-------------------/ \-------------------/
377 : */
378 : template <typename AlgTopoMatch, typename InsAlgTemplate0, typename InsAlgTemplate1>
379 0 : HcclResult InsAllReduceParallelExecutor<AlgTopoMatch, InsAlgTemplate0, InsAlgTemplate1>::GenInsQues(
380 : InsAlgTemplate0& tempAlgIntra, InsAlgTemplate1& tempAlgInter)
381 : {
382 0 : u64 alignedSize = 128; // 假设需要128字节对齐，太大会导致后续maxCountPerLoop计算有问题
383 0 : u32 multipleIntra = tempAlgIntra.CalcScratchMultiple(BufferType::INPUT, BufferType::OUTPUT);
384 0 : u32 multipleInter = tempAlgInter.CalcScratchMultiple(BufferType::INPUT, BufferType::OUTPUT);
385 0 : u64 memBlockSize = UB_MAX_DATA_SIZE;
386 0 : CalcSendDataSize(memBlockSize, dataSplitRate, multipleIntra, multipleInter);
387 : // dataSplitSize为分数，这里maxCountPerLoop对10取整，ScratchBufferSize为1M时可能会导致maxCountPerLoop为0；
388 0 : u64 maxCountPerLoop = (memBlockSize / dataTypeSize_ / 10 / alignedSize) * 10 * alignedSize;
389 0 : CHK_PRT_RET(
390 : maxCountPerLoop == 0,
391 : HCCL_ERROR("[InsAllReduceParallelExecutor] memBlockSize:%llu,maxCountPerLoop==0!.", memBlockSize),
392 : HcclResult::HCCL_E_INTERNAL);
393 0 : u32 loopTimes = dataCount_ / maxCountPerLoop + ((dataCount_ % maxCountPerLoop == 0) ? 0 : 1);
394 :
395 0 : TemplateDataParams tempAlgParamsIntra0, tempAlgParamsInter0, tempAlgParamsInter1, tempAlgParamsIntra1;
396 0 : TempFuncs tempFuncs;
397 0 : tempFuncs.enableCounterNotify = false;
398 0 : tempFuncs.opMode = opMode_;
399 0 : tempFuncs.isBottom = true;
400 0 : tempFuncs.isForepart = true;
401 0 : for (u32 loopIndex = 0; loopIndex < loopTimes; loopIndex++) {
402 0 : u64 currCount = (loopIndex == loopTimes - 1) ? (dataCount_ - loopIndex * maxCountPerLoop) : maxCountPerLoop;
403 0 : u64 dataCountPerLoopAixs0 = static_cast<u64>(dataSplitRate * currCount);
404 0 : u64 dataCountPerLoopAixs1 = currCount - dataCountPerLoopAixs0;
405 : // 第一步开始前同步
406 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
407 0 : u64 dataOffset0 = loopIndex * maxCountPerLoop * dataTypeSize_;
408 0 : u64 dataOffset1 = dataOffset0 + dataCountPerLoopAixs0 * dataTypeSize_;
409 :
410 0 : tempAlgParamsIntra0.buffInfo.scratchBuffSize = Intra0ScratchSize;
411 0 : GenTemplateAlgParams0(dataOffset0, dataCountPerLoopAixs0, 0, tempAlgParamsIntra0);
412 : // 把每个template需要的queue传进去，比如stars的mesh要传多条queue
413 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra0, intraLinks_, intraQue_));
414 0 : tempAlgParamsInter0.buffInfo.scratchBuffSize = Inter0ScratchSize;
415 0 : GenTemplateAlgParams0(dataOffset1, dataCountPerLoopAixs1, interScratchOffset0, tempAlgParamsInter0);
416 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter0, interLinks_, interQue_));
417 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
418 :
419 : // 第二步开始前同步
420 0 : CHK_RET(PreSyncQues(syncQueues_, 0));
421 0 : tempAlgParamsInter1.buffInfo.scratchBuffSize = Inter1ScratchSize;
422 0 : GenTemplateAlgParams1(dataOffset0, dataCountPerLoopAixs0, interScratchOffset1, tempAlgParamsInter1);
423 0 : CHK_RET(tempAlgInter.GenExtIns(tempFuncs, tempAlgParamsInter1, interLinks_, interQue_));
424 0 : tempAlgParamsIntra1.buffInfo.scratchBuffSize = Intra1ScratchSize;
425 0 : GenTemplateAlgParams1(dataOffset1, dataCountPerLoopAixs1, 0, tempAlgParamsIntra1);
426 0 : CHK_RET(tempAlgIntra.GenExtIns(tempFuncs, tempAlgParamsIntra1, intraLinks_, intraQue_));
427 0 : CHK_RET(PostSyncQues(syncQueues_, 0));
428 : }
429 0 : return HcclResult::HCCL_SUCCESS;
430 0 : }
431 :
432 : // 算法注册
433 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
434 : OpType::ALLREDUCE, InsAllReduceParallelMesh1DNHR, InsAllReduceParallelExecutor, TopoMatchMeshNHR,
435 : InsTempAllReduceMesh1DTwoShot, InsTempAllReduceNHR);
436 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
437 : OpType::ALLREDUCE, InsAllReduceParallelMesh2DNHR, InsAllReduceParallelExecutor, TopoMatchConcurrMeshNHR,
438 : InsTempAllReduceMesh2DTwoShot, InsTempAllReduceNHR);
439 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
440 : OpType::ALLREDUCE, InsAllReduceParallelNHRNHR, InsAllReduceParallelExecutor, TopoMatchMeshNHR, InsTempAllReduceNHR,
441 : InsTempAllReduceNHR);
442 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
443 : OpType::ALLREDUCE, InsAllReduceParallelMesh1DNHRPcie, InsAllReduceParallelExecutor, TopoMatchMeshNHRPcie,
444 : InsTempAllReduceMesh1DTwoShot, InsTempAllReduceNHR);
445 :
446 : #ifndef CCL_KERNEL_AICPU
447 : INS_REGISTER_IMPL_BY_TWO_TEMPS(
448 : OpType::ALLREDUCE, CcuAllReduceParallelMesh1DNHR, InsAllReduceParallelExecutor, TopoMatchMeshNHR,
449 : CcuTempAllReduceMeshMem2Mem1D, CcuTempAllReduceNHRMem2Mem1D);
450 : #endif
451 : } // namespace Hccl
|
