xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #2386 from taosdata/hotfix/rpcUdp 

close UDP thread
This commit is contained in:
Shengliang Guan 2020-06-20 16:54:05 +08:00 committed by GitHub
parent 9c19a70f67 de8c9fc46f
commit cec2369c57
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 7 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/rpc/src/rpcTcp.c
View File

@ -437,7 +437,7 @@ static void *taosProcessTcpData(void *param) {
while (1) { while (1) {
int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, -1); int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, -1);
if (pThreadObj->stop) { if (pThreadObj->stop) {
tTrace("%s, tcp thread get stop event, exiting...", pThreadObj->label); tTrace("%s TCP thread get stop event, exiting...", pThreadObj->label);
break; break;
} }
if (fdNum < 0) continue; if (fdNum < 0) continue;

13
src/rpc/src/rpcUdp.c
View File

@ -142,16 +142,15 @@ void taosCleanUpUdpConnection(void *handle) {
pConn = pSet->udpConn + i; pConn = pSet->udpConn + i;
pConn->signature = NULL; pConn->signature = NULL;
// shutdown to signal the thread to exit if (pConn->fd >=0) shutdown(pConn->fd, SHUT_RDWR);
if ( pConn->fd >=0) shutdown(pConn->fd, SHUT_RD); if (pConn->fd >=0) taosCloseSocket(pConn->fd);
} }
for (int i = 0; i < pSet->threads; ++i) { for (int i = 0; i < pSet->threads; ++i) {
pConn = pSet->udpConn + i; pConn = pSet->udpConn + i;
if (pConn->thread) pthread_join(pConn->thread, NULL); if (pConn->thread) pthread_join(pConn->thread, NULL);
if (pConn->fd >=0) taosCloseSocket(pConn->fd);
tfree(pConn->buffer); tfree(pConn->buffer);
tTrace("UDP chandle:%p is closed", pConn); tTrace("%s UDP thread is closed, inedx:%d", pConn->label, i);
} }
tfree(pSet); tfree(pSet);
@ -185,15 +184,15 @@ static void *taosRecvUdpData(void *param) {
while (1) { while (1) {
dataLen = recvfrom(pConn->fd, pConn->buffer, RPC_MAX_UDP_SIZE, 0, (struct sockaddr *)&sourceAdd, &addLen); dataLen = recvfrom(pConn->fd, pConn->buffer, RPC_MAX_UDP_SIZE, 0, (struct sockaddr *)&sourceAdd, &addLen);
if(dataLen == 0) { if(dataLen <= 0) {
tTrace("data length is 0, socket was closed, exiting"); tTrace("%s UDP socket was closed, exiting", pConn->label);
break; break;
} }
port = ntohs(sourceAdd.sin_port); port = ntohs(sourceAdd.sin_port);
if (dataLen < sizeof(SRpcHead)) { if (dataLen < sizeof(SRpcHead)) {
tError("%s recvfrom failed, reason:%s\n", pConn->label, strerror(errno)); tError("%s recvfrom failed(%s)", pConn->label, strerror(errno));
continue; continue;
} }