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