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