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