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 "topo_match_mesh.h"
14 : #include "ins_v2_reduce_scatter_v_sole_executor.h"
15 :
16 : #ifndef CCL_KERNEL_AICPU
17 : #include "ccu_temp_reduce_scatter_v_mesh_1D.h"
18 : #include "ccu_temp_reduce_scatter_v_mesh_1D_mem2mem.h"
19 : #endif
20 :
21 : namespace Hccl {
22 : template <typename AlgTopoMatch, typename InsAlgTemplate>
23 0 : InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::InsV2ReduceScatterVSoleExecutor() : InsCollAlgBase()
24 0 : {}
25 :
26 : template <typename AlgTopoMatch, typename InsAlgTemplate>
27 0 : InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::~InsV2ReduceScatterVSoleExecutor()
28 0 : {}
29 :
30 : template <typename AlgTopoMatch, typename InsAlgTemplate>
31 0 : HcclResult InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::InitCommInfo(const RankGraph* rankGraph)
32 : {
33 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
34 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
35 0 : return HcclResult::HCCL_SUCCESS;
36 0 : }
37 :
38 : template <typename AlgTopoMatch, typename InsAlgTemplate>
39 0 : HcclResult InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::InitCommInfo(const AlgTopoInfo& topoInfo)
40 : {
41 0 : CHK_PRT_RET(
42 : topoInfo.vTopo.empty(), HCCL_ERROR("[InsV2ReduceSoleExecutor][InitCommInfo] vTopo size is invalid"),
43 : HCCL_E_PARA);
44 0 : CHK_PRT_RET(
45 : topoInfo.virtRankMap.empty(), HCCL_ERROR("[InsV2ReduceSoleExecutor][InitCommInfo] virtRankMap size is invalid"),
46 : HCCL_E_PARA);
47 0 : CHK_PRT_RET(
48 : topoInfo.virtRanks.empty(), HCCL_ERROR("[InsV2ReduceSoleExecutor][InitCommInfo] virtRanks size is invalid"),
49 : HCCL_E_PARA);
50 0 : vTopo_ = topoInfo.vTopo[0]; // 本通信域内的通信平面
51 0 : virtRankMap_ = topoInfo.virtRankMap[0]; // 本通信域内的 rank 映射表
52 0 : virtRanks_ = topoInfo.virtRanks[0]; // 本通信域内的 rank 集合
53 0 : return HcclResult::HCCL_SUCCESS;
54 : }
55 :
56 : template <typename AlgTopoMatch, typename InsAlgTemplate>
57 0 : HcclResult InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CreateTemplates(
58 : std::shared_ptr<InsAlgTemplate>& algTemplatePtr)
59 : {
60 0 : HCCL_DEBUG("[InsV2ReduceScatterVSoleExecutor][CreateTemplates]");
61 0 : algTemplatePtr = std::make_shared<InsAlgTemplate>(myRank_, rankSize_, vTopo_, virtRankMap_);
62 0 : CHK_PTR_NULL(algTemplatePtr);
63 0 : algTemplatePtr->SetDmaMode(dmaMode_);
64 0 : algTemplatePtr->SetDataType(dataType_);
65 0 : algTemplatePtr->SetCollOp(op_);
66 0 : algTemplatePtr->InitReduceInfo(redOp_, dataType_);
67 0 : return HcclResult::HCCL_SUCCESS;
68 : }
69 :
70 : // HOST 侧算法入口，将对应的 instruction 添加到指令队列中
71 : template <typename AlgTopoMatch, typename InsAlgTemplate>
72 0 : HcclResult InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::Orchestrate(
73 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
74 : {
75 0 : HCCL_INFO("[InsV2ReduceScatterVSoleExecutor][Orchestrate] Orchestrate host Start");
76 :
77 0 : CHK_RET(Init(op, params, insQue));
78 0 : CHK_RET(InitCommInfo(rankGraph));
79 0 : u32 dataSizePerVolume = DataTypeSizeGet(dataType_);
80 0 : dataSize_ = dataCount_ * dataSizePerVolume;
81 0 : std::shared_ptr<InsAlgTemplate> algTemplate = nullptr;
82 0 : CHK_RET(CreateTemplates(algTemplate));
83 :
84 0 : AlgTempResReq tempResReq;
85 0 : if (enableDetour_) {
86 0 : HCCL_DEBUG(
87 : "[InsV2ReduceScatterVSoleExecutor][Orchestrate] [%s] Rank[%d], CalcRes with detouring enabled.", __func__,
88 : myRank_);
89 0 : CHK_RET(algTemplate->CalcResDetour(rankGraph, tempResReq));
90 : } else {
91 0 : HCCL_DEBUG(
92 : "[InsV2ReduceScatterVSoleExecutor][Orchestrate] [%s] Rank[%d], CalcRes with detouring disabled.", __func__,
93 : myRank_);
94 0 : CHK_RET(algTemplate->CalcRes(tempResReq));
95 : }
96 :
97 0 : HCCL_DEBUG(
98 : "[InsV2ReduceScatterVSoleExecutor][Orchestrate] Rank[%d], template [%s], requiredQue Num [%u].", myRank_,
99 : algTemplate->Describe().c_str(), tempResReq.queNum);
100 0 : CHK_RET(InitQueue(tempResReq.queNum, tempInsQue_));
101 0 : CHK_RET(PrepResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, tempResLinks_));
102 0 : CHK_RET(OrchestrateLoop(algTemplate));
103 0 : return HcclResult::HCCL_SUCCESS;
104 0 : }
105 :
106 : template <typename AlgTopoMatch, typename InsAlgTemplate>
107 0 : HcclResult InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::Orchestrate(
108 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
109 : InsQuePtr insQue)
110 : {
111 0 : HCCL_INFO("[InsV2ReduceScatterVSoleExecutor][Orchestrate] Orchestrate AICPU Start");
112 0 : CHK_RET(Init(op, params, insQue));
113 0 : CHK_RET(InitCommInfo(topoInfo));
114 0 : u32 dataSizePerVolume = DataTypeSizeGet(dataType_);
115 0 : dataSize_ = dataCount_ * dataSizePerVolume;
116 0 : std::shared_ptr<InsAlgTemplate> algTemplate = nullptr;
117 0 : CHK_RET(CreateTemplates(algTemplate));
118 0 : AlgTempResReq tempResReq;
119 0 : if (enableDetour_) {
120 0 : CHK_RET(algTemplate->CalcResDetour(linkMgr, tempResReq));
121 : } else {
122 0 : CHK_RET(algTemplate->CalcRes(tempResReq));
123 : }
124 0 : HCCL_DEBUG(
125 : "[InsV2ReduceScatterVSoleExecutor][Orchestrate] Rank[%d], template [%s], requiredQue Num [%u].", myRank_,
126 : algTemplate->Describe().c_str(), tempResReq.queNum);
127 0 : CHK_RET(InitQueue(tempResReq.queNum, tempInsQue_));
128 0 : CHK_RET(PrepResLinks(myRank_, tempResReq.links, linkMgr, tempResLinks_));
129 0 : CHK_RET(OrchestrateLoop(algTemplate));
130 0 : return HcclResult::HCCL_SUCCESS;
131 0 : }
132 :
133 : // 单算子模式资源计算接口
134 : template <typename AlgTopoMatch, typename InsAlgTemplate>
135 0 : HcclResult InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::OrchestrateLoop(
136 : std::shared_ptr<InsAlgTemplate> algTemplate)
137 : {
138 0 : HCCL_INFO(
139 : "[InsV2ReduceScatterVSoleExecutor][OrchestrateOpbase] Start, template[%s]", algTemplate->Describe().c_str());
140 :
141 0 : TemplateDataParams tempAlgParams;
142 0 : tempAlgParams.buffInfo.inBuffType = BufferType::INPUT;
143 0 : tempAlgParams.buffInfo.outBuffType = BufferType::OUTPUT;
144 0 : tempAlgParams.buffInfo.scratBuffType = BufferType::SCRATCH;
145 0 : tempAlgParams.repeatNum = 1; // 不需要重复
146 0 : tempAlgParams.inputRepeatStride = 0;
147 0 : tempAlgParams.outputRepeatStride = 0;
148 :
149 0 : TempFuncs tempFuncs;
150 0 : tempFuncs.opMode = opMode_;
151 0 : tempFuncs.isForepart = true;
152 0 : tempFuncs.isBottom = true;
153 0 : tempFuncs.enableCounterNotify = IsEnableCounterNotify();
154 :
155 0 : u64 maxDataSizePerLoop = 0;
156 : u64 transportBoundDataSize;
157 0 : CHK_RET(algTemplate->GetMaxTransPortDataSize(transportBoundDataSize));
158 0 : u64 templateScratchMultiplier
159 0 : = algTemplate->CalcScratchMultiple(tempAlgParams.buffInfo.inBuffType, tempAlgParams.buffInfo.outBuffType);
160 0 : if (templateScratchMultiplier != 0) {
161 0 : Buffer* scratchBuffer = op_.GetBuffer(BufferType::SCRATCH);
162 0 : CHK_PRT_RET(
163 : scratchBuffer == nullptr || scratchBuffer->GetAddr() == 0 || scratchBuffer->GetSize() == 0,
164 : HCCL_ERROR("Scratch buffer is null while ScratchMultiplier is not 0"), HCCL_E_MEMORY);
165 0 : u64 scratchBoundDataSize = scratchBuffer->GetSize() / templateScratchMultiplier;
166 0 : maxDataSizePerLoop = std::min(transportBoundDataSize, scratchBoundDataSize);
167 : } else {
168 0 : maxDataSizePerLoop = transportBoundDataSize;
169 : }
170 0 : u64 maxDataCountPerLoop = maxDataSizePerLoop / dataTypeSize_;
171 0 : HCCL_INFO(
172 : "[InsAllReduceCombExecutor][OrchestrateOpbase] maxDataCountPerLoop[%llu], maxDataSizePerLoop[%llu], "
173 : "transportBoundDataSize[%llu], templateScratchMultiplier[%llu]",
174 : maxDataCountPerLoop, maxDataSizePerLoop, transportBoundDataSize, templateScratchMultiplier);
175 0 : CHK_PRT_RET(
176 : maxDataCountPerLoop == 0, HCCL_ERROR("[InsAllReduceCombExecutor][OrchestrateOpbase] maxDataCountPerLoop is 0"),
177 : HCCL_E_INTERNAL);
178 :
179 0 : CHK_PRT_RET(
180 : op_.vDataDes.counts == nullptr || op_.vDataDes.displs == nullptr,
181 : HCCL_ERROR("[InsAllReduceCombExecutor][OrchestrateOpbase] counts or displs is nullptr"), HCCL_E_PTR);
182 0 : u64 myRankRecvCount = static_cast<u64*>(op_.vDataDes.counts)[myRank_];
183 0 : u64 myDisplacementCount = static_cast<u64*>(op_.vDataDes.displs)[myRank_];
184 0 : u64 maxRecvDataCount = 0;
185 0 : for (u64 i = 0; i < rankSize_; i++) {
186 0 : maxRecvDataCount = max(maxRecvDataCount, static_cast<u64*>(op_.vDataDes.counts)[i]);
187 : }
188 :
189 0 : u64 loopTimes = 1 + ((maxRecvDataCount - 1) / maxDataCountPerLoop); // 向上取整
190 0 : u64 processedDataCount = 0;
191 0 : for (u64 loop = 0; loop < loopTimes; loop++) {
192 0 : u64 currDataCount
193 0 : = processedDataCount < myRankRecvCount ? min(maxDataCountPerLoop, myRankRecvCount - processedDataCount) : 0;
194 0 : tempAlgParams.buffInfo.inBuffBaseOff = processedDataCount * dataTypeSize_;
195 0 : tempAlgParams.buffInfo.outBuffBaseOff = processedDataCount * dataTypeSize_;
196 0 : tempAlgParams.buffInfo.scratchBuffBaseOff = 0;
197 0 : tempAlgParams.sliceSize = currDataCount * dataTypeSize_;
198 0 : tempAlgParams.tailSize = 0; // 变长算子不涉及
199 : tempAlgParams.inputSliceStride
200 0 : = myDisplacementCount * dataTypeSize_; // 变长算子, 表示自己的这片输入数据的起始位置
201 0 : tempAlgParams.outputSliceStride = 0; // 变长算子， 表示自己的这片输出数据的起始位置
202 :
203 0 : CHK_RET(algTemplate->GenExtIns(tempFuncs, tempAlgParams, tempResLinks_, tempInsQue_));
204 0 : processedDataCount += maxDataCountPerLoop;
205 : }
206 0 : return HcclResult::HCCL_SUCCESS;
207 0 : }
208 :
209 : template <typename AlgTopoMatch, typename InsAlgTemplate>
210 0 : HcclResult InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CalcRes(
211 : const RankGraph* rankGraph, CollAlgResReq& algResReq)
212 : {
213 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
214 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
215 0 : algResReq.topoInfo.UpdateSingleLevelTopo(virtRanks_, virtRankMap_, vTopo_);
216 :
217 0 : InsAlgTemplate tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
218 :
219 0 : AlgTempResReq tempResReq;
220 0 : if (enableDetour_) {
221 0 : HCCL_DEBUG(
222 : "[InsV2ReduceScatterVSoleExecutor][CalcRes] [%s] Rank[%d], CalcRes with detouring enabled.", __func__,
223 : myRank_);
224 0 : CHK_RET(tempAlg.CalcResDetour(rankGraph, tempResReq));
225 : } else {
226 0 : HCCL_DEBUG(
227 : "[InsV2ReduceScatterVSoleExecutor][CalcRes] [%s] Rank[%d], CalcRes with detouring disabled.", __func__,
228 : myRank_);
229 0 : CHK_RET(tempAlg.CalcRes(tempResReq));
230 : }
231 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
232 0 : algResReq.primQueueNum = tempResReq.streamNum;
233 0 : algResReq.queueNotifys = tempResReq.queNotifys;
234 0 : algResReq.localWaitGroupCntNotify = tempResReq.localWaitGroupCntNotify;
235 0 : algResReq.localBcastPostCntNotify = tempResReq.localBcastPostCntNotify;
236 0 : HCCL_INFO("[%s] Rank[%d], requiredQueNum [%u].", __func__, myRank_, algResReq.primQueueNum);
237 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
238 :
239 0 : return HcclResult::HCCL_SUCCESS;
240 0 : }
241 :
242 : template <typename AlgTopoMatch, typename InsAlgTemplate>
243 0 : HcclResult InsV2ReduceScatterVSoleExecutor<AlgTopoMatch, InsAlgTemplate>::CalcResOffload(
244 : const RankGraph* rankGraph, const u64& dataSize, CollOffloadOpResReq& resReq)
245 : {
246 : (void)dataSize;
247 : // Topo Match
248 0 : CHK_RET(InitCommInfo(rankGraph));
249 :
250 0 : std::shared_ptr<InsAlgTemplate> algTemplate = nullptr;
251 0 : CHK_RET(CreateTemplates(algTemplate));
252 :
253 0 : AlgTempResReq tempResReq;
254 0 : if (enableDetour_) {
255 0 : HCCL_DEBUG(
256 : "[InsV2ReduceScatterVSoleExecutor][CalcResOffload] [%s] Rank[%d], CalcRes with detouring enabled.",
257 : __func__, myRank_);
258 0 : CHK_RET(algTemplate->CalcResDetour(rankGraph, tempResReq));
259 : } else {
260 0 : HCCL_DEBUG(
261 : "[InsV2ReduceScatterVSoleExecutor][CalcResOffload] [%s] Rank[%d], CalcRes with detouring disabled.",
262 : __func__, myRank_);
263 0 : CHK_RET(algTemplate->CalcRes(tempResReq));
264 : }
265 0 : resReq.requiredScratchMemSize = UB_MAX_DATA_SIZE;
266 0 : resReq.requiredSubQueNum = tempResReq.streamNum - 1;
267 :
268 0 : return HcclResult::HCCL_SUCCESS;
269 0 : }
270 :
271 : #ifndef CCL_KERNEL_AICPU
272 : INS_REGISTER_IMPL_BY_TEMP(
273 : OpType::REDUCESCATTERV, CcuReduceScatterVMeshMem2Mem1D, InsV2ReduceScatterVSoleExecutor, TopoMatchMesh,
274 : CcuTempReduceScatterVMeshMem2Mem1D);
275 : INS_REGISTER_IMPL_BY_TEMP(
276 : OpType::REDUCESCATTERV, CcuReduceScatterVMesh1D, InsV2ReduceScatterVSoleExecutor, TopoMatchMesh,
277 : CcuTempReduceScatterVMesh1D);
278 : #endif
279 :
280 : } // namespace Hccl
|
