refactor rpc

source/libs/transport/inc/transComm.h

@@ -199,4 +199,6 @@ void transBuildAuthHead(void* pMsg, int msgLen, void* pAuth, void* pKey);
 bool transCompressMsg(char* msg, int32_t len, int32_t* flen);
 bool transDecompressMsg(char* msg, int32_t len, int32_t* flen);
 
+void transConnCtxDestroy(STransConnCtx* ctx);
+
 #endif

source/libs/transport/src/transCli.c

@@ -63,6 +63,8 @@ static void clientAsyncCb(uv_async_t* handle);
 static void clientDestroy(uv_handle_t* handle);
 static void clientConnDestroy(SCliConn* pConn);
 
+static void clientMsgDestroy(SCliMsg* pMsg);
+
 static void* clientThread(void* arg);
 
 static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
@@ -235,9 +237,22 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
   }
   return cli;
 }
+static void clientMsgDestroy(SCliMsg* pMsg) {
+  // impl later
+  free(pMsg);
+}
 void taosCloseClient(void* arg) {
   // impl later
   SClientObj* cli = arg;
+  for (int i = 0; i < cli->numOfThreads; i++) {
+    SCliThrdObj* pThrd = cli->pThreadObj[i];
+    pthread_join(pThrd->thread, NULL);
+    pthread_mutex_destroy(&pThrd->msgMtx);
+    free(pThrd->loop);
+    free(pThrd);
+  }
+  free(cli->pThreadObj);
+  free(cli);
 }
 
 void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t* pRid) {
@@ -247,19 +262,18 @@ void rpcSendRequest(void* shandle, const SEpSet* pEpSet, SRpcMsg* pMsg, int64_t*
 
   SRpcInfo* pRpc = (SRpcInfo*)shandle;
 
-  int len = rpcCompressRpcMsg(pMsg->pCont, pMsg->contLen);
+  int32_t flen = 0;
+  if (transCompressMsg(pMsg->pCont, pMsg->contLen, &flen)) {
+    // imp later
+  }
 
   STransConnCtx* pCtx = calloc(1, sizeof(STransConnCtx));
 
   pCtx->pRpc = (SRpcInfo*)shandle;
   pCtx->ahandle = pMsg->ahandle;
-  // pContext->contLen = len;
-  // pContext->pCont = pMsg->pCont;
   pCtx->msgType = pMsg->msgType;
   pCtx->ip = strdup(ip);
   pCtx->port = port;
-  // pContext->epSet = *pEpSet;
-  // pContext->oldInUse = pEpSet->inUse;
 
   assert(pRpc->connType == TAOS_CONN_CLIENT);
   // atomic or not

source/libs/transport/src/transComm.c

@@ -107,6 +107,7 @@ int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
 }
 
 bool transCompressMsg(char* msg, int32_t len, int32_t* flen) {
+  return false;
   // SRpcHead* pHead = rpcHeadFromCont(pCont);
   bool succ = false;
   int  overhead = sizeof(STransCompMsg);
@@ -186,4 +187,8 @@ SRpcHead* rpcDecompressRpcMsg(SRpcHead* pHead) {
   return pHead;
 }
 
+void transConnCtxDestroy(STransConnCtx* ctx) {
+  free(ctx->ip);
+  free(ctx);
+}
 #endif

source/libs/transport/src/transSrv.c

@@ -496,6 +496,7 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
 
   for (int i = 0; i < srv->numOfThreads; i++) {
     SWorkThrdObj* thrd = (SWorkThrdObj*)calloc(1, sizeof(SWorkThrdObj));
+
     srv->pipe[i] = (uv_pipe_t*)calloc(2, sizeof(uv_pipe_t));
     int fds[2];
     if (uv_socketpair(AF_UNIX, SOCK_STREAM, fds, UV_NONBLOCK_PIPE, UV_NONBLOCK_PIPE) != 0) {
@@ -530,6 +531,17 @@ void* taosInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
 void taosCloseServer(void* arg) {
   // impl later
   SServerObj* srv = arg;
+  for (int i = 0; i < srv->numOfThreads; i++) {
+    SWorkThrdObj* pThrd = srv->pThreadObj[i];
+    pthread_join(pThrd->thread, NULL);
+    free(srv->pipe[i]);
+    free(pThrd->loop);
+    free(pThrd);
+  }
+  free(srv->loop);
+  free(srv->pipe);
+  free(srv->pThreadObj);
+  pthread_join(srv->thread, NULL);
 }
 
 void rpcSendResponse(const SRpcMsg* pMsg) {