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 "server_socket_manager.h"
12 :
13 : namespace hcomm {
14 26 : HcclResult ServerSocketManager::ServerSocketStartListen(
15 : const Hccl::PortData& localPort, const Hccl::NicType nicType, const uint32_t devPhyId, uint32_t* port)
16 : {
17 26 : if (nicType == Hccl::NicType::HOST_NIC_TYPE) {
18 15 : CHK_RET(HostSocketListen(localPort, devPhyId, port));
19 11 : } else if (nicType == Hccl::NicType::DEVICE_NIC_TYPE) {
20 10 : CHK_RET(DeviceSocketListen(localPort, devPhyId, port));
21 : } else {
22 1 : HCCL_ERROR("[ServerSocketManager][%s] illegal NicType[%s]", __func__, nicType.Describe().c_str());
23 1 : return HCCL_E_PARA;
24 : }
25 25 : return HCCL_SUCCESS;
26 : }
27 :
28 : HcclResult
29 15 : ServerSocketManager::HostSocketListen(const Hccl::PortData& localPort, const uint32_t devPhyId, uint32_t* port)
30 : {
31 15 : std::lock_guard<std::mutex> lock(hostMutex_);
32 15 : uint32_t requestedPort = *port;
33 15 : if (hostServerSocketMap_.find(localPort) != hostServerSocketMap_.end()
34 15 : && hostServerSocketMap_[localPort].find(requestedPort) != hostServerSocketMap_[localPort].end()) {
35 6 : if (hostServerSocketMap_[localPort][requestedPort].second == UINT32_MAX) {
36 0 : HCCL_ERROR("[ServerSocketManager][%s]port listening count overflow UINT32_MAX", __func__);
37 0 : return HCCL_E_INTERNAL;
38 : }
39 6 : hostServerSocketMap_[localPort][requestedPort].second
40 6 : = hostServerSocketMap_[localPort][requestedPort].second + 1; // 计数+1
41 6 : HCCL_INFO("[ServerSocketManager][%s] reuse serverSocket", __func__);
42 6 : return HCCL_SUCCESS;
43 : }
44 :
45 9 : if (socketMgrCompat_ == nullptr) {
46 1 : EXCEPTION_CATCH(socketMgrCompat_ = std::make_unique<Hccl::SocketManager>(), return HCCL_E_INTERNAL);
47 : }
48 9 : bool isListen = socketMgrCompat_->CheckServerPortListening(localPort, requestedPort);
49 :
50 9 : Hccl::SocketHandle socketHandle{};
51 9 : EXCEPTION_CATCH(
52 : socketHandle = Hccl::HostSocketHandleManager::GetInstance().Create(devPhyId, localPort.GetAddr()),
53 : return HCCL_E_PARA);
54 :
55 9 : std::unique_ptr<Hccl::Socket> serverSocket{};
56 9 : EXCEPTION_CATCH(
57 : serverSocket = std::make_unique<Hccl::Socket>(
58 : socketHandle, localPort.GetAddr(), requestedPort, localPort.GetAddr(), "server", Hccl::SocketRole::SERVER,
59 : Hccl::NicType::HOST_NIC_TYPE),
60 : return HCCL_E_PARA); // 端口号可能冲突，需要SE做决定
61 9 : HCCL_INFO("[ServerSocketManager][%s] listen_socket_info[%s]", __func__, serverSocket->Describe().c_str());
62 9 : uint32_t actualPort = requestedPort;
63 9 : CHK_RET(ListenSocketIfNeeded(serverSocket.get(), isListen, actualPort, port));
64 :
65 9 : hostServerSocketMap_[localPort][actualPort] = std::make_pair(std::move(serverSocket), 1);
66 :
67 9 : return HCCL_SUCCESS;
68 15 : }
69 :
70 : HcclResult
71 10 : ServerSocketManager::DeviceSocketListen(const Hccl::PortData& localPort, const uint32_t devPhyId, uint32_t* port)
72 : {
73 10 : std::lock_guard<std::mutex> lock(deviceMutex_);
74 10 : uint32_t requestedPort = *port;
75 10 : if (deviceServerSocketMap_.find(localPort) != deviceServerSocketMap_.end()
76 10 : && deviceServerSocketMap_[localPort].find(requestedPort) != deviceServerSocketMap_[localPort].end()) {
77 1 : if (deviceServerSocketMap_[localPort][requestedPort].second == UINT32_MAX) {
78 0 : HCCL_ERROR("[ServerSocketManager][%s]port listening count overflow UINT32_MAX", __func__);
79 0 : return HCCL_E_INTERNAL;
80 : }
81 1 : deviceServerSocketMap_[localPort][requestedPort].second
82 1 : = deviceServerSocketMap_[localPort][requestedPort].second + 1; // 计数+1
83 1 : HCCL_INFO("[ServerSocketManager][%s] reuse serverSocket", __func__);
84 1 : return HCCL_SUCCESS;
85 : }
86 :
87 9 : if (socketMgrCompat_ == nullptr) {
88 0 : EXCEPTION_CATCH(socketMgrCompat_ = std::make_unique<Hccl::SocketManager>(), return HCCL_E_INTERNAL);
89 : }
90 : // 查询socketMgrCompat_，如果查询到已有serversocket，？new一个空壳，放进map里面，计数+1
91 9 : bool isListen = socketMgrCompat_->CheckServerPortListening(localPort, requestedPort);
92 :
93 9 : Hccl::SocketHandle socketHandle{};
94 9 : EXCEPTION_CATCH(
95 : socketHandle = Hccl::SocketHandleManager::GetInstance().Create(devPhyId, localPort), return HCCL_E_PARA);
96 :
97 9 : std::unique_ptr<Hccl::Socket> serverSocket;
98 9 : EXCEPTION_CATCH(
99 : serverSocket = std::make_unique<Hccl::Socket>(
100 : socketHandle, localPort.GetAddr(), requestedPort, localPort.GetAddr(), "server", Hccl::SocketRole::SERVER,
101 : Hccl::NicType::DEVICE_NIC_TYPE),
102 : return HCCL_E_PARA); // 端口号可能冲突，需要SE做决定
103 9 : HCCL_INFO("[ServerSocketManager][%s] listen_socket_info[%s]", __func__, serverSocket->Describe().c_str());
104 9 : uint32_t actualPort = requestedPort;
105 9 : CHK_RET(ListenSocketIfNeeded(serverSocket.get(), isListen, actualPort, port));
106 9 : deviceServerSocketMap_[localPort][actualPort] = std::make_pair(std::move(serverSocket), 1);
107 :
108 9 : return HCCL_SUCCESS;
109 10 : }
110 :
111 16 : HcclResult ServerSocketManager::ServerSocketStopListen(
112 : const Hccl::PortData& localPort, const Hccl::NicType nicType, const uint32_t port)
113 : {
114 16 : if (nicType == Hccl::NicType::DEVICE_NIC_TYPE) {
115 7 : CHK_RET(DeviceSocketStopListen(localPort, port));
116 9 : } else if (nicType == Hccl::NicType::HOST_NIC_TYPE) {
117 9 : CHK_RET(HostSocketStopListen(localPort, port));
118 : } else {
119 0 : HCCL_ERROR("[ServerSocketManager][%s] illegal NicType[%s]", __func__, nicType.Describe().c_str());
120 0 : return HCCL_E_PARA;
121 : }
122 11 : return HCCL_SUCCESS;
123 : }
124 :
125 7 : HcclResult ServerSocketManager::DeviceSocketStopListen(const Hccl::PortData& localPort, const uint32_t port)
126 : {
127 7 : std::lock_guard<std::mutex> lock(deviceMutex_);
128 7 : if (deviceServerSocketMap_.find(localPort) != deviceServerSocketMap_.end()
129 7 : && deviceServerSocketMap_[localPort].find(port) != deviceServerSocketMap_[localPort].end()) {
130 4 : if (deviceServerSocketMap_[localPort][port].second == 0) {
131 0 : HCCL_ERROR("[ServerSocketManager][%s]port[%u] listening count already zero", __func__, port);
132 0 : return HCCL_E_INTERNAL;
133 : }
134 4 : deviceServerSocketMap_[localPort][port].second = deviceServerSocketMap_[localPort][port].second - 1; // 计数-1
135 4 : if (deviceServerSocketMap_[localPort][port].second == 0) {
136 3 : deviceServerSocketMap_[localPort].erase(port);
137 3 : CHK_RET(DeInitCompatServerIfListening(localPort, port));
138 : }
139 4 : if (deviceServerSocketMap_[localPort].empty()) {
140 3 : deviceServerSocketMap_.erase(localPort);
141 : }
142 4 : return HCCL_SUCCESS;
143 : }
144 3 : HCCL_ERROR(
145 : "[ServerSocketManager][%s] Can not stop listen cause {PortData[%s], port[%u]} is Not Listening", __func__,
146 : localPort.Describe().c_str(), port);
147 3 : return HCCL_E_NOT_FOUND;
148 7 : }
149 :
150 9 : HcclResult ServerSocketManager::HostSocketStopListen(const Hccl::PortData& localPort, const uint32_t port)
151 : {
152 9 : std::lock_guard<std::mutex> lock(hostMutex_);
153 9 : if (hostServerSocketMap_.find(localPort) != hostServerSocketMap_.end()
154 9 : && hostServerSocketMap_[localPort].find(port) != hostServerSocketMap_[localPort].end()) {
155 7 : if (hostServerSocketMap_[localPort][port].second == 0) {
156 0 : HCCL_ERROR("[ServerSocketManager][%s]port[%u] listening count already zero", __func__, port);
157 0 : return HCCL_E_INTERNAL;
158 : }
159 7 : hostServerSocketMap_[localPort][port].second = hostServerSocketMap_[localPort][port].second - 1; // 计数-1
160 7 : if (hostServerSocketMap_[localPort][port].second == 0) {
161 3 : hostServerSocketMap_[localPort].erase(port);
162 3 : CHK_RET(DeInitCompatServerIfListening(localPort, port));
163 : }
164 7 : if (hostServerSocketMap_[localPort].empty()) {
165 3 : hostServerSocketMap_.erase(localPort);
166 : }
167 7 : return HCCL_SUCCESS;
168 : }
169 2 : HCCL_ERROR(
170 : "[ServerSocketManager][%s] Can not stop listen cause {PortData[%s], port[%u]} is Not Listening", __func__,
171 : localPort.Describe().c_str(), port);
172 2 : return HCCL_E_NOT_FOUND;
173 9 : }
174 :
175 18 : HcclResult ServerSocketManager::ListenSocketIfNeeded(
176 : Hccl::Socket* serverSocket, bool isListen, uint32_t& actualPort, uint32_t* outPort)
177 : {
178 : // port=0 表示自动分配端口，必须调 Listen(actualPort) 获取实际端口，不受 isListen 影响
179 18 : if (actualPort == 0) {
180 6 : EXCEPTION_CATCH(serverSocket->Listen(actualPort), return HCCL_E_INTERNAL);
181 6 : HCCL_INFO("[ServerSocketManager][%s] allocated port[%u]", __func__, actualPort);
182 6 : *outPort = actualPort;
183 6 : return HCCL_SUCCESS;
184 : }
185 : // 指定端口且已有监听则跳过，避免重复 Listen
186 12 : if (isListen) {
187 12 : HCCL_INFO("[ServerSocketManager][%s] port[%u] has been listen", __func__, actualPort);
188 12 : return HCCL_SUCCESS;
189 : }
190 0 : EXCEPTION_CATCH(serverSocket->Listen(), return HCCL_E_INTERNAL);
191 0 : return HCCL_SUCCESS;
192 : }
193 :
194 6 : HcclResult ServerSocketManager::DeInitCompatServerIfListening(const Hccl::PortData& localPort, const uint32_t port)
195 : {
196 6 : if (socketMgrCompat_ == nullptr) {
197 0 : EXCEPTION_CATCH(socketMgrCompat_ = std::make_unique<Hccl::SocketManager>(), return HCCL_E_INTERNAL);
198 : }
199 6 : bool isListen = socketMgrCompat_->CheckServerPortListening(localPort, port);
200 6 : if (isListen) {
201 : Hccl::PortData portDataCopy(
202 6 : localPort.GetRankId(), localPort.GetType(), localPort.GetProto(), localPort.GetId(), localPort.GetAddr());
203 6 : socketMgrCompat_->ServerDeInit(portDataCopy);
204 : }
205 6 : return HCCL_SUCCESS;
206 : }
207 :
208 14 : void ServerSocketManager::DeInitDeviceSockets(u32 devPhyId)
209 : {
210 14 : std::lock_guard<std::mutex> lock(deviceMutex_);
211 22 : for (auto it = deviceServerSocketMap_.begin(); it != deviceServerSocketMap_.end();) {
212 8 : if (static_cast<uint32_t>(it->first.GetRankId()) == devPhyId) {
213 12 : for (auto& portEntry : it->second) {
214 6 : if (portEntry.second.first != nullptr) {
215 6 : portEntry.second.first->Destroy();
216 6 : portEntry.second.first.reset();
217 : }
218 : }
219 6 : it = deviceServerSocketMap_.erase(it);
220 : } else {
221 2 : ++it;
222 : }
223 : }
224 14 : }
225 :
226 14 : void ServerSocketManager::DeInitHostSockets(u32 devPhyId)
227 : {
228 14 : std::lock_guard<std::mutex> lock(hostMutex_);
229 20 : for (auto it = hostServerSocketMap_.begin(); it != hostServerSocketMap_.end();) {
230 6 : if (static_cast<uint32_t>(it->first.GetRankId()) == devPhyId) {
231 10 : for (auto& portEntry : it->second) {
232 5 : if (portEntry.second.first != nullptr) {
233 5 : portEntry.second.first->Destroy();
234 5 : portEntry.second.first.reset();
235 : }
236 : }
237 5 : it = hostServerSocketMap_.erase(it);
238 : } else {
239 1 : ++it;
240 : }
241 : }
242 14 : }
243 :
244 11 : void ServerSocketManager::DeInit(u32 devPhyId)
245 : {
246 11 : HCCL_INFO("[ServerSocketManager][%s] DeInit[%u]", __func__, devPhyId);
247 11 : DeInitDeviceSockets(devPhyId);
248 11 : DeInitHostSockets(devPhyId);
249 11 : }
250 :
251 : } // namespace hcomm
|
