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