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