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 236 : HcclSocket::HcclSocket(
26 : const std::string& tag, HcclNetDevCtx netDevCtx, const HcclIpAddress& remoteIp, u32 remotePort,
27 236 : HcclSocketRole localRole)
28 236 : : tag_(tag),
29 237 : netDevCtx_(netDevCtx),
30 237 : remoteIp_(remoteIp),
31 237 : remotePort_(remotePort),
32 237 : localRole_(localRole),
33 237 : status_(HcclSocketStatus::SOCKET_INIT),
34 237 : fdHandle_(nullptr)
35 237 : {}
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 324 : HcclSocket::~HcclSocket() { DeInit(); }
46 :
47 260 : HcclResult HcclSocket::Init()
48 : {
49 260 : CHK_PTR_NULL(netDevCtx_);
50 260 : socketType_ = (static_cast<hccl::NetDevContext*>(netDevCtx_))->GetNicType();
51 260 : localDevicePhyId_ = (static_cast<hccl::NetDevContext*>(netDevCtx_))->GetPhyId();
52 260 : localDeviceLogicId_ = (static_cast<hccl::NetDevContext*>(netDevCtx_))->GetLogicId();
53 260 : localIp_ = (static_cast<hccl::NetDevContext*>(netDevCtx_))->GetLocalIp();
54 260 : backupIp_ = (static_cast<hccl::NetDevContext*>(netDevCtx_))->GetBackupIp();
55 :
56 : // 默认场景下，只有VNIC使用强制断链，其它场景还是走优雅断链
57 260 : forceClose_ = (socketType_ == NicType::VNIC_TYPE);
58 :
59 260 : 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 2 : errormessage = "The IP address " + std::string(localIp_.GetReadableIP()) + " and port "
172 3 : + std::to_string(localPort_) + " 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 2 : errormessage = "The IP address " + std::string(localIp_.GetReadableIP()) + " and port "
186 3 : + std::to_string(localPort_) + " 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 : strcpy_s(connectInfo.tag, SOCK_CONN_TAG_SIZE, tag_.c_str());
326 :
327 2 : HcclResult ret = hrtRaSocketNonBlockBatchAbort(&connectInfo, 1);
328 2 : if (ret != HCCL_SUCCESS) {
329 0 : HCCL_ERROR("[Abort] call ra socket abort failed. ret[%d]", ret);
330 : } else {
331 2 : HCCL_RUN_INFO("[Abort] call ra socket abort success. tag[%s]", connectInfo.tag);
332 : }
333 : }
334 : } else {
335 : // socket处于其他状态时调用close接口关闭socket
336 323 : if (fdHandle_ == nullptr) {
337 312 : HCCL_WARNING("[Close] socket's fdHandle is null, do not need close.");
338 312 : return;
339 : }
340 11 : SocketCloseInfoT closeInfo = {};
341 11 : closeInfo.socketHandle = nicSocketHandle_;
342 11 : closeInfo.fdHandle = fdHandle_;
343 11 : closeInfo.disuseLinger = static_cast<s32>(forceClose_);
344 11 : HCCL_DEBUG(
345 : "[HcclSocket][Close] socketType[%d] nicHandle[%p] fdHandle[%p] disuseLinger[%d]", socketType_,
346 : nicSocketHandle_, fdHandle_, closeInfo.disuseLinger);
347 :
348 11 : HcclResult sRet = hrtRaSocketBatchClose(&closeInfo, 1, 1);
349 11 : if (sRet != HCCL_SUCCESS) {
350 0 : HCCL_WARNING("[Close] errNo[0x%016llx] server socket batch close ret[%d] failed. not fatal", sRet);
351 : }
352 :
353 11 : fdHandle_ = nullptr;
354 : }
355 :
356 13 : return;
357 : }
358 :
359 : // 此接口用于DFX, 仅在 HcclSocketManager 中判断为连接异常时调用
360 41 : void HcclSocket::SetStatus(HcclSocketStatus status)
361 : {
362 41 : if (status != HcclSocketStatus::SOCKET_TIMEOUT && status != HcclSocketStatus::SOCKET_ERROR) {
363 0 : HCCL_WARNING("[Set]]Status] Only support set SOCKET_TIMEOUT or SOCKET_ERROR, status[%d]", status);
364 0 : return;
365 : }
366 41 : status_ = status;
367 : }
368 :
369 15 : HcclSocketStatus HcclSocket::ConvertRaSocketStatus(int raStatus)
370 : {
371 15 : HcclSocketStatus status = HcclSocketStatus::SOCKET_INIT;
372 : /**< socket status:0 not connected 1:connected 2:connect timeout 3:connecting */
373 15 : switch (raStatus) {
374 0 : case 0: // 0 not connected
375 0 : status = HcclSocketStatus::SOCKET_INIT;
376 0 : break;
377 15 : case 1: // 1:connected
378 15 : status = HcclSocketStatus::SOCKET_OK;
379 15 : break;
380 0 : case 2: // 2:connect timeout
381 0 : status = HcclSocketStatus::SOCKET_TIMEOUT;
382 0 : break;
383 0 : case 3: // 3:connecting
384 0 : status = HcclSocketStatus::SOCKET_CONNECTING;
385 0 : break;
386 0 : default:
387 0 : status = HcclSocketStatus::SOCKET_ERROR;
388 0 : break;
389 : }
390 15 : return status;
391 : }
392 :
393 266 : HcclSocketStatus HcclSocket::GetStatus()
394 : {
395 266 : if (status_ == HcclSocketStatus::SOCKET_OK || status_ == HcclSocketStatus::SOCKET_TIMEOUT
396 265 : || status_ == HcclSocketStatus::SOCKET_ERROR) {
397 77 : HCCL_DEBUG("[Get][Status]socket status is [%d].", status_);
398 78 : return status_;
399 : }
400 :
401 189 : if (GetNicSocketHandle()) {
402 143 : return HcclSocketStatus::SOCKET_INIT;
403 : }
404 :
405 : // 疑问: Listen Socket 会是什么状态？
406 :
407 45 : SocketInfoT socketInfo{};
408 45 : socketInfo.remoteIp.addr = remoteIp_.GetBinaryAddress().addr;
409 45 : socketInfo.remoteIp.addr6 = remoteIp_.GetBinaryAddress().addr6;
410 45 : socketInfo.socketHandle = nicSocketHandle_;
411 45 : s32 ret = strcpy_s(socketInfo.tag, SOCK_CONN_TAG_SIZE, tag_.c_str());
412 45 : CHK_PRT_RET(ret != 0, HCCL_ERROR("[Get][Status]strcpy_s failed. ret[%u]", ret), HcclSocketStatus::SOCKET_ERROR);
413 :
414 45 : u32 connectedNum = 0;
415 45 : s32 sockRet = hrtRaGetSockets(static_cast<u32>(localRole_), &socketInfo, 1, &connectedNum);
416 45 : if ((connectedNum == 0 && sockRet == 0) || (sockRet == SOCK_EAGAIN)) {
417 30 : return HcclSocketStatus::SOCKET_CONNECTING;
418 15 : } else if (sockRet != 0) {
419 0 : HCCL_ERROR("[Get][Status]get rasocket error. role[%u] sockRet[%d] ", localRole_, sockRet);
420 0 : return HcclSocketStatus::SOCKET_ERROR;
421 : } else {
422 15 : if (connectedNum == 1) {
423 15 : status_ = ConvertRaSocketStatus(socketInfo.status);
424 15 : fdHandle_ = socketInfo.fdHandle;
425 15 : HCCL_INFO("[Get][Status]status_[%u] ", status_);
426 15 : return status_;
427 : } else {
428 0 : HCCL_ERROR("[Get][Status]total Sockets[%u], more than needed num[1]!", connectedNum);
429 0 : return HcclSocketStatus::SOCKET_ERROR;
430 : }
431 : }
432 : }
433 :
434 7 : HcclResult HcclSocket::Accept(const std::string& tag, std::shared_ptr<HcclSocket>& socket, u32 acceptTimeOut)
435 : {
436 7 : if (listened_ == false) {
437 0 : HCCL_ERROR("[Accept]socket no listen, can not accepted.");
438 0 : return HCCL_E_PARA;
439 : }
440 :
441 7 : EXCEPTION_CATCH(
442 : (socket = std::make_shared<HcclSocket>(tag, netDevCtx_, remoteIp_, 0, HcclSocketRole::SOCKET_ROLE_SERVER)),
443 : return HCCL_E_PTR);
444 7 : CHK_SMART_PTR_NULL(socket);
445 7 : CHK_RET(socket->Init());
446 :
447 7 : HCCL_INFO(
448 : "[Accept]localIp[%s], remoteIp[%s], socketHandle[%p], tag[%s]", localIp_.GetReadableAddress(),
449 : remoteIp_.GetReadableAddress(), nicSocketHandle_, tag.c_str());
450 :
451 7 : s32 acceptTimeOutTmp = static_cast<s32>(acceptTimeOut);
452 7 : s32 timer = (acceptTimeOutTmp > 0 && acceptTimeOutTmp < GetExternalInputHcclLinkTimeOut()) ?
453 0 : acceptTimeOutTmp :
454 7 : GetExternalInputHcclLinkTimeOut();
455 :
456 7 : auto startTime = std::chrono::steady_clock::now();
457 7 : auto timeout = std::chrono::seconds(timer);
458 7 : u32 count = 0;
459 :
460 : while (1) {
461 27 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
462 2 : if (acceptTimeOutTmp != 0) {
463 0 : HCCL_WARNING("[Get][Connection]topo exchange server get socket timeout! timeout[%d s]", timer);
464 : } else {
465 2 : HCCL_ERROR("[Get][Connection]topo exchange server get socket timeout! timeout[%d s]", timer);
466 : }
467 2 : return HCCL_E_TIMEOUT;
468 : }
469 :
470 25 : HcclSocketStatus status = socket->GetStatus();
471 25 : if (status == HcclSocketStatus::SOCKET_OK) {
472 5 : HCCL_DEBUG(
473 : "[Accept]socket is established. localIp[%s], remoteIp[%s]", socket->GetLocalIp().GetReadableIP(),
474 : socket->GetRemoteIp().GetReadableIP());
475 5 : return HCCL_SUCCESS;
476 20 : } else if (status == HcclSocketStatus::SOCKET_CONNECTING) {
477 20 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
478 : // 日志过滤, 50次才打印一次
479 20 : if (count % 50 == 0) {
480 2 : HCCL_DEBUG("[Wait][LinkEstablish]socket is connecting ");
481 : }
482 20 : count++;
483 20 : continue;
484 0 : } else if (status == HcclSocketStatus::SOCKET_TIMEOUT) {
485 0 : return HCCL_E_TIMEOUT;
486 : } else {
487 0 : HCCL_ERROR("[Accept]get socket fail");
488 0 : return HCCL_E_TCP_CONNECT;
489 : }
490 20 : }
491 :
492 : return HCCL_SUCCESS;
493 : }
494 :
495 10 : HcclResult HcclSocket::Send(const void* data, u64 size)
496 : {
497 10 : CHK_PTR_NULL(data);
498 10 : CHK_PTR_NULL(fdHandle_);
499 20 : CHK_RET(hrtRaSocketBlockSend(fdHandle_, data, size, [this]() -> bool {
500 : return this->GetStopFlag();
501 : }));
502 10 : return HCCL_SUCCESS;
503 : }
504 :
505 9 : HcclResult HcclSocket::Recv(void* recvBuf, u32 recvBufLen, u32 timeout)
506 : {
507 9 : CHK_PTR_NULL(fdHandle_);
508 9 : CHK_PTR_NULL(recvBuf);
509 18 : CHK_RET(hrtRaSocketBlockRecv(
510 : fdHandle_, recvBuf, recvBufLen,
511 : [this]() -> bool {
512 : return this->GetStopFlag();
513 : },
514 : timeout));
515 9 : return HCCL_SUCCESS;
516 : }
517 :
518 0 : HcclResult HcclSocket::Send(const std::string& sendMsg)
519 : {
520 0 : CHK_PTR_NULL(fdHandle_);
521 0 : u32 msgLen = sendMsg.length();
522 0 : u8 buff[MAX_MSG_STR_LEN] = {0};
523 0 : s32 sRet = strcpy_s(reinterpret_cast<char*>(buff), MAX_MSG_STR_LEN, sendMsg.c_str());
524 0 : if (sRet != 0) {
525 0 : HCCL_ERROR("[Send] Block send message length[%u] is illegal", msgLen);
526 0 : return HCCL_E_PARA;
527 : }
528 :
529 : // 与 HcclSocket::Recv(std::string &recvMsg) 对应, 发送的消息长度和接收的消息长度一致, 才能保证正常Recv
530 0 : CHK_RET(hrtRaSocketBlockSend(fdHandle_, buff, MAX_MSG_STR_LEN, [this]() -> bool {
531 : return this->GetStopFlag();
532 : }));
533 0 : return HCCL_SUCCESS;
534 : }
535 :
536 0 : HcclResult HcclSocket::Recv(std::string& recvMsg, u32 timeout)
537 : {
538 0 : CHK_PTR_NULL(fdHandle_);
539 0 : recvMsg.clear();
540 0 : u8 recvBuf[MAX_MSG_STR_LEN] = {0};
541 0 : CHK_RET(hrtRaSocketBlockRecv(
542 : fdHandle_, reinterpret_cast<void*>(recvBuf), MAX_MSG_STR_LEN,
543 : [this]() -> bool {
544 : return this->GetStopFlag();
545 : },
546 : timeout));
547 0 : recvMsg.assign(reinterpret_cast<char*>(recvBuf));
548 0 : return HCCL_SUCCESS;
549 : }
550 :
551 8 : HcclResult HcclSocket::ISend(void* data, u64 size, u64& compSize)
552 : {
553 8 : CHK_PTR_NULL(data);
554 8 : CHK_PTR_NULL(fdHandle_);
555 8 : if (sendStatus_)
556 0 : return HCCL_E_NETWORK;
557 8 : if (size > SOCKET_SEND_MAX_SIZE) {
558 0 : HCCL_ERROR(
559 : "[ISend]errNo[0x%016llx] ra socket send size is too large, "
560 : "data[%p], size[%llu Byte]",
561 : HCCL_ERROR_CODE(HCCL_E_NETWORK), data, size);
562 0 : return HCCL_E_PARA;
563 : }
564 8 : s32 ret = hrtRaSocketNonBlockSend(fdHandle_, data, size, &compSize);
565 8 : HCCL_DEBUG("[ISend]except size [%llu Byte], actual size [%llu Byte], ret[%d]", size, compSize, ret);
566 :
567 8 : if (ret && ret != SOCK_EAGAIN) {
568 0 : sendStatus_ = ret;
569 0 : HCCL_RUN_WARNING("[ISend]except size [%llu Byte], actual size [%llu Byte], ret[%d]", size, compSize, ret);
570 0 : return HCCL_E_NETWORK;
571 : }
572 8 : return HCCL_SUCCESS; // EAGAIN和success都要返回HCCL_SUCCESS
573 : }
574 :
575 0 : HcclResult HcclSocket::IRecv(void* recvBuf, u32 recvBufLen, u64& compSize)
576 : {
577 0 : CHK_PTR_NULL(fdHandle_);
578 0 : CHK_PTR_NULL(recvBuf);
579 0 : if (recvStatus_)
580 0 : return HCCL_E_NETWORK;
581 0 : s32 ret = hrtRaSocketNonBlockRecv(fdHandle_, recvBuf, recvBufLen, &compSize);
582 0 : HCCL_DEBUG("[IRecv]except size [%u Byte], actual size [%u Byte], ret[%u]", recvBufLen, compSize, ret);
583 :
584 0 : if (ret && ret != SOCK_EAGAIN) {
585 0 : recvStatus_ = ret;
586 0 : HCCL_RUN_INFO("[IRecv]except size [%u Byte], actual size [%u Byte], ret[%u]", recvBufLen, compSize, ret);
587 0 : return HCCL_E_NETWORK;
588 : }
589 0 : return HCCL_SUCCESS; // EAGAIN和success都要返回HCCL_SUCCESS
590 : }
591 :
592 9 : HcclResult HcclSocket::SendAsync(const void* data, u64 size, u64* sentSize, void** reqHandle)
593 : {
594 9 : CHK_PTR_NULL(fdHandle_);
595 8 : CHK_PTR_NULL(data);
596 7 : CHK_PTR_NULL(sentSize);
597 6 : CHK_PTR_NULL(reqHandle);
598 5 : CHK_PRT_RET(
599 : (size == 0) || (size > MAX_MSG_STR_LEN),
600 : HCCL_ERROR("[SendAsync]send size[%llu] is 0 or large than %u", size, MAX_MSG_STR_LEN), HCCL_E_PARA);
601 :
602 3 : s32 ret = hrtRaSocketSendAsync(fdHandle_, data, size, sentSize, reqHandle);
603 3 : if (ret == 0) {
604 1 : return HCCL_SUCCESS;
605 : }
606 :
607 2 : if (ret == SOCK_EAGAIN) {
608 1 : return HCCL_E_AGAIN;
609 : }
610 1 : HCCL_ERROR("[SendAsync]RaSocketSendAsync failed, data[%p] size[%llu] ret[%d]", data, size, ret);
611 1 : return HCCL_E_NETWORK;
612 : }
613 :
614 8 : HcclResult HcclSocket::RecvAsync(void* recvBuf, u64 recvBufLen, u64* receivedSize, void** reqHandle)
615 : {
616 8 : CHK_PTR_NULL(fdHandle_);
617 7 : CHK_PTR_NULL(recvBuf);
618 6 : CHK_PTR_NULL(receivedSize);
619 5 : CHK_PTR_NULL(reqHandle);
620 4 : CHK_PRT_RET(recvBufLen == 0, HCCL_ERROR("[RecvAsync]recvBufLen is 0"), HCCL_E_PARA);
621 :
622 3 : s32 ret = hrtRaSocketRecvAsync(fdHandle_, recvBuf, recvBufLen, receivedSize, reqHandle);
623 3 : if (ret == 0) {
624 1 : return HCCL_SUCCESS;
625 : }
626 :
627 2 : if (ret == SOCK_EAGAIN) {
628 1 : return HCCL_E_AGAIN;
629 : }
630 1 : HCCL_ERROR("[RecvAsync]RaSocketRecvAsync failed, recvBuf[%p] recvBufLen[%llu] ret[%d]", recvBuf, recvBufLen, ret);
631 1 : return HCCL_E_NETWORK;
632 : }
633 :
634 6 : HcclResult HcclSocket::GetAsyncReqResult(void* reqHandle, HcclResult& reqResult)
635 : {
636 6 : CHK_PTR_NULL(reqHandle);
637 5 : s32 asyncReqRet = 0;
638 5 : s32 ret = hrtRaSocketGetAsyncReqResult(reqHandle, &asyncReqRet);
639 5 : if (ret == 0) {
640 : reqResult
641 2 : = (asyncReqRet == 0) ? HCCL_SUCCESS : (asyncReqRet == SOCK_EAGAIN ? HCCL_E_AGAIN : HCCL_E_TCP_TRANSFER);
642 2 : return HCCL_SUCCESS;
643 : }
644 :
645 3 : if (ret == OTHERS_EAGAIN) {
646 2 : return HCCL_E_AGAIN;
647 : }
648 1 : HCCL_ERROR("[GetAsyncReqResult]RaSocketGetAsyncReqResult failed, ret[%d]", ret);
649 1 : return HCCL_E_NETWORK;
650 : }
651 :
652 : // static
653 1 : bool HcclSocket::IsSupportAsync()
654 : {
655 1 : bool isSupportRaSocketAsync = false;
656 1 : HcclResult ret = IsSupportHdcAsync(isSupportRaSocketAsync);
657 1 : if (ret != HCCL_SUCCESS) { // 失败时默认不支持异步收发
658 1 : HCCL_WARNING("[IsSupportAsync] IsSupportHdcAsync failed ret[%d]", ret);
659 : }
660 1 : return isSupportRaSocketAsync;
661 : }
662 :
663 0 : std::string HcclSocket::GetTag() const { return tag_; }
664 :
665 59 : NicType HcclSocket::GetSocketType() const { return socketType_; }
666 :
667 239 : HcclResult HcclSocket::GetNicSocketHandle(
668 : std::map<HcclIpAddress, IpSocket>& socketMap, const HcclIpAddress& ip, SocketHandle& nicSocketHandle)
669 : {
670 239 : if (ip.IsInvalid()) {
671 6 : HCCL_ERROR("[Get][NicHandleInfo]phyId[%u] nicIp is invalid", localDevicePhyId_);
672 6 : return HCCL_E_PARA;
673 : }
674 :
675 233 : auto it = socketMap.find(ip);
676 233 : if (it == socketMap.end()) {
677 183 : HCCL_ERROR("[Get][NicHandleInfo]can not find nic socket handle, ip[%s]", ip.GetReadableAddress());
678 183 : return HCCL_E_PARA;
679 : } else {
680 50 : if (it->second.nicSocketHandle == nullptr) {
681 0 : HCCL_ERROR(
682 : "[Get][NicHandleInfo]get nic socket handle failed! phyId[%u] IP addr[%s]", localDevicePhyId_,
683 : ip.GetReadableAddress());
684 0 : return HCCL_E_PARA;
685 : }
686 50 : nicSocketHandle = it->second.nicSocketHandle;
687 : }
688 50 : return HCCL_SUCCESS;
689 : }
690 :
691 272 : HcclResult HcclSocket::GetNicSocketHandle()
692 : {
693 272 : if (nicSocketHandle_ != nullptr) {
694 34 : return HCCL_SUCCESS;
695 : }
696 238 : RaResourceInfo raResourceInfo;
697 238 : CHK_RET(NetworkManager::GetInstance(localDeviceLogicId_).GetRaResourceInfo(raResourceInfo));
698 :
699 238 : std::map<HcclIpAddress, IpSocket> tempSocketMap;
700 :
701 239 : if (socketType_ == NicType::DEVICE_NIC_TYPE) {
702 72 : tempSocketMap = raResourceInfo.nicSocketMap;
703 71 : HCCL_INFO(
704 : "[Get][NicHandleInfo]phyId[%u], nicSocketMap[%u] localIp[[%s]", localDeviceLogicId_, tempSocketMap.size(),
705 : localIp_.GetReadableAddress());
706 71 : CHK_RET(GetNicSocketHandle(tempSocketMap, localIp_, nicSocketHandle_));
707 167 : } else if (socketType_ == NicType::HOST_NIC_TYPE) {
708 33 : tempSocketMap = raResourceInfo.hostNetSocketMap;
709 33 : HCCL_INFO(
710 : "[Get][NicHandleInfo]phyId[%u], nicSocketMap[%u] localIp[[%s]", localDeviceLogicId_, tempSocketMap.size(),
711 : localIp_.GetReadableAddress());
712 33 : CHK_RET(GetNicSocketHandle(tempSocketMap, localIp_, nicSocketHandle_));
713 134 : } else if (socketType_ == NicType::VNIC_TYPE) {
714 135 : tempSocketMap = raResourceInfo.vnicSocketMap;
715 134 : HCCL_INFO(
716 : "[Get][NicHandleInfo]phyId[%u], vnicSocketMap size[%u] localIp[[%s]", localDeviceLogicId_,
717 : tempSocketMap.size(), localIp_.GetReadableAddress());
718 135 : CHK_RET(GetNicSocketHandle(tempSocketMap, localIp_, nicSocketHandle_));
719 : } else {
720 0 : return HCCL_E_INTERNAL;
721 : }
722 :
723 50 : return HCCL_SUCCESS;
724 238 : }
725 :
726 57 : HcclIpAddress HcclSocket::GetRemoteIp() const { return remoteIp_; }
727 :
728 44 : u32 HcclSocket::GetRemotePort() const { return remotePort_; }
729 :
730 74 : HcclIpAddress HcclSocket::GetLocalIp() const { return localIp_; }
731 :
732 65 : u32 HcclSocket::GetLocalPort() const { return localPort_; }
733 :
734 39 : HcclSocketRole HcclSocket::GetLocalRole() const { return localRole_; }
735 :
736 0 : FdHandle HcclSocket::GetFdHandle() const { return fdHandle_; }
737 :
738 10 : void HcclSocket::SetForceClose(bool forceClose) { forceClose_ = forceClose; }
739 :
740 0 : HcclResult HcclSocket::SetStopFlag(bool value)
741 : {
742 0 : stopFlag_.store(value);
743 0 : return HCCL_SUCCESS;
744 : }
745 :
746 26 : bool HcclSocket::GetStopFlag() { return stopFlag_.load(); }
747 : } // namespace hccl
|
