Merge pull request #2009 from taosdata/hotfix/td-364

Hotfix/td 364

src/rpc/src/rpcMain.c

@@ -867,9 +867,8 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
   // underlying UDP layer does not know it is server or client
   pRecv->connType = pRecv->connType | pRpc->connType;  
 
-  if (pRecv->ip==0 && pConn) {
-    rpcProcessBrokenLink(pConn); 
-    rpcFreeMsg(pRecv->msg);
+  if (pRecv->ip == 0 && pConn) {
+    rpcProcessBrokenLink(pConn);
     return NULL;
   }
 

src/rpc/src/rpcTcp.c

@@ -215,7 +215,6 @@ static void* taosAcceptTcpConnection(void *arg) {
       continue;
     }
 
-    tTrace("%s TCP connection from ip:%s:%hu", pServerObj->label, inet_ntoa(caddr.sin_addr), caddr.sin_port);
     taosKeepTcpAlive(connFd);
 
     // pick up the thread to handle this connection
@@ -229,7 +228,8 @@ static void* taosAcceptTcpConnection(void *arg) {
               inet_ntoa(caddr.sin_addr), pFdObj->port, pFdObj, pThreadObj->numOfFds);
     } else {
       close(connFd);
-      tError("%s failed to malloc FdObj(%s)", pServerObj->label, strerror(errno));
+      tError("%s failed to malloc FdObj(%s) for connection from:%s:%hu", pServerObj->label, strerror(errno),
+             inet_ntoa(caddr.sin_addr), caddr.sin_port);
     }  
 
     // pick up next thread for next connection
@@ -341,7 +341,9 @@ static void taosReportBrokenLink(SFdObj *pFdObj) {
     recvInfo.chandle = NULL;
     recvInfo.connType = RPC_CONN_TCP;
     (*(pThreadObj->processData))(&recvInfo);
-  } 
+  } else {
+    taosFreeFdObj(pFdObj);
+  }
 }
 
 #define maxEvents 10
@@ -352,7 +354,7 @@ static void *taosProcessTcpData(void *param) {
   struct epoll_event events[maxEvents];
   SRecvInfo          recvInfo;
   SRpcHead           rpcHead;
-
+ 
   while (1) {
     int fdNum = epoll_wait(pThreadObj->pollFd, events, maxEvents, -1);
     if (pThreadObj->stop) {
@@ -466,7 +468,7 @@ static void taosFreeFdObj(SFdObj *pFdObj) {
 
   pFdObj->signature = NULL;
   epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_DEL, pFdObj->fd, NULL);
-  close(pFdObj->fd);
+  taosCloseTcpSocket(pFdObj->fd);
 
   pThreadObj->numOfFds--;
 

src/rpc/test/rserver.c

@@ -127,6 +127,8 @@ int main(int argc, char *argv[]) {
   SRpcInit rpcInit;
   char     dataName[20] = "server.data";
 
+  taosBlockSIGPIPE();
+
   memset(&rpcInit, 0, sizeof(rpcInit));
   rpcInit.localPort    = 7000;
   rpcInit.label        = "SER";