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 : #define _GNU_SOURCE
12 : #include <unistd.h>
13 : #include <stdlib.h>
14 : #include <netinet/in.h>
15 : #include <arpa/inet.h>
16 : #include <sys/types.h>
17 : #include <sys/eventfd.h>
18 : #include <sys/epoll.h>
19 : #include <sys/syscall.h>
20 : #include <sys/prctl.h>
21 :
22 : #include "rs_tls.h"
23 : #include "ssl_adp.h"
24 : #include "securec.h"
25 : #include "rs.h"
26 : #include "rs_inner.h"
27 : #include "ra_rs_comm.h"
28 : #include "ra_rs_err.h"
29 : #include "rs_drv_rdma.h"
30 : #include "dl_hal_function.h"
31 : #include "rs_drv_socket.h"
32 : #include "rs_socket.h"
33 : #ifdef CONFIG_TLV
34 : #include "rs_adp_nslb.h"
35 : #endif
36 : #include "rs_ub_dfx.h"
37 : #include "rs_epoll.h"
38 :
39 : #define BIND_MIN_DCPU_NUM 1
40 : #define COMMAND_LENGTH 100
41 : #define ADD_TID_SHELL_PATH 50
42 : #define TID_LENGTH 20
43 : #define HOST ((uint32_t)1)
44 :
45 : struct RsPthreadInfo gEpollThreadInfo = {0}; // lint !e17
46 : struct RsPthreadInfo gConnectThreadInfo = {0};
47 :
48 : /*
49 : * opcode:
50 : * ADD: EPOLL_CTL_ADD
51 : * MOD: EPOLL_CTL_MOD
52 : * DEL: EPOLL_CTL_DEL
53 : */
54 230 : int RsEpollCtl(int epollfd, int op, int fd, unsigned int state)
55 : {
56 : int ret;
57 : struct epoll_event ev;
58 :
59 230 : ev.events = state;
60 230 : ev.data.fd = fd;
61 230 : ret = epoll_ctl(epollfd, op, fd, &ev);
62 230 : if (ret) {
63 108 : hccp_warn("epoll_ctl for fd %d unsuccessful! ret:%d errno:%d op:%d state:%u", fd, ret, errno, op, state);
64 : }
65 230 : return ret;
66 : }
67 :
68 1 : int RsEpollCtlFdHandle(int epollfd, int op, int fd, unsigned int state, void *fdHandle)
69 : {
70 : int ret;
71 : struct epoll_event ev;
72 :
73 1 : ev.events = state;
74 1 : ev.data.ptr = fdHandle;
75 1 : ret = epoll_ctl(epollfd, op, fd, &ev);
76 1 : if (ret) {
77 1 : hccp_warn("epoll_ctl for fd %d unsuccessful! ret:%d errno:%d op:%d state:%u", fd, ret, errno, op, state);
78 : }
79 1 : return ret;
80 : }
81 :
82 : int RsWlistCheckConnAdd(struct rs_cb *rsCb, struct RsConnInfo *connTmp)
83 : {
84 : int ret;
85 : int retClose;
86 16 : struct RsConnInfo *conn = NULL;
87 :
88 16 : if (rsCb->connCb.wlistEnable == 1) {
89 0 : ret = RsWhiteListCheckValid(rsCb->chipId, &rsCb->connCb, connTmp);
90 0 : if (ret) {
91 0 : hccp_info("invalid client node found: chip_id %u, fd %d accept, server ip:%s, client ip:%s, port:%u, "
92 : "state:%d, tag:%s",
93 : rsCb->chipId, connTmp->connfd, connTmp->serverIp.readAddr, connTmp->clientIp.readAddr, connTmp->port,
94 : connTmp->state, connTmp->tag);
95 0 : return ret;
96 : }
97 : }
98 :
99 : /* add conn node to server list */
100 16 : ret = RsAllocConnNode(&conn, connTmp->port);
101 16 : if (ret) {
102 0 : hccp_err("server IP:0x%s add conn info to list failed, fd:%d, ret:%d", connTmp->serverIp.readAddr,
103 : connTmp->connfd, ret);
104 0 : goto alloc_err;
105 : }
106 :
107 16 : ret = RsSocketCopyConnInfo(connTmp, conn);
108 16 : if (ret) {
109 0 : hccp_err("rs_socket_copy_conn_info failed, ret[%d]", ret);
110 0 : goto out;
111 : }
112 :
113 16 : if (rsCb->sslEnable == RS_SSL_ENABLE) {
114 0 : ret = RsEpollCtl(rsCb->connCb.epollfd, EPOLL_CTL_DEL, connTmp->connfd, EPOLLIN);
115 0 : if (ret) {
116 0 : hccp_err("rs epoll ctl failed, ret %d", ret);
117 0 : goto out;
118 : }
119 : }
120 :
121 16 : RS_PTHREAD_MUTEX_LOCK(&rsCb->connCb.connMutex);
122 16 : RsListAddTail(&conn->list, &rsCb->connCb.serverConnList);
123 16 : RS_PTHREAD_MUTEX_ULOCK(&rsCb->connCb.connMutex);
124 :
125 16 : hccp_info("[Server]chip_id %u, fd %d accept, server ip:%s, client ip:%s, state:%d, tag:%s", rsCb->chipId,
126 : connTmp->connfd, connTmp->serverIp.readAddr, connTmp->clientIp.readAddr, connTmp->state, connTmp->tag);
127 16 : ShowConnNode(&(rsCb->connCb.serverConnList));
128 16 : return 0;
129 :
130 0 : out:
131 0 : free(conn);
132 0 : conn = NULL;
133 0 : alloc_err:
134 0 : RS_CLOSE_RETRY_FOR_EINTR(retClose, connTmp->connfd);
135 0 : return ret;
136 : }
137 :
138 1 : STATIC int RsSslRecvTagInHandle(struct RsAcceptInfo *acceptInfo, struct RsConnInfo *connTmp)
139 : {
140 1 : int expSize = SOCK_CONN_TAG_SIZE + SOCK_CONN_DEV_ID_SIZE;
141 1 : char *recvBuff = connTmp->tag;
142 : struct timeval startTime, now;
143 1 : float timeCost = 0.0;
144 1 : int size = 0;
145 1 : int ret = 0;
146 :
147 1 : RsGetCurTime(&startTime);
148 1 : while (expSize > 0) {
149 1 : connTmp->tagSyncTimes++;
150 1 : size = ssl_adp_read(acceptInfo->ssl, recvBuff, expSize);
151 1 : if (size > 0) {
152 0 : expSize -= size;
153 0 : recvBuff += size;
154 : } else {
155 1 : ret = RsSslReadInnerCheck(acceptInfo->ssl, acceptInfo->connFd, size, (uint64_t)expSize);
156 1 : if (ret == -EAGAIN) {
157 0 : connTmp->tagEintrTimes++;
158 : } else {
159 1 : hccp_run_warn("ssl_adp_read unsuccessful, fd:%d ret:%d tagSyncTime:%u tagEintrTime:%u",
160 : acceptInfo->connFd, ret, connTmp->tagSyncTimes, connTmp->tagEintrTimes);
161 1 : return -ESOCKCLOSED;
162 : }
163 : }
164 :
165 0 : RsGetCurTime(&now);
166 0 : HccpTimeInterval(&now, &startTime, &timeCost);
167 0 : if (timeCost >= RS_RECV_MAX_TIME) {
168 0 : hccp_run_info("recv tag time out, server:{%s:%u} client:%s tagSyncTime:%u tagEintrTime:%u",
169 : acceptInfo->serverIpAddr.readAddr, acceptInfo->sockPort, acceptInfo->clientIpAddr.readAddr,
170 : connTmp->tagSyncTimes, connTmp->tagEintrTimes);
171 0 : return -ETIME;
172 : }
173 :
174 0 : if (timeCost <= 0) {
175 0 : RsGetCurTime(&startTime);
176 : }
177 : }
178 :
179 0 : connTmp->serverIp = acceptInfo->serverIpAddr;
180 0 : connTmp->clientIp = acceptInfo->clientIpAddr;
181 0 : connTmp->connfd = acceptInfo->connFd;
182 0 : connTmp->state = RS_CONN_STATE_TAG_SYNC;
183 0 : connTmp->port = acceptInfo->sockPort;
184 0 : connTmp->ssl = acceptInfo->ssl;
185 :
186 0 : hccp_info("recv tag success, server:{%s:%u} client:%s timeCost:%fms tagSyncTime:%u tagEintrTime:%u",
187 : acceptInfo->serverIpAddr.readAddr, acceptInfo->sockPort, acceptInfo->clientIpAddr.readAddr, timeCost,
188 : connTmp->tagSyncTimes, connTmp->tagEintrTimes);
189 0 : return 0;
190 : }
191 :
192 1 : STATIC void RsEpollEventSslRecvTagInHandle(struct rs_cb *rsCb, struct RsAcceptInfo *acceptInfo)
193 : {
194 1 : struct RsConnInfo connTmp = {0};
195 : int retClose;
196 : int ret;
197 :
198 1 : ret = RsSslRecvTagInHandle(acceptInfo, &connTmp);
199 1 : if (ret != 0) {
200 0 : RS_CLOSE_RETRY_FOR_EINTR(retClose, acceptInfo->connFd);
201 0 : goto out;
202 : }
203 :
204 1 : ret = RsWlistCheckConnAdd(rsCb, &connTmp);
205 1 : out:
206 1 : if (ret != 0) {
207 1 : hccp_warn("recv tag or add conn unsuccessful ret:%d", ret);
208 1 : ssl_adp_shutdown(acceptInfo->ssl);
209 1 : ssl_adp_free(acceptInfo->ssl);
210 1 : acceptInfo->ssl = NULL;
211 : }
212 :
213 : /* do not shutdown ssl */
214 1 : RS_PTHREAD_MUTEX_LOCK(&rsCb->connCb.connMutex);
215 1 : RsListDel(&acceptInfo->list);
216 1 : free(acceptInfo);
217 1 : acceptInfo = NULL;
218 1 : RS_PTHREAD_MUTEX_ULOCK(&rsCb->connCb.connMutex);
219 :
220 1 : return;
221 : }
222 :
223 0 : STATIC void RsDoSslHandshake(struct RsAcceptInfo *acceptInfo, struct rs_cb *rscb)
224 : {
225 : int ret;
226 : int err;
227 :
228 0 : ret = ssl_adp_do_handshake(acceptInfo->ssl);
229 0 : if (ret == 1) {
230 0 : ret = rs_tls_peer_cert_verify(acceptInfo->ssl, rscb);
231 0 : if (ret) {
232 0 : hccp_err("tls verify peer cert failed");
233 0 : return;
234 : }
235 0 : acceptInfo->state = RS_CONN_STATE_SSL_CONNECTED;
236 : } else {
237 0 : err = ssl_adp_get_error(acceptInfo->ssl, ret);
238 0 : if (err == SSL_ERROR_WANT_WRITE) {
239 0 : hccp_info("return want write");
240 0 : return;
241 0 : } else if (err == SSL_ERROR_WANT_READ) {
242 0 : hccp_info("return want read");
243 0 : return;
244 : } else {
245 0 : rs_ssl_err_string(acceptInfo->connFd, err);
246 0 : return;
247 : }
248 : }
249 :
250 0 : return;
251 : }
252 :
253 1 : STATIC int RsEpollEventSslAcceptInHandle(struct rs_cb *rsCb, int fd)
254 : {
255 : int ret;
256 :
257 1 : struct RsAcceptInfo *acceptInfo = NULL;
258 1 : struct RsAcceptInfo *acceptInfo2 = NULL;
259 :
260 : /* Server event: ssl accept */
261 1 : RS_LIST_GET_HEAD_ENTRY(acceptInfo, acceptInfo2, &rsCb->connCb.serverAcceptList, list, struct RsAcceptInfo);
262 1 : for (; (&acceptInfo->list) != &rsCb->connCb.serverAcceptList;
263 0 : acceptInfo = acceptInfo2, acceptInfo2 = list_entry(acceptInfo2->list.next, struct RsAcceptInfo, list)) {
264 : /* connection request for Server */
265 0 : if (fd == acceptInfo->connFd) {
266 0 : if (acceptInfo->ssl == NULL) {
267 0 : acceptInfo->ssl = ssl_adp_new(rsCb->serverSslCtx);
268 0 : CHK_PRT_RETURN(acceptInfo->ssl == NULL, hccp_err("server ssl ctx alloc failed"), -ENOMEM);
269 :
270 0 : ret = ssl_adp_set_fd(acceptInfo->ssl, acceptInfo->connFd);
271 0 : if (ret != 1) {
272 0 : hccp_err("bind connfd and ssl failed, ret %d", ret);
273 0 : ssl_adp_shutdown(acceptInfo->ssl);
274 0 : ssl_adp_free(acceptInfo->ssl);
275 0 : acceptInfo->ssl = NULL;
276 0 : return -EINVAL;
277 : }
278 :
279 0 : ssl_adp_set_mode(acceptInfo->ssl, SSL_MODE_AUTO_RETRY);
280 0 : ssl_adp_set_accept_state(acceptInfo->ssl);
281 : }
282 :
283 0 : if (acceptInfo->state == RS_CONN_STATE_RESET) {
284 0 : RsDoSslHandshake(acceptInfo, rsCb);
285 0 : return 0;
286 : }
287 :
288 0 : if (acceptInfo->state == RS_CONN_STATE_SSL_CONNECTED) {
289 0 : RsEpollEventSslRecvTagInHandle(rsCb, acceptInfo);
290 0 : return 0;
291 : }
292 : }
293 : }
294 :
295 1 : return -ENODEV;
296 : }
297 :
298 2 : STATIC int RsEpollTcpRecv(struct rs_cb *rsCb, int fd)
299 : {
300 2 : if (rsCb->tcpRecvCallback != NULL) {
301 1 : rsCb->tcpRecvCallback(rsCb->fdMap[fd]);
302 : } else {
303 1 : hccp_err("[rs_epoll_tcp_recv]tcp_recv_callback is null.");
304 1 : return -EINVAL;
305 : }
306 1 : return 0;
307 : }
308 :
309 0 : STATIC int RsEpollEventHeterogTcpRecvInHandle(struct rs_cb *rsCb, int fd)
310 : {
311 : int ret;
312 0 : struct RsHeterogTcpFdInfo *fdNode = NULL;
313 0 : struct RsHeterogTcpFdInfo *fdNode1 = NULL;
314 :
315 0 : RS_LIST_GET_HEAD_ENTRY(fdNode, fdNode1, &rsCb->heterogTcpFdList, list, struct RsHeterogTcpFdInfo);
316 0 : for (; (&fdNode->list) != &rsCb->heterogTcpFdList;
317 0 : fdNode = fdNode1, fdNode1 = list_entry(fdNode1->list.next, struct RsHeterogTcpFdInfo, list)) {
318 0 : if (fdNode->fd == fd) {
319 : // 处理tcp recv
320 0 : ret = RsEpollTcpRecv(rsCb, fd);
321 0 : return ret;
322 : }
323 : }
324 0 : return -ENODEV;
325 : }
326 :
327 18 : STATIC void RsEpollEventInHandle(struct rs_cb *rsCb, struct epoll_event *events)
328 : {
329 18 : int fd = events->data.fd;
330 : int ret;
331 :
332 18 : ret = RsEpollEventPingHandle(rsCb, fd);
333 18 : if (ret != -ENODEV) {
334 0 : hccp_info("the fd:%d is for ping, no need to go on, ret:%d", fd, ret);
335 0 : return;
336 : }
337 :
338 18 : ret = RsEpollEventListenInHandle(rsCb, fd);
339 18 : if (ret != -ENODEV) {
340 16 : hccp_info("the fd:%d is tcp listened, no need to go on, ret:%d", fd, ret);
341 16 : return;
342 : }
343 :
344 2 : if (rsCb->sslEnable == RS_SSL_ENABLE) {
345 1 : ret = RsEpollEventSslAcceptInHandle(rsCb, fd);
346 1 : if (ret != -ENODEV) {
347 1 : hccp_info("the fd:%d is ssl accept, no need to go on, ret:%d", fd, ret);
348 1 : return;
349 : }
350 : }
351 :
352 1 : ret = RsEpollEventQpMrInHandle(rsCb, fd);
353 1 : if (ret != -ENODEV) {
354 0 : hccp_info("the fd:%d is for qp mr, no need to go on, ret:%d", fd, ret);
355 0 : return;
356 : }
357 :
358 1 : ret = RsEpollEventHeterogTcpRecvInHandle(rsCb, fd);
359 1 : if (ret != -ENODEV) {
360 1 : hccp_info("the fd:%d is for tcp recv, no need to go on, ret:%d", fd, ret);
361 1 : return;
362 : }
363 :
364 0 : if (RsIsUdmaSupported()) {
365 0 : ret = RsEpollEventJfcInHandle(rsCb, fd);
366 0 : if (ret != -ENODEV) {
367 0 : hccp_info("the fd:%d is for poll jfc, no need to go on, ret:%d", fd, ret);
368 0 : return;
369 : }
370 :
371 0 : ret = RsEpollEventUrmaAsyncEventInHandle(rsCb, fd);
372 0 : if (ret != -ENODEV) {
373 0 : hccp_info("the fd:%d is for urma async event, no need to go on, ret:%d", fd, ret);
374 0 : return;
375 : }
376 : }
377 :
378 0 : return;
379 : }
380 :
381 13 : STATIC void RsEpollEventHandleOne(struct rs_cb *rsCb, struct epoll_event *events)
382 : {
383 13 : int ret = 0;
384 :
385 : (void)ret;
386 13 : RS_CHECK_POINTER_NULL_RETURN_VOID(events);
387 13 : RS_CHECK_POINTER_NULL_RETURN_VOID(rsCb);
388 :
389 : #ifdef CONFIG_TLV
390 : if (RsIsTlvSupported()) {
391 : ret = RsEpollNslbEventHandle(&rsCb->tlvCb.nslbCb, events->data.fd, events->events);
392 : if (ret != -ENODEV) {
393 : hccp_info("the fd:%d is nslb event, no need to go on, ret:%d", events->data.fd, ret);
394 : return;
395 : }
396 : }
397 : #endif
398 :
399 13 : if (events->events & EPOLLIN) {
400 12 : if (events->events & EPOLLRDHUP) {
401 0 : hccp_dbg("Peer socket has been closed!");
402 0 : return;
403 : }
404 12 : RsEpollEventInHandle(rsCb, events);
405 : } else {
406 1 : hccp_warn("unknown event(0x%x)!", events->events);
407 : }
408 :
409 13 : return;
410 : }
411 :
412 1 : STATIC int SetAffinity(unsigned int chipId, unsigned int cpuId)
413 : {
414 : cpu_set_t mask;
415 : cpu_set_t get;
416 : int ret;
417 :
418 1 : CPU_ZERO(&mask);
419 :
420 : // 设置线程CPU亲和力
421 1 : CPU_SET(cpuId, &mask);
422 1 : ret = pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask);
423 1 : CHK_PRT_RETURN(ret < 0, hccp_err("could not set CPU affinity"), ret);
424 :
425 : // 获取线程cpu亲和力
426 1 : CPU_ZERO(&get);
427 1 : ret = pthread_getaffinity_np(pthread_self(), sizeof(get), &get);
428 1 : CHK_PRT_RETURN(ret, hccp_err("could not get CPU affinity"), ret);
429 :
430 1 : if (CPU_ISSET(cpuId, &get)) { // 检查cpuid是否在这个集合中
431 1 : hccp_info("dev is %d thread %llu is running in processor %d.", chipId, pthread_self(), cpuId);
432 : } else {
433 0 : hccp_warn("dev is %d thread %llu is not running in processor %d.", chipId, pthread_self(), cpuId);
434 : }
435 :
436 1 : return ret;
437 : }
438 :
439 41 : STATIC int BindDataCpu(unsigned int chipId)
440 : {
441 : int ret;
442 : unsigned int cpuId;
443 41 : uint32_t info = 0;
444 41 : uint32_t devNum = 0;
445 : int64_t ccpuNum;
446 : int64_t dcpuNum;
447 : int64_t acpuNum;
448 : int64_t cpuNum;
449 :
450 : // 判断当前处于host or device侧，如果在host侧，无需进行绑核操作
451 41 : ret = DlDrvGetPlatformInfo(&info);
452 41 : CHK_PRT_RETURN(ret, hccp_err("get PlatformInfo failed, ret[%d]", ret), ret);
453 41 : if (info == HOST) {
454 0 : hccp_info("host not need bind cpu, info[%u]", info);
455 0 : return 0;
456 : }
457 :
458 : // 获取当前os的device数量
459 41 : ret = DlDrvGetDevNum(&devNum);
460 41 : CHK_PRT_RETURN(ret, hccp_err("get device_num failed, ret[%d]", ret), ret);
461 39 : if (devNum == 0) {
462 0 : hccp_info("no device need bind cpu, device num %u", devNum);
463 0 : return 0;
464 : }
465 :
466 : // 获取data cpu数量
467 39 : ret = DlHalGetDeviceInfo(chipId, MODULE_TYPE_DCPU, INFO_TYPE_CORE_NUM, &dcpuNum);
468 39 : CHK_PRT_RETURN(ret, hccp_err("get dcpu_num failed, ret(%d)", ret), ret);
469 :
470 : // 如果dcpu_num < 1，无需绑核直接返回
471 39 : if (dcpuNum < BIND_MIN_DCPU_NUM) {
472 38 : hccp_info("data cpu num %d, device not need bind cpu", dcpuNum);
473 38 : return 0;
474 : }
475 :
476 : // 获取ctrl cpu数量
477 1 : ret = DlHalGetDeviceInfo(chipId, MODULE_TYPE_CCPU, INFO_TYPE_CORE_NUM, &ccpuNum);
478 1 : CHK_PRT_RETURN(ret, hccp_err("get ccpu_num failed, ret(%d)", ret), ret);
479 :
480 : // 获取ai cpu数量
481 1 : ret = DlHalGetDeviceInfo(chipId, MODULE_TYPE_AICPU, INFO_TYPE_CORE_NUM, &acpuNum);
482 1 : CHK_PRT_RETURN(ret, hccp_err("get acpu_num failed, ret(%d)", ret), ret);
483 :
484 : // 计算单个device上的核数
485 1 : cpuNum = ccpuNum + dcpuNum + acpuNum;
486 1 : hccp_info("halGetDeviceInf chip = %u dev_num = %u, ccpu = %lld dcpu = %lld acpu = %lld cpuNum = %lld", chipId,
487 : devNum, ccpuNum, dcpuNum, acpuNum, cpuNum);
488 :
489 1 : cpuId = (unsigned int)((int64_t)(chipId % devNum) * cpuNum + ccpuNum);
490 :
491 : // 进行绑核
492 1 : ret = DlHalBindCgroup(BIND_DATACPU_CGROUP);
493 1 : CHK_PRT_RETURN(ret, hccp_err("bind cgroup failed, ret[%d], strerror[%s]", ret, strerror(errno)), ret);
494 1 : hccp_info("bind cgroup success!");
495 :
496 1 : ret = SetAffinity(chipId, cpuId);
497 1 : CHK_PRT_RETURN(ret, hccp_err("set affinity with cpu[%u] failed", cpuId), ret);
498 :
499 1 : hccp_info("chip_id[%u] bind data cpu[%u] success", chipId, cpuId);
500 :
501 1 : return ret;
502 : }
503 :
504 57 : STATIC void RsSocketSaveEpollWaitErrInfo(int eventNum, int errNo, struct SocketErrInfo *epollErrInfo)
505 : {
506 57 : if (eventNum > 0) {
507 57 : return;
508 : }
509 :
510 0 : RsSocketSaveErrInfo(RS_CONN_STATE_RESET, errNo, epollErrInfo);
511 : }
512 :
513 41 : STATIC void *RsEpollHandle(void *arg)
514 : {
515 41 : struct RsConnCb *connCb = NULL;
516 41 : struct rs_cb *rsCb = NULL;
517 : eventfd_t val;
518 : int ret;
519 : int num;
520 : int i;
521 :
522 41 : RS_CHECK_POINTER_NULL_RETURN_NULL(arg);
523 :
524 41 : hccp_info("<EPOLL> thread begin! thread_id:%lu, pid:%d, ppid:%d", pthread_self(), getpid(), getppid());
525 :
526 : struct epoll_event events[RS_EPOLL_EVENT];
527 :
528 41 : CHK_PRT_RETURN(pthread_detach(pthread_self()),
529 : hccp_err("pthread_detach failed! thread_id:%lu, errno:%d", pthread_self(), errno), NULL);
530 :
531 41 : (void)prctl(PR_SET_NAME, (uintptr_t) "hccp_epoll", 0, 0, 0);
532 :
533 41 : rsCb = (struct rs_cb *)arg;
534 41 : gRsCb = rsCb;
535 41 : connCb = &rsCb->connCb;
536 :
537 41 : ret = BindDataCpu(rsCb->chipId);
538 41 : if (ret) {
539 2 : hccp_warn("bind data cpu unsuccessful! thread_id:%lu, errno:%d", pthread_self(), errno);
540 : }
541 :
542 41 : rsCb->state &= ~RS_STATE_HALT;
543 41 : RsGetCurTime(&gEpollThreadInfo.lastCheckTime);
544 41 : ret = strncpy_s((char *)gEpollThreadInfo.pthreadName, sizeof(gEpollThreadInfo.pthreadName), "epoll_pthread",
545 : strlen("epoll_pthread"));
546 41 : CHK_PRT_RETURN(ret, hccp_err("strncpy_s pthread name failed, ret[%d]", ret), NULL);
547 :
548 41 : hccp_run_info("pthread[%s] is alive!", gEpollThreadInfo.pthreadName);
549 :
550 41 : ret = RsDrvInitCqeErrInfo();
551 41 : CHK_PRT_RETURN(ret, hccp_err("rs_drv_init_cqe_err_info failed, ret[%d]", ret), NULL);
552 16 : while (1) {
553 : do {
554 57 : num = epoll_wait(connCb->epollfd, events, RS_EPOLL_EVENT, -1);
555 57 : } while ((num < 0) && (errno == EINTR));
556 57 : RsSocketSaveEpollWaitErrInfo(num, -errno, &connCb->epollErrInfo);
557 :
558 : /* eventfd is for wake up epoll wait, value is ignored */
559 57 : if (events[0].data.fd == connCb->eventfd) {
560 41 : hccp_warn("<EPOLL> SHUT DOWN event eventfd:%d", connCb->eventfd);
561 : do {
562 41 : num = read(connCb->eventfd, &val, sizeof(eventfd_t));
563 41 : } while ((num < 0) && (errno == EINTR));
564 41 : RS_PTHREAD_MUTEX_LOCK(&rsCb->mutex);
565 41 : rsCb->state |= RS_STATE_HALT;
566 41 : RS_PTHREAD_MUTEX_ULOCK(&rsCb->mutex);
567 :
568 41 : break;
569 : }
570 :
571 16 : RsHeartbeatAlivePrint(&gEpollThreadInfo);
572 16 : RS_PTHREAD_MUTEX_LOCK(&rsCb->mutex);
573 : /* events handle one by one */
574 32 : for (i = 0; i < num; i++) {
575 16 : RsEpollEventHandleOne(rsCb, &events[i]);
576 : }
577 :
578 16 : RS_PTHREAD_MUTEX_ULOCK(&rsCb->mutex);
579 : }
580 41 : RsDrvDeinitCqeErrInfo();
581 41 : hccp_info("<EPOLL> QUIT thread_id:%lu, pid:%d, read num:%d", pthread_self(), getpid(), num);
582 :
583 41 : return NULL;
584 : }
585 :
586 44 : STATIC int RsCreateEpoll(struct rs_cb *rsCb)
587 : {
588 : int ret, retFd;
589 44 : struct RsConnCb *connCb = NULL;
590 :
591 44 : connCb = &rsCb->connCb;
592 :
593 44 : connCb->epollfd = epoll_create(1);
594 44 : CHK_PRT_RETURN(connCb->epollfd < 0, hccp_err("epollfd[%d] failed, errno[%d]", connCb->epollfd, errno), -EINVAL);
595 :
596 43 : connCb->eventfd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
597 43 : if (connCb->eventfd == RS_FD_INVALID) {
598 0 : RS_CLOSE_RETRY_FOR_EINTR(retFd, connCb->epollfd);
599 0 : connCb->epollfd = RS_FD_INVALID;
600 0 : hccp_err("create eventfd for rs cb failed !");
601 0 : return -EINVAL;
602 : }
603 :
604 43 : ret = RsEpollCtl(connCb->epollfd, EPOLL_CTL_ADD, connCb->eventfd, EPOLLIN);
605 43 : if (ret) {
606 0 : hccp_err("re epoll ctl failed, epollfd[%d], eventfd[%d]", connCb->epollfd, connCb->eventfd);
607 0 : RS_CLOSE_RETRY_FOR_EINTR(retFd, connCb->eventfd);
608 0 : connCb->eventfd = RS_FD_INVALID;
609 0 : RS_CLOSE_RETRY_FOR_EINTR(retFd, connCb->epollfd);
610 0 : connCb->epollfd = RS_FD_INVALID;
611 0 : return ret;
612 : }
613 :
614 43 : return 0;
615 : }
616 :
617 43 : void RsDestroyEpoll(struct rs_cb *rsCb)
618 : {
619 : int ret;
620 43 : struct RsConnCb *connCb = NULL;
621 :
622 43 : connCb = &rsCb->connCb;
623 :
624 43 : ret = RsEpollCtl(connCb->epollfd, EPOLL_CTL_DEL, connCb->eventfd, EPOLLIN);
625 43 : if (ret) {
626 0 : hccp_warn("re epoll ctl unsuccessful, epollfd[%d], eventfd[%d]", connCb->epollfd, connCb->eventfd);
627 : }
628 :
629 43 : RS_CLOSE_RETRY_FOR_EINTR(ret, connCb->eventfd);
630 43 : connCb->eventfd = RS_FD_INVALID;
631 :
632 43 : RS_CLOSE_RETRY_FOR_EINTR(ret, connCb->epollfd);
633 43 : connCb->epollfd = RS_FD_INVALID;
634 :
635 43 : return;
636 : }
637 :
638 61 : STATIC void RsUsleepWaitConn(sem_t *sem, uint32_t timeoutUs)
639 : {
640 61 : uint32_t sleepUs = 1000; // 每 1000us 查看sem是否为零
641 61 : uint32_t elapsedUs = 0;
642 : do {
643 9266 : if (sem_trywait(sem) == 0) {
644 15 : return;
645 : }
646 9251 : usleep(sleepUs);
647 9251 : elapsedUs += sleepUs;
648 9251 : } while (elapsedUs < timeoutUs);
649 46 : return;
650 : }
651 :
652 43 : STATIC void *RsConnectHandle(void *arg)
653 : {
654 43 : struct RsConnInfo *connTmp2 = NULL;
655 43 : struct RsConnInfo *connTmp = NULL;
656 43 : bool promoteConnect = false;
657 : int ret;
658 :
659 43 : hccp_info("<SOCKET> thread begin! thread_id:%lu, pid:%d, ppid:%d", pthread_self(), getpid(), getppid());
660 43 : CHK_PRT_RETURN(pthread_detach(pthread_self()),
661 : hccp_err("pthread_detach failed! thread_id:%lu, errno:%d", pthread_self(), errno), NULL);
662 :
663 42 : (void)prctl(PR_SET_NAME, (uintptr_t) "hccp_connect", 0, 0, 0);
664 :
665 42 : RsGetCurTime(&gConnectThreadInfo.lastCheckTime);
666 42 : ret = strncpy_s((char *)gConnectThreadInfo.pthreadName, sizeof(gConnectThreadInfo.pthreadName), "connect_pthread",
667 : strlen("connect_pthread"));
668 42 : CHK_PRT_RETURN(ret, hccp_err("strncpy_s pthread name failed, ret[%d]", ret), NULL);
669 42 : struct rs_cb *rsCb = (struct rs_cb *)arg;
670 42 : if (rsCb == NULL) {
671 1 : return NULL;
672 : }
673 :
674 41 : gRsCb = rsCb;
675 41 : sem_init(&rsCb->connectTrigSem, 0, 0);
676 :
677 41 : hccp_run_info("pthread[%s] is alive!", gConnectThreadInfo.pthreadName);
678 : while (1) {
679 102 : if (rsCb == NULL) {
680 0 : return NULL;
681 : }
682 :
683 102 : if (rsCb->connFlag == 0) {
684 41 : break;
685 : }
686 :
687 61 : promoteConnect = false;
688 61 : RsHeartbeatAlivePrint(&gConnectThreadInfo);
689 61 : RS_PTHREAD_MUTEX_LOCK(&rsCb->mutex);
690 61 : RS_LIST_GET_HEAD_ENTRY(connTmp, connTmp2, &rsCb->connCb.clientConnList, list, struct RsConnInfo);
691 82 : for (; (&connTmp->list) != &rsCb->connCb.clientConnList;
692 21 : connTmp = connTmp2, connTmp2 = list_entry(connTmp2->list.next, struct RsConnInfo, list)) {
693 21 : ret = RsSocketConnectAsync(connTmp, rsCb);
694 21 : if (ret != 0 && connTmp->state == RS_CONN_STATE_RESET) {
695 0 : connTmp->state = RS_CONN_STATE_ERR;
696 0 : hccp_err("[client]rs_socket_connect_async failed at RS_CONN_STATE_RESET state, ret:%d, clientIp:%s "
697 : "serverIp:%s server_port:%u tag:%s",
698 : ret, connTmp->clientIp.readAddr, connTmp->serverIp.readAddr, connTmp->port, connTmp->tag);
699 0 : continue;
700 : }
701 21 : if ((promoteConnect == false) && (RsGetSocketConnectState(connTmp) == 0)) {
702 0 : promoteConnect = true;
703 : }
704 : }
705 61 : RS_PTHREAD_MUTEX_ULOCK(&rsCb->mutex);
706 :
707 61 : if (promoteConnect) {
708 0 : usleep(RS_PROMOTE_CONN_USLEEP_TIME);
709 : } else {
710 61 : RsUsleepWaitConn(&rsCb->connectTrigSem, RS_CONN_USLEEP_TIME);
711 : }
712 : }
713 41 : sem_destroy(&rsCb->connectTrigSem);
714 41 : rsCb->connFlag = RS_CONN_EXIT_FLAG;
715 41 : hccp_info("<SOCKET> QUIT thread_id:%lu, pid:%d", pthread_self(), getpid());
716 :
717 41 : return NULL;
718 : }
719 :
720 44 : int RsEpollConnectHandleInit(struct rs_cb *rscb)
721 : {
722 : int ret;
723 : pthread_t ntid;
724 :
725 44 : ret = RsCreateEpoll(rscb);
726 44 : CHK_PRT_RETURN(ret, hccp_err("rs_create_epoll failed ! ret:%d", ret), ret);
727 :
728 43 : hccp_info("rs_create_epoll ok");
729 43 : gRsCb = rscb;
730 :
731 43 : ret = pthread_create(&ntid, NULL, RsEpollHandle, (void *)rscb);
732 43 : if (ret != 0) {
733 1 : gRsCb = NULL;
734 1 : RsDestroyEpoll(rscb);
735 1 : hccp_err("pthread_create failed ! ret:%d, errno:%d", ret, errno);
736 1 : return -ESYSFUNC;
737 : }
738 :
739 : pthread_t tidp;
740 42 : rscb->connFlag = 1;
741 42 : ret = pthread_create(&tidp, NULL, RsConnectHandle, (void *)rscb);
742 42 : if (ret != 0) {
743 1 : gRsCb = NULL;
744 1 : RsDestroyEpoll(rscb);
745 1 : hccp_err("conn pthread_create failed ! ret:%d, errno:%d", ret, errno);
746 1 : return -ESYSFUNC;
747 : }
748 :
749 41 : hccp_info("RS INIT OK!");
750 :
751 41 : return ret;
752 : }
753 :
754 6 : int RsEpollCreateEpollfd(int *epollfd)
755 : {
756 6 : RS_CHECK_POINTER_NULL_WITH_RET(epollfd);
757 :
758 : // 1024 specify the max fd num, this arg will be ignored since Linux 2.6.8
759 4 : *epollfd = epoll_create(1024);
760 4 : CHK_PRT_RETURN(*epollfd < 0, hccp_err("create epollfd[%d] failed, errno[%d]", *epollfd, errno), -EINVAL);
761 :
762 4 : return 0;
763 : }
764 :
765 6 : int RsEpollDestroyFd(int *fd)
766 : {
767 6 : int ret = 0;
768 :
769 6 : RS_CHECK_POINTER_NULL_WITH_RET(fd);
770 :
771 4 : RS_CLOSE_RETRY_FOR_EINTR(ret, *fd);
772 4 : *fd = RS_FD_INVALID;
773 :
774 4 : return ret;
775 : }
776 :
777 2 : int RsEpollWaitHandle(int eventHandle, struct epoll_event *events, int timeout, unsigned int maxevents,
778 : unsigned int *eventsNum)
779 : {
780 : int eventCount;
781 :
782 2 : RS_CHECK_POINTER_NULL_WITH_RET(events);
783 1 : RS_CHECK_POINTER_NULL_WITH_RET(eventsNum);
784 :
785 1 : eventCount = epoll_wait(eventHandle, events, (int)maxevents, timeout);
786 1 : if (eventCount < 0) {
787 0 : hccp_err("[rs_epoll_wait_handle]epoll_wait failed, strerror[%s]", strerror(errno));
788 0 : return -EIO;
789 : }
790 :
791 1 : *eventsNum = (unsigned int)eventCount;
792 1 : return 0;
793 : }
794 :
795 : RS_ATTRI_VISI_DEF int RsCreateEventHandle(int *eventHandle)
796 : {
797 : int ret;
798 :
799 4 : ret = RsEpollCreateEpollfd(eventHandle);
800 4 : CHK_PRT_RETURN(ret, hccp_err("rs_epoll_create_epollfd failed ret(%d)", ret), ret);
801 3 : return 0;
802 : }
803 :
804 : RS_ATTRI_VISI_DEF int RsCtlEventHandle(int eventHandle, const void *fdHandle, int opcode, enum RaEpollEvent event)
805 : {
806 5 : int fd = RS_FD_INVALID;
807 : unsigned int tmpEvent;
808 : int ret;
809 :
810 5 : if (eventHandle < 0) {
811 1 : hccp_err("event_handle[%d] is invalid", eventHandle);
812 1 : return -EINVAL;
813 : }
814 4 : if (fdHandle == NULL) {
815 1 : hccp_err("fd_handle is NULL");
816 1 : return -EINVAL;
817 : }
818 3 : if (opcode != EPOLL_CTL_ADD && opcode != EPOLL_CTL_DEL && opcode != EPOLL_CTL_MOD) {
819 1 : hccp_err("opcode[%d] invalid, valid opcode includes {%d, %d, %d}", opcode, EPOLL_CTL_ADD, EPOLL_CTL_DEL,
820 : EPOLL_CTL_MOD);
821 1 : return -EINVAL;
822 : }
823 :
824 2 : if (event == RA_EPOLLONESHOT) {
825 1 : tmpEvent = EPOLLIN | EPOLLET | EPOLLONESHOT;
826 1 : } else if (event == RA_EPOLLIN) {
827 0 : tmpEvent = EPOLLIN;
828 1 : } else if (event == RA_EPOLLOUT) {
829 0 : tmpEvent = EPOLLOUT;
830 1 : } else if (event == RA_EPOLLOUT_LET_ONESHOT) {
831 0 : tmpEvent = EPOLLOUT | EPOLLET | EPOLLONESHOT;
832 : } else {
833 1 : hccp_err("unknown event[%d]", event);
834 1 : return -EINVAL;
835 : }
836 :
837 1 : tmpEvent = tmpEvent | EPOLLRDHUP;
838 1 : fd = ((const struct SocketPeerInfo *)fdHandle)->fd;
839 :
840 1 : ret = RsEpollCtlFdHandle(eventHandle, opcode, fd, tmpEvent, (void *)fdHandle);
841 1 : CHK_PRT_RETURN(ret, hccp_err("rs_epoll_ctl_fd_handle failed ret(%d), fd:%d", ret, fd), ret);
842 0 : return 0;
843 : }
844 :
845 : RS_ATTRI_VISI_DEF int RsWaitEventHandle(int eventHandle, struct SocketEventInfoT *eventInfos, int timeout,
846 : unsigned int maxevents, unsigned int *eventsNum)
847 : {
848 : int ret;
849 :
850 4 : if (eventHandle < 0) {
851 1 : hccp_err("event_handle[%d] is invalid", eventHandle);
852 1 : return -EINVAL;
853 : }
854 :
855 3 : if (eventInfos == NULL) {
856 1 : hccp_err("event_info is NULL");
857 1 : return -EINVAL;
858 : }
859 :
860 2 : if (timeout < -1) {
861 1 : hccp_err("timeout[%d ms] is invalid", timeout);
862 1 : return -EINVAL;
863 : }
864 :
865 1 : if (maxevents > MAX_SOCKET_EVENT_NUM) {
866 0 : hccp_err("maxevents[%u] exceeds %u", maxevents, MAX_SOCKET_EVENT_NUM);
867 0 : return -EINVAL;
868 : }
869 :
870 1 : if (eventsNum == NULL) {
871 0 : hccp_err("events_num is NULL");
872 0 : return -EINVAL;
873 : }
874 :
875 1 : ret = RsEpollWaitHandle(eventHandle, (struct epoll_event *)eventInfos, timeout, maxevents, eventsNum);
876 1 : CHK_PRT_RETURN(ret, hccp_err("rs_epoll_wait_handle failed ret(%d)", ret), ret);
877 :
878 1 : return 0;
879 : }
880 :
881 : RS_ATTRI_VISI_DEF int RsDestroyEventHandle(int *eventHandle)
882 : {
883 : int ret;
884 :
885 4 : ret = RsEpollDestroyFd(eventHandle);
886 4 : CHK_PRT_RETURN(ret, hccp_err("rs_epoll_destroy_fd failed ret(%d)", ret), ret);
887 3 : return 0;
888 : }
|
