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