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