Merge pull request #4869 from taosdata/feature/sim

Feature/sim

src/vnode/inc/vnodeRead.h

@@ -27,7 +27,7 @@ void    vnodeCleanupRead(void);
 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);
-void    vnodeWaitReadCompleted(void *pVnode);
+void    vnodeWaitReadCompleted(SVnodeObj *pVnode);
 
 #ifdef __cplusplus
 }

src/vnode/inc/vnodeWrite.h

@@ -27,7 +27,7 @@ void    vnodeCleanupWrite(void);
 int32_t vnodeWriteToWQueue(void *pVnode, void *pHead, int32_t qtype, void *pRpcMsg);
 void    vnodeFreeFromWQueue(void *pVnode, SVWriteMsg *pWrite);
 int32_t vnodeProcessWrite(void *pVnode, void *pHead, int32_t qtype, void *pRspRet);
-void    vnodeWaitWriteCompleted(void *pVnode);
+void    vnodeWaitWriteCompleted(SVnodeObj *pVnode);
 
 #ifdef __cplusplus
 }

src/vnode/src/vnodeRead.c

@@ -436,4 +436,9 @@ int32_t vnodeNotifyCurrentQhandle(void *handle, void *qhandle, int32_t vgId) {
   return rpcReportProgress(handle, (char *)pMsg, sizeof(SRetrieveTableMsg));
 }
 
-void vnodeWaitReadCompleted(void *pVnode) {}
+void vnodeWaitReadCompleted(SVnodeObj *pVnode) {
+  while (pVnode->queuedRMsg > 0) {
+    vTrace("vgId:%d, queued rmsg num:%d", pVnode->vgId, pVnode->queuedRMsg);
+    taosMsleep(10);
+  }
+}

src/vnode/src/vnodeStatus.c

@@ -18,6 +18,8 @@
 #include "taosmsg.h"
 #include "query.h"
 #include "vnodeStatus.h"
+#include "vnodeRead.h"
+#include "vnodeWrite.h"
 
 char* vnodeStatus[] = {
   "init",
@@ -77,6 +79,8 @@ bool vnodeSetClosingStatus(SVnodeObj* pVnode) {
 
   // release local resources only after cutting off outside connections
   qQueryMgmtNotifyClosed(pVnode->qMgmt);
+  vnodeWaitReadCompleted(pVnode);
+  vnodeWaitWriteCompleted(pVnode);
 
   return true;
 }
@@ -121,6 +125,8 @@ bool vnodeSetResetStatus(SVnodeObj* pVnode) {
 
   // release local resources only after cutting off outside connections
   qQueryMgmtNotifyClosed(pVnode->qMgmt);
+  vnodeWaitReadCompleted(pVnode);
+  vnodeWaitWriteCompleted(pVnode);
 
   return true;
 }

src/vnode/src/vnodeWrite.c

@@ -345,4 +345,9 @@ static int32_t vnodePerformFlowCtrl(SVWriteMsg *pWrite) {
   }
 }
 
-void vnodeWaitWriteCompleted(void *pVnode) {}
+void vnodeWaitWriteCompleted(SVnodeObj *pVnode) {
+  while (pVnode->queuedWMsg > 0) {
+    vTrace("vgId:%d, queued wmsg num:%d", pVnode->vgId, pVnode->queuedWMsg);
+    taosMsleep(10);
+  }
+}