refactor rpc code

source/libs/transport/inc/transComm.h

@@ -229,8 +229,8 @@ typedef struct {
   int8_t      stop;
 } SAsyncPool;
 
-SAsyncPool* transCreateAsyncPool(uv_loop_t* loop, int sz, void* arg, AsyncCB cb);
-void        transDestroyAsyncPool(SAsyncPool* pool);
+SAsyncPool* transAsyncPoolCreate(uv_loop_t* loop, int sz, void* arg, AsyncCB cb);
+void        transAsyncPoolDestroy(SAsyncPool* pool);
 int         transAsyncSend(SAsyncPool* pool, queue* mq);
 bool        transAsyncPoolIsEmpty(SAsyncPool* pool);
 
@@ -322,7 +322,7 @@ typedef struct STransReq {
 } STransReq;
 
 void  transReqQueueInit(queue* q);
-void* transReqQueuePushReq(queue* q);
+void* transReqQueuePush(queue* q);
 void* transReqQueueRemove(void* arg);
 void  transReqQueueClear(queue* q);
 

source/libs/transport/src/transCli.c

@@ -26,7 +26,7 @@ typedef struct SCliConn {
 
   SConnBuffer readBuf;
   STransQueue cliMsgs;
-  queue       conn;
+  queue       q;
   uint64_t    expireTime;
 
   STransCtx  ctx;
@@ -451,7 +451,7 @@ void cliTimeoutCb(uv_timer_t* handle) {
   while (p != NULL) {
     while (!QUEUE_IS_EMPTY(&p->conn)) {
       queue*    h = QUEUE_HEAD(&p->conn);
-      SCliConn* c = QUEUE_DATA(h, SCliConn, conn);
+      SCliConn* c = QUEUE_DATA(h, SCliConn, q);
       if (c->expireTime < currentTime) {
         QUEUE_REMOVE(h);
         transUnrefCliHandle(c);
@@ -475,7 +475,7 @@ void* destroyConnPool(void* pool) {
   while (connList != NULL) {
     while (!QUEUE_IS_EMPTY(&connList->conn)) {
       queue*    h = QUEUE_HEAD(&connList->conn);
-      SCliConn* c = QUEUE_DATA(h, SCliConn, conn);
+      SCliConn* c = QUEUE_DATA(h, SCliConn, q);
       cliDestroyConn(c, true);
     }
     connList = taosHashIterate((SHashObj*)pool, connList);
@@ -501,11 +501,11 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
     return NULL;
   }
   queue*    h = QUEUE_HEAD(&plist->conn);
-  SCliConn* conn = QUEUE_DATA(h, SCliConn, conn);
+  SCliConn* conn = QUEUE_DATA(h, SCliConn, q);
   conn->status = ConnNormal;
-  QUEUE_REMOVE(&conn->conn);
-  QUEUE_INIT(&conn->conn);
-  assert(h == &conn->conn);
+  QUEUE_REMOVE(&conn->q);
+  QUEUE_INIT(&conn->q);
+  assert(h == &conn->q);
   return conn;
 }
 static int32_t allocConnRef(SCliConn* conn, bool update) {
@@ -560,8 +560,8 @@ static void addConnToPool(void* pool, SCliConn* conn) {
   SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key));
   // list already create before
   assert(plist != NULL);
-  QUEUE_INIT(&conn->conn);
-  QUEUE_PUSH(&plist->conn, &conn->conn);
+  QUEUE_INIT(&conn->q);
+  QUEUE_PUSH(&plist->conn, &conn->q);
   assert(!QUEUE_IS_EMPTY(&plist->conn));
 }
 static void cliAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
@@ -614,7 +614,7 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) {
   transReqQueueInit(&conn->wreqQueue);
 
   transQueueInit(&conn->cliMsgs, NULL);
-  QUEUE_INIT(&conn->conn);
+  QUEUE_INIT(&conn->q);
   conn->hostThrd = pThrd;
   conn->status = ConnNormal;
   conn->broken = 0;
@@ -626,8 +626,8 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) {
 }
 static void cliDestroyConn(SCliConn* conn, bool clear) {
   tTrace("%s conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn);
-  QUEUE_REMOVE(&conn->conn);
-  QUEUE_INIT(&conn->conn);
+  QUEUE_REMOVE(&conn->q);
+  QUEUE_INIT(&conn->q);
   transRemoveExHandle(transGetRefMgt(), conn->refId);
   conn->refId = -1;
 
@@ -735,7 +735,7 @@ void cliSend(SCliConn* pConn) {
     CONN_SET_PERSIST_BY_APP(pConn);
   }
 
-  uv_write_t* req = transReqQueuePushReq(&pConn->wreqQueue);
+  uv_write_t* req = transReqQueuePush(&pConn->wreqQueue);
   uv_write(req, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb);
   return;
 _RETURN:
@@ -990,7 +990,7 @@ static SCliThrd* createThrdObj() {
   pThrd->loop = (uv_loop_t*)taosMemoryMalloc(sizeof(uv_loop_t));
   uv_loop_init(pThrd->loop);
 
-  pThrd->asyncPool = transCreateAsyncPool(pThrd->loop, 5, pThrd, cliAsyncCb);
+  pThrd->asyncPool = transAsyncPoolCreate(pThrd->loop, 5, pThrd, cliAsyncCb);
   uv_timer_init(pThrd->loop, &pThrd->timer);
   pThrd->timer.data = pThrd;
 
@@ -1009,7 +1009,7 @@ static void destroyThrdObj(SCliThrd* pThrd) {
   CLI_RELEASE_UV(pThrd->loop);
   taosThreadMutexDestroy(&pThrd->msgMtx);
   TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyCmsg);
-  transDestroyAsyncPool(pThrd->asyncPool);
+  transAsyncPoolDestroy(pThrd->asyncPool);
 
   transDQDestroy(pThrd->delayQueue, destroyCmsg);
   taosMemoryFree(pThrd->loop);
@@ -1336,19 +1336,18 @@ int transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMs
   tGDebug("%s send request at thread:%08" PRId64 ", dst:%s:%d, app:%p", transLabel(pTransInst), pThrd->pid,
           EPSET_GET_INUSE_IP(&pCtx->epSet), EPSET_GET_INUSE_PORT(&pCtx->epSet), pReq->info.ahandle);
 
-  if (0 != transAsyncSend(pThrd->asyncPool, &cliMsg->q)) {
-    tsem_destroy(sem);
-    taosMemoryFree(sem);
+  int ret = transAsyncSend(pThrd->asyncPool, &cliMsg->q);
+  if (ret != 0) {
     destroyCmsg(cliMsg);
-    transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
-    return -1;
+    goto _RETURN;
   }
   tsem_wait(sem);
+
+_RETURN:
   tsem_destroy(sem);
   taosMemoryFree(sem);
-
   transReleaseExHandle(transGetInstMgt(), (int64_t)shandle);
-  return 0;
+  return ret;
 }
 /*
  *

source/libs/transport/src/transComm.c

@@ -175,7 +175,7 @@ int transSetConnOption(uv_tcp_t* stream) {
   return ret;
 }
 
-SAsyncPool* transCreateAsyncPool(uv_loop_t* loop, int sz, void* arg, AsyncCB cb) {
+SAsyncPool* transAsyncPoolCreate(uv_loop_t* loop, int sz, void* arg, AsyncCB cb) {
   SAsyncPool* pool = taosMemoryCalloc(1, sizeof(SAsyncPool));
   pool->nAsync = sz;
   pool->asyncs = taosMemoryCalloc(1, sizeof(uv_async_t) * pool->nAsync);
@@ -194,7 +194,7 @@ SAsyncPool* transCreateAsyncPool(uv_loop_t* loop, int sz, void* arg, AsyncCB cb)
   return pool;
 }
 
-void transDestroyAsyncPool(SAsyncPool* pool) {
+void transAsyncPoolDestroy(SAsyncPool* pool) {
   for (int i = 0; i < pool->nAsync; i++) {
     uv_async_t* async = &(pool->asyncs[i]);
     // uv_close((uv_handle_t*)async, NULL);
@@ -205,6 +205,14 @@ void transDestroyAsyncPool(SAsyncPool* pool) {
   taosMemoryFree(pool->asyncs);
   taosMemoryFree(pool);
 }
+bool transAsyncPoolIsEmpty(SAsyncPool* pool) {
+  for (int i = 0; i < pool->nAsync; i++) {
+    uv_async_t* async = &(pool->asyncs[i]);
+    SAsyncItem* item = async->data;
+    if (!QUEUE_IS_EMPTY(&item->qmsg)) return false;
+  }
+  return true;
+}
 int transAsyncSend(SAsyncPool* pool, queue* q) {
   if (atomic_load_8(&pool->stop) == 1) {
     return -1;
@@ -228,14 +236,6 @@ int transAsyncSend(SAsyncPool* pool, queue* q) {
   }
   return uv_async_send(async);
 }
-bool transAsyncPoolIsEmpty(SAsyncPool* pool) {
-  for (int i = 0; i < pool->nAsync; i++) {
-    uv_async_t* async = &(pool->asyncs[i]);
-    SAsyncItem* item = async->data;
-    if (!QUEUE_IS_EMPTY(&item->qmsg)) return false;
-  }
-  return true;
-}
 
 void transCtxInit(STransCtx* ctx) {
   // init transCtx
@@ -308,7 +308,7 @@ void transReqQueueInit(queue* q) {
   // init req queue
   QUEUE_INIT(q);
 }
-void* transReqQueuePushReq(queue* q) {
+void* transReqQueuePush(queue* q) {
   uv_write_t* req = taosMemoryCalloc(1, sizeof(uv_write_t));
   STransReq*  wreq = taosMemoryCalloc(1, sizeof(STransReq));
   wreq->data = req;

source/libs/transport/src/transSvr.c

@@ -434,7 +434,7 @@ static void uvStartSendRespInternal(SSvrMsg* smsg) {
   uvPrepareSendData(smsg, &wb);
 
   transRefSrvHandle(pConn);
-  uv_write_t* req = transReqQueuePushReq(&pConn->wreqQueue);
+  uv_write_t* req = transReqQueuePush(&pConn->wreqQueue);
   uv_write(req, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnSendCb);
 }
 static void uvStartSendResp(SSvrMsg* smsg) {
@@ -697,7 +697,7 @@ static bool addHandleToWorkloop(SWorkThrd* pThrd, char* pipeName) {
   // conn set
   QUEUE_INIT(&pThrd->conn);
 
-  pThrd->asyncPool = transCreateAsyncPool(pThrd->loop, 1, pThrd, uvWorkerAsyncCb);
+  pThrd->asyncPool = transAsyncPoolCreate(pThrd->loop, 1, pThrd, uvWorkerAsyncCb);
   uv_pipe_connect(&pThrd->connect_req, pThrd->pipe, pipeName, uvOnPipeConnectionCb);
   // uv_read_start((uv_stream_t*)pThrd->pipe, uvAllocConnBufferCb, uvOnConnectionCb);
   return true;
@@ -976,7 +976,7 @@ void destroyWorkThrd(SWorkThrd* pThrd) {
   taosThreadJoin(pThrd->thread, NULL);
   SRV_RELEASE_UV(pThrd->loop);
   TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SSvrMsg, destroySmsg);
-  transDestroyAsyncPool(pThrd->asyncPool);
+  transAsyncPoolDestroy(pThrd->asyncPool);
   taosMemoryFree(pThrd->loop);
   taosMemoryFree(pThrd);
 }