TD-1899

src/dnode/inc/dnodeVRead.h

@@ -24,7 +24,7 @@ int32_t dnodeInitVRead();
 void    dnodeCleanupVRead();
 void    dnodeDispatchToVReadQueue(SRpcMsg *pMsg);
 void *  dnodeAllocVReadQueue(void *pVnode);
-void    dnodeFreeVReadQueue(void *rqueue);
+void    dnodeFreeVReadQueue(void *pRqueue);
 
 #ifdef __cplusplus
 }

src/dnode/src/dnodeVRead.c

@@ -35,7 +35,7 @@ typedef struct {
   pthread_mutex_t mutex;
 } SVReadWorkerPool;
 
-static void *dnodeProcessReadQueue(void *param);
+static void *dnodeProcessReadQueue(void *pWorker);
 
 // module global variable
 static SVReadWorkerPool tsVReadWP;
@@ -47,7 +47,7 @@ int32_t dnodeInitVRead() {
   tsVReadWP.min = tsNumOfCores;
   tsVReadWP.max = tsNumOfCores * tsNumOfThreadsPerCore;
   if (tsVReadWP.max <= tsVReadWP.min * 2) tsVReadWP.max = 2 * tsVReadWP.min;
-  tsVReadWP.worker = (SVReadWorker *)calloc(sizeof(SVReadWorker), tsVReadWP.max);
+  tsVReadWP.worker = calloc(sizeof(SVReadWorker), tsVReadWP.max);
   pthread_mutex_init(&tsVReadWP.mutex, NULL);
 
   if (tsVReadWP.worker == NULL) return -1;
@@ -85,7 +85,7 @@ void dnodeCleanupVRead() {
 void dnodeDispatchToVReadQueue(SRpcMsg *pMsg) {
   int32_t queuedMsgNum = 0;
   int32_t leftLen = pMsg->contLen;
-  char *  pCont = (char *)pMsg->pCont;
+  char *  pCont = pMsg->pCont;
 
   while (leftLen > 0) {
     SMsgHead *pHead = (SMsgHead *)pCont;
@@ -146,8 +146,8 @@ void *dnodeAllocVReadQueue(void *pVnode) {
   return queue;
 }
 
-void dnodeFreeVReadQueue(void *rqueue) {
-  taosCloseQueue(rqueue);
+void dnodeFreeVReadQueue(void *pRqueue) {
+  taosCloseQueue(pRqueue);
 }
 
 void dnodeSendRpcVReadRsp(void *pVnode, SVReadMsg *pRead, int32_t code) {
@@ -159,14 +159,12 @@ void dnodeSendRpcVReadRsp(void *pVnode, SVReadMsg *pRead, int32_t code) {
   };
 
   rpcSendResponse(&rpcRsp);
-  vnodeRelease(pVnode);
 }
 
 void dnodeDispatchNonRspMsg(void *pVnode, SVReadMsg *pRead, int32_t code) {
-  vnodeRelease(pVnode);
 }
 
-static void *dnodeProcessReadQueue(void *param) {
+static void *dnodeProcessReadQueue(void *pWorker) {
   SVReadMsg *pRead;
   int32_t    qtype;
   void *     pVnode;
@@ -193,7 +191,7 @@ static void *dnodeProcessReadQueue(void *param) {
       }
     }
 
-    taosFreeQitem(pRead);
+    vnodeFreeFromRQueue(pVnode, pRead);
   }
 
   return NULL;

src/inc/dnode.h

@@ -57,7 +57,7 @@ void *dnodeAllocVWriteQueue(void *pVnode);
 void  dnodeFreeVWriteQueue(void *pWqueue);
 void  dnodeSendRpcVWriteRsp(void *pVnode, void *pWrite, int32_t code);
 void *dnodeAllocVReadQueue(void *pVnode);
-void  dnodeFreeVReadQueue(void *rqueue);
+void  dnodeFreeVReadQueue(void *pRqueue);
 
 int32_t dnodeAllocateMPeerQueue();
 void    dnodeFreeMPeerQueue();

src/inc/vnode.h

@@ -23,12 +23,12 @@ extern "C" {
 #include "twal.h"
 
 typedef enum _VN_STATUS {
-  TAOS_VN_STATUS_INIT,
-  TAOS_VN_STATUS_READY,
-  TAOS_VN_STATUS_CLOSING,
-  TAOS_VN_STATUS_UPDATING,
-  TAOS_VN_STATUS_RESET,
-} EVnStatus;
+  TAOS_VN_STATUS_INIT = 0,
+  TAOS_VN_STATUS_READY = 1,
+  TAOS_VN_STATUS_CLOSING = 2,
+  TAOS_VN_STATUS_UPDATING = 3,
+  TAOS_VN_STATUS_RESET = 4,
+} EVnodeStatus;
 
 typedef struct {
   int32_t len;
@@ -81,7 +81,8 @@ void    vnodeSetAccess(SVgroupAccess *pAccess, int32_t numOfVnodes);
 int32_t vnodeInitResources();
 void    vnodeCleanupResources();
 
-int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qtype, void *rparam);
+int32_t vnodeWriteToRQueue(void *pVnode, void *pCont, int32_t contLen, int8_t qtype, void *rparam);
+void    vnodeFreeFromRQueue(void *pVnode, SVReadMsg *pRead);
 int32_t vnodeProcessRead(void *pVnode, SVReadMsg *pRead);
 
 #ifdef __cplusplus

src/vnode/inc/vnodeInt.h

@@ -37,8 +37,9 @@ extern int32_t vDebugFlag;
 typedef struct {
   int32_t      vgId;      // global vnode group ID
   int32_t      refCount;  // reference count
-  int32_t      queuedMsg;
-  int32_t      delay;
+  int32_t      queuedWMsg;
+  int32_t      queuedRMsg;
+  int32_t      delayMs;
   int8_t       status; 
   int8_t       role;   
   int8_t       accessState;

src/vnode/src/vnodeMain.c

@@ -378,9 +378,9 @@ int32_t vnodeClose(int32_t vgId) {
   return 0;
 }
 
-void vnodeRelease(void *pVnodeRaw) {
-  if (pVnodeRaw == NULL) return;
-  SVnodeObj *pVnode = pVnodeRaw;
+void vnodeRelease(void *vparam) {
+  if (vparam == NULL) return;
+  SVnodeObj *pVnode = vparam;
   int32_t    code = 0;
   int32_t    vgId = pVnode->vgId;
 
@@ -643,18 +643,19 @@ static void vnodeNotifyRole(void *ahandle, int8_t role) {
   pVnode->role = role;
   dnodeSendStatusMsgToMnode();
 
-  if (pVnode->role == TAOS_SYNC_ROLE_MASTER)
+  if (pVnode->role == TAOS_SYNC_ROLE_MASTER) {
     cqStart(pVnode->cq);
-  else
+  } else {
     cqStop(pVnode->cq);
+  }
 }
 
 static void vnodeCtrlFlow(void *ahandle, int32_t mseconds) {
   SVnodeObj *pVnode = ahandle;
-  if (pVnode->delay != mseconds) {
-    vInfo("vgId:%d, sync flow control, mseconds:%d", pVnode->vgId, mseconds);
+  if (pVnode->delayMs != mseconds) {
+    pVnode->delayMs = mseconds;
+    vDebug("vgId:%d, sync flow control, mseconds:%d", pVnode->vgId, mseconds);
   }
-  pVnode->delay = mseconds;
 }
 
 static int vnodeResetTsdb(SVnodeObj *pVnode) {

src/vnode/src/vnodeRead.c

@@ -41,8 +41,8 @@ void vnodeInitReadFp(void) {
 // still required, or there will be a deadlock, so we don’t do any check here, but put the check codes before the
 // request enters the queue
 //
-int32_t vnodeProcessRead(void *param, SVReadMsg *pRead) {
-  SVnodeObj *pVnode = (SVnodeObj *)param;
+int32_t vnodeProcessRead(void *vparam, SVReadMsg *pRead) {
+  SVnodeObj *pVnode = vparam;
   int32_t    msgType = pRead->msgType;
 
   if (vnodeProcessReadMsgFp[msgType] == NULL) {
@@ -53,8 +53,8 @@ int32_t vnodeProcessRead(void *param, SVReadMsg *pRead) {
   return (*vnodeProcessReadMsgFp[msgType])(pVnode, pRead);
 }
 
-static int32_t vnodeCheckRead(void *param) {
-  SVnodeObj *pVnode = param;
+static int32_t vnodeCheckRead(void *vparam) {
+  SVnodeObj *pVnode = vparam;
   if (pVnode->status != TAOS_VN_STATUS_READY) {
     vDebug("vgId:%d, vnode status is %s, refCount:%d pVnode:%p", pVnode->vgId, vnodeStatus[pVnode->status],
            pVnode->refCount, pVnode);
@@ -76,6 +76,16 @@ static int32_t vnodeCheckRead(void *param) {
   return TSDB_CODE_SUCCESS;
 }
 
+void vnodeFreeFromRQueue(void *vparam, SVReadMsg *pRead) {
+  SVnodeObj *pVnode = vparam;
+
+  atomic_sub_fetch_32(&pVnode->queuedRMsg, 1);
+  vTrace("vgId:%d, free from vrqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedRMsg);
+
+  taosFreeQitem(pRead);
+  vnodeRelease(pVnode);
+}
+
 int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qtype, void *rparam) {
   SVnodeObj *pVnode = vparam;
 
@@ -108,7 +118,8 @@ int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qt
   pRead->qtype = qtype;
 
   atomic_add_fetch_32(&pVnode->refCount, 1);
-  vTrace("vgId:%d, get vnode rqueue, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
+  atomic_add_fetch_32(&pVnode->queuedRMsg, 1);
+  vTrace("vgId:%d, write into vrqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedRMsg);
 
   taosWriteQitem(pVnode->rqueue, qtype, pRead);
   return TSDB_CODE_SUCCESS;

src/vnode/src/vnodeWrite.c

@@ -77,7 +77,7 @@ int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rpara
 
     // assign version
     pHead->version = pVnode->version + 1;
-    if (pVnode->delay) usleep(pVnode->delay * 1000);
+    if (pVnode->delayMs) taosMsleep(pVnode->delayMs);
 
   } else {  // from wal or forward
     // for data from WAL or forward, version may be smaller
@@ -245,13 +245,13 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar
 
   atomic_add_fetch_32(&pVnode->refCount, 1);
 
-  int32_t queued = atomic_add_fetch_32(&pVnode->queuedMsg, 1);
+  int32_t queued = atomic_add_fetch_32(&pVnode->queuedWMsg, 1);
   if (queued > MAX_QUEUED_MSG_NUM) {
     vDebug("vgId:%d, too many msg:%d in vwqueue, flow control", pVnode->vgId, queued);
     taosMsleep(1);
   }
 
-  vTrace("vgId:%d, write into vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedMsg);
+  vTrace("vgId:%d, write into vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedWMsg);
 
   taosWriteQitem(pVnode->wqueue, qtype, pWrite);
   return TSDB_CODE_SUCCESS;
@@ -260,8 +260,8 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar
 void vnodeFreeFromWQueue(void *vparam, SVWriteMsg *pWrite) {
   SVnodeObj *pVnode = vparam;
 
-  atomic_sub_fetch_32(&pVnode->queuedMsg, 1);
-  vTrace("vgId:%d, free from vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedMsg);
+  atomic_sub_fetch_32(&pVnode->queuedWMsg, 1);
+  vTrace("vgId:%d, free from vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedWMsg);
 
   taosFreeQitem(pWrite);
   vnodeRelease(pVnode);