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