From c9279cdc115636da78bf041323ad464dbaaf1070 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Tue, 6 Apr 2021 17:04:31 +0800 Subject: [PATCH 1/6] refactor rpc --- src/rpc/src/rpcTcp.c | 155 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 143 insertions(+), 12 deletions(-) diff --git a/src/rpc/src/rpcTcp.c b/src/rpc/src/rpcTcp.c index 3162ab2e4c..286ed223c7 100644 --- a/src/rpc/src/rpcTcp.c +++ b/src/rpc/src/rpcTcp.c @@ -21,6 +21,13 @@ #include "rpcLog.h" #include "rpcHead.h" #include "rpcTcp.h" +#include "tlist.h" + +typedef struct SConnItem { + SOCKET fd; + uint32_t ip; + uint16_t port; +} SConnItem; typedef struct SFdObj { void *signature; @@ -38,6 +45,12 @@ typedef struct SThreadObj { pthread_t thread; SFdObj * pHead; pthread_mutex_t mutex; + // receive the notify from dispatch thread + + int notifyReceiveFd; + int notifySendFd; + SList *connQueue; + uint32_t ip; bool stop; EpollFd pollFd; @@ -69,6 +82,7 @@ typedef struct { } SServerObj; static void *taosProcessTcpData(void *param); +static void *taosProcessServerTcpData(void *param); static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd); static void taosFreeFdObj(SFdObj *pFdObj); static void taosReportBrokenLink(SFdObj *pFdObj); @@ -124,6 +138,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label)); pThreadObj->shandle = shandle; pThreadObj->stop = false; + pThreadObj->connQueue = tdListNew(sizeof(SConnItem)); } // initialize mutex, thread, fd which may fail @@ -142,7 +157,25 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread break; } - code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj)); + int fds[2]; + if (pipe(fds)) { + tError("%s failed to create pipe", label); + code = -1; + break; + } + + pThreadObj->notifyReceiveFd = fds[0]; + pThreadObj->notifySendFd = fds[1]; + struct epoll_event event; + event.events = EPOLLIN | EPOLLRDHUP; + event.data.fd = pThreadObj->notifyReceiveFd; + if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, pThreadObj->notifyReceiveFd , &event) < 0) { + tError("%s failed to create pipe", label); + code = -1; + break; + } + + code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessServerTcpData, (void *)(pThreadObj)); if (code != 0) { tError("%s failed to create TCP process data thread(%s)", label, strerror(errno)); break; @@ -275,17 +308,12 @@ static void *taosAcceptTcpConnection(void *arg) { // pick up the thread to handle this connection pThreadObj = pServerObj->pThreadObj[threadId]; - SFdObj *pFdObj = taosMallocFdObj(pThreadObj, connFd); - if (pFdObj) { - pFdObj->ip = caddr.sin_addr.s_addr; - pFdObj->port = htons(caddr.sin_port); - tDebug("%s new TCP connection from %s:%hu, fd:%d FD:%p numOfFds:%d", pServerObj->label, - taosInetNtoa(caddr.sin_addr), pFdObj->port, connFd, pFdObj, pThreadObj->numOfFds); - } else { - taosCloseSocket(connFd); - tError("%s failed to malloc FdObj(%s) for connection from:%s:%hu", pServerObj->label, strerror(errno), - taosInetNtoa(caddr.sin_addr), htons(caddr.sin_port)); - } + pthread_mutex_lock(&(pThreadObj->mutex)); + SConnItem item = {.fd = connFd, .ip = caddr.sin_addr.s_addr, .port = htons(caddr.sin_port)}; + tdListAppend(pThreadObj->connQueue, &item); + pthread_mutex_unlock(&(pThreadObj->mutex)); + + write(pThreadObj->notifySendFd, "", 1); // pick up next thread for next connection threadId++; @@ -591,6 +619,109 @@ static void *taosProcessTcpData(void *param) { return NULL; } +static void *taosProcessServerTcpData(void *param) { + SThreadObj *pThreadObj = param; + SFdObj *pFdObj; + struct epoll_event events[maxEvents]; + SRecvInfo recvInfo; + + char bb[1]; +#ifdef __APPLE__ + taos_block_sigalrm(); +#endif // __APPLE__ + while (1) { + int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, TAOS_EPOLL_WAIT_TIME); + if (pThreadObj->stop) { + tDebug("%s TCP thread get stop event, exiting...", pThreadObj->label); + break; + } + if (fdNum < 0) continue; + + for (int i = 0; i < fdNum; ++i) { + if (events[i].data.fd == pThreadObj->notifyReceiveFd) { + if (events[i].events & EPOLLIN) { + read(pThreadObj->notifyReceiveFd, bb, 1); + + pthread_mutex_lock(&(pThreadObj->mutex)); + SListNode *head = tdListPopHead(pThreadObj->connQueue); + pthread_mutex_unlock(&(pThreadObj->mutex)); + + SConnItem item = {0}; + tdListNodeGetData(pThreadObj->connQueue, head, &item); + tfree(head); + + // register fd on epoll + SFdObj *pFdObj = taosMallocFdObj(pThreadObj, item.fd); + if (pFdObj) { + pFdObj->ip = item.ip; + pFdObj->port = item.port; + tDebug("%s new TCP connection from %u:%hu, fd:%d FD:%p numOfFds:%d", pThreadObj->label, + pFdObj->ip, pFdObj->port, item.fd, pFdObj, pThreadObj->numOfFds); + } else { + taosCloseSocket(item.fd); + tError("%s failed to malloc FdObj(%s) for connection from:%u:%hu", pThreadObj->label, strerror(errno), + pFdObj->ip, pFdObj->port); + } + } + continue; + } + pFdObj = events[i].data.ptr; + + if (events[i].events & EPOLLERR) { + tDebug("%s %p FD:%p epoll errors", pThreadObj->label, pFdObj->thandle, pFdObj); + taosReportBrokenLink(pFdObj); + continue; + } + + if (events[i].events & EPOLLRDHUP) { + tDebug("%s %p FD:%p RD hang up", pThreadObj->label, pFdObj->thandle, pFdObj); + taosReportBrokenLink(pFdObj); + continue; + } + + if (events[i].events & EPOLLHUP) { + tDebug("%s %p FD:%p hang up", pThreadObj->label, pFdObj->thandle, pFdObj); + taosReportBrokenLink(pFdObj); + continue; + } + + if (taosReadTcpData(pFdObj, &recvInfo) < 0) { + shutdown(pFdObj->fd, SHUT_WR); + continue; + } + + pFdObj->thandle = (*(pThreadObj->processData))(&recvInfo); + if (pFdObj->thandle == NULL) taosFreeFdObj(pFdObj); + } + + if (pThreadObj->stop) break; + } + + if (pThreadObj->connQueue) { + pThreadObj->connQueue = tdListFree(pThreadObj->connQueue); + } + // close pipe + close(pThreadObj->notifySendFd); + close(pThreadObj->notifyReceiveFd); + + if (pThreadObj->pollFd >=0) { + EpollClose(pThreadObj->pollFd); + pThreadObj->pollFd = -1; + } + + while (pThreadObj->pHead) { + SFdObj *pFdObj = pThreadObj->pHead; + pThreadObj->pHead = pFdObj->next; + taosReportBrokenLink(pFdObj); + } + + pthread_mutex_destroy(&(pThreadObj->mutex)); + tDebug("%s TCP thread exits ...", pThreadObj->label); + tfree(pThreadObj); + + return NULL; +} + static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd) { struct epoll_event event; From 57e77b5f416eba4164196d6c8438fd6af119d6ab Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 8 Apr 2021 12:10:51 +0800 Subject: [PATCH 2/6] enlarge default time --- src/rpc/src/rpcMain.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c index 133ae6d0ab..58d611ebb5 100644 --- a/src/rpc/src/rpcMain.c +++ b/src/rpc/src/rpcMain.c @@ -295,7 +295,7 @@ void *rpcOpen(const SRpcInit *pInit) { return NULL; } } else { - pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime); + pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime * 30); if ( pRpc->pCache == NULL ) { tError("%s failed to init connection cache", pRpc->label); rpcClose(pRpc); @@ -470,7 +470,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) { taosTmrStopA(&pConn->pTimer); // set the idle timer to monitor the activity - taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer); + taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime * 30, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer); rpcSendMsgToPeer(pConn, msg, msgLen); // if not set to secured, set it expcet NOT_READY case, since client wont treat it as secured From 3a14b45f40977f6d46f67c17d8019f1a66678aaa Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Fri, 9 Apr 2021 14:56:25 +0800 Subject: [PATCH 3/6] fixbug crash --- src/rpc/src/rpcMain.c | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c index 58d611ebb5..98d5c1ed54 100644 --- a/src/rpc/src/rpcMain.c +++ b/src/rpc/src/rpcMain.c @@ -997,8 +997,8 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont } if ( rpcIsReq(pHead->msgType) ) { - terrno = rpcProcessReqHead(pConn, pHead); pConn->connType = pRecv->connType; + terrno = rpcProcessReqHead(pConn, pHead); // stop idle timer taosTmrStopA(&pConn->pIdleTimer); @@ -1367,7 +1367,8 @@ static void rpcProcessConnError(void *param, void *id) { tDebug("%s %p, connection error happens", pRpc->label, pContext->ahandle); - if (pContext->numOfTry >= pContext->epSet.numOfEps) { + if (pContext->numOfTry >= pContext->epSet.numOfEps + || pContex->msgType == TSDB_MSG_TYPE_FETCH) { rpcMsg.msgType = pContext->msgType+1; rpcMsg.ahandle = pContext->ahandle; rpcMsg.code = pContext->code; From 1511e0c32f13255564836f9d9b9e48255b8b615f Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Fri, 9 Apr 2021 15:01:29 +0800 Subject: [PATCH 4/6] fixbug crash --- src/rpc/src/rpcMain.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c index 98d5c1ed54..3e8c1d9180 100644 --- a/src/rpc/src/rpcMain.c +++ b/src/rpc/src/rpcMain.c @@ -1368,7 +1368,7 @@ static void rpcProcessConnError(void *param, void *id) { tDebug("%s %p, connection error happens", pRpc->label, pContext->ahandle); if (pContext->numOfTry >= pContext->epSet.numOfEps - || pContex->msgType == TSDB_MSG_TYPE_FETCH) { + || pContext->msgType == TSDB_MSG_TYPE_FETCH) { rpcMsg.msgType = pContext->msgType+1; rpcMsg.ahandle = pContext->ahandle; rpcMsg.code = pContext->code; From f5db431c19973e56a11c1e644e521634a7d83fb2 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Sat, 10 Apr 2021 17:11:58 +0800 Subject: [PATCH 5/6] reset timeout --- src/rpc/src/rpcMain.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c index 7205bafd7d..cd6b3cf80c 100644 --- a/src/rpc/src/rpcMain.c +++ b/src/rpc/src/rpcMain.c @@ -295,7 +295,7 @@ void *rpcOpen(const SRpcInit *pInit) { return NULL; } } else { - pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime*30); + pRpc->pCache = rpcOpenConnCache(pRpc->sessions, rpcCloseConn, pRpc->tmrCtrl, pRpc->idleTime); if ( pRpc->pCache == NULL ) { tError("%s failed to init connection cache", pRpc->label); rpcClose(pRpc); @@ -470,7 +470,7 @@ void rpcSendResponse(const SRpcMsg *pRsp) { taosTmrStopA(&pConn->pTimer); // set the idle timer to monitor the activity - taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime*30, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer); + taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer); rpcSendMsgToPeer(pConn, msg, msgLen); // if not set to secured, set it expcet NOT_READY case, since client wont treat it as secured From 35c7506dd096edd534c5b93eef4b4afc6ac647cc Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Sat, 10 Apr 2021 22:48:35 +0800 Subject: [PATCH 6/6] reset tcp --- src/rpc/src/rpcTcp.c | 155 ++++--------------------------------------- 1 file changed, 12 insertions(+), 143 deletions(-) diff --git a/src/rpc/src/rpcTcp.c b/src/rpc/src/rpcTcp.c index 286ed223c7..3162ab2e4c 100644 --- a/src/rpc/src/rpcTcp.c +++ b/src/rpc/src/rpcTcp.c @@ -21,13 +21,6 @@ #include "rpcLog.h" #include "rpcHead.h" #include "rpcTcp.h" -#include "tlist.h" - -typedef struct SConnItem { - SOCKET fd; - uint32_t ip; - uint16_t port; -} SConnItem; typedef struct SFdObj { void *signature; @@ -45,12 +38,6 @@ typedef struct SThreadObj { pthread_t thread; SFdObj * pHead; pthread_mutex_t mutex; - // receive the notify from dispatch thread - - int notifyReceiveFd; - int notifySendFd; - SList *connQueue; - uint32_t ip; bool stop; EpollFd pollFd; @@ -82,7 +69,6 @@ typedef struct { } SServerObj; static void *taosProcessTcpData(void *param); -static void *taosProcessServerTcpData(void *param); static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd); static void taosFreeFdObj(SFdObj *pFdObj); static void taosReportBrokenLink(SFdObj *pFdObj); @@ -138,7 +124,6 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label)); pThreadObj->shandle = shandle; pThreadObj->stop = false; - pThreadObj->connQueue = tdListNew(sizeof(SConnItem)); } // initialize mutex, thread, fd which may fail @@ -157,25 +142,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread break; } - int fds[2]; - if (pipe(fds)) { - tError("%s failed to create pipe", label); - code = -1; - break; - } - - pThreadObj->notifyReceiveFd = fds[0]; - pThreadObj->notifySendFd = fds[1]; - struct epoll_event event; - event.events = EPOLLIN | EPOLLRDHUP; - event.data.fd = pThreadObj->notifyReceiveFd; - if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, pThreadObj->notifyReceiveFd , &event) < 0) { - tError("%s failed to create pipe", label); - code = -1; - break; - } - - code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessServerTcpData, (void *)(pThreadObj)); + code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj)); if (code != 0) { tError("%s failed to create TCP process data thread(%s)", label, strerror(errno)); break; @@ -308,12 +275,17 @@ static void *taosAcceptTcpConnection(void *arg) { // pick up the thread to handle this connection pThreadObj = pServerObj->pThreadObj[threadId]; - pthread_mutex_lock(&(pThreadObj->mutex)); - SConnItem item = {.fd = connFd, .ip = caddr.sin_addr.s_addr, .port = htons(caddr.sin_port)}; - tdListAppend(pThreadObj->connQueue, &item); - pthread_mutex_unlock(&(pThreadObj->mutex)); - - write(pThreadObj->notifySendFd, "", 1); + SFdObj *pFdObj = taosMallocFdObj(pThreadObj, connFd); + if (pFdObj) { + pFdObj->ip = caddr.sin_addr.s_addr; + pFdObj->port = htons(caddr.sin_port); + tDebug("%s new TCP connection from %s:%hu, fd:%d FD:%p numOfFds:%d", pServerObj->label, + taosInetNtoa(caddr.sin_addr), pFdObj->port, connFd, pFdObj, pThreadObj->numOfFds); + } else { + taosCloseSocket(connFd); + tError("%s failed to malloc FdObj(%s) for connection from:%s:%hu", pServerObj->label, strerror(errno), + taosInetNtoa(caddr.sin_addr), htons(caddr.sin_port)); + } // pick up next thread for next connection threadId++; @@ -619,109 +591,6 @@ static void *taosProcessTcpData(void *param) { return NULL; } -static void *taosProcessServerTcpData(void *param) { - SThreadObj *pThreadObj = param; - SFdObj *pFdObj; - struct epoll_event events[maxEvents]; - SRecvInfo recvInfo; - - char bb[1]; -#ifdef __APPLE__ - taos_block_sigalrm(); -#endif // __APPLE__ - while (1) { - int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, TAOS_EPOLL_WAIT_TIME); - if (pThreadObj->stop) { - tDebug("%s TCP thread get stop event, exiting...", pThreadObj->label); - break; - } - if (fdNum < 0) continue; - - for (int i = 0; i < fdNum; ++i) { - if (events[i].data.fd == pThreadObj->notifyReceiveFd) { - if (events[i].events & EPOLLIN) { - read(pThreadObj->notifyReceiveFd, bb, 1); - - pthread_mutex_lock(&(pThreadObj->mutex)); - SListNode *head = tdListPopHead(pThreadObj->connQueue); - pthread_mutex_unlock(&(pThreadObj->mutex)); - - SConnItem item = {0}; - tdListNodeGetData(pThreadObj->connQueue, head, &item); - tfree(head); - - // register fd on epoll - SFdObj *pFdObj = taosMallocFdObj(pThreadObj, item.fd); - if (pFdObj) { - pFdObj->ip = item.ip; - pFdObj->port = item.port; - tDebug("%s new TCP connection from %u:%hu, fd:%d FD:%p numOfFds:%d", pThreadObj->label, - pFdObj->ip, pFdObj->port, item.fd, pFdObj, pThreadObj->numOfFds); - } else { - taosCloseSocket(item.fd); - tError("%s failed to malloc FdObj(%s) for connection from:%u:%hu", pThreadObj->label, strerror(errno), - pFdObj->ip, pFdObj->port); - } - } - continue; - } - pFdObj = events[i].data.ptr; - - if (events[i].events & EPOLLERR) { - tDebug("%s %p FD:%p epoll errors", pThreadObj->label, pFdObj->thandle, pFdObj); - taosReportBrokenLink(pFdObj); - continue; - } - - if (events[i].events & EPOLLRDHUP) { - tDebug("%s %p FD:%p RD hang up", pThreadObj->label, pFdObj->thandle, pFdObj); - taosReportBrokenLink(pFdObj); - continue; - } - - if (events[i].events & EPOLLHUP) { - tDebug("%s %p FD:%p hang up", pThreadObj->label, pFdObj->thandle, pFdObj); - taosReportBrokenLink(pFdObj); - continue; - } - - if (taosReadTcpData(pFdObj, &recvInfo) < 0) { - shutdown(pFdObj->fd, SHUT_WR); - continue; - } - - pFdObj->thandle = (*(pThreadObj->processData))(&recvInfo); - if (pFdObj->thandle == NULL) taosFreeFdObj(pFdObj); - } - - if (pThreadObj->stop) break; - } - - if (pThreadObj->connQueue) { - pThreadObj->connQueue = tdListFree(pThreadObj->connQueue); - } - // close pipe - close(pThreadObj->notifySendFd); - close(pThreadObj->notifyReceiveFd); - - if (pThreadObj->pollFd >=0) { - EpollClose(pThreadObj->pollFd); - pThreadObj->pollFd = -1; - } - - while (pThreadObj->pHead) { - SFdObj *pFdObj = pThreadObj->pHead; - pThreadObj->pHead = pFdObj->next; - taosReportBrokenLink(pFdObj); - } - - pthread_mutex_destroy(&(pThreadObj->mutex)); - tDebug("%s TCP thread exits ...", pThreadObj->label); - tfree(pThreadObj); - - return NULL; -} - static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd) { struct epoll_event event;