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 "network_manager.h"
12 : #include "externalinput_pub.h"
13 : #include "device_capacity.h"
14 : #include "adapter_tdt.h"
15 : #include "adapter_rts.h"
16 : #include "adapter_hccp.h"
17 : #include "adapter_hal.h"
18 : #include "adapter_error_manager.h"
19 : #include "dlhal_function.h"
20 : #include "adapter_hccp_common.h"
21 : namespace hccl {
22 :
23 : using namespace std;
24 :
25 : constexpr u32 SOCKET_LISTEN_AUTO_INTERFACE = 4;
26 : constexpr u32 SOCKET_LISTEN_AUTO_INTERFACE_VERSION = 3;
27 :
28 : NetworkManager* NetworkManager::nmInstance[MAX_DEV_NUM] = {nullptr};
29 : std::atomic<unsigned> NetworkManager::InitTool::initCount(0);
30 :
31 1579 : NetworkManager::InitTool::InitTool()
32 : {
33 1579 : if (initCount.load() == 0) {
34 1551 : for (u32 i = 0; i < MAX_DEV_NUM; i++) {
35 1504 : NetworkManager::nmInstance[i] = new (std::nothrow) NetworkManager;
36 1504 : CHK_PRT_CONT(
37 : NetworkManager::nmInstance[i] == nullptr,
38 : HCCL_ERROR("[NetworkManager][InitTool] nmInstance[%u] new failed", i));
39 : }
40 : }
41 1579 : ++initCount;
42 1579 : }
43 :
44 1579 : NetworkManager::InitTool::~InitTool()
45 : {
46 1579 : --initCount;
47 1579 : if (initCount.load() == 0) {
48 1551 : for (u32 i = 0; i < MAX_DEV_NUM; i++) {
49 1504 : if (NetworkManager::nmInstance[i] != nullptr) {
50 1504 : delete NetworkManager::nmInstance[i];
51 1504 : NetworkManager::nmInstance[i] = nullptr;
52 : }
53 : }
54 : }
55 1579 : }
56 :
57 1504 : NetworkManager::NetworkManager()
58 1504 : : deviceLogicId_(INVALID_INT),
59 1504 : devicePhyId_(INVALID_UINT),
60 1504 : isHostUseDevNic_(false),
61 1504 : notifyType_(NO_USE)
62 1504 : {}
63 :
64 1504 : NetworkManager::~NetworkManager()
65 : {
66 1504 : Destroy();
67 1504 : isRaDeInit_ = false;
68 1504 : }
69 :
70 1178 : NetworkManager& NetworkManager::GetInstance(s32 deviceLogicID)
71 : {
72 1178 : HCCL_INFO("NetworkManager::GetInstance deviceLogicID[%u].", deviceLogicID);
73 1179 : if (deviceLogicID == HOST_DEVICE_ID) {
74 4 : nmInstance[DEFAULT_DEVICE_LOGIC_ID]->deviceLogicId_ = DEFAULT_DEVICE_LOGIC_ID;
75 4 : return *(nmInstance[DEFAULT_DEVICE_LOGIC_ID]);
76 : }
77 :
78 1175 : if (static_cast<u32>(deviceLogicID) >= MAX_DEV_NUM || deviceLogicID <= HOST_DEVICE_ID) {
79 0 : HCCL_WARNING("[Get][Instance]deviceLogicID[%d] is invalid", deviceLogicID);
80 0 : nmInstance[DEFAULT_DEVICE_LOGIC_ID]->deviceLogicId_ = DEFAULT_DEVICE_LOGIC_ID;
81 0 : return *(nmInstance[DEFAULT_DEVICE_LOGIC_ID]);
82 : }
83 1175 : nmInstance[deviceLogicID]->deviceLogicId_ = deviceLogicID;
84 1175 : return *(nmInstance[deviceLogicID]);
85 : }
86 :
87 411 : HcclResult NetworkManager::TsdCapabilityGet(bool& supportMultiProcHCCP)
88 : {
89 411 : int32_t type = TSD_CAPABILITY_MUTIPLE_HCCP;
90 411 : bool* resultPtr = &supportMultiProcHCCP;
91 411 : CHK_RET(hrtTsdCapabilityGet(deviceLogicId_, type, static_cast<uint64_t>(reinterpret_cast<uintptr_t>(resultPtr))));
92 411 : return HCCL_SUCCESS;
93 : }
94 :
95 0 : HcclResult NetworkManager::GetNicIp(uint32_t devicePhyId, HcclAddress** addr, uint32_t* len)
96 : {
97 0 : vector<HcclIpAddress> tempIp;
98 0 : std::unique_lock<std::mutex> lock(memResMutex_);
99 0 : if (nicIpAddrs_.size() == 0) {
100 0 : CHK_RET(hrtRaGetDeviceIP(devicePhyId, tempIp));
101 0 : nicIpAddrs_.resize(tempIp.size());
102 0 : for (size_t i = 0; i < tempIp.size(); ++i) {
103 0 : CHK_RET(HcclIpAddressConvertHcclAddr(&nicIpAddrs_[i], &tempIp[i]));
104 : }
105 : }
106 0 : *addr = this->nicIpAddrs_.data();
107 0 : *len = this->nicIpAddrs_.size();
108 0 : lock.unlock();
109 0 : return HCCL_SUCCESS;
110 0 : }
111 :
112 0 : HcclResult NetworkManager::TsdProcessOpen(bool hasBackup)
113 : {
114 0 : s32 locaLogDevid = 0;
115 0 : hrtGetDevice(&locaLogDevid);
116 0 : if (locaLogDevid != deviceLogicId_) {
117 0 : hrtSetDevice(deviceLogicId_);
118 : }
119 : // 校验是否为新版本驱动，旧版本驱动不支持配置backupPhyId，报错返回
120 0 : s32 halAPIVersion = 0;
121 0 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
122 0 : CHK_RET(hrtHalGetAPIVersion(halAPIVersion));
123 0 : HCCL_INFO(
124 : "[%s]params: halAPIVersion[%d], BACKUP_DEVICE_LOG_DEV_VERSION[%d]", __func__, halAPIVersion,
125 : BACKUP_DEVICE_LOG_DEV_VERSION);
126 0 : if (halAPIVersion < BACKUP_DEVICE_LOG_DEV_VERSION) {
127 0 : HCCL_WARNING(
128 : "[%s]this package does not support obtaining backUp HCCP Log in PLOG. halAPIVersion[%d]", __func__,
129 : halAPIVersion);
130 : }
131 0 : isTsdProcessOpen_ = true;
132 0 : if (!hasBackup || halAPIVersion < BACKUP_DEVICE_LOG_DEV_VERSION) {
133 0 : std::string extPam("--hdcType=" + std::to_string(HDC_SERVICE_TYPE_RDMA_V2));
134 : rtNetServiceOpenArgs openArgs;
135 0 : rtProcExtParam extParam{};
136 0 : extParam.paramInfo = extPam.c_str();
137 0 : extParam.paramLen = extPam.size();
138 0 : openArgs.extParamList = &extParam;
139 0 : openArgs.extParamCnt = 1UL;
140 : // 根据pid粒度拉起hccp的rs进程
141 0 : CHK_RET(hrtOpenNetService(&openArgs));
142 0 : HCCL_INFO(
143 : "[%s]hrtOpenNetService success, subPid[%d], devicePhyId_[%u], deviceLogicId_[%d], hasBackup[%u]", __func__,
144 : subPid_, devicePhyId_, deviceLogicId_, hasBackup);
145 0 : } else {
146 : // 获取chip上另一个die的logicalID
147 0 : u32 deviceBackUpPhyId = 0;
148 0 : CHK_RET(hrtGetPairDevicePhyId(devicePhyId_, deviceBackUpPhyId));
149 : rtNetServiceOpenArgs openArgs;
150 : std::string extPams[TSD_OPEN_EXT_PARA_NUM]
151 0 : = {std::string("--hdcType=" + std::to_string(HDC_SERVICE_TYPE_RDMA_V2)),
152 0 : std::string("--backupPhyId=" + std::to_string(deviceBackUpPhyId))};
153 0 : rtProcExtParam extParams[TSD_OPEN_EXT_PARA_NUM]{};
154 0 : for (u32 i = 0; i < TSD_OPEN_EXT_PARA_NUM; i++) {
155 0 : extParams[i].paramInfo = extPams[i].c_str();
156 0 : extParams[i].paramLen = extPams[i].size();
157 : }
158 0 : openArgs.extParamList = extParams;
159 0 : openArgs.extParamCnt = TSD_OPEN_EXT_PARA_NUM;
160 : // 根据pid粒度拉起hccp的rs进程
161 0 : CHK_RET(hrtOpenNetService(&openArgs));
162 0 : HCCL_INFO(
163 : "[%s]hrtOpenNetService success, subPid[%u], "
164 : "devicePhyId_[%u], deviceLogicId_[%d], deviceBackUpPhyId[%u], hasBackup[%u]",
165 : __func__, subPid_, devicePhyId_, deviceLogicId_, deviceBackUpPhyId, hasBackup);
166 0 : }
167 :
168 0 : if (locaLogDevid != deviceLogicId_) {
169 0 : hrtSetDevice(locaLogDevid);
170 : }
171 0 : return HCCL_SUCCESS;
172 : }
173 :
174 1 : HcclResult NetworkManager::PrepareInit(NICDeployment nicDeploy, u32 devicePhyId, s32& ref)
175 : {
176 1 : HCCL_INFO("nicDeploy = [%u], devicePhyId = [%u] ", nicDeploy, devicePhyId);
177 1 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) {
178 0 : ref = hostNicInitRef_.Ref();
179 1 : } else if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
180 1 : ref = deviceNicInitRef_.Ref();
181 : } else {
182 0 : HCCL_ERROR(
183 : "[NetworkManager][PrepareInit]NetworkManager: init nic failed, nicPosition[%u] is not supported.",
184 : static_cast<u32>(nicDeploy));
185 0 : return HCCL_E_INTERNAL;
186 : }
187 1 : if (ref > 1) {
188 0 : HCCL_INFO("NetworkManager: init nic, nicPosition[%u] ref[%u], skip.", static_cast<u32>(nicDeploy), ref);
189 0 : return HCCL_SUCCESS;
190 : }
191 :
192 1 : if (devicePhyId != INVALID_UINT) {
193 0 : devicePhyId_ = devicePhyId;
194 : } else {
195 : // 初始化ra资源(dev信息带入逻辑ID)
196 1 : CHK_RET(hrtGetDevice(&deviceLogicId_));
197 1 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId_), devicePhyId_));
198 : }
199 1 : return HCCL_SUCCESS;
200 : }
201 1 : HcclResult NetworkManager::GetConfigAndRaInit(struct RaInitConfig& config, bool isHdcV2, NICDeployment nicDeploy)
202 : {
203 : // DC场景网卡与进程在同一侧，需要设置为类似host网卡模式
204 1 : config.nicPosition = Is310PDevice() ? 0 : static_cast<u32>(nicDeploy);
205 1 : config.phyId = devicePhyId_;
206 :
207 1 : if (isHdcV2) {
208 : // 使用HDC_SERVICE_TYPE_RDMA_V2指定进程粒度
209 0 : config.hdcType = HDC_SERVICE_TYPE_RDMA_V2;
210 0 : HCCL_DEBUG(
211 : "[%s]hdcType is set to HDC_SERVICE_TYPE_RDMA_V2"
212 : "devicePhyId[%u], deviceLogicId_[%d]",
213 : __func__, devicePhyId_, deviceLogicId_);
214 : }
215 1 : HcclResult ret = HrtRaInit(&config);
216 1 : RPT_CALL_ERR(
217 : ret != HCCL_SUCCESS, "ra init failed,return[%d] devicePhyId_[%u], nicPosition[%u]", ret, devicePhyId_,
218 : static_cast<u32>(nicDeploy));
219 1 : CHK_PRT_RET(
220 : ret != HCCL_SUCCESS,
221 : HCCL_ERROR(
222 : "[NetworkManager][Init]errNo[0x%016llx] ra init failed,return[%d] devicePhyId_[%u], "
223 : "nicPosition[%u]",
224 : HCCL_ERROR_CODE(ret), ret, devicePhyId_, static_cast<u32>(nicDeploy)),
225 : HCCL_E_NETWORK);
226 :
227 1 : return HCCL_SUCCESS;
228 : }
229 :
230 1 : HcclResult NetworkManager::GetTsdOpen(NICDeployment nicDeploy, bool hasBackup, bool& supportMultiProcHCCP)
231 : {
232 1 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !Is310PDevice()) { // DC场景不需要拉起tsd
233 1 : CHK_RET(DlTdtFunction::GetInstance().DlTdtFunctionInit());
234 1 : CHK_RET(TsdCapabilityGet(supportMultiProcHCCP));
235 1 : HCCL_INFO("[NetworkManager][Init]supportMultiProcHCCP[%u], hasBackup[%u]", supportMultiProcHCCP, hasBackup);
236 1 : if (supportMultiProcHCCP || hasBackup) {
237 : // 根据pid粒度拉起hccp的rs进程
238 0 : CHK_RET(TsdProcessOpen(hasBackup));
239 0 : HCCL_INFO(
240 : "[NetworkManager][Init]open tsd by process success, devicePhyId[%u], deviceLogicId_[%d].", devicePhyId_,
241 : deviceLogicId_);
242 0 : } else {
243 : // device 网卡初始化前需要拉起 hccp.
244 1 : CHK_RET(hrtOpenTsd());
245 1 : HCCL_INFO(
246 : "[%s]NetworkManager open tsd success, devicePhyId[%u], deviceLogicId_[%d]", __func__, devicePhyId_,
247 : deviceLogicId_);
248 : }
249 : }
250 1 : return HCCL_SUCCESS;
251 : }
252 : /* init network resource */
253 252 : HcclResult NetworkManager::Init(
254 : NICDeployment nicDeploy, bool enableWhitelistFlag, u32 devicePhyId, bool isHostUseDevNic, bool hasBackup)
255 : {
256 252 : s32 ref = 0;
257 252 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) {
258 25 : ref = hostNicInitRef_.Ref();
259 227 : } else if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
260 227 : ref = deviceNicInitRef_.Ref();
261 : } else {
262 0 : HCCL_ERROR(
263 : "[NetworkManager][Init]NetworkManager: init nic failed, nicPosition[%u] is not supported.",
264 : static_cast<u32>(nicDeploy));
265 0 : return HCCL_E_INTERNAL;
266 : }
267 252 : if (ref > 1) {
268 25 : HCCL_INFO("NetworkManager: init nic, nicPosition[%u] ref[%u], skip.", static_cast<u32>(nicDeploy), ref);
269 25 : return HCCL_SUCCESS;
270 : }
271 :
272 227 : if (devicePhyId != INVALID_UINT) {
273 37 : devicePhyId_ = devicePhyId;
274 : } else {
275 : // 初始化ra资源(dev信息带入逻辑ID)
276 190 : CHK_RET(hrtGetDevice(&deviceLogicId_));
277 190 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId_), devicePhyId_));
278 : }
279 227 : HCCL_INFO(
280 : "NetworkManager: devicePhyId[%u], devicePhyId_[%u] deviceLogicId_[%u], hasBackup[%d]", devicePhyId,
281 : devicePhyId_, deviceLogicId_, hasBackup);
282 :
283 227 : bool supportMultiProcHCCP = false;
284 227 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !Is310PDevice()) { // DC场景不需要拉起tsd
285 206 : CHK_RET(DlTdtFunction::GetInstance().DlTdtFunctionInit());
286 206 : CHK_RET(TsdCapabilityGet(supportMultiProcHCCP));
287 206 : HCCL_INFO("[NetworkManager][Init]supportMultiProcHCCP[%u], hasBackup[%u]", supportMultiProcHCCP, hasBackup);
288 206 : if (supportMultiProcHCCP || hasBackup) {
289 : // 根据pid粒度拉起hccp的rs进程
290 0 : CHK_RET(TsdProcessOpen(hasBackup));
291 0 : HCCL_INFO(
292 : "[NetworkManager][Init]open tsd by process success, devicePhyId[%u], deviceLogicId_[%d].", devicePhyId_,
293 : deviceLogicId_);
294 0 : } else {
295 : // device 网卡初始化前需要拉起 hccp.
296 206 : CHK_RET(hrtOpenTsd());
297 206 : HCCL_INFO(
298 : "[%s]NetworkManager open tsd success, devicePhyId[%u], deviceLogicId_[%d]", __func__, devicePhyId_,
299 : deviceLogicId_);
300 : }
301 : }
302 :
303 227 : struct RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, false};
304 227 : u32 enableWhiteList = (GetExternalInputHcclEnableWhitelist() == HCCL_WHITELIST_ON) ? 1 : 0;
305 227 : if (GetRemoteIsHdc() && IsGeneralServer()) {
306 0 : HCCL_INFO("General server NetworkManager open Whitelist");
307 0 : enableWhitelistFlag = true;
308 0 : enableWhiteList = true;
309 : }
310 227 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST && enableWhitelistFlag) {
311 16 : CHK_RET(hrtRaSocketSetWhiteListStatus(enableWhiteList));
312 : }
313 : // DC场景网卡与进程在同一侧，需要设置为类似host网卡模式
314 227 : config.nicPosition = Is310PDevice() ? 0 : static_cast<u32>(nicDeploy);
315 227 : config.phyId = devicePhyId_;
316 227 : HCCL_DEBUG(
317 : "[%s]config.phyId = %u, nicPosition[%u], hasBackup[%d], devicePhyId_[%u], deviceLogicId_[%d], "
318 : "devicePhyId[%u]",
319 : __func__, config.phyId, config.nicPosition, hasBackup, devicePhyId_, deviceLogicId_, devicePhyId);
320 :
321 227 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !Is310PDevice() && (supportMultiProcHCCP || hasBackup)) {
322 : // 使用HDC_SERVICE_TYPE_RDMA_V2指定进程粒度
323 0 : config.hdcType = HDC_SERVICE_TYPE_RDMA_V2;
324 0 : HCCL_DEBUG(
325 : "[%s]hdcType is set to HDC_SERVICE_TYPE_RDMA_V2, hasBackup[%d], nicDeploy[%d], "
326 : "devicePhyId[%u], deviceLogicId_[%d]",
327 : __func__, hasBackup, nicDeploy, devicePhyId_, deviceLogicId_);
328 : }
329 : DevType devType;
330 227 : CHK_RET(hrtGetDeviceType(devType));
331 227 : if (devType == DevType::DEV_TYPE_910_93 && nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
332 8 : isEnableHdcAsync_ = true;
333 8 : config.enableHdcAsync = true;
334 : }
335 227 : HCCL_INFO(
336 : "[%s]config.phyId[%u], config.nicPosition[%u], config.hdcType[%d], config.enableHdcAsync[%d]", __func__,
337 : config.phyId, config.nicPosition, config.hdcType, config.enableHdcAsync);
338 227 : HcclResult ret = HrtRaInit(&config);
339 227 : RPT_CALL_ERR(
340 : ret != HCCL_SUCCESS, "ra init failed,return[%d] devicePhyId_[%u], nicPosition[%u]", ret, devicePhyId_,
341 : static_cast<u32>(nicDeploy));
342 227 : CHK_PRT_RET(
343 : ret != HCCL_SUCCESS,
344 : HCCL_ERROR(
345 : "[NetworkManager][Init]errNo[0x%016llx] ra init failed,return[%d] devicePhyId_[%u], "
346 : "nicPosition[%u]",
347 : HCCL_ERROR_CODE(ret), ret, devicePhyId_, static_cast<u32>(nicDeploy)),
348 : HCCL_E_NETWORK);
349 :
350 226 : HCCL_INFO(
351 : "NetworkManager nicDeploy[%u] deviceLogicId_[%d] devicePhyId_[%u] init ra OK, nicSocketMap size[%u], "
352 : "isHostUseDevNic[%d]",
353 : static_cast<u32>(nicDeploy), deviceLogicId_, devicePhyId_, raResourceInfo_.nicSocketMap.size(),
354 : isHostUseDevNic);
355 226 : return HCCL_SUCCESS;
356 : }
357 :
358 1 : HcclResult NetworkManager::InitV2(NICDeployment nicDeploy, bool isBackup, u32 devicePhyId, bool isHostUseDevNic)
359 : {
360 1 : s32 ref = 0;
361 1 : CHK_RET(PrepareInit(nicDeploy, devicePhyId, ref));
362 1 : if (ref > 1) {
363 0 : HCCL_INFO("NetworkManager: initv2 nic, nicPosition[%u] ref[%u], skip.", static_cast<u32>(nicDeploy), ref);
364 0 : return HCCL_SUCCESS;
365 : }
366 1 : HCCL_INFO(
367 : "NetworkManager InitV2: devicePhyId[%u], devicePhyId_[%u] deviceLogicId_[%u], isBackup[%d]", devicePhyId,
368 : devicePhyId_, deviceLogicId_, isBackup);
369 :
370 1 : bool supportMultiProcHCCP = false;
371 1 : CHK_RET(GetTsdOpen(nicDeploy, isBackup, supportMultiProcHCCP));
372 :
373 1 : struct RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, false};
374 : bool isHdcV2
375 1 : = (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !Is310PDevice() && (supportMultiProcHCCP || isBackup));
376 1 : if (isHdcV2) {
377 0 : HCCL_DEBUG(
378 : "[%s]hdcType is set to HDC_SERVICE_TYPE_RDMA_V2, isBackup[%d], nicDeploy[%d], "
379 : "devicePhyId[%u], deviceLogicId_[%d]",
380 : __func__, isBackup, nicDeploy, devicePhyId_, deviceLogicId_);
381 : }
382 1 : CHK_RET(GetConfigAndRaInit(config, isHdcV2, nicDeploy));
383 :
384 1 : HCCL_INFO(
385 : "NetworkManager nicDeploy[%u] deviceLogicId_[%d] devicePhyId_[%u] init ra OK, nicSocketMap size[%u], "
386 : "isHostUseDevNic[%d]",
387 : static_cast<u32>(nicDeploy), deviceLogicId_, devicePhyId_, raResourceInfo_.nicSocketMap.size(),
388 : isHostUseDevNic);
389 1 : return HCCL_SUCCESS;
390 : }
391 :
392 0 : HcclResult NetworkManager::HeterogStartListen(const HcclIpAddress& ipAddr, u32 port)
393 : {
394 0 : HCCL_DEBUG("HeterogStartListen ipAddr[%s] port[%u]", ipAddr.GetReadableAddress(), port);
395 0 : SocketHandle nicSocketHandle = raResourceInfo_.nicSocketMap[ipAddr].nicSocketHandle;
396 :
397 0 : CHK_PRT_RET(
398 : port > MAX_PORT_ID || port < MIN_PORT_ID,
399 : HCCL_ERROR("[NetworkManager][HeterogStartListen] port error[%u]", port), HCCL_E_NETWORK);
400 :
401 0 : if (nicSocketHandle != nullptr
402 0 : && raResourceInfo_.nicSocketMap[ipAddr].listenedPort.find(port)
403 0 : == raResourceInfo_.nicSocketMap[ipAddr].listenedPort.end()
404 0 : && IPPortListenRefMapHost_[ipAddr][port].Ref() == FIRST_LISTEN) {
405 0 : CHK_RET(StartListenSocket(nicSocketHandle, port));
406 0 : raResourceInfo_.nicSocketMap[ipAddr].listenedPort.insert(port);
407 0 : raResourceInfo_.hostNetSocketMap[ipAddr].listenedPort.insert(port);
408 : }
409 0 : return HCCL_SUCCESS;
410 : }
411 :
412 1 : HcclResult NetworkManager::HeterogInit(u32 devId, const HcclIpAddress& ipAddr, u32 port)
413 : {
414 1 : s32 ref = hostNicInitRef_.Ref();
415 1 : if (ref > 1) {
416 0 : HCCL_INFO("NetworkManager: heterog init nic, ref[%u], skip", ref);
417 0 : return HCCL_SUCCESS;
418 : }
419 :
420 1 : CHK_RET(hrtRaSocketSetWhiteListStatus(0));
421 :
422 : // 暂缺获取物理id的手段
423 1 : RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, false};
424 1 : config.phyId = ((static_cast<s32>(devId) == HOST_DEVICE_ID) ? 0 : devId);
425 1 : config.nicPosition = static_cast<u32>(NICDeployment::NIC_DEPLOYMENT_HOST);
426 1 : HCCL_INFO("HeterogInit call HrtRaInit.");
427 1 : CHK_RET(HrtRaInit(&config));
428 :
429 1 : struct rdev nicRdevInfo = {};
430 1 : nicRdevInfo.phyId = devId;
431 1 : nicRdevInfo.family = ipAddr.GetFamily();
432 1 : nicRdevInfo.localIp.addr = ipAddr.GetBinaryAddress().addr;
433 1 : nicRdevInfo.localIp.addr6 = ipAddr.GetBinaryAddress().addr6;
434 1 : SocketHandle socketHandle = nullptr;
435 1 : HcclResult ret = hrtRaSocketInit(NETWORK_PEER_ONLINE, nicRdevInfo, socketHandle);
436 1 : if (ret != HCCL_SUCCESS) {
437 1 : HCCL_ERROR(
438 : "[HeterogInit]errNo[0x%016llx] ra socket init failed, ip[%s], return[%d]",
439 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), ipAddr.GetReadableAddress(), ret);
440 1 : HrtRaDeInit(&config);
441 1 : return HCCL_E_TCP_CONNECT;
442 : }
443 0 : HCCL_INFO("ip[%s] socket init OK", ipAddr.GetReadableAddress());
444 :
445 0 : IpSocket ipSocketInfo;
446 0 : ipSocketInfo.nicSocketHandle = socketHandle;
447 :
448 0 : raResourceInfo_.nicSocketMap.insert(std::make_pair(ipAddr, ipSocketInfo));
449 0 : raResourceInfo_.hostNetSocketMap.insert(std::make_pair(ipAddr, ipSocketInfo));
450 :
451 : /* device网卡初始化暂不考虑 */
452 0 : if (!GetExternalInputHcclIsTcpMode()) {
453 0 : ret = InitRdmaHandle(devId, ipAddr);
454 0 : if (ret != HCCL_SUCCESS) {
455 0 : HCCL_ERROR("[HeterogInit] InitRdmaHandle fail, ret[%d], destroy resource.", ret);
456 0 : hrtRaSocketDeInit(socketHandle);
457 0 : HrtRaDeInit(&config);
458 0 : return ret;
459 : }
460 : }
461 :
462 0 : ret = HeterogStartListen(ipAddr, port);
463 0 : if (ret != HCCL_SUCCESS) {
464 0 : HCCL_ERROR("[HeterogInit] HeterogStartListen fail, ret[%d], destroy resource.", ret);
465 0 : if (!GetExternalInputHcclIsTcpMode()) {
466 0 : auto& ipSock = raResourceInfo_.nicSocketMap[ipAddr];
467 0 : if (ipSock.nicRdmaHandle != nullptr) {
468 0 : (void)HrtRaRdmaDeInit(ipSock.nicRdmaHandle, NO_USE);
469 : }
470 : }
471 0 : hrtRaSocketDeInit(socketHandle);
472 0 : HrtRaDeInit(&config);
473 0 : raResourceInfo_.nicSocketMap.erase(ipAddr);
474 0 : return ret;
475 : }
476 0 : return HCCL_SUCCESS;
477 0 : }
478 :
479 0 : HcclResult NetworkManager::HeterogStopListen(const HcclIpAddress& ipAddr, u32 port, bool isNeedDeinit)
480 : {
481 0 : HCCL_DEBUG("HeterogStopListen ipAddr[%s] port[%u]", ipAddr.GetReadableAddress(), port);
482 0 : SocketHandle nicSocketHandle = raResourceInfo_.nicSocketMap[ipAddr].nicSocketHandle;
483 :
484 0 : if (nicSocketHandle != nullptr && raResourceInfo_.nicSocketMap[ipAddr].listenedPort.size() > 0) {
485 0 : if (raResourceInfo_.nicSocketMap[ipAddr].listenedPort.find(port)
486 0 : != raResourceInfo_.nicSocketMap[ipAddr].listenedPort.end()
487 0 : && IPPortListenRefMapHost_[ipAddr][port].Unref() == LAST_RELEASE) {
488 : // 重复raInit时不用去停止监听，但是要处理引用计数，否则析构会出问题
489 0 : if (isRaInitRepeated_) {
490 0 : return HCCL_SUCCESS;
491 : }
492 0 : HCCL_INFO("HeterogStopListen socket listen stop, socket port[%u]", port);
493 0 : CHK_RET(StopListenSocket(nicSocketHandle, port));
494 0 : raResourceInfo_.nicSocketMap[ipAddr].listenedPort.erase(port);
495 0 : raResourceInfo_.hostNetSocketMap[ipAddr].listenedPort.erase(port);
496 : }
497 :
498 0 : if (isNeedDeinit) {
499 0 : CHK_RET(hrtRaSocketDeInit(nicSocketHandle));
500 : }
501 : }
502 0 : return HCCL_SUCCESS;
503 : }
504 :
505 1 : HcclResult NetworkManager::HeterogDeinit(u32 devId, const HcclIpAddress& ipAddr, u32 port)
506 : {
507 1 : s32 ref = hostNicInitRef_.Unref();
508 1 : if (ref > 0) {
509 0 : HCCL_INFO("NetworkManager: heterog deinit nic success, ref[%u], skip.", ref);
510 0 : return HCCL_SUCCESS;
511 1 : } else if (ref < 0) {
512 1 : HCCL_ERROR("[NetworkManager][DeInit]NetworkManager: heterog deinit nic failed, nic has already deinit.");
513 1 : return HCCL_E_INTERNAL;
514 : }
515 :
516 0 : CHK_RET(HeterogStopListen(ipAddr, port, true));
517 :
518 0 : SocketHandle nicRdmaHandle = raResourceInfo_.nicSocketMap[ipAddr].nicRdmaHandle;
519 0 : if (!GetExternalInputHcclIsTcpMode() && nicRdmaHandle != nullptr) {
520 0 : CHK_RET(HrtRaRdmaDeInit(nicRdmaHandle, NO_USE));
521 : }
522 :
523 0 : RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, false};
524 0 : config.phyId = devId;
525 0 : config.nicPosition = static_cast<u32>(NICDeployment::NIC_DEPLOYMENT_HOST);
526 :
527 0 : HCCL_INFO("HeterogDeinit call HrtRaDeInit.");
528 0 : CHK_RET(HrtRaDeInit(&config));
529 :
530 0 : raResourceInfo_.nicSocketMap.erase(raResourceInfo_.nicSocketMap.find(ipAddr));
531 0 : raResourceInfo_.hostNetSocketMap.erase(raResourceInfo_.hostNetSocketMap.find(ipAddr));
532 :
533 0 : return HCCL_SUCCESS;
534 : }
535 :
536 1 : HcclResult NetworkManager::CloseHccpProcess()
537 : {
538 1 : std::unique_lock<std::mutex> lock(hccpProcInfoMutex_);
539 1 : if (isTsdProcessOpen_ == true) {
540 0 : s32 locaLogDevid = 0;
541 0 : hrtGetDevice(&locaLogDevid);
542 0 : if (locaLogDevid != deviceLogicId_) {
543 0 : hrtSetDevice(deviceLogicId_);
544 : }
545 0 : CHK_RET(hrtCloseNetService());
546 0 : isTsdProcessOpen_ = false;
547 0 : if (locaLogDevid != deviceLogicId_) {
548 0 : hrtSetDevice(locaLogDevid);
549 : }
550 : }
551 1 : return HCCL_SUCCESS;
552 1 : }
553 :
554 1 : HcclResult NetworkManager::PrepareDeInit(s32& ref, NICDeployment nicDeploy)
555 : {
556 1 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) {
557 1 : ref = hostNicInitRef_.Unref();
558 0 : } else if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
559 0 : ref = deviceNicInitRef_.Unref();
560 : } else {
561 0 : HCCL_ERROR(
562 : "[NetworkManager][PrepareDeInit]NetworkManager: deinit nic failed, nicPosition[%u] is not supported.",
563 : static_cast<u32>(nicDeploy));
564 0 : return HCCL_E_INTERNAL;
565 : }
566 :
567 1 : if (ref > 0) {
568 0 : HCCL_INFO(
569 : "NetworkManager: PrepareDeInit nic success, nicPosition[%u] ref[%u], skip.", static_cast<u32>(nicDeploy),
570 : ref);
571 0 : return HCCL_SUCCESS;
572 1 : } else if (ref < 0) {
573 1 : HCCL_ERROR(
574 : "[NetworkManager][PrepareDeInit]NetworkManager: deinit nic failed, nicPosition[%u] has already deinit.",
575 : nicDeploy);
576 1 : return HCCL_E_INTERNAL;
577 : }
578 0 : return HCCL_SUCCESS;
579 : }
580 :
581 0 : HcclResult NetworkManager::GetConfigAndRaDeinit(
582 : struct RaInitConfig& config, NICDeployment nicDeploy, bool& isMultiProc, bool hasBackup)
583 : {
584 0 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !Is310PDevice()) {
585 0 : CHK_RET(DlTdtFunction::GetInstance().DlTdtFunctionInit());
586 0 : bool supportMultiProcHCCP = false;
587 0 : CHK_RET(TsdCapabilityGet(supportMultiProcHCCP));
588 0 : if (supportMultiProcHCCP || hasBackup) {
589 0 : isMultiProc = true;
590 0 : config.hdcType = HDC_SERVICE_TYPE_RDMA_V2;
591 : }
592 : }
593 :
594 0 : HcclResult ret = HrtRaDeInit(&config);
595 0 : CHK_PRT_RET(
596 : ret != HCCL_SUCCESS,
597 : HCCL_ERROR(
598 : "[NetworkManager][DeInit]ra deinit failed. para: nicdeploy[%u], phyId[%u], ret[%u]", config.nicPosition,
599 : config.phyId, ret),
600 : ret);
601 0 : if (IsGeneralServer()) {
602 0 : isRaDeInit_ = true;
603 : }
604 0 : return HCCL_SUCCESS;
605 : }
606 :
607 234 : HcclResult NetworkManager::DeInit(NICDeployment nicDeploy, [[maybe_unused]] bool resetDeviceFlag, bool hasBackup)
608 : {
609 234 : s32 ref = 0;
610 234 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) {
611 20 : ref = hostNicInitRef_.Unref();
612 214 : } else if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
613 217 : ref = deviceNicInitRef_.Unref();
614 : } else {
615 0 : HCCL_ERROR(
616 : "[NetworkManager][DeInit]NetworkManager: deinit nic failed, nicPosition[%u] is not supported.",
617 : static_cast<u32>(nicDeploy));
618 0 : return HCCL_E_INTERNAL;
619 : }
620 :
621 237 : if (ref > 0) {
622 16 : HCCL_INFO(
623 : "NetworkManager: deinit nic success, nicPosition[%u] ref[%u], skip.", static_cast<u32>(nicDeploy), ref);
624 16 : return HCCL_SUCCESS;
625 221 : } else if (ref < 0) {
626 1 : HCCL_ERROR(
627 : "[NetworkManager][DeInit]NetworkManager: deinit nic failed, nicPosition[%u] has already deinit.",
628 : nicDeploy);
629 1 : return HCCL_E_INTERNAL;
630 : }
631 :
632 220 : struct RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, isEnableHdcAsync_};
633 220 : config.nicPosition = Is310PDevice() ? 0 : static_cast<u32>(nicDeploy);
634 220 : config.phyId = devicePhyId_;
635 :
636 220 : bool isMultiProc = false;
637 220 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !Is310PDevice()) {
638 205 : CHK_RET(DlTdtFunction::GetInstance().DlTdtFunctionInit());
639 205 : bool supportMultiProcHCCP = false;
640 205 : CHK_RET(TsdCapabilityGet(supportMultiProcHCCP));
641 205 : if (supportMultiProcHCCP || hasBackup) {
642 1 : isMultiProc = true;
643 1 : config.hdcType = HDC_SERVICE_TYPE_RDMA_V2;
644 : }
645 : }
646 :
647 220 : HcclResult ret = HrtRaDeInit(&config);
648 220 : CHK_PRT_RET(
649 : ret != HCCL_SUCCESS,
650 : HCCL_ERROR(
651 : "[NetworkManager][DeInit]ra deinit failed. para: nicdeploy[%u], phyId[%u], ret[%u]", config.nicPosition,
652 : config.phyId, ret),
653 : ret);
654 220 : if (IsGeneralServer()) {
655 0 : isRaDeInit_ = true;
656 : }
657 :
658 220 : if (isMultiProc) {
659 1 : CHK_RET(CloseHccpProcess());
660 1 : HCCL_INFO(
661 : "[%s]finish CloseHccpProcess, phyId[%u], hdcType[%d], nicDeployment[%d], hasBackup[%d]", __func__,
662 : config.phyId, config.hdcType, nicDeploy, hasBackup);
663 : }
664 :
665 220 : HCCL_INFO("NetworkManager: deinit nic success, nicPosition[%u] ref[%u].", static_cast<u32>(nicDeploy), ref);
666 220 : return HCCL_SUCCESS;
667 : }
668 :
669 0 : HcclResult NetworkManager::DeInitV2(NICDeployment nicDeploy, bool isBackup, [[maybe_unused]] bool resetDeviceFlag)
670 : {
671 0 : s32 ref = 0;
672 0 : CHK_RET(PrepareDeInit(ref, nicDeploy));
673 0 : if (ref > 0) {
674 0 : HCCL_INFO(
675 : "NetworkManager: DeInitV2 nic success, nicPosition[%u] ref[%u], skip.", static_cast<u32>(nicDeploy), ref);
676 0 : return HCCL_SUCCESS;
677 : }
678 :
679 0 : struct RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, false};
680 0 : config.nicPosition = Is310PDevice() ? 0 : static_cast<u32>(nicDeploy);
681 0 : config.phyId = devicePhyId_;
682 :
683 0 : bool isMultiProc = false;
684 0 : CHK_RET(GetConfigAndRaDeinit(config, nicDeploy, isMultiProc, isBackup));
685 :
686 0 : if (isMultiProc) {
687 0 : CHK_RET(CloseHccpProcess());
688 0 : HCCL_INFO(
689 : "[%s]finish CloseHccpProcess, phyId[%u], hdcType[%d], nicDeployment[%d], isBackup[%d]", __func__,
690 : config.phyId, config.hdcType, nicDeploy, isBackup);
691 : }
692 :
693 0 : HCCL_INFO("NetworkManager: DeInitV2 success, nicPosition[%u] ref[%u].", static_cast<u32>(nicDeploy), ref);
694 0 : return HCCL_SUCCESS;
695 : }
696 :
697 24 : HcclResult NetworkManager::StartVnic(HcclIpAddress localIp, u32& port)
698 : {
699 24 : CHK_PRT_RET(
700 : deviceNicInitRef_.Count() <= 0, HCCL_ERROR("[Start][Vnic]can't start vnic socket before init device nic!"),
701 : HCCL_E_INTERNAL);
702 19 : CHK_PRT_RET(Is310PDevice(), HCCL_INFO("DC does not need vnic"), HCCL_SUCCESS);
703 19 : CHK_PRT_RET(port > MAX_PORT_ID, HCCL_ERROR("[Start][Vnic]invalid port id[%u]", port), HCCL_E_PARA);
704 :
705 19 : auto sockInfo = raResourceInfo_.vnicSocketMap.find(localIp);
706 19 : if (sockInfo == raResourceInfo_.vnicSocketMap.end()) {
707 14 : IpSocket tempSock;
708 14 : raResourceInfo_.vnicSocketMap.insert(std::make_pair(localIp, tempSock)); // 本IP占位
709 14 : HCCL_INFO("[Start][Vnic]device[%u] Start Vnic insert ip[%s]", devicePhyId_, localIp.GetReadableAddress());
710 14 : }
711 :
712 19 : IpSocket& sock = raResourceInfo_.vnicSocketMap[localIp];
713 19 : if (sock.listenedPort.size() == 0 && sock.nicSocketHandle == nullptr) {
714 14 : HcclResult ret = InitDeviceSocket(devicePhyId_, localIp, sock.nicSocketHandle);
715 14 : CHK_PRT_RET(
716 : ret != HCCL_SUCCESS,
717 : HCCL_ERROR(
718 : "[Start][Vnic]errNo[0x%016llx] ra vnic init socket failed, devid[%u], ipAddr[%s], return[%d]",
719 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, localIp.GetReadableAddress(), ret),
720 : ret);
721 : } else {
722 5 : HCCL_INFO(
723 : "[Start][Vnic]vnic socket has inited, ipAddr[%s] port[%u], skip init.", localIp.GetReadableAddress(), port);
724 : }
725 :
726 19 : if (sock.listenedPort.find(port) != sock.listenedPort.end()) {
727 5 : HCCL_WARNING("[Start][Vnic]ipAddr[%s] port[%u] is already listened.", localIp.GetReadableAddress(), port);
728 : } else {
729 : // 作为socket server端启动监听(虚拟网卡)
730 14 : bool isAutoPort = port == 0;
731 14 : HCCL_INFO("[Start][Vnic]trying to listen on ip[%s] port[%u].", localIp.GetReadableAddress(), port);
732 14 : CHK_RET(CheckAutoListenVersion(isAutoPort));
733 14 : HcclResult ret = StartListenSocket(sock.nicSocketHandle, port); /* 只拉起1个vnic */
734 14 : CHK_PRT_RET(
735 : ret == HCCL_E_UNAVAIL,
736 : HCCL_INFO(
737 : "[Start][StartVnic]Could not start listening socket for IP [%s] and port [%u].",
738 : localIp.GetReadableAddress(), port),
739 : ret);
740 14 : CHK_PRT_RET(
741 : ret != HCCL_SUCCESS,
742 : HCCL_ERROR(
743 : "[Start][Vnic]errNo[0x%016llx] ra inner listen start failed, "
744 : "devid[%u], ip[%s], port[%u], return[%d]",
745 : HCCL_ERROR_CODE(ret), devicePhyId_, localIp.GetReadableAddress(), port, ret),
746 : ret);
747 14 : sock.listenedPort.insert(port);
748 14 : HCCL_RUN_INFO(
749 : "[Start][Vnic]Listen on ip[%s], port[%u] success, devPhyId[%u], devLogicId[%u], isAutoPort[%d]",
750 : localIp.GetReadableAddress(), port, devicePhyId_, deviceLogicId_, isAutoPort);
751 : }
752 19 : int refCount = IPPortListenRefMapVnicDevice_[localIp][port].Ref();
753 19 : HCCL_INFO(
754 : "NetworkManager devicePhyId_[%u] ip[%s] port[%u] start vnic OK. refCount[%d]", devicePhyId_,
755 : localIp.GetReadableAddress(), port, refCount);
756 19 : return HCCL_SUCCESS;
757 : }
758 :
759 : // 此处只进行socket的创建 不listen
760 0 : HcclResult NetworkManager::CreateVnicSocketHandle(HcclIpAddress localIp)
761 : {
762 0 : CHK_PRT_RET(
763 : !deviceNicInitRef_.Count(),
764 : HCCL_ERROR("[NetworkManager][CreateVnicSocketHandle]"
765 : "can't start vnic socket before init device nic!"),
766 : HCCL_E_INTERNAL);
767 0 : OccupyIp(localIp, raResourceInfo_.vnicSocketMap);
768 0 : CHK_PRT_RET(Is310PDevice(), HCCL_INFO("DC does not need vnic"), HCCL_SUCCESS);
769 :
770 0 : IpSocket& sock = raResourceInfo_.vnicSocketMap[localIp];
771 0 : if (sock.listenedPort.size() == 0 && sock.nicSocketHandle == nullptr) {
772 0 : HcclResult ret = InitDeviceSocket(devicePhyId_, localIp, sock.nicSocketHandle);
773 0 : CHK_PRT_RET(
774 : ret != HCCL_SUCCESS,
775 : HCCL_ERROR(
776 : "[NetworkManager][CreateVnicSocketHandle]errNo[0x%016llx] ra vnic init socket failed, "
777 : "devid[%u], return[%d]",
778 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, ret),
779 : ret);
780 : } else {
781 0 : HCCL_INFO(
782 : "[NetworkManager][CreateVnicSocketHandle] socket has inited, ipAddr[%s], skip.",
783 : localIp.GetReadableAddress());
784 : }
785 0 : HCCL_INFO(
786 : "[NetworkManager][CreateVnicSocketHandle] CreateVnicSocketHandle OK, ipAddr[%s].",
787 : localIp.GetReadableAddress());
788 0 : return HCCL_SUCCESS;
789 : }
790 :
791 6 : HcclResult NetworkManager::StopVnic(const HcclIpAddress& localIp, u32 port)
792 : {
793 6 : auto it = raResourceInfo_.vnicSocketMap.find(localIp);
794 6 : CHK_PRT_RET(
795 : it == raResourceInfo_.vnicSocketMap.end(),
796 : HCCL_ERROR("[Stop][Vnic]ip[%s] is not found in vnicSocketMap, port[%u].", localIp.GetReadableAddress(), port),
797 : HCCL_E_INTERNAL);
798 6 : IpSocket& ipSock = it->second;
799 :
800 6 : int count = IPPortListenRefMapVnicDevice_[localIp][port].Unref();
801 6 : CHK_PRT_RET(
802 : count > 0,
803 : HCCL_INFO("[Stop][Vnic]ip[%s] port[%u] ref[%d] skip stop.", localIp.GetReadableAddress(), port, count),
804 : HCCL_SUCCESS);
805 6 : CHK_PRT_RET(
806 : count < 0,
807 : HCCL_INFO(
808 : "[Stop][Vnic]ip[%s] port[%u] devicePhyId_[%u] vnic stopped ERROR, refcount[%d].",
809 : localIp.GetReadableAddress(), port, devicePhyId_, count),
810 : HCCL_SUCCESS);
811 :
812 : // Stop Listen
813 6 : CHK_PRT_RET(
814 : ipSock.nicSocketHandle != nullptr && StopListenSocket(ipSock.nicSocketHandle, port) != HCCL_SUCCESS,
815 : HCCL_ERROR(
816 : "[Stop][Vnic]errNo[0x%016llx] stop vnic socket failed,devid[%u], ip[%s], port[%u]",
817 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, localIp.GetReadableAddress(), port),
818 : HCCL_E_INTERNAL);
819 6 : ipSock.listenedPort.erase(port);
820 :
821 : // DeInit Socket
822 6 : if (ipSock.listenedPort.size() == 0 && ipSock.nicSocketHandle != nullptr) {
823 6 : HcclResult ret = hrtRaSocketDeInit(ipSock.nicSocketHandle);
824 6 : CHK_PRT_RET(
825 : ret != HCCL_SUCCESS,
826 : HCCL_ERROR(
827 : "[Stop][Vnic]errNo[0x%016llx] stop vnic socket failed,devid[%u], ip[%s], return[%d]",
828 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, localIp.GetReadableAddress(), ret),
829 : HCCL_E_INTERNAL);
830 6 : raResourceInfo_.vnicSocketMap.erase(localIp);
831 : }
832 6 : return HCCL_SUCCESS;
833 : }
834 :
835 1 : HcclResult NetworkManager::StopVnicSocketHandle(const HcclIpAddress& localIp)
836 : {
837 1 : auto it = raResourceInfo_.vnicSocketMap.find(localIp);
838 1 : CHK_PRT_RET(
839 : it == raResourceInfo_.vnicSocketMap.end(),
840 : HCCL_ERROR(
841 : "[NetworkManager][StopVnicSocketHandle]ip[%s] is not found in vnicSocketMap.",
842 : localIp.GetReadableAddress()),
843 : HCCL_E_INTERNAL);
844 :
845 1 : IpSocket& ipSock = it->second;
846 : // 关闭该ip下的全部端口的listen
847 2 : for (auto& port : ipSock.listenedPort) {
848 1 : if (ipSock.nicSocketHandle != nullptr && StopListenSocket(ipSock.nicSocketHandle, port) != HCCL_SUCCESS) {
849 0 : HCCL_ERROR(
850 : "[NetworkManager][StopVnicSocketHandle]errNo[0x%016llx] stop vnic listen failed, "
851 : "devid[%u], ip[%s], port[%u]",
852 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, localIp.GetReadableAddress(), port);
853 0 : break;
854 : }
855 1 : IPPortListenRefMapVnicDevice_[localIp][port].Clear();
856 1 : HCCL_INFO(
857 : "[NetworkManager][StopVnicSocketHandle] ip[%s] stop listen port[%u]", localIp.GetReadableAddress(), port);
858 : }
859 1 : ipSock.listenedPort.clear();
860 :
861 : // 销毁socket
862 1 : CHK_PRT_RET(
863 : ipSock.nicSocketHandle != nullptr && hrtRaSocketDeInit(ipSock.nicSocketHandle),
864 : HCCL_ERROR("[Stop][NicsSocket]VNIC socket deInit not successfully"), HCCL_E_NETWORK);
865 1 : ipSock.nicSocketHandle = nullptr;
866 1 : raResourceInfo_.vnicSocketMap.erase(localIp);
867 :
868 1 : HCCL_INFO(
869 : "[NetworkManager][StopVnicSocketHandle] devid[%u] ip[%s] stop vnic socket success", devicePhyId_,
870 : localIp.GetReadableAddress());
871 1 : return HCCL_SUCCESS;
872 : }
873 :
874 39 : HcclResult NetworkManager::StartNic(const HcclIpAddress& ipAddr, u32& port, bool rdmaFlag, HcclIpAddress ipAddrBackup)
875 : {
876 39 : CHK_PRT_RET(
877 : !deviceNicInitRef_.Count(), HCCL_ERROR("[Start][Nic]can't start nic socket before init device nic!"),
878 : HCCL_E_INTERNAL);
879 38 : auto sockInfo = raResourceInfo_.nicSocketMap.find(ipAddr);
880 38 : if (sockInfo == raResourceInfo_.nicSocketMap.end()) {
881 37 : IpSocket tempSock;
882 37 : raResourceInfo_.nicSocketMap.insert(std::make_pair(ipAddr, tempSock)); // 本IP占位
883 37 : HCCL_INFO("device[%u] Start Nic insert ip[%s]", devicePhyId_, ipAddr.GetReadableAddress());
884 37 : }
885 :
886 38 : IpSocket& sock = raResourceInfo_.nicSocketMap[ipAddr];
887 38 : if (sock.listenedPort.size() == 0) {
888 37 : if (sock.nicSocketHandle == nullptr) {
889 37 : HcclResult ret = InitDeviceSocket(devicePhyId_, ipAddr, sock.nicSocketHandle);
890 37 : CHK_PRT_RET(
891 : ret != HCCL_SUCCESS,
892 : HCCL_ERROR(
893 : "[Start][Nic]errNo[0x%016llx] ra nic init socket failed, devid[%u], return[%d]",
894 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, ret),
895 : ret);
896 : }
897 : } else {
898 1 : HCCL_INFO("NetworkManager: socket has inited, ipAddr[%s] port[%u], skip.", ipAddr.GetReadableAddress(), port);
899 : }
900 38 : if (sock.nicRdmaHandle == nullptr && rdmaFlag) {
901 : NetworkMode netMode;
902 34 : GetNetworkMode(netMode);
903 34 : CHK_RET(GetNotifyType(notifyType_));
904 68 : HcclResult ret = InitRDMA(
905 34 : devicePhyId_, ipAddr, netMode, notifyType_, sock.nicRdmaHandle, isDisableLiteThread_, false, ipAddrBackup);
906 34 : CHK_PRT_RET(
907 : ret != HCCL_SUCCESS,
908 : HCCL_ERROR(
909 : "[Start][Nic]errNo[0x%016llx] ra nic init rdma failed, devid[%u], return[%d]",
910 : HCCL_ERROR_CODE(HCCL_E_NETWORK), devicePhyId_, ret),
911 : HCCL_E_NETWORK);
912 :
913 : int supportLite;
914 34 : ret = HrtGetRdmaLiteStatus(sock.nicRdmaHandle, &supportLite);
915 34 : CHK_PRT_RET(
916 : ret != HCCL_SUCCESS,
917 : HCCL_ERROR(
918 : "[Get][RdmaLiteStatus]errNo[0x%016llx] get rdma lite status failed, return[%d]",
919 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), ret),
920 : HCCL_E_TCP_CONNECT);
921 34 : isRdmaLiteEn_ = (supportLite == 1);
922 34 : } else {
923 4 : HCCL_INFO(
924 : "[Start][Nic] requesting not init rdma , or already init rdma, "
925 : "rdmaFlag[%u], ipAddr[%s], port[%u], skip.",
926 : rdmaFlag, ipAddr.GetReadableAddress(), port);
927 : }
928 : // 如果port id传入的值无效值0xFFFFFFFF, 不启动监听
929 38 : CHK_PRT_RET(port == MAX_VALUE_U32, HCCL_INFO("[Start][Nic] port id[%u], skip listen socket", port), HCCL_SUCCESS);
930 6 : CHK_PRT_RET(port > MAX_PORT_ID, HCCL_ERROR("[Start][Nic]invalid port id[%u]", port), HCCL_E_INTERNAL);
931 6 : if (sock.listenedPort.find(port) != sock.listenedPort.end()) {
932 0 : HCCL_WARNING("port[%u] is already listened.", port);
933 : } else {
934 6 : bool isAutoPort = port == 0;
935 6 : HCCL_INFO("[Start][Nic]trying to listen on ip[%s] port[%u].", ipAddr.GetReadableAddress(), port);
936 6 : CHK_RET(CheckAutoListenVersion(isAutoPort));
937 6 : HcclResult ret = StartListenSocket(sock.nicSocketHandle, port);
938 6 : CHK_PRT_RET(
939 : ret == HCCL_E_UNAVAIL,
940 : HCCL_INFO(
941 : "[Start][Nic]Could not start listening socket for IP [%s] and port [%u].", ipAddr.GetReadableAddress(),
942 : port),
943 : ret);
944 6 : CHK_PRT_RET(
945 : ret != HCCL_SUCCESS,
946 : HCCL_ERROR(
947 : "[Start][Nic]errNo[0x%016llx] ra inner listen start failed, "
948 : "devid[%u], ip[%s], port[%u], return[%d]",
949 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, ipAddr.GetReadableAddress(), port, ret),
950 : HCCL_E_TCP_CONNECT);
951 6 : HCCL_INFO("port[%u] listen start OK", port);
952 6 : sock.listenedPort.insert(port);
953 6 : HCCL_RUN_INFO(
954 : "[Start][Nic]Listen on ip[%s], port[%u] success, devPhyId[%u], devLogicId[%u], isAutoPort[%d]",
955 : ipAddr.GetReadableAddress(), port, devicePhyId_, deviceLogicId_, isAutoPort);
956 : }
957 6 : int refCount = IPPortListenRefMapDevice_[ipAddr][port].Ref();
958 6 : HCCL_INFO("Nic ip[%s] port[%u] refcount is [%d]", ipAddr.GetReadableAddress(), port, refCount);
959 6 : return HCCL_SUCCESS;
960 : }
961 :
962 2 : void NetworkManager::SetDisableLiteThread(bool disable) { isDisableLiteThread_ = disable; }
963 :
964 0 : HcclResult NetworkManager::CreateNicSocketHandle(const HcclIpAddress& ipAddr)
965 : {
966 0 : CHK_PRT_RET(
967 : !deviceNicInitRef_.Count(),
968 : HCCL_ERROR("[NetworkManager][CreateNicSocketHandle]can't start nic socket before init device nic!"),
969 : HCCL_E_INTERNAL);
970 0 : OccupyIp(ipAddr, raResourceInfo_.nicSocketMap);
971 :
972 0 : IpSocket& sock = raResourceInfo_.nicSocketMap[ipAddr];
973 0 : if (sock.listenedPort.size() == 0) {
974 0 : if (sock.nicSocketHandle == nullptr) {
975 0 : HcclResult ret = InitDeviceSocket(devicePhyId_, ipAddr, sock.nicSocketHandle);
976 0 : CHK_PRT_RET(
977 : ret != HCCL_SUCCESS,
978 : HCCL_ERROR(
979 : "[NetworkManager][CreateNicSocketHandle]errNo[0x%016llx] ra nic init socket failed, devid[%u], "
980 : "return[%d]",
981 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, ret),
982 : ret);
983 : }
984 : } else {
985 0 : HCCL_INFO(
986 : "[NetworkManager][CreateNicSocketHandle] socket has inited, ipAddr[%s], skip.",
987 : ipAddr.GetReadableAddress());
988 : }
989 0 : HCCL_INFO(
990 : "[NetworkManager][CreateNicSocketHandle] CreateNicSocketHandle OK, ipAddr[%s].", ipAddr.GetReadableAddress());
991 0 : return HCCL_SUCCESS;
992 : }
993 :
994 0 : HcclResult NetworkManager::CreateRdmaHandle(
995 : const HcclIpAddress& ipAddr, bool isBackup, NetworkMode netMode, NotifyTypeT notifyType,
996 : HcclNetDevDeployment netDevDeployment)
997 : {
998 : // device侧如果传入的ip是备份ip,那么给InitRDMA传递的两个主备ip(两个相同)都是备份ip hostrdma没有备份 不受影响
999 0 : HcclIpAddress ipAddrBackup(ipAddr.GetFamily(), ipAddr.GetBinaryAddress());
1000 0 : if (!isBackup) {
1001 0 : HCCL_INFO("[NetworkManager][CreateRdmaHandle] ipAddr[%s] is not backup", ipAddr.GetReadableAddress());
1002 0 : ipAddrBackup.clear();
1003 : }
1004 0 : CHK_PRT_RET(
1005 : !(deviceNicInitRef_.Count() || hostNicInitRef_.Count()),
1006 : HCCL_ERROR("[NetworkManager][CreateRdmaHandle]can't start nic socket before init device nic!"),
1007 : HCCL_E_INTERNAL);
1008 0 : switch (netDevDeployment) {
1009 0 : case HcclNetDevDeployment::HCCL_NETDEV_DEPLOYMENT_DEVICE: {
1010 : // ip占位
1011 0 : OccupyIp(ipAddr, raResourceInfo_.nicSocketMap);
1012 : // 初始化socket
1013 0 : IpSocket& sock = raResourceInfo_.nicSocketMap[ipAddr];
1014 0 : if (sock.nicRdmaHandle == nullptr) {
1015 0 : notifyType_ = notifyType;
1016 0 : HcclResult ret = InitRDMA(
1017 0 : devicePhyId_, ipAddr, netMode, notifyType_, sock.nicRdmaHandle, false, false, ipAddrBackup);
1018 0 : CHK_PRT_RET(
1019 : ret != HCCL_SUCCESS,
1020 : HCCL_ERROR(
1021 : "[NetworkManager][CreateRdmaHandle]errNo[0x%016llx] ra nic init rdma failed, devid[%u], "
1022 : "return[%d]",
1023 : HCCL_ERROR_CODE(HCCL_E_NETWORK), devicePhyId_, ret),
1024 : HCCL_E_NETWORK);
1025 : } else {
1026 0 : HCCL_INFO(
1027 : "[NetworkManager][CreateRdmaHandle] requesting not init rdma , or already init rdma, "
1028 : "ipAddr[%s], skip.",
1029 : ipAddr.GetReadableAddress());
1030 : }
1031 : // device-roce需要开启rdmalite
1032 0 : RdmaSupportLite(sock.nicRdmaHandle);
1033 0 : break;
1034 : }
1035 0 : case HcclNetDevDeployment::HCCL_NETDEV_DEPLOYMENT_HOST: {
1036 : // 本ip占位
1037 0 : OccupyIp(ipAddr, raResourceInfo_.hostNetSocketMap);
1038 : // 初始化socket
1039 0 : IpSocket& sock = raResourceInfo_.hostNetSocketMap[ipAddr];
1040 0 : if (sock.nicRdmaHandle == nullptr) {
1041 0 : notifyType_ = NOTIFY;
1042 0 : HcclResult ret = InitRDMA(devicePhyId_, ipAddr, netMode, notifyType_, sock.nicRdmaHandle);
1043 0 : CHK_PRT_RET(
1044 : ret != HCCL_SUCCESS,
1045 : HCCL_ERROR(
1046 : "[Start][Nic]errNo[0x%016llx] ra nic init rdma failed, devid[%u], return[%d]",
1047 : HCCL_ERROR_CODE(HCCL_E_NETWORK), devicePhyId_, ret),
1048 : HCCL_E_NETWORK);
1049 : } else {
1050 0 : HCCL_INFO(
1051 : "[NetworkManager][CreateRdmaHandle] requesting not init rdma , or already init rdma, "
1052 : "ipAddr[%s], skip.",
1053 : ipAddr.GetReadableAddress());
1054 : }
1055 : // host rdma的额外占位
1056 0 : auto temp = raResourceInfo_.nicSocketMap.find(ipAddr);
1057 0 : if (temp != raResourceInfo_.nicSocketMap.end()) {
1058 0 : HCCL_ERROR(
1059 : "[NetworkManager][CreateRdmaHandle] ipAddr[%s] has already occupied.", ipAddr.GetReadableAddress());
1060 0 : return HCCL_E_INTERNAL;
1061 : }
1062 0 : raResourceInfo_.nicSocketMap.insert(std::make_pair(ipAddr, sock));
1063 0 : break;
1064 : }
1065 0 : default: {
1066 0 : HCCL_ERROR(
1067 : "[NetworkManager][CreateRdmaHandle]this Deployment [%u] is not supported, please check the "
1068 : "configuration.",
1069 : netDevDeployment);
1070 0 : return HCCL_E_NOT_SUPPORT;
1071 : }
1072 : }
1073 0 : return HCCL_SUCCESS;
1074 0 : }
1075 :
1076 0 : HcclResult NetworkManager::RdmaSupportLite(RdmaHandle rdmaHandle)
1077 : {
1078 : int supportLite;
1079 0 : HcclResult ret = HrtGetRdmaLiteStatus(rdmaHandle, &supportLite);
1080 0 : CHK_PRT_RET(
1081 : ret != HCCL_SUCCESS,
1082 : HCCL_ERROR(
1083 : "[Get][RdmaLiteStatus]errNo[0x%016llx] get rdma lite status failed, return[%d]",
1084 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), ret),
1085 : HCCL_E_TCP_CONNECT);
1086 0 : isRdmaLiteEn_ = (supportLite == 1);
1087 0 : return HCCL_SUCCESS;
1088 : }
1089 :
1090 0 : HcclResult NetworkManager::StopRdmaHandle(const HcclIpAddress& ipAddr, HcclNetDevDeployment netDevDeployment)
1091 : {
1092 : // rdma没有socket 没有listen
1093 :
1094 : // 销毁socket
1095 0 : switch (netDevDeployment) {
1096 0 : case HcclNetDevDeployment::HCCL_NETDEV_DEPLOYMENT_DEVICE: {
1097 0 : auto it = raResourceInfo_.nicSocketMap.find(ipAddr);
1098 0 : if (it == raResourceInfo_.nicSocketMap.end()) {
1099 0 : HCCL_WARNING(
1100 : "[Stop][NicsSocket]ip[%s] not found in nicSocketMap, may already cleaned.",
1101 : ipAddr.GetReadableAddress());
1102 0 : return HCCL_SUCCESS;
1103 : }
1104 0 : IpSocket& ipSock = it->second;
1105 0 : if (ipSock.nicRdmaHandle != nullptr && HrtRaRdmaDeInit(ipSock.nicRdmaHandle, notifyType_)) {
1106 0 : HCCL_ERROR("[Stop][rmda]NIC rdev deInit not successfully, notifyType_[%d]", notifyType_);
1107 0 : return HCCL_E_NETWORK;
1108 : }
1109 0 : ipSock.nicRdmaHandle = nullptr;
1110 : // 如果该ip下既没有tcp也没有rdma那么移除该ip
1111 0 : if (ipSock.nicSocketHandle == nullptr && ipSock.nicRdmaHandle == nullptr) {
1112 0 : raResourceInfo_.nicSocketMap.erase(ipAddr);
1113 : }
1114 0 : break;
1115 : }
1116 0 : case HcclNetDevDeployment::HCCL_NETDEV_DEPLOYMENT_HOST: {
1117 0 : auto it = raResourceInfo_.hostNetSocketMap.find(ipAddr);
1118 0 : CHK_PRT_RET(
1119 : it == raResourceInfo_.hostNetSocketMap.end(),
1120 : HCCL_ERROR("[Stop][rdma]ip[%s] is not found in hostNetSocketMap.", ipAddr.GetReadableAddress()),
1121 : HCCL_E_INTERNAL);
1122 0 : IpSocket& ipSock = it->second;
1123 0 : if (ipSock.nicRdmaHandle != nullptr && HrtRaRdmaDeInit(ipSock.nicRdmaHandle, notifyType_)) {
1124 0 : HCCL_ERROR("[Stop][rdma]NIC rdev deInit not successfully, notifyType_[%d]", notifyType_);
1125 0 : return HCCL_E_NETWORK;
1126 : }
1127 0 : ipSock.nicRdmaHandle = nullptr;
1128 0 : raResourceInfo_.nicSocketMap.erase(ipAddr); // 移除额外的占位
1129 : // 如果该ip下没有tcp也没有rdma那么移除该ip
1130 0 : if (ipSock.nicSocketHandle == nullptr && ipSock.nicRdmaHandle == nullptr) {
1131 0 : raResourceInfo_.hostNetSocketMap.erase(ipAddr);
1132 : }
1133 0 : break;
1134 : }
1135 0 : default: {
1136 0 : HCCL_ERROR(
1137 : "[NetworkManager][StopRdmaHandle]this Deployment [%u] is not supported, please check the "
1138 : "configuration.",
1139 : netDevDeployment);
1140 0 : return HCCL_E_NOT_SUPPORT;
1141 : }
1142 : }
1143 :
1144 0 : HCCL_INFO(
1145 : "[NetworkManager] [StopRdmaHandle] devicePhyId_[%u] StopRdmaHandle success ip [%s]", devicePhyId_,
1146 : ipAddr.GetReadableAddress());
1147 0 : return HCCL_SUCCESS; // stop socket。port 数清零时自动关闭socket
1148 : }
1149 :
1150 2 : HcclResult NetworkManager::StopNicSocketHandle(const HcclIpAddress& ipAddr)
1151 : {
1152 2 : auto it = raResourceInfo_.nicSocketMap.find(ipAddr);
1153 2 : CHK_PRT_RET(
1154 : it == raResourceInfo_.nicSocketMap.end(),
1155 : HCCL_ERROR(
1156 : "[NetworkManager][StopNicSocketHandle]ip[%s] is not found in nicSocketMap.", ipAddr.GetReadableAddress()),
1157 : HCCL_E_INTERNAL);
1158 0 : IpSocket& ipSock = it->second;
1159 : HcclResult ret;
1160 : // 关闭该ip下的全部端口的listen
1161 0 : for (auto& port : ipSock.listenedPort) {
1162 0 : ret = StopNicsSocketListen(ipAddr, port);
1163 0 : CHK_PRT_RET(
1164 : ret != HCCL_SUCCESS,
1165 : HCCL_ERROR(
1166 : "[NetworkManager][StopNicSocketHandle]errNo[0x%016llx] stop nic listen failed,devid[%u], ip[%s], "
1167 : "port[%u], return[%d]",
1168 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, ipAddr.GetReadableAddress(), port, ret),
1169 : HCCL_E_INTERNAL);
1170 0 : ipSock.listenedPort.erase(port);
1171 0 : IPPortListenRefMapDevice_[ipAddr][port].Clear(); // port计数更新
1172 0 : HCCL_WARNING(
1173 : "[NetworkManager][StopNicSocketHandle] ip [%s] stop listen port [%u] refcount is [%d]",
1174 : ipAddr.GetReadableAddress(), port, IPPortListenRefMapDevice_[ipAddr][port].Count());
1175 : }
1176 :
1177 : // 销毁socket
1178 0 : if (ipSock.nicSocketHandle != nullptr && hrtRaSocketDeInit(ipSock.nicSocketHandle)) {
1179 0 : HCCL_ERROR("[Stop][NicsSocket]NIC socket deInit not successfully");
1180 0 : return HCCL_E_NETWORK;
1181 : }
1182 0 : ipSock.nicSocketHandle = nullptr;
1183 :
1184 : // 如果该ip下既没有tcp也没有rdma那么移除该ip
1185 0 : if (ipSock.nicSocketHandle == nullptr && ipSock.nicRdmaHandle == nullptr) {
1186 0 : raResourceInfo_.nicSocketMap.erase(ipAddr);
1187 : }
1188 :
1189 0 : return HCCL_SUCCESS;
1190 : }
1191 :
1192 36 : HcclResult NetworkManager::StopNic(const HcclIpAddress& ipAddr, u32 port)
1193 : {
1194 36 : auto it = raResourceInfo_.nicSocketMap.find(ipAddr);
1195 36 : CHK_PRT_RET(
1196 : it == raResourceInfo_.nicSocketMap.end(),
1197 : HCCL_ERROR("[Stop][Nic]ip[%s] is not found in nicSocketMap, port[%u].", ipAddr.GetReadableAddress(), port),
1198 : HCCL_E_INTERNAL);
1199 36 : IpSocket& ipSock = it->second;
1200 : HcclResult ret;
1201 : // 传入端口号为无效值0xFFFFFFFF，未启动监听，不需要stop listen
1202 36 : if (port != MAX_VALUE_U32) {
1203 4 : CHK_PRT_RET(
1204 : IPPortListenRefMapDevice_[ipAddr][port].Unref() > 0,
1205 : HCCL_INFO(
1206 : "[Stop][Nic]ip[%s] port[%u] ref[%d] skip stop.", ipAddr.GetReadableAddress(), port,
1207 : IPPortListenRefMapDevice_[ipAddr][port].Count()),
1208 : HCCL_SUCCESS);
1209 4 : ret = StopNicsSocketListen(ipAddr, port);
1210 4 : CHK_PRT_RET(
1211 : ret != HCCL_SUCCESS,
1212 : HCCL_ERROR(
1213 : "[Stop][Nic]errNo[0x%016llx] stop nic socket failed,devid[%u], ip[%s], port[%u], return[%d]",
1214 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, ipAddr.GetReadableAddress(), port, ret),
1215 : HCCL_E_INTERNAL);
1216 4 : ipSock.listenedPort.erase(port);
1217 : }
1218 :
1219 36 : if (ipSock.listenedPort.size() == 0) {
1220 35 : ret = StopNicsSocket(ipAddr);
1221 35 : CHK_PRT_RET(
1222 : ret != HCCL_SUCCESS,
1223 : HCCL_ERROR(
1224 : "[Stop][Nic]errNo[0x%016llx] stop nic socket failed,devid[%u], ip[%s], return[%d]",
1225 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, ipAddr.GetReadableAddress(), ret),
1226 : HCCL_E_INTERNAL);
1227 33 : raResourceInfo_.nicSocketMap.erase(ipAddr);
1228 : }
1229 34 : return HCCL_SUCCESS;
1230 : }
1231 5 : HcclResult NetworkManager::StopAllDeviceNicSockets()
1232 : {
1233 : HcclResult ret;
1234 11 : for (auto& itSocket : raResourceInfo_.nicSocketMap) {
1235 6 : std::set<u32> listenedPorts = itSocket.second.listenedPort;
1236 7 : for (auto itPort : listenedPorts) {
1237 3 : ret = StopNicsSocketListen(itSocket.first, itPort);
1238 3 : if (ret != HCCL_SUCCESS) {
1239 2 : HCCL_ERROR(
1240 : "[Stop][AllDeviceNicSockets]errNo[0x%016llx] stop nic socket failed,devid[%u],ip[%s], "
1241 : "port[%u],return[%d]",
1242 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, itSocket.first.GetReadableAddress(), itPort,
1243 : ret);
1244 2 : itSocket.second.listenedPort.erase(itPort);
1245 2 : break;
1246 : }
1247 : }
1248 6 : ret = StopNicsSocket(itSocket.first);
1249 6 : if (ret != HCCL_SUCCESS) {
1250 2 : HCCL_ERROR(
1251 : "[Stop][AllDeviceNicSockets]errNo[0x%016llx] stop nic socket failed,devid[%u],ip[%s],return[%d]",
1252 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, itSocket.first.GetReadableAddress(), ret);
1253 2 : if (itSocket.second.nicRdmaHandle != nullptr) {
1254 0 : HCCL_ERROR("[StopVnicSocketHandle] itSocket.second.nicRdmaHandle is not nullptr.");
1255 0 : (void)HrtRaRdmaDeInit(itSocket.second.nicRdmaHandle, notifyType_);
1256 0 : itSocket.second.nicRdmaHandle = nullptr;
1257 : }
1258 2 : if (itSocket.second.nicSocketHandle != nullptr) {
1259 2 : HCCL_ERROR("[StopVnicSocketHandle] itSocket.second.nicSocketHandle is not nullptr.");
1260 2 : (void)HrtRaRdmaDeInit(itSocket.second.nicSocketHandle, notifyType_);
1261 2 : itSocket.second.nicSocketHandle = nullptr;
1262 : }
1263 : }
1264 6 : }
1265 :
1266 5 : raResourceInfo_.nicSocketMap.clear();
1267 5 : return HCCL_SUCCESS;
1268 : }
1269 :
1270 7 : HcclResult NetworkManager::StopAllDeviceVnicSockets()
1271 : {
1272 : HcclResult ret;
1273 :
1274 14 : for (auto itSocket : raResourceInfo_.vnicSocketMap) {
1275 7 : HCCL_WARNING(
1276 : "vnicSocketMap ip[%s] is not released when NetworkManager Destroy, force releasing",
1277 : itSocket.first.GetReadableAddress());
1278 7 : if (itSocket.second.nicSocketHandle != nullptr) {
1279 14 : for (auto itPort : itSocket.second.listenedPort) {
1280 7 : ret = StopListenSocket(itSocket.second.nicSocketHandle, itPort);
1281 7 : if (ret != HCCL_SUCCESS) {
1282 0 : HCCL_ERROR(
1283 : "[Stop][AllDeviceVnicSockets]errNo[0x%016llx] stop vnic socket listen failed, "
1284 : "devid[%u], ip[%s], port[%u], return[%d]",
1285 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, itSocket.first.GetReadableAddress(), itPort,
1286 : ret);
1287 : }
1288 7 : IPPortListenRefMapVnicDevice_[itSocket.first][itPort].Clear();
1289 : }
1290 7 : ret = hrtRaSocketDeInit(itSocket.second.nicSocketHandle);
1291 7 : if (ret != HCCL_SUCCESS) {
1292 1 : HCCL_ERROR(
1293 : "[Stop][AllDeviceVnicSockets]errNo[0x%016llx] deinit vnic socket failed, "
1294 : "devid[%u], ip[%s], return[%d]",
1295 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, itSocket.first.GetReadableAddress(), ret);
1296 : }
1297 7 : itSocket.second.nicSocketHandle = nullptr;
1298 : }
1299 7 : }
1300 7 : raResourceInfo_.vnicSocketMap.clear();
1301 :
1302 7 : return HCCL_SUCCESS;
1303 : }
1304 :
1305 34 : HcclResult NetworkManager::InitRDMA(
1306 : u32 devicePhysicID, const HcclIpAddress& ipAddr, NetworkMode netMode, NotifyTypeT notifyType,
1307 : RdmaHandle& rdmaHandle, bool disabledLiteThread, bool enable910ALite, HcclIpAddress ipAddrBackup)
1308 : {
1309 : struct rdev nicRdevInfo;
1310 34 : nicRdevInfo.phyId = devicePhysicID;
1311 34 : nicRdevInfo.family = ipAddr.GetFamily();
1312 34 : nicRdevInfo.localIp.addr = ipAddr.GetBinaryAddress().addr;
1313 34 : nicRdevInfo.localIp.addr6 = ipAddr.GetBinaryAddress().addr6;
1314 :
1315 34 : struct RdevInitInfo init_info = {};
1316 34 : init_info.mode = netMode;
1317 34 : init_info.notifyType = notifyType;
1318 34 : init_info.disabledLiteThread = disabledLiteThread;
1319 34 : init_info.enabled910aLite = enable910ALite;
1320 34 : init_info.enabled2mbLite = GetExternalInputRdmaFastPost();
1321 :
1322 : HcclResult ret;
1323 34 : HCCL_DEBUG("isRaInitRepeated_[%d]", isRaInitRepeated_);
1324 34 : if (isRaInitRepeated_) {
1325 : // 重复RaInit时，调用此接口获取相同的rdmaHandle，防止重新生成
1326 0 : ret = HrtRaRdmaGetHandle(devicePhysicID, rdmaHandle);
1327 : } else {
1328 34 : if (!ipAddrBackup.IsInvalid()) {
1329 : struct rdev nicRdevInfoback;
1330 0 : CHK_RET(hrtGetPairDevicePhyId(devicePhysicID, nicRdevInfoback.phyId));
1331 0 : nicRdevInfoback.family = ipAddrBackup.GetFamily();
1332 0 : nicRdevInfoback.localIp.addr = ipAddrBackup.GetBinaryAddress().addr;
1333 0 : nicRdevInfoback.localIp.addr6 = ipAddrBackup.GetBinaryAddress().addr6;
1334 0 : HCCL_DEBUG(
1335 : "[%s]backup rdev info: ipAddr[%s], ipAddrBackup[%s]", __func__, ipAddr.GetReadableIP(),
1336 : ipAddrBackup.GetReadableIP());
1337 0 : ret = HrtRdmaInitWithBackupAttr(init_info, nicRdevInfo, nicRdevInfoback, rdmaHandle);
1338 : } else {
1339 34 : ret = HrtRaRdmaInitWithAttr(init_info, nicRdevInfo, rdmaHandle);
1340 : }
1341 : }
1342 :
1343 34 : CHK_PRT_RET(
1344 : ret != HCCL_SUCCESS,
1345 : HCCL_ERROR(
1346 : "[Init][RDMA]errNo[0x%016llx] ra rdma init failed, devid[%u] ip[%s], notifyType[%d], return[%d]",
1347 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhysicID, ipAddr.GetReadableAddress(), notifyType, ret),
1348 : HCCL_E_TCP_CONNECT);
1349 34 : HCCL_INFO(
1350 : "devicePhyId[%u], ip[%s] disabledLiteThread[%u] enabled910aLite[%u] rdmaHandle[%p] rdma init OK",
1351 : devicePhysicID, ipAddr.GetReadableAddress(), disabledLiteThread, enable910ALite, rdmaHandle);
1352 :
1353 34 : return HCCL_SUCCESS;
1354 : }
1355 :
1356 153 : bool NetworkManager::GetRdmaLiteStatus() { return isRdmaLiteEn_; }
1357 :
1358 34 : HcclResult NetworkManager::GetNotifyType(NotifyTypeT& notifyType) const
1359 : {
1360 : DevType deviceType;
1361 34 : CHK_RET(hrtGetDeviceType(deviceType));
1362 34 : if (deviceType == DevType::DEV_TYPE_910 || deviceType == DevType::DEV_TYPE_910B
1363 0 : || deviceType == DevType::DEV_TYPE_910_93) {
1364 34 : notifyType = NOTIFY;
1365 0 : } else if (deviceType == DevType::DEV_TYPE_310P3 || deviceType == DevType::DEV_TYPE_310P1) {
1366 0 : notifyType = EVENTID;
1367 : } else {
1368 0 : HCCL_ERROR("[Init][DeviceRDMA]devType[%d] is invalid", deviceType);
1369 0 : return HCCL_E_PARA;
1370 : }
1371 34 : return HCCL_SUCCESS;
1372 : }
1373 :
1374 85 : void NetworkManager::GetNetworkMode(NetworkMode& netMode) const
1375 : {
1376 85 : if (Is310PDevice()) {
1377 0 : netMode = NETWORK_PEER_ONLINE;
1378 : } else {
1379 85 : netMode = NETWORK_OFFLINE;
1380 : }
1381 85 : }
1382 :
1383 51 : HcclResult NetworkManager::InitDeviceSocket(u32 devicePhysicID, const HcclIpAddress& ipAddr, SocketHandle& socketHandle)
1384 : {
1385 : struct rdev nicRdevInfo;
1386 51 : nicRdevInfo.phyId = devicePhysicID;
1387 51 : nicRdevInfo.family = ipAddr.GetFamily();
1388 51 : nicRdevInfo.localIp.addr = ipAddr.GetBinaryAddress().addr;
1389 51 : nicRdevInfo.localIp.addr6 = ipAddr.GetBinaryAddress().addr6;
1390 :
1391 : NetworkMode netMode;
1392 51 : GetNetworkMode(netMode);
1393 :
1394 51 : HcclResult ret = hrtRaSocketInit(netMode, nicRdevInfo, socketHandle);
1395 51 : CHK_PRT_RET(
1396 : ret != HCCL_SUCCESS,
1397 : HCCL_ERROR(
1398 : "[Init][DeviceSocket]ra socket init failed, network mode[%d] devid[%u] ip[%s], return[%d]", netMode,
1399 : devicePhysicID, ipAddr.GetReadableAddress(), ret),
1400 : HCCL_E_TCP_CONNECT);
1401 51 : HCCL_INFO("devicePhyId[%u], ip[%s] socket init OK", devicePhysicID, ipAddr.GetReadableAddress());
1402 :
1403 51 : return HCCL_SUCCESS;
1404 : }
1405 :
1406 : HcclResult
1407 22 : NetworkManager::StartHostNetAndListen(const HcclIpAddress& ipAddr, SocketHandle& socketHandle, u32& port, bool rdmaFlag)
1408 : {
1409 22 : CHK_PRT_RET(
1410 : (hostNicInitRef_.Count() == 0),
1411 : HCCL_ERROR("[Start][HostNetAndListen]cannot start nic socket before host nic inited!"), HCCL_E_INTERNAL);
1412 22 : auto sockInfo = raResourceInfo_.hostNetSocketMap.find(ipAddr);
1413 22 : if (sockInfo == raResourceInfo_.hostNetSocketMap.end()) {
1414 2 : IpSocket tempSock;
1415 2 : raResourceInfo_.hostNetSocketMap.insert(std::make_pair(ipAddr, tempSock)); // 本IP占位
1416 2 : HCCL_INFO("device[%u] Start host nic insert Ip[%s]", devicePhyId_, ipAddr.GetReadableAddress());
1417 2 : }
1418 :
1419 : HcclResult ret;
1420 22 : IpSocket& sock = raResourceInfo_.hostNetSocketMap[ipAddr];
1421 22 : if (sock.nicSocketHandle == nullptr) {
1422 2 : ret = InitHostSocket(ipAddr, sock.nicSocketHandle);
1423 2 : CHK_PRT_RET(
1424 : ret != HCCL_SUCCESS,
1425 : HCCL_ERROR(
1426 : "[Start][HostNetAndListen]start host socket failed, devid[%u], ip[%s] return[%d]", devicePhyId_,
1427 : ipAddr.GetReadableAddress(), ret),
1428 : ret);
1429 : }
1430 22 : if (sock.nicRdmaHandle == nullptr && rdmaFlag) {
1431 0 : notifyType_ = EVENTID;
1432 0 : ret = InitRDMA(devicePhyId_, ipAddr, NETWORK_PEER_ONLINE, notifyType_, sock.nicRdmaHandle);
1433 0 : CHK_PRT_RET(
1434 : ret != HCCL_SUCCESS,
1435 : HCCL_ERROR(
1436 : "[Start][Nic]errNo[0x%016llx] ra nic init rdma failed, devid[%u], return[%d]",
1437 : HCCL_ERROR_CODE(HCCL_E_NETWORK), devicePhyId_, ret),
1438 : HCCL_E_NETWORK);
1439 : }
1440 22 : if (IPPortListenRefMapHost_[ipAddr][port].Count() == 0 && sock.listenedPort.find(port) == sock.listenedPort.end()) {
1441 21 : bool isAutoPort = port == 0;
1442 21 : HCCL_INFO("[Start][HostNetAndListen]trying to listen on ip[%s] port[%u].", ipAddr.GetReadableAddress(), port);
1443 21 : ret = StartListenSocket(sock.nicSocketHandle, port);
1444 21 : CHK_PRT_RET(
1445 : ret == HCCL_E_UNAVAIL,
1446 : HCCL_INFO(
1447 : "[Start][HostNetAndListen]Could not start listening socket for IP [%s] and port [%u].",
1448 : ipAddr.GetReadableAddress(), port),
1449 : ret);
1450 21 : CHK_PRT_RET(
1451 : ret != HCCL_SUCCESS,
1452 : HCCL_ERROR(
1453 : "[Start][HostNetAndListen]start host socket failed, devid[%u], ip[%s], port[%u], return[%d]",
1454 : devicePhyId_, ipAddr.GetReadableAddress(), port, ret),
1455 : ret);
1456 21 : sock.listenedPort.insert(port);
1457 21 : HCCL_RUN_INFO(
1458 : "[Start][HostNetAndListen]Listen on ip[%s], port[%u] success, "
1459 : "devPhyId[%u], devLogicId[%u], isAutoPort[%d]",
1460 : ipAddr.GetReadableAddress(), port, devicePhyId_, deviceLogicId_, isAutoPort);
1461 : }
1462 22 : int refCount = IPPortListenRefMapHost_[ipAddr][port].Ref();
1463 22 : HCCL_INFO("host ip[%s] port[%u] refcount is [%d]", ipAddr.GetReadableAddress(), port, refCount);
1464 :
1465 22 : socketHandle = sock.nicSocketHandle;
1466 22 : raResourceInfo_.nicSocketMap.insert(std::make_pair(ipAddr, sock));
1467 22 : hostNicSocketClientRef_[ipAddr].Ref();
1468 22 : HCCL_INFO("HostNet, ip[%s] port[%u] socket init OK", ipAddr.GetReadableAddress(), port);
1469 22 : return HCCL_SUCCESS;
1470 : }
1471 :
1472 45 : HcclResult NetworkManager::CheckSocketInfo(const SocketHandle socketHandle, const HcclIpAddress& ipAddr, u32 port) const
1473 : {
1474 45 : auto iterIP = raResourceInfo_.hostNetSocketMap.find(ipAddr);
1475 45 : CHK_PRT_RET(
1476 : (iterIP == raResourceInfo_.hostNetSocketMap.end()),
1477 : HCCL_ERROR(
1478 : "[Check][SocketInfo]ip[%s] port[%u] has not been started. ip is invalid.", ipAddr.GetReadableAddress(),
1479 : port),
1480 : HCCL_E_INTERNAL);
1481 :
1482 45 : CHK_PRT_RET(
1483 : (socketHandle != iterIP->second.nicSocketHandle),
1484 : HCCL_ERROR(
1485 : "[Check][SocketInfo]ip[%s] port[%u] has not been started. socketHandle is invalid",
1486 : ipAddr.GetReadableAddress(), port),
1487 : HCCL_E_INTERNAL);
1488 :
1489 45 : if (port != NO_LISTEN_PORT) {
1490 20 : CHK_PRT_RET(
1491 : (iterIP->second.listenedPort.count(port) == 0),
1492 : HCCL_ERROR(
1493 : "[Check][SocketInfo]ip[%s] port[%u] has not been started. port is invalid", ipAddr.GetReadableAddress(),
1494 : port),
1495 : HCCL_E_INTERNAL);
1496 : }
1497 45 : return HCCL_SUCCESS;
1498 : }
1499 :
1500 20 : HcclResult NetworkManager::StopHostNetAndListen(SocketHandle socketHandle, const HcclIpAddress& ipAddr, u32 port)
1501 : {
1502 20 : std::unique_lock<std::mutex> lock(raLock_);
1503 20 : CHK_PRT_RET(
1504 : (hostNicInitRef_.Count() == 0),
1505 : HCCL_ERROR("[Stop][HostNetAndListen]cannot start nic socket before host nic inited!"), HCCL_E_INTERNAL);
1506 :
1507 20 : CHK_RET(CheckSocketInfo(socketHandle, ipAddr, port));
1508 :
1509 20 : if (IPPortListenRefMapHost_[ipAddr][port].Unref() == 0) {
1510 20 : CHK_RET(StopListenSocket(socketHandle, port)); /* 当前只拉起一个server */
1511 :
1512 20 : HCCL_INFO("ip[%s] port[%u] stop success.", ipAddr.GetReadableAddress(), port);
1513 20 : raResourceInfo_.hostNetSocketMap[ipAddr].listenedPort.erase(port);
1514 20 : raResourceInfo_.nicSocketMap[ipAddr].listenedPort.erase(port);
1515 : } else {
1516 0 : HCCL_INFO(
1517 : "ip[%s] port[%u] skip stop. ref[%d].", ipAddr.GetReadableAddress(), port,
1518 : IPPortListenRefMapHost_[ipAddr][port].Count());
1519 : }
1520 :
1521 20 : if (hostNicSocketClientRef_[ipAddr].Unref() == 0
1522 20 : && raResourceInfo_.hostNetSocketMap[ipAddr].listenedPort.size() == 0) {
1523 0 : CHK_RET(hrtRaSocketDeInit(socketHandle));
1524 0 : raResourceInfo_.hostNetSocketMap.erase(ipAddr);
1525 0 : raResourceInfo_.nicSocketMap.erase(ipAddr);
1526 0 : HCCL_INFO("ip[%s] port[%u] deinit success.", ipAddr.GetReadableAddress(), port);
1527 : }
1528 :
1529 20 : return HCCL_SUCCESS;
1530 20 : }
1531 2 : HcclResult NetworkManager::StopAllHostNicSockets()
1532 : {
1533 : HcclResult ret;
1534 4 : for (auto itSocket : raResourceInfo_.hostNetSocketMap) {
1535 3 : for (auto itPort : itSocket.second.listenedPort) {
1536 1 : ret = (StopListenSocket(itSocket.second.nicSocketHandle, itPort));
1537 1 : if (ret != HCCL_SUCCESS) {
1538 1 : HCCL_ERROR(
1539 : "[Stop][StopAllHostNicSockets]errNo[0x%016llx] stop listen socket failed,"
1540 : "devid[%u], ip[%s], port[%u], return[%d].",
1541 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, itSocket.first.GetReadableAddress(), itPort,
1542 : ret);
1543 : } else {
1544 0 : HCCL_INFO("ip[%s] port[%u] stop success.", itSocket.first.GetReadableAddress(), itPort);
1545 : }
1546 : }
1547 2 : ret = hrtRaSocketDeInit(itSocket.second.nicSocketHandle);
1548 2 : if (ret != HCCL_SUCCESS) {
1549 1 : HCCL_ERROR(
1550 : "[Stop][StopAllHostNicSockets]errNo[0x%016llx] deinit socket failed, devid[%u], ip[%s], return[%d]",
1551 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), devicePhyId_, itSocket.first.GetReadableAddress(), ret);
1552 : } else {
1553 1 : HCCL_INFO("ip[%s] deinit success.", itSocket.first.GetReadableAddress());
1554 : }
1555 2 : hostNicSocketClientRef_[itSocket.first].Clear();
1556 2 : }
1557 :
1558 2 : raResourceInfo_.hostNetSocketMap.clear();
1559 2 : return HCCL_SUCCESS;
1560 : }
1561 : /* destroy network resource */
1562 1513 : HcclResult NetworkManager::Destroy()
1563 : {
1564 : /* 停止nic ra的监听 */
1565 1513 : if (raResourceInfo_.nicSocketMap.size() != 0) {
1566 7 : for (auto it : raResourceInfo_.nicSocketMap) {
1567 4 : HCCL_WARNING("nicSocketMap[%s] is not stopped when NetworkManager Destroy", it.first.GetReadableAddress());
1568 4 : }
1569 : // StartHostNetAndListen等函数中，同一地址在nicSocketMap和hostNetSocketMap内同时插入
1570 : // 此处StopAllDeviceNicSockets()销毁nicSocketMap后，StopAllHostNicSockets内会发生重复销毁导致core
1571 : // 为了避免此种情况，同时尽量减少对既有函数的修改扩散影响，此处同步对hostNetSocketMap进行清理
1572 : // 最终修改方案需要重构本类，解除两个MAP的耦合
1573 :
1574 7 : for (auto& it : raResourceInfo_.nicSocketMap) {
1575 4 : raResourceInfo_.hostNetSocketMap.erase(it.first); // key不存在则不会删除
1576 : }
1577 3 : StopAllDeviceNicSockets();
1578 : }
1579 :
1580 : /* 停止vnic ra的监听 */
1581 1513 : if (raResourceInfo_.vnicSocketMap.size() != 0) {
1582 5 : StopAllDeviceVnicSockets();
1583 : }
1584 :
1585 : /* 停止host nic ra的监听 */
1586 1513 : if (raResourceInfo_.hostNetSocketMap.size() != 0) {
1587 0 : for (auto it : raResourceInfo_.hostNetSocketMap) {
1588 0 : HCCL_WARNING("hostNicSocketMap[%s] is not stop when NetworkManager Destroy", it.first.GetReadableAddress());
1589 0 : }
1590 0 : StopAllHostNicSockets();
1591 : }
1592 :
1593 : /* 释放ra资源 重复RaInit时，不再调用内部DeInit */
1594 1513 : HCCL_DEBUG("Destroy call HrtRaDeInit.");
1595 1513 : if (deviceNicInitRef_.Count() != 0 && !isRaDeInit_) {
1596 5 : HCCL_WARNING("device Nic is not deinit when NetworkManager Destroy. ref[%d]", deviceNicInitRef_.Count());
1597 5 : struct RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, isEnableHdcAsync_};
1598 5 : GetDeviceRaInitConfig(config);
1599 5 : config.nicPosition = static_cast<u32>(NICDeployment::NIC_DEPLOYMENT_DEVICE);
1600 5 : if (HrtRaDeInit(&config) != HCCL_SUCCESS) {
1601 0 : HCCL_ERROR("ra deinit failed. para: nicdeploy[%u], phyId[%u]", config.nicPosition, config.phyId);
1602 : }
1603 :
1604 5 : isRaDeInit_ = true;
1605 5 : deviceNicInitRef_.Clear();
1606 : }
1607 1513 : if (hostNicInitRef_.Count() != 0) {
1608 6 : HCCL_WARNING("host Nic is not deinit when NetworkManager Destroy. ref[%d]", hostNicInitRef_.Count());
1609 6 : struct RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, isEnableHdcAsync_};
1610 6 : config.nicPosition = static_cast<u32>(NICDeployment::NIC_DEPLOYMENT_HOST);
1611 6 : config.phyId = devicePhyId_;
1612 6 : if (HrtRaDeInit(&config) != HCCL_SUCCESS) {
1613 1 : HCCL_ERROR("ra deinit failed. para: nicdeploy[%u], phyId[%u]", config.nicPosition, config.phyId);
1614 : }
1615 6 : hostNicInitRef_.Clear();
1616 : }
1617 1513 : HCCL_INFO("destroy all nic/vnic.");
1618 1513 : return HCCL_SUCCESS;
1619 : }
1620 :
1621 6 : HcclResult NetworkManager::StopNicsSocketListen(const HcclIpAddress& ipAddr, u32 port)
1622 : {
1623 6 : auto it = raResourceInfo_.nicSocketMap.find(ipAddr);
1624 6 : CHK_PRT_RET(
1625 : it == raResourceInfo_.nicSocketMap.end(),
1626 : HCCL_ERROR(
1627 : "[Stop][NicsSocketPort]ip[%s] port[%u] is not found in nicSocketMap.", ipAddr.GetReadableAddress(), port),
1628 : HCCL_E_INTERNAL);
1629 6 : IpSocket& ipSock = it->second;
1630 6 : bool portFound = false;
1631 13 : for (auto itPort : ipSock.listenedPort) {
1632 7 : if (itPort == port) {
1633 6 : portFound = true;
1634 : }
1635 : }
1636 6 : CHK_PRT_RET(
1637 : !portFound,
1638 : HCCL_ERROR("[Stop][NicsSocketPort]PORT(ip[%s] port[%u]) is not found.", ipAddr.GetReadableAddress(), port),
1639 : HCCL_E_INTERNAL);
1640 6 : if (ipSock.nicSocketHandle != nullptr && StopListenSocket(ipSock.nicSocketHandle, port)) {
1641 1 : HCCL_ERROR("[Stop][NicsSocketPort]NIC socket listen is not stopped successfully");
1642 1 : return HCCL_E_NETWORK;
1643 : }
1644 5 : return HCCL_SUCCESS;
1645 : }
1646 :
1647 40 : HcclResult NetworkManager::StopNicsSocket(const HcclIpAddress& ipAddr)
1648 : {
1649 40 : auto it = raResourceInfo_.nicSocketMap.find(ipAddr);
1650 40 : CHK_PRT_RET(
1651 : it == raResourceInfo_.nicSocketMap.end(),
1652 : HCCL_ERROR("[Stop][NicsSocket]ip[%s] is not found in nicSocketMap.", ipAddr.GetReadableAddress()),
1653 : HCCL_E_INTERNAL);
1654 40 : IpSocket& ipSock = it->second;
1655 40 : if (ipSock.nicRdmaHandle != nullptr && HrtRaRdmaDeInit(ipSock.nicRdmaHandle, notifyType_)) {
1656 2 : HCCL_ERROR("[Stop][NicsSocket]NIC rdev deInit not successfully, notifyType_[%d]", notifyType_);
1657 2 : return HCCL_E_NETWORK;
1658 : }
1659 38 : ipSock.nicRdmaHandle = nullptr;
1660 38 : if (ipSock.nicSocketHandle != nullptr && hrtRaSocketDeInit(ipSock.nicSocketHandle)) {
1661 1 : HCCL_ERROR("[Stop][NicsSocket]NIC socket deInit not successfully");
1662 1 : return HCCL_E_NETWORK;
1663 : }
1664 37 : ipSock.nicSocketHandle = nullptr;
1665 37 : return HCCL_SUCCESS;
1666 : }
1667 :
1668 : HcclResult
1669 0 : NetworkManager::InitRdmaHandle(u32 devId, const HcclIpAddress& ipAddr, bool disabledLiteThread, bool enable910ALite)
1670 : {
1671 0 : if (raResourceInfo_.nicSocketMap[ipAddr].nicRdmaHandle != nullptr
1672 0 : && raResourceInfo_.hostNetSocketMap[ipAddr].nicRdmaHandle != nullptr) {
1673 0 : HCCL_INFO("NetworkManager: RdmaInit already nic");
1674 0 : return HCCL_SUCCESS;
1675 : }
1676 :
1677 0 : u32 devicePhyId = ((static_cast<s32>(devId) == HOST_DEVICE_ID) ? 0 : devId);
1678 0 : RdmaHandle rdmaHandle = nullptr;
1679 : // 模式和notify类型按照是否为hdc模式进行赋值
1680 0 : NetworkMode initRdmaMode = (isHostUseDevNic_) ? NETWORK_OFFLINE : NETWORK_PEER_ONLINE;
1681 0 : NotifyTypeT notifyType = (isHostUseDevNic_) ? NotifyTypeT::NOTIFY : NotifyTypeT::NO_USE;
1682 : HcclResult ret
1683 0 : = InitRDMA(devicePhyId, ipAddr, initRdmaMode, notifyType, rdmaHandle, disabledLiteThread, enable910ALite);
1684 0 : CHK_PRT_RET(
1685 : ret != HCCL_SUCCESS,
1686 : HCCL_ERROR(
1687 : "[Init][RDMA]errNo[0x%016llx] ra rdma init failed, ip[%s], return[%d]", HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT),
1688 : ipAddr.GetReadableAddress(), ret),
1689 : HCCL_E_TCP_CONNECT);
1690 0 : HCCL_INFO("ip[%s] rdma init OK", ipAddr.GetReadableAddress());
1691 0 : CHK_PTR_NULL(rdmaHandle);
1692 0 : raResourceInfo_.nicSocketMap[ipAddr].nicRdmaHandle = rdmaHandle;
1693 0 : raResourceInfo_.hostNetSocketMap[ipAddr].nicRdmaHandle = rdmaHandle;
1694 0 : return HCCL_SUCCESS;
1695 : }
1696 :
1697 238 : HcclResult NetworkManager::GetRaResourceInfo(RaResourceInfo& raResourceInfo)
1698 : {
1699 238 : raResourceInfo = raResourceInfo_;
1700 238 : return HCCL_SUCCESS;
1701 : }
1702 :
1703 1 : HcclResult NetworkManager::PsWorkerRaInit(
1704 : u32 devId, const HcclIpAddress& ipAddr, u32 port, bool isHostUseDevNic, bool remoteIsHdc, bool isBoardVersion)
1705 : {
1706 1 : HCCL_INFO("PsWorkerRaInit, devicePhyId[%u], deviceLogicId_[%d]", devicePhyId_, deviceLogicId_);
1707 : // 引用计数
1708 1 : u32 devicePhyId = ((static_cast<s32>(devId) == HOST_DEVICE_ID) ? 0 : devId);
1709 1 : std::string ipAddrStr(ipAddr.GetReadableAddress());
1710 1 : if (ipAddrStr == "127.0.0.1") {
1711 0 : hostNicInitRef_.Ref();
1712 0 : HCCL_INFO("hostNicInitRef_[%d]", hostNicInitRef_.Count());
1713 : } else {
1714 1 : bool fistUsed{false};
1715 1 : deviceNicInitRef_.Ref();
1716 1 : CHK_RET(hrtRaIsFirstUsed(devicePhyId, fistUsed));
1717 1 : HCCL_INFO(
1718 : "deviceNicInitRef_[%d] fistUsed[%u] devicePhyId[%u]", deviceNicInitRef_.Count(), fistUsed, devicePhyId);
1719 1 : if (deviceNicInitRef_.Count() == 1 && !fistUsed) {
1720 0 : isRaInitRepeated_ = true;
1721 1 : } else if (!fistUsed) {
1722 0 : HCCL_INFO("[NetworkManager] PsWorkerRa is not fistUsed");
1723 0 : return HCCL_SUCCESS;
1724 : }
1725 : }
1726 :
1727 1 : CHK_PRT_RET(
1728 : (raResourceInfo_.nicSocketMap.count(ipAddr) != 0),
1729 : HCCL_INFO("NetworkManager: PsWorkerRa already Init, ipAddr[%s]", ipAddr.GetReadableAddress()), HCCL_SUCCESS);
1730 1 : isHostUseDevNic_ = isHostUseDevNic;
1731 :
1732 : // hdc模式下需要先拉起device上的hccp进程
1733 1 : if (isHostUseDevNic_) {
1734 : // 拉起device进程
1735 0 : if (devId != INVALID_UINT) {
1736 0 : devicePhyId_ = devId;
1737 : } else {
1738 : // 初始化ra资源(dev信息带入逻辑ID)
1739 0 : CHK_RET(hrtGetDevice(&deviceLogicId_));
1740 0 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId_), devicePhyId_));
1741 : }
1742 :
1743 : // device 网卡初始化前需要拉起 hccp .
1744 : rtNetServiceOpenArgs openArgs;
1745 0 : rtProcExtParam extParam{};
1746 0 : std::string extPam("--hdcType=" + std::to_string(PID_HDC_TYPE));
1747 :
1748 0 : extParam.paramInfo = extPam.c_str();
1749 0 : extParam.paramLen = extPam.size();
1750 0 : openArgs.extParamList = &extParam;
1751 0 : openArgs.extParamCnt = 1UL;
1752 0 : isTsdProcessOpen_ = true;
1753 0 : CHK_RET(hrtOpenNetService(&openArgs));
1754 0 : HCCL_INFO(
1755 : "NetworkManager open tsd success, devicePhyId[%u], deviceLogicId_[%d], subPid[%lld]", devicePhyId_,
1756 : deviceLogicId_, static_cast<s64>(subPid_));
1757 0 : }
1758 :
1759 1 : bool isOpenWhiteList = false;
1760 1 : if (!isBoardVersion && remoteIsHdc && IsGeneralServer()) {
1761 0 : HCCL_INFO("general server, ps open WhiteList");
1762 0 : isOpenWhiteList = true;
1763 : }
1764 :
1765 1 : CHK_RET(hrtRaSocketSetWhiteListStatus(static_cast<u32>(isOpenWhiteList)));
1766 :
1767 1 : RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, false};
1768 1 : config.phyId = devicePhyId;
1769 1 : if (ipAddrStr == "127.0.0.1") {
1770 0 : config.nicPosition = static_cast<u32>(NICDeployment::NIC_DEPLOYMENT_HOST);
1771 : } else {
1772 1 : config.nicPosition = static_cast<u32>(isHostUseDevNic_);
1773 : }
1774 1 : if (isHostUseDevNic_) {
1775 0 : config.hdcType = PID_HDC_TYPE;
1776 : }
1777 :
1778 1 : if (!isRaInitRepeated_) {
1779 1 : HCCL_INFO(
1780 : "PsWorkerRaInit call HrtRaInit. devicePhyId[%u] isRaInitRepeated_[%u]", devicePhyId_, isRaInitRepeated_);
1781 1 : HcclResult hcclRet = HrtRaInit(&config);
1782 1 : if (hcclRet != HCCL_SUCCESS) {
1783 0 : HCCL_ERROR("[PsWorkerRaInit] ra init failed, ret[%d].", hcclRet);
1784 0 : if (isTsdProcessOpen_) {
1785 0 : (void)hrtCloseNetService();
1786 0 : isTsdProcessOpen_ = false;
1787 : }
1788 0 : return hcclRet;
1789 : }
1790 : }
1791 :
1792 1 : struct rdev nicRdevInfo = {};
1793 1 : nicRdevInfo.phyId = devicePhyId;
1794 1 : nicRdevInfo.family = ipAddr.GetFamily();
1795 1 : nicRdevInfo.localIp.addr = ipAddr.GetBinaryAddress().addr;
1796 1 : nicRdevInfo.localIp.addr6 = ipAddr.GetBinaryAddress().addr6;
1797 1 : SocketHandle socketHandle = nullptr;
1798 1 : NetworkMode raSocketInitMode = (isHostUseDevNic_) ? NETWORK_OFFLINE : NETWORK_PEER_ONLINE;
1799 1 : HcclResult ret = hrtRaSocketInit(raSocketInitMode, nicRdevInfo, socketHandle);
1800 1 : if (ret != HCCL_SUCCESS) {
1801 1 : HCCL_ERROR(
1802 : "[Init][HostSocket]errNo[0x%016llx] ra socket init failed, ip[%s], return[%d]",
1803 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), ipAddr.GetReadableAddress(), ret);
1804 1 : if (!isRaInitRepeated_) {
1805 1 : (void)HrtRaDeInit(&config);
1806 : }
1807 1 : if (isTsdProcessOpen_) {
1808 0 : (void)hrtCloseNetService();
1809 0 : isTsdProcessOpen_ = false;
1810 : }
1811 1 : return HCCL_E_TCP_CONNECT;
1812 : }
1813 0 : HCCL_INFO(
1814 : "ip[%s] socket init OK, devicePhyId_[%u], socketHandle[%llu]", ipAddr.GetReadableAddress(), devicePhyId_,
1815 : hash<void*>{}(socketHandle));
1816 :
1817 0 : IpSocket ipSocketInfo;
1818 0 : ipSocketInfo.nicSocketHandle = socketHandle;
1819 :
1820 0 : raResourceInfo_.nicSocketMap.insert(std::make_pair(ipAddr, ipSocketInfo));
1821 0 : raResourceInfo_.hostNetSocketMap.insert(std::make_pair(ipAddr, ipSocketInfo));
1822 :
1823 0 : ret = HeterogStartListen(ipAddr, port);
1824 0 : if (ret != HCCL_SUCCESS) {
1825 0 : HCCL_ERROR("[PsWorkerRaInit] HeterogStartListen failed, ret[%d]", ret);
1826 0 : (void)hrtRaSocketDeInit(socketHandle);
1827 0 : raResourceInfo_.nicSocketMap.erase(ipAddr);
1828 0 : raResourceInfo_.hostNetSocketMap.erase(ipAddr);
1829 0 : if (!isRaInitRepeated_) {
1830 0 : (void)HrtRaDeInit(&config);
1831 : }
1832 0 : if (isTsdProcessOpen_) {
1833 0 : (void)hrtCloseNetService();
1834 0 : isTsdProcessOpen_ = false;
1835 : }
1836 0 : return ret;
1837 : }
1838 :
1839 0 : return HCCL_SUCCESS;
1840 1 : }
1841 :
1842 : // 最后一次ra_deinit时才关闭device的hccp进程。ES场景主要使用
1843 0 : HcclResult NetworkManager::CloseHccpSubProc()
1844 : {
1845 0 : if (!isHostUseDevNic_ || subPid_ == 0) {
1846 0 : HCCL_INFO("No need to close hccp sub proc, devicePhyId[%u], subPid[%lld]", devicePhyId_, subPid_);
1847 0 : return HCCL_SUCCESS;
1848 : }
1849 0 : HCCL_INFO(
1850 : "NetworkManager ProcessCloseSubProcList HDC devicePhyId[%u], deviceLogicId_[%d], subPid[%lld]", devicePhyId_,
1851 : deviceLogicId_, static_cast<s64>(subPid_));
1852 0 : s32 locaLogDevid = 0;
1853 0 : hrtGetDevice(&locaLogDevid);
1854 0 : if (locaLogDevid != deviceLogicId_) {
1855 0 : hrtSetDevice(deviceLogicId_);
1856 : }
1857 0 : HcclResult ret = hrtCloseNetService();
1858 0 : if (locaLogDevid != deviceLogicId_) {
1859 0 : hrtSetDevice(locaLogDevid);
1860 : }
1861 0 : CHK_RET(ret);
1862 0 : subPid_ = 0;
1863 :
1864 0 : return HCCL_SUCCESS;
1865 : }
1866 :
1867 16 : HcclResult NetworkManager::PingMeshRaPingInit(u32 devLogicId, u32 devPhyId, RaInitConfig* config)
1868 : {
1869 : // 引用计数
1870 16 : deviceLogicId_ = static_cast<s32>(devLogicId);
1871 16 : devicePhyId_ = ((static_cast<s32>(devPhyId) == HOST_DEVICE_ID) ? 0 : devPhyId);
1872 16 : isHostUseDevNic_ = true;
1873 16 : isRaInitRepeated_ = false;
1874 :
1875 : // hccp侧初始化ping mesh资源
1876 16 : CHK_RET(HrtRaInit(config));
1877 16 : HCCL_INFO(
1878 : "[HCCN][PingMeshRaPingInit]Device[%u] config.hdcType[%d], config.nicPosition[%u], config.phyId[%u].",
1879 : deviceLogicId_, config->hdcType, config->nicPosition, config->phyId);
1880 16 : deviceNicInitRef_.Ref();
1881 :
1882 16 : return HCCL_SUCCESS;
1883 : }
1884 :
1885 15 : HcclResult NetworkManager::PingMeshRaPingDeinit()
1886 : {
1887 : // 引用计数
1888 15 : isRaInitRepeated_ = false;
1889 :
1890 15 : struct RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, false};
1891 15 : GetDeviceRaInitConfig(config);
1892 15 : CHK_RET(HrtRaDeInit(&config));
1893 15 : deviceNicInitRef_.Unref();
1894 :
1895 15 : return HCCL_SUCCESS;
1896 : }
1897 :
1898 20 : void NetworkManager::GetDeviceRaInitConfig(RaInitConfig& config)
1899 : {
1900 20 : u32 devicePhyId = ((static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) ? 0 : devicePhyId_);
1901 20 : HCCL_INFO("RaDeinit devicePhyId_[%u] devicePhyId[%u]", devicePhyId_, devicePhyId);
1902 :
1903 20 : config.phyId = devicePhyId;
1904 20 : config.nicPosition = static_cast<u32>(isHostUseDevNic_);
1905 :
1906 20 : if (isHostUseDevNic_) {
1907 16 : config.hdcType = PID_HDC_TYPE;
1908 : }
1909 20 : }
1910 :
1911 0 : HcclResult NetworkManager::PsWorkerRaDeinit(u32 devId, const HcclIpAddress& ipAddr, u32 port)
1912 : {
1913 0 : string ipAddrStr(ipAddr.GetReadableAddress());
1914 0 : u32 devicePhyId = ((static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) ? 0 : devicePhyId_);
1915 0 : if (ipAddrStr == "127.0.0.1") {
1916 0 : hostNicInitRef_.Unref();
1917 0 : HCCL_INFO("hostNicInitRef_[%d]", hostNicInitRef_.Count());
1918 : } else {
1919 0 : bool lastUsed{false};
1920 0 : deviceNicInitRef_.Unref();
1921 0 : CHK_RET(hrtRaIsLastUsed(devicePhyId, lastUsed));
1922 0 : HCCL_INFO(
1923 : "deviceNicInitRef_[%d] lastUsed[%u] devicePhyId[%u]", deviceNicInitRef_.Count(), lastUsed, devicePhyId);
1924 0 : if (deviceNicInitRef_.Count() == 0 && !lastUsed) {
1925 0 : isRaInitRepeated_ = true;
1926 0 : } else if (lastUsed) {
1927 0 : isRaInitRepeated_ = false;
1928 0 : } else if (deviceNicInitRef_.Count() > 0) {
1929 0 : HCCL_INFO("[NetworkManager] PsWorkerRa is not lastUsed");
1930 0 : return HCCL_SUCCESS;
1931 : }
1932 : }
1933 0 : CHK_PRT_RET(
1934 : (raResourceInfo_.nicSocketMap.count(ipAddr) == 0),
1935 : HCCL_INFO("NetworkManager: PsWorkerRa already Deinit, ipAddr[%s]", ipAddr.GetReadableAddress()), HCCL_SUCCESS);
1936 :
1937 0 : HCCL_INFO("PsWorkerRaDeinit devId[%u], isRaInitRepeated[%d]", devId, isRaInitRepeated_);
1938 : // 重复RaInit时，不再调用内部DeInit
1939 0 : if (!isRaInitRepeated_) {
1940 0 : CHK_RET(HeterogStopListen(ipAddr, port, true));
1941 : }
1942 :
1943 0 : SocketHandle nicRdmaHandle = raResourceInfo_.nicSocketMap[ipAddr].nicRdmaHandle;
1944 0 : if (!GetExternalInputHcclIsTcpMode() && nicRdmaHandle != nullptr && !isRaInitRepeated_) {
1945 : // Helper的PS临时暂不调用 CHK_RET(HrtRaRdmaDeInit(nicRdmaHandle, (isHostUseDevNic_) ?
1946 : // NotifyTypeT::NOTIFY : NotifyTypeT::NO_USE));
1947 0 : HCCL_INFO("Not call RaRdmaDeInit devicePhyId[%u]", devicePhyId_);
1948 : }
1949 :
1950 0 : struct RaInitConfig config = {DEFAULT_INIT_PHY_ID, DEFAULT_INIT_NIC_POS, DEFAULT_HDC_TYPE, false};
1951 0 : GetDeviceRaInitConfig(config);
1952 :
1953 0 : if (ipAddrStr == "127.0.0.1") {
1954 0 : if (hostNicInitRef_.Count() == 0) {
1955 0 : HCCL_INFO("PsWorkerRaDeinit call hrtRaDeInit. devicePhyId[%u]", devicePhyId_);
1956 0 : config.nicPosition = static_cast<u32>(NICDeployment::NIC_DEPLOYMENT_HOST);
1957 0 : CHK_RET(HrtRaDeInit(&config));
1958 : }
1959 : } else {
1960 0 : if (deviceNicInitRef_.Count() == 0) {
1961 0 : HCCL_INFO("PsWorkerRaDeinit call HrtRaDeInit. devicePhyId[%u]", devicePhyId_);
1962 0 : if (!isRaInitRepeated_) {
1963 0 : CHK_RET(HrtRaDeInit(&config));
1964 0 : CHK_RET(CloseHccpSubProc());
1965 : }
1966 : }
1967 : }
1968 :
1969 0 : raResourceInfo_.nicSocketMap.erase(raResourceInfo_.nicSocketMap.find(ipAddr));
1970 0 : raResourceInfo_.hostNetSocketMap.erase(raResourceInfo_.hostNetSocketMap.find(ipAddr));
1971 :
1972 0 : return HCCL_SUCCESS;
1973 0 : }
1974 :
1975 22 : HcclResult NetworkManager::InitHostSocket(const HcclIpAddress& addr, SocketHandle& socketHandle) const
1976 : {
1977 22 : struct SocketInitInfoT socketInitInfo = {};
1978 22 : socketInitInfo.rdevInfo.family = addr.GetFamily();
1979 22 : socketInitInfo.rdevInfo.phyId = devicePhyId_;
1980 22 : socketInitInfo.rdevInfo.localIp.addr = addr.GetBinaryAddress().addr;
1981 22 : socketInitInfo.rdevInfo.localIp.addr6 = addr.GetBinaryAddress().addr6;
1982 22 : socketInitInfo.scopeId = addr.GetScopeID();
1983 22 : HcclResult ret = hrtRaSocketInitV1(NETWORK_PEER_ONLINE, socketInitInfo, socketHandle);
1984 22 : CHK_PRT_RET(
1985 : ret != HCCL_SUCCESS,
1986 : HCCL_ERROR(
1987 : "[Init][HostSocket]errNo[0x%016llx] ra socket init v1 failed, ip[%s], return[%d]",
1988 : HCCL_ERROR_CODE(HCCL_E_TCP_CONNECT), addr.GetReadableAddress(), ret),
1989 : HCCL_E_TCP_CONNECT);
1990 22 : HCCL_INFO("ip[%s] socket init OK", addr.GetReadableAddress());
1991 22 : return HCCL_SUCCESS;
1992 : }
1993 :
1994 25 : HcclResult NetworkManager::StopHostNet(SocketHandle socketHandle, const HcclIpAddress& ipAddr)
1995 : {
1996 25 : std::unique_lock<std::mutex> lock(raLock_);
1997 25 : CHK_PRT_RET(
1998 : (hostNicInitRef_.Count() == 0), HCCL_ERROR("[Stop][HostNet]cannot start nic socket before host nic inited!"),
1999 : HCCL_E_INTERNAL);
2000 :
2001 25 : CHK_RET(CheckSocketInfo(socketHandle, ipAddr));
2002 :
2003 25 : if (hostNicSocketClientRef_[ipAddr].Unref() == 0
2004 25 : && raResourceInfo_.hostNetSocketMap[ipAddr].listenedPort.size() == 0) {
2005 20 : CHK_RET(hrtRaSocketDeInit(socketHandle));
2006 20 : HCCL_INFO("ip[%s] deinit success.", ipAddr.GetReadableAddress());
2007 :
2008 20 : raResourceInfo_.hostNetSocketMap.erase(ipAddr);
2009 20 : raResourceInfo_.nicSocketMap.erase(ipAddr);
2010 : }
2011 25 : return HCCL_SUCCESS;
2012 25 : }
2013 :
2014 : // 从ip查handle 把ip下所有的listen全stop
2015 0 : HcclResult NetworkManager::StopHostSocketHandle(const HcclIpAddress& ipAddr)
2016 : {
2017 0 : std::unique_lock<std::mutex> lock(raLock_);
2018 0 : CHK_PRT_RET(
2019 : (hostNicInitRef_.Count() == 0),
2020 : HCCL_ERROR("[NetworkManager][StopHostSocketHandle]cannot start nic socket before host nic inited!"),
2021 : HCCL_E_INTERNAL);
2022 :
2023 0 : auto sockInfo = raResourceInfo_.hostNetSocketMap.find(ipAddr);
2024 0 : auto ipIt = IPPortListenRefMapHost_.find(ipAddr);
2025 0 : CHK_PRT_RET(
2026 : (sockInfo == raResourceInfo_.hostNetSocketMap.end()),
2027 : HCCL_ERROR("[NetworkManager][StopHostSocketHandle]ipAddr is invalid"), HCCL_E_INTERNAL);
2028 0 : IpSocket& sock = raResourceInfo_.hostNetSocketMap[ipAddr];
2029 0 : CHK_RET(CheckSocketInfo(sock.nicSocketHandle, ipAddr));
2030 :
2031 : // 停止所有的listen
2032 0 : if (ipIt != IPPortListenRefMapHost_.end()) {
2033 0 : for (auto& portIt : ipIt->second) {
2034 0 : u32 port = portIt.first;
2035 0 : if (IPPortListenRefMapHost_[ipAddr][port].Count() > 0) {
2036 0 : CHK_RET(StopListenSocket(sock.nicSocketHandle, port));
2037 0 : HCCL_WARNING(
2038 : "[NetworkManager][StopHostSocketHandle] ip [%s] stop listen port [%u]", ipAddr.GetReadableAddress(),
2039 : port);
2040 0 : raResourceInfo_.hostNetSocketMap[ipAddr].listenedPort.erase(port);
2041 0 : IPPortListenRefMapHost_[ipAddr][port].Clear(); // 引用计数归0
2042 : }
2043 : }
2044 : }
2045 0 : CHK_PRT_RET(
2046 : (raResourceInfo_.hostNetSocketMap[ipAddr].listenedPort.size() != 0),
2047 : HCCL_ERROR(
2048 : "[NetworkManager][StopHostSocketHandle]IPPortListenRefMapHost_[%s] is unequal to "
2049 : "hostNetSocketMap[%s].listenedPort",
2050 : ipAddr.GetReadableAddress(), ipAddr.GetReadableAddress()),
2051 : HCCL_E_INTERNAL);
2052 : // 删除socket 删除IP
2053 0 : CHK_RET(hrtRaSocketDeInit(sock.nicSocketHandle));
2054 0 : HCCL_INFO("[NetworkManager][StopHostSocketHandle] ip [%s] deinit success.", ipAddr.GetReadableAddress());
2055 0 : sock.nicSocketHandle = nullptr;
2056 : // 没有host和rdma时 删除ip
2057 0 : if (sock.nicSocketHandle == nullptr && sock.nicRdmaHandle == nullptr) {
2058 0 : raResourceInfo_.hostNetSocketMap.erase(ipAddr);
2059 : }
2060 0 : hostNicSocketClientRef_[ipAddr].Clear();
2061 0 : return HCCL_SUCCESS;
2062 0 : }
2063 :
2064 25 : HcclResult NetworkManager::StartHostNet(const HcclIpAddress& ipAddr, SocketHandle& socketHandle)
2065 : {
2066 25 : CHK_RET(CreateHostSocketHandle(ipAddr, socketHandle));
2067 25 : return HCCL_SUCCESS;
2068 : }
2069 :
2070 25 : HcclResult NetworkManager::CreateHostSocketHandle(const HcclIpAddress& ipAddr, SocketHandle& socketHandle)
2071 : {
2072 25 : CHK_PRT_RET(
2073 : (hostNicInitRef_.Count() == 0),
2074 : HCCL_ERROR("[CreateHostSocketHandle]cannot start nic socket before host nic inited!"), HCCL_E_INTERNAL);
2075 25 : OccupyIp(ipAddr, raResourceInfo_.hostNetSocketMap);
2076 25 : IpSocket& sock = raResourceInfo_.hostNetSocketMap[ipAddr];
2077 25 : if (sock.nicSocketHandle == nullptr) {
2078 20 : CHK_RET(InitHostSocket(ipAddr, sock.nicSocketHandle));
2079 : }
2080 :
2081 25 : socketHandle = sock.nicSocketHandle;
2082 25 : HCCL_INFO("ip[%s] socket start success socketHandle[%p]", ipAddr.GetReadableAddress(), socketHandle);
2083 25 : hostNicSocketClientRef_[ipAddr].Ref(); // 引用计数
2084 25 : return HCCL_SUCCESS;
2085 : }
2086 :
2087 41 : HcclResult NetworkManager::StartListenSocket(const SocketHandle socketHandle, u32& port) const
2088 : {
2089 41 : struct SocketListenInfoT serverInfo = {};
2090 41 : serverInfo.socketHandle = const_cast<SocketHandle>(socketHandle);
2091 41 : serverInfo.port = port;
2092 41 : if (isRaInitRepeated_) {
2093 0 : return HCCL_SUCCESS;
2094 : }
2095 41 : bool isAutoPort = port == AUTO_LISTEN_PORT;
2096 41 : HcclResult ret = hrtRaSocketListenStart(&serverInfo, 1);
2097 41 : CHK_PRT_RET(
2098 : ret == HCCL_E_UNAVAIL,
2099 : HCCL_INFO(
2100 : "socket port[%u] has already been bound. Could not start listening host nic. please use an idle port.",
2101 : port),
2102 : ret);
2103 41 : RPT_CALL_ERR(ret != HCCL_SUCCESS, "host nic listen start failed, port[%u], return[%d]", port, ret);
2104 41 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("host nic listen start failed, port[%u], return[%d]", port, ret), ret);
2105 41 : if (isAutoPort) {
2106 0 : port = serverInfo.port;
2107 0 : CHK_PRT_RET(
2108 : port == AUTO_LISTEN_PORT, HCCL_ERROR("start listen on a port selected by os automatically failed"),
2109 : HCCL_E_NOT_SUPPORT);
2110 0 : HCCL_RUN_INFO("start listen on port[%u] by auto success.", port);
2111 : }
2112 41 : return HCCL_SUCCESS;
2113 : }
2114 :
2115 39 : HcclResult NetworkManager::StopListenSocket(const SocketHandle socketHandle, u32 port) const
2116 : {
2117 : struct SocketListenInfoT serverInfo;
2118 39 : serverInfo.socketHandle = const_cast<SocketHandle>(socketHandle);
2119 39 : serverInfo.port = port;
2120 39 : HcclResult ret = hrtRaSocketListenStop(&serverInfo, 1);
2121 39 : RPT_CALL_ERR(ret != HCCL_SUCCESS, "socket listen stop failed, port[%u], return[%d]", port, ret);
2122 39 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("socket listen stop failed, port[%u], return[%d]", port, ret), ret);
2123 39 : return HCCL_SUCCESS;
2124 : }
2125 :
2126 32 : HcclResult NetworkManager::GetRdmaHandleByIpAddr(const HcclIpAddress& ipAddr, RdmaHandle& rdmaHandle)
2127 : {
2128 32 : auto it = raResourceInfo_.nicSocketMap.find(ipAddr);
2129 32 : CHK_PRT_RET(
2130 : it == raResourceInfo_.nicSocketMap.end(),
2131 : HCCL_ERROR("GetRdmaHandleByIpAddr ip[%s] is not found in nicSocketMap.", ipAddr.GetReadableAddress()),
2132 : HCCL_E_INTERNAL);
2133 32 : rdmaHandle = raResourceInfo_.nicSocketMap[ipAddr].nicRdmaHandle;
2134 32 : CHK_PTR_NULL(rdmaHandle);
2135 32 : return HCCL_SUCCESS;
2136 : }
2137 :
2138 0 : HcclResult NetworkManager::GetNicHandleByIpAddr(const HcclIpAddress& ipAddr, SocketHandle& nicHandle)
2139 : {
2140 0 : auto it = raResourceInfo_.nicSocketMap.find(ipAddr);
2141 0 : CHK_PRT_RET(
2142 : it == raResourceInfo_.nicSocketMap.end(),
2143 : HCCL_ERROR("GetNicHandleByIpAddr ip[%s] is not found in nicSocketMap.", ipAddr.GetReadableAddress()),
2144 : HCCL_E_INTERNAL);
2145 0 : nicHandle = raResourceInfo_.nicSocketMap[ipAddr].nicSocketHandle;
2146 0 : CHK_PTR_NULL(nicHandle);
2147 0 : return HCCL_SUCCESS;
2148 : }
2149 :
2150 21 : HcclResult NetworkManager::CheckAutoListenVersion(bool isAutoPort)
2151 : {
2152 21 : if (isAutoPort) {
2153 1 : u32 listenStartVersion = 0;
2154 1 : HcclResult vRet = hrtRaGetInterfaceVersion(devicePhyId_, SOCKET_LISTEN_AUTO_INTERFACE, &listenStartVersion);
2155 1 : HCCL_INFO("[CheckAutoListenVersion] listen start version[%u].", listenStartVersion);
2156 1 : CHK_PRT_RET(
2157 : vRet != HCCL_SUCCESS || listenStartVersion < SOCKET_LISTEN_AUTO_INTERFACE_VERSION,
2158 : HCCL_ERROR(
2159 : "this package does not support hrtRaSocketNonBlockListenStart to "
2160 : "listen automatically with port %u, please change new package.",
2161 : AUTO_LISTEN_PORT),
2162 : HCCL_E_NOT_SUPPORT);
2163 : }
2164 20 : return HCCL_SUCCESS;
2165 : }
2166 :
2167 : // 新旧ip类型转换
2168 0 : HcclResult NetworkManager::HcclIpAddressConvertHcclAddr(HcclAddress* hccladdr, HcclIpAddress* hcclIP)
2169 : {
2170 0 : CHK_PTR_NULL(hcclIP);
2171 0 : CHK_PTR_NULL(hccladdr);
2172 0 : if (hcclIP->GetFamily() == AF_INET) {
2173 0 : hccladdr->type = HCCL_ADDR_TYPE_IP_V4;
2174 0 : hccladdr->addr = hcclIP->GetBinaryAddress().addr;
2175 0 : } else if (hcclIP->GetFamily() == AF_INET6) {
2176 0 : hccladdr->type = HCCL_ADDR_TYPE_IP_V6;
2177 0 : hccladdr->addr6 = hcclIP->GetBinaryAddress().addr6;
2178 : } else {
2179 0 : HCCL_ERROR("[HcclIpAddressConvertingHcclAddr]ERROR IP type!");
2180 0 : return HCCL_E_PARA;
2181 : }
2182 0 : return HCCL_SUCCESS;
2183 : }
2184 :
2185 25 : HcclResult NetworkManager::OccupyIp(const HcclIpAddress& ipAddr, std::map<hccl::HcclIpAddress, IpSocket>& socketMap)
2186 : {
2187 25 : auto sockInfo = socketMap.find(ipAddr);
2188 25 : if (sockInfo == socketMap.end()) {
2189 20 : IpSocket tempSock;
2190 20 : socketMap.insert(std::make_pair(ipAddr, tempSock)); // 本IP占位
2191 20 : HCCL_INFO("[NetworkManager][OccupyIp] device[%u] insert ip[%s]", devicePhyId_, ipAddr.GetReadableAddress());
2192 20 : } else {
2193 5 : HCCL_INFO(
2194 : "[NetworkManager][OccupyIp] device[%u] ip[%s] has already occupied.", devicePhyId_,
2195 : ipAddr.GetReadableAddress());
2196 : }
2197 25 : return HCCL_SUCCESS;
2198 : }
2199 : } // namespace hccl
|
