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 <cmath>
12 : #include <algorithm>
13 : #include "stream_active_manager.h"
14 : #include "profiling_manager_pub.h"
15 : #include "coll_alg_utils.h"
16 : #include "sal_pub.h"
17 : #include "hccl_impl.h"
18 :
19 : using namespace std;
20 :
21 : namespace hccl {
22 :
23 : std::array<DeviceMem, MAX_MODULE_DEVICE_NUM> hcclImpl::inOutPutTempMem_;
24 : std::array<std::mutex, MAX_MODULE_DEVICE_NUM> hcclImpl::inOutPutTempMemMutex_;
25 : std::array<Referenced, MAX_MODULE_DEVICE_NUM> hcclImpl::instanceRef_;
26 : RegisterToHeartBeatCallBack g_RegisterToHeartBeatCallBack = nullptr;
27 : UnRegisterToHeartBeatCallBack g_UnRegisterToHeartBeatCallBack = nullptr;
28 : SetRankPortInfoCallBack g_SetRankPortInfoCallBack = nullptr;
29 :
30 523 : hcclImpl::hcclImpl(
31 : const HcclDispatcher dispatcher, const std::unique_ptr<NotifyPool>& notifyPool,
32 : std::map<HcclIpAddress, HcclNetDevCtx>& netDevCtxMap, const std::unique_ptr<QueueNotifyManager>& queueNotifyManager,
33 : std::unique_ptr<WorkspaceResource>& workSpaceRes, CCLBufferManager& cclBufferManager, HcclAlgoAttr& algoAttr,
34 : HcclTopoAttr& topoAttr, std::shared_ptr<AlgConfigurator> algConfigurator,
35 523 : std::shared_ptr<TopoInfoExtractor> topoInfoEx)
36 523 : : dispatcher_(dispatcher),
37 523 : notifyPool_(notifyPool),
38 523 : netDevCtxMap_(netDevCtxMap),
39 523 : queueNotifyManager_(queueNotifyManager),
40 523 : workSpaceRes_(workSpaceRes),
41 523 : cclBufferManager_(cclBufferManager),
42 523 : algConfigurator_(algConfigurator),
43 523 : topoInfoEx_(topoInfoEx),
44 523 : topoAttr_(topoAttr),
45 523 : algoAttr_(algoAttr)
46 : {
47 523 : SetAlgoAttr(algoAttr);
48 523 : SetTopoAttr(topoAttr);
49 :
50 523 : s32 deviceLogicId = 0;
51 523 : if (hrtGetDevice(&deviceLogicId) != HCCL_SUCCESS) {
52 0 : HCCL_INFO("start hccl resources build:no get deviceLogicId[%d]", deviceLogicId);
53 0 : return;
54 : }
55 523 : if ((static_cast<u32>(deviceLogicId) >= MAX_MODULE_DEVICE_NUM) || (deviceLogicId < 0)) {
56 0 : HCCL_WARNING("start hccl resources build:get fail deviceLogicId[%d]", deviceLogicId);
57 0 : return;
58 : }
59 :
60 523 : HCCL_INFO("start hccl resources build:get deviceLogicId[%d]", deviceLogicId_);
61 523 : instanceRef_[deviceLogicId].Ref();
62 523 : if (SalGetBareTgid(&pid_) != HCCL_SUCCESS) {
63 0 : HCCL_INFO("get pid is unsuccessful");
64 0 : return;
65 : }
66 0 : }
67 :
68 1040 : hcclImpl::~hcclImpl()
69 : {
70 523 : HCCL_INFO("start hccl resources destruction:deviceLogicId[%d]", deviceLogicId_);
71 :
72 521 : WaitCommThread(commThreadPtrLevel0_);
73 520 : WaitCommThread(commThreadPtrLevel1_);
74 520 : WaitCommThread(commThreadPtrLevel2_);
75 :
76 : /* 销毁通信域关联资源 */
77 520 : for (auto& iter : tagCommInfo_) {
78 0 : DestroyLevel0Comm(iter.first);
79 0 : DestroyLevel1Comm(iter.first);
80 0 : DestroyIntraServerComm(iter.first);
81 : // Workspace资源需要根据tag销毁（临时方案）
82 0 : workSpaceRes_->DestroyWorkspaceResource(iter.first);
83 : }
84 :
85 522 : cclBufferManager_.ReleaseAlltoAllvParaBuffer();
86 :
87 521 : for (auto& level1_stream_info : tagStreamInfo_) {
88 0 : if (ReleaseSignal(level1_stream_info.second) != HCCL_SUCCESS) {
89 0 : HCCL_WARNING("tag[%s],signal is not released successfully", level1_stream_info.first.c_str());
90 : }
91 0 : (void)StreamActiveManager::GetInstance(deviceLogicId_).StreamsUnactive(level1_stream_info.second.ringStreams);
92 : }
93 :
94 519 : tagCommInfo_.clear();
95 520 : tagStreamInfo_.clear();
96 522 : commMeshPtr_.reset();
97 519 : commMeshLevel2_.reset();
98 519 : commMeshMap_.clear();
99 :
100 521 : commFactory_ = nullptr;
101 :
102 523 : if ((static_cast<u32>(deviceLogicId_) >= MAX_MODULE_DEVICE_NUM) || (deviceLogicId_ < 0)) {
103 0 : HCCL_WARNING("start hccl resources destruction:get fail deviceLogicId[%d]", deviceLogicId_);
104 0 : return;
105 : }
106 :
107 523 : if (instanceRef_[deviceLogicId_].Unref() == 0) {
108 520 : std::unique_lock<std::mutex> lock(inOutPutTempMemMutex_[deviceLogicId_]);
109 514 : inOutPutTempMem_[deviceLogicId_].free();
110 516 : }
111 519 : }
112 :
113 523 : void hcclImpl::SetAlgoAttr(HcclAlgoAttr& algoAttr)
114 : {
115 523 : isHaveCpuRank_ = algoAttr.isHaveCpuRank;
116 523 : inlineReduceSwitchOn_ = algoAttr.inlineReduceSwitchOn;
117 523 : isUsedRdmaLevel0_ = algoAttr.isUsedRdmaLevel0;
118 523 : isUsedInterHccsMode_ = algoAttr.isUsedInterHccsMode;
119 :
120 523 : identifier_ = algoAttr.identifier;
121 523 : collectiveId_ = algoAttr.collectiveId;
122 :
123 523 : nicDeployment_ = algoAttr.nicDeployment;
124 523 : commWorkMode_ = algoAttr.commWorkMode;
125 523 : return;
126 : }
127 :
128 523 : void hcclImpl::SetTopoAttr(HcclTopoAttr& topoAttr)
129 : {
130 523 : serverNum_ = topoAttr.serverNum;
131 523 : superPodNum_ = topoAttr.superPodNum;
132 523 : moduleNum_ = topoAttr.moduleNum;
133 523 : deviceNumPerServer_ = topoAttr.deviceNumPerServer;
134 523 : deviceNumPerAggregation_ = topoAttr.deviceNumPerAggregation;
135 523 : multiModuleDiffDeviceNumMode_ = topoAttr.multiModuleDiffDeviceNumMode;
136 523 : multiSuperPodDiffServerNumMode_ = topoAttr.multiSuperPodDiffServerNumMode;
137 523 : multiSuperPodDiffDeviceNumMode_ = topoAttr.multiSuperPodDiffDeviceNumMode;
138 :
139 523 : meshAggregationRankSize_ = topoAttr.meshAggregationRankSize;
140 523 : isDiffDeviceModule_ = topoAttr.isDiffDeviceModule;
141 523 : isSingleMeshAggregation_ = topoAttr.isSingleMeshAggregation;
142 523 : isAllRankSamePlane_ = topoAttr.isAllRankSamePlane;
143 :
144 523 : userRank_ = topoAttr.userRank;
145 523 : realUserRank_ = topoAttr.realUserRank;
146 523 : userRankSize_ = topoAttr.userRankSize;
147 523 : rankInfoList_ = topoAttr.rankInfoList;
148 :
149 523 : devicePhyId_ = topoAttr.devicePhyId;
150 523 : deviceLogicId_ = topoAttr.deviceLogicId;
151 523 : useSuperPodMode_ = topoAttr.useSuperPodMode;
152 523 : deviceType_ = topoAttr.deviceType;
153 523 : isStandardCard_ = topoAttr.isStandardCard;
154 523 : is310PDuoCard_ = topoAttr.is310PDuoCard;
155 :
156 523 : nicList_ = topoAttr.nicList;
157 523 : pairLinkCounter_ = topoAttr.pairLinkCounter;
158 523 : pairLinkInfo_ = topoAttr.pairLinkInfo;
159 523 : isSupportRdmaLite_ = topoAttr.isSupportRdmaLite;
160 523 : isSupportHccsAndSio_ = topoAttr_.isSupportHccsAndSio;
161 523 : localNicPort_ = topoAttr.localNicPort;
162 523 : isNeedInitNic_ = topoAttr.isNeedInitNic;
163 523 : return;
164 : }
165 :
166 523 : HcclResult hcclImpl::Init(bool isHeterogComm)
167 : {
168 523 : algConfigurator_->GetTopoType(topoType_);
169 :
170 523 : commFactory_.reset(new (std::nothrow) CommFactory(
171 1046 : identifier_, userRank_, userRankSize_, dispatcher_, notifyPool_, netDevCtxMap_, topoInfoEx_, isUsedRdmaLevel0_,
172 523 : topoType_, deviceType_, rankInfoList_, nicDeployment_, isHeterogComm, meshAggregationRankSize_, isHaveCpuRank_,
173 1046 : isUsedInterHccsMode_, useSuperPodMode_));
174 523 : CHK_SMART_PTR_NULL(commFactory_);
175 523 : CHK_RET(commFactory_->Init());
176 :
177 523 : HCCL_INFO("hcclImpl init success.");
178 523 : return HCCL_SUCCESS;
179 : }
180 :
181 0 : HcclResult hcclImpl::ReleaseCommInfos()
182 : {
183 0 : auto iter = tagCommInfo_.begin();
184 0 : while (iter != tagCommInfo_.end()) {
185 0 : for (auto& comm : iter->second.commLevel1) {
186 0 : if (comm != nullptr) {
187 0 : CHK_RET(comm->DeInit());
188 : }
189 : }
190 0 : iter++;
191 : }
192 0 : return HCCL_SUCCESS;
193 : }
194 :
195 0 : HcclResult hcclImpl::CreateP2pComm(const std::string& tag, CommInfo& commInfo, DeviceMem& inOutMem, u32 peerUserRank)
196 : {
197 0 : CommParaInfo commP2P(COMM_COMBINE, CommType::COMM_TAG_P2P);
198 0 : commP2P.peerUserRank = peerUserRank;
199 0 : CHK_RET(commFactory_->CreateCommPlane(tag, inOutMem, inOutMem, commP2P, commInfo.commP2P));
200 0 : return HCCL_SUCCESS;
201 0 : }
202 :
203 1575 : HcclResult hcclImpl::WaitCommThread(std::unique_ptr<std::thread>& ThreadPtr) const
204 : {
205 : // 若线程指针为空，为此线程从未被拉起使能，不返回异常日志
206 1575 : if (ThreadPtr != nullptr && ThreadPtr->joinable()) {
207 12 : ThreadPtr->join(); // 等待线程执行完毕
208 12 : CHK_RET(hrtResetDevice(deviceLogicId_)); // 防止线程里面异常退出，在进程中reset
209 : }
210 1571 : return HCCL_SUCCESS;
211 : }
212 :
213 1 : HcclResult hcclImpl::InitMultiStreamResource(
214 : const std::string& tag, level1StreamInfo_t& streamInfo, AlgType algType, bool isAicpuModeEn, bool isBatchSendRecv,
215 : u32 ringNum)
216 : {
217 1 : if (!isBatchSendRecv) {
218 1 : if (algType.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_NP_SINGLE_RING) {
219 1 : if (deviceType_ == DevType::DEV_TYPE_910_93) {
220 0 : if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
221 0 : streamInfo.ringNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
222 : } else {
223 0 : streamInfo.ringNum = LEVEL0_PLANE_NUM_IN_NPRING_SINGLE;
224 : }
225 : }
226 0 : } else if (algType.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_NP_DOUBLE_RING) {
227 0 : if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
228 0 : streamInfo.ringNum = LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
229 : } else {
230 0 : streamInfo.ringNum = LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE;
231 : }
232 0 : } else if (algType.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_8P_RING) {
233 0 : streamInfo.ringNum = LEVEL0_PLANE_NUM_IN_8PRING;
234 0 : } else if (algType.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_NP_MESH) {
235 0 : if ((GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE)
236 0 : && (deviceType_ == DevType::DEV_TYPE_910B) && isSingleMeshAggregation_) {
237 0 : streamInfo.ringNum = deviceNumPerAggregation_;
238 0 : } else if ((deviceType_ == DevType::DEV_TYPE_910_93) && (isAicpuModeEn == true)) {
239 0 : streamInfo.ringNum = deviceNumPerAggregation_;
240 0 : } else if (
241 0 : (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE)
242 0 : && (deviceType_ == DevType::DEV_TYPE_910B)
243 0 : && algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_PIPELINE) {
244 0 : streamInfo.ringNum = deviceNumPerAggregation_ + 1; /* pipeline ring场景下性能优化 */
245 : } else {
246 0 : streamInfo.ringNum = deviceNumPerAggregation_ - 1;
247 : }
248 0 : } else if (algType.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_4P_MESH) {
249 0 : streamInfo.ringNum = LEVEL0_PLANE_NUM_IN_4PMESH;
250 : }
251 : } else {
252 : // 批量send/recv需要2条流
253 0 : streamInfo.ringNum = 2;
254 : }
255 :
256 1 : if (piplineSliceNum_ > 0) {
257 0 : streamInfo.ringNum++; // 流水并行算法, Server间需要额外一条从流
258 : }
259 1 : streamInfo.ringNum = std::max(streamInfo.ringNum, ringNum);
260 1 : HCCL_INFO("algType:[%u] InitMultiStreamResource streamInfo.ringNum %u", algType.algoLevel0, streamInfo.ringNum);
261 1 : if (streamInfo.ringNum > 1) {
262 0 : u32 resNum = streamInfo.ringNum - 1;
263 0 : streamInfo.ringStreams.resize(resNum); // 只有主环以外会用,减去主环1
264 0 : streamInfo.ringSignal.resize(resNum); // 只有主环以外会用,减去主环1
265 0 : streamInfo.ringSignalAux.resize(resNum); // 只有主环以外会用,减去主环1
266 0 : streamInfo.ringThreadsManage.resize(resNum);
267 0 : streamInfo.tidInfo.resize(resNum);
268 :
269 0 : for (auto& signal : streamInfo.ringSignal) {
270 0 : signal = nullptr;
271 : }
272 0 : for (auto& signal : streamInfo.ringSignalAux) {
273 0 : signal = nullptr;
274 : }
275 :
276 0 : u32 notifyNum = resNum * 2; // 2:Signal + SignalAux
277 0 : std::vector<std::shared_ptr<LocalNotify>> notifys(notifyNum, nullptr);
278 0 : CHK_RET(queueNotifyManager_->Alloc(tag, notifyNum, notifys));
279 0 : for (u32 i = 0; i < resNum; i++) {
280 0 : streamInfo.ringSignal[i] = notifys[2 * i];
281 0 : streamInfo.ringSignalAux[i] = notifys[2 * i + 1];
282 : }
283 0 : for (u32 ringIndex = 0; ringIndex < resNum; ringIndex++) {
284 0 : if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
285 0 : streamInfo.ringThreadsManage[ringIndex].reset(new (std::nothrow)
286 0 : ThreadManage(deviceLogicId_, userRank_, dispatcher_));
287 0 : CHK_SMART_PTR_NULL(streamInfo.ringThreadsManage[ringIndex]);
288 0 : HcclResult ret = streamInfo.ringThreadsManage[ringIndex]->Init();
289 0 : CHK_PRT_RET(
290 : ret != HCCL_SUCCESS,
291 : HCCL_ERROR("[Init][MultiRingResource]ringIndex[%u] ThreadManage failed,return[%d]", ringIndex, ret),
292 : ret);
293 0 : streamInfo.tidInfo[ringIndex] = streamInfo.ringThreadsManage[ringIndex]->GetTid();
294 0 : HCCL_INFO("ringThreadsManage Init success[%u]", ringIndex);
295 : }
296 : }
297 0 : }
298 1 : if (isAicpuModeEn == true) {
299 1 : HCCL_INFO("aicpu resource num[%u]", streamInfo.ringNum);
300 1 : streamInfo.ringDeviceStreams.resize(streamInfo.ringNum);
301 :
302 1 : if (streamInfo.ringNum > 1) {
303 0 : u32 resNum = streamInfo.ringNum - 1;
304 0 : streamInfo.ringDeviceSignal.resize(resNum);
305 0 : streamInfo.ringDeviceSignalAux.resize(resNum);
306 :
307 0 : for (auto& signal : streamInfo.ringDeviceSignal) {
308 0 : signal = nullptr;
309 : }
310 :
311 0 : for (auto& signal : streamInfo.ringDeviceSignalAux) {
312 0 : signal = nullptr;
313 : }
314 :
315 0 : u32 notifyNum = resNum * 2; // 2:Signal + SignalAux
316 0 : std::vector<std::shared_ptr<LocalNotify>> notifys(notifyNum, nullptr);
317 0 : CHK_RET(queueNotifyManager_->Alloc(tag, notifyNum, notifys, NotifyLoadType::DEVICE_NOTIFY));
318 0 : for (u32 i = 0; i < resNum; i++) {
319 0 : streamInfo.ringDeviceSignal[i] = notifys[2 * i];
320 0 : streamInfo.ringDeviceSignalAux[i] = notifys[2 * i + 1];
321 : }
322 0 : }
323 : }
324 1 : return HCCL_SUCCESS;
325 : }
326 :
327 0 : HcclResult hcclImpl::ReplaceCommInfoByTag(const std::string& tag, std::unique_ptr<CommInfo>& commInfo)
328 : {
329 0 : std::unique_lock<std::mutex> replLock(commLock_);
330 0 : tagCommInfo_.erase(tag);
331 0 : tagCommInfo_.insert(std::pair<std::string, CommInfo>(tag, std::move(*commInfo)));
332 0 : return HCCL_SUCCESS;
333 0 : }
334 :
335 1 : HcclResult hcclImpl::CreateComm(
336 : const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem, AlgType algType,
337 : std::unique_ptr<CommInfo>& commInfo, u32 root, bool isP2p, bool isAicpuModeEn, bool isBatchSendRecv,
338 : bool meshSinglePlane, bool aivMode, std::set<u32> batchSendRecvtargetRanks)
339 : {
340 : (void)batchSendRecvtargetRanks;
341 : (void)isBatchSendRecv;
342 : // Comm资源的唯一性，由上层调用保证
343 : // tag 多线程并行调度时唯一标识，不能为空
344 1 : CHK_PRT_RET(
345 : tag.empty(), HCCL_ERROR("[Create][Comm]errNo[0x%016llx] tag is empty", HCCL_ERROR_CODE(HCCL_E_PARA)),
346 : HCCL_E_PARA);
347 :
348 1 : commInfo.reset(new (std::nothrow) CommInfo);
349 1 : CHK_SMART_PTR_NULL(commInfo);
350 :
351 1 : DeviceMem inputMemComm(inputMem);
352 1 : DeviceMem outputMemComm(outputMem);
353 1 : DeviceMem expMemComm = cclBufferManager_.GetCommCCLBuffer();
354 1 : if (!isHaveCpuRank_) {
355 1 : inputMemComm = cclBufferManager_.GetCommRegMem(inputMem, MemAttr::IN_CCL_BUFFER, aivMode);
356 1 : outputMemComm = cclBufferManager_.GetCommRegMem(outputMem, MemAttr::OUT_CCL_BUFFER, aivMode);
357 : }
358 :
359 1 : if (isP2p) {
360 0 : CHK_RET(CreateP2pComm(tag, *commInfo, inputMemComm, root));
361 1 : } else if (isAicpuModeEn && deviceType_ == DevType::DEV_TYPE_910_93) {
362 : // level0 mesh通信域
363 0 : std::vector<std::unique_ptr<CommBase>> commMeshL0;
364 0 : CommParaInfo commCombinePara(COMM_MESH_L0, CommType::COMM_TAG_MESH);
365 0 : commCombinePara.isAicpuModeEn = isAicpuModeEn;
366 0 : CHK_RET(commFactory_->CreateCommPlane(tag, inputMemComm, outputMemComm, commCombinePara, commInfo->commLevel0));
367 0 : } else {
368 1 : bool isA2MC2MultiServer = false;
369 1 : const std::string& suffix = HCCL_MC2_MULTISERVER_SUFFIX;
370 1 : if (tag.size() > suffix.size() && tag.compare(tag.size() - suffix.size(), suffix.size(), suffix) == 0) {
371 0 : isA2MC2MultiServer = true;
372 : }
373 1 : CHK_RET(CreateCommByAlg(
374 : tag, algType, *commInfo, inputMemComm, outputMemComm, expMemComm, root, isAicpuModeEn, meshSinglePlane,
375 : isA2MC2MultiServer));
376 : }
377 :
378 1 : return HCCL_SUCCESS;
379 1 : }
380 :
381 1 : HcclResult hcclImpl::CreateComm(
382 : const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem, AlgType algType, u32 root, bool isP2p,
383 : bool isBatchSendRecv, bool meshSinglePlane, bool aivMode, std::set<u32> batchSendRecvtargetRanks)
384 : {
385 : // tag 多线程并行调度时唯一标识，不能为空
386 1 : CHK_PRT_RET(
387 : tag.empty(), HCCL_ERROR("[Create][Comm]errNo[0x%016llx] tag is empty", HCCL_ERROR_CODE(HCCL_E_PARA)),
388 : HCCL_E_PARA);
389 :
390 : // 作下重复的判断，在Gather等逻辑梳理清楚后，再清理
391 0 : CHK_PRT_RET(
392 : IsExistCommRes(tag), HCCL_DEBUG("[HcclImpl][CreateComm] tag[%s] comm has existed, do nothing", tag.c_str()),
393 : HCCL_SUCCESS);
394 :
395 0 : std::unique_ptr<CommInfo> commInfo = nullptr;
396 0 : HcclResult ret = CreateComm(
397 : tag, inputMem, outputMem, algType, commInfo, root, isP2p, false, isBatchSendRecv, meshSinglePlane, aivMode,
398 : batchSendRecvtargetRanks);
399 0 : CHK_PRT_RET(
400 : ret != HCCL_SUCCESS,
401 : HCCL_ERROR("[hcclImpl][CreateComm]create comminfo by tag[%s] failed. return[%d]", tag.c_str(), ret), ret);
402 :
403 : // 根据上下层逻辑，这里其实只是Save/Insert。
404 0 : CHK_RET(ReplaceCommInfoByTag(tag, commInfo));
405 0 : return HCCL_SUCCESS;
406 0 : }
407 :
408 7 : HcclResult hcclImpl::GetCommTypeInLevel0(const AlgType algType, const TopoType topoType, CommType& commType)
409 : {
410 7 : if (Is310P3Common(isHaveCpuRank_, deviceType_)) {
411 0 : if (algType.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_NP_HD) {
412 0 : commType = CommType::COMM_TAG_HALVING_DOUBLING;
413 : } else {
414 0 : commType = CommType::COMM_TAG_RING_INNER;
415 : }
416 0 : HCCL_DEBUG(
417 : "[Get][CommTypeForLevel0]The algType is %s, topoType is %d, while commType is %d",
418 : AlgTypeToStr(algType).c_str(), topoType, commType);
419 0 : return HCCL_SUCCESS;
420 : }
421 :
422 7 : bool isMesh
423 7 : = ((topoType_ == TopoType::TOPO_TYPE_4P_MESH) || (topoType_ == TopoType::TOPO_TYPE_2P_MESH)
424 14 : || (topoType_ == TopoType::TOPO_TYPE_1P_MESH) || (topoType_ == TopoType::TOPO_TYPE_NP_MESH));
425 :
426 : // 根据算法类型创建内层拓扑
427 7 : if (algType.algoLevel0 == AlgTypeLevel0::ALG_LEVEL0_NP_STAR) {
428 0 : commType = CommType::COMM_TAG_STAR;
429 7 : } else if (isMesh) {
430 0 : commType = CommType::COMM_TAG_MESH;
431 : } else {
432 7 : commType = CommType::COMM_TAG_RING_INNER;
433 : }
434 7 : HCCL_DEBUG(
435 : "[Get][CommTypeForLevel0]The algType is %s, topoType is %d, while commType is %d",
436 : AlgTypeToStr(algType).c_str(), topoType, commType);
437 7 : return HCCL_SUCCESS;
438 : }
439 :
440 7 : HcclResult hcclImpl::GetCommTypeInLevel1(const AlgType algType, CommType& commType)
441 : {
442 : // 根据算法类型创建内层拓扑
443 7 : if (algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_WHOLE_RING) {
444 0 : commType = CommType::COMM_TAG_RING_COMBINED;
445 7 : } else if (algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
446 0 : commType = CommType::COMM_TAG_HALVING_DOUBLING;
447 : /* pipeline ring场景下性能优化 */
448 7 : } else if (
449 7 : algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_PIPELINE
450 7 : || algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
451 1 : commType = CommType::COMM_TAG_RING_INNER;
452 6 : } else if (algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_STAR) {
453 1 : commType = CommType::COMM_TAG_STAR;
454 5 : } else if (algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
455 1 : if (algType.algoLevel0 != AlgTypeLevel0::ALG_LEVEL0_RESERVED) {
456 0 : commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
457 : } else {
458 1 : commType = CommType::COMM_TAG_WHOLE_NHR;
459 : }
460 4 : } else if (algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR_V1) {
461 1 : if (algType.algoLevel0 != AlgTypeLevel0::ALG_LEVEL0_RESERVED) {
462 0 : commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING_V1;
463 : } else {
464 1 : commType = CommType::COMM_TAG_WHOLE_NHR_V1;
465 : }
466 3 : } else if (algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC) {
467 0 : if (algType.algoLevel0 != AlgTypeLevel0::ALG_LEVEL0_RESERVED) {
468 0 : commType = CommType::COMM_TAG_ASYMMETRIC_HIERARCHICAL_CONCATENATE;
469 : } else {
470 0 : commType = CommType::COMM_TAG_WHOLE_AHC;
471 : }
472 3 : } else if (algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
473 1 : if (algType.algoLevel0 != AlgTypeLevel0::ALG_LEVEL0_RESERVED) {
474 0 : commType = CommType::COMM_TAG_ASYMMETRIC_HIERARCHICAL_CONCATENATE_BROKE;
475 : } else {
476 1 : commType = CommType::COMM_TAG_WHOLE_AHC_BROKE;
477 : }
478 2 : } else if (algType.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
479 1 : if (algType.algoLevel0 != AlgTypeLevel0::ALG_LEVEL0_RESERVED) {
480 0 : commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
481 : } else {
482 1 : commType = CommType::COMM_TAG_WHOLE_NB;
483 : }
484 : } else {
485 1 : HCCL_ERROR("[Get][CommTypeInLevel1]algType[%s] is not support", AlgTypeToStr(algType).c_str());
486 1 : return HCCL_E_PARA;
487 : }
488 6 : HCCL_DEBUG(
489 : "[Get][CommTypeInLevel1]The algType is %s, while commType is %d", AlgTypeToStr(algType).c_str(), commType);
490 6 : return HCCL_SUCCESS;
491 : }
492 :
493 6 : CommPlane hcclImpl::GetCommPlaneInLevel1(CommType& commType)
494 : {
495 : CommPlane commPlane;
496 6 : switch (commType) {
497 0 : case CommType::COMM_TAG_RING_COMBINED: {
498 0 : commPlane = COMM_COMBINE;
499 0 : break;
500 : }
501 :
502 3 : case CommType::COMM_TAG_WHOLE_NB:
503 : case CommType::COMM_TAG_WHOLE_NHR:
504 : case CommType::COMM_TAG_WHOLE_NHR_V1:
505 : case CommType::COMM_TAG_MESH_COMBINED: {
506 3 : commPlane = COMM_COMBINE_ORDER;
507 3 : break;
508 : }
509 :
510 3 : default: {
511 3 : commPlane = COMM_LEVEL1;
512 3 : break;
513 : }
514 : }
515 6 : HCCL_DEBUG("[Get][CommPlaneInLevel1]The commType is %d, commPlane is %d", commType, commPlane);
516 6 : return commPlane;
517 : }
518 :
519 7 : HcclResult hcclImpl::CreateCommByAlg(
520 : const std::string& tag, const AlgType algType, CommInfo& commInfo, DeviceMem& inputMem, DeviceMem& outputMem,
521 : DeviceMem& expMem, u32 root, bool isAicpuModeEn, bool meshSinglePlane, bool isA2MC2MultiServer)
522 : {
523 7 : CHK_RET(algConfigurator_->CheckAlgType(algType));
524 7 : CHK_RET(commFactory_->SetHDCModeInfo(
525 : rankDevicePhyIdNicInfoMap_, ranksPort_, vnicRanksPort_, isSetHDCModeInfo_, isUseRankPort_));
526 :
527 7 : HcclResult commThreadWaitResultLevel0 = HCCL_SUCCESS;
528 7 : HcclResult commThreadWaitResultLevel0Rdma = HCCL_SUCCESS;
529 7 : HcclResult commThreadWaitResultLevel1 = HCCL_SUCCESS;
530 7 : HcclResult commThreadWaitResultLevel1Rdma = HCCL_SUCCESS;
531 7 : HcclResult commThreadWaitResultLevel2 = HCCL_SUCCESS;
532 :
533 7 : workflowMode_ = GetWorkflowMode(); // 后续会起新线程，因此更新workflowMode
534 : /* Level0通信域 */
535 : CommType commTypeInLevel0;
536 7 : HcclResult commThreadResultLevel0 = HCCL_SUCCESS;
537 7 : HcclResult commThreadResultLevel0Rdma = HCCL_SUCCESS;
538 7 : CHK_RET(GetCommTypeInLevel0(algType, topoType_, commTypeInLevel0));
539 7 : bool isUsedRdma = false;
540 7 : if (isA2MC2MultiServer) {
541 0 : HCCL_INFO("commInfo create commLevel0Rdma/commLevel1Rdma for EnableRdmaSdma start");
542 0 : isUsedRdma = true;
543 : }
544 :
545 7 : if (Is310P3Common(isHaveCpuRank_, deviceType_)) {
546 0 : if (isAicpuModeEn) {
547 0 : commTypeInLevel0 = CommType::COMM_TAG_MESH;
548 : }
549 : // level0 通信域
550 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, commTypeInLevel0);
551 0 : commParaLevel0.isAicpuModeEn = isAicpuModeEn;
552 0 : std::vector<std::unique_ptr<CommBase>> commVec;
553 0 : CHK_RET(commFactory_->CreateCommPlane(tag, inputMem, outputMem, commParaLevel0, commVec));
554 :
555 0 : CHK_PRT_RET(
556 : commVec.empty() || !commVec[0],
557 : HCCL_ERROR(
558 : "[Create][CommIntraServer]errNo[0x%016llx] tag[%s], created commIntraServer fail.",
559 : HCCL_ERROR_CODE(HCCL_E_NOT_FOUND), tag.c_str()),
560 : HCCL_E_NOT_FOUND);
561 0 : commInfo.commIntraServer = std::move(commVec[0]);
562 0 : return HCCL_SUCCESS;
563 0 : }
564 : CommParaInfo commInfoLevel0(
565 7 : COMM_LEVEL0, commTypeInLevel0, root, INVALID_VALUE_RANKID, isAicpuModeEn, meshSinglePlane);
566 : // default、whole_nhr和whole_nb算法不创建外层拓扑
567 7 : if (algType.algoLevel0 != AlgTypeLevel0::ALG_LEVEL0_WHOLE_RING
568 6 : && algType.algoLevel0 != AlgTypeLevel0::ALG_LEVEL0_RESERVED && !isA2MC2MultiServer) {
569 1 : commThreadPtrLevel0_.reset(new (std::nothrow) std::thread(
570 1 : &hcclImpl::CreateCommThread, this, hrtErrMGetErrorContextPub(), std::ref(tag), std::ref(inputMem),
571 1 : std::ref(outputMem), std::ref(expMem), std::ref(commInfoLevel0), std::ref(commInfo.commLevel0),
572 2 : std::ref(commThreadResultLevel0)));
573 1 : CHK_PRT_RET(
574 : !commThreadPtrLevel0_,
575 : HCCL_ERROR("[Create][CommByAlg]commTypeInLevel0[%d] threads reset failed.", commInfoLevel0.commType),
576 : HCCL_E_INTERNAL);
577 1 : commThreadWaitResultLevel0 = WaitCommThread(commThreadPtrLevel0_);
578 1 : if (isUsedRdma) {
579 0 : commInfoLevel0.forceRdma = isUsedRdma;
580 0 : commThreadPtrLevel0Rdma_.reset(new (std::nothrow) std::thread(
581 0 : &hcclImpl::CreateCommThread, this, hrtErrMGetErrorContextPub(), std::ref(tag), std::ref(inputMem),
582 0 : std::ref(outputMem), std::ref(expMem), std::ref(commInfoLevel0), std::ref(commInfo.commLevel0Rdma),
583 0 : std::ref(commThreadResultLevel0Rdma)));
584 0 : CHK_PRT_RET(
585 : !commThreadPtrLevel0Rdma_,
586 : HCCL_ERROR(
587 : "[Create][CommByAlg]commTypeInLevel0[%d]"
588 : " commLevel0Rdma threads reset failed.",
589 : commInfoLevel0.commType),
590 : HCCL_E_INTERNAL);
591 0 : commThreadWaitResultLevel0Rdma = WaitCommThread(commThreadPtrLevel0Rdma_);
592 : }
593 : }
594 :
595 : /* Level1通信域 */
596 7 : HcclResult commThreadResultLevel1 = HCCL_SUCCESS;
597 7 : HcclResult commThreadResultLevel1Rdma = HCCL_SUCCESS;
598 : CommType commTypeInLevel1;
599 7 : CHK_RET(GetCommTypeInLevel1(algType, commTypeInLevel1));
600 6 : if (isA2MC2MultiServer) {
601 0 : commTypeInLevel1 = CommType::COMM_TAG_MESH_COMBINED;
602 : }
603 :
604 6 : CommPlane commPlaneInLevel1 = GetCommPlaneInLevel1(commTypeInLevel1);
605 6 : CommParaInfo commInfoLevel1(commPlaneInLevel1, commTypeInLevel1, root, INVALID_VALUE_RANKID, isAicpuModeEn);
606 6 : if (commTypeInLevel1 != CommType::COMM_TAG_STAR) {
607 5 : if (!isA2MC2MultiServer) {
608 5 : commThreadPtrLevel1_.reset(new (std::nothrow) std::thread(
609 5 : &hcclImpl::CreateCommThread, this, hrtErrMGetErrorContextPub(), std::ref(tag), std::ref(inputMem),
610 5 : std::ref(outputMem), std::ref(expMem), std::ref(commInfoLevel1), std::ref(commInfo.commLevel1),
611 10 : std::ref(commThreadResultLevel1)));
612 5 : CHK_PRT_RET(
613 : !commThreadPtrLevel1_,
614 : HCCL_ERROR("[Create][CommByAlg]commTypeInLevel1[%d] threads reset failed.", commInfoLevel1.commType),
615 : HCCL_E_INTERNAL);
616 5 : commThreadWaitResultLevel1 = WaitCommThread(commThreadPtrLevel1_);
617 : }
618 :
619 5 : if (isUsedRdma) {
620 0 : commInfoLevel1.forceRdma = isUsedRdma;
621 0 : commThreadPtrLevel1Rdma_.reset(new (std::nothrow) std::thread(
622 0 : &hcclImpl::CreateCommThread, this, hrtErrMGetErrorContextPub(), std::ref(tag), std::ref(inputMem),
623 0 : std::ref(outputMem), std::ref(expMem), std::ref(commInfoLevel1), std::ref(commInfo.commLevel1Rdma),
624 0 : std::ref(commThreadResultLevel1Rdma)));
625 0 : CHK_PRT_RET(
626 : !commThreadPtrLevel1Rdma_,
627 : HCCL_ERROR(
628 : "[Create][CommByAlg]commTypeInLevel1[%d]"
629 : " commLevel1Rdma threads reset failed.",
630 : commInfoLevel1.commType),
631 : HCCL_E_INTERNAL);
632 0 : commThreadWaitResultLevel1Rdma = WaitCommThread(commThreadPtrLevel1Rdma_);
633 : }
634 : }
635 :
636 : /* Level2通信域 */
637 6 : HcclResult commThreadResultLevel2 = HCCL_SUCCESS;
638 6 : CommParaInfo commInfoLevel2(COMM_LEVEL2, CommType::COMM_TAG_RING_INNER);
639 6 : commThreadPtrLevel2_.reset(new (std::nothrow) std::thread(
640 6 : &hcclImpl::CreateCommThread, this, hrtErrMGetErrorContextPub(), std::ref(tag), std::ref(inputMem),
641 6 : std::ref(outputMem), std::ref(expMem), std::ref(commInfoLevel2), std::ref(commInfo.commLevel2),
642 12 : std::ref(commThreadResultLevel2)));
643 6 : CHK_PRT_RET(
644 : !commThreadPtrLevel2_,
645 : HCCL_ERROR("[Create][CommByAlg]commTypeInLevel2[%d] threads reset failed.", commInfoLevel2.commType),
646 : HCCL_E_INTERNAL);
647 6 : commThreadWaitResultLevel2 = WaitCommThread(commThreadPtrLevel2_);
648 :
649 6 : CHK_PRT_RET(
650 : static_cast<bool>(commThreadWaitResultLevel0) || static_cast<bool>(commThreadWaitResultLevel1)
651 : || static_cast<bool>(commThreadWaitResultLevel2) || static_cast<bool>(commThreadWaitResultLevel0Rdma)
652 : || static_cast<bool>(commThreadWaitResultLevel1Rdma),
653 : HCCL_ERROR(
654 : "[Create][CommByAlg]wait thread failed.algoLevel0[%d] Level1[%d] Level2[%d] Level0rdma[%d]"
655 : " Level1rdma[%d]",
656 : commThreadWaitResultLevel0, commThreadWaitResultLevel1, commThreadWaitResultLevel2,
657 : commThreadWaitResultLevel0Rdma, commThreadWaitResultLevel1Rdma),
658 : HCCL_E_INTERNAL);
659 :
660 6 : CHK_PRT_RET(
661 : static_cast<bool>(commThreadResultLevel0) || static_cast<bool>(commThreadResultLevel1)
662 : || static_cast<bool>(commThreadResultLevel2) || static_cast<bool>(commThreadResultLevel0Rdma)
663 : || static_cast<bool>(commThreadResultLevel1Rdma),
664 : HCCL_ERROR(
665 : "[Create][CommByAlg]CreateComm failed. result: Level0[%d] Level1[%d] Level2[%d]"
666 : " Level0rdma[%d] Level1rdma[%d].",
667 : commThreadResultLevel0, commThreadResultLevel1, commThreadResultLevel2, commThreadResultLevel0Rdma,
668 : commThreadResultLevel1Rdma),
669 : HCCL_E_INTERNAL);
670 2 : return HCCL_SUCCESS;
671 7 : }
672 :
673 14 : HcclResult hcclImpl::CreateCommThread(
674 : const ErrContextPub& error_context, const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem,
675 : DeviceMem& expMem, const CommParaInfo& commParaInfo, std::vector<std::unique_ptr<CommBase>>& commVec,
676 : HcclResult& retOut)
677 : {
678 : // 给当前线程添加名字
679 14 : SetThreadName("Hccl_CreateComm");
680 :
681 14 : hrtErrMSetErrorContextPub(error_context);
682 14 : retOut = hrtSetDevice(deviceLogicId_);
683 14 : CHK_PRT_RET(
684 : retOut != HCCL_SUCCESS, HCCL_ERROR("[Create][CommThread]set device[%d] failed", deviceLogicId_), retOut);
685 14 : SetWorkflowMode(workflowMode_);
686 :
687 14 : retOut = commFactory_->CreateCommPlane(tag, inputMem, outputMem, commParaInfo, commVec, expMem);
688 14 : CHK_PRT_RET(
689 : retOut != HCCL_SUCCESS,
690 : HCCL_ERROR(
691 : "[Create][CommThread]tag[%s], create comm level[%d] commType[%d] fail", tag.c_str(), commParaInfo.commPlane,
692 : commParaInfo.commType),
693 : retOut);
694 :
695 9 : return HCCL_SUCCESS;
696 : }
697 :
698 1 : HcclResult hcclImpl::CreateMutiStreamRes(
699 : const std::string& tag, Stream& stream, level1StreamInfo_t& streamInfo, AlgType algType, bool isAicpuModeEn,
700 : bool isBatchSendRecv, u32 ringNum)
701 : {
702 : /* 多环资源初始化 */
703 1 : HcclResult ret = InitMultiStreamResource(tag, streamInfo, algType, isAicpuModeEn, isBatchSendRecv, ringNum);
704 1 : CHK_PRT_RET(
705 : ret != HCCL_SUCCESS,
706 : HCCL_ERROR("[Create][MutiStreamRes]tag[%s] init multi ring resource failed, return[%d]", tag.c_str(), ret),
707 : ret);
708 :
709 1 : CHK_RET(hccl::ProfilingManagerPub::CallMsprofReportMultiThreadInfo(streamInfo.tidInfo));
710 :
711 1 : if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
712 : // GE OffloadStreamManager中set的流都是从流
713 0 : CHK_RET(workSpaceRes_->RegisterMaster(tag, stream));
714 0 : streamInfo.ringStreams = workSpaceRes_->AllocSlaveStreams(tag, streamInfo.ringNum - 1);
715 1 : } else if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
716 1 : CHK_RET(opBaseStreamManager_.RegisterMaster(stream));
717 : streamInfo.ringStreams
718 1 : = opBaseStreamManager_.AllocSlaves(StreamType::STREAM_TYPE_ONLINE, streamInfo.ringNum - 1);
719 :
720 1 : if (isAicpuModeEn == true) {
721 1 : if (auxRingStreamsDev_.empty()) {
722 1 : auxRingStreamsDev_.reserve(MAX_SUBSTREAM_NUM + 1);
723 1 : HCCL_DEBUG("CreateMutiStreamRes: reserve auxRingStreamsDev_[%u]", MAX_SUBSTREAM_NUM);
724 : }
725 1 : if (auxRingStreamsDev_.size() < streamInfo.ringNum) {
726 1 : HCCL_DEBUG(
727 : "CreateMutiStreamRes:tag[%s], auxRingStreamsDev_.size[%u], less than [%u], need create new streams",
728 : tag.c_str(), auxRingStreamsDev_.size(), streamInfo.ringNum);
729 1 : CHK_PRT_RET(
730 : streamInfo.ringNum > MAX_SUBSTREAM_NUM + 1,
731 : HCCL_ERROR(
732 : "[Create][MutiStreamRes]tag[%s] streamInfo.ringNum[%u] is larger than MAX_SUBSTREAM_NUM+1[%u].",
733 : tag.c_str(), streamInfo.ringNum, MAX_SUBSTREAM_NUM + 1),
734 : HCCL_E_INTERNAL);
735 1 : u32 ringNum = auxRingStreamsDev_.size();
736 2 : for (u32 ringIndex = ringNum; ringIndex < streamInfo.ringNum; ringIndex++) {
737 1 : auxRingStreamsDev_.emplace_back(Stream(StreamType::STREAM_TYPE_DEVICE));
738 : // 给device侧申请的流不需要setmode，否则rts会捕获流成员Flags为1024的异常
739 : }
740 : }
741 2 : for (u32 ringIndex = 0; ringIndex < streamInfo.ringNum; ringIndex++) {
742 1 : streamInfo.ringDeviceStreams[ringIndex] = auxRingStreamsDev_[ringIndex];
743 1 : CHK_SMART_PTR_NULL(streamInfo.ringDeviceStreams[ringIndex]);
744 : }
745 : }
746 : } else {
747 0 : HCCL_ERROR("[Create][MutiStreamRes]WorkflowMode[%d] invalid", GetWorkflowMode());
748 0 : return HCCL_E_INTERNAL;
749 : }
750 1 : CHK_PRT_RET(
751 : (streamInfo.ringStreams.size() != streamInfo.ringNum - 1),
752 : HCCL_ERROR(
753 : "[Create][MutiStreamRes]tag[%s] get slave stream failed, "
754 : "expect to get size [%u], but only alloc [%u].",
755 : tag.c_str(), streamInfo.ringNum - 1, streamInfo.ringStreams.size()),
756 : HCCL_E_INTERNAL);
757 :
758 1 : return HCCL_SUCCESS;
759 : }
760 :
761 0 : HcclResult hcclImpl::CreateMutiStreamRes(
762 : const std::string& tag, Stream& stream, AlgType algType, bool isBatchSendRecv, u32 ringNum)
763 : {
764 0 : std::unique_lock<std::mutex> mutiStreamLock(tagStreamInfoLock_);
765 0 : CHK_PRT_RET(
766 : tagStreamInfo_.find(tag) != tagStreamInfo_.end(),
767 : HCCL_DEBUG("[Create][MutiStreamRes]tag[%s] is already exit, do nothing", tag.c_str()), HCCL_SUCCESS);
768 :
769 0 : level1StreamInfo_t streamInfo;
770 0 : CHK_RET(CreateMutiStreamRes(tag, stream, streamInfo, algType, false, isBatchSendRecv, ringNum));
771 :
772 : // 构建线程和内部流维护关系
773 0 : tagStreamInfo_.insert(std::pair<std::string, Level1StreamInfo>(tag, std::move(streamInfo)));
774 0 : mutiStreamLock.unlock();
775 0 : HCCL_INFO("[Create][MutiStreamRes]tag[%s], ringNum[%u]", tag.c_str(), streamInfo.ringNum);
776 0 : return HCCL_SUCCESS;
777 0 : }
778 :
779 0 : void hcclImpl::DestroyLevel1Comm(const std::string& tag)
780 : {
781 : // vector成员是智能指针, 自动destroy
782 0 : tagCommInfo_t::iterator itr = tagCommInfo_.find(tag);
783 0 : if (itr != tagCommInfo_.end()) {
784 0 : itr->second.commLevel1.clear();
785 : }
786 0 : }
787 :
788 0 : void hcclImpl::DestroyLevel0Comm(const std::string& tag)
789 : {
790 : // vector成员是智能指针, 自动destroy
791 0 : tagCommInfo_t::iterator itr = tagCommInfo_.find(tag);
792 0 : if (itr != tagCommInfo_.end()) {
793 0 : itr->second.commLevel0.clear();
794 : }
795 0 : }
796 :
797 0 : void hcclImpl::DestroyIntraServerComm(const std::string& tag)
798 : {
799 0 : tagCommInfo_t::iterator itr = tagCommInfo_.find(tag);
800 0 : if (itr != tagCommInfo_.end()) {
801 0 : itr->second.commIntraServer.reset();
802 : }
803 0 : }
804 :
805 0 : HcclResult hcclImpl::ReleaseSignal(level1StreamInfo_t& level1Stream)
806 : {
807 0 : for (auto& signal : level1Stream.ringSignal) {
808 0 : if (signal != nullptr) {
809 0 : signal = nullptr;
810 : }
811 : }
812 :
813 0 : for (auto& signal : level1Stream.ringSignalAux) {
814 0 : if (signal != nullptr) {
815 0 : signal = nullptr;
816 : }
817 : }
818 :
819 0 : for (auto& signal : level1Stream.ringDeviceSignal) {
820 0 : if (signal != nullptr) {
821 0 : signal = nullptr;
822 : }
823 : }
824 :
825 0 : for (auto& signal : level1Stream.ringDeviceSignalAux) {
826 0 : if (signal != nullptr) {
827 0 : signal = nullptr;
828 : }
829 : }
830 :
831 0 : return HCCL_SUCCESS;
832 : }
833 :
834 16 : HcclResult hcclImpl::ClearOpResource(const std::string& tag)
835 : {
836 : // 链接资源释放
837 16 : commMeshMap_.erase(tag);
838 13 : tagCommInfo_.erase(tag);
839 : // stream解绑定
840 14 : auto iterStream = tagStreamInfo_.find(tag);
841 13 : if (iterStream != tagStreamInfo_.end()) {
842 0 : CHK_RET(StreamActiveManager::GetInstance(deviceLogicId_).StreamsUnactive(iterStream->second.ringStreams));
843 : }
844 13 : tagStreamInfo_.erase(tag);
845 : // scratchMemMap_清理
846 13 : scratchMemMap_.erase(tag);
847 14 : return HCCL_SUCCESS;
848 : }
849 :
850 0 : HcclResult hcclImpl::SetRankPortInfo(s32 deviceLogicID, bool isUseRankPort, std::vector<u32>& ranksPort)
851 : {
852 0 : if (g_SetRankPortInfoCallBack != nullptr) {
853 0 : return g_SetRankPortInfoCallBack(deviceLogicID, isUseRankPort, ranksPort);
854 : } else {
855 0 : HCCL_RUN_WARNING("[SetRankPortInfo] g_SetRankPortInfoCallBack is nullptr");
856 : }
857 0 : return HCCL_SUCCESS;
858 : }
859 :
860 145 : void hcclImpl::SetHDCModeInfo(
861 : std::unordered_map<std::string, std::map<u32, HcclIpAddress>>& rankDevicePhyIdNicInfoMap,
862 : std::vector<u32>& ranksPort, std::vector<u32>& vnicRanksPort, bool isSetHDCModeInfo, bool isUseRankPort)
863 : {
864 145 : rankDevicePhyIdNicInfoMap_ = rankDevicePhyIdNicInfoMap;
865 140 : ranksPort_ = ranksPort;
866 143 : vnicRanksPort_ = vnicRanksPort;
867 142 : isSetHDCModeInfo_ = isSetHDCModeInfo;
868 142 : isUseRankPort_ = isUseRankPort;
869 142 : }
870 : #ifdef __cplusplus
871 : extern "C" {
872 : #endif // __cplusplus
873 47 : void RegisterHeartBeatCallBack(
874 : RegisterToHeartBeatCallBack p1, UnRegisterToHeartBeatCallBack p2, SetRankPortInfoCallBack p3)
875 : {
876 47 : g_RegisterToHeartBeatCallBack = p1;
877 47 : g_UnRegisterToHeartBeatCallBack = p2;
878 47 : g_SetRankPortInfoCallBack = p3;
879 47 : }
880 : #ifdef __cplusplus
881 : }
882 : #endif // __cplusplus
883 : } // namespace hccl
884 : // namespace hccl
|
