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