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 "socket.h"
12 :
13 : #include "sal.h"
14 : #include "socket_exception.h"
15 : #include "network_api_exception.h"
16 :
17 : namespace Hccl {
18 :
19 : constexpr u32 MAX_TRANSFER_SIZE = 20 * 1024 * 1024;
20 : constexpr u32 MAX_LOG_TIMEOUT_MS = 30000;
21 : constexpr u32 ONE_MILLISECOND_OF_USLEEP = 1000;
22 : constexpr u32 AUTO_LISTEN_PORT = 0;
23 :
24 788 : Socket::~Socket()
25 : {
26 342 : if (!isDestroyed) {
27 293 : DECTOR_TRY_CATCH("Socket", this->Destroy());
28 : }
29 446 : }
30 :
31 11 : void Socket::Listen()
32 : {
33 33 : HCCL_INFO("[Socket::%s] listen start, listenPort[%u]", __func__, listenPort);
34 11 : HrtNetworkMode netMode = nicType == NicType::HOST_NIC_TYPE ? HrtNetworkMode::PEER : HrtNetworkMode::HDC;
35 11 : RaSocketListenParam param(socketHandle, listenPort, localIp);
36 11 : HrtRaSocketListenOneStart(param, netMode);
37 11 : isListening = true;
38 11 : socketStatus = SocketStatus::LISTENING;
39 11 : }
40 :
41 2 : bool Socket::Listen(u32 &port)
42 : {
43 6 : HCCL_INFO("[Socket::%s] trying to listen on port[%u]", __func__, port);
44 2 : HrtNetworkMode netMode = nicType == NicType::HOST_NIC_TYPE ? HrtNetworkMode::PEER : HrtNetworkMode::HDC;
45 2 : RaSocketListenParam param(socketHandle, port, localIp);
46 2 : bool ret = HrtRaSocketTryListenOneStart(param, netMode);
47 2 : CHK_PRT_RET(!ret, HCCL_WARNING("[Socket::%s] socket[%s] listen unsuccessful, port[%u] is in use",
48 : __func__, Describe().c_str(), port), ret);
49 :
50 2 : port = port == AUTO_LISTEN_PORT ? param.port : port;
51 2 : listenPort = port;
52 2 : isListening = true;
53 2 : socketStatus = SocketStatus::LISTENING;
54 :
55 6 : HCCL_INFO("[Socket::%s] socket[%s] listen success.", __func__, Describe().c_str());
56 2 : return true;
57 : }
58 :
59 5 : void Socket::Connect()
60 : {
61 15 : HCCL_INFO("socket role_ is %u, %s", static_cast<u32>(role), role.Describe().c_str());
62 5 : if (role == SocketRole::SERVER || socketStatus == SocketStatus::OK) {
63 5 : return;
64 : }
65 :
66 0 : RaSocketConnectParam param(socketHandle, remoteIp, listenPort, tag);
67 0 : HrtRaSocketConnectOne(param);
68 0 : HCCL_INFO("conn.tag %s", tag.c_str());
69 :
70 0 : socketStatus = SocketStatus::CONNECTING;
71 0 : }
72 :
73 0 : void Socket::PrintErrorSocketInfo()
74 : {
75 0 : HCCL_ERROR("Socket::GetStatus failed.");
76 0 : HCCL_ERROR("Please check if env HCCL_SOCKET_IFNAME is set correctly, "
77 : "which can be verified by checking localIp and remoteIp in the socket info as follows.");
78 0 : HCCL_ERROR("%s", Describe().c_str());
79 0 : }
80 :
81 4 : SocketStatus Socket::GetStatus(u32 timeout)
82 : {
83 4 : if (socketStatus == SocketStatus::OK) {
84 0 : HCCL_INFO("socketinfo.tag=%s status is OK, role=%s", tag.c_str(), role.Describe().c_str());
85 0 : return SocketStatus::OK;
86 : }
87 :
88 4 : RaSocketGetParam param(socketHandle, remoteIp, tag, fdHandle);
89 4 : RaSocketFdHandleParam result(nullptr, 0);
90 4 : TRY_CATCH_PROCESS_THROW(
91 : NetworkApiException,
92 : result = HrtRaBlockGetOneSocket(static_cast<u32>(role), param, timeout),
93 : "Socket::GetStatus failed",
94 : PrintErrorSocketInfo()
95 : );
96 :
97 4 : fdHandle = result.fdHandle;
98 :
99 : // socket status:0 not connected 1:connected 2:connect timeout 3:connecting
100 4 : if (result.status == SOCKET_CONNECTED) {
101 1 : socketStatus = SocketStatus::OK;
102 1 : isConnected = true;
103 3 : } else if (result.status == SOCKET_CONNECT_TIMEOUT) {
104 1 : socketStatus = SocketStatus::TIMEOUT;
105 2 : } else if (result.status == SOCKET_CONNECTING) {
106 1 : socketStatus = SocketStatus::CONNECTING;
107 : } else {
108 1 : socketStatus = SocketStatus::INIT;
109 : }
110 12 : HCCL_INFO("socketinfo.tag=%s status=%s, role=%s", tag.c_str(), socketStatus.Describe().c_str(),
111 : role.Describe().c_str());
112 4 : return socketStatus;
113 4 : }
114 :
115 6 : bool Socket::Send(const void *sendBuf, u32 size) const
116 : {
117 6 : HrtRaSocketBlockSend(fdHandle, sendBuf, size);
118 6 : return true;
119 : }
120 :
121 1 : bool Socket::Recv(void *recvBuf, u32 size) const
122 : {
123 1 : HrtRaSocketBlockRecv(fdHandle, recvBuf, size);
124 1 : return true;
125 : }
126 :
127 5 : bool Socket::ISend(void *data, u64 size, u64& compSize) const
128 : {
129 5 : return HrtRaSocketNonBlockSend(fdHandle, data, size, &compSize);
130 : }
131 :
132 0 : HcclResult Socket::ISendWithHeart(void *data, u64 size, u64& compSize) const
133 : {
134 0 : return HrtRaSocketNonBlockSendHeart(fdHandle, data, size, &compSize);
135 : }
136 :
137 0 : HcclResult Socket::IRecvWithHeart(void *data, u64 size, u64& compSize) const
138 : {
139 0 : return HrtRaSocketNonBlockRecvHeart(fdHandle, data, size, &compSize);
140 : }
141 :
142 342 : void Socket::Destroy()
143 : {
144 342 : isDestroyed = true;
145 342 : EXCEPTION_CATCH(StopListen(), return);
146 342 : EXCEPTION_CATCH(Close(), return);
147 : }
148 :
149 378 : void Socket::Close()
150 : {
151 378 : if (isConnected) {
152 15 : RaSocketCloseParam param(socketHandle, fdHandle);
153 15 : HrtRaSocketCloseOne(param);
154 15 : isConnected = false;
155 : }
156 378 : }
157 :
158 347 : void Socket::StopListen()
159 : {
160 347 : if (isListening) {
161 13 : RaSocketListenParam param(socketHandle, listenPort, localIp);
162 13 : HrtRaSocketListenOneStop(param);
163 13 : isListening = false;
164 : }
165 347 : }
166 :
167 : // 抑制日志刷屏，同一类型日志超时前只打印一次
168 0 : inline bool CheckLogTime(std::chrono::steady_clock::time_point &lastTime)
169 : {
170 0 : auto nowTime = std::chrono::steady_clock::now();
171 0 : if (nowTime - lastTime <= std::chrono::milliseconds(MAX_LOG_TIMEOUT_MS)) {
172 0 : return false;
173 : }
174 :
175 0 : lastTime = nowTime;
176 0 : return true;
177 : }
178 :
179 3 : inline void HandleSocketEAgain(RequestHandle lastReqHandle, std::chrono::steady_clock::time_point lastLogTime)
180 : {
181 3 : if (CheckLogTime(lastLogTime)) {
182 0 : HCCL_WARNING("[Socket][%s] reqhandle[%llu] get request result [SOCK_EAGAIN], sleep 1ms and retry.",
183 : __func__, lastReqHandle);
184 : }
185 :
186 3 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
187 3 : }
188 :
189 7 : bool Socket::CheckStartRequestResult()
190 : {
191 7 : if (reqHandle == 0) {
192 0 : return true;
193 : }
194 :
195 7 : RequestHandle lastReqHandle = reqHandle;
196 7 : ReqHandleResult result = HrtRaGetAsyncReqResult(reqHandle);
197 7 : if (result == ReqHandleResult::NOT_COMPLETED) {
198 2 : if (CheckLogTime(lastLogTime)) {
199 3 : HCCL_INFO("[Socket][%s] connect is not completed, reqHandle[%llu], [%s].",
200 : __func__, lastReqHandle, this->Describe().c_str());
201 : }
202 :
203 2 : return false;
204 : }
205 :
206 : // COMPLETED 表示调用接口成功，可以更新数据信息
207 : // SOCK_E_AGAIN 表示接口调用失败，需要重新调用接口
208 : // 其余结果为异常场景，抛出异常
209 5 : if (result == ReqHandleResult::COMPLETED) {
210 2 : lastLogTime = {};
211 2 : return true;
212 3 : } else if (result == ReqHandleResult::SOCK_E_AGAIN) {
213 1 : HandleSocketEAgain(lastReqHandle, lastLogTime);
214 : } else {
215 2 : THROW<SocketException>(
216 8 : StringFormat("[Socket][%s] failed, request handle[%llu] result[%s] is unexpected, [%s].",
217 10 : __func__, lastReqHandle, result.Describe().c_str(), this->Describe().c_str()));
218 : }
219 :
220 1 : if (socketStatus == SocketStatus::CONNECT_STARTING) {
221 1 : ConnectAsync();
222 0 : } else if (socketStatus == SocketStatus::LISTEN_STARTING) {
223 0 : ListenAsync();
224 : } else {
225 0 : THROW<SocketException>(
226 0 : StringFormat("[Socket][%s] failed, socket status[%s] is not expected, [%s].",
227 0 : __func__, socketStatus.Describe().c_str(), this->Describe().c_str()));
228 : }
229 :
230 1 : return false;
231 : }
232 :
233 10 : bool Socket::CheckSendRequestResult()
234 : {
235 10 : if (reqHandle == 0) {
236 0 : return true;
237 : }
238 :
239 10 : RequestHandle lastReqHandle = reqHandle;
240 10 : ReqHandleResult result = HrtRaGetAsyncReqResult(reqHandle);
241 10 : if (result == ReqHandleResult::NOT_COMPLETED) {
242 1 : if (CheckLogTime(lastLogTime)) {
243 3 : HCCL_INFO("[Socket][%s] reqHandle[%llu] send is not completed, [%s].",
244 : __func__, lastReqHandle, this->Describe().c_str());
245 : }
246 :
247 1 : return false;
248 : }
249 :
250 9 : if (sendSize > sendLeftSize) {
251 0 : THROW<SocketException>(StringFormat("[Socket][%s] prev send request handle[%llu] failed, "
252 : "send size[%u] is greater than expected[%u], [%s].", __func__,
253 0 : lastReqHandle, sendSize, sendLeftSize, this->Describe().c_str()));
254 : }
255 :
256 : // COMPLETED 表示调用接口成功，可以更新数据信息
257 : // SOCK_E_AGAIN 表示接口调用失败，需要重新调用接口
258 : // 其余结果为异常场景，抛出异常
259 9 : if (result == ReqHandleResult::COMPLETED) {
260 8 : totalSendSize += sendSize;
261 8 : sendLeftSize -= sendSize;
262 1 : } else if (result == ReqHandleResult::SOCK_E_AGAIN) {
263 1 : HandleSocketEAgain(lastReqHandle, lastLogTime);
264 : } else {
265 0 : THROW<SocketException>(
266 0 : StringFormat("[Socket][%s] failed, request handle[%llu] result[%s] is unexpected, [%s].",
267 0 : __func__, lastReqHandle, result.Describe().c_str(), this->Describe().c_str()));
268 : }
269 :
270 : // 如果仍有数据未处理，则继续调用接口
271 9 : if (sendLeftSize != 0) {
272 1 : sendSize = 0;
273 2 : reqHandle = HrtRaSocketSendAsync(fdHandle,
274 1 : reinterpret_cast<void *>(reinterpret_cast<uintptr_t>(sendDataBuff) + totalSendSize),
275 1 : sendLeftSize, sendSize);
276 :
277 1 : if (CheckLogTime(lastLogTime)) {
278 0 : HCCL_INFO("[Socket][%s] reqHandle[%llu] need to retry send, start to send left size[%u], [%s].",
279 : __func__, reqHandle, sendLeftSize, this->Describe().c_str());
280 : }
281 :
282 1 : return false;
283 : }
284 :
285 8 : lastLogTime = {};
286 24 : HCCL_INFO("[Socket][%s] pre send request[%llu] is completed, total send size[%u], [%s]",
287 : __func__, lastReqHandle, totalSendSize, this->Describe().c_str());
288 8 : return true;
289 : }
290 :
291 9 : bool Socket::CheckRecvRequestResult()
292 : {
293 9 : if (reqHandle == 0) {
294 0 : return true;
295 : }
296 :
297 9 : RequestHandle lastReqHandle = reqHandle;
298 9 : ReqHandleResult result = HrtRaGetAsyncReqResult(reqHandle);
299 9 : if (result == ReqHandleResult::NOT_COMPLETED) {
300 1 : if (CheckLogTime(lastLogTime)) {
301 3 : HCCL_INFO("[Socket][%s] reqHandle[%llu] recv is not completed, [%s].",
302 : __func__, lastReqHandle, this->Describe().c_str());
303 : }
304 :
305 1 : return false;
306 : }
307 :
308 8 : if (recvSize > recvLeftSize) {
309 0 : THROW<SocketException>(StringFormat("[Socket][%s] prev recv request handle[%llu] failed, "
310 : "recv size[%u] is greater than expected[%u], [%s].", __func__,
311 0 : lastReqHandle, recvSize, recvLeftSize, this->Describe().c_str()));
312 : }
313 :
314 : // COMPLETED 表示调用接口成功，可以更新数据信息
315 : // SOCK_E_AGAIN 表示接口调用失败，需要重新调用接口
316 : // 其余结果为异常场景，抛出异常
317 8 : if (result == ReqHandleResult::COMPLETED) {
318 7 : totalRecvSize += recvSize;
319 7 : recvLeftSize -= recvSize;
320 1 : } else if (result == ReqHandleResult::SOCK_E_AGAIN) {
321 1 : HandleSocketEAgain(lastReqHandle, lastLogTime);
322 : } else {
323 0 : THROW<SocketException>(
324 0 : StringFormat("[Socket][%s] failed, request handle[%llu] result[%s] is unexpected, [%s].",
325 0 : __func__, lastReqHandle, result.Describe().c_str(), this->Describe().c_str()));
326 : }
327 :
328 : // 如果仍有数据未处理，则继续调用接口
329 8 : if (recvLeftSize != 0) {
330 1 : recvSize = 0;
331 2 : reqHandle = HrtRaSocketRecvAsync(fdHandle,
332 1 : reinterpret_cast<void *>(reinterpret_cast<uintptr_t>(recvDataBuff) + totalRecvSize),
333 1 : recvLeftSize, recvSize);
334 :
335 1 : if (CheckLogTime(lastLogTime)) {
336 0 : HCCL_INFO("[Socket][%s] reqHandle[%llu] need to retry recv, start to recv left size[%u], [%s].",
337 : __func__, reqHandle, recvLeftSize, this->Describe().c_str());
338 : }
339 :
340 1 : return false;
341 : }
342 :
343 7 : lastLogTime = {};
344 21 : HCCL_INFO("[Socket][%s] pre recv request[%llu] is completed, total recv size[%u], [%s].",
345 : __func__, lastReqHandle, totalRecvSize, this->Describe().c_str());
346 7 : return true;
347 : }
348 :
349 46 : SocketStatus Socket::GetAsyncStatus()
350 : {
351 46 : switch (socketStatus) {
352 8 : case SocketStatus::OK:
353 : case SocketStatus::TIMEOUT:
354 : case SocketStatus::LISTENING:
355 8 : break;
356 1 : case SocketStatus::LISTEN_STARTING: {
357 1 : if (CheckStartRequestResult()) {
358 0 : isListening = true;
359 0 : socketStatus = SocketStatus::LISTENING;
360 : }
361 0 : break;
362 : }
363 10 : case SocketStatus::SENDING: {
364 10 : if (CheckSendRequestResult()) {
365 8 : socketStatus = SocketStatus::OK;
366 : }
367 10 : break;
368 : }
369 9 : case SocketStatus::RECVING: {
370 9 : if (CheckRecvRequestResult()) {
371 7 : socketStatus = SocketStatus::OK;
372 : }
373 :
374 9 : break;
375 : }
376 6 : case SocketStatus::CONNECT_STARTING: {
377 6 : if (CheckStartRequestResult()) {
378 2 : GetOneSocket();
379 : }
380 5 : break;
381 : }
382 12 : case SocketStatus::INIT:
383 : case SocketStatus::CONNECTING:
384 : default:
385 12 : GetOneSocket();
386 : }
387 :
388 44 : return socketStatus;
389 : }
390 :
391 14 : void Socket::GetOneSocket()
392 : {
393 14 : RaSocketGetParam param(socketHandle, remoteIp, tag, fdHandle);
394 :
395 14 : RaSocketFdHandleParam fdHandleParam(nullptr, 0);
396 14 : EXCEPTION_CATCH(fdHandleParam = RaGetOneSocket(static_cast<u32>(role), param), return);
397 : // socket status:0 not connected 1:connected 2:connect timeout 3:connecting
398 14 : if (fdHandleParam.status == SOCKET_CONNECTED) {
399 : // sockete 准备好时，可以读取信息
400 14 : fdHandle = fdHandleParam.fdHandle;
401 14 : socketStatus = SocketStatus::OK;
402 14 : isConnected = true;
403 14 : lastLogTime = {};
404 0 : } else if (fdHandleParam.status == SOCKET_CONNECT_TIMEOUT) {
405 0 : socketStatus = SocketStatus::TIMEOUT;
406 0 : } else if (fdHandleParam.status == SOCKET_CONNECTING) {
407 0 : if (CheckLogTime(lastLogTime)) {
408 0 : HCCL_INFO("[Socket][%s] socket is connecting, [%s]", __func__, this->Describe().c_str());
409 : }
410 :
411 0 : socketStatus = SocketStatus::CONNECTING;
412 : } else {
413 0 : socketStatus = SocketStatus::INIT;
414 : }
415 14 : }
416 :
417 3 : void Socket::ListenAsync()
418 : {
419 3 : listenInfo_ = std::make_unique<SocketListenInfoT>();
420 3 : listenInfo_->socketHandle = socketHandle;
421 3 : listenInfo_->port = listenPort;
422 3 : reqHandle = RaSocketListenOneStartAsync(listenInfo_.get());
423 3 : socketStatus = SocketStatus::LISTEN_STARTING;
424 3 : }
425 :
426 16 : void Socket::ConnectAsync()
427 : {
428 16 : if (role == SocketRole::SERVER || socketStatus == SocketStatus::OK) {
429 10 : return;
430 : }
431 :
432 6 : RaSocketConnectParam param(socketHandle, remoteIp, listenPort, tag);
433 6 : reqHandle = RaSocketConnectOneAsync(param);
434 18 : HCCL_INFO("[Socket][%s] conn.tag %s", __func__, tag.c_str());
435 :
436 6 : socketStatus = SocketStatus::CONNECT_STARTING;
437 6 : }
438 :
439 11 : void Socket::SendAsync(const void *sendBuf, u32 size)
440 : {
441 11 : if (!sendBuf) {
442 2 : THROW<SocketException>(StringFormat("[Socket][%s] failed to send, "
443 3 : "sendBuf is nullptr, [%s].", __func__, this->Describe().c_str()));
444 : }
445 :
446 10 : if (size == 0) {
447 2 : THROW<SocketException>(StringFormat("[Socket][%s] failed to send, "
448 3 : "size is 0, [%s].", __func__, this->Describe().c_str()));
449 : }
450 :
451 9 : if (size > MAX_TRANSFER_SIZE) {
452 0 : THROW<SocketException>(StringFormat("[Socket][%s] failed to send, "
453 : "size[%u] is greater than max size[%u], [%s]",
454 0 : __func__, size, MAX_TRANSFER_SIZE, this->Describe().c_str()));
455 : }
456 :
457 9 : if (socketStatus != SocketStatus::OK) {
458 3 : THROW<SocketException>(StringFormat("[Socket][%s] failed to send, "
459 : "status[%s] is not ok, [%s].", __func__,
460 5 : socketStatus.Describe().c_str(), this->Describe().c_str()));
461 : }
462 :
463 8 : sendSize = 0;
464 8 : totalSendSize = 0;
465 8 : sendLeftSize = size;
466 8 : sendDataBuff = sendBuf;
467 :
468 8 : reqHandle = HrtRaSocketSendAsync(fdHandle, sendBuf, sendLeftSize, sendSize);
469 24 : HCCL_INFO("[Socket][%s] reqHandle[%llu] start to send size[%u], [%s].",
470 : __func__, reqHandle, sendLeftSize, this->Describe().c_str());
471 :
472 8 : lastLogTime = {};
473 8 : socketStatus = SocketStatus::SENDING;
474 8 : }
475 :
476 10 : void Socket::RecvAsync(u8 *recvBuf, u32 size)
477 : {
478 10 : if (!recvBuf) {
479 2 : THROW<SocketException>(StringFormat("[Socket][%s] failed to recv, "
480 3 : "recvBuf is nullptr, [%s].", __func__, this->Describe().c_str()));
481 : }
482 :
483 9 : if (size == 0) {
484 2 : THROW<SocketException>(StringFormat("[Socket][%s] failed to recv, "
485 3 : "size is 0, [%s].", __func__, this->Describe().c_str()));
486 : }
487 :
488 8 : if (size > MAX_TRANSFER_SIZE) {
489 0 : THROW<SocketException>(StringFormat("[Socket][%s] failed to recv, "
490 : "size[%u] is greater than max size[%u], [%s].",
491 0 : __func__, size, MAX_TRANSFER_SIZE, this->Describe().c_str()));
492 : }
493 :
494 8 : if (socketStatus != SocketStatus::OK) {
495 0 : THROW<SocketException>(StringFormat("[Socket][%s] failed to recv, "
496 : "status[%s] is not ok, [%s].", __func__,
497 0 : socketStatus.Describe().c_str(), this->Describe().c_str()));
498 : }
499 :
500 8 : recvSize = 0;
501 8 : totalRecvSize = 0;
502 8 : recvLeftSize = size;
503 8 : recvDataBuff = static_cast<void *>(recvBuf);
504 :
505 8 : reqHandle = HrtRaSocketRecvAsync(fdHandle, recvDataBuff, recvLeftSize, recvSize);
506 24 : HCCL_INFO("[Socket][%s] reqHandle[%llu] start to recv size[%u], [%s].",
507 : __func__, reqHandle, recvLeftSize, this->Describe().c_str());
508 :
509 8 : lastLogTime = {};
510 8 : socketStatus = SocketStatus::RECVING;
511 8 : }
512 :
513 : } // namespace Hccl
|
