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 <cmath>
12 :
13 : #include "log.h"
14 :
15 : #include "coll_alg_registry.h"
16 : #include "all_reduce_comb_executor.h"
17 :
18 : namespace Hccl {
19 : template <typename AlgTopoMatch, typename AlgTempRS, typename AlgTempAG>
20 0 : AllReduceCombExecutor<AlgTopoMatch, AlgTempRS, AlgTempAG>::AllReduceCombExecutor() : CollAlgBase()
21 : {
22 0 : }
23 :
24 : template <typename AlgTopoMatch, typename AlgTempRS, typename AlgTempAG>
25 0 : AllReduceCombExecutor<AlgTopoMatch, AlgTempRS, AlgTempAG>::~AllReduceCombExecutor()
26 : {
27 0 : }
28 :
29 : template <typename AlgTopoMatch, typename AlgTempRS, typename AlgTempAG>
30 0 : HcclResult AllReduceCombExecutor<AlgTopoMatch, AlgTempRS, AlgTempAG>::CalcResOffload(const RankGraph *rankGraph,
31 : const u64 &dataSize,
32 : CollOffloadOpResReq &resReq)
33 : {
34 : (void)dataSize;
35 0 : resReq.requiredScratchMemSize = 0;
36 :
37 : // Topo Match
38 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
39 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
40 0 : HCCL_INFO("[CollAlgFactory] Rank[%d], [%s].", myRank_, topoMatch.Describe().c_str());
41 :
42 : // instantiate templates
43 0 : AlgTempRS tempRSAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
44 0 : AlgTempAG tempAGAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
45 :
46 : // calculate required primQues and prepare queue
47 0 : AlgTempResReq tempResReqRS;
48 0 : u32 requiredScratchMultiplier = 0;
49 0 : if (enableDetour_) {
50 0 : CHK_RET(tempRSAlg.CalcResDetour(true, rankGraph, tempResReqRS, requiredScratchMultiplier));
51 : } else {
52 0 : CHK_RET(tempRSAlg.CalcRes(true, tempResReqRS, requiredScratchMultiplier));
53 : }
54 :
55 0 : AlgTempResReq tempResReqAG;
56 0 : if (enableDetour_) {
57 0 : CHK_RET(tempAGAlg.CalcResDetour(rankGraph, tempResReqAG));
58 : } else {
59 0 : CHK_RET(tempAGAlg.CalcRes(tempResReqAG));
60 : }
61 :
62 0 : CHK_PRT_RET(
63 : tempResReqRS.queNum != tempResReqAG.queNum,
64 : HCCL_ERROR(
65 : "[CollAlgFactory] Rank [%d], required QueNum for RS template [%u] not equals to it for AG template [%u].",
66 : myRank_, tempResReqRS.queNum, tempResReqAG.queNum),
67 : HcclResult::HCCL_E_INTERNAL);
68 :
69 0 : resReq.requiredSubQueNum = tempResReqRS.queNum - 1;
70 :
71 0 : return HcclResult::HCCL_SUCCESS;
72 0 : }
73 :
74 : template <typename AlgTopoMatch, typename AlgTempRS, typename AlgTempAG>
75 0 : HcclResult AllReduceCombExecutor<AlgTopoMatch, AlgTempRS, AlgTempAG>::GenPrimQues(const RankGraph *rankGraph,
76 : const CollAlgOperator &op,
77 : const CollAlgParams ¶ms,
78 : PrimQuePtr primQue)
79 : {
80 : // init and check params
81 0 : CHK_RET(Init(op, params, primQue));
82 :
83 : // Topo Match
84 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
85 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
86 :
87 : // instantiate templates
88 0 : AlgTempRS tempRSAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
89 0 : tempRSAlg.InitReduceInfo(redOp_, dataType_);
90 0 : tempRSAlg.SetDmaMode(dmaMode_);
91 0 : AlgTempAG tempAGAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
92 0 : tempAGAlg.SetDmaMode(dmaMode_);
93 :
94 : // calculate required primQues and prepare queue
95 0 : AlgTempResReq tempResReqRS;
96 0 : u32 requiredScratchMultiplier = 0;
97 0 : if (enableDetour_) {
98 0 : tempRSAlg.SetDataType(dataType_);
99 0 : CHK_RET(tempRSAlg.CalcResDetour(true, rankGraph, tempResReqRS, requiredScratchMultiplier));
100 : } else {
101 0 : CHK_RET(tempRSAlg.CalcRes(true, tempResReqRS, requiredScratchMultiplier));
102 : }
103 :
104 0 : CHK_RET(InitQueue(tempResReqRS.queNum, requiredQue_));
105 :
106 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, tempResReqRS.links, tempResLinks_));
107 :
108 0 : AlgTempResReq tempResReqAG;
109 0 : if (enableDetour_) {
110 0 : tempAGAlg.SetDataType(dataType_);
111 0 : CHK_RET(tempAGAlg.CalcResDetour(rankGraph, tempResReqAG));
112 : } else {
113 0 : CHK_RET(tempAGAlg.CalcRes(tempResReqAG));
114 : }
115 :
116 0 : CHK_PRT_RET(
117 : tempResReqAG.queNum != tempResReqRS.queNum,
118 : HCCL_ERROR(
119 : "[CollAlgFactory] Rank [%d], required QueNum for RS template [%u] not equals to it for AG template [%u].",
120 : myRank_, tempResReqRS.queNum, tempResReqAG.queNum),
121 : HcclResult::HCCL_E_INTERNAL);
122 :
123 0 : HCCL_INFO(
124 : "[CollAlgFactory] Rank[%d], reduce scatter template [%s], all gather template [%s]: requiredQue Num [%u].",
125 : myRank_, tempRSAlg.Describe().c_str(), tempAGAlg.Describe().c_str(), tempResReqRS.queNum);
126 :
127 0 : u32 dataSizePerVolume = DataTypeSizeGet(dataType_);
128 0 : dataSize_ = dataCount_ * dataSizePerVolume; // for allreduce, dataSize is the size of whole data
129 :
130 0 : if (opMode_ == OpMode::OFFLOAD) {
131 0 : HCCL_INFO("[CollAlgFactory] Rank[%d], Generating Primitive Queues in OFFLOAD Mode for Host.", myRank_);
132 0 : CHK_RET(GenPrimQues4Offload(tempRSAlg, tempAGAlg));
133 : } else { // OPBASE
134 0 : HCCL_INFO("[CollAlgFactory] Rank[%d], Generating Primitive Queues in OPBASE Mode for Host.", myRank_);
135 0 : CHK_RET(GenPrimQues4Opbase(requiredScratchMultiplier, dataSizePerVolume, tempRSAlg, tempAGAlg));
136 : }
137 0 : return HcclResult::HCCL_SUCCESS;
138 0 : }
139 :
140 : template <typename AlgTopoMatch, typename AlgTempRS, typename AlgTempAG>
141 0 : HcclResult AllReduceCombExecutor<AlgTopoMatch, AlgTempRS, AlgTempAG>::CalcRes(const RankGraph *rankGraph,
142 : CollAlgResReq &algResReq)
143 : {
144 : // Topo Match
145 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
146 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
147 0 : algResReq.topoInfo.UpdateSingleLevelTopo(virtRanks_, virtRankMap_, vTopo_);
148 :
149 : // instantiate a template
150 0 : AlgTempRS tempRSAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
151 0 : tempRSAlg.InitReduceInfo(redOp_, dataType_);
152 :
153 : // calculate required primQues and prepare queue
154 0 : AlgTempResReq tempResReqRS;
155 0 : u32 requiredScratchMultiplier = 0;
156 0 : if (enableDetour_) {
157 0 : tempRSAlg.SetDataType(dataType_);
158 0 : CHK_RET(tempRSAlg.CalcResDetour(true, rankGraph, tempResReqRS, requiredScratchMultiplier));
159 : } else {
160 0 : CHK_RET(tempRSAlg.CalcRes(true, tempResReqRS, requiredScratchMultiplier));
161 : }
162 :
163 0 : algResReq.primQueueNum = tempResReqRS.queNum;
164 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReqRS.links, algResReq.links));
165 :
166 0 : return HcclResult::HCCL_SUCCESS;
167 0 : }
168 :
169 : template <typename AlgTopoMatch, typename AlgTempRS, typename AlgTempAG>
170 0 : HcclResult AllReduceCombExecutor<AlgTopoMatch, AlgTempRS, AlgTempAG>::GenPrimQuesAIC(const AlgTopoInfo &topoInfo,
171 : const CollAlgOperator &op,
172 : const CollAlgParams ¶ms,
173 : ConnectedLinkMgr *linkMgr,
174 : PrimQuePtr primQue)
175 : {
176 : // init and check params
177 0 : CHK_RET(Init(op, params, primQue));
178 :
179 : // instantiate templates
180 0 : AlgTempRS tempRSAlg(myRank_, rankSize_, topoInfo.vTopo[0], topoInfo.virtRankMap[0]);
181 0 : tempRSAlg.InitReduceInfo(redOp_, dataType_);
182 0 : tempRSAlg.SetDmaMode(dmaMode_);
183 0 : AlgTempAG tempAGAlg(myRank_, rankSize_, topoInfo.vTopo[0], topoInfo.virtRankMap[0]);
184 0 : tempAGAlg.SetDmaMode(dmaMode_);
185 :
186 : // calculate required primQues and prepare queue
187 0 : AlgTempResReq tempResReqRS;
188 0 : u32 requiredScratchMultiplier = 0;
189 0 : if (enableDetour_) {
190 0 : tempRSAlg.SetDataType(dataType_);
191 0 : CHK_RET(tempRSAlg.CalcResDetour(true, linkMgr, tempResReqRS, requiredScratchMultiplier));
192 : } else {
193 0 : CHK_RET(tempRSAlg.CalcRes(true, tempResReqRS, requiredScratchMultiplier));
194 : }
195 :
196 0 : CHK_RET(InitQueue(tempResReqRS.queNum, requiredQue_));
197 :
198 0 : CHK_RET(PrepResLinks(myRank_, tempResReqRS.links, linkMgr, tempResLinks_));
199 :
200 0 : AlgTempResReq tempResReqAG;
201 :
202 0 : if (enableDetour_) {
203 0 : tempAGAlg.SetDataType(dataType_);
204 0 : CHK_RET(tempAGAlg.CalcResDetour(linkMgr, tempResReqAG));
205 : } else {
206 0 : CHK_RET(tempAGAlg.CalcRes(tempResReqAG));
207 : }
208 :
209 0 : HCCL_INFO(
210 : "[CollAlgFactory] Rank[%d], reduce scatter template [%s], all gather template [%s]: requiredQue Num [%u].",
211 : myRank_, tempRSAlg.Describe().c_str(), tempAGAlg.Describe().c_str(), tempResReqRS.queNum);
212 :
213 0 : u32 dataSizePerVolume = DataTypeSizeGet(dataType_);
214 0 : dataSize_ = dataCount_ * dataSizePerVolume; // for allreduce, dataSize is the size of whole data
215 :
216 0 : if (opMode_ == OpMode::OFFLOAD) {
217 0 : HCCL_INFO("[CollAlgFactory] Rank[%d], Generating Primitive Queues in OFFLOAD Mode for AICPU.", myRank_);
218 0 : CHK_RET(GenPrimQues4Offload(tempRSAlg, tempAGAlg));
219 : } else { // OPBASE
220 0 : HCCL_INFO("[CollAlgFactory] Rank[%d], Generating Primitive Queues in OPBASE Mode for AICPU.", myRank_);
221 0 : CHK_RET(GenPrimQues4Opbase(requiredScratchMultiplier, dataSizePerVolume, tempRSAlg, tempAGAlg));
222 : }
223 0 : return HcclResult::HCCL_SUCCESS;
224 0 : }
225 :
226 : template <typename AlgTopoMatch, typename AlgTempRS, typename AlgTempAG>
227 0 : HcclResult AllReduceCombExecutor<AlgTopoMatch, AlgTempRS, AlgTempAG>::GenPrimQues4Offload(AlgTemplateBase &tempRSAlg,
228 : AlgTemplateBase &tempAGAlg)
229 : {
230 0 : RankSliceInfo sliceInfoVec;
231 0 : AllignInfo allignInfo = {enableAllign_, allignSize_, dataType_};
232 0 : CHK_RET(tempRSAlg.CalcSliceInfo(allignInfo, true, dataSize_, sliceInfoVec));
233 :
234 0 : BuffInfo buffInfo;
235 0 : buffInfo.inBuffType = BufferType::INPUT;
236 0 : buffInfo.outBuffType = BufferType::OUTPUT;
237 0 : buffInfo.scratBuffType = BufferType::OUTPUT;
238 0 : buffInfo.inBuffBaseOff = 0;
239 0 : buffInfo.outBuffBaseOff = 0;
240 0 : buffInfo.scratchBuffBaseOff = 0;
241 :
242 0 : TempFuncs tempFuncs;
243 0 : tempFuncs.opMode = opMode_;
244 0 : tempFuncs.enableCounterNotify = IsEnableCounterNotify();
245 0 : tempFuncs.forAllReduce = true;
246 :
247 0 : CHK_RET(tempRSAlg.GenPrimQue(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
248 0 : CHK_RET(tempAGAlg.GenPrimQue(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
249 :
250 0 : return HcclResult::HCCL_SUCCESS;
251 0 : }
252 :
253 : template <typename AlgTopoMatch, typename AlgTempRS, typename AlgTempAG>
254 0 : HcclResult AllReduceCombExecutor<AlgTopoMatch, AlgTempRS, AlgTempAG>::GenPrimQues4Opbase(
255 : const u32 requiredScratchMultiplier, const u32 dataSizePerVolume, AlgTemplateBase &tempRSAlg,
256 : AlgTemplateBase &tempAGAlg)
257 : {
258 0 : u64 scratchInputMemSize
259 0 : = (rankSize_ % dataSizePerVolume == 0)
260 0 : ? static_cast<int>(floor(maxTmpMemSize_ / (rankSize_ + requiredScratchMultiplier)) * rankSize_)
261 0 : : static_cast<int>(floor(maxTmpMemSize_ / ((rankSize_ + requiredScratchMultiplier) * dataSizePerVolume))
262 0 : * rankSize_ * dataSizePerVolume);
263 :
264 0 : CHK_PRT_RET(scratchInputMemSize == 0,
265 : HCCL_ERROR("[CollAlgFactory] Rank [%d], Invalid input maxTmpMemSize [%u].", myRank_, maxTmpMemSize_),
266 : HcclResult::HCCL_E_PARA);
267 :
268 0 : BuffInfo buffInfo;
269 0 : buffInfo.inBuffType = BufferType::SCRATCH;
270 0 : buffInfo.outBuffType = BufferType::SCRATCH;
271 0 : buffInfo.scratBuffType = BufferType::SCRATCH;
272 :
273 0 : u32 sendRecvTimes = (dataSize_ / scratchInputMemSize) + ((dataSize_ % scratchInputMemSize) == 0 ? 0 : 1);
274 0 : HCCL_INFO("[CollAlgFactory] Rank [%d], datasize [%u], sendRecvTimes [%u].", myRank_, dataSize_, sendRecvTimes);
275 :
276 0 : for (u32 idx = 0; idx < sendRecvTimes; idx++) {
277 0 : u64 currDataSize = (idx == sendRecvTimes - 1) ? (dataSize_ - idx * scratchInputMemSize) : scratchInputMemSize;
278 :
279 0 : buffInfo.inBuffBaseOff = 0;
280 0 : buffInfo.outBuffBaseOff = currDataSize;
281 0 : buffInfo.scratchBuffBaseOff = currDataSize;
282 :
283 0 : RankSliceInfo sliceInfoVec;
284 0 : AllignInfo allignInfo = {enableAllign_, allignSize_, dataType_};
285 0 : CHK_RET(tempRSAlg.CalcSliceInfo(allignInfo, true, currDataSize, sliceInfoVec));
286 0 : TempFuncs tempFuncs;
287 0 : tempFuncs.opMode = opMode_;
288 0 : tempFuncs.enableCounterNotify = IsEnableCounterNotify();
289 0 : tempFuncs.forAllReduce = true;
290 0 : tempFuncs.isForepart = true; // Usr Buff to CCL Buff required
291 :
292 0 : UsrData usrData;
293 0 : DataSlice usrInSlice = DataSlice(BufferType::INPUT, idx * scratchInputMemSize, currDataSize);
294 0 : DataSlice scratchInSlice = DataSlice(BufferType::SCRATCH, 0, currDataSize);
295 0 : usrData.usrInSlices.push_back(usrInSlice);
296 0 : usrData.scratchInSlices.push_back(scratchInSlice);
297 :
298 0 : tempFuncs.usrData = usrData;
299 0 : CHK_RET(tempRSAlg.GenPrimQue(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
300 :
301 0 : buffInfo.outBuffBaseOff = 0;
302 0 : buffInfo.inBuffBaseOff = currDataSize; // will not be used in allgather
303 0 : buffInfo.scratchBuffBaseOff = currDataSize;
304 0 : tempFuncs.isForepart = false; // Usr Buff to CCL Buff required
305 0 : tempFuncs.isBottom = true; // CCL Buff to Usr Buff required
306 :
307 0 : DataSlice scratchOutSlice = DataSlice(BufferType::SCRATCH, 0, currDataSize);
308 0 : DataSlice usrOutSlice = DataSlice(BufferType::OUTPUT, idx * scratchInputMemSize, currDataSize);
309 0 : tempFuncs.usrData.scratchOutSlices.push_back(scratchOutSlice);
310 0 : tempFuncs.usrData.usrOutSlices.push_back(usrOutSlice);
311 0 : CHK_RET(tempAGAlg.GenPrimQue(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
312 : }
313 :
314 0 : return HcclResult::HCCL_SUCCESS;
315 : }
316 :
317 : REGISTER_IMPL_BY_TWO_TEMPS(OpType::ALLREDUCE, AllReduceConcurrMesh, AllReduceCombExecutor, TopoMatchConcurrMesh,
318 : TempReduceScatterConcurrMesh, TempAllGatherConcurrMesh);
319 : REGISTER_IMPL_BY_TWO_TEMPS(OpType::ALLREDUCE, AllReduceMesh, AllReduceCombExecutor, TopoMatchMesh,
320 : TempReduceScatterMesh, TempAllGatherMesh);
321 : } // namespace Hccl
|
