refactor rpc code

source/libs/transport/inc/transComm.h

@@ -393,9 +393,9 @@ typedef struct SDelayQueue {
   uv_loop_t*  loop;
 } SDelayQueue;
 
-int  transDQCreate(uv_loop_t* loop, SDelayQueue** queue);
-void transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg));
-int  transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs);
+int         transDQCreate(uv_loop_t* loop, SDelayQueue** queue);
+void        transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg));
+SDelayTask* transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs);
 
 bool transEpSetIsEqual(SEpSet* a, SEpSet* b);
 /*

source/libs/transport/src/transCli.c

@@ -1054,6 +1054,12 @@ static void doDelayTask(void* param) {
   cliHandleReq(pMsg, pThrd);
 }
 
+static void doCloseIdleConn(void* param) {
+  STaskArg* arg = param;
+  SCliConn* conn = arg->param1;
+  SCliThrd* pThrd = arg->param2;
+}
+
 static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) {
   STransConnCtx* pCtx = pMsg->ctx;
 
@@ -1075,7 +1081,7 @@ void cliCompareAndSwap(int8_t* val, int8_t exp, int8_t newVal) {
   }
 }
 
-bool cliTryToExtractEpSet(STransMsg* pResp, SEpSet* dst) {
+bool cliTryExtractEpSet(STransMsg* pResp, SEpSet* dst) {
   if ((pResp == NULL || pResp->info.hasEpSet == 0)) {
     return false;
   }
@@ -1116,7 +1122,8 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
    */
   STransConnCtx* pCtx = pMsg->ctx;
   int32_t        code = pResp->code;
-  bool           retry = (pTransInst->retry != NULL && pTransInst->retry(code, pResp->msgType - 1)) ? true : false;
+
+  bool retry = (pTransInst->retry != NULL && pTransInst->retry(code, pResp->msgType - 1)) ? true : false;
   if (retry) {
     pMsg->sent = 0;
     pCtx->retryCnt += 1;
@@ -1125,6 +1132,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
       if (pCtx->retryCnt < pCtx->retryLimit) {
         transUnrefCliHandle(pConn);
         EPSET_FORWARD_INUSE(&pCtx->epSet);
+        transFreeMsg(pResp->pCont);
         cliSchedMsgToNextNode(pMsg, pThrd);
         return -1;
       }
@@ -1148,7 +1156,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
 
   STraceId* trace = &pResp->info.traceId;
 
-  bool hasEpSet = cliTryToExtractEpSet(pResp, &pCtx->epSet);
+  bool hasEpSet = cliTryExtractEpSet(pResp, &pCtx->epSet);
   if (hasEpSet) {
     char tbuf[256] = {0};
     EPSET_DEBUG_STR(&pCtx->epSet, tbuf);

source/libs/transport/src/transComm.c

@@ -488,8 +488,25 @@ void transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg)) {
   heapDestroy(queue->heap);
   taosMemoryFree(queue);
 }
+void transDQCancel(SDelayQueue* queue, SDelayTask* task) {
+  uv_timer_stop(queue->timer);
 
-int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs) {
+  if (heapSize(queue->heap) <= 0) return;
+  heapRemove(queue->heap, &task->node);
+
+  if (heapSize(queue->heap) != 0) {
+    HeapNode* minNode = heapMin(queue->heap);
+    if (minNode != NULL) return;
+
+    uint64_t    now = taosGetTimestampMs();
+    SDelayTask* task = container_of(minNode, SDelayTask, node);
+    uint64_t    timeout = now > task->execTime ? now - task->execTime : 0;
+
+    uv_timer_start(queue->timer, transDQTimeout, timeout, 0);
+  }
+}
+
+SDelayTask* transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs) {
   uint64_t    now = taosGetTimestampMs();
   SDelayTask* task = taosMemoryCalloc(1, sizeof(SDelayTask));
   task->func = func;
@@ -507,7 +524,7 @@ int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_
   tTrace("timer %p put task into delay queue, timeoutMs:%" PRIu64, queue->timer, timeoutMs);
   heapInsert(queue->heap, &task->node);
   uv_timer_start(queue->timer, transDQTimeout, timeoutMs, 0);
-  return 0;
+  return task;
 }
 
 void transPrintEpSet(SEpSet* pEpSet) {