refactor: multi-process test mode

source/dnode/mgmt/node_mgmt/src/dmEnv.c

@@ -70,8 +70,9 @@ static int32_t dmCheckRepeatCleanup(SDnode *pDnode) {
 
 void dmCleanup() {
   dDebug("start to cleanup env");
-  if (dmCheckRepeatCleanup != 0) return;
-  dmCleanupDnode(dmInstance());
+  SDnode *pDnode = dmInstance();
+  if (dmCheckRepeatCleanup(pDnode) != 0) return;
+  dmCleanupDnode(pDnode);
   monCleanup();
   syncCleanUp();
   walCleanUp();

source/dnode/mgmt/node_mgmt/src/dmMgmt.c

@@ -25,9 +25,18 @@ static bool dmRequireNode(SDnode *pDnode, SMgmtWrapper *pWrapper) {
     dDebug("node:%s, does not require startup", pWrapper->name);
   }
 
-  if (pWrapper->ntype == DNODE && pDnode->rtype != DNODE && pDnode->rtype != NODE_END) {
-    required = false;
-    dDebug("node:%s, does not require startup in child process", pWrapper->name);
+  if (pWrapper->ntype == DNODE) {
+    if (pDnode->rtype != DNODE && pDnode->rtype != NODE_END) {
+      required = false;
+      dDebug("node:%s, does not require startup in child process", pWrapper->name);
+    }
+  } else {
+    if (OnlyInChildProc(pWrapper)) {
+      if (pWrapper->ntype != pDnode->rtype) {
+        dDebug("node:%s, does not require startup in child process", pWrapper->name);
+        required = false;
+      }
+    }
   }
 
   if (required) {
@@ -111,7 +120,6 @@ static void dmClearVars(SDnode *pDnode) {
 
   taosThreadMutexDestroy(&pDnode->mutex);
   memset(&pDnode->mutex, 0, sizeof(pDnode->mutex));
-  taosMemoryFree(pDnode);
 }
 
 int32_t dmInitDnode(SDnode *pDnode, EDndNodeType rtype) {

source/dnode/mgmt/node_mgmt/src/dmProc.c

@@ -392,12 +392,13 @@ static void *dmConsumParentQueue(void *param) {
       rpcSendResponse(pRsp);
     } else if (ftype == DND_FUNC_REGIST) {
       pRsp = pHead;
+      pRsp->pCont = pBody;
       dTrace("node:%s, get regist msg:%p from pqueue, code:0x%04x handle:%p", proc->name, pRsp, code,
              pRsp->info.handle);
       rpcRegisterBrokenLinkArg(pRsp);
-      rpcFreeCont(pBody);
     } else if (ftype == DND_FUNC_RELEASE) {
       pRsp = pHead;
+      pRsp->pCont = NULL;
       dTrace("node:%s, get release msg:%p from pqueue, code:0x%04x handle:%p", proc->name, pRsp, code,
              pRsp->info.handle);
       dmRemoveProcRpcHandle(proc, pRsp->info.handle);

source/dnode/mgmt/node_mgmt/src/dmTransport.c

@@ -43,6 +43,7 @@ int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
     return -1;
   }
 
+  pMsg->info.wrapper = pWrapper;
   dTrace("msg:%p, will be processed by %s, handle:%p", pMsg, pWrapper->name, pMsg->info.handle);
   return (*msgFp)(pWrapper->pMgmt, pMsg);
 }