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