change the API rpcFreeCont implementation, so app can release the request message

fix a few minor bugs

src/inc/trpc.h

@@ -27,6 +27,7 @@ extern "C" {
 #define TAOS_CONN_CLIENT   1
 
 extern int tsRpcHeadSize;
+extern int tsRpcMaxUdpSize;
 
 typedef struct {
   int8_t    inUse; 

src/rpc/inc/rpcHead.h

@@ -26,6 +26,8 @@ extern "C" {
 #define RPC_CONN_TCPC   3
 #define RPC_CONN_TCP    2
 
+extern int tsRpcOverhead;
+
 typedef struct {
   void    *msg;
   int      msgLen;

src/rpc/src/rpcClient.c

@@ -236,6 +236,7 @@ static void *taosReadTcpData(void *param) {
   STcpFd            *pFdObj;
   struct epoll_event events[maxTcpEvents];
   SRecvInfo          recvInfo;
+  SRpcHead           rpcHead;
 
   while (1) {
     pthread_mutex_lock(&pTcp->mutex);
@@ -260,37 +261,24 @@ static void *taosReadTcpData(void *param) {
         continue;
       }
 
-      void *buffer = malloc(1024);
-      if (NULL == buffer) {
-        tTrace("%s TCP malloc(size:1024) fail\n", pTcp->label);
-        taosCleanUpTcpFdObj(pFdObj);
-        continue;
-      }
-
-      int headLen = taosReadMsg(pFdObj->fd, buffer, sizeof(SRpcHead));
+      int headLen = taosReadMsg(pFdObj->fd, &rpcHead, sizeof(SRpcHead));
       if (headLen != sizeof(SRpcHead)) {
         tError("%s read error, headLen:%d", pTcp->label, headLen);
-        tfree(buffer);
         taosCleanUpTcpFdObj(pFdObj);
         continue;
       }
 
-      int dataLen = (int32_t)htonl((uint32_t)((SRpcHead *)buffer)->msgLen);
-      if (dataLen > 1024) {
-        void *b = realloc(buffer, (size_t)dataLen);
-        if (NULL == b) {
-          tTrace("%s TCP malloc(size:%d) fail\n", pTcp->label, dataLen);
-          tfree(buffer);
-          taosCleanUpTcpFdObj(pFdObj);
-          continue;
-        }
-        buffer = b;
+      int32_t msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen);
+      char   *buffer = (char *)malloc((size_t)msgLen + tsRpcOverhead);
+      if (NULL == buffer) {
+        tTrace("%s TCP malloc(size:%d) fail\n", pTcp->label, msgLen);
+        taosCleanUpTcpFdObj(pFdObj);
+        continue;
       }
 
-      int leftLen = dataLen - headLen;
-      int retLen = taosReadMsg(pFdObj->fd, buffer + headLen, leftLen);
-
-      // tTrace("%s TCP data is received, ip:%s port:%u len:%d", pTcp->label, pFdObj->ipstr, pFdObj->port, dataLen);
+      char    *msg = buffer + tsRpcOverhead;
+      int32_t  leftLen = msgLen - headLen;
+      int32_t  retLen = taosReadMsg(pFdObj->fd, msg + headLen, leftLen);
 
       if (leftLen != retLen) {
         tError("%s read error, leftLen:%d retLen:%d", pTcp->label, leftLen, retLen);
@@ -299,8 +287,11 @@ static void *taosReadTcpData(void *param) {
         continue;
       }
 
-      recvInfo.msg = buffer;
-      recvInfo.msgLen = dataLen;
+      // tTrace("%s TCP data is received, ip:%s:%u len:%d", pTcp->label, pFdObj->ipstr, pFdObj->port, msgLen);
+
+      memcpy(msg, &rpcHead, sizeof(SRpcHead));
+      recvInfo.msg = msg;
+      recvInfo.msgLen = msgLen;
       recvInfo.ip = pFdObj->ip;
       recvInfo.port = pFdObj->port;
       recvInfo.shandle = pTcp->shandle;

src/rpc/src/rpcMain.c

@@ -122,10 +122,12 @@ typedef struct _RpcConn {
 } SRpcConn;
 
 int tsRpcProgressTime = 10;  // milliseocnds
+int tsRpcMaxUdpSize = 15000; // bytes;
 
 // not configurable
 int tsRpcMaxRetry;
 int tsRpcHeadSize;
+int tsRpcOverhead;
 
 // server:0 client:1  tcp:2 udp:0
 #define RPC_CONN_UDPS   0
@@ -188,7 +190,7 @@ static void  rpcProcessRetryTimer(void *, void *);
 static void  rpcProcessIdleTimer(void *param, void *tmrId);
 static void  rpcProcessProgressTimer(void *param, void *tmrId);
 
-static void  rpcFreeOutMsg(void *msg);
+static void  rpcFreeMsg(void *msg);
 static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen);
 static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead);
 static int   rpcAddAuthPart(SRpcConn *pConn, char *msg, int msgLen);
@@ -201,6 +203,7 @@ void *rpcOpen(SRpcInit *pInit) {
 
   tsRpcMaxRetry = tsRpcMaxTime * 1000 / tsRpcProgressTime;
   tsRpcHeadSize = RPC_MSG_OVERHEAD; 
+  tsRpcOverhead = sizeof(SRpcReqContext);
 
   pRpc = (SRpcInfo *)calloc(1, sizeof(SRpcInfo));
   if (pRpc == NULL) return NULL;
@@ -313,8 +316,8 @@ void *rpcMallocCont(int contLen) {
 
 void rpcFreeCont(void *cont) {
   if ( cont ) {
-    char *msg = ((char *)cont) - sizeof(SRpcHead);
-    free(msg);
+    char *temp = ((char *)cont) - sizeof(SRpcHead) - sizeof(SRpcReqContext);
+    free(temp);
   }
 }
 
@@ -351,7 +354,7 @@ void rpcSendRequest(void *shandle, SRpcIpSet *pIpSet, char type, void *pCont, in
   pContext->oldInUse = pIpSet->inUse;
 
   pContext->connType = RPC_CONN_UDPC; 
-  if (contLen > 16000) pContext->connType = RPC_CONN_TCPC;
+  if (contLen > tsRpcMaxUdpSize) pContext->connType = RPC_CONN_TCPC;
 
   // connection type is application specific. 
   // for TDengine, all the query, show commands shall have TCP connection
@@ -406,7 +409,7 @@ void rpcSendResponse(void *handle, int32_t code, void *pCont, int contLen) {
   pConn->inType = 0;
 
   // response message is released until new response is sent
-  rpcFreeOutMsg(pConn->pRspMsg); 
+  rpcFreeMsg(pConn->pRspMsg); 
   pConn->pRspMsg = msg;
   pConn->rspMsgLen = msgLen;
   if (pHead->content[0] == TSDB_CODE_ACTION_IN_PROGRESS) pConn->inTranId--;
@@ -442,6 +445,13 @@ void rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
   strcpy(pInfo->user, pConn->user);
 }
 
+static void rpcFreeMsg(void *msg) {
+  if ( msg ) {
+    char *temp = (char *)msg - sizeof(SRpcReqContext);
+    free(temp);
+  }
+}
+
 static SRpcConn *rpcOpenConn(SRpcInfo *pRpc, char *peerIpStr, uint16_t peerPort, int8_t connType) {
   SRpcConn *pConn;
 
@@ -490,7 +500,7 @@ static void rpcCloseConn(void *thandle) {
       char hashstr[40] = {0};
       sprintf(hashstr, "%x:%x:%x:%d", pConn->peerIp, pConn->peerUid, pConn->peerId, pConn->connType);
       taosDeleteStrHash(pRpc->hash, hashstr);
-      rpcFreeOutMsg(pConn->pRspMsg); // it may have a response msg saved, but not request msg
+      rpcFreeMsg(pConn->pRspMsg); // it may have a response msg saved, but not request msg
       pConn->pRspMsg = NULL;
       pConn->inType = 0;
       pConn->inTranId = 0;
@@ -822,7 +832,7 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
     }
   }
 
-  if ( terrno ) free (pRecv->msg);
+  if (terrno) rpcFreeMsg(pRecv->msg);
   return pConn;
 }
 
@@ -855,7 +865,7 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead) {
       if ( pRpc->ufp && (pContext->ipSet.inUse != pContext->oldInUse || pContext->redirect) ) 
         (*pRpc->ufp)(pContext->ahandle, &pContext->ipSet);  // notify the update of ipSet
       (*pRpc->cfp)(pHead->msgType, pCont, contLen, pContext->ahandle, code);
-      rpcFreeOutMsg(rpcHeadFromCont(pContext->pCont)); // free the request msg
+      rpcFreeCont(pContext->pCont); // free the request msg
     }
   }
 }
@@ -996,8 +1006,8 @@ static void rpcProcessConnError(void *param, void *id) {
   tTrace("%s connection error happens", pRpc->label);
 
   if ( pContext->numOfTry >= pContext->ipSet.numOfIps ) {
-    rpcFreeOutMsg(rpcHeadFromCont(pContext->pCont)); // free the request msg
     (*(pRpc->cfp))(pContext->msgType+1, NULL, 0, pContext->ahandle, pContext->code);  
+    rpcFreeCont(pContext->pCont); // free the request msg
   } else {
     // move to next IP 
     pContext->ipSet.inUse++;
@@ -1127,7 +1137,8 @@ static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead) {
     int contLen = htonl(pComp->contLen);
   
     // prepare the temporary buffer to decompress message
-    pNewHead = (SRpcHead *)malloc(contLen + RPC_MSG_OVERHEAD);
+    char *temp = (char *)malloc(contLen + RPC_MSG_OVERHEAD);
+    pNewHead = (SRpcHead *)(temp + sizeof(SRpcReqContext)); // reserve SRpcReqContext
   
     if (pNewHead) {
       int compLen = rpcContLenFromMsg(pHead->msgLen) - overhead;
@@ -1136,7 +1147,7 @@ static SRpcHead *rpcDecompressRpcMsg(SRpcHead *pHead) {
     
       memcpy(pNewHead, pHead, sizeof(SRpcHead));
       pNewHead->msgLen = rpcMsgLenFromCont(origLen);
-      free(pHead); // free the compressed message buffer
+      rpcFreeMsg(pHead); // free the compressed message buffer
       pHead = pNewHead; 
       tTrace("decompress rpc msg, compLen:%d, after:%d", compLen, contLen);
     } else {

src/rpc/src/rpcServer.c

@@ -191,8 +191,9 @@ static void taosProcessTcpData(void *param) {
   int                i, fdNum;
   SFdObj *           pFdObj;
   struct epoll_event events[maxEvents];
-  SRecvInfo         recvInfo;
+  SRecvInfo          recvInfo;
   pThreadObj = (SThreadObj *)param;
+  SRpcHead           rpcHead;
 
   while (1) {
     pthread_mutex_lock(&pThreadObj->threadMutex);
@@ -219,24 +220,24 @@ static void taosProcessTcpData(void *param) {
         continue;
       }
 
-      void *buffer = malloc(1024);
-      int   headLen = taosReadMsg(pFdObj->fd, buffer, sizeof(SRpcHead));
-
+      int32_t headLen = taosReadMsg(pFdObj->fd, &rpcHead, sizeof(SRpcHead));
       if (headLen != sizeof(SRpcHead)) {
         tError("%s read error, headLen:%d, errno:%d", pThreadObj->label, headLen, errno);
         taosCleanUpFdObj(pFdObj);
-        tfree(buffer);
         continue;
       }
 
-      int dataLen = (int32_t)htonl((uint32_t)((SRpcHead *)buffer)->msgLen);
-      if (dataLen > 1024) buffer = realloc(buffer, (size_t)dataLen);
-
-      int leftLen = dataLen - headLen;
-      int retLen = taosReadMsg(pFdObj->fd, buffer + headLen, leftLen);
+      int32_t msgLen = (int32_t)htonl((uint32_t)rpcHead.msgLen);
+      char   *buffer = malloc(msgLen + tsRpcOverhead);
+      if ( NULL == buffer) {
+        tError("%s TCP malloc(size:%d) fail\n", pThreadObj->label, msgLen);
+        taosCleanUpFdObj(pFdObj);
+        continue;
+      }
 
-      // tTrace("%s TCP data is received, ip:%s port:%u len:%d",
-      // pThreadObj->label, pFdObj->ipstr, pFdObj->port, dataLen);
+      char   *msg = buffer + tsRpcOverhead;
+      int32_t leftLen = msgLen - headLen;
+      int32_t retLen = taosReadMsg(pFdObj->fd, msg + headLen, leftLen);
 
       if (leftLen != retLen) {
         tError("%s read error, leftLen:%d retLen:%d", pThreadObj->label, leftLen, retLen);
@@ -245,8 +246,11 @@ static void taosProcessTcpData(void *param) {
         continue;
       }
 
-      recvInfo.msg = buffer;
-      recvInfo.msgLen = dataLen;
+      // tTrace("%s TCP data is received, ip:%s:%u len:%d", pTcp->label, pFdObj->ipstr, pFdObj->port, msgLen);
+
+      memcpy(msg, &rpcHead, sizeof(SRpcHead));
+      recvInfo.msg = msg;
+      recvInfo.msgLen = msgLen;
       recvInfo.ip = pFdObj->ip;
       recvInfo.port = pFdObj->port;
       recvInfo.shandle = pThreadObj->shandle;

src/rpc/src/rpcUdp.c

@@ -40,12 +40,12 @@ typedef struct {
   char            label[12];  // copy from udpConnSet;
   pthread_t       thread;
   pthread_mutex_t mutex;
-  void *          tmrCtrl;  // copy from UdpConnSet;
-  void *          hash;
-  void *          shandle;  // handle passed by upper layer during server initialization
-  void *          pSet;
-  void            *(*processData)(SRecvInfo *pRecv);
-  char            buffer[RPC_MAX_UDP_SIZE];  // buffer to receive data
+  void           *tmrCtrl;  // copy from UdpConnSet;
+  void           *hash;
+  void           *shandle;  // handle passed by upper layer during server initialization
+  void           *pSet;
+  void         *(*processData)(SRecvInfo *pRecv);
+  char           *buffer;  // buffer to receive data
 } SUdpConn;
 
 typedef struct {
@@ -96,15 +96,15 @@ void *taosInitUdpConnection(char *ip, uint16_t port, char *label, int threads, v
   pSet->fp = fp;
   strcpy(pSet->label, label);
 
-  //  if ( tsUdpDelay ) {
-  char udplabel[12];
-  sprintf(udplabel, "%s.b", label);
-  pSet->tmrCtrl = taosTmrInit(RPC_MAX_UDP_CONNS * threads, 5, 5000, udplabel);
-  if (pSet->tmrCtrl == NULL) {
-    tError("%s failed to initialize tmrCtrl") taosCleanUpUdpConnection(pSet);
-    return NULL;
+  if ( tsUdpDelay ) {
+    char udplabel[12];
+    sprintf(udplabel, "%s.b", label);
+    pSet->tmrCtrl = taosTmrInit(RPC_MAX_UDP_CONNS * threads, 5, 5000, udplabel);
+    if (pSet->tmrCtrl == NULL) {
+      tError("%s failed to initialize tmrCtrl") taosCleanUpUdpConnection(pSet);
+      return NULL;
+    }
   }
-  //  }
 
   pthread_attr_init(&thAttr);
   pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
@@ -120,6 +120,13 @@ void *taosInitUdpConnection(char *ip, uint16_t port, char *label, int threads, v
       return NULL;
     }
 
+    pConn->buffer = malloc(RPC_MAX_UDP_SIZE);
+    if (NULL == pConn->buffer) {
+      tError("%s failed to malloc recv buffer", label);
+      taosCleanUpUdpConnection(pSet);
+      return NULL;
+    }
+
     struct sockaddr_in sin;
     unsigned int       addrlen = sizeof(sin);
     if (getsockname(pConn->fd, (struct sockaddr *)&sin, &addrlen) == 0 && sin.sin_family == AF_INET &&
@@ -128,14 +135,6 @@ void *taosInitUdpConnection(char *ip, uint16_t port, char *label, int threads, v
     }
 
     strcpy(pConn->label, label);
-
-    if (pthread_create(&pConn->thread, &thAttr, taosRecvUdpData, pConn) != 0) {
-      tError("%s failed to create thread to process UDP data, reason:%s", label, strerror(errno));
-      taosCloseSocket(pConn->fd);
-      taosCleanUpUdpConnection(pSet);
-      return NULL;
-    }
-
     pConn->shandle = shandle;
     pConn->processData = fp;
     pConn->index = i;
@@ -146,6 +145,14 @@ void *taosInitUdpConnection(char *ip, uint16_t port, char *label, int threads, v
       pthread_mutex_init(&pConn->mutex, NULL);
       pConn->tmrCtrl = pSet->tmrCtrl;
     }
+
+    if (pthread_create(&pConn->thread, &thAttr, taosRecvUdpData, pConn) != 0) {
+      tError("%s failed to create thread to process UDP data, reason:%s", label, strerror(errno));
+      taosCloseSocket(pConn->fd);
+      taosCleanUpUdpConnection(pSet);
+      return NULL;
+    }
+
     ++pSet->threads;
   }
 
@@ -164,6 +171,7 @@ void taosCleanUpUdpConnection(void *handle) {
   for (int i = 0; i < pSet->threads; ++i) {
     pConn = pSet->udpConn + i;
     pConn->signature = NULL;
+    free(pConn->buffer);
     pthread_cancel(pConn->thread);
     taosCloseSocket(pConn->fd);
     if (pConn->hash) {
@@ -210,7 +218,7 @@ static void *taosRecvUdpData(void *param) {
   tTrace("%s UDP thread is created, index:%d", pConn->label, pConn->index);
 
   while (1) {
-    dataLen = recvfrom(pConn->fd, pConn->buffer, sizeof(pConn->buffer), 0, (struct sockaddr *)&sourceAdd, &addLen);
+    dataLen = recvfrom(pConn->fd, pConn->buffer, RPC_MAX_UDP_SIZE, 0, (struct sockaddr *)&sourceAdd, &addLen);
     tTrace("%s msg is recv from 0x%x:%hu len:%d", pConn->label, sourceAdd.sin_addr.s_addr, ntohs(sourceAdd.sin_port),
            dataLen);
 
@@ -235,9 +243,15 @@ static void *taosRecvUdpData(void *param) {
         break;
       }
 
-      char *data = malloc((size_t)msgLen);
-      memcpy(data, msg, (size_t)msgLen);
-      recvInfo.msg = data;
+      char *tmsg = malloc((size_t)msgLen + tsRpcOverhead);
+      if (NULL == tmsg) {
+        tError("%s failed to allocate memory, size:%d", pConn->label, msgLen);
+        break;
+      }
+
+      tmsg += tsRpcOverhead;  // overhead for SRpcReqContext
+      memcpy(tmsg, msg, (size_t)msgLen);
+      recvInfo.msg = tmsg;
       recvInfo.msgLen = msgLen;
       recvInfo.ip = sourceAdd.sin_addr.s_addr;
       recvInfo.port = port;