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 "coll_native_executor_base.h"
12 : #include "profiling_manager_pub.h"
13 : namespace hccl {
14 :
15 181 : CollNativeExecutorBase::CollNativeExecutorBase(const HcclDispatcher dispatcher,
16 181 : std::unique_ptr<TopoMatcher> &topoMatcher)
17 168 : : CollExecutorBase(dispatcher, topoMatcher), topoAttr_(topoMatcher_->GetTopoInfo()),
18 364 : algoAttr_(topoMatcher_->GetAlgoInfo()), workflowMode_(GetWorkflowMode())
19 : {
20 182 : topoType_ = topoAttr_.topoType;
21 182 : is310P3Common_ = topoAttr_.is310P3Common;
22 182 : }
23 :
24 60 : void CollNativeExecutorBase::ParseParam(const OpParam& param)
25 : {
26 60 : tag_ = param.tag;
27 60 : root_ = param.root;
28 60 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
29 60 : opType_ = param.opType;
30 60 : }
31 :
32 : // ----------------------资源计算接口----------------------
33 107 : HcclResult CollNativeExecutorBase::CalcResRequest(const OpParam& param, AlgResourceRequest& resourceRequest)
34 : {
35 107 : (void)ParseParam(param);
36 :
37 109 : u64 scratchMemSize = 0U;
38 109 : u32 streamNum = 0U;
39 109 : u32 notifyNum = 0U;
40 109 : u64 aivBufferRequest = 0U;
41 : std::vector<LevelNSubCommTransport> opTransport {
42 0 : std::vector<LevelNSubCommTransport>(static_cast<u32>(COMM_LEVEL_RESERVED))
43 109 : };
44 :
45 103 : CHK_RET(CalcScratchMemSize(scratchMemSize));
46 109 : CHK_RET(CalcOptimalIntraRing(param));
47 109 : CHK_RET(CalcStreamNum(streamNum));
48 107 : CHK_RET(CalcNotifyNum(streamNum, notifyNum));
49 109 : CHK_RET(CalcAivBufferRequest(aivBufferRequest));
50 109 : CHK_RET(CalcCommInfo(opTransport));
51 :
52 106 : CHK_RET(BuildResourceRequest(scratchMemSize, streamNum, notifyNum, aivBufferRequest, opTransport, resourceRequest));
53 109 : HCCL_INFO("streamNum[%u], notifyNum[%u], sctrachMemSize[%llu], aivBufferRequest[%llu]",
54 : resourceRequest.streamNum, resourceRequest.notifyNum, resourceRequest.scratchMemSize,
55 : resourceRequest.aivBufferRequest);
56 : // 打印建链诉求
57 109 : PrintTransportRequest(resourceRequest);
58 109 : return HCCL_SUCCESS;
59 109 : }
60 :
61 74 : HcclResult CollNativeExecutorBase::CalcScratchMemSize(u64& scratchMemSize)
62 : {
63 74 : scratchMemSize = 0U;
64 74 : HCCL_INFO("[CollNativeExecutorBase][CalcScratchMemSize]tag[%s] scratchMemSize_ is [%llu]",
65 : tag_.c_str(), scratchMemSize);
66 76 : return HCCL_SUCCESS;
67 : }
68 :
69 40 : HcclResult CollNativeExecutorBase::CalcStreamNum(u32& streamNum)
70 : {
71 : // 只传递从流数量
72 40 : streamNum = 0;
73 40 : HCCL_INFO("[CollNativeExecutorBase][CalcStreamNum]tag[%s] streamNum_ is [%u]", tag_.c_str(), streamNum);
74 40 : return HCCL_SUCCESS;
75 : }
76 :
77 112 : HcclResult CollNativeExecutorBase::CalcNotifyNum(u32 streamNum, u32 ¬ifyNum)
78 : {
79 : // notify数量是从流的两倍
80 112 : notifyNum = 2U * streamNum;
81 112 : HCCL_INFO("[CollNativeExecutorBase][CalcNotifyNum]tag[%s] notifyNum_ is [%u]", tag_.c_str(), notifyNum);
82 114 : return HCCL_SUCCESS;
83 : }
84 :
85 114 : HcclResult CollNativeExecutorBase::CalcAivBufferRequest(u64 &aivBufferRequest)
86 : {
87 114 : if (desc_.isAivMode) {
88 5 : SalSetBitOne(aivBufferRequest, ATTR_POS_AIV_COMM_BUFFER);
89 : }
90 114 : if (desc_.isAivCrossNode) {
91 0 : SalSetBitOne(aivBufferRequest, ATTR_POS_AIV_COMM_INFO_BUFFER);
92 : }
93 114 : HCCL_INFO("[CollNativeExecutorBase][CalcAivBufferRequest]tag[%s] aivBufferRequest is [%llu]", tag_.c_str(),
94 : aivBufferRequest);
95 114 : return HCCL_SUCCESS;
96 : }
97 :
98 1 : HcclResult CollNativeExecutorBase::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
99 : {
100 1 : return HCCL_SUCCESS;
101 : }
102 :
103 109 : HcclResult CollNativeExecutorBase::CalcOptimalIntraRing(const OpParam& param) {
104 109 : return HCCL_SUCCESS;
105 : }
106 :
107 0 : HcclResult CollNativeExecutorBase::SetCommInfoForARS(u32 ringSize)
108 : {
109 0 : std::vector<u32> commPlaneVector = topoMatcher_->GetCommPlaneRanks(COMM_ARS)[0];
110 0 : std::sort(commPlaneVector.begin(), commPlaneVector.end());
111 0 : u32 intraRingsize = ringSize;
112 0 : u32 userRank = topoAttr_.userRank;
113 0 : u32 userRankSize = topoAttr_.userRankSize;
114 0 : HCCL_DEBUG("[SetCommInfoForARS]set topo info for ARS, USERRANK:%u, userRankSize:%u", userRank, userRankSize);
115 :
116 0 : SetCommInfoForIntraARS(intraRingsize, commPlaneVector);
117 0 : SetCommInfoForInterARS(intraRingsize, commPlaneVector);
118 0 : topoMatcher_->SetRankMap();//一定要刷新RankMap
119 0 : HCCL_DEBUG("[SetTopoInfoForARS] outer userRank[%u] ,COMM_LEVEL0_LOGICAL total num [%d]",
120 : userRank, topoMatcher_->GetCommPlaneRanks(COMM_LEVEL0_LOGICAL).size());
121 0 : HCCL_DEBUG("[SetTopoInfoForARS] outer userRank[%u] ,COMM_LEVEL1_LOGICAL total num [%d]",
122 : userRank, topoMatcher_->GetCommPlaneRanks(COMM_LEVEL1_LOGICAL).size());
123 0 : return HCCL_SUCCESS;
124 0 : }
125 :
126 0 : HcclResult CollNativeExecutorBase::SetCommInfoForIntraARS(u32 intraRingsize, std::vector<u32> commPlaneVector)
127 : {
128 0 : std::vector<u32> comLevelARSVector = topoMatcher_->GetCommPlaneRanks(COMM_ARS)[0];
129 0 : u32 superPodRankSize = commPlaneVector.size();
130 0 : bool ringIntra = (comLevelARSVector.size() > 2 && topoAttr_.isARSDoubleRing);
131 0 : std::vector<u32> ringVectorIntra;
132 0 : for (u32 i = 0; i < superPodRankSize && ringIntra; i += intraRingsize) {
133 0 : u32 maxValue = i + intraRingsize;
134 0 : u32 rankval = topoAttr_.userRank % superPodRankSize;
135 0 : if (rankval < i || rankval >= maxValue) {
136 0 : continue;
137 : }
138 0 : for (u32 j = 0; j < intraRingsize; j++) {
139 0 : ringVectorIntra.push_back(commPlaneVector[i+j]);
140 : }
141 : }
142 0 : std::vector<std::vector<u32>> ARSmultiOuterOrder;
143 0 : std::vector<std::vector<u32>> intraRingVec;
144 0 : if (ringIntra) {
145 0 : ARSmultiOuterOrder = GetARSRingsOrder(intraRingsize, TopoType::TOPO_TYPE_NP_DOUBLE_RING, ringVectorIntra);
146 0 : for (u32 ringIndex = 0; ringIndex < ARSmultiOuterOrder.size();ringIndex++) {
147 0 : std::string outLogInfo = "userRank:";
148 0 : std::vector<u32> tmpOuterVector;
149 0 : for (u32 startIndex = 0; startIndex < ARSmultiOuterOrder[ringIndex].size();startIndex++) {
150 0 : u32 userRank = ARSmultiOuterOrder[ringIndex][startIndex];
151 0 : outLogInfo.append(std::to_string(userRank));
152 0 : outLogInfo.append("/");
153 0 : tmpOuterVector.push_back(userRank);
154 : }
155 0 : outLogInfo.append("; ");
156 0 : intraRingVec.push_back(tmpOuterVector);
157 0 : HCCL_INFO("[COMM_LEVEL0_LOGICAL]: userRank[%u], userRankSize[%u], topoRankInfo[%s]",
158 : topoAttr_.userRank, topoAttr_.userRankSize, outLogInfo.c_str());
159 0 : }
160 : } else {
161 0 : std::string outLogInfo = "userRank: ";
162 0 : std::vector<u32> tmpOuterVector;
163 0 : outLogInfo.append(std::to_string(topoAttr_.userRank));
164 0 : tmpOuterVector.push_back(topoAttr_.userRank);
165 0 : intraRingVec.push_back(tmpOuterVector);
166 0 : HCCL_INFO("[COMM_LEVEL0_LOGICAL]: userRank[%u], userRankSize[%u], topoRankInfo[%s]",
167 : topoAttr_.userRank, topoAttr_.userRankSize, outLogInfo.c_str());
168 0 : }
169 0 : topoMatcher_->EditCommPlaneVector(COMM_LEVEL0_LOGICAL, intraRingVec);
170 0 : return HCCL_SUCCESS;
171 0 : }
172 :
173 0 : HcclResult CollNativeExecutorBase::SetCommInfoForInterARS(u32 intraRingsize, std::vector<u32> commPlaneVector)
174 : {
175 0 : u32 superPodRankSize = commPlaneVector.size();
176 0 : std::vector<u32> ringVectorInter;
177 0 : std::vector<std::vector<u32>> ringVectorInterOrder;
178 0 : for (u32 i = 0; i < intraRingsize; i++) {
179 0 : ringVectorInter.clear();
180 0 : for (u32 j = 0; j < superPodRankSize; j += intraRingsize) {
181 0 : ringVectorInter.push_back(commPlaneVector[i + j]);
182 : }
183 0 : ringVectorInterOrder.push_back(ringVectorInter);
184 : }
185 0 : std::vector<std::vector<u32>> interRingVec;
186 0 : for (u32 ringIndex = 0; ringIndex < ringVectorInterOrder.size();ringIndex++) {
187 0 : std::string outLogInfo = "userRank: ";
188 0 : std::vector<u32> tmpOuterVector;
189 0 : for (u32 startIndex = 0; startIndex < ringVectorInterOrder[ringIndex].size();startIndex++) {
190 0 : u32 userRank = ringVectorInterOrder[ringIndex][startIndex];
191 0 : outLogInfo.append(std::to_string(userRank));
192 0 : outLogInfo.append("/");
193 0 : tmpOuterVector.push_back(userRank);
194 : }
195 0 : outLogInfo.append("; ");
196 0 : interRingVec.push_back(tmpOuterVector);
197 0 : HCCL_INFO("[COMM_LEVEL1_LOGICAL]:userRank[%u], userRankSize[%u], topoRankInfo[%s]",
198 : topoAttr_.userRank, topoAttr_.userRankSize, outLogInfo.c_str());
199 0 : }
200 0 : topoMatcher_->EditCommPlaneVector(COMM_LEVEL1_LOGICAL, interRingVec);
201 0 : return HCCL_SUCCESS;
202 0 : }
203 :
204 171 : HcclResult CollNativeExecutorBase::CalcCommPlaneInfo(const std::string &tag, const CommParaInfo &commParaInfo,
205 : std::vector<SingleSubCommTransport> &commTransport, TransportMemType inPutMemType,
206 : TransportMemType outPutMemType)
207 : {
208 171 : return topoMatcher_->CalcCommPlaneInfo(tag, commParaInfo, commTransport, inPutMemType, outPutMemType);
209 : }
210 :
211 47 : HcclResult CollNativeExecutorBase::CalcLevel1CommInfo(TransportMemType inputType,
212 : TransportMemType outputType,
213 : std::vector<LevelNSubCommTransport>& opTransport)
214 : {
215 47 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] start", tag_.c_str());
216 48 : u32 root = root_;
217 48 : if (opType_ == HcclCMDType::HCCL_CMD_BROADCAST && topoAttr_.superPodNum > 1) {
218 0 : root = topoMatcher_->GetSubRootWithSuperPod(topoAttr_.userRank, root_);
219 0 : HCCL_DEBUG("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] subroot is %u usrRank is %u root_ is %u",
220 : tag_.c_str(), root, topoAttr_.userRank, root_);
221 : }
222 48 : CommParaInfo commParaLevel1(COMM_LEVEL1, CommType::COMM_TAG_MAX, root);
223 :
224 48 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
225 13 : commParaLevel1.commType = CommType::COMM_TAG_RING_INNER;
226 13 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] Calc RingCommInfo", tag_.c_str());
227 35 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
228 1 : commParaLevel1.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
229 1 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] Calc NHRCommInfo", tag_.c_str());
230 34 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
231 0 : commParaLevel1.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING_V1;
232 0 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] Calc NHRV1CommInfo", tag_.c_str());
233 34 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
234 0 : commParaLevel1.commPlane = CommPlane::COMM_LEVEL1_AHC;
235 0 : commParaLevel1.commType = CommType::COMM_TAG_ASYMMETRIC_HIERARCHICAL_CONCATENATE;
236 0 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] Calc AHCCommInfo", tag_.c_str());
237 34 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
238 0 : commParaLevel1.commPlane = CommPlane::COMM_LEVEL1_AHC;
239 0 : commParaLevel1.commType = CommType::COMM_TAG_ASYMMETRIC_HIERARCHICAL_CONCATENATE_BROKE;
240 0 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] Calc AHCBrokeCommInfo", tag_.c_str());
241 34 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
242 0 : commParaLevel1.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
243 0 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] Calc NBCommInfo", tag_.c_str());
244 : } else {
245 34 : commParaLevel1.commType = CommType::COMM_TAG_HALVING_DOUBLING;
246 34 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] Calc HDCommInfo", tag_.c_str());
247 : }
248 48 : commParaLevel1.forceRdma = false;
249 48 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel1, opTransport[commParaLevel1.commPlane], inputType, outputType));
250 48 : HCCL_INFO("[CollNativeExecutorBase][COMM_LEVEL1]tag[%s] Calc CommInfo Finish", tag_.c_str());
251 :
252 48 : HCCL_INFO("[CollNativeExecutorBase][CalcLevel1CommInfo]tag[%s] Calc CommInfo Finish", tag_.c_str());
253 :
254 48 : return HCCL_SUCCESS;
255 48 : }
256 :
257 0 : HcclResult CollNativeExecutorBase::CalcLevel0CommInfo(TransportMemType inputType,
258 : TransportMemType outputType,
259 : std::vector<LevelNSubCommTransport>& opTransport)
260 : {
261 0 : return HCCL_SUCCESS;
262 : }
263 :
264 0 : HcclResult CollNativeExecutorBase::CalcLevel2CommInfo(TransportMemType inputType,
265 : TransportMemType outputType,
266 : std::vector<LevelNSubCommTransport>& opTransport)
267 : {
268 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC ||
269 0 : algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
270 0 : HCCL_INFO("[%s] select AHC bypass level2 comm calculate", __func__);
271 0 : return HCCL_SUCCESS;
272 : }
273 :
274 0 : CommParaInfo commParaLevel2(COMM_LEVEL2, CommType::COMM_TAG_MAX, root_);
275 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
276 0 : commParaLevel2.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
277 0 : HCCL_INFO("[%s] Calc NHRCommInfo", __func__);
278 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
279 0 : commParaLevel2.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
280 0 : HCCL_INFO("[%s] Calc NBCommInfo", __func__);
281 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_HD) {
282 0 : commParaLevel2.commType = CommType::COMM_TAG_HALVING_DOUBLING;
283 0 : HCCL_INFO("[%s] Calc HDCommInfo", __func__);
284 : } else {
285 0 : commParaLevel2.commType = CommType::COMM_TAG_RING_INNER;
286 0 : HCCL_INFO("[%s] Calc RingCommInfo", __func__);
287 : }
288 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel2, opTransport[COMM_LEVEL2], inputType, outputType));
289 :
290 0 : return HCCL_SUCCESS;
291 0 : }
292 :
293 109 : HcclResult CollNativeExecutorBase::PrintTransportRequest(AlgResourceRequest& resourceRequest)
294 : {
295 1840 : for (u32 levelIndex = 0; levelIndex < COMM_LEVEL_RESERVED; levelIndex++) {
296 1732 : LevelNSubCommTransport &levelTransport = resourceRequest.opTransport[levelIndex];
297 1732 : u32 ringSize = levelTransport.size();
298 2042 : for (u32 ringIndex = 0; ringIndex < ringSize; ringIndex++) {
299 311 : SingleSubCommTransport &subCommTransport = levelTransport[ringIndex];
300 312 : u32 rankSize = subCommTransport.transportRequests.size();
301 954 : for (u32 rankIndex = 0; rankIndex < rankSize; rankIndex++) {
302 642 : if (subCommTransport.transportRequests[rankIndex].isValid == true) {
303 364 : HCCL_INFO("[CollNativeExecutorBase][PrintTransportRequest]" \
304 : "levelIndex[%u], ringIndex[%u], rankIndex[%u], userRank[%u], remoteRank[%u], isUsedRdma[%d]",
305 : levelIndex, ringIndex, rankIndex, subCommTransport.transportRequests[rankIndex].localUserRank,
306 : subCommTransport.transportRequests[rankIndex].remoteUserRank,
307 : subCommTransport.transportRequests[rankIndex].isUsedRdma);
308 : }
309 : }
310 : }
311 : }
312 108 : return HCCL_SUCCESS;
313 : }
314 : // ----------------------算法编排接口----------------------
315 0 : HcclResult CollNativeExecutorBase::KernelRun(const OpParam ¶m, ExecMem &execMem)
316 : {
317 0 : HCCL_WARNING("[CollNativeExecutorBase][KernelRun]Using the default kernel run, nothing is done.");
318 0 : return HCCL_SUCCESS;
319 : }
320 :
321 61 : HcclResult CollNativeExecutorBase::ActiveSlaveStreams(const Stream &stream)
322 : {
323 61 : HcclResult ret = HCCL_SUCCESS;
324 61 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) { // offline
325 6 : for (u32 streamIndex = 0; streamIndex < algResResp_->slaveStreams.size(); streamIndex++) {
326 0 : ret = StreamActiveManager::GetInstance(topoAttr_.deviceLogicId).StreamActive(
327 0 : algResResp_->slaveStreams[streamIndex].ptr(), stream.ptr());
328 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
329 : HCCL_ERROR("[CollNativeExecutorBase][ActiveSlaveStreams]tag[%s], stream[%u] active failed,return[%d]",
330 : tag_.c_str(), streamIndex, ret), ret);
331 : }
332 : }
333 61 : return ret;
334 : }
335 :
336 0 : HcclResult CollNativeExecutorBase::AddSubStreamToProfiling()
337 : {
338 : #ifndef OPEN_HCCL_TEST
339 0 : if (((workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) &&
340 0 : !hccl::ProfilingManagerPub::GetAdditionInfoState() &&
341 0 : !hccl::ProfilingManagerPub::GetTaskApiState() &&
342 0 : !hccl::ProfilingManagerPub::GetThreadCaptureStatus())) {
343 0 : return HCCL_SUCCESS;
344 : }
345 :
346 0 : for (u32 streamIndex = 0; streamIndex < algResResp_->slaveStreams.size(); streamIndex++) {
347 : // profiling加入从环的stream
348 0 : HCCL_PROFILER_ADD_STREAM_BY_STREAMID(algResResp_->slaveStreams[streamIndex].id(), tag_, streamIndex + 1, algType_);
349 : }
350 : #endif
351 0 : return HCCL_SUCCESS;
352 : }
353 :
354 :
355 297 : HcclResult CollNativeExecutorBase::CheckCommSize(const CommPlane levelIndex, const u32 subLevelIndex)
356 : {
357 297 : if (algResResp_->opTransportResponse[levelIndex].size() < subLevelIndex) {
358 0 : HCCL_ERROR("[CollNativeExecutorBase][CheckCommSize]tag[%s], levelIndex[%u], " \
359 : "ring size[%zu] is less than expected[%u]",
360 : tag_.c_str(), levelIndex, algResResp_->opTransportResponse[levelIndex].size(), subLevelIndex);
361 0 : return HCCL_E_INTERNAL;
362 : }
363 297 : return HCCL_SUCCESS;
364 : }
365 :
366 366 : SubCommInfo CollNativeExecutorBase::GetSubCommInfo(const CommPlane levelIndex, const u32 subLevelIndex)
367 : {
368 366 : SubCommInfo info;
369 : SingleSubCommTransport &transportInfo =
370 366 : const_cast<SingleSubCommTransport&>(algResResp_->opTransportResponse[levelIndex][subLevelIndex]);
371 366 : info.localRank = transportInfo.userRank2subCommRank[topoAttr_.userRank];
372 366 : info.localRankSize = transportInfo.transportRequests.size();
373 366 : info.links = transportInfo.links;
374 366 : info.virtualLinks = transportInfo.virtualLinks;
375 366 : return info;
376 0 : }
377 :
378 110 : HcclResult CollNativeExecutorBase::BuildResourceRequest(u64 scratchMemSize, u32 streamNum, u32 notifyNum,
379 : u64 aivBufferRequest, std::vector<LevelNSubCommTransport>& opTransport,
380 : AlgResourceRequest& resourceRequest)
381 : {
382 110 : resourceRequest.scratchMemSize = scratchMemSize;
383 110 : resourceRequest.streamNum = streamNum;
384 110 : resourceRequest.notifyNum = notifyNum;
385 110 : resourceRequest.aivBufferRequest = aivBufferRequest;
386 110 : resourceRequest.opTransport = opTransport;
387 114 : return HCCL_SUCCESS;
388 : }
389 :
390 1 : HcclResult CollNativeExecutorBase::GetRankByUserRank(CommPlane levelIndex, u32 subLevelIndex, u32 userRank, u32 &rank)
391 : {
392 1 : CHK_RET(CheckCommSize(levelIndex, subLevelIndex + 1));
393 : SingleSubCommTransport &transportInfo =
394 1 : const_cast<SingleSubCommTransport&>(algResResp_->opTransportResponse[levelIndex][subLevelIndex]);
395 1 : rank = transportInfo.userRank2subCommRank[userRank];
396 1 : HCCL_DEBUG("[GetRankByUserRank]levelIndex[%u] subLevelIndex[%u], userRank[%u], rank[%u]",
397 : levelIndex, subLevelIndex, userRank, rank);
398 1 : return HCCL_SUCCESS;
399 : }
400 :
401 32 : HcclResult CollNativeExecutorBase::GetUserRankByRank(CommPlane levelIndex, u32 subLevelIndex, u32 rank, u32 &userRank)
402 : {
403 32 : CHK_RET(CheckCommSize(levelIndex, subLevelIndex + 1));
404 : SingleSubCommTransport &transportInfo =
405 32 : const_cast<SingleSubCommTransport&>(algResResp_->opTransportResponse[levelIndex][subLevelIndex]);
406 32 : userRank = transportInfo.subCommRank2UserRank[rank];
407 32 : return HCCL_SUCCESS;
408 : }
409 :
410 0 : HcclResult CollNativeExecutorBase::GenerateStreams(PrepareData &prepareData, std::vector<Stream> &streams)
411 : {
412 : // 主流 + 从流
413 0 : std::vector<Stream> substreams = *prepareData.subStreamsPtr;
414 0 : u32 streamIndexNum = substreams.size() + 1;
415 0 : u32 index = 0;
416 0 : for (u32 streamIndex = 0; streamIndex < streamIndexNum; streamIndex++) {
417 0 : if (streamIndex == 0) {
418 0 : streams.push_back(prepareData.stream);
419 : } else {
420 0 : streams.push_back(substreams[index]);
421 0 : index++;
422 : }
423 : }
424 0 : return HCCL_SUCCESS;
425 0 : }
426 :
427 0 : HcclResult CollNativeExecutorBase::NotifySubStreamStart(
428 : Stream &stream,
429 : std::vector<Stream> &substreams,
430 : std::vector<std::shared_ptr<LocalNotify>> &signalsSubToMain,
431 : u32 substreamNum)
432 : {
433 0 : for (u32 streamIndex = 0; streamIndex < substreamNum; streamIndex++) {
434 0 : CHK_RET(LocalNotify::Post(stream, dispatcher_, signalsSubToMain[streamIndex],
435 : INVALID_VALUE_STAGE));
436 0 : CHK_RET(LocalNotify::Wait(substreams[streamIndex], dispatcher_, signalsSubToMain[streamIndex],
437 : INVALID_VALUE_STAGE));
438 : }
439 0 : return HCCL_SUCCESS;
440 : }
441 :
442 0 : HcclResult CollNativeExecutorBase::WaitSubStreamFinish(
443 : Stream &stream,
444 : std::vector<Stream> &substreams,
445 : std::vector<std::shared_ptr<LocalNotify>> &signalsMainToSub,
446 : u32 substreamNum)
447 : {
448 0 : for (u32 streamIndex = 0; streamIndex < substreamNum; streamIndex++) {
449 0 : CHK_RET(LocalNotify::Post(substreams[streamIndex], dispatcher_, signalsMainToSub[streamIndex],
450 : INVALID_VALUE_STAGE));
451 0 : CHK_RET(LocalNotify::Wait(stream, dispatcher_, signalsMainToSub[streamIndex],
452 : INVALID_VALUE_STAGE));
453 : }
454 0 : return HCCL_SUCCESS;
455 : }
456 :
457 0 : HcclResult CollNativeExecutorBase::GenerateRecordWaitStreams(
458 : std::vector<Stream> &streams,
459 : u32 recordStreamNum, u32 waitStreamNum,
460 : std::vector<Stream> &recordStreams, std::vector<Stream> &waitStreams)
461 : {
462 : // 生成 record wait Streams
463 0 : for (u32 i = 0; i < recordStreamNum; i++) {
464 0 : recordStreams.push_back(streams[i]);
465 : }
466 0 : for (u32 i = recordStreamNum; i < recordStreamNum + waitStreamNum; i++) {
467 0 : waitStreams.push_back(streams[i]);
468 : }
469 0 : return HCCL_SUCCESS;
470 : }
471 :
472 0 : HcclResult CollNativeExecutorBase::HoldAllRanksOnCurrentOp(
473 : const OpParam ¶m, ExecMem &execMem, PrepareData &prepareData, std::vector<LINK> links)
474 : {
475 : (void) param;
476 0 : u32 subStreamsNum = (*prepareData.subStreamsPtr).size();
477 0 : u32 signalNum = (*prepareData.signalPtr).size();
478 0 : u32 signalAuxNum = (*prepareData.signalAuxPtr).size();
479 0 : std::vector<Stream> substreams = *prepareData.subStreamsPtr;
480 0 : std::vector<std::shared_ptr<LocalNotify>> signalsMainToSub = *prepareData.signalPtr;
481 0 : std::vector<std::shared_ptr<LocalNotify>> signalsSubToMain = *prepareData.signalAuxPtr;
482 : // 校验数据是否对齐
483 0 : if (subStreamsNum != signalNum || subStreamsNum != signalAuxNum) {
484 0 : HCCL_ERROR("[CollNativeExecutorBase][HoldAllRanksOnCurrentOp] The subStreamsNum[%u] != signalNum[%u] or "
485 : "subStreamsNum[%u] != signalAuxNum[%u]", subStreamsNum, signalNum, subStreamsNum, signalAuxNum);
486 0 : return HCCL_E_PARA;
487 : }
488 :
489 0 : std::vector<Stream> streams;
490 0 : CHK_RET(GenerateStreams(prepareData, streams));
491 : // 支持Record和Wait信号分stream排版布
492 0 : u32 recordStreamNum = (subStreamsNum + 1) / 2; // 2代表均分所有流
493 0 : u32 waitStreamNum = (subStreamsNum + 1) / 2; // 2代表均分所有流
494 0 : std::vector<Stream> recordStreams;
495 0 : std::vector<Stream> waitStreams;
496 0 : CHK_RET(GenerateRecordWaitStreams(
497 : streams, recordStreamNum, waitStreamNum, recordStreams, waitStreams));
498 :
499 : // 主流record从流
500 0 : u32 neededSubstreamNum = recordStreamNum + waitStreamNum - 1;
501 0 : CHK_RET(NotifySubStreamStart(prepareData.stream, substreams, signalsSubToMain, neededSubstreamNum));
502 :
503 0 : u32 recordIndex = 0;
504 0 : u32 waitIndex = 0;
505 : // 防止某一个rank在link未通的情况下继续执行下一个算子
506 0 : for (size_t i = 0; i < links.size(); i++) {
507 0 : if (links[i] == nullptr) {
508 0 : HCCL_DEBUG("[CollNativeExecutorBase][HoldAllRanksOnCurrentOp]links[%zu] == nullptr.", i);
509 0 : continue;
510 : }
511 0 : HCCL_INFO("[CollNativeExecutorBase][HoldAllRanksOnCurrentOp]links[%zu]. recordIndex[%u], waitIndex[%u], "
512 : "recordStreams.size()[%zu], waitStreams.size()[%zu]",
513 : i, recordIndex, waitIndex, recordStreams.size(), waitStreams.size());
514 0 : CHK_RET(links[i]->TxAck(recordStreams[recordIndex]));
515 0 : CHK_RET(links[i]->RxAck(waitStreams[waitIndex]));
516 0 : recordIndex = (recordIndex + 1) % recordStreams.size();
517 0 : waitIndex = (waitIndex + 1) % waitStreams.size();
518 : }
519 0 : CHK_RET(WaitSubStreamFinish(prepareData.stream, substreams, signalsMainToSub, neededSubstreamNum));
520 0 : CHK_RET(NotifySubStreamStart(prepareData.stream, substreams, signalsSubToMain, neededSubstreamNum));
521 0 : recordIndex = 0;
522 0 : waitIndex = 0;
523 0 : for (size_t i = 0; i < links.size(); i++) {
524 0 : if (links[i] == nullptr) {
525 0 : HCCL_DEBUG("[CollNativeExecutorBase][HoldAllRanksOnCurrentOp]links[%zu] == nullptr.", i);
526 0 : continue;
527 : }
528 0 : u64 size = std::min(execMem.inputMem.size(), HCCL_POST_SYNC_MEMCOPY_SIZE); // 传128K数据量占满所有端口
529 0 : HCCL_INFO("[CollNativeExecutorBase][HoldAllRanksOnCurrentOp]links[%zu] start to memcopy data [%llu]B.", i, size);
530 0 : CHK_RET(links[i]->TxAsync(UserMemType::INPUT_MEM, 0, execMem.inputMem.ptr(), size, recordStreams[recordIndex]));
531 0 : CHK_RET(links[i]->RxAsync(UserMemType::INPUT_MEM, 0, execMem.inputMem.ptr(), size, waitStreams[waitIndex]));
532 0 : HCCL_INFO("[CollNativeExecutorBase][HoldAllRanksOnCurrentOp]links[%zu]. recordIndex[%u], waitIndex[%u], "
533 : "recordStreams.size()[%zu], waitStreams.size()[%zu]",
534 : i, recordIndex, waitIndex, recordStreams.size(), waitStreams.size());
535 0 : CHK_RET(links[i]->PostFinAck(recordStreams[recordIndex]));
536 0 : CHK_RET(links[i]->WaitFinAck(waitStreams[waitIndex]));
537 0 : recordIndex = (recordIndex + 1) % recordStreams.size();
538 0 : waitIndex = (waitIndex + 1) % waitStreams.size();
539 : }
540 : // 从流record主流
541 0 : CHK_RET(WaitSubStreamFinish(prepareData.stream, substreams, signalsMainToSub, neededSubstreamNum));
542 0 : CHK_RET(LaunchTaskExtend(dispatcher_, prepareData.stream, substreams));
543 0 : return HCCL_SUCCESS;
544 0 : }
545 :
546 0 : HcclResult CollNativeExecutorBase::HoldAllRanksOnCurrentOpWithSingleStream(
547 : OpParam ¶m, ExecMem &execMem, std::vector<LINK> links)
548 : {
549 : // 防止某一个rank在link未通的情况下继续执行下一个算子
550 0 : for (size_t i = 0; i < links.size(); i++) {
551 0 : if (links[i] == nullptr) {
552 0 : HCCL_DEBUG("[CollNativeExecutorBase][HoldAllRanksOnCurrentOpWithSingleStream]links[%zu] == nullptr.", i);
553 0 : continue;
554 : }
555 0 : HCCL_INFO("[CollNativeExecutorBase][HoldAllRanksOnCurrentOpWithSingleStream]links[%zu].", i);
556 0 : CHK_RET(links[i]->TxAck(param.stream));
557 0 : CHK_RET(links[i]->RxAck(param.stream));
558 0 : u64 size = std::min(execMem.inputMem.size(), HCCL_POST_SYNC_MEMCOPY_SIZE); // 传128K数据量占满所有端口
559 0 : HCCL_INFO("[CollNativeExecutorBase][HoldAllRanksOnCurrentOpWithSingleStream]"
560 : "links[%zu] start to memcopy data [%llu]B.", i, size);
561 0 : CHK_RET(links[i]->TxAsync(UserMemType::INPUT_MEM, 0, execMem.inputMem.ptr(), size, param.stream));
562 0 : CHK_RET(links[i]->RxAsync(UserMemType::INPUT_MEM, 0, execMem.inputMem.ptr(), size, param.stream));
563 0 : CHK_RET(links[i]->PostFinAck(param.stream));
564 0 : CHK_RET(links[i]->WaitFinAck(param.stream));
565 : }
566 0 : return HCCL_SUCCESS;
567 : }
568 :
569 0 : HcclResult CollNativeExecutorBase::SendRecvSignalOnLinks(OpParam ¶m, ExecMem &execMem, std::vector<LINK> links)
570 : {
571 : // 实验结果: 算子间隔1s能够被PreSync阻拦
572 : // 收发信号校验
573 0 : for (size_t i = 0; i < links.size(); i++) {
574 0 : if (links[i] == nullptr) {
575 0 : HCCL_DEBUG("[CollNativeExecutorBase][SendRecvSignalOnLinks]links[%zu] == nullptr.", i);
576 0 : continue;
577 : }
578 0 : HCCL_INFO("[CollNativeExecutorBase][SendRecvSignalOnLinks]links[%zu].", i);
579 0 : CHK_RET(links[i]->TxAck(param.stream));
580 0 : CHK_RET(links[i]->RxAck(param.stream));
581 : }
582 : // 拷贝数据从而占满端口，才能在注入故障时在PreSync算子触发重执行
583 0 : for (size_t i = 0; i < links.size(); i++) {
584 0 : if (links[i] == nullptr) {
585 0 : HCCL_DEBUG("[CollNativeExecutorBase][SendRecvSignalOnLinks]links[%zu] == nullptr.", i);
586 0 : continue;
587 : }
588 0 : u64 size = std::min(execMem.inputMem.size(), HCCL_INPLACE_MEMCOPY_SIZE); // 传128K数据量占满所有端口
589 0 : HCCL_INFO("[CollNativeExecutorBase][SendRecvSignalOnLinks]"
590 : "links[%zu] start memcopy start to memcopy data [%llu]B.", i, size);
591 0 : CHK_RET(links[i]->TxAsync(UserMemType::INPUT_MEM, 0, execMem.inputMem.ptr(), size, param.stream));
592 0 : CHK_RET(links[i]->RxAsync(UserMemType::INPUT_MEM, 0, execMem.inputMem.ptr(), size, param.stream));
593 0 : CHK_RET(links[i]->PostFinAck(param.stream));
594 0 : CHK_RET(links[i]->WaitFinAck(param.stream));
595 : }
596 : // 防止某一个rank在link未通的情况下继续执行下一个算子
597 0 : for (size_t i = 0; i < links.size(); i++) {
598 0 : if (links[i] == nullptr) {
599 0 : HCCL_DEBUG("[CollNativeExecutorBase][SendRecvSignalOnLinks]links[%zu] == nullptr.", i);
600 0 : continue;
601 : }
602 0 : HCCL_INFO("[CollNativeExecutorBase][SendRecvSignalOnLinks]links[%zu].", i);
603 0 : CHK_RET(links[i]->TxAck(param.stream));
604 0 : CHK_RET(links[i]->RxAck(param.stream));
605 0 : CHK_RET(links[i]->TxDataSignal(param.stream));
606 0 : CHK_RET(links[i]->RxDataSignal(param.stream));
607 : }
608 0 : return HCCL_SUCCESS;
609 : }
610 :
611 0 : bool CollNativeExecutorBase::OpSyncCheckCommSize(const CommPlane levelIndex, const u32 expectedSize)
612 : {
613 0 : if (algResResp_->opTransportResponse[levelIndex].size() < expectedSize) {
614 0 : HCCL_WARNING("[CollNativeExecutorBase][CheckCommSize]tag[%s], levelIndex[%u], " \
615 : "ring size[%zu] is less than expected[%u]",
616 : tag_.c_str(), levelIndex, algResResp_->opTransportResponse[levelIndex].size(), expectedSize);
617 0 : return false;
618 : }
619 0 : return true;
620 : }
621 :
622 0 : HcclResult CollNativeExecutorBase::PostSyncWithSubstream(OpParam ¶m, ExecMem &execMem, PrepareData &prepareData)
623 : {
624 : // COMM_COMBINE_ORDER 是不是只有alltoall类算子使用? 不是，有一些打平场景也会用到
625 : // 所以需要另起新函数，用于alltoall类算子的postsync调用
626 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream] "
627 : "The op with algOpContext_.opRetryHandler.isPostSync[%d] starts.",
628 : algOpContext_.opRetryHandler.isPostSync);
629 0 : u32 level0ServerIndex = 0;
630 0 : if (OpSyncCheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1)) {
631 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
632 0 : level0ServerIndex = level0CommInfo.localRank;
633 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream]level0CommInfo.links check starts.");
634 0 : CHK_RET(HoldAllRanksOnCurrentOp(param, execMem, prepareData, level0CommInfo.links));
635 0 : }
636 0 : if (OpSyncCheckCommSize(COMM_LEVEL1, level0ServerIndex + 1)) {
637 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0ServerIndex);
638 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream]level1CommInfo.links check starts.");
639 0 : CHK_RET(HoldAllRanksOnCurrentOp(param, execMem, prepareData, level1CommInfo.links));
640 0 : }
641 0 : if (OpSyncCheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1)) {
642 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
643 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream]level2CommInfo.links check starts.");
644 0 : CHK_RET(HoldAllRanksOnCurrentOp(param, execMem, prepareData, level2CommInfo.links));
645 0 : }
646 0 : if (OpSyncCheckCommSize(COMM_LEVEL1, level0ServerIndex + 1)) {
647 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0ServerIndex);
648 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream]level1CommInfo.links check starts again.");
649 0 : CHK_RET(HoldAllRanksOnCurrentOp(param, execMem, prepareData, level1CommInfo.links));
650 0 : }
651 0 : if (OpSyncCheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1)) {
652 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
653 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream]level0CommInfo.links check starts again.");
654 0 : CHK_RET(HoldAllRanksOnCurrentOp(param, execMem, prepareData, level0CommInfo.links));
655 0 : }
656 0 : if (OpSyncCheckCommSize(COMM_COMBINE_ORDER, COMM_INDEX_0 + 1)) {
657 0 : SubCommInfo combineOrderCommInfo = GetSubCommInfo(COMM_COMBINE_ORDER, COMM_INDEX_0);
658 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream]combineOrderCommInfo.links check starts.");
659 0 : CHK_RET(HoldAllRanksOnCurrentOp(param, execMem, prepareData, combineOrderCommInfo.links));
660 0 : }
661 0 : if (OpSyncCheckCommSize(COMM_COMBINE_ORDER, COMM_INDEX_0 + 1)) {
662 0 : SubCommInfo combineOrderCommInfo = GetSubCommInfo(COMM_COMBINE_ORDER, COMM_INDEX_0);
663 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream]combineOrderCommInfo.links check starts again.");
664 0 : CHK_RET(HoldAllRanksOnCurrentOp(param, execMem, prepareData, combineOrderCommInfo.links));
665 0 : }
666 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithSubstream] "
667 : "The op with algOpContext_.opRetryHandler.isPostSync[%d] ends.",
668 : algOpContext_.opRetryHandler.isPostSync);
669 0 : return HCCL_SUCCESS;
670 : }
671 :
672 0 : HcclResult CollNativeExecutorBase::PostSyncWithoutSubstream(OpParam ¶m, ExecMem &execMem)
673 : {
674 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream] "
675 : "The op with algOpContext_.opRetryHandler.isPostSync[%d] starts.",
676 : algOpContext_.opRetryHandler.isPostSync);
677 0 : u32 level0ServerIndex = 0;
678 0 : if (OpSyncCheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1)) {
679 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
680 0 : level0ServerIndex = level0CommInfo.localRank;
681 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream]level0CommInfo.links check starts.");
682 0 : CHK_RET(HoldAllRanksOnCurrentOpWithSingleStream(param, execMem, level0CommInfo.links));
683 0 : }
684 0 : if (OpSyncCheckCommSize(COMM_LEVEL1, level0ServerIndex + 1)) {
685 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0ServerIndex);
686 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream]level1CommInfo.links check starts.");
687 0 : CHK_RET(HoldAllRanksOnCurrentOpWithSingleStream(param, execMem, level1CommInfo.links));
688 0 : }
689 0 : if (OpSyncCheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1)) {
690 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
691 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream]level2CommInfo.links check starts.");
692 0 : CHK_RET(HoldAllRanksOnCurrentOpWithSingleStream(param, execMem, level2CommInfo.links));
693 0 : }
694 0 : if (OpSyncCheckCommSize(COMM_LEVEL1, level0ServerIndex + 1)) {
695 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0ServerIndex);
696 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream]level1CommInfo.links check starts again.");
697 0 : CHK_RET(HoldAllRanksOnCurrentOpWithSingleStream(param, execMem, level1CommInfo.links));
698 0 : }
699 0 : if (OpSyncCheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1)) {
700 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
701 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream]level0CommInfo.links check starts again.");
702 0 : CHK_RET(HoldAllRanksOnCurrentOpWithSingleStream(param, execMem, level0CommInfo.links));
703 0 : }
704 0 : if (OpSyncCheckCommSize(COMM_COMBINE_ORDER, COMM_INDEX_0 + 1)) {
705 0 : SubCommInfo combineOrderCommInfo = GetSubCommInfo(COMM_COMBINE_ORDER, COMM_INDEX_0);
706 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream]combineOrderCommInfo.links check starts.");
707 0 : CHK_RET(HoldAllRanksOnCurrentOpWithSingleStream(param, execMem, combineOrderCommInfo.links));
708 0 : }
709 0 : if (OpSyncCheckCommSize(COMM_COMBINE_ORDER, COMM_INDEX_0 + 1)) {
710 0 : SubCommInfo combineOrderCommInfo = GetSubCommInfo(COMM_COMBINE_ORDER, COMM_INDEX_0);
711 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream]combineOrderCommInfo.links check starts again.");
712 0 : CHK_RET(HoldAllRanksOnCurrentOpWithSingleStream(param, execMem, combineOrderCommInfo.links));
713 0 : }
714 0 : HCCL_INFO("[CollNativeExecutorBase][PostSyncWithoutSubstream] "
715 : "The op with algOpContext_.opRetryHandler.isPostSync[%d] ends.",
716 : algOpContext_.opRetryHandler.isPostSync);
717 :
718 0 : CHK_RET(LaunchTaskExtend(dispatcher_,
719 : const_cast<Stream &>(param.stream),
720 : const_cast<std::vector<Stream> &>(algResResp_->slaveStreams)));
721 :
722 0 : return HCCL_SUCCESS;
723 : }
724 :
725 0 : HcclResult CollNativeExecutorBase::InplaceOpSync(OpParam ¶m, ExecMem &execMem)
726 : {
727 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync] The op with algOpContext_.opRetryHandler.isInplacePreSync[%d] "
728 : "or algOpContext_.opRetryHandler.isPostSync[%d] starts.",
729 : algOpContext_.opRetryHandler.isInplacePreSync, algOpContext_.opRetryHandler.isPostSync);
730 0 : u32 level0ServerIndex = 0;
731 0 : if (OpSyncCheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1)) {
732 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
733 0 : level0ServerIndex = level0CommInfo.localRank;
734 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync]level0CommInfo.links check starts.");
735 0 : CHK_RET(SendRecvSignalOnLinks(param, execMem, level0CommInfo.links));
736 0 : }
737 0 : if (OpSyncCheckCommSize(COMM_LEVEL1, level0ServerIndex + 1)) {
738 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0ServerIndex);
739 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync]level1CommInfo.links check starts.");
740 0 : CHK_RET(SendRecvSignalOnLinks(param, execMem, level1CommInfo.links));
741 0 : }
742 0 : if (OpSyncCheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1)) {
743 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
744 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync]level2CommInfo.links check starts.");
745 0 : CHK_RET(SendRecvSignalOnLinks(param, execMem, level2CommInfo.links));
746 0 : }
747 0 : if (OpSyncCheckCommSize(COMM_LEVEL1, level0ServerIndex + 1)) {
748 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0ServerIndex);
749 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync]level1CommInfo.links check starts again.");
750 0 : CHK_RET(SendRecvSignalOnLinks(param, execMem, level1CommInfo.links));
751 0 : }
752 0 : if (OpSyncCheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1)) {
753 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
754 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync]level0CommInfo.links check starts again.");
755 0 : CHK_RET(SendRecvSignalOnLinks(param, execMem, level0CommInfo.links));
756 0 : }
757 : // alltoall-like opType
758 0 : if (OpSyncCheckCommSize(COMM_COMBINE_ORDER, COMM_INDEX_0 + 1)) {
759 0 : SubCommInfo combineOrderCommInfo = GetSubCommInfo(COMM_COMBINE_ORDER, COMM_INDEX_0);
760 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync]combineOrderCommInfo.links check starts.");
761 0 : CHK_RET(SendRecvSignalOnLinks(param, execMem, combineOrderCommInfo.links));
762 0 : }
763 0 : if (OpSyncCheckCommSize(COMM_COMBINE_ORDER, COMM_INDEX_0 + 1)) {
764 0 : SubCommInfo combineOrderCommInfo = GetSubCommInfo(COMM_COMBINE_ORDER, COMM_INDEX_0);
765 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync]combineOrderCommInfo.links check starts again.");
766 0 : CHK_RET(SendRecvSignalOnLinks(param, execMem, combineOrderCommInfo.links));
767 0 : }
768 0 : HCCL_INFO("[CollNativeExecutorBase][InplaceOpSync] The op with algOpContext_.opRetryHandler.isInplacePreSync[%d] "
769 : "or algOpContext_.opRetryHandler.isPostSync[%d] ends.",
770 : algOpContext_.opRetryHandler.isInplacePreSync, algOpContext_.opRetryHandler.isPostSync);
771 :
772 0 : CHK_RET(LaunchTaskExtend(dispatcher_,
773 : const_cast<Stream &>(param.stream),
774 : const_cast<std::vector<Stream> &>(algResResp_->slaveStreams)));
775 :
776 0 : return HCCL_SUCCESS;
777 : }
778 :
779 0 : std::vector<std::vector<u32>> GetARSRingsOrder(u32 ranksSize, TopoType topoType, std::vector<u32> &RingList)
780 : {
781 0 : std::vector<std::vector<u32>> ARSmultiRingOrder;
782 0 : std::vector<u32> tmpOuter0 = RingList; // 环0
783 0 : if (topoType == TopoType::TOPO_TYPE_NP_DOUBLE_RING && ranksSize > FACTOR_TWO ) { //两环
784 0 : std::vector<u32> tmpOuter1; // 环1
785 0 : tmpOuter1.reserve(ranksSize);
786 0 : tmpOuter1.push_back(RingList[0]);
787 0 : tmpOuter1.insert(tmpOuter1.end(), tmpOuter0.rbegin(), tmpOuter0.rend() - 1);
788 0 : ARSmultiRingOrder.push_back(tmpOuter0);
789 0 : ARSmultiRingOrder.push_back(tmpOuter1);
790 0 : } else {
791 0 : ARSmultiRingOrder.push_back(tmpOuter0);
792 : }
793 0 : return ARSmultiRingOrder;
794 0 : }
795 :
796 0 : HcclResult CollNativeExecutorBase::CopyAivCommInfoToDevice(const CommPlane levelIndex, const u32 subLevelIndex,
797 : AlgResourceResponse& algResource)
798 : {
799 0 : algResResp_ = &algResource;
800 0 : CHK_RET(CheckCommSize(levelIndex, subLevelIndex + 1));
801 0 : SubCommInfo commInfo = GetSubCommInfo(levelIndex, subLevelIndex);
802 0 : u32 localRank = commInfo.localRank;
803 0 : u32 localRankSize = commInfo.localRankSize;
804 :
805 0 : void* buffersInOut[MAX_RANK_SIZE_A3 * 2] = {};
806 0 : bool isOpbaseMode = GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
807 :
808 0 : for (u32 i = 0; i < localRankSize; i++) {
809 0 : u32 idx = (i << 1);
810 0 : if (i != localRank) {
811 0 : CHK_RET(commInfo.links[i]->GetRemoteMem(UserMemType::INPUT_MEM, &(buffersInOut[idx])));
812 0 : CHK_RET(commInfo.links[i]->GetRemoteMem(UserMemType::OUTPUT_MEM, &(buffersInOut[idx + 1])));
813 : } else {
814 0 : buffersInOut[idx] = isOpbaseMode ? algResource.cclInputMem.ptr() : algResource.paramInputMem.ptr();
815 0 : buffersInOut[idx + 1] = algResource.aivOutputMem.ptr();
816 : }
817 : }
818 0 : const u32 bufferNum = 2;
819 0 : CHK_RET(hrtMemSyncCopy(algResource.aivCommInfoMem.ptr(), sizeof(u64) * localRankSize * bufferNum,
820 : buffersInOut, sizeof(u64) * localRankSize * bufferNum, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
821 0 : return HCCL_SUCCESS;
822 0 : }
823 :
824 0 : HcclResult CollNativeExecutorBase::Getlevel1CommRank(SubCommInfo& level1CommInfo)
825 : {
826 : (void) level1CommInfo;
827 0 : return HCCL_SUCCESS;
828 : }
829 0 : HcclResult CollNativeExecutorBase::SelectTempAlg(std::unique_ptr<AlgTemplateBase> &level1TempAlg, u32 level1RankSize)
830 : {
831 : (void) level1TempAlg;
832 : (void) level1RankSize;
833 0 : return HCCL_SUCCESS;
834 : }
835 0 : HcclResult CollNativeExecutorBase::GetDevNumInlocalPod(u32& devNumInlocalPod)
836 : {
837 : (void) devNumInlocalPod;
838 0 : return HCCL_SUCCESS;
839 : }
840 :
841 0 : HcclResult CollNativeExecutorBase::SetOpCache(const AivOpArgs& opArgs, const AivTopoArgs& topoArgs, const AivResourceArgs& resourceArgs,
842 : const AivAlgArgs& algArgs, ExtraArgs& extraArgs, AivProfilingInfo& aivProfilingInfo, bool isA3CrossNode)
843 : {
844 0 : cacheInfo_.opArgs = opArgs;
845 0 : cacheInfo_.topoArgs = topoArgs;
846 0 : cacheInfo_.resourceArgs = resourceArgs;
847 0 : cacheInfo_.algArgs = algArgs;
848 0 : cacheInfo_.profilingInfo = aivProfilingInfo;
849 0 : cacheInfo_.extraArgs = extraArgs;
850 0 : cacheInfo_.isUseCache = true;
851 :
852 0 : if (isA3CrossNode) {
853 0 : u8 buffersOutSize = 2 * sizeof(void *);
854 0 : CHK_SAFETY_FUNC_RET(memcpy_s(cacheInfo_.buffersIn, sizeof(void *), resourceArgs.buffersIn, sizeof(void *)));
855 0 : CHK_SAFETY_FUNC_RET(memcpy_s(cacheInfo_.buffersOut, buffersOutSize, resourceArgs.buffersOut, buffersOutSize));
856 : } else {
857 0 : u64 bufferInfoSize = sizeof(void *) * topoArgs.rankSize;
858 0 : CHK_SAFETY_FUNC_RET(memcpy_s(cacheInfo_.buffersIn, bufferInfoSize, resourceArgs.buffersIn, bufferInfoSize));
859 0 : CHK_SAFETY_FUNC_RET(memcpy_s(cacheInfo_.buffersOut, bufferInfoSize, resourceArgs.buffersOut, bufferInfoSize));
860 : }
861 :
862 0 : HCCL_INFO("[CollNativeExecutorBase][SetOpCache] cmdType:%d, count:%llu, dataType:%d, op:%d, " \
863 : "rank:%u, rankSize:%u, serverNum:%u, isA3CrossNode:%d, buffersIn:%p, buffersOut:%p", opArgs.cmdType, opArgs.count, opArgs.dataType,
864 : opArgs.op, topoArgs.rank, topoArgs.rankSize, topoArgs.serverNum, isA3CrossNode,
865 : cacheInfo_.buffersIn, cacheInfo_.buffersOut);
866 :
867 0 : return HCCL_SUCCESS;
868 : }
869 :
870 0 : void CollNativeExecutorBase::SetWorkflowMode(HcclWorkflowMode workflowMode)
871 : {
872 0 : workflowMode_ = workflowMode;
873 0 : }
874 :
875 0 : HcclWorkflowMode CollNativeExecutorBase::GetExecutorWorkflowMode() const
876 : {
877 0 : return workflowMode_;
878 : }
879 : }
|
