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