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 : #include <mutex>
11 : #include <set>
12 : #include <algorithm>
13 : #include "socket_manager.h"
14 : #include "socket_handle_manager.h"
15 : #include "communicator_impl.h"
16 : #include "null_ptr_exception.h"
17 : #include "exception_util.h"
18 : #include "stl_util.h"
19 : #include "preempt_port_manager.h"
20 : #include "timeout_exception.h"
21 : #include "p2p_enable_manager.h"
22 : #include "phy_topo_builder.h"
23 :
24 : namespace Hccl {
25 : std::mutex SocketManager::socketLock;
26 :
27 2 : void SocketManager::PrepareLinkAndServerInit(const SocketConfig &socketConfig)
28 : {
29 2 : LinkData link = socketConfig.link;
30 :
31 2 : if (!Contain(availableLinks, link)) {
32 2 : if (link.GetLinkProtocol() == LinkProtocol::PCIE) {
33 0 : std::vector<uint32_t> remoteDevices;
34 0 : remoteDevices.push_back(link.GetRemoteDeviceId());
35 0 : auto ret = P2PEnableManager::GetInstance().WaitP2PEnabled(remoteDevices);
36 0 : if (ret != HCCL_SUCCESS) {
37 0 : THROW<TimeoutException>(
38 0 : StringFormat("WaitP2PEnabled failed, devicePhyId=%d", link.GetRemoteDeviceId()));
39 : }
40 0 : }
41 2 : availableLinks.insert({link});
42 : }
43 :
44 2 : if (GetConnectedSocket(socketConfig) == nullptr) {
45 2 : auto portData = link.GetLocalPort();
46 2 : SocketRole role = link.GetLocalRankId() < link.GetRemoteRankId() ? SocketRole::SERVER : SocketRole::CLIENT;
47 2 : if (role == SocketRole::SERVER) {
48 2 : ServerInit(portData);
49 : }
50 : }
51 2 : }
52 :
53 1 : void SocketManager::ServerListen(const SocketConfig &socketConfig)
54 : {
55 1 : PrepareLinkAndServerInit(socketConfig);
56 1 : }
57 :
58 1 : void SocketManager::ConnectSockets(const SocketConfig &socketConfig)
59 : {
60 1 : if (GetConnectedSocket(socketConfig) == nullptr) {
61 1 : AddWhiteList(socketConfig);
62 1 : CreateConnectedSocket(socketConfig);
63 : }
64 1 : }
65 :
66 20 : void SocketManager::BatchCreateSockets(const vector<LinkData> &links)
67 : {
68 20 : vector<LinkData> pendingLinks;
69 28 : for (auto &link : links) {
70 8 : if (Contain(availableLinks, link)) {
71 0 : continue;
72 : }
73 8 : pendingLinks.emplace_back(link);
74 : }
75 :
76 20 : if (pendingLinks.empty()) {
77 18 : return;
78 : }
79 :
80 10 : for (auto &link: pendingLinks) {
81 8 : if (link.GetLinkProtocol() == LinkProtocol::PCIE) {
82 0 : std::vector<uint32_t> remoteDevices;
83 0 : remoteDevices.push_back(link.GetRemoteDeviceId());
84 0 : auto ret = P2PEnableManager::GetInstance().WaitP2PEnabled(remoteDevices);
85 0 : if (ret != HCCL_SUCCESS) {
86 0 : THROW<TimeoutException>(StringFormat("WaitP2PEnabled failed, devicePhyId=%d", link.GetRemoteDeviceId()));
87 : }
88 0 : }
89 : }
90 2 : BatchServerInit(pendingLinks);
91 2 : BatchAddWhiteList(pendingLinks);
92 2 : BatchCreateConnectedSockets(pendingLinks);
93 :
94 2 : availableLinks.insert(pendingLinks.begin(), pendingLinks.end());
95 20 : }
96 :
97 1 : void SocketManager::BatchCreateSockets(const SocketConfig &socketConfig)
98 : {
99 1 : PrepareLinkAndServerInit(socketConfig);
100 1 : if (GetConnectedSocket(socketConfig) == nullptr) {
101 1 : AddWhiteList(socketConfig);
102 1 : CreateConnectedSocket(socketConfig);
103 : }
104 1 : }
105 :
106 2 : void SocketManager::AddWhiteList(const SocketConfig &socketConfig)
107 : {
108 2 : unordered_map<PortData, vector<RaSocketWhitelist>> wlistMap{};
109 2 : LinkData link = socketConfig.link;
110 :
111 : // 通过虚拟拓扑获取Peer可能为空，如果为空，需要抛异，NullPtrException
112 : // 这里检查rankGraph完整性的逻辑是什么？
113 2 : SocketRole role = link.GetLocalRankId() < link.GetRemoteRankId() ? SocketRole::SERVER : SocketRole::CLIENT;
114 2 : if (role == SocketRole::SERVER) {
115 2 : if (comm) {
116 0 : auto peer = comm->GetRankGraph()->GetPeer(link.GetRemoteRankId());
117 0 : if (peer == nullptr) {
118 0 : auto msg = StringFormat("Fail to get peer of rank %d!", link.GetRemoteRankId());
119 0 : THROW<NullPtrException>(msg);
120 0 : }
121 0 : }
122 :
123 2 : RaSocketWhitelist wlistInfo{};
124 2 : wlistInfo.connLimit = 1;
125 2 : wlistInfo.remoteIp = link.GetRemoteAddr();
126 2 : wlistInfo.tag = socketConfig.GetHccpTag();
127 :
128 2 : auto port = link.GetLocalPort();
129 6 : vector<RaSocketWhitelist> wlistInfoVec{wlistInfo};
130 2 : AddWhiteList(port, wlistInfoVec);
131 2 : socketWlistMap[port] = wlistInfoVec;
132 2 : }
133 4 : }
134 :
135 2 : void SocketManager::BatchServerInit(const vector<LinkData> &links)
136 : {
137 10 : for (auto &link : links) {
138 8 : SocketRole role = link.GetLocalRankId() < link.GetRemoteRankId() ? SocketRole::SERVER : SocketRole::CLIENT;
139 8 : if (role == SocketRole::SERVER) {
140 6 : auto portData = link.GetLocalPort();
141 6 : ServerInit(portData);
142 : }
143 : }
144 2 : }
145 :
146 0 : void SocketManager::BatchAddWhiteList(const vector<LinkData> &links)
147 : {
148 0 : unordered_map<PortData, vector<RaSocketWhitelist>> wlistMap{};
149 :
150 0 : for (const auto &link : links) {
151 : // 通过虚拟拓扑获取Peer可能为空，如果为空，需要抛异，NullPtrException
152 0 : SocketRole role = link.GetLocalRankId() < link.GetRemoteRankId() ? SocketRole::SERVER : SocketRole::CLIENT;
153 0 : if (role == SocketRole::SERVER) {
154 0 : if (comm) {
155 0 : auto peer = comm->GetRankGraph()->GetPeer(link.GetRemoteRankId());
156 0 : if (peer == nullptr) {
157 0 : auto msg = StringFormat("Fail to get peer of rank %d!", link.GetRemoteRankId());
158 0 : THROW<NullPtrException>(msg);
159 0 : }
160 0 : }
161 :
162 0 : RaSocketWhitelist wlistInfo{};;
163 0 : wlistInfo.connLimit = 1;
164 0 : wlistInfo.remoteIp = link.GetRemoteAddr();
165 :
166 0 : std::string linkTag = socketTag_;
167 : // 获取到reuseIdx不为0时，tag需要拼接_reuseIdx；为0时不拼接，不影响原socket公用
168 0 : if (link.GetReuseIdx() != "0") {
169 0 : linkTag += ("_" + link.GetReuseIdx());
170 : }
171 0 : SocketConfig socketConfig(link.GetRemoteRankId(), link, linkTag);
172 0 : string hccpSocketTag = socketConfig.GetHccpTag();
173 :
174 0 : wlistInfo.tag = hccpSocketTag;
175 0 : wlistMap[link.GetLocalPort()].push_back(wlistInfo);
176 0 : }
177 : }
178 :
179 0 : for (auto &i : wlistMap) {
180 0 : auto port = i.first;
181 0 : AddWhiteList(port, i.second);
182 0 : socketWlistMap[port] = i.second;
183 : }
184 0 : }
185 :
186 2 : void SocketManager::BatchCreateConnectedSockets(const vector<LinkData> &links)
187 : {
188 10 : for (auto &link : links) {
189 8 : auto remoteRank = link.GetRemoteRankId();
190 8 : std::string socketTag = socketTag_;
191 8 : if (link.GetReuseIdx() != "0") {
192 0 : socketTag += ("_" + link.GetReuseIdx());
193 : }
194 8 : SocketConfig socketConfig(remoteRank, link, socketTag);
195 8 : CreateConnectedSocket(socketConfig);
196 8 : }
197 2 : }
198 :
199 8 : void SocketManager::ServerInit(PortData &localPort)
200 : {
201 8 : std::lock_guard<std::mutex> lock(socketLock);
202 8 : IpAddress ipAddress = localPort.GetAddr();
203 8 : u32 serverListenPort = localPort.GetType() == PortDeploymentType::P2P
204 8 : ? GetDeviceListenPort(localPort.GetRankId(), DEVICE_PORT_KEY_IPADDRESS)
205 8 : : GetDeviceListenPort(localPort.GetRankId(), ipAddress);
206 :
207 8 : auto &serverSocketMap = SocketManager::GetServerSocketMap();
208 8 : auto serverSocketInMap = serverSocketMap.find(localPort);
209 8 : if (serverSocketInMap != serverSocketMap.end()) {
210 6 : auto oldServerSocket = serverSocketMap.at(localPort);
211 6 : u32 oldServerListenPort = oldServerSocket->GetListenPort();
212 6 : if (oldServerListenPort != serverListenPort) {
213 : // 自定义算子的时候，会持有一个不关联通信域的SocketManager, 从而获取到的是默认端口，在单卡多进程的时候需要重新导向合适的端口。
214 : // 通信域算子又可以切换回来。
215 0 : bool success = oldServerSocket->Listen(serverListenPort);
216 0 : HCCL_INFO("[SocketManager::%s] %s change listen port %u to %u, ret[%u]", __func__,
217 : localPort.Describe().c_str(), oldServerListenPort, serverListenPort, success);
218 : }
219 18 : HCCL_INFO("[%s] find localPort in serverSocketMap, localPort [%s]", __func__, localPort.Describe().c_str());
220 6 : return;
221 6 : }
222 :
223 2 : SocketHandle hccpSocketHandle = SocketHandleManager::GetInstance().Create(devicePhyId, localPort);
224 2 : NicType nicType = localPort.GetType() == PortDeploymentType::P2P
225 : ? NicType::DEVICE_VNIC_TYPE
226 2 : : NicType::DEVICE_NIC_TYPE;
227 4 : auto serverSocket = socketProducer(ipAddress, ipAddress, serverListenPort, hccpSocketHandle, "server", SocketRole::SERVER, nicType);
228 2 : bool success = serverSocket->Listen(serverListenPort);
229 2 : if (success) {
230 6 : HCCL_RUN_INFO("[SocketManager::%s] Local %s listen the port %u success", __func__, localPort.Describe().c_str(), serverListenPort);
231 : } else {
232 0 : string msg = StringFormat("[SocketManager::%s] Local %s listen the port %u failed, maybe other process be listen it", __func__, localPort.Describe().c_str(), serverListenPort);
233 0 : MACRO_THROW(InvalidParamsException, msg);
234 0 : }
235 2 : serverSocketMap[localPort] = std::move(serverSocket);
236 8 : }
237 :
238 2 : void SocketManager::ServerInitAll(NewRankInfo &rankInfo)
239 : {
240 2 : vector<SocketPortRange> listenPortRanges = EnvConfig::GetInstance().GetHostNicConfig().GetDeviceSocketPortRange();
241 2 : if (listenPortRanges.empty()) {
242 3 : HCCL_RUN_INFO("[SocketManager::%s] socket port range not configured.", __func__);
243 1 : return;
244 : }
245 :
246 1 : const std::string &topoPath = CommunicatorImpl::GetTopoFilePath();
247 1 : PhyTopoBuilder::GetInstance().Build(topoPath);
248 :
249 1 : std::lock_guard<std::mutex> lock(socketLock);
250 1 : auto devLogicId = HrtGetDevice();
251 1 : auto &serverSocketMap = SocketManager::GetServerSocketMap();
252 1 : u32 rankId = rankInfo.rankId;
253 1 : u32 localId = rankInfo.localId;
254 1 : u32 devicePhyId = rankInfo.deviceId;
255 2 : for (auto &rankLevelInfo : rankInfo.rankLevelInfos) {
256 1 : shared_ptr<Graph<PhyTopo::Node, PhyTopo::Link>> graph = PhyTopo::GetInstance()->GetTopoGraph(rankLevelInfo.netLayer);
257 1 : if (graph == nullptr) {
258 0 : HCCL_DEBUG("[SocketManager::%s]Can't find the layout %u Graph!", __func__, rankLevelInfo.netLayer);
259 0 : continue;
260 0 : }
261 1 : std::vector<std::shared_ptr<PhyTopo::Link>> links = graph->GetEdges(localId);
262 7 : for (auto &link : links) {
263 6 : if (link->GetSourceIFace()->GetPos() == AddrPosition::HOST) {
264 0 : continue;
265 : }
266 18 : HCCL_DEBUG("[SocketManager::%s] find the device link %s", __func__, link->Describe().c_str());
267 6 : const std::set<LinkProtocol> &protocols = link->GetLinkProtocols();
268 12 : for (auto &protocol : protocols) {
269 6 : PortDeploymentType deployType = AddrPos2PortDeploymentType(link->GetSourceIFace()->GetPos(), protocol);
270 6 : LinkProtoType protoType = LinkProtocol2LinkProtoType(protocol);
271 6 : const std::set<std::string> &ports = link->GetSourceIFace()->GetPorts();
272 36 : for (auto &rankAddr : rankLevelInfo.rankAddrs) {
273 : // topo查得网口使用则打开建链
274 30 : std::set<std::string> intersectSet;
275 30 : std::set_intersection(ports.begin(), ports.end(), rankAddr.ports.begin(), rankAddr.ports.end(), std::inserter(intersectSet, intersectSet.begin()));
276 30 : if (intersectSet.empty()) {
277 27 : continue;
278 : }
279 3 : PortData localPort{static_cast<RankId>(rankId), deployType, protoType, 0, rankAddr.addr};
280 3 : u32 listenPort = DEFAULT_VALUE_TCPPORT;
281 3 : if (serverSocketMap.find(localPort) != serverSocketMap.end()) {
282 : // 单进程多通信域，找到老端口直接返回老端口
283 0 : listenPort = serverSocketMap[localPort]->GetListenPort();
284 0 : HCCL_INFO("[SocketManager::%s] Device %s use the old device port %u in same process.", __func__, localPort.Describe().c_str(), listenPort);
285 : } else {
286 : // 首次执行启用新端口
287 3 : SocketHandle hccpSocketHandle = SocketHandleManager::GetInstance().Create(devicePhyId, localPort);
288 3 : IpAddress ipAddress = localPort.GetAddr();
289 3 : NicType nicType = localPort.GetType() == PortDeploymentType::P2P ? NicType::DEVICE_VNIC_TYPE : NicType::DEVICE_NIC_TYPE;
290 3 : auto serverSocket = std::make_shared<Socket>(hccpSocketHandle, ipAddress, listenPort, ipAddress, "server", SocketRole::SERVER, nicType);
291 3 : PreemptPortManager::GetInstance(devLogicId).ListenPreempt(serverSocket, listenPortRanges, listenPort);
292 3 : serverSocketMap[localPort] = std::move(serverSocket);
293 9 : HCCL_RUN_INFO("[SocketManager::%s] Device %s listen the preempt port %u", __func__, localPort.Describe().c_str(), listenPort);
294 3 : }
295 3 : rankAddr.socketPort_ = listenPort;
296 3 : rankInfo.devicePort = listenPort;
297 30 : }
298 6 : }
299 6 : }
300 1 : }
301 2 : }
302 :
303 4 : bool SocketManager::ServerDeInit(PortData &localPort) const
304 : {
305 4 : std::lock_guard<std::mutex> lock(socketLock);
306 4 : auto &serverSocketMap = SocketManager::GetServerSocketMap();
307 4 : auto res = GetServerListenSocket(localPort);
308 : // 待修改 stop listen maybe needed
309 4 : if (res != nullptr) {
310 1 : serverSocketMap.erase(localPort);
311 : }
312 :
313 4 : return true;
314 4 : }
315 :
316 10 : Socket *SocketManager::CreateConnectedSocket(const SocketConfig &socketConfig)
317 : {
318 10 : auto res = GetConnectedSocket(socketConfig);
319 10 : if (res != nullptr) {
320 0 : return res;
321 : }
322 :
323 30 : HCCL_INFO("[SocketManager::%s] Create connected socket for tag %s.", __func__, socketConfig.tag.c_str());
324 :
325 10 : const PortData &localPort = socketConfig.link.GetLocalPort();
326 10 : const PortData &remotePort = socketConfig.link.GetRemotePort();
327 :
328 10 : auto socketHandle = SocketHandleManager::GetInstance().Get(devicePhyId, localPort);
329 10 : if (socketHandle == nullptr) {
330 0 : socketHandle = SocketHandleManager::GetInstance().Create(devicePhyId, socketConfig.link.GetLocalPort());
331 : }
332 :
333 10 : if (socketHandle == nullptr) {
334 0 : THROW<NullPtrException>(StringFormat("socketHandle of is nullptr, devicePhyId=%d, port=%s", devicePhyId,
335 0 : localPort.Describe().c_str()));
336 : }
337 10 : IpAddress localIpAddress = socketConfig.link.GetLocalAddr();
338 10 : IpAddress remoteIpAddress = socketConfig.link.GetRemoteAddr();
339 10 : SocketRole socketRole = socketConfig.GetRole();
340 10 : string hccpSocketTag = socketConfig.GetHccpTag();
341 :
342 10 : u32 serverListenPort = localPort.GetType() == PortDeploymentType::P2P
343 10 : ? GetDeviceListenPort(remotePort.GetRankId(), DEVICE_PORT_KEY_IPADDRESS)
344 10 : : GetDeviceListenPort(remotePort.GetRankId(), remoteIpAddress);
345 10 : NicType nicType = localPort.GetType() == PortDeploymentType::P2P
346 : ? NicType::DEVICE_VNIC_TYPE
347 10 : : NicType::DEVICE_NIC_TYPE;
348 : auto tmpSocket = socketProducer(localIpAddress, remoteIpAddress, serverListenPort, socketHandle, hccpSocketTag,
349 10 : socketRole, nicType);
350 30 : HCCL_INFO("[SocketManager::%s] Connect async the remote %s port %u.", __func__, remotePort.Describe().c_str(), serverListenPort);
351 10 : tmpSocket->ConnectAsync();
352 10 : connectedSocketMap[socketConfig] = std::move(tmpSocket);
353 10 : return connectedSocketMap[socketConfig].get();
354 10 : }
355 :
356 34 : Socket *SocketManager::GetConnectedSocket(const SocketConfig &socketConfig) const
357 : {
358 102 : HCCL_INFO("[SocketManager::%s] Get connected socket for tag %s.", __func__, socketConfig.tag.c_str());
359 34 : auto res = connectedSocketMap.find(socketConfig);
360 34 : if (res != connectedSocketMap.end()) {
361 20 : return res->second.get();
362 : }
363 :
364 14 : return nullptr;
365 : }
366 :
367 283 : void SocketManager::DestroyAll()
368 : {
369 285 : for (auto &i : socketWlistMap) {
370 2 : auto port = i.first;
371 2 : DelWhiteList(port, i.second);
372 : }
373 283 : socketWlistMap.clear();
374 :
375 331 : for (auto &socket : connectedSocketMap) {
376 48 : if (socket.second != nullptr) {
377 48 : socket.second->Destroy();
378 : }
379 : }
380 283 : connectedSocketMap.clear();
381 283 : availableLinks.clear();
382 283 : }
383 :
384 8 : Socket *SocketManager::GetServerListenSocket(const PortData &localPort) const
385 : {
386 8 : auto &serverSocketMap = SocketManager::GetServerSocketMap();
387 8 : auto res = serverSocketMap.find(localPort);
388 8 : if (res != serverSocketMap.end()) {
389 1 : return (res->second).get();
390 : }
391 :
392 7 : return nullptr;
393 : }
394 :
395 280 : SocketManager::SocketManager(
396 : const CommunicatorImpl &communicator, u32 localRank, u32 devicePhyId, u32 deviceLogicId,
397 : std::function<shared_ptr<Socket>(IpAddress &localIpAddress, IpAddress &remoteIpAddress, u32 listenPort,
398 : SocketHandle socketHandle, const std::string &tag, SocketRole socketRole,
399 : NicType nicType)>
400 280 : socketProducer)
401 280 : : comm(&communicator), localRank(localRank), devicePhyId(devicePhyId), deviceLogicId_(deviceLogicId)
402 : {
403 280 : if (socketProducer != nullptr) {
404 0 : this->socketProducer = socketProducer;
405 : }
406 :
407 280 : if (comm != nullptr) {
408 280 : socketTag_ = comm->GetEstablishLinkSocketTag();
409 : }
410 280 : }
411 :
412 3 : SocketManager::SocketManager(u32 localRank, u32 devicePhyId, u32 deviceLogicId, const std::string &socketTag)
413 3 : : comm(nullptr), localRank(localRank), devicePhyId(devicePhyId), deviceLogicId_(deviceLogicId)
414 : {
415 3 : socketTag_ = socketTag;
416 3 : }
417 :
418 2 : void SocketManager::AddWhiteList(PortData &localPort, vector<RaSocketWhitelist> &wlistInfoVec) const
419 : {
420 2 : auto socketHandle = SocketHandleManager::GetInstance().Get(devicePhyId, localPort);
421 2 : if (socketHandle == nullptr) {
422 0 : THROW<NullPtrException>(StringFormat("socketHandle of is nullptr, devicePhyId=%d, port=%s", devicePhyId,
423 0 : localPort.Describe().c_str()));
424 : }
425 2 : HrtRaSocketWhiteListAdd(socketHandle, wlistInfoVec);
426 2 : }
427 :
428 2 : bool SocketManager::DelWhiteList(PortData &localPort, vector<RaSocketWhitelist> &wlistInfoVec) const
429 : {
430 2 : auto socketHandle = SocketHandleManager::GetInstance().Get(devicePhyId, localPort);
431 2 : if (socketHandle == nullptr) {
432 0 : return false;
433 : }
434 2 : HrtRaSocketWhiteListDel(socketHandle, wlistInfoVec);
435 2 : return true;
436 : }
437 :
438 5 : void SocketManager::SetDeviceServerListenPortMap(const std::unordered_map<u32, std::unordered_map<IpAddress, u32>> &rankListenPortMap)
439 : {
440 5 : std::lock_guard<std::mutex> lock(socketLock);
441 5 : rankListenPortMap_ = rankListenPortMap;
442 5 : }
443 :
444 0 : std::unordered_map<u32, std::unordered_map<IpAddress, u32>> SocketManager::GetSubCommDeviceServerListenPortMap(const std::vector<u32> &rankIds) const
445 : {
446 0 : std::lock_guard<std::mutex> lock(socketLock);
447 0 : std::unordered_map<u32, std::unordered_map<IpAddress, u32>> subRankListenPortMap;
448 0 : for (u32 subRankId = 0; subRankId < rankIds.size(); ++subRankId) {
449 0 : u32 rankId = rankIds[subRankId];
450 0 : if (rankListenPortMap_.find(rankId) == rankListenPortMap_.end()) {
451 0 : HCCL_WARNING("[SocketManager::%s]Cant't find listen port for rank %u to sub comm.", __func__, rankId);
452 : } else {
453 0 : subRankListenPortMap.insert(std::make_pair(subRankId, rankListenPortMap_.at(rankId)));
454 : }
455 : }
456 0 : return subRankListenPortMap;
457 0 : }
458 :
459 18 : u32 SocketManager::GetDeviceListenPort(const u32 &rankId, const IpAddress &ipAddress)
460 : {
461 18 : u32 listenPort = rankListenPortMap_[rankId][ipAddress];
462 18 : if (listenPort == 0) {
463 12 : listenPort = DEFAULT_VALUE_TCPPORT;
464 12 : rankListenPortMap_[rankId][ipAddress] = listenPort;
465 36 : HCCL_WARNING("[SocketManager::%s] Can't find rankId[%u], addr[%s] listen port, use default", __func__, rankId, ipAddress.Describe().c_str());
466 : }
467 18 : return listenPort;
468 : }
469 :
470 283 : SocketManager::~SocketManager()
471 : {
472 283 : DECTOR_TRY_CATCH("SocketManager", DestroyAll());
473 283 : }
474 :
475 25 : std::unordered_map<PortData, shared_ptr<Socket>> &SocketManager::GetServerSocketMap()
476 : {
477 25 : static std::unordered_map<PortData, shared_ptr<Socket>> serverSocketMap;
478 25 : return serverSocketMap;
479 : }
480 :
481 1 : bool SocketManager::CheckServerPortListening(const PortData &portData, const uint32_t port) const
482 : {
483 1 : std::lock_guard<std::mutex> lock(socketLock);
484 1 : auto &serverSocketMap = SocketManager::GetServerSocketMap();
485 1 : auto iterSocket = serverSocketMap.find(portData);
486 1 : if (iterSocket == serverSocketMap.end()) {
487 0 : return false;
488 : }
489 1 : if (iterSocket->second->GetListenPort() != port) {
490 1 : return false;
491 : }
492 0 : return true;
493 1 : }
494 :
495 : } // namespace Hccl
|
