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 <chrono>
12 : #include <algorithm>
13 :
14 : #include "socket_mgr.h"
15 : #include "hcomm_adapter_runtime.h"
16 : #include "../channels/channel.h"
17 : #include "orion_adpt_utils.h"
18 : #include "host_socket_handle_manager.h"
19 : #include "exception_handler.h"
20 : #include "adapter_rts.h"
21 : #include "env_config/env_config.h"
22 :
23 : namespace hcomm {
24 :
25 : constexpr uint32_t TempServerListenPort = 60001; // 临时固定监听端口，用于功能验证
26 : constexpr uint32_t kHostResourceId = 0U;
27 :
28 : s32 g_linkTimeout = 0;
29 1 : inline s32 EnvLinkTimeoutGet()
30 : {
31 1 : g_linkTimeout = g_linkTimeout != 0 ? g_linkTimeout : Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut();
32 1 : return g_linkTimeout;
33 : }
34 :
35 40 : SocketMgr& SocketMgr::GetInstance(s32 phyId)
36 : {
37 170 : static SocketMgr instances[MAX_MODULE_DEVICE_NUM]; // C++11 保证线程安全
38 40 : if (static_cast<u32>(phyId) >= MAX_MODULE_DEVICE_NUM) {
39 0 : HCCL_WARNING("[SocketMgr] devicePhyId >= MAX_MODULE_DEVICE_NUM, devicePhyId=%d, MAX_MODULE_DEVICE_NUM=%d",
40 : phyId, MAX_MODULE_DEVICE_NUM);
41 0 : return instances[0];
42 : }
43 40 : instances[phyId].devicePhyId_ = phyId;
44 40 : return instances[phyId];
45 : }
46 :
47 16 : HcclResult SocketMgr::Init()
48 : {
49 16 : uint32_t runtimeDevicePhyId = 0;
50 16 : bool noDevice = false;
51 16 : CHK_RET(ResolveRuntimeDevicePhyId(runtimeDevicePhyId, noDevice));
52 16 : if (isLoaded_ && isHostOnlyInit_ == noDevice) {
53 9 : return HCCL_SUCCESS;
54 : }
55 : // 覆盖 GetInstance 和 EndpointPair 直接构造 SocketMgr 两种路径，保持 devicePhyId_ 与 runtime 当前设备一致。
56 7 : devicePhyId_ = noDevice ? kHostResourceId : runtimeDevicePhyId;
57 7 : isLoaded_ = true;
58 7 : isHostOnlyInit_ = noDevice;
59 7 : serverListenPort_ = TempServerListenPort;
60 7 : HCCL_INFO("[SocketMgr][%s] init socket mgr, noDevice[%d], runtimeDevicePhyId[%u], devicePhyId[%u].",
61 : __func__, noDevice, runtimeDevicePhyId, devicePhyId_);
62 7 : return HCCL_SUCCESS;
63 : }
64 :
65 11 : HcclResult SocketMgr::AddWhiteList(const Hccl::SocketConfig &socketConfig, const Hccl::SocketHandle &socketHandle)
66 : {
67 : EXCEPTION_HANDLE_BEGIN
68 :
69 : // 1. 创建 wlistInfo 对象
70 11 : Hccl::RaSocketWhitelist wlistInfo{};;
71 11 : wlistInfo.connLimit = 1;
72 11 : wlistInfo.remoteIp = socketConfig.link.GetRemoteAddr();
73 11 : wlistInfo.tag = socketConfig.GetHccpTag();
74 11 : handle2WhiteListMap_[socketHandle].push_back(wlistInfo);
75 :
76 11 : std::vector<Hccl::RaSocketWhitelist> wlistInfoVec;
77 11 : wlistInfoVec.clear();
78 11 : wlistInfoVec.push_back(wlistInfo);
79 :
80 : // 2. 加入白名单
81 11 : Hccl::HrtRaSocketWhiteListAdd(socketHandle, wlistInfoVec);
82 :
83 11 : EXCEPTION_HANDLE_END
84 11 : return HCCL_SUCCESS;
85 : }
86 :
87 12 : HcclResult SocketMgr::GetSocketHandle(const Hccl::SocketConfig &socketConfig, Hccl::SocketHandle &socketHandle)
88 : {
89 : EXCEPTION_HANDLE_BEGIN
90 :
91 : // 加异常捕获
92 12 : auto localPort = socketConfig.link.GetLocalPort();
93 12 : if (localPort.GetType() == Hccl::PortDeploymentType::DEV_NET) {
94 6 : socketHandle = Hccl::SocketHandleManager::GetInstance().Get(devicePhyId_, localPort);
95 6 : if (socketHandle == nullptr) {
96 2 : socketHandle = Hccl::SocketHandleManager::GetInstance().Create(devicePhyId_, localPort);
97 : }
98 6 : } else if (localPort.GetType() == Hccl::PortDeploymentType::HOST_NET){
99 5 : socketHandle = Hccl::HostSocketHandleManager::GetInstance().Get(devicePhyId_, localPort.GetAddr());
100 5 : if (socketHandle == nullptr) {
101 0 : socketHandle = Hccl::HostSocketHandleManager::GetInstance().Create(devicePhyId_, localPort.GetAddr());
102 : }
103 : } else {
104 1 : HCCL_ERROR(
105 : "[SocketMgr] PortDeploymentType = %d, not support create socket.", localPort.GetType().Describe().c_str());
106 1 : return HCCL_E_NOT_SUPPORT;
107 : }
108 11 : if (socketHandle == nullptr) {
109 0 : HCCL_ERROR("[SocketMgr] socketHandle is nullptr, devicePhyId=%d, localPort[%s]",
110 : devicePhyId_, localPort.Describe().c_str());
111 0 : return HCCL_E_INTERNAL;
112 : }
113 11 : HCCL_INFO("[SocketMgr][%s] socketHandle[%p] devicePhyId[%u] localPort[%s]",
114 : __func__, socketHandle, devicePhyId_, localPort.Describe().c_str());
115 :
116 0 : EXCEPTION_HANDLE_END
117 11 : return HCCL_SUCCESS;
118 : }
119 :
120 11 : HcclResult SocketMgr::CreateSocket(const Hccl::SocketConfig &socketConfig, const Hccl::SocketHandle &socketHandle)
121 : {
122 : EXCEPTION_HANDLE_BEGIN
123 :
124 11 : Hccl::IpAddress localIpAddress = socketConfig.link.GetLocalAddr();
125 11 : Hccl::IpAddress remoteIpAddress = socketConfig.link.GetRemoteAddr();
126 11 : Hccl::SocketRole socketRole = socketConfig.GetRole();
127 11 : std::string hccpSocketTag = socketConfig.GetHccpTag();
128 11 : serverListenPort_ = socketConfig.listeningPort; // serverListenPort_这个变量似乎没用
129 :
130 11 : std::unique_ptr<Hccl::Socket> tmpSocket = nullptr;
131 11 : if (socketConfig.link.GetType() == Hccl::PortDeploymentType::DEV_NET) {
132 6 : EXCEPTION_CATCH(
133 : tmpSocket = std::make_unique<Hccl::Socket>(
134 : socketHandle, localIpAddress, socketConfig.listeningPort,
135 : remoteIpAddress, hccpSocketTag,
136 : socketRole, Hccl::NicType::DEVICE_NIC_TYPE
137 : ),
138 : return HCCL_E_PTR
139 : );
140 6 : HCCL_INFO("[SocketMgr][%s] client_socket_info[%s]", __func__, tmpSocket->Describe().c_str());
141 6 : tmpSocket->ConnectAsync();
142 5 : } else if (socketConfig.link.GetType() == Hccl::PortDeploymentType::HOST_NET) {
143 5 : EXCEPTION_CATCH(
144 : tmpSocket = std::make_unique<Hccl::Socket>(socketHandle,
145 : localIpAddress,
146 : socketConfig.listeningPort,
147 : remoteIpAddress,
148 : hccpSocketTag,
149 : socketRole,
150 : Hccl::NicType::HOST_NIC_TYPE),
151 : return HCCL_E_PTR
152 : );
153 5 : HCCL_INFO("[SocketMgr][%s] client_socket_info[%s]", __func__, tmpSocket->Describe().c_str());
154 5 : tmpSocket->Connect();
155 : } else {
156 0 : HCCL_ERROR(
157 : "[SocketMgr] PortDeploymentType = %d, not support create socket.", socketConfig.link.GetType().Describe().c_str());
158 0 : return HCCL_E_NOT_SUPPORT;
159 : }
160 :
161 11 : socketMap_[socketConfig] = std::move(tmpSocket);
162 11 : socketInUseMap_[socketMap_[socketConfig].get()] = false;
163 :
164 11 : EXCEPTION_HANDLE_END
165 11 : return HCCL_SUCCESS;
166 : }
167 :
168 12 : HcclResult SocketMgr::CreateSocketWithSocketHandle(const Hccl::SocketConfig &socketConfig)
169 : {
170 : Hccl::SocketHandle socketHandle;
171 12 : CHK_RET(GetSocketHandle(socketConfig, socketHandle));
172 11 : CHK_RET(AddWhiteList(socketConfig, socketHandle));
173 11 : CHK_RET(CreateSocket(socketConfig, socketHandle));
174 :
175 11 : return HCCL_SUCCESS;
176 : }
177 :
178 10 : HcclResult SocketMgr::MakeSocketInUse(Hccl::Socket*& socket)
179 : {
180 10 : if (socketInUseMap_.find(socket) != socketInUseMap_.end()) {
181 10 : socketInUseMap_[socket] = true;
182 : } else {
183 0 : HCCL_ERROR("[SocketMgr][%s] CreateSocket succeeded but socket not found in socketInUseMap",
184 : __func__);
185 0 : return HCCL_E_INTERNAL;
186 : }
187 10 : return HCCL_SUCCESS;
188 : }
189 :
190 12 : HcclResult SocketMgr::GetNewSocket(const Hccl::SocketConfig &socketConfig, Hccl::Socket*& socket)
191 : {
192 12 : CHK_RET(CreateSocketWithSocketHandle(socketConfig));
193 :
194 : // 再次查找
195 : std::unordered_map<Hccl::SocketConfig,
196 : std::unique_ptr<Hccl::Socket>>::iterator it =
197 11 : socketMap_.find(socketConfig);
198 11 : if (it == socketMap_.end()) {
199 0 : HCCL_ERROR("[SocketMgr][%s] CreateSocket succeeded but socket not found in socketMap",
200 : __func__);
201 0 : return HCCL_E_INTERNAL;
202 : }
203 11 : socket = it->second.get();
204 11 : return HCCL_SUCCESS;
205 : }
206 :
207 11 : HcclResult SocketMgr::GetSocket(const Hccl::SocketConfig &socketConfig, Hccl::Socket*& socket)
208 : {
209 11 : std::unique_lock<std::mutex> lock(mutex_);
210 11 : CHK_RET(Init());
211 : // 1. 先查找
212 : std::unordered_map<Hccl::SocketConfig,
213 : std::unique_ptr<Hccl::Socket>>::iterator it =
214 11 : socketMap_.begin();
215 :
216 14 : for (; it != socketMap_.end(); ++it) {
217 4 : if (std::equal_to<Hccl::SocketConfig>{}(socketConfig, it->first)) {
218 1 : socket = it->second.get();
219 1 : break;
220 : }
221 : }
222 11 : if (it != socketMap_.end()) {
223 1 : if (socketConfig.hostNic2DeviceNicMode_) {
224 0 : HCCL_INFO("[SocketMgr][%s] destroy a socket[%p] in hostNic2DeviceNicMode", __func__, static_cast<void*>(socket));
225 0 : socket->Destroy();
226 0 : socketMap_.erase(it);
227 0 : socketInUseMap_.erase(socket);
228 : } else {
229 1 : HCCL_INFO("[SocketMgr][%s] find a correct socket in map", __func__);
230 1 : auto timeoutPoint = std::chrono::steady_clock::now() +
231 2 : std::chrono::seconds(EnvLinkTimeoutGet()) - std::chrono::seconds(10);
232 1 : while(socketInUseMap_[socket] == true) {
233 0 : auto currentTime = std::chrono::steady_clock::now();
234 0 : if (currentTime >= timeoutPoint) {
235 0 : HCCL_ERROR("[SocketMgr][%s] Get Socket Time Out", __func__);
236 0 : return HCCL_E_TIMEOUT;
237 : }
238 : }
239 1 : CHK_RET(MakeSocketInUse(socket));
240 1 : return HCCL_SUCCESS;
241 : }
242 : }
243 :
244 : // 2. 不存在则创建
245 10 : CHK_RET(GetNewSocket(socketConfig, socket));
246 9 : CHK_RET(MakeSocketInUse(socket));
247 9 : return HCCL_SUCCESS;
248 11 : }
249 :
250 : // 仅通信域管理层的host网卡使用，后续需归一到通信域管理层的socket管理模块
251 3 : HcclResult SocketMgr::GetHostSocket(const Hccl::SocketConfig &socketConfig, Hccl::Socket*& socket)
252 : {
253 3 : CHK_RET(Init());
254 : // 1. 先查找
255 3 : auto it = socketMap_.find(socketConfig);
256 :
257 3 : if (it != socketMap_.end()) {
258 1 : if (socketConfig.hostNic2DeviceNicMode_) {
259 0 : socket = it->second.get();
260 0 : HCCL_INFO("[SocketMgr][%s] destroy a socket[%p] in hostNic2DeviceNicMode", __func__, static_cast<void*>(socket));
261 0 : socket->Destroy();
262 0 : socketMap_.erase(it);
263 0 : socketInUseMap_.erase(socket);
264 : } else {
265 1 : socket = it->second.get();
266 1 : return HCCL_SUCCESS;
267 : }
268 : }
269 :
270 : // 2. 不存在则创建
271 2 : CHK_RET(GetNewSocket(socketConfig, socket));
272 2 : return HCCL_SUCCESS;
273 : }
274 :
275 4 : HcclResult SocketMgr::PutSocket(const Hccl::SocketConfig*& socketConfig, Hccl::Socket*& socket)
276 : {
277 4 : HCCL_INFO("[SocketMgr][%s] start to put a socket", __func__);
278 4 : CHK_PTR_NULL(socket);
279 4 : CHK_RET(UpdateSocketConfig(socketConfig, socket));
280 4 : for (auto it = socketMap_.begin(); it != socketMap_.end(); ++it) {
281 1 : if (it->second.get() == socket) {
282 1 : socketInUseMap_[it->second.get()] = false;
283 1 : socketAvailableCv_.notify_all();
284 1 : socket = nullptr;
285 1 : return HCCL_SUCCESS;
286 : }
287 : }
288 3 : HCCL_INFO("[SocketMgr][%s] socket not found in socketInUseMap", __func__);
289 3 : return HCCL_SUCCESS;
290 : }
291 :
292 4 : HcclResult SocketMgr::UpdateSocketConfig(const Hccl::SocketConfig*& socketConfig, Hccl::Socket*& socket)
293 : {
294 4 : for (auto it = socketMap_.begin(); it != socketMap_.end(); ++it) {
295 1 : if (it->second.get() == socket) {
296 1 : socketConfig = &(it->first);
297 1 : return HCCL_SUCCESS;
298 : }
299 : }
300 3 : HCCL_INFO("[SocketMgr][%s] socket not found in socketMap", __func__);
301 3 : return HCCL_SUCCESS;
302 : }
303 :
304 6 : HcclResult SocketMgr::DeleteWhiteList(Hccl::Socket* socket)
305 : {
306 6 : std::unique_lock<std::mutex> lock(mutex_);
307 6 : CHK_PTR_NULL(socket);
308 6 : bool socketExist = false;
309 6 : for (auto it = socketMap_.begin(); it != socketMap_.end(); ++it) {
310 6 : if (it->second.get() == socket) {
311 6 : socketExist = true;
312 6 : break;
313 : }
314 : }
315 6 : if (!socketExist) {
316 0 : HCCL_WARNING("[DeleteWhiteList] socket[%p] not found in socketMap_, nothing to delete.",
317 : static_cast<void*>(socket));
318 0 : return HCCL_SUCCESS;
319 : }
320 6 : auto iter = handle2WhiteListMap_.find(socket->GetFdHandle());
321 6 : if (iter == handle2WhiteListMap_.end()) {
322 6 : HCCL_WARNING("[DeleteWhiteList] socketHandle[%p] not found in handle2WhiteListMap_, nothing to delete.",
323 : socket->GetFdHandle());
324 6 : return HCCL_SUCCESS;
325 : }
326 :
327 0 : std::vector<Hccl::RaSocketWhitelist> &wlistInfoVec = iter->second;
328 0 : if (wlistInfoVec.empty()) {
329 0 : HCCL_WARNING("[DeleteWhiteList] socketHandle[%p] has empty white list, nothing to delete.", socket->GetFdHandle());
330 0 : return HCCL_SUCCESS;
331 : }
332 :
333 0 : EXCEPTION_CATCH(Hccl::HrtRaSocketWhiteListDel(socket->GetFdHandle(), wlistInfoVec), return HCCL_E_INTERNAL);
334 0 : handle2WhiteListMap_.erase(iter);
335 :
336 0 : return HCCL_SUCCESS;
337 6 : }
338 :
339 8 : HcclResult SocketMgr::DestroySocket(Hccl::Socket* socket)
340 : {
341 8 : std::unique_lock<std::mutex> lock(mutex_);
342 8 : if (socket == nullptr) {
343 0 : HCCL_WARNING("[DestroySocket] socket is nullptr, nothing to destroy.");
344 0 : return HCCL_SUCCESS;
345 : }
346 8 : bool socketExist = false;
347 8 : for (auto it = socketMap_.begin(); it != socketMap_.end(); ++it) {
348 6 : if (it->second.get() == socket) {
349 6 : socketExist = true;
350 6 : HCCL_INFO("[DestroySocket] Erasing socket inuse info with tag[%s] from socketInUseMap.", it->first.GetHccpTag().c_str());
351 6 : socketInUseMap_.erase(socket);
352 6 : HCCL_INFO("[DestroySocket] Erasing socket with tag[%s] from socketMap.", it->first.GetHccpTag().c_str());
353 6 : socketMap_.erase(it);
354 6 : break;
355 : }
356 : }
357 8 : if (!socketExist) {
358 2 : HCCL_WARNING("[DestroySocket] socket is not exist in socketMap_, nothing to destroy.");
359 2 : return HCCL_SUCCESS;
360 : }
361 6 : return HCCL_SUCCESS;
362 8 : }
363 :
364 3 : void SocketMgr::DeInit(u32 devPhyId)
365 : {
366 3 : HCCL_INFO("[SocketMgr][%s] DeInit devPhyId[%u]", __func__, devPhyId);
367 3 : auto &inst = GetInstance(static_cast<s32>(devPhyId));
368 3 : std::lock_guard<std::mutex> lock(inst.mutex_);
369 3 : for (auto &it : inst.socketMap_) {
370 0 : if (it.second != nullptr) {
371 0 : it.second->Destroy();
372 0 : it.second.reset();
373 : }
374 : }
375 3 : inst.socketMap_.clear();
376 3 : inst.socketInUseMap_.clear();
377 3 : inst.handle2WhiteListMap_.clear();
378 3 : inst.isLoaded_ = false;
379 3 : }
380 :
381 : } // namespace hcomm
|
