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 "dlhal_function.h"
12 : #include "externalinput_pub.h"
13 : #include "adapter_hccp.h"
14 : #include "network_manager_pub.h"
15 : #include "sal_pub.h"
16 : #include "hccl_net_dev_defs.h"
17 : #include "hccl_network.h"
18 : #include "network/hccp_common.h"
19 : #include "adapter_error_manager_pub.h"
20 : #include "hccl_socket.h"
21 :
22 : namespace hccl {
23 : constexpr u32 MAX_MSG_STR_LEN = 2 * 1024;
24 :
25 213 : HcclSocket::HcclSocket(const std::string &tag, HcclNetDevCtx netDevCtx,
26 : const HcclIpAddress &remoteIp, u32 remotePort,
27 213 : HcclSocketRole localRole)
28 213 : : tag_(tag), netDevCtx_(netDevCtx),
29 639 : remoteIp_(remoteIp), remotePort_(remotePort), localRole_(localRole),
30 213 : status_(HcclSocketStatus::SOCKET_INIT), fdHandle_(nullptr)
31 : {
32 213 : }
33 :
34 87 : HcclSocket::HcclSocket(HcclNetDevCtx netDevCtx, u32 localPort)
35 174 : : netDevCtx_(netDevCtx), localPort_(localPort), localRole_(HcclSocketRole::SOCKET_ROLE_RESERVED),
36 87 : status_(HcclSocketStatus::SOCKET_INIT), fdHandle_(nullptr)
37 : {
38 87 : }
39 :
40 300 : HcclSocket::~HcclSocket()
41 : {
42 300 : DeInit();
43 300 : }
44 :
45 236 : HcclResult HcclSocket::Init()
46 : {
47 236 : CHK_PTR_NULL(netDevCtx_);
48 236 : socketType_ = (static_cast<hccl::NetDevContext *>(netDevCtx_))->GetNicType();
49 238 : localDevicePhyId_ = (static_cast<hccl::NetDevContext *>(netDevCtx_))->GetPhyId();
50 238 : localDeviceLogicId_ = (static_cast<hccl::NetDevContext *>(netDevCtx_))->GetLogicId();
51 238 : localIp_ = (static_cast<hccl::NetDevContext *>(netDevCtx_))->GetLocalIp();
52 238 : backupIp_ = (static_cast<hccl::NetDevContext *>(netDevCtx_))->GetBackupIp();
53 :
54 : // 默认场景下，只有VNIC使用强制断链，其它场景还是走优雅断链
55 238 : forceClose_ = (socketType_ == NicType::VNIC_TYPE);
56 :
57 238 : return HCCL_SUCCESS;
58 : }
59 :
60 253 : HcclResult HcclSocket::DeInit()
61 : {
62 253 : Close();
63 253 : if (listened_) {
64 30 : if (socketType_ == NicType::VNIC_TYPE) {
65 6 : CHK_RET(NetworkManager::GetInstance(localDeviceLogicId_).StopVnic(localIp_, localPort_));
66 24 : } else if (socketType_ == NicType::DEVICE_NIC_TYPE) {
67 4 : CHK_RET(NetworkManager::GetInstance(localDeviceLogicId_).StopNic(localIp_, localPort_));
68 : } else {
69 20 : CHK_RET(NetworkManager::GetInstance(localDeviceLogicId_).StopHostNetAndListen(
70 : nicSocketHandle_, localIp_, localPort_));
71 : }
72 :
73 28 : listened_ = false;
74 28 : HCCL_INFO("[HcclSocket][DeInit] device[%d] stops listen on ip[%s], port[%u] success, socketType[%u].",
75 : localDeviceLogicId_, localIp_.GetReadableAddress(), localPort_, socketType_);
76 : }
77 :
78 251 : return HCCL_SUCCESS;
79 : }
80 :
81 20 : HcclResult HcclSocket::ListenHostNet(HcclResult &ret, std::string &errormessage, u32 &port)
82 : {
83 20 : bool rdmaFlag = !GetExternalInputHcclIsTcpMode();
84 20 : u32 proto = 0;
85 : SocketHandle hostSocketHandle;
86 20 : HcclNetDevGetProtoType(netDevCtx_, proto);
87 20 : if (rdmaFlag && proto == HCCL_PROTO_TYPE_ROCE) {
88 0 : rdmaFlag = true;
89 : } else {
90 20 : rdmaFlag = false;
91 : }
92 20 : ret = NetworkManager::GetInstance(localDeviceLogicId_).StartHostNetAndListen(localIp_, hostSocketHandle,
93 : port, rdmaFlag);
94 40 : errormessage = "The IP address " + std::string(localIp_.GetReadableIP()) + " and port " +
95 60 : std::to_string(port) + " have already been bound.";
96 20 : RPT_INPUT_ERR(ret == HCCL_E_UNAVAIL, "EI0019", std::vector<std::string>({"reason"}),
97 : std::vector<std::string>({errormessage}));
98 20 : return HCCL_SUCCESS;
99 0 : }
100 :
101 27 : HcclResult HcclSocket::Listen()
102 : {
103 27 : CHK_PRT_RET(localPort_ == HCCL_INVALID_PORT,
104 : HCCL_ERROR("[HcclSocket][Listen]No port is set, please listen with a valid port."), HCCL_E_INTERNAL);
105 27 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
106 :
107 27 : HcclResult ret = HCCL_E_RESERVED;
108 27 : std::string errormessage = "";
109 27 : if (socketType_ == NicType::VNIC_TYPE) {
110 16 : ret = NetworkManager::GetInstance(localDeviceLogicId_).StartVnic(localIp_, localPort_);
111 32 : errormessage = "The IP address " + std::string(localIp_.GetReadableIP()) +
112 48 : " and port " + std::to_string(localPort_) + " have already been bound.";
113 16 : RPT_INPUT_ERR(ret == HCCL_E_UNAVAIL, "EI0020", std::vector<std::string>({"reason"}),
114 : std::vector<std::string>({errormessage}));
115 11 : } else if (socketType_ == NicType::DEVICE_NIC_TYPE) {
116 4 : bool rdmaFlag = !GetExternalInputHcclIsTcpMode();
117 4 : HCCL_DEBUG("[%s]StartNic localDeviceLogicId_[%d], localIp_[%s], localPort_[%u], rdmaFlag[%d], "
118 : "socketType_[%d], backupIp_[%s]", __func__, localDeviceLogicId_, localIp_.GetReadableIP(),
119 : localPort_, rdmaFlag, socketType_, backupIp_.GetReadableIP());
120 : // 如果是backup，传入额外的rdev信息
121 4 : ret = NetworkManager::GetInstance(localDeviceLogicId_).StartNic(localIp_, localPort_, rdmaFlag, backupIp_);
122 8 : errormessage = "The IP address " + std::string(localIp_.GetReadableIP()) + " and port " +
123 12 : std::to_string(localPort_) + " have already been bound.";
124 4 : RPT_INPUT_ERR(ret == HCCL_E_UNAVAIL, "EI0020", std::vector<std::string>({"reason"}),
125 : std::vector<std::string>({errormessage}));
126 : } else {
127 7 : CHK_RET(ListenHostNet(ret, errormessage, localPort_));
128 : }
129 27 : std::stringstream tmpMsgstream;
130 81 : tmpMsgstream << ((socketType_ == NicType::HOST_NIC_TYPE) ? ("[" + LOG_KEYWORDS_INIT_CHANNEL + "]") :
131 74 : ("[" + LOG_KEYWORDS_INIT_GROUP + "]")) << "[" << LOG_KEYWORDS_RANKTABLE_DETECT << "]";
132 27 : std::string errmsg = tmpMsgstream.str();
133 27 : CHK_PRT_RET(ret != HCCL_SUCCESS,
134 : HCCL_ERROR("%s socket type[%u], listen on ip[%s] and specific port[%u] fail. "
135 : "Please check the port status and whether the port is being used by other process.",
136 : errmsg.c_str(), socketType_, localIp_.GetReadableAddress(), localPort_), ret);
137 :
138 21 : CHK_RET(GetNicSocketHandle());
139 :
140 15 : listened_ = true;
141 15 : HCCL_INFO("[HcclSocket][Listen] device[%d] listens on ip[%s] port[%u] success, socketType[%u].",
142 : localDeviceLogicId_, localIp_.GetReadableAddress(), localPort_, socketType_);
143 :
144 15 : return HCCL_SUCCESS;
145 27 : }
146 :
147 15 : HcclResult HcclSocket::Listen(u32 port)
148 : {
149 15 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
150 :
151 15 : HcclResult ret = HCCL_E_RESERVED;
152 15 : std::string errormessage = "";
153 15 : HCCL_INFO("[HcclSocket][Listen] device[%d] trying to listen on port[%u]", localDeviceLogicId_, port);
154 15 : if (socketType_ == NicType::VNIC_TYPE) {
155 1 : ret = NetworkManager::GetInstance(localDeviceLogicId_).StartVnic(localIp_, port);
156 2 : errormessage = "The IP address " + std::string(localIp_.GetReadableIP()) +
157 3 : " and port " + std::to_string(localPort_) + " have already been bound.";
158 1 : RPT_INPUT_ERR(ret == HCCL_E_UNAVAIL, "EI0020", std::vector<std::string>({"reason"}),
159 : std::vector<std::string>({errormessage}));
160 14 : } else if (socketType_ == NicType::DEVICE_NIC_TYPE) {
161 1 : bool rdmaFlag = false;
162 1 : HCCL_DEBUG("[%s]StartNic localDeviceLogicId_[%d], localIp_[%s], localPort_[%u], rdmaFlag[%d], "
163 : "socketType_[%d], backupIp_[%s]", __func__, localDeviceLogicId_, localIp_.GetReadableIP(),
164 : port, rdmaFlag, socketType_, backupIp_.GetReadableIP());
165 : // 如果是backup，传入额外的rdev信息
166 1 : ret = NetworkManager::GetInstance(localDeviceLogicId_).StartNic(localIp_, port, rdmaFlag, backupIp_);
167 2 : errormessage = "The IP address " + std::string(localIp_.GetReadableIP()) +
168 3 : " and port " + std::to_string(localPort_) + " have already been bound.";
169 1 : RPT_INPUT_ERR(ret == HCCL_E_UNAVAIL, "EI0020", std::vector<std::string>({"reason"}),
170 : std::vector<std::string>({errormessage}));
171 : } else {
172 13 : CHK_RET(ListenHostNet(ret, errormessage, port));
173 : }
174 15 : std::stringstream tmpMsgstream;
175 45 : tmpMsgstream << ((socketType_ == NicType::HOST_NIC_TYPE) ? ("[" + LOG_KEYWORDS_INIT_CHANNEL + "]") :
176 32 : ("[" + LOG_KEYWORDS_INIT_GROUP + "]")) << "[" << LOG_KEYWORDS_RANKTABLE_DETECT << "]";
177 15 : std::string errmsg = tmpMsgstream.str();
178 15 : CHK_PRT_RET(ret == HCCL_E_UNAVAIL,
179 : HCCL_INFO("%s socket type[%u], Could not listen on IP [%s] and port [%u], port already in use.",
180 : errmsg.c_str(), socketType_, localIp_.GetReadableAddress(), port), ret);
181 15 : CHK_PRT_RET(ret != HCCL_SUCCESS,
182 : HCCL_ERROR("%s socket type[%u], listen on ip[%s] and port[%u] fail,.",
183 : errmsg.c_str(), socketType_, localIp_.GetReadableAddress(), port), ret);
184 :
185 15 : CHK_RET(GetNicSocketHandle());
186 :
187 15 : localPort_ = port;
188 15 : listened_ = true;
189 15 : HCCL_INFO("[HcclSocket][Listen] device[%d] listens on ip[%s] port[%u] success, socketType[%u].",
190 : localDeviceLogicId_, localIp_.GetReadableAddress(), localPort_, socketType_);
191 :
192 15 : return HCCL_SUCCESS;
193 15 : }
194 3 : HcclResult HcclSocket::AddWhiteList(std::vector<SocketWlistInfo> &wlistInfoVec)
195 : {
196 3 : if (listened_ == false) {
197 0 : return HCCL_E_NOT_FOUND;
198 : }
199 :
200 3 : std::vector<struct SocketWlistInfoT> wlistInfosVec;
201 6 : for (auto remote : wlistInfoVec) {
202 3 : struct SocketWlistInfoT wlistInfo = {0};
203 3 : wlistInfo.connLimit = remote.connLimit;
204 3 : wlistInfo.remoteIp.addr = remote.remoteIp.addr;
205 3 : wlistInfo.remoteIp.addr6 = remote.remoteIp.addr6;
206 3 : s32 sRet = memcpy_s(&wlistInfo.tag[0], sizeof(wlistInfo.tag), remote.tag, sizeof(remote.tag));
207 3 : if (sRet != EOK) {
208 0 : HCCL_ERROR("[Add][SocketWhiteList]memory copy failed. errorno[%d]", sRet);
209 0 : return HCCL_E_MEMORY;
210 : }
211 3 : wlistInfosVec.push_back(wlistInfo);
212 : }
213 :
214 3 : CHK_RET(hrtRaSocketWhiteListAdd(nicSocketHandle_, wlistInfosVec.data(), wlistInfosVec.size()));
215 :
216 3 : return HCCL_SUCCESS;
217 3 : }
218 :
219 1 : HcclResult HcclSocket::DelWhiteList(std::vector<SocketWlistInfo> &wlistInfoVec)
220 : {
221 1 : if (listened_ == false) {
222 0 : return HCCL_E_NOT_FOUND;
223 : }
224 :
225 1 : std::vector<struct SocketWlistInfoT> wlistInfosVec;
226 2 : for (auto remote : wlistInfoVec) {
227 1 : struct SocketWlistInfoT wlistInfo = {0};
228 1 : wlistInfo.connLimit = remote.connLimit;
229 1 : wlistInfo.remoteIp.addr = remote.remoteIp.addr;
230 1 : wlistInfo.remoteIp.addr6 = remote.remoteIp.addr6;
231 1 : s32 sRet = memcpy_s(&wlistInfo.tag[0], sizeof(wlistInfo.tag), remote.tag, sizeof(remote.tag));
232 1 : if (sRet != EOK) {
233 0 : HCCL_ERROR("[Delete][SocketWhiteList]memory copy failed. errorno[%d]", sRet);
234 0 : return HCCL_E_MEMORY;
235 : }
236 1 : wlistInfosVec.push_back(wlistInfo);
237 : }
238 :
239 1 : CHK_RET(hrtRaSocketWhiteListDel(nicSocketHandle_, wlistInfosVec.data(), wlistInfosVec.size()));
240 :
241 1 : return HCCL_SUCCESS;
242 1 : }
243 :
244 47 : HcclResult HcclSocket::Connect()
245 : {
246 47 : if (status_ != HcclSocketStatus::SOCKET_INIT) {
247 0 : HCCL_ERROR("[Connect]socket status[%d] is not SOCKET_INIT, can not connect", status_);
248 0 : return HCCL_E_TCP_CONNECT;
249 : }
250 :
251 47 : CHK_RET(GetNicSocketHandle());
252 :
253 : // 作为客户端时, 向远端发起 Connect 请求; 作为服务端时, 暂什么也不做
254 9 : if (localRole_ == HcclSocketRole::SOCKET_ROLE_CLIENT) {
255 9 : SocketConnectInfoT connectInfo {};
256 9 : connectInfo.remoteIp.addr = remoteIp_.GetBinaryAddress().addr;
257 9 : connectInfo.remoteIp.addr6 = remoteIp_.GetBinaryAddress().addr6;
258 9 : connectInfo.socketHandle = nicSocketHandle_;
259 9 : connectInfo.port = remotePort_;
260 9 : CHK_SAFETY_FUNC_RET(strcpy_s(connectInfo.tag, SOCK_CONN_TAG_SIZE, tag_.c_str()));
261 :
262 9 : HCCL_INFO("[Connect] localIp[%s], remoteIp[%s], socketHandle[%lu], tag[%s], port[%u]",
263 : localIp_.GetReadableAddress(), remoteIp_.GetReadableAddress(),
264 : nicSocketHandle_, connectInfo.tag, remotePort_);
265 :
266 16 : HcclResult ret = hrtRaSocketBatchConnect(&connectInfo, 1, MAX_VALUE_U32, [this]() -> bool { return this->GetStopFlag(); });
267 9 : CHK_PRT_RET(ret != HCCL_SUCCESS,
268 : HCCL_ERROR("[Connect] call ra socket connect failed. errorno[%d]", ret), ret);
269 :
270 9 : status_ = HcclSocketStatus::SOCKET_CONNECTING;
271 : }
272 :
273 9 : return HCCL_SUCCESS;
274 : }
275 :
276 323 : void HcclSocket::Close()
277 : {
278 323 : HCCL_INFO("[Close] localIp[%s], remoteIp[%s], socketHandle[%lu], tag[%s], port[%u] status[%d]",
279 : localIp_.GetReadableAddress(), remoteIp_.GetReadableAddress(),
280 : nicSocketHandle_, tag_.c_str(), remotePort_, status_);
281 : // 若socket处于超时状态，调用abort接口终止连接请求
282 323 : if (status_ == HcclSocketStatus::SOCKET_TIMEOUT) {
283 : // 刷新status，防止重复调用abort接口
284 2 : status_ = HcclSocketStatus::SOCKET_ERROR;
285 2 : bool isSupportRaSocketAbort = false;
286 2 : (void)IsSupportRaSocketAbort(isSupportRaSocketAbort);
287 : // 作为客户端时, 终止向远端发起的 connect 请求; 作为服务端时, 暂什么也不做
288 4 : if (isSupportRaSocketAbort && localRole_ == HcclSocketRole::SOCKET_ROLE_CLIENT && !remoteIp_.IsInvalid() &&
289 2 : nicSocketHandle_) {
290 2 : SocketConnectInfoT connectInfo {};
291 2 : connectInfo.remoteIp.addr = remoteIp_.GetBinaryAddress().addr;
292 2 : connectInfo.remoteIp.addr6 = remoteIp_.GetBinaryAddress().addr6;
293 2 : connectInfo.socketHandle = nicSocketHandle_;
294 2 : connectInfo.port = remotePort_;
295 2 : strcpy_s(connectInfo.tag, SOCK_CONN_TAG_SIZE, tag_.c_str());
296 :
297 2 : HcclResult ret = hrtRaSocketNonBlockBatchAbort(&connectInfo, 1);
298 2 : if (ret != HCCL_SUCCESS) {
299 0 : HCCL_ERROR("[Abort] call ra socket abort failed. ret[%d]", ret);
300 : } else {
301 2 : HCCL_RUN_INFO("[Abort] call ra socket abort success. tag[%s]", connectInfo.tag);
302 : }
303 : }
304 : } else {
305 : // socket处于其他状态时调用close接口关闭socket
306 321 : if (fdHandle_ == nullptr) {
307 310 : HCCL_WARNING("[Close] socket's fdHandle is null, do not need close.");
308 310 : return;
309 : }
310 11 : SocketCloseInfoT closeInfo = {0};
311 11 : closeInfo.socketHandle = nicSocketHandle_;
312 11 : closeInfo.fdHandle = fdHandle_;
313 11 : closeInfo.disuseLinger = static_cast<s32>(forceClose_);
314 11 : HCCL_DEBUG("[HcclSocket][Close] socketType[%d] nicHandle[%p] fdHandle[%p] disuseLinger[%d]", socketType_,
315 : nicSocketHandle_, fdHandle_, closeInfo.disuseLinger);
316 :
317 11 : HcclResult sRet = hrtRaSocketBatchClose(&closeInfo, 1, 1);
318 11 : if (sRet != HCCL_SUCCESS) {
319 0 : HCCL_WARNING("[Close] errNo[0x%016llx] server socket batch close ret[%d] failed. not fatal", sRet);
320 : }
321 :
322 11 : fdHandle_ = nullptr;
323 : }
324 :
325 13 : return;
326 : }
327 :
328 : // 此接口用于DFX, 仅在 HcclSocketManager 中判断为连接异常时调用
329 41 : void HcclSocket::SetStatus(HcclSocketStatus status)
330 : {
331 41 : if (status != HcclSocketStatus::SOCKET_TIMEOUT && status != HcclSocketStatus::SOCKET_ERROR) {
332 0 : HCCL_WARNING("[Set]]Status] Only support set SOCKET_TIMEOUT or SOCKET_ERROR, status[%d]", status);
333 0 : return;
334 : }
335 41 : status_ = status;
336 : }
337 :
338 15 : HcclSocketStatus HcclSocket::ConvertRaSocketStatus(int raStatus)
339 : {
340 15 : HcclSocketStatus status = HcclSocketStatus::SOCKET_INIT;
341 : /**< socket status:0 not connected 1:connected 2:connect timeout 3:connecting */
342 15 : switch (raStatus) {
343 0 : case 0: // 0 not connected
344 0 : status = HcclSocketStatus::SOCKET_INIT;
345 0 : break;
346 15 : case 1: // 1:connected
347 15 : status = HcclSocketStatus::SOCKET_OK;
348 15 : break;
349 0 : case 2: // 2:connect timeout
350 0 : status = HcclSocketStatus::SOCKET_TIMEOUT;
351 0 : break;
352 0 : case 3: // 3:connecting
353 0 : status = HcclSocketStatus::SOCKET_CONNECTING;
354 0 : break;
355 0 : default:
356 0 : status = HcclSocketStatus::SOCKET_ERROR;
357 0 : break;
358 : }
359 15 : return status;
360 : }
361 :
362 248 : HcclSocketStatus HcclSocket::GetStatus()
363 : {
364 248 : if (status_ == HcclSocketStatus::SOCKET_OK ||
365 247 : status_ == HcclSocketStatus::SOCKET_TIMEOUT ||
366 247 : status_ == HcclSocketStatus::SOCKET_ERROR) {
367 77 : HCCL_DEBUG("[Get][Status]socket status is [%d].", status_);
368 78 : return status_;
369 : }
370 :
371 171 : if (GetNicSocketHandle()) {
372 125 : return HcclSocketStatus::SOCKET_INIT;
373 : }
374 :
375 : // 疑问: Listen Socket 会是什么状态？
376 :
377 45 : SocketInfoT socketInfo {};
378 45 : socketInfo.remoteIp.addr = remoteIp_.GetBinaryAddress().addr;
379 45 : socketInfo.remoteIp.addr6 = remoteIp_.GetBinaryAddress().addr6;
380 45 : socketInfo.socketHandle = nicSocketHandle_;
381 45 : s32 ret = strcpy_s(socketInfo.tag, SOCK_CONN_TAG_SIZE, tag_.c_str());
382 45 : CHK_PRT_RET(ret != 0,
383 : HCCL_ERROR("[Get][Status]strcpy_s failed. ret[%u]", ret), HcclSocketStatus::SOCKET_ERROR);
384 :
385 45 : u32 connectedNum = 0;
386 45 : s32 sockRet = hrtRaGetSockets(static_cast<u32>(localRole_), &socketInfo, 1, &connectedNum);
387 45 : if ((connectedNum == 0 && sockRet == 0) || (sockRet == SOCK_EAGAIN)) {
388 30 : return HcclSocketStatus::SOCKET_CONNECTING;
389 15 : } else if (sockRet != 0) {
390 0 : HCCL_ERROR("[Get][Status]get rasocket error. role[%u] sockRet[%d] ", localRole_, sockRet);
391 0 : return HcclSocketStatus::SOCKET_ERROR;
392 : } else {
393 15 : if (connectedNum == 1) {
394 15 : status_ = ConvertRaSocketStatus(socketInfo.status);
395 15 : fdHandle_ = socketInfo.fdHandle;
396 15 : HCCL_INFO("[Get][Status]status_[%u] ", status_);
397 15 : return status_;
398 : } else {
399 0 : HCCL_ERROR("[Get][Status]total Sockets[%u], more than needed num[1]!", connectedNum);
400 0 : return HcclSocketStatus::SOCKET_ERROR;
401 : }
402 : }
403 : }
404 :
405 7 : HcclResult HcclSocket::Accept(const std::string &tag, std::shared_ptr<HcclSocket> &socket, u32 acceptTimeOut)
406 : {
407 7 : if (listened_ == false) {
408 0 : HCCL_ERROR("[Accept]socket no listen, can not accepted.");
409 0 : return HCCL_E_PARA;
410 : }
411 :
412 7 : EXCEPTION_CATCH((socket = std::make_shared<HcclSocket>(tag,
413 : netDevCtx_, remoteIp_, 0, HcclSocketRole::SOCKET_ROLE_SERVER)), return HCCL_E_PTR);
414 7 : CHK_SMART_PTR_NULL(socket);
415 7 : CHK_RET(socket->Init());
416 :
417 7 : HCCL_INFO("[Accept]localIp[%s], remoteIp[%s], socketHandle[%p], tag[%s]",
418 : localIp_.GetReadableAddress(), remoteIp_.GetReadableAddress(), nicSocketHandle_, tag.c_str());
419 :
420 7 : s32 acceptTimeOutTmp = static_cast<s32>(acceptTimeOut);
421 7 : s32 timer = (acceptTimeOutTmp > 0 && acceptTimeOutTmp < GetExternalInputHcclLinkTimeOut()) ?
422 7 : acceptTimeOutTmp: GetExternalInputHcclLinkTimeOut();
423 :
424 7 : auto startTime = std::chrono::steady_clock::now();
425 7 : auto timeout = std::chrono::seconds(timer);
426 7 : u32 count = 0;
427 :
428 : while (1) {
429 27 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
430 2 : if (acceptTimeOutTmp!=0) {
431 0 : HCCL_WARNING("[Get][Connection]topo exchange server get socket timeout! timeout[%d s]", timer);
432 : } else {
433 2 : HCCL_ERROR("[Get][Connection]topo exchange server get socket timeout! timeout[%d s]", timer);
434 : }
435 2 : return HCCL_E_TIMEOUT;
436 : }
437 :
438 25 : HcclSocketStatus status = socket->GetStatus();
439 25 : if (status == HcclSocketStatus::SOCKET_OK) {
440 5 : HCCL_DEBUG("[Accept]socket is established. localIp[%s], remoteIp[%s]",
441 : socket->GetLocalIp().GetReadableIP(), socket->GetRemoteIp().GetReadableIP());
442 5 : return HCCL_SUCCESS;
443 20 : } else if (status == HcclSocketStatus::SOCKET_CONNECTING) {
444 20 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
445 : // 日志过滤, 50次才打印一次
446 20 : if (count % 50 == 0) {
447 2 : HCCL_DEBUG("[Wait][LinkEstablish]socket is connecting ");
448 : }
449 20 : count++;
450 20 : continue;
451 0 : } else if (status == HcclSocketStatus::SOCKET_TIMEOUT) {
452 0 : return HCCL_E_TIMEOUT;
453 : } else {
454 0 : HCCL_ERROR("[Accept]get socket fail");
455 0 : return HCCL_E_TCP_CONNECT;
456 : }
457 20 : }
458 :
459 : return HCCL_SUCCESS;
460 : }
461 :
462 10 : HcclResult HcclSocket::Send(const void *data, u64 size)
463 : {
464 10 : CHK_PTR_NULL(data);
465 10 : CHK_PTR_NULL(fdHandle_);
466 20 : CHK_RET(hrtRaSocketBlockSend(fdHandle_, data, size, [this]() -> bool { return this->GetStopFlag(); }));
467 10 : return HCCL_SUCCESS;
468 : }
469 :
470 9 : HcclResult HcclSocket::Recv(void *recvBuf, u32 recvBufLen, u32 timeout)
471 : {
472 9 : CHK_PTR_NULL(fdHandle_);
473 9 : CHK_PTR_NULL(recvBuf);
474 18 : CHK_RET(hrtRaSocketBlockRecv(fdHandle_, recvBuf, recvBufLen, [this]() -> bool { return this->GetStopFlag(); }, timeout));
475 9 : return HCCL_SUCCESS;
476 : }
477 :
478 0 : HcclResult HcclSocket::Send(const std::string &sendMsg)
479 : {
480 0 : CHK_PTR_NULL(fdHandle_);
481 0 : u32 msgLen = sendMsg.length();
482 0 : u8 buff[MAX_MSG_STR_LEN] = {0};
483 0 : s32 sRet = strcpy_s(reinterpret_cast<char *>(buff), MAX_MSG_STR_LEN, sendMsg.c_str());
484 0 : if (sRet != 0) {
485 0 : HCCL_ERROR("[Send] Block send message length[%u] is illegal", msgLen);
486 0 : return HCCL_E_PARA;
487 : }
488 :
489 : // 与 HcclSocket::Recv(std::string &recvMsg) 对应, 发送的消息长度和接收的消息长度一致, 才能保证正常Recv
490 0 : CHK_RET(hrtRaSocketBlockSend(fdHandle_, buff, MAX_MSG_STR_LEN, [this]() -> bool { return this->GetStopFlag(); }));
491 0 : return HCCL_SUCCESS;
492 : }
493 :
494 0 : HcclResult HcclSocket::Recv(std::string &recvMsg, u32 timeout)
495 : {
496 0 : CHK_PTR_NULL(fdHandle_);
497 0 : recvMsg.clear();
498 0 : u8 recvBuf[MAX_MSG_STR_LEN] = {0};
499 0 : CHK_RET(hrtRaSocketBlockRecv(fdHandle_, reinterpret_cast<void *>(recvBuf), MAX_MSG_STR_LEN,
500 : [this]() -> bool { return this->GetStopFlag(); }, timeout));
501 0 : recvMsg.assign(reinterpret_cast<char *>(recvBuf));
502 0 : return HCCL_SUCCESS;
503 : }
504 :
505 8 : HcclResult HcclSocket::ISend(void *data, u64 size, u64& compSize)
506 : {
507 8 : CHK_PTR_NULL(data);
508 8 : CHK_PTR_NULL(fdHandle_);
509 8 : if (sendStatus_) return HCCL_E_NETWORK;
510 8 : if (size > SOCKET_SEND_MAX_SIZE) {
511 0 : HCCL_ERROR("[ISend]errNo[0x%016llx] ra socket send size is too large, " \
512 : "data[%p], size[%llu Byte]", HCCL_ERROR_CODE(HCCL_E_NETWORK), data, size);
513 0 : return HCCL_E_PARA;
514 : }
515 8 : s32 ret = hrtRaSocketNonBlockSend(fdHandle_, data, size, &compSize);
516 8 : HCCL_DEBUG("[ISend]except size [%llu Byte], actual size [%llu Byte], ret[%d]", size, compSize, ret);
517 :
518 8 : if (ret && ret != SOCK_EAGAIN) {
519 0 : sendStatus_ = ret;
520 0 : HCCL_RUN_WARNING("[ISend]except size [%llu Byte], actual size [%llu Byte], ret[%d]", size, compSize, ret);
521 0 : return HCCL_E_NETWORK;
522 : }
523 8 : return HCCL_SUCCESS; // EAGAIN和success都要返回HCCL_SUCCESS
524 : }
525 :
526 0 : HcclResult HcclSocket::IRecv(void *recvBuf, u32 recvBufLen, u64& compSize)
527 : {
528 0 : CHK_PTR_NULL(fdHandle_);
529 0 : CHK_PTR_NULL(recvBuf);
530 0 : if (recvStatus_) return HCCL_E_NETWORK;
531 0 : s32 ret = hrtRaSocketNonBlockRecv(fdHandle_, recvBuf, recvBufLen, &compSize);
532 0 : HCCL_DEBUG("[IRecv]except size [%u Byte], actual size [%u Byte], ret[%u]", recvBufLen, compSize, ret);
533 :
534 0 : if (ret && ret != SOCK_EAGAIN) {
535 0 : recvStatus_ = ret;
536 0 : HCCL_RUN_INFO("[IRecv]except size [%u Byte], actual size [%u Byte], ret[%u]", recvBufLen, compSize, ret);
537 0 : return HCCL_E_NETWORK;
538 : }
539 0 : return HCCL_SUCCESS; // EAGAIN和success都要返回HCCL_SUCCESS
540 : }
541 :
542 9 : HcclResult HcclSocket::SendAsync(const void *data, u64 size, u64 *sentSize, void **reqHandle)
543 : {
544 9 : CHK_PTR_NULL(fdHandle_);
545 8 : CHK_PTR_NULL(data);
546 7 : CHK_PTR_NULL(sentSize);
547 6 : CHK_PTR_NULL(reqHandle);
548 5 : CHK_PRT_RET((size == 0) || (size > MAX_MSG_STR_LEN),
549 : HCCL_ERROR("[SendAsync]send size[%llu] is 0 or large than %u", size, MAX_MSG_STR_LEN), HCCL_E_PARA);
550 :
551 3 : s32 ret = hrtRaSocketSendAsync(fdHandle_, data, size, sentSize, reqHandle);
552 3 : if (ret == 0) {
553 1 : return HCCL_SUCCESS;
554 : }
555 :
556 2 : if (ret == SOCK_EAGAIN) {
557 1 : return HCCL_E_AGAIN;
558 : }
559 1 : HCCL_ERROR("[SendAsync]RaSocketSendAsync failed, data[%p] size[%llu] ret[%d]", data, size, ret);
560 1 : return HCCL_E_NETWORK;
561 : }
562 :
563 8 : HcclResult HcclSocket::RecvAsync(void *recvBuf, u64 recvBufLen, u64 *receivedSize, void **reqHandle)
564 : {
565 8 : CHK_PTR_NULL(fdHandle_);
566 7 : CHK_PTR_NULL(recvBuf);
567 6 : CHK_PTR_NULL(receivedSize);
568 5 : CHK_PTR_NULL(reqHandle);
569 4 : CHK_PRT_RET(recvBufLen == 0, HCCL_ERROR("[RecvAsync]recvBufLen is 0"), HCCL_E_PARA);
570 :
571 3 : s32 ret = hrtRaSocketRecvAsync(fdHandle_, recvBuf, recvBufLen, receivedSize, reqHandle);
572 3 : if (ret == 0) {
573 1 : return HCCL_SUCCESS;
574 : }
575 :
576 2 : if (ret == SOCK_EAGAIN) {
577 1 : return HCCL_E_AGAIN;
578 : }
579 1 : HCCL_ERROR("[RecvAsync]RaSocketRecvAsync failed, recvBuf[%p] recvBufLen[%llu] ret[%d]", recvBuf, recvBufLen, ret);
580 1 : return HCCL_E_NETWORK;
581 : }
582 :
583 6 : HcclResult HcclSocket::GetAsyncReqResult(void *reqHandle, HcclResult &reqResult)
584 : {
585 6 : CHK_PTR_NULL(reqHandle);
586 5 : s32 asyncReqRet = 0;
587 5 : s32 ret = hrtRaSocketGetAsyncReqResult(reqHandle, &asyncReqRet);
588 5 : if (ret == 0) {
589 2 : reqResult = (asyncReqRet == 0) ? HCCL_SUCCESS : (asyncReqRet == SOCK_EAGAIN ? HCCL_E_AGAIN : HCCL_E_TCP_TRANSFER);
590 2 : return HCCL_SUCCESS;
591 : }
592 :
593 3 : if (ret == OTHERS_EAGAIN) {
594 2 : return HCCL_E_AGAIN;
595 : }
596 1 : HCCL_ERROR("[GetAsyncReqResult]RaSocketGetAsyncReqResult failed, ret[%d]", ret);
597 1 : return HCCL_E_NETWORK;
598 : }
599 :
600 : // static
601 1 : bool HcclSocket::IsSupportAsync()
602 : {
603 1 : bool isSupportRaSocketAsync = false;
604 1 : HcclResult ret = IsSupportHdcAsync(isSupportRaSocketAsync);
605 1 : if (ret != HCCL_SUCCESS) { // 失败时默认不支持异步收发
606 1 : HCCL_WARNING("[IsSupportAsync] IsSupportHdcAsync failed ret[%d]", ret);
607 : }
608 1 : return isSupportRaSocketAsync;
609 : }
610 :
611 0 : std::string HcclSocket::GetTag() const
612 : {
613 0 : return tag_;
614 : }
615 :
616 59 : NicType HcclSocket::GetSocketType() const
617 : {
618 59 : return socketType_;
619 : }
620 :
621 221 : HcclResult HcclSocket::GetNicSocketHandle(std::map<HcclIpAddress, IpSocket> &socketMap,
622 : const HcclIpAddress &ip, SocketHandle &nicSocketHandle)
623 : {
624 221 : if (ip.IsInvalid()) {
625 6 : HCCL_ERROR("[Get][NicHandleInfo]phyId[%u] nicIp is invalid", localDevicePhyId_);
626 6 : return HCCL_E_PARA;
627 : }
628 :
629 215 : auto it = socketMap.find(ip);
630 213 : if (it == socketMap.end()) {
631 164 : HCCL_ERROR("[Get][NicHandleInfo]can not find nic socket handle, ip[%s]",
632 : ip.GetReadableAddress());
633 165 : return HCCL_E_PARA;
634 : } else {
635 50 : if (it->second.nicSocketHandle == nullptr) {
636 0 : HCCL_ERROR("[Get][NicHandleInfo]get nic socket handle failed! phyId[%u] IP addr[%s]",
637 : localDevicePhyId_, ip.GetReadableAddress());
638 0 : return HCCL_E_PARA;
639 : }
640 50 : nicSocketHandle = it->second.nicSocketHandle;
641 : }
642 50 : return HCCL_SUCCESS;
643 : }
644 :
645 253 : HcclResult HcclSocket::GetNicSocketHandle()
646 : {
647 253 : if (nicSocketHandle_ != nullptr) {
648 34 : return HCCL_SUCCESS;
649 : }
650 219 : RaResourceInfo raResourceInfo;
651 218 : CHK_RET(NetworkManager::GetInstance(localDeviceLogicId_).GetRaResourceInfo(raResourceInfo));
652 :
653 219 : std::map<HcclIpAddress, IpSocket> tempSocketMap;
654 :
655 218 : if (socketType_ == NicType::DEVICE_NIC_TYPE) {
656 53 : tempSocketMap = raResourceInfo.nicSocketMap;
657 53 : HCCL_INFO("[Get][NicHandleInfo]phyId[%u], nicSocketMap[%u] localIp[[%s]",
658 : localDeviceLogicId_, tempSocketMap.size(), localIp_.GetReadableAddress());
659 53 : CHK_RET(GetNicSocketHandle(tempSocketMap, localIp_, nicSocketHandle_));
660 165 : } else if (socketType_ == NicType::HOST_NIC_TYPE) {
661 33 : tempSocketMap = raResourceInfo.hostNetSocketMap;
662 33 : HCCL_INFO("[Get][NicHandleInfo]phyId[%u], nicSocketMap[%u] localIp[[%s]",
663 : localDeviceLogicId_, tempSocketMap.size(), localIp_.GetReadableAddress());
664 33 : CHK_RET(GetNicSocketHandle(tempSocketMap, localIp_, nicSocketHandle_));
665 132 : } else if (socketType_ == NicType::VNIC_TYPE) {
666 132 : tempSocketMap = raResourceInfo.vnicSocketMap;
667 132 : HCCL_INFO("[Get][NicHandleInfo]phyId[%u], vnicSocketMap size[%u] localIp[[%s]",
668 : localDeviceLogicId_, tempSocketMap.size(), localIp_.GetReadableAddress());
669 135 : CHK_RET(GetNicSocketHandle(tempSocketMap, localIp_, nicSocketHandle_));
670 : } else {
671 0 : return HCCL_E_INTERNAL;
672 : }
673 :
674 50 : return HCCL_SUCCESS;
675 221 : }
676 :
677 57 : HcclIpAddress HcclSocket::GetRemoteIp() const
678 : {
679 57 : return remoteIp_;
680 : }
681 :
682 44 : u32 HcclSocket::GetRemotePort() const
683 : {
684 44 : return remotePort_;
685 : }
686 :
687 74 : HcclIpAddress HcclSocket::GetLocalIp() const
688 : {
689 74 : return localIp_;
690 : }
691 :
692 65 : u32 HcclSocket::GetLocalPort() const
693 : {
694 65 : return localPort_;
695 : }
696 :
697 39 : HcclSocketRole HcclSocket::GetLocalRole() const
698 : {
699 39 : return localRole_;
700 : }
701 :
702 0 : FdHandle HcclSocket::GetFdHandle() const
703 : {
704 0 : return fdHandle_;
705 : }
706 :
707 10 : void HcclSocket::SetForceClose(bool forceClose)
708 : {
709 10 : forceClose_ = forceClose;
710 10 : }
711 :
712 0 : HcclResult HcclSocket::SetStopFlag(bool value)
713 : {
714 0 : stopFlag_.store(value);
715 0 : return HCCL_SUCCESS;
716 : }
717 :
718 26 : bool HcclSocket::GetStopFlag()
719 : {
720 26 : return stopFlag_.load();
721 : }
722 : }
|
