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