tune the code

src/rpc/inc/rpcTcp.h

@@ -25,6 +25,7 @@ void taosStopTcpServer(void *param);
 void taosCleanUpTcpServer(void *param);
 
 void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *fp, void *shandle);
+void taosStopTcpClient(void *chandle);
 void taosCleanUpTcpClient(void *chandle);
 void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port);
 

src/rpc/inc/rpcUdp.h

@@ -23,6 +23,7 @@ extern "C" {
 #include "taosdef.h"
 
 void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int, void *fp, void *shandle);
+void  taosStopUdpConnection(void *handle);
 void  taosCleanUpUdpConnection(void *handle);
 int   taosSendUdpData(uint32_t ip, uint16_t port, void *data, int dataLen, void *chandle);
 void *taosOpenUdpConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port);

src/rpc/src/rpcMain.c

@@ -147,17 +147,17 @@ void *(*taosInitConn[])(uint32_t ip, uint16_t port, char *label, int threads, vo
 };
 
 void (*taosCleanUpConn[])(void *thandle) = {
-    NULL, 
-    NULL,
+    taosCleanUpUdpConnection, 
+    taosCleanUpUdpConnection, 
     taosCleanUpTcpServer,
     taosCleanUpTcpClient
 };
 
 void (*taosStopConn[])(void *thandle) = {
-    taosCleanUpUdpConnection, 
-    taosCleanUpUdpConnection, 
+    taosStopUdpConnection, 
+    taosStopUdpConnection, 
     taosStopTcpServer,
-    NULL 
+    taosStopTcpClient,
 };
 
 int (*taosSendData[])(uint32_t ip, uint16_t port, void *data, int len, void *chandle) = {
@@ -297,11 +297,8 @@ void rpcClose(void *param) {
   SRpcInfo *pRpc = (SRpcInfo *)param;
 
   // stop connection to outside first
-  if (taosStopConn[pRpc->connType | RPC_CONN_TCP])
-    (*taosStopConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
-
-  if (taosStopConn[pRpc->connType])
-    (*taosStopConn[pRpc->connType])(pRpc->udphandle);
+  (*taosStopConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
+  (*taosStopConn[pRpc->connType])(pRpc->udphandle);
 
   // close all connections 
   for (int i = 0; i < pRpc->sessions; ++i) {
@@ -311,11 +308,8 @@ void rpcClose(void *param) {
   }
 
   // clean up
-  if (taosCleanUpConn[pRpc->connType | RPC_CONN_TCP])
-    (*taosCleanUpConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
-
-  if (taosCleanUpConn[pRpc->connType])
-    (*taosCleanUpConn[pRpc->connType])(pRpc->udphandle);
+  (*taosCleanUpConn[pRpc->connType | RPC_CONN_TCP])(pRpc->tcphandle);
+  (*taosCleanUpConn[pRpc->connType])(pRpc->udphandle);
 
   tTrace("%s rpc is closed", pRpc->label);
   rpcDecRef(pRpc);

src/rpc/src/rpcTcp.c

@@ -193,10 +193,11 @@ static void taosStopTcpThread(SThreadObj* pThreadObj) {
 void taosStopTcpServer(void *handle) {
   SServerObj *pServerObj = handle;
 
-  tTrace("TCP:%s, stop accept new connections", pServerObj->label);
   if (pServerObj == NULL) return;
   if(pServerObj->fd >=0) shutdown(pServerObj->fd, SHUT_RD);
   if(pServerObj->thread) pthread_join(pServerObj->thread, NULL);
+
+  tTrace("%s TCP server is stopped", pServerObj->label);
 }
 
 void taosCleanUpTcpServer(void *handle) {
@@ -210,7 +211,7 @@ void taosCleanUpTcpServer(void *handle) {
     pthread_mutex_destroy(&(pThreadObj->mutex));
   }
 
-  tTrace("TCP:%s, TCP server is cleaned up", pServerObj->label);
+  tTrace("%s TCP server is cleaned up", pServerObj->label);
 
   tfree(pServerObj->pThreadObj);
   tfree(pServerObj);
@@ -309,12 +310,19 @@ void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *
   return pThreadObj;
 }
 
+void taosStopTcpClient(void *chandle) {
+  SThreadObj *pThreadObj = chandle;
+  if (pThreadObj == NULL) return;
+
+  tTrace ("%s TCP client is stopped", pThreadObj->label);
+}
+
 void taosCleanUpTcpClient(void *chandle) {
   SThreadObj *pThreadObj = chandle;
   if (pThreadObj == NULL) return;
 
   taosStopTcpThread(pThreadObj);
-  tTrace ("%s, all connections are cleaned up", pThreadObj->label);
+  tTrace ("%s TCP client is cleaned up", pThreadObj->label);
 
   tfree(pThreadObj);
 }

src/rpc/src/rpcUdp.c

@@ -130,7 +130,7 @@ void *taosInitUdpConnection(uint32_t ip, uint16_t port, char *label, int threads
   return pSet;
 }
 
-void taosCleanUpUdpConnection(void *handle) {
+void taosStopUdpConnection(void *handle) {
   SUdpConnSet *pSet = (SUdpConnSet *)handle;
   SUdpConn    *pConn;
 
@@ -146,9 +146,24 @@ void taosCleanUpUdpConnection(void *handle) {
     pConn = pSet->udpConn + i;
     if (pConn->thread) pthread_join(pConn->thread, NULL);
     tfree(pConn->buffer);
-    tTrace("%s UDP thread is closed, inedx:%d", pConn->label, i);
+    // tTrace("%s UDP thread is closed, index:%d", pConn->label, i);
+  }
+
+  tTrace("%s UDP is stopped", pSet->label);
+}
+
+void taosCleanUpUdpConnection(void *handle) {
+  SUdpConnSet *pSet = (SUdpConnSet *)handle;
+  SUdpConn    *pConn;
+
+  if (pSet == NULL) return;
+
+  for (int i = 0; i < pSet->threads; ++i) {
+    pConn = pSet->udpConn + i;
+    if (pConn->fd >=0) taosCloseSocket(pConn->fd);
   }
 
+  tTrace("%s UDP is cleaned up", pSet->label);
   tfree(pSet);
 }