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 "global_net_dev_manager.h"
12 : #include <string>
13 : #include "hccl_mem.h"
14 : // for hccl_network.h
15 : #include "hccl_network.h"
16 : #include "network_manager_pub.h"
17 : #include "dlhal_function.h"
18 : #include "dlra_function.h"
19 :
20 : using namespace hccl;
21 :
22 : namespace hccl {
23 : std::map<PortInfo, std::pair<NicType, HcclNetDevCtx>> GlobalNetDevMgr::netDevCtxMap_;
24 : std::map<PortInfo, Referenced> GlobalNetDevMgr::netDevCtxRefMap_;
25 : std::mutex GlobalNetDevMgr::netDevCtxMtx_;
26 : bool GlobalNetDevMgr::isDlRaInited_{false};
27 :
28 : std::map<PortInfo, std::shared_ptr<HcclSocket>> GlobalNetDevMgr::serverSocketMap_;
29 : std::map<PortInfo, Referenced> GlobalNetDevMgr::serverSocketRefMap_;
30 : std::mutex GlobalNetDevMgr::serverMapMutex_;
31 :
32 : // reserve 1 instance for invalid deviceid and host
33 : static GlobalNetDevMgr netDevMgrInstance[MAX_MODULE_DEVICE_NUM + 1];
34 3432 : GlobalNetDevMgr::~GlobalNetDevMgr()
35 : {
36 3432 : if (isInited_.load(std::memory_order_acquire)) {
37 1 : HCCL_INFO("[GlobalNetDevMgr][%s] start.", __func__);
38 1 : UnInit();
39 1 : HCCL_INFO("[GlobalNetDevMgr][%s] end.", __func__);
40 : }
41 3432 : }
42 :
43 36 : GlobalNetDevMgr& GlobalNetDevMgr::GetInstance(u32 devicePhyId)
44 : {
45 : u32 deviceLogicId;
46 36 : HcclResult hcclRet = hrtGetDeviceIndexByPhyId(devicePhyId, deviceLogicId);
47 36 : if (hcclRet != HCCL_SUCCESS) {
48 0 : HCCL_RUN_WARNING(
49 : "GlobalNetDevMgr::GetInstance hrtGetDeviceIndexByPhyId failed, ret[%d], "
50 : "return reserve instance",
51 : hcclRet);
52 0 : return netDevMgrInstance[MAX_MODULE_DEVICE_NUM];
53 : }
54 :
55 36 : if (deviceLogicId >= MAX_MODULE_DEVICE_NUM) {
56 0 : HCCL_RUN_WARNING("[Get][Instance]deviceLogicId[%u] is invalid, return reserve instance", deviceLogicId);
57 0 : return netDevMgrInstance[MAX_MODULE_DEVICE_NUM];
58 : }
59 :
60 36 : if (!netDevMgrInstance[deviceLogicId].isInited_.load(std::memory_order_acquire)) {
61 12 : hcclRet = Init(devicePhyId, deviceLogicId);
62 12 : if (hcclRet != HCCL_SUCCESS) {
63 0 : HCCL_RUN_WARNING("[Get][Instance]Init deviceLogicId[%u]fail, return reserve instance", deviceLogicId);
64 0 : return netDevMgrInstance[MAX_MODULE_DEVICE_NUM];
65 : }
66 : }
67 :
68 36 : HCCL_DEBUG("GlobalNetDevMgr::GetInstance deviceLogicId[%u], devicePhyId[%u] done.", deviceLogicId, devicePhyId);
69 36 : return netDevMgrInstance[deviceLogicId];
70 : }
71 :
72 12 : HcclResult GlobalNetDevMgr::Init(u32 devicePhyId, u32 deviceLogicId)
73 : {
74 : // init after get the lock
75 12 : std::unique_lock<std::mutex> lock(netDevCtxMtx_);
76 12 : if (netDevMgrInstance[deviceLogicId].isInited_.load(std::memory_order_relaxed)) {
77 0 : return HCCL_SUCCESS;
78 : }
79 :
80 12 : if (!isDlRaInited_) {
81 1 : CHK_RET(hccl::DlRaFunction::GetInstance().DlRaFunctionInit());
82 1 : CHK_RET(hccl::DlHalFunction::GetInstance().DlHalFunctionInit());
83 1 : isDlRaInited_ = true;
84 : }
85 :
86 : // need to check again
87 12 : if (netDevMgrInstance[deviceLogicId].isInited_.load(std::memory_order_relaxed)) {
88 0 : HCCL_INFO(
89 : "[GlobalNetDevMgr][Init]Has been inited. devicePhyId[%u], deviceLogicId[%u]", devicePhyId, deviceLogicId);
90 0 : return HCCL_SUCCESS;
91 : }
92 :
93 12 : netDevMgrInstance[deviceLogicId].devicePhyId_ = devicePhyId;
94 12 : netDevMgrInstance[deviceLogicId].deviceLogicId_ = deviceLogicId;
95 12 : CHK_RET(HcclNetInit(NICDeployment::NIC_DEPLOYMENT_DEVICE, devicePhyId, static_cast<u32>(deviceLogicId), false));
96 12 : netDevMgrInstance[deviceLogicId].isInited_.store(true, std::memory_order_release);
97 12 : HCCL_INFO("[GlobalNetDevMgr][Init]Init success, devicePhyId[%u], deviceLogicId[%u]", devicePhyId, deviceLogicId);
98 12 : return HCCL_SUCCESS;
99 12 : }
100 :
101 12 : void GlobalNetDevMgr::UnInit()
102 : {
103 12 : if (!isInited_.load(std::memory_order_acquire)) {
104 0 : HCCL_INFO(
105 : "[GlobalNetDevMgr][UnInit]has been deinited. devicePhyId[%u], deviceLogicId[%d]", devicePhyId_,
106 : deviceLogicId_);
107 0 : return;
108 : }
109 :
110 12 : (void)HcclNetDeInit(NICDeployment::NIC_DEPLOYMENT_DEVICE, devicePhyId_, static_cast<u32>(deviceLogicId_));
111 12 : netDevCtx_ = nullptr;
112 12 : isInited_.store(false, std::memory_order_release);
113 12 : HCCL_INFO(
114 : "[GlobalNetDevMgr][UnInit]UnInit success. devicePhyId[%u], deviceLogicId[%d]", devicePhyId_, deviceLogicId_);
115 : }
116 :
117 2 : HcclResult GlobalNetDevMgr::GetDeviceVnicIP(u32 devicePhyId, u32 superDeviceId, hccl::HcclIpAddress& vnicIP)
118 : {
119 : s32 localDeviceLogicId;
120 : u32 localDeviceId;
121 2 : CHK_RET(hrtGetDevice(&localDeviceLogicId));
122 2 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(localDeviceLogicId), localDeviceId));
123 :
124 : // 先创建进程
125 : bool isHostUseDevNic;
126 2 : CHK_RET(IsHostUseDevNic(isHostUseDevNic));
127 2 : u32 tempDevicePhyId = hccl::DEFAULT_PHY_ID;
128 2 : HCCL_DEBUG(
129 : "[GlobalNetDevMgr][%s]GetDeviceVnicIP, deviceLogicId[%d], devicePhyId[%u], "
130 : "nicDeploy[%d], hasBackup[%d], tempDevicePhyId[%u]",
131 : __func__, localDeviceLogicId, devicePhyId, static_cast<int>(NICDeployment::NIC_DEPLOYMENT_DEVICE), false,
132 : tempDevicePhyId);
133 2 : CHK_RET(hccl::NetworkManager::GetInstance(localDeviceLogicId)
134 : .InitV2(NICDeployment::NIC_DEPLOYMENT_DEVICE, false, tempDevicePhyId, isHostUseDevNic));
135 :
136 : // 参考 Heartbeat::GetConnInfo
137 : // hccl::HcclIpAddress vnicIP(localDeviceId);
138 2 : if (superDeviceId != SUPER_DEVICE_ID_INVALID) {
139 2 : CHK_RET(
140 : hrtRaGetSingleSocketVnicIpInfo(localDeviceId, DeviceIdType::DEVICE_ID_TYPE_SDID, superDeviceId, vnicIP));
141 : } else {
142 0 : CHK_RET(
143 : hrtRaGetSingleSocketVnicIpInfo(localDeviceId, DeviceIdType::DEVICE_ID_TYPE_PHY_ID, devicePhyId, vnicIP));
144 : }
145 :
146 0 : HCCL_INFO(
147 : "[GlobalNetDevMgr][GetDeviceVnicIP] vnicIP [%s] for devicePhyId[%u], superDeviceId[%u]",
148 : vnicIP.GetReadableAddress(), devicePhyId, superDeviceId);
149 :
150 : // 销毁进程
151 0 : CHK_RET(hccl::NetworkManager::GetInstance(localDeviceLogicId)
152 : .DeInitV2(NICDeployment::NIC_DEPLOYMENT_DEVICE, false, false));
153 0 : return HCCL_SUCCESS;
154 : }
155 :
156 : HcclResult
157 14 : GlobalNetDevMgr::RefNetDevCtx(NicType nicType, const HcclIpAddress& ipAddr, u32 port, HcclNetDevCtx& netDevCtx)
158 : {
159 14 : HCCL_INFO(
160 : "[GlobalNetDevMgr][RefNetDevCtx] nicType[%d], ip[%s]", static_cast<int>(nicType), ipAddr.GetReadableAddress());
161 14 : std::lock_guard<std::mutex> lock(netDevCtxMtx_);
162 :
163 : // 进程粒度open dev，如果已open，直接复用
164 14 : PortInfo portInfo(ipAddr, port);
165 14 : if (netDevCtxMap_.find(portInfo) != netDevCtxMap_.end()) {
166 0 : netDevCtx = netDevCtxMap_[portInfo].second;
167 0 : CHK_PTR_NULL(netDevCtx);
168 :
169 0 : auto& netDevCtxRef = netDevCtxRefMap_[portInfo];
170 0 : netDevCtxRef.Ref();
171 0 : netDevCtx_ = netDevCtx;
172 :
173 0 : HCCL_INFO(
174 : "[GlobalNetDevMgr][RefNetDevCtx] nicType[%d] ip[%s] has been Ref.", static_cast<int>(nicType),
175 : ipAddr.GetReadableAddress());
176 0 : return HCCL_SUCCESS;
177 : }
178 :
179 : HcclNetDevCtx tempNetDevCtx;
180 14 : CHK_RET(HcclNetOpenDev(&tempNetDevCtx, nicType, devicePhyId_, deviceLogicId_, ipAddr));
181 14 : CHK_PTR_NULL(tempNetDevCtx);
182 :
183 : try {
184 14 : netDevCtxMap_.insert(std::make_pair(portInfo, std::make_pair(nicType, tempNetDevCtx)));
185 0 : } catch (...) {
186 0 : (void)HcclNetCloseDev(tempNetDevCtx);
187 0 : return HCCL_E_MEMORY;
188 0 : }
189 :
190 14 : Referenced ref;
191 14 : ref.Ref();
192 : try {
193 14 : netDevCtxRefMap_.insert(std::make_pair(portInfo, ref));
194 0 : } catch (...) {
195 0 : netDevCtxMap_.erase(portInfo);
196 0 : (void)HcclNetCloseDev(tempNetDevCtx);
197 0 : return HCCL_E_MEMORY;
198 0 : }
199 :
200 14 : netDevCtx = tempNetDevCtx;
201 14 : netDevCtx_ = netDevCtx;
202 14 : HCCL_INFO(
203 : "[GlobalNetDevMgr][RefNetDevCtx] nicType[%d] ip[%s] has been Init.", static_cast<int>(nicType),
204 : ipAddr.GetReadableAddress());
205 14 : return HCCL_SUCCESS;
206 14 : }
207 :
208 14 : HcclResult GlobalNetDevMgr::UnRefNetDevCtx(NicType nicType, const HcclIpAddress& ipAddr, u32 port)
209 : {
210 14 : HCCL_INFO(
211 : "[GlobalNetDevMgr][UnRefNetDevCtx] nicType[%d], ip[%s]", static_cast<int>(nicType),
212 : ipAddr.GetReadableAddress());
213 :
214 14 : std::lock_guard<std::mutex> lock(netDevCtxMtx_);
215 :
216 : HcclNetDevCtx netDevCtx;
217 14 : PortInfo portInfo(ipAddr, port);
218 14 : if (netDevCtxMap_.find(portInfo) != netDevCtxMap_.end()) {
219 14 : netDevCtx = netDevCtxMap_[portInfo].second;
220 14 : CHK_PTR_NULL(netDevCtx);
221 :
222 14 : auto& netDevCtxRef = netDevCtxRefMap_[portInfo];
223 14 : netDevCtxRef.Unref();
224 14 : HCCL_INFO(
225 : "[GlobalNetDevMgr][UnRefNetDevCtx] nicType[%d] ip[%s] has been UnRef.", static_cast<int>(nicType),
226 : ipAddr.GetReadableAddress());
227 :
228 14 : if (netDevCtxRef.Count() == 0) {
229 14 : netDevCtxMap_.erase(portInfo);
230 14 : netDevCtxRefMap_.erase(portInfo);
231 14 : HcclNetCloseDev(netDevCtx);
232 14 : HCCL_INFO(
233 : "[GlobalNetDevMgr][UnRefNetDevCtx] nicType[%d] ip[%s] has been Deinit.", static_cast<int>(nicType),
234 : ipAddr.GetReadableAddress());
235 : }
236 : }
237 :
238 14 : if (netDevCtxMap_.empty()) {
239 11 : UnInit();
240 : }
241 14 : return HCCL_SUCCESS;
242 14 : }
243 :
244 2 : HcclResult GlobalNetDevMgr::ServerInit(u32 port)
245 : {
246 2 : HcclIpAddress localIp{0};
247 2 : std::shared_ptr<HcclSocket> tempSocket;
248 : {
249 2 : std::lock_guard<std::mutex> lock(netDevCtxMtx_);
250 2 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx_, localIp));
251 2 : HCCL_INFO("[GlobalNetDevMgr][ServerInit]ip[%s] port[%u]", localIp.GetReadableAddress(), port);
252 2 : EXCEPTION_CATCH((tempSocket = std::make_shared<HcclSocket>(netDevCtx_, port)), return HCCL_E_PTR);
253 2 : }
254 :
255 2 : PortInfo portInfo(localIp, port);
256 2 : std::unique_lock<std::mutex> lock(serverMapMutex_);
257 2 : auto serverSocketInMap = serverSocketMap_.find(portInfo);
258 2 : if (serverSocketInMap != serverSocketMap_.end()) {
259 0 : auto& serverSocketRef = serverSocketRefMap_[portInfo];
260 0 : serverSocketRef.Ref();
261 0 : HCCL_INFO("[GlobalNetDevMgr][ServerInit]ip[%s] port[%u] inited", localIp.GetReadableAddress(), port);
262 0 : return HCCL_SUCCESS;
263 : }
264 :
265 2 : CHK_RET(tempSocket->Init());
266 2 : CHK_RET(tempSocket->Listen());
267 2 : serverSocketMap_.insert(std::make_pair(portInfo, tempSocket));
268 :
269 2 : Referenced ref;
270 2 : ref.Ref();
271 2 : serverSocketRefMap_.insert(std::make_pair(portInfo, ref));
272 2 : HCCL_INFO("[GlobalNetDevMgr][ServerInit]ip[%s] port[%u] init done", localIp.GetReadableAddress(), port);
273 2 : return HCCL_SUCCESS;
274 2 : }
275 :
276 2 : HcclResult GlobalNetDevMgr::ServerDeInit(u32 port)
277 : {
278 2 : HcclIpAddress localIp{0};
279 : {
280 2 : std::lock_guard<std::mutex> lock(netDevCtxMtx_);
281 2 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx_, localIp));
282 2 : }
283 2 : CHK_RET(ServerDeInit(localIp, port));
284 :
285 2 : return HCCL_SUCCESS;
286 2 : }
287 :
288 2 : HcclResult GlobalNetDevMgr::ServerDeInit(const HcclIpAddress& localIp, u32 port)
289 : {
290 2 : PortInfo portInfo(localIp, port);
291 :
292 2 : std::unique_lock<std::mutex> lock(serverMapMutex_);
293 2 : auto res = serverSocketMap_.find(portInfo);
294 2 : if (res == serverSocketMap_.end()) {
295 0 : HCCL_INFO("[DeInit][Server]ip[%s] port[%u] not found", localIp.GetReadableAddress(), port);
296 0 : return HCCL_SUCCESS;
297 : }
298 :
299 2 : auto& serverSocketRef = serverSocketRefMap_[portInfo];
300 2 : serverSocketRef.Unref();
301 :
302 2 : HCCL_INFO(
303 : "[DeInit][Server]ip[%s] port[%u] serverSocketRef.Count() = %d", localIp.GetReadableAddress(), port,
304 : serverSocketRef.Count());
305 2 : if (serverSocketRef.Count() == 0) {
306 2 : HCCL_INFO("[DeInit][Server]ip[%s] port[%u]", localIp.GetReadableAddress(), port);
307 2 : serverSocketMap_[portInfo]->DeInit();
308 2 : serverSocketMap_.erase(portInfo);
309 2 : serverSocketRefMap_.erase(portInfo);
310 : }
311 :
312 2 : return HCCL_SUCCESS;
313 2 : }
314 :
315 2 : HcclResult GlobalNetDevMgr::GetListenSocket(
316 : const HcclIpAddress& localIp, uint32_t port, std::shared_ptr<hccl::HcclSocket>& listenSocket)
317 : {
318 2 : PortInfo portInfo(localIp, port);
319 2 : std::lock_guard<std::mutex> lock(serverMapMutex_);
320 2 : auto it = serverSocketMap_.find(portInfo);
321 2 : if (it == serverSocketMap_.end() || it->second == nullptr) {
322 0 : HCCL_ERROR("[GlobalNetDevMgr][%s] no listen socket for port[%u]", __func__, port);
323 0 : return HCCL_E_NOT_FOUND;
324 : }
325 2 : listenSocket = it->second;
326 2 : return HCCL_SUCCESS;
327 2 : }
328 :
329 1 : HcclResult GlobalNetDevMgr::AddListenSocketWhiteList(
330 : const HcclIpAddress& localIp, uint32_t port, const std::vector<SocketWlistInfo>& wlistInfos)
331 : {
332 1 : if (wlistInfos.empty()) {
333 0 : HCCL_ERROR("[GlobalNetDevMgr][%s] empty whitelist", __func__);
334 0 : return HCCL_E_PARA;
335 : }
336 :
337 1 : std::shared_ptr<hccl::HcclSocket> listenSocket;
338 1 : CHK_RET(GetListenSocket(localIp, port, listenSocket));
339 1 : std::vector<SocketWlistInfo> mutableCopy = wlistInfos;
340 1 : return listenSocket->AddWhiteList(mutableCopy);
341 1 : }
342 :
343 1 : HcclResult GlobalNetDevMgr::AcceptDataSocket(
344 : const HcclIpAddress& localIp, uint32_t port, const std::string& tag,
345 : std::shared_ptr<hccl::HcclSocket>& outConnected, uint32_t acceptTimeoutMs)
346 : {
347 1 : std::shared_ptr<hccl::HcclSocket> listenSocket;
348 1 : CHK_RET(GetListenSocket(localIp, port, listenSocket));
349 1 : return listenSocket->Accept(tag, outConnected, acceptTimeoutMs);
350 1 : }
351 :
352 : HcclResult
353 1 : GlobalNetDevMgr::WaitClientSocketLinkEstablished(const std::shared_ptr<hccl::HcclSocket>& socket, s32 timeoutSec)
354 : {
355 1 : CHK_SMART_PTR_NULL(socket);
356 1 : u32 pollCount = 0;
357 1 : const auto startTime = std::chrono::steady_clock::now();
358 1 : const auto timeout = std::chrono::seconds(timeoutSec > 0 ? timeoutSec : GetExternalInputHcclLinkTimeOut());
359 1 : HCCL_DEBUG("[GlobalNetDevMgr][client][WaitLink] waiting for socket link up...");
360 : while (true) {
361 1 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
362 0 : HCCL_ERROR(
363 : "[GlobalNetDevMgr][client][WaitLink] wait socket establish timeout, timeout[%lld s]",
364 : static_cast<long long>(timeout.count()));
365 0 : socket->SetStatus(hccl::HcclSocketStatus::SOCKET_TIMEOUT);
366 0 : return HCCL_E_TIMEOUT;
367 : }
368 1 : const hccl::HcclSocketStatus status = socket->GetStatus();
369 1 : if (status == hccl::HcclSocketStatus::SOCKET_OK) {
370 1 : HCCL_DEBUG(
371 : "[GlobalNetDevMgr][client][WaitLink] socket established. localIp[%s], remoteIp[%s]",
372 : socket->GetLocalIp().GetReadableIP(), socket->GetRemoteIp().GetReadableIP());
373 1 : return HCCL_SUCCESS;
374 : }
375 0 : if (status == hccl::HcclSocketStatus::SOCKET_CONNECTING) {
376 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
377 0 : if (pollCount % 50U == 0U) {
378 0 : HCCL_DEBUG("[GlobalNetDevMgr][client][WaitLink] socket is connecting");
379 : }
380 0 : ++pollCount;
381 0 : continue;
382 : }
383 0 : if (status == hccl::HcclSocketStatus::SOCKET_TIMEOUT) {
384 0 : return HCCL_E_TIMEOUT;
385 : }
386 0 : socket->SetStatus(hccl::HcclSocketStatus::SOCKET_ERROR);
387 0 : return HCCL_E_TCP_CONNECT;
388 0 : }
389 : }
390 :
391 2 : void GlobalNetDevMgr::MakeSocketTag(
392 : hccl::HcclIpAddress tagServerIp, uint32_t tagServerPort, hccl::HcclIpAddress tagClientIp, std::string& socketTag)
393 : {
394 8 : socketTag = tagServerIp.GetReadableIP() + std::string(":") + std::to_string(tagServerPort) + std::string(":")
395 4 : + tagClientIp.GetReadableIP();
396 2 : }
397 :
398 1 : HcclResult GlobalNetDevMgr::ConnectToServer(
399 : uint32_t localPort, hccl::HcclIpAddress remoteIp, uint32_t remotePort, std::string& socketTag,
400 : std::shared_ptr<hccl::HcclSocket>& socket)
401 : {
402 1 : HCCL_INFO("[GlobalNetDevMgr]ConnectToServer start");
403 :
404 1 : hccl::HcclIpAddress localIpAddr;
405 1 : std::shared_ptr<hccl::HcclSocket> socketTemp = nullptr;
406 : {
407 1 : std::lock_guard<std::mutex> lock(netDevCtxMtx_);
408 1 : CHK_PTR_NULL(netDevCtx_);
409 1 : auto* netDevCtxPtr = static_cast<hccl::NetDevContext*>(netDevCtx_);
410 1 : localIpAddr = netDevCtxPtr->GetLocalIp();
411 :
412 1 : HCCL_INFO(
413 : "[GlobalNetDevMgr]ConnectToServer localIp[%s] localPort[%u] remoteIp[%s] remotePort[%u] socketTag[%s]",
414 : localIpAddr.GetReadableIP(), localPort, remoteIp.GetReadableIP(), remotePort, socketTag.c_str());
415 1 : HCCL_INFO("[GlobalNetDevMgr][client] ConnectToServer connect to server");
416 :
417 1 : EXCEPTION_CATCH(
418 : socketTemp = std::make_shared<hccl::HcclSocket>(
419 : socketTag, netDevCtx_, remoteIp, remotePort, hccl::HcclSocketRole::SOCKET_ROLE_CLIENT),
420 : return HCCL_E_PTR);
421 1 : }
422 :
423 1 : CHK_SMART_PTR_NULL(socketTemp);
424 1 : CHK_RET(socketTemp->Init());
425 1 : CHK_RET(socketTemp->Connect());
426 1 : HcclResult waitRet = WaitClientSocketLinkEstablished(socketTemp, 0);
427 1 : if (waitRet != HCCL_SUCCESS) {
428 0 : socketTemp->Close();
429 0 : return waitRet;
430 : }
431 :
432 1 : socket = socketTemp;
433 1 : HCCL_INFO(
434 : "[GlobalNetDevMgr]ConnectToServer done localPort[%u] remotePort[%u]", socket->GetLocalPort(),
435 : socket->GetRemotePort());
436 :
437 1 : return HCCL_SUCCESS;
438 1 : }
439 :
440 1 : HcclResult GlobalNetDevMgr::AcceptClient(
441 : uint32_t localPort, hccl::HcclIpAddress remoteIp, std::string& socketTag, std::shared_ptr<hccl::HcclSocket>& socket)
442 : {
443 1 : HCCL_INFO("[GlobalNetDevMgr]AcceptClient start");
444 :
445 1 : hccl::HcclIpAddress localIpAddr;
446 : {
447 1 : std::lock_guard<std::mutex> lock(netDevCtxMtx_);
448 1 : CHK_PTR_NULL(netDevCtx_);
449 1 : auto* netDevCtxPtr = static_cast<hccl::NetDevContext*>(netDevCtx_);
450 1 : localIpAddr = netDevCtxPtr->GetLocalIp();
451 1 : }
452 :
453 1 : HCCL_INFO(
454 : "[GlobalNetDevMgr]AcceptClient localIp[%s] localPort[%u] remoteIp[%s] socketTag[%s]",
455 : localIpAddr.GetReadableIP(), localPort, remoteIp.GetReadableIP(), socketTag.c_str());
456 :
457 1 : HCCL_INFO("[GlobalNetDevMgr][server] AcceptClient listen and accept");
458 1 : SocketWlistInfo wlistEntry{};
459 1 : wlistEntry.connLimit = 1U;
460 1 : const auto bin = remoteIp.GetBinaryAddress();
461 1 : wlistEntry.remoteIp.addr = bin.addr;
462 1 : wlistEntry.remoteIp.addr6 = bin.addr6;
463 1 : s32 mw = memcpy_s(wlistEntry.tag, sizeof(wlistEntry.tag), socketTag.c_str(), socketTag.size() + 1U);
464 1 : CHK_PRT_RET(mw != EOK, HCCL_ERROR("[GlobalNetDevMgr]memcpy_s whitelist tag failed"), HCCL_E_MEMORY);
465 2 : const std::vector<SocketWlistInfo> wlistVec = {wlistEntry};
466 1 : CHK_RET(AddListenSocketWhiteList(localIpAddr, localPort, wlistVec));
467 :
468 1 : std::shared_ptr<hccl::HcclSocket> socketTemp = nullptr;
469 1 : CHK_RET(AcceptDataSocket(localIpAddr, localPort, socketTag, socketTemp, 0));
470 1 : CHK_SMART_PTR_NULL(socketTemp);
471 :
472 1 : socket = socketTemp;
473 1 : HCCL_INFO(
474 : "[GlobalNetDevMgr]AcceptClient done localPort[%u] remotePort[%u]", socket->GetLocalPort(),
475 : socket->GetRemotePort());
476 :
477 1 : return HCCL_SUCCESS;
478 1 : }
479 :
480 2 : void GlobalNetDevMgr::CloseSocket(std::shared_ptr<hccl::HcclSocket>& socket)
481 : {
482 2 : if (socket != nullptr) {
483 2 : socket->Close();
484 2 : socket = nullptr;
485 : }
486 2 : }
487 : } // namespace hccl
|
