TD-2324

src/dnode/src/dnodeMain.c

@@ -20,6 +20,7 @@
 #include "tfile.h"
 #include "twal.h"
 #include "tfs.h"
+#include "tsync.h"
 #include "dnodePeer.h"
 #include "dnodeModule.h"
 #include "dnodeEps.h"
@@ -53,6 +54,7 @@ static SStep tsDnodeSteps[] = {
   {"dnode-eps",       dnodeInitEps,        dnodeCleanupEps},
   {"dnode-minfos",    dnodeInitMInfos,     dnodeCleanupMInfos},
   {"dnode-wal",       walInit,             walCleanUp},
+  {"dnode-sync",      syncInit,            syncCleanUp},
   {"dnode-check",     dnodeInitCheck,      dnodeCleanupCheck},     // NOTES: dnodeInitCheck must be behind the dnodeinitStorage component !!!
   {"dnode-vread",     dnodeInitVRead,      dnodeCleanupVRead},
   {"dnode-vwrite",    dnodeInitVWrite,     dnodeCleanupVWrite},

src/dnode/src/dnodeVMgmt.c

@@ -49,13 +49,13 @@ int32_t dnodeInitVMgmt() {
 
   tsMgmtQset = taosOpenQset();
   if (tsMgmtQset == NULL) {
-    dError("failed to create the mgmt queue set");
+    dError("failed to create the vmgmt queue set");
     return -1;
   }
 
   tsMgmtQueue = taosOpenQueue();
   if (tsMgmtQueue == NULL) {
-    dError("failed to create the mgmt queue");
+    dError("failed to create the vmgmt queue");
     return -1;
   }
 
@@ -68,11 +68,11 @@ int32_t dnodeInitVMgmt() {
   code = pthread_create(&tsQthread, &thAttr, dnodeProcessMgmtQueue, NULL);
   pthread_attr_destroy(&thAttr);
   if (code != 0) {
-    dError("failed to create thread to process mgmt queue, reason:%s", strerror(errno));
+    dError("failed to create thread to process vmgmt queue, reason:%s", strerror(errno));
     return -1;
   }
 
-  dInfo("dnode mgmt is initialized");
+  dInfo("dnode vmgmt is initialized");
   return TSDB_CODE_SUCCESS;
 }
 
@@ -133,9 +133,12 @@ static void *dnodeProcessMgmtQueue(void *param) {
       rsp.code = TSDB_CODE_DND_MSG_NOT_PROCESSED;
     }
 
-    rsp.handle = pMsg->handle;
-    rsp.pCont = NULL;
-    rpcSendResponse(&rsp);
+    dDebug("msg:%p, is processed, code:0x%x", pMgmt, rsp.code);
+    if (rsp.code != TSDB_CODE_DND_ACTION_IN_PROGRESS) {
+      rsp.handle = pMsg->handle;
+      rsp.pCont = NULL;
+      rpcSendResponse(&rsp);
+    }
 
     taosFreeQitem(pMsg);
   }

src/inc/taoserror.h

@@ -192,6 +192,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED,        0, 0x0400, "Message no
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY,            0, 0x0401, "Dnode out of memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_NO_WRITE_ACCESS,          0, 0x0402, "No permission for disk files in dnode")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_MSG_LEN,          0, 0x0403, "Invalid message length")
+TAOS_DEFINE_ERROR(TSDB_CODE_DND_ACTION_IN_PROGRESS,       0, 0x0404, "Action in progress")
 
 // vnode 
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_ACTION_IN_PROGRESS,       0, 0x0500, "Action in progress")

src/vnode/src/vnodeMgmt.c

@@ -17,6 +17,7 @@
 #include "os.h"
 #include "dnode.h"
 #include "vnodeStatus.h"
+#include "vnodeWorker.h"
 #include "vnodeRead.h"
 #include "vnodeWrite.h"
 #include "vnodeMain.h"
@@ -28,11 +29,11 @@ static void    vnodeCleanupHash(void);
 static void    vnodeIncRef(void *ptNode);
 
 static SStep tsVnodeSteps[] = {
-  {"vsync",  syncInit,            syncCleanUp},
-  {"vwrite", vnodeInitWrite,      vnodeCleanupWrite},
-  {"vread",  vnodeInitRead,       vnodeCleanupRead},
-  {"vhash",  vnodeInitHash,       vnodeCleanupHash},
-  {"vqueue", tsdbInitCommitQueue, tsdbDestroyCommitQueue}
+  {"vnode-worker", vnodeInitMWorker,    vnodeCleanupMWorker},
+  {"vnode-write",  vnodeInitWrite,      vnodeCleanupWrite},
+  {"vnode-read",   vnodeInitRead,       vnodeCleanupRead},
+  {"vnode-hash",   vnodeInitHash,       vnodeCleanupHash},
+  {"tsdb-queue",   tsdbInitCommitQueue, tsdbDestroyCommitQueue}
 };
 
 int32_t vnodeInitMgmt() {

src/vnode/src/vnodeWorker.c

@@ -143,7 +143,11 @@ static int32_t vnodeWriteIntoMWorker(int32_t vgId, EVMWorkerAction action,void *
   pMsg->pVnode = pVnode;
   pMsg->rpcHandle = rpcHandle;
   pMsg->action = action;
-  return taosWriteQitem(tsVMWorkerQueue, TAOS_QTYPE_RPC, pMsg);
+
+  int32_t code = taosWriteQitem(tsVMWorkerQueue, TAOS_QTYPE_RPC, pMsg);
+  if (code == 0) code = TSDB_CODE_DND_ACTION_IN_PROGRESS;
+
+  return code;
 }
 
 int32_t vnodeOpenInMWorker(int32_t vgId, void *rpcHandle) {