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_sole_executor.h"
12 : #include "log.h"
13 : #include "ins_coll_alg_registry.h"
14 : #ifndef CCL_KERNEL_AICPU
15 : #include "ccu_temp_reduce_mesh_2D.h"
16 : #endif
17 : #include "topo_match_concurr_mesh.h"
18 : #include "alg_data_trans_wrapper.h"
19 :
20 : namespace Hccl {
21 : template <typename AlgTopoMatch, typename InsAlgTemplate>
22 0 : InsReduceSoleExecutor<AlgTopoMatch, InsAlgTemplate>::InsReduceSoleExecutor() : InsCollAlgBase()
23 : {
24 0 : }
25 :
26 : template <typename AlgTopoMatch, typename InsAlgTemplate>
27 0 : InsReduceSoleExecutor<AlgTopoMatch, InsAlgTemplate>::~InsReduceSoleExecutor()
28 : {
29 0 : }
30 :
31 : template <typename AlgTopoMatch, typename InsAlgTemplate>
32 0 : HcclResult InsReduceSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CalcResOffload(const RankGraph *rankGraph,
33 : const u64 &dataSize,
34 : CollOffloadOpResReq &resReq)
35 : {
36 : (void)dataSize;
37 0 : constexpr u64 needScratchSize = 200 * 1024 * 1024; // 需要申请200MB临时内存
38 0 : resReq.requiredScratchMemSize = needScratchSize;
39 :
40 : // Topo Match
41 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
42 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
43 :
44 : // instantiate a template
45 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
46 0 : tempAlg.InitReduceInfo(redOp_, dataType_);
47 :
48 : // calculate required insQueues and prepare queue
49 0 : AlgTempResReq tempResReq;
50 0 : if (enableDetour_) {
51 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor], CalcRes with detouring enabled.");
52 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
53 : } else {
54 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor], CalcRes with detouring disabled.");
55 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
56 : }
57 :
58 0 : resReq.requiredSubQueNum = tempResReq.streamNum - 1;
59 :
60 0 : return HcclResult::HCCL_SUCCESS;
61 0 : }
62 :
63 : template <typename AlgTopoMatch, typename InsAlgTemplate>
64 0 : HcclResult InsReduceSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CalcRes(const RankGraph *rankGraph,
65 : CollAlgResReq &algResReq)
66 : {
67 : // Topo Match
68 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
69 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
70 0 : algResReq.topoInfo.UpdateSingleLevelTopo(virtRanks_, virtRankMap_, vTopo_);
71 :
72 : // instantiate a template
73 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
74 0 : tempAlg.InitReduceInfo(redOp_, dataType_);
75 :
76 : // calculate required insQues and prepare queue
77 0 : AlgTempResReq tempResReq;
78 0 : if (enableDetour_) {
79 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
80 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
81 : } else {
82 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
83 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
84 : }
85 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
86 0 : algResReq.primQueueNum = tempResReq.streamNum;
87 0 : algResReq.queueNotifys = tempResReq.queNotifys;
88 0 : algResReq.localWaitGroupCntNotify = tempResReq.localWaitGroupCntNotify;
89 0 : algResReq.localBcastPostCntNotify = tempResReq.localBcastPostCntNotify;
90 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor] Rank[%d], requiredQueNum [%u].", myRank_, algResReq.primQueueNum);
91 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
92 :
93 0 : return HcclResult::HCCL_SUCCESS;
94 0 : }
95 :
96 : // dataSize_ as input
97 : template <typename AlgTopoMatch, typename InsAlgTemplate>
98 0 : HcclResult InsReduceSoleExecutor<AlgTopoMatch, InsAlgTemplate>::Orchestrate(const RankGraph *rankGraph,
99 : const CollAlgOperator &op, const CollAlgParams ¶ms, InsQuePtr insQue)
100 : {
101 0 : HCCL_INFO("[InsCollAlgFactory] [InsReduceSoleExecutor] Host Orchestrate begins.");
102 : // init and check params
103 0 : CHK_RET(Init(op, params, insQue));
104 :
105 : // Topo Match
106 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
107 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
108 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], [%s].", myRank_, topoMatch.Describe().c_str());
109 0 : dataTypeSize_ = DataTypeSizeGet(dataType_);
110 0 : dataSize_ = dataCount_ * dataTypeSize_;
111 0 : CHK_PRT_RET(dataTypeSize_ == 0,
112 : HCCL_ERROR("Reduce_[CollAlgFactory] Rank [%d], Invalid dataTypeSize_ [%u].", myRank_, dataTypeSize_),
113 : HcclResult::HCCL_E_INTERNAL);
114 :
115 : // 实例化算法模板类
116 0 : HCCL_DEBUG("Reduce_[InsReduceSoleExecutor] Rank[%d], Init insAlgTemplate with rankSize [%u] and dmaMode [%s].",
117 : myRank_, rankSize_, dmaMode_.Describe().c_str());
118 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
119 0 : tempAlg.SetDmaMode(dmaMode_);
120 0 : tempAlg.InitReduceInfo(redOp_, dataType_);
121 0 : tempAlg.SetCollOp(op); // CCU template需要传递op信息
122 0 : tempAlg.SetRoot(root_);
123 :
124 : // 计算算法模板所需资源
125 0 : AlgTempResReq tempResReq;
126 0 : if (enableDetour_) {
127 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor] Rank[%d], CalcRes with detouring enabled for Orchestrate.", myRank_);
128 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
129 : } else {
130 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor] Rank[%d], CalcRes with detouring disabled for Orchestrate.", myRank_);
131 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
132 : }
133 : // 申请算法模板所需资源
134 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
135 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, tempResLinks_));
136 :
137 0 : ParamPool paramPool = {op_, params};
138 0 : if (opMode_ == OpMode::OFFLOAD) {
139 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for HOST.", myRank_);
140 0 : CHK_RET(OrchestrateOffload(tempAlg, paramPool));
141 : } else { // OPBASE
142 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
143 0 : CHK_RET(OrchestrateOpbase(tempAlg, paramPool));
144 : }
145 :
146 0 : return HcclResult::HCCL_SUCCESS;
147 0 : }
148 :
149 : // 算子执行aicpu接口
150 : template <typename AlgTopoMatch, typename InsAlgTemplate>
151 0 : HcclResult InsReduceSoleExecutor<AlgTopoMatch, InsAlgTemplate>::Orchestrate(const AlgTopoInfo &topoInfo,
152 : const CollAlgOperator &op, const CollAlgParams ¶ms, ConnectedLinkMgr *linkMgr,
153 : InsQuePtr insQue)
154 : {
155 0 : HCCL_INFO("[InsCollAlgFactory] [InsReduceSoleExecutor] [InsReduceSoleExecutor] AiCpu Orchestrate begins.");
156 : // 参数校验和初始化
157 0 : CHK_RET(Init(op, params, insQue));
158 :
159 : // soleEsecutor 只支持单层拓扑, 所以只取第 0 级通信域的信息
160 0 : vTopo_ = topoInfo.vTopo[0]; // 本通信域内的通信平面
161 0 : virtRankMap_ = topoInfo.virtRankMap[0]; // 本通信域内的 rank 映射表
162 0 : virtRanks_ = topoInfo.virtRanks[0]; // 本通信域内的 rank 集合
163 0 : dataTypeSize_ = DataTypeSizeGet(dataType_);
164 0 : dataSize_ = dataCount_ * dataTypeSize_;
165 0 : CHK_PRT_RET(dataTypeSize_ == 0,
166 : HCCL_ERROR("Reduce_[CollAlgFactory] Rank [%d], Invalid dataTypeSize_ [%u].", myRank_, dataTypeSize_),
167 : HcclResult::HCCL_E_INTERNAL);
168 :
169 : // 实例化算法模板类
170 0 : HCCL_DEBUG("Reduce_[InsReduceSoleExecutor] Rank[%d], Init insAlgTemplate with rankSize [%u] and dmaMode [%s].",
171 : myRank_, rankSize_, dmaMode_.Describe().c_str());
172 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
173 0 : tempAlg.SetDmaMode(dmaMode_);
174 0 : tempAlg.InitReduceInfo(redOp_, dataType_);
175 0 : tempAlg.SetCollOp(op); // CCU template需要传递op信息
176 0 : tempAlg.SetRoot(root_);
177 :
178 : // 计算算法模板所需资源
179 0 : AlgTempResReq tempResReq;
180 0 : if (enableDetour_) {
181 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
182 0 : CHK_RET(tempAlg.CalcResDetour(linkMgr, tempResReq));
183 : } else {
184 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceSoleExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
185 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
186 : }
187 :
188 : // 申请算法模板所需资源
189 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
190 0 : CHK_RET(PrepResLinks(myRank_, tempResReq.links, linkMgr, tempResLinks_));
191 :
192 0 : ParamPool paramPool = {op_, params};
193 0 : if (opMode_ == OpMode::OFFLOAD) {
194 0 : HCCL_DEBUG("[InsReduceSoleExecutor] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for AICPU.",
195 : myRank_);
196 0 : CHK_RET(OrchestrateOffload(tempAlg, paramPool));
197 : } else { // OPBASE
198 0 : HCCL_DEBUG("[InsReduceSoleExecutor] Rank[%d], Generating Instruction Queues in OPBASE Mode for AICPU.",
199 : myRank_);
200 0 : CHK_RET(OrchestrateOpbase(tempAlg, paramPool));
201 : }
202 :
203 0 : return HcclResult::HCCL_SUCCESS;
204 0 : }
205 :
206 : template <typename AlgTopoMatch, typename InsAlgTemplate>
207 0 : HcclResult InsReduceSoleExecutor<AlgTopoMatch, InsAlgTemplate>::OrchestrateOffload(InsAlgTemplate &tempAlg,
208 : const ParamPool ¶mPool)
209 : {
210 : (void)paramPool;
211 0 : u32 dataSizePerVolume = DataTypeSizeGet(dataType_);
212 0 : CHK_PRT_RET(dataSizePerVolume == 0,
213 : HCCL_ERROR("[CollAlgFactory] Rank [%d], Invalid dataSizePerVolume [%u].", myRank_, dataSizePerVolume),
214 : HcclResult::HCCL_E_INTERNAL);
215 0 : u64 transportBoundDataSize = UB_MAX_DATA_SIZE;
216 0 : BuffInfo buffInfo;
217 0 : buffInfo.inBuffType = BufferType::INPUT;
218 0 : buffInfo.outBuffType = BufferType::OUTPUT;
219 0 : buffInfo.inBuffBaseOff = 0;
220 0 : buffInfo.outBuffBaseOff = 0;
221 0 : HCCL_INFO("[InsCollAlgFactory] Rank[%d], input buffer type [%s], output buffer type [%s], input buffer base "
222 : "offset [%u], output buffer base offset [%u].", myRank_, buffInfo.inBuffType.Describe().c_str(),
223 : buffInfo.outBuffType.Describe().c_str(), buffInfo.inBuffBaseOff, buffInfo.outBuffBaseOff);
224 0 : u64 sendRecvTimes = (dataSize_ / transportBoundDataSize) + ((dataSize_ % transportBoundDataSize) == 0 ? 0 : 1);
225 0 : HCCL_INFO("[CollAlgFactory] Rank [%d], sendRecvTimes [%u].", myRank_, sendRecvTimes);
226 0 : for (u64 idx = 0; idx < sendRecvTimes; idx++) {
227 0 : u64 currDataSize = (idx == (sendRecvTimes - 1)) ? (dataSize_ - idx * transportBoundDataSize)
228 : : transportBoundDataSize; // 判断是否为最后一轮
229 0 : RankSliceInfo sliceInfoVec;
230 0 : AllignInfo allignInfo = {enableAllign_, allignSize_, dataType_};
231 0 : TempFuncs tempFuncs;
232 0 : tempFuncs.opMode = opMode_;
233 0 : tempFuncs.enableCounterNotify = IsEnableCounterNotify();
234 0 : tempFuncs.isForepart = true; // Usr Buff to CCL Buff required
235 0 : tempFuncs.isBottom = true; // CCL Buff to Usr Buff required
236 0 : UsrData usrData;
237 : // 将整块数据一次性从 UserIn 搬运到 CclIn 上
238 0 : usrData.usrInSlices.emplace_back(BufferType::INPUT, idx * transportBoundDataSize, currDataSize);
239 0 : usrData.scratchInSlices.emplace_back(BufferType::SCRATCH, 0, currDataSize);
240 : // 将整块数据一次性从 CclOut 搬运到 UserOut 上
241 0 : usrData.scratchOutSlices.emplace_back(BufferType::SCRATCH, 0, currDataSize);
242 0 : usrData.usrOutSlices.emplace_back(BufferType::OUTPUT, idx * transportBoundDataSize, currDataSize);
243 0 : tempFuncs.usrData = usrData;
244 0 : CHK_RET(tempAlg.CalcSliceInfo(allignInfo, transportBoundDataSize, sliceInfoVec));
245 0 : CHK_RET(tempAlg.Run(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
246 : }
247 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], done generating instruction queues.", myRank_);
248 0 : return HcclResult::HCCL_SUCCESS;
249 : }
250 :
251 : template <typename AlgTopoMatch, typename InsAlgTemplate>
252 0 : HcclResult InsReduceSoleExecutor<AlgTopoMatch, InsAlgTemplate>::OrchestrateOpbase(InsAlgTemplate &tempAlg,
253 : ParamPool ¶mPool)
254 : {
255 0 : BuffInfo buffInfo;
256 0 : buffInfo.inBuffType = BufferType::SCRATCH;
257 0 : buffInfo.outBuffType = BufferType::SCRATCH;
258 0 : buffInfo.inBuffBaseOff = 0;
259 0 : buffInfo.outBuffBaseOff = 0;
260 :
261 : // 基本参数配置
262 0 : AllignInfo allignInfo = {enableAllign_, allignSize_, dataType_};
263 0 : TempFuncs tempFuncs;
264 0 : tempFuncs.isForepart = true; // Usr Buff to CCL Buff required
265 0 : tempFuncs.isBottom = true; // CCL Buff to Usr Buff required
266 0 : tempFuncs.opMode = opMode_;
267 0 : tempFuncs.enableCounterNotify = IsEnableCounterNotify();
268 :
269 0 : u64 outputCount = dataCount_;
270 : // 根据CCL Buffer 大小，计算出一轮中最多能输出多少数据
271 0 : u64 maxLoopOutputCount = tempAlg.CalcLoopMaxCount(paramPool);
272 0 : if (maxLoopOutputCount == 0) {
273 0 : HCCL_ERROR("[InsReduceSoleExecutor][OrchestrateOpbase] maxLoopOutputCount is zero!");
274 0 : return HcclResult::HCCL_E_INTERNAL;
275 : }
276 0 : HCCL_INFO("[InsReduceSoleExecutor][OrchestrateOpbase] Actual maxLoopOutputCount: [%lu].", maxLoopOutputCount);
277 :
278 0 : u64 loopTimes = outputCount / maxLoopOutputCount + static_cast<u64>(outputCount % maxLoopOutputCount != 0);
279 0 : for (u32 loop = 0; loop < loopTimes; loop++) {
280 0 : u64 loopOffsetCount = loop * maxLoopOutputCount;
281 0 : u64 loopOffsetSize = loopOffsetCount * dataTypeSize_;
282 : // 本轮需要处理的数据量
283 0 : u64 currOutputCount = (loop == (loopTimes - 1)) ? outputCount - loopOffsetCount : maxLoopOutputCount;
284 0 : u64 currOutputSize = currOutputCount * dataTypeSize_;
285 0 : UsrData usrData;
286 : // 将整块数据一次性从 UserIn 搬运到 CclIn 上
287 0 : usrData.usrInSlices.emplace_back(BufferType::INPUT, loopOffsetSize, currOutputSize);
288 0 : usrData.scratchInSlices.emplace_back(BufferType::SCRATCH, 0, currOutputSize);
289 : // 将整块数据一次性从 CclOut 搬运到 UserOut 上
290 0 : usrData.scratchOutSlices.emplace_back(BufferType::SCRATCH, 0, currOutputSize);
291 0 : usrData.usrOutSlices.emplace_back(BufferType::OUTPUT, loopOffsetSize, currOutputSize);
292 0 : tempFuncs.usrData = usrData;
293 :
294 0 : RankSliceInfo sliceInfoVec;
295 0 : CHK_RET(tempAlg.CalcSliceInfo(allignInfo, currOutputSize, sliceInfoVec));
296 0 : CHK_RET(tempAlg.Run(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
297 : }
298 0 : return HcclResult::HCCL_SUCCESS;
299 0 : }
300 :
301 : #ifndef CCL_KERNEL_AICPU
302 : INS_REGISTER_IMPL_BY_TEMP(OpType::REDUCE, CcuReduceMesh2D, InsReduceSoleExecutor, TopoMatchConcurrMesh,
303 : CcuTempReduceMesh2D);
304 : #endif
305 : } // namespace Hccl
|
