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