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