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