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 "log.h"
12 : #include "ins_coll_alg_registry.h"
13 : #include "ins_reduce_scatter_sole_executor.h"
14 : #ifndef CCL_KERNEL_AICPU
15 : #include "ccu_temp_reduce_scatter_mesh_1D.h"
16 : #include "ccu_temp_reduce_scatter_mesh_2D.h"
17 : #include "ccu_temp_reduce_scatter_mesh_2D_mem2mem.h"
18 : #include "ccu_temp_reduce_scatter_mesh_detour_1D.h"
19 : #endif
20 :
21 : #include "topo_match_mesh.h"
22 : #include "topo_match_concurr_mesh.h"
23 :
24 : namespace Hccl {
25 : template <typename AlgTopoMatch, typename InsAlgTemplate>
26 0 : InsReduceScatterSoleExecutor<AlgTopoMatch, InsAlgTemplate>::InsReduceScatterSoleExecutor() : InsCollAlgBase()
27 0 : {}
28 :
29 : template <typename AlgTopoMatch, typename InsAlgTemplate>
30 0 : InsReduceScatterSoleExecutor<AlgTopoMatch, InsAlgTemplate>::~InsReduceScatterSoleExecutor()
31 0 : {}
32 :
33 : template <typename AlgTopoMatch, typename InsAlgTemplate>
34 0 : HcclResult InsReduceScatterSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CalcResOffload(
35 : const RankGraph* rankGraph, const u64& dataSize, CollOffloadOpResReq& resReq)
36 : {
37 : (void)dataSize;
38 0 : resReq.requiredScratchMemSize = 200 * 1024 * 1024; // scratch memory size 200 * 1024K
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] [InsReduceScatterSoleExecutor], CalcRes with detouring enabled.");
52 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
53 : } else {
54 0 : HCCL_DEBUG("[InsCollAlgFactory] [InsReduceScatterSoleExecutor], 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 InsReduceScatterSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CalcRes(
65 : const RankGraph* rankGraph, 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(
80 : "[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
81 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
82 : } else {
83 0 : HCCL_DEBUG(
84 : "[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
85 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
86 : }
87 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
88 0 : algResReq.primQueueNum = tempResReq.streamNum;
89 0 : algResReq.queueNotifys = tempResReq.queNotifys;
90 0 : algResReq.localWaitGroupCntNotify = tempResReq.localWaitGroupCntNotify;
91 0 : algResReq.localBcastPostCntNotify = tempResReq.localBcastPostCntNotify;
92 0 : HCCL_DEBUG(
93 : "[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Rank[%d], requiredQueNum [%u].", myRank_,
94 : algResReq.primQueueNum);
95 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
96 :
97 0 : return HcclResult::HCCL_SUCCESS;
98 0 : }
99 :
100 : // dataSize_ as input
101 : template <typename AlgTopoMatch, typename InsAlgTemplate>
102 0 : HcclResult InsReduceScatterSoleExecutor<AlgTopoMatch, InsAlgTemplate>::Orchestrate(
103 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
104 : {
105 0 : HCCL_INFO("[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Host Orchestrate begins.");
106 : // init and check params
107 0 : CHK_RET(Init(op, params, insQue));
108 :
109 : // Topo Match
110 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
111 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
112 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], [%s].", myRank_, topoMatch.Describe().c_str());
113 0 : dataTypeSize_ = DataTypeSizeGet(dataType_);
114 0 : dataSize_ = dataCount_ * dataTypeSize_;
115 0 : CHK_PRT_RET(
116 : dataTypeSize_ == 0,
117 : HCCL_ERROR("Reducescatter_[CollAlgFactory] Rank [%d], Invalid dataTypeSize_ [%u].", myRank_, dataTypeSize_),
118 : HcclResult::HCCL_E_INTERNAL);
119 :
120 : // 实例化算法模板类
121 0 : HCCL_DEBUG(
122 : "[InsReduceScatterSoleExecutor] Rank[%d], Init insAlgTemplate with rankSize [%u] and dmaMode [%s].", myRank_,
123 : rankSize_, dmaMode_.Describe().c_str());
124 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
125 0 : tempAlg.SetDmaMode(dmaMode_);
126 0 : tempAlg.InitReduceInfo(redOp_, dataType_);
127 0 : tempAlg.SetCollOp(op); // CCU template需要传递op信息
128 :
129 : // 计算算法模板所需资源
130 0 : AlgTempResReq tempResReq;
131 0 : if (enableDetour_) {
132 0 : HCCL_DEBUG(
133 : "[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Rank[%d], CalcRes with detouring enabled for "
134 : "Orchestrate.",
135 : myRank_);
136 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
137 : } else {
138 0 : HCCL_DEBUG(
139 : "[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Rank[%d], CalcRes with detouring disabled for "
140 : "Orchestrate.",
141 : myRank_);
142 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
143 : }
144 : // 申请算法模板所需资源
145 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
146 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, tempResLinks_));
147 0 : ParamPool paramPool = {op, params};
148 0 : HCCL_DEBUG(
149 : "[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Rank[%d], Generating Instruction Queues in OFFLOAD Mode "
150 : "for HOST.",
151 : myRank_);
152 0 : CHK_RET(OrchestrateLoop(tempAlg, paramPool));
153 :
154 0 : return HcclResult::HCCL_SUCCESS;
155 0 : }
156 :
157 : // 算子执行aicpu接口
158 : template <typename AlgTopoMatch, typename InsAlgTemplate>
159 0 : HcclResult InsReduceScatterSoleExecutor<AlgTopoMatch, InsAlgTemplate>::Orchestrate(
160 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
161 : InsQuePtr insQue)
162 : {
163 0 : CHK_PTR_NULL(linkMgr);
164 0 : HCCL_INFO("[InsCollAlgFactory] [InsReduceScatterSoleExecutor] AiCpu Orchestrate begins.");
165 : // 参数校验和初始化
166 0 : CHK_RET(Init(op, params, insQue));
167 :
168 : // soleEsecutor 只支持单层拓扑, 所以只取第0级通信域的信息
169 0 : vTopo_ = topoInfo.vTopo[0]; // 本通信域内的通信平面
170 0 : virtRankMap_ = topoInfo.virtRankMap[0]; // 本通信域内的 rank 映射表
171 0 : virtRanks_ = topoInfo.virtRanks[0]; // 本通信域内的 rank 集合
172 0 : dataTypeSize_ = DataTypeSizeGet(dataType_);
173 0 : dataSize_ = dataCount_ * dataTypeSize_;
174 0 : CHK_PRT_RET(
175 : dataTypeSize_ == 0,
176 : HCCL_ERROR("Reducescatter_[CollAlgFactory] Rank [%d], Invalid dataTypeSize_ [%u].", myRank_, dataTypeSize_),
177 : HcclResult::HCCL_E_INTERNAL);
178 :
179 : // 实例化算法模板类
180 0 : HCCL_DEBUG(
181 : "[InsReduceScatterSoleExecutor] Rank[%d], Init insAlgTemplate with rankSize [%u] and dmaMode [%s].", myRank_,
182 : rankSize_, dmaMode_.Describe().c_str());
183 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
184 0 : tempAlg.SetDmaMode(dmaMode_);
185 0 : tempAlg.InitReduceInfo(redOp_, dataType_);
186 0 : tempAlg.SetCollOp(op); // CCU template需要传递op信息
187 :
188 : // 计算算法模板所需资源
189 0 : AlgTempResReq tempResReq;
190 0 : if (enableDetour_) {
191 0 : HCCL_DEBUG(
192 : "[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
193 0 : CHK_RET(tempAlg.CalcResDetour(linkMgr, tempResReq));
194 : } else {
195 0 : HCCL_DEBUG(
196 : "[InsCollAlgFactory] [InsReduceScatterSoleExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
197 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
198 : }
199 :
200 : // 申请算法模板所需资源
201 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
202 0 : CHK_RET(PrepResLinks(myRank_, tempResReq.links, linkMgr, tempResLinks_));
203 0 : ParamPool paramPool = {op, params};
204 0 : HCCL_DEBUG(
205 : "[InsReduceScatterSoleExecutor] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for AICPU.", myRank_);
206 0 : CHK_RET(OrchestrateLoop(tempAlg, paramPool));
207 :
208 0 : return HcclResult::HCCL_SUCCESS;
209 0 : }
210 :
211 : template <typename AlgTopoMatch, typename InsAlgTemplate>
212 0 : HcclResult InsReduceScatterSoleExecutor<AlgTopoMatch, InsAlgTemplate>::OrchestrateLoop(
213 : InsAlgTemplate& tempAlg, const ParamPool& paramPool)
214 : {
215 : (void)paramPool;
216 0 : BuffInfo buffInfo;
217 0 : TempFuncs tempFuncs;
218 0 : tempFuncs.opMode = opMode_;
219 0 : tempFuncs.enableCounterNotify = IsEnableCounterNotify();
220 0 : if (opMode_ == OpMode::OFFLOAD) {
221 0 : buffInfo.inBuffType = BufferType::INPUT;
222 0 : buffInfo.outBuffType = BufferType::OUTPUT;
223 : } else {
224 0 : buffInfo.inBuffType = BufferType::SCRATCH;
225 0 : buffInfo.outBuffType = BufferType::SCRATCH;
226 0 : tempFuncs.isForepart = true; // Usr Buff to CCL Buff required
227 0 : tempFuncs.isBottom = true; // CCL Buff to Usr Buff required
228 : }
229 0 : buffInfo.scratBuffType = BufferType::SCRATCH;
230 0 : buffInfo.inBuffBaseOff = 0;
231 0 : buffInfo.outBuffBaseOff = 0;
232 0 : HCCL_DEBUG(
233 : "[InsCollAlgFactory] Rank[%d], input buffer type [%s], output buffer type [%s], input buffer base "
234 : "offset [%u], output buffer base offset [%u].",
235 : myRank_, buffInfo.inBuffType.Describe().c_str(), buffInfo.outBuffType.Describe().c_str(),
236 : buffInfo.inBuffBaseOff, buffInfo.outBuffBaseOff);
237 0 : u64 outputCount = dataCount_;
238 0 : u64 outputSize = outputCount * dataTypeSize_;
239 0 : CHK_PRT_RET(rankSize_ == 0, HCCL_ERROR("[CollAlgFactory] RankSize is zero!"), HcclResult::HCCL_E_PARA);
240 0 : AllignInfo allignInfo = {enableAllign_, allignSize_, dataType_};
241 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], done calculating slice information.", myRank_);
242 :
243 0 : u64 maxLoopOutputCount = 0;
244 0 : u64 transportBoundDataSize = UB_MAX_DATA_SIZE;
245 0 : if (opMode_ == OpMode::OFFLOAD) {
246 0 : HCCL_INFO("[InsV2ReduceScatterSoleExecutor]transportBoundDataSize [%u]", transportBoundDataSize);
247 0 : maxLoopOutputCount = transportBoundDataSize / dataTypeSize_;
248 : } else {
249 0 : HCCL_INFO("[InsV2ReduceScatterSoleExecutor]maxTmpMemSize_ [%u]", maxTmpMemSize_);
250 0 : maxLoopOutputCount = maxTmpMemSize_ / (rankSize_ * dataTypeSize_);
251 : }
252 0 : CHK_PRT_RET(
253 : maxLoopOutputCount == 0, HCCL_ERROR("[InsReduceScatterSoleExecutor] maxLoopOutputCount is zero!"),
254 : HcclResult::HCCL_E_PARA);
255 0 : u64 loopTimes = outputCount / maxLoopOutputCount + static_cast<u64>(outputCount % maxLoopOutputCount != 0);
256 0 : for (u64 loop = 0; loop < loopTimes; loop++) {
257 0 : u64 loopOffsetCount = loop * maxLoopOutputCount;
258 0 : u64 loopOffsetSize = loopOffsetCount * dataTypeSize_;
259 : // 本轮需要处理的output数据总量
260 0 : u64 currOutputCount = (loop == (loopTimes - 1)) ? outputCount - loopOffsetCount : maxLoopOutputCount;
261 0 : u64 currSliceSize = currOutputCount * dataTypeSize_;
262 0 : UsrData usrData;
263 0 : for (RankId rankId : virtRanks_) {
264 0 : u32 rankIdx = virtRankMap_[rankId];
265 0 : u64 rankOffsetSize = rankIdx * outputSize;
266 : // 需要处理每一个rank的userIn数据
267 0 : usrData.usrInSlices.emplace_back(BufferType::INPUT, rankOffsetSize + loopOffsetSize, currSliceSize);
268 : // userIn数据搬到CCLIn上时的对应位置
269 0 : usrData.scratchInSlices.emplace_back(BufferType::SCRATCH, rankIdx * currSliceSize, currSliceSize);
270 : }
271 : // 直接将对应rank的整块数据搬到userOut上
272 0 : usrData.scratchOutSlices.emplace_back(
273 0 : BufferType::SCRATCH, virtRankMap_[myRank_] * currSliceSize, currSliceSize);
274 0 : usrData.usrOutSlices.emplace_back(BufferType::OUTPUT, loopOffsetSize, currSliceSize);
275 0 : tempFuncs.usrData = usrData;
276 0 : RankSliceInfo sliceInfoVec;
277 0 : CHK_RET(tempAlg.CalcSliceInfo(allignInfo, currSliceSize, sliceInfoVec));
278 0 : CHK_RET(tempAlg.Run(tempFuncs, sliceInfoVec, buffInfo, tempResLinks_, requiredQue_));
279 : }
280 0 : return HcclResult::HCCL_SUCCESS;
281 0 : }
282 :
283 : #ifndef CCL_KERNEL_AICPU
284 : INS_REGISTER_IMPL_BY_TEMP(
285 : OpType::REDUCESCATTER, CcuReduceScatterMesh1D, InsReduceScatterSoleExecutor, TopoMatchMesh,
286 : CcuTempReduceScatterMesh1D);
287 : INS_REGISTER_IMPL_BY_TEMP(
288 : OpType::REDUCESCATTER, CcuReduceScatterMesh2D, InsReduceScatterSoleExecutor, TopoMatchConcurrMesh,
289 : CcuTempReduceScatterMesh2D);
290 : INS_REGISTER_IMPL_BY_TEMP(
291 : OpType::REDUCESCATTER, CcuReduceScatterMeshMem2Mem2D, InsReduceScatterSoleExecutor, TopoMatchConcurrMesh,
292 : CcuTempReduceScatterMeshMem2Mem2D);
293 : INS_REGISTER_IMPL_BY_TEMP(
294 : OpType::REDUCESCATTER, CcuReduceScatterMeshDetour1D, InsReduceScatterSoleExecutor, TopoMatchMesh,
295 : CcuTempReduceScatterMeshDetour1D);
296 : #endif
297 : } // namespace Hccl
|
