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