shm

include/common/tmsg.h

@@ -2271,10 +2271,11 @@ static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* p
 
 struct SRpcMsg;
 struct SEpSet;
-typedef int32_t (*PutToQueueFp)(void *pMgmt, struct SRpcMsg *pReq);
-typedef int32_t (*SendReqFp)(void *pMgmt, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
-typedef int32_t (*SendMnodeReqFp)(void *pMgmt, struct SRpcMsg *rpcMsg);
-typedef int32_t (*SendRspFp)(void *pMgmt, struct SRpcMsg *rpcMsg);
+struct SMgmtWrapper;
+typedef int32_t (*PutToQueueFp)(struct SMgmtWrapper* pWrapper, struct SRpcMsg* pReq);
+typedef int32_t (*SendReqFp)(struct SMgmtWrapper* pWrapper, struct SEpSet* epSet, struct SRpcMsg* rpcMsg);
+typedef int32_t (*SendMnodeReqFp)(struct SMgmtWrapper* pWrapper, struct SRpcMsg* rpcMsg);
+typedef int32_t (*SendRspFp)(struct SMgmtWrapper* pWrapper, struct SRpcMsg* rpcMsg);
 
 #ifdef __cplusplus
 }

include/dnode/bnode/bnode.h

@@ -21,24 +21,21 @@ extern "C" {
 #endif
 
 /* ------------------------ TYPES EXPOSED ------------------------ */
-typedef struct SDnode SDnode;
-typedef struct SBnode SBnode;
-typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
-typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
-typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
+typedef struct SMgmtWrapper SMgmtWrapper;
+typedef struct SBnode       SBnode;
 
 typedef struct {
   int64_t numOfErrors;
 } SBnodeLoad;
 
 typedef struct {
-  int32_t           sver;
-  int32_t           dnodeId;
-  int64_t           clusterId;
-  SDnode           *pDnode;
-  SendReqToDnodeFp  sendReqFp;
-  SendReqToMnodeFp  sendReqToMnodeFp;
-  SendRedirectRspFp sendRedirectRspFp;
+  int32_t        sver;
+  int32_t        dnodeId;
+  int64_t        clusterId;
+  SMgmtWrapper  *pWrapper;
+  SendReqFp      sendReqFp;
+  SendMnodeReqFp sendMnodeReqFp;
+  SendRspFp      sendRspFp;
 } SBnodeOpt;
 
 /* ------------------------ SBnode ------------------------ */

include/dnode/mnode/mnode.h

@@ -23,13 +23,8 @@ extern "C" {
 #endif
 
 /* ------------------------ TYPES EXPOSED ------------------------ */
-typedef struct SDnode    SDnode;
-typedef struct SMnode    SMnode;
-typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
-typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
-typedef int32_t (*PutReqToMWriteQFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
-typedef int32_t (*PutReqToMReadQFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
-typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
+typedef struct SMgmtWrapper SMgmtWrapper;
+typedef struct SMnode       SMnode;
 
 typedef struct {
   int32_t        dnodeId;
@@ -37,11 +32,12 @@ typedef struct {
   int8_t         replica;
   int8_t         selfIndex;
   SReplica       replicas[TSDB_MAX_REPLICA];
-  SDnode        *pDnode;
-  PutToQueueFp   putReqToMWriteQFp;
-  PutToQueueFp   putReqToMReadQFp;
+  SMgmtWrapper  *pWrapper;
+  PutToQueueFp   putToWriteQFp;
+  PutToQueueFp   putToReadQFp;
   SendReqFp      sendReqFp;
-  SendMnodeReqFp sendReqToMnodeFp;
+  SendMnodeReqFp sendMnodeReqFp;
+  SendRspFp      sendRspFp;
 } SMnodeOpt;
 
 /* ------------------------ SMnode ------------------------ */

include/dnode/qnode/qnode.h

@@ -21,11 +21,8 @@ extern "C" {
 #endif
 
 /* ------------------------ TYPES EXPOSED ------------------------ */
-typedef struct SDnode SDnode;
-typedef struct SQnode SQnode;
-typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
-typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
-typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
+typedef struct SMgmtWrapper SMgmtWrapper;
+typedef struct SQnode       SQnode;
 
 typedef struct {
   int64_t numOfStartTask;
@@ -39,13 +36,13 @@ typedef struct {
 } SQnodeLoad;
 
 typedef struct {
-  int32_t           sver;
-  int32_t           dnodeId;
-  int64_t           clusterId;
-  SDnode           *pDnode;
-  SendReqToDnodeFp  sendReqFp;
-  SendReqToMnodeFp  sendReqToMnodeFp;
-  SendRedirectRspFp sendRedirectRspFp;
+  int32_t        sver;
+  int32_t        dnodeId;
+  int64_t        clusterId;
+  SMgmtWrapper  *pWrapper;
+  SendReqFp      sendReqFp;
+  SendMnodeReqFp sendMnodeReqFp;
+  SendRspFp      sendRspFp;
 } SQnodeOpt;
 
 /* ------------------------ SQnode ------------------------ */

include/dnode/snode/snode.h

@@ -24,24 +24,21 @@ extern "C" {
 #endif
 
 /* ------------------------ TYPES EXPOSED ------------------------ */
-typedef struct SDnode SDnode;
-typedef struct SSnode SSnode;
-typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
-typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
-typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
+typedef struct SMgmtWrapper SMgmtWrapper;
+typedef struct SSnode       SSnode;
 
 typedef struct {
   int64_t numOfErrors;
 } SSnodeLoad;
 
 typedef struct {
-  int32_t           sver;
-  int32_t           dnodeId;
-  int64_t           clusterId;
-  SDnode           *pDnode;
-  SendReqToDnodeFp  sendReqFp;
-  SendReqToMnodeFp  sendReqToMnodeFp;
-  SendRedirectRspFp sendRedirectRspFp;
+  int32_t        sver;
+  int32_t        dnodeId;
+  int64_t        clusterId;
+  SMgmtWrapper  *pWrapper;
+  SendReqFp      sendReqFp;
+  SendMnodeReqFp sendMnodeReqFp;
+  SendRspFp      sendRspFp;
 } SSnodeOpt;
 
 /* ------------------------ SSnode ------------------------ */

source/dnode/mgmt/bnode/src/bmMgmt.c

@@ -177,7 +177,7 @@ static void dndStopBnodeWorker(SDnode *pDnode) {
 static void dndBuildBnodeOption(SDnode *pDnode, SBnodeOpt *pOption) {
   pOption->pDnode = pDnode;
   pOption->sendReqFp = dndSendReqToDnode;
-  pOption->sendReqToMnodeFp = dndSendReqToMnode;
+  pOption->sendMnodeReqFp = dndSendReqToMnode;
   pOption->sendRedirectRspFp = dmSendRedirectRsp;
   pOption->dnodeId = pDnode->dnodeId;
   pOption->clusterId = pDnode->clusterId;

source/dnode/mgmt/container/inc/dnd.h

@@ -149,9 +149,9 @@ void          dndSetMsgHandle(SMgmtWrapper *pWrapper, int32_t msgType, NodeMsgFp
 void          dndReportStartup(SDnode *pDnode, char *pName, char *pDesc);
 void          dndSendMonitorReport(SDnode *pDnode);
 
-int32_t dndSendReqToMnode(void *wrapper, SRpcMsg *pMsg);
-int32_t dndSendReqToDnode(void *wrapper, SEpSet *pEpSet, SRpcMsg *pMsg);
-void    dndSendRsp(void *pWrapper, SRpcMsg *pRsp);
+int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
+int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, SEpSet *pEpSet, SRpcMsg *pMsg);
+void    dndSendRsp(SMgmtWrapper *pWrapper, SRpcMsg *pRsp);
 
 int32_t dndInitWorker(void *param, SDnodeWorker *pWorker, EWorkerType type, const char *name, int32_t minNum,
                       int32_t maxNum, void *queueFp);

source/dnode/mgmt/container/src/dndTransport.c

@@ -279,9 +279,7 @@ static int32_t dndSendRpcReq(STransMgmt *pMgmt, SEpSet *pEpSet, SRpcMsg *pReq) {
   return 0;
 }
 
-int32_t dndSendReqToDnode(void *wrapper, SEpSet *pEpSet, SRpcMsg *pReq) {
-  SMgmtWrapper *pWrapper = wrapper;
-
+int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, SEpSet *pEpSet, SRpcMsg *pReq) {
   if (pWrapper->procType == PROC_CHILD) {
   } else {
     STransMgmt *pTrans = &pWrapper->pDnode->trans;
@@ -289,9 +287,7 @@ int32_t dndSendReqToDnode(void *wrapper, SEpSet *pEpSet, SRpcMsg *pReq) {
   }
 }
 
-int32_t dndSendReqToMnode(void *wrapper, SRpcMsg *pReq) {
-  SMgmtWrapper *pWrapper = wrapper;
-
+int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pReq) {
   if (pWrapper->procType == PROC_CHILD) {
   } else {
     SDnode     *pDnode = pWrapper->pDnode;
@@ -311,9 +307,7 @@ void dndSendRpcRsp(SMgmtWrapper *pWrapper, SRpcMsg *pRsp) {
   }
 }
 
-void dndSendRsp(void *wrapper, SRpcMsg *pRsp) {
-  SMgmtWrapper *pWrapper = wrapper;
-
+void dndSendRsp(SMgmtWrapper *pWrapper, SRpcMsg *pRsp) {
   if (pWrapper->procType == PROC_CHILD) {
     int32_t code = -1;
     do {

source/dnode/mgmt/dnode/src/dmMsg.c

@@ -30,7 +30,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
   req.updateTime = pMgmt->updateTime;
   req.numOfCores = tsNumOfCores;
   req.numOfSupportVnodes = pDnode->numOfSupportVnodes;
-  memcpy(req.dnodeEp, pDnode->localEp, TSDB_EP_LEN);
+  tstrncpy(req.dnodeEp, pDnode->localEp, TSDB_EP_LEN);
 
   req.clusterCfg.statusInterval = tsStatusInterval;
   req.clusterCfg.checkTime = 0;

source/dnode/mgmt/mnode/inc/mmInt.h

@@ -62,8 +62,8 @@ int32_t mmProcessWriteMsg(SMnodeMgmt *pMgmt, SNodeMsg *pMsg);
 int32_t mmProcessSyncMsg(SMnodeMgmt *pMgmt, SNodeMsg *pMsg);
 int32_t mmProcessReadMsg(SMnodeMgmt *pMgmt, SNodeMsg *pMsg);
 
-int32_t mmPutMsgToWriteQueue(void *wrapper, SRpcMsg *pRpcMsg);
-int32_t mmPutMsgToReadQueue(void *wrapper, SRpcMsg *pRpcMsg);
+int32_t mmPutMsgToWriteQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpcMsg);
+int32_t mmPutMsgToReadQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpcMsg);
 
 #ifdef __cplusplus
 }

source/dnode/mgmt/mnode/src/mmInt.c

@@ -135,11 +135,11 @@ int32_t mmDrop(SMnodeMgmt *pMgmt) {
 static void mmInitOption(SMnodeMgmt *pMgmt, SMnodeOpt *pOption) {
   SDnode *pDnode = pMgmt->pDnode;
 
-  pOption->pDnode = pDnode;
+  pOption->pWrapper = pMgmt->pWrapper;
   pOption->sendReqFp = dndSendReqToDnode;
-  pOption->sendReqToMnodeFp = dndSendReqToMnode;
-  pOption->putReqToMWriteQFp = mmPutMsgToWriteQueue;
-  pOption->putReqToMReadQFp = mmPutMsgToReadQueue;
+  pOption->sendMnodeReqFp = dndSendReqToMnode;
+  pOption->putToWriteQFp = mmPutMsgToWriteQueue;
+  pOption->putToReadQFp = mmPutMsgToReadQueue;
   pOption->dnodeId = pDnode->dnodeId;
   pOption->clusterId = pDnode->clusterId;
 }

source/dnode/mgmt/mnode/src/mmWorker.c

@@ -17,7 +17,7 @@
 #include "mmInt.h"
 
 static void mmProcessQueue(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
-  dTrace("msg:%p, will be processed", pMsg);
+  dTrace("msg:%p, will be processed in mnode queue", pMsg);
   SMnode  *pMnode = mmAcquire(pMgmt);
   SRpcMsg *pRpc = &pMsg->rpcMsg;
   bool     isReq = (pRpc->msgType & 1U);
@@ -102,7 +102,7 @@ int32_t mmProcessReadMsg(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
   return mmPutMsgToWorker(pMgmt, &pMgmt->readWorker, pMsg);
 }
 
-static int32_t mmPutRpcMsgToWorker(SMgmtWrapper *pWrapper, SDnodeWorker *pWorker, SRpcMsg *pRpc) {
+static int32_t mmPutRpcMsgToWorker(SMnodeMgmt *pMgmt, SDnodeWorker *pWorker, SRpcMsg *pRpc) {
   SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
   if (pMsg == NULL) {
     return -1;
@@ -111,7 +111,7 @@ static int32_t mmPutRpcMsgToWorker(SMgmtWrapper *pWrapper, SDnodeWorker *pWorker
   dTrace("msg:%p, is created", pMsg);
   pMsg->rpcMsg = *pRpc;
 
-  int32_t code = mmPutMsgToWorker(pWrapper->pMgmt, pWorker, pMsg);
+  int32_t code = mmPutMsgToWorker(pMgmt, pWorker, pMsg);
   if (code != 0) {
     dTrace("msg:%p, is freed", pMsg);
     taosFreeQitem(pMsg);
@@ -121,16 +121,12 @@ static int32_t mmPutRpcMsgToWorker(SMgmtWrapper *pWrapper, SDnodeWorker *pWorker
   return code;
 }
 
-int32_t mmPutMsgToWriteQueue(void *wrapper, SRpcMsg *pRpc) {
-  // SMgmtWrapper *pWrapper = dndGetWrapper(pDnode, MNODE);
-  // SMnodeMgmt   *pMgmt = pWrapper->pMgmt;
-  // return mmPutRpcMsgToWorker(pWrapper, &pMgmt->writeWorker, pRpc);
-  return 0;
+int32_t mmPutMsgToWriteQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
+  SMnodeMgmt *pMgmt = pWrapper->pMgmt;
+  return mmPutRpcMsgToWorker(pMgmt, &pMgmt->writeWorker, pRpc);
 }
 
-int32_t mmPutMsgToReadQueue(void *wrapper, SRpcMsg *pRpc) {
-  // SMgmtWrapper *pWrapper = dndGetWrapper(pDnode, MNODE);
-  // SMnodeMgmt   *pMgmt = pWrapper->pMgmt;
-  // return mmPutRpcMsgToWorker(pWrapper, &pMgmt->readWorker, pRpc);
-  return 0;
+int32_t mmPutMsgToReadQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
+  SMnodeMgmt *pMgmt = pWrapper->pMgmt;
+  return mmPutRpcMsgToWorker(pMgmt, &pMgmt->readWorker, pRpc);
 }

source/dnode/mgmt/qnode/src/qmMgmt.c

@@ -184,7 +184,7 @@ static void dndStopQnodeWorker(SDnode *pDnode) {
 static void dndBuildQnodeOption(SDnode *pDnode, SQnodeOpt *pOption) {
   pOption->pDnode = pDnode;
   pOption->sendReqFp = dndSendReqToDnode;
-  pOption->sendReqToMnodeFp = dndSendReqToMnode;
+  pOption->sendMnodeReqFp = dndSendReqToMnode;
   pOption->sendRedirectRspFp = dndSendRedirectRsp;
   pOption->dnodeId = pDnode->dnodeId;
   pOption->clusterId = pDnode->clusterId;

source/dnode/mgmt/snode/src/smMgmt.c

@@ -209,7 +209,7 @@ static void dndStopSnodeWorker(SDnode *pDnode) {
 static void dndBuildSnodeOption(SDnode *pDnode, SSnodeOpt *pOption) {
   pOption->pDnode = pDnode;
   pOption->sendReqFp = dndSendReqToDnode;
-  pOption->sendReqToMnodeFp = dndSendReqToMnode;
+  pOption->sendMnodeReqFp = dndSendReqToMnode;
   pOption->sendRedirectRspFp = dndSendRedirectRsp;
   pOption->dnodeId = pDnode->dnodeId;
   pOption->clusterId = pDnode->clusterId;

source/dnode/mgmt/vnode/inc/vmInt.h

@@ -34,7 +34,6 @@ typedef struct SVnodesMgmt {
   SWWorkerPool  syncPool;
   SWWorkerPool  writePool;
   const char   *path;
-  SMnode       *pMnode;
   SDnode       *pDnode;
   SMgmtWrapper *pWrapper;
   SDnodeWorker  mgmtWorker;
@@ -100,8 +99,8 @@ void    vmStopWorker(SVnodesMgmt *pMgmt);
 int32_t vmAllocQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode);
 void    vmFreeQueue(SVnodesMgmt *pMgmt, SVnodeObj *pVnode);
 
-int32_t vmPutMsgToQueryQueue(SVnodesMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutMsgToApplyQueue(SVnodesMgmt *pMgmt, int32_t vgId, SRpcMsg *pMsg);
+int32_t vmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
+int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, int32_t vgId, SRpcMsg *pMsg);
 
 int32_t vmProcessWriteMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg);
 int32_t vmProcessSyncMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg);

source/dnode/mgmt/vnode/src/vmInt.c

@@ -257,10 +257,6 @@ static void vmCleanup(SMgmtWrapper *pWrapper) {
   dInfo("vnodes-mgmt is cleaned up");
 }
 
-static int32_t vmSendReqToDnode(SVnodesMgmt *pMgmt, struct SEpSet *epSet, struct SRpcMsg *rpcMsg) {
-  return dndSendReqToDnode(pMgmt->pDnode, epSet, rpcMsg);
-}
-
 static int32_t vmInit(SMgmtWrapper *pWrapper) {
   SDnode      *pDnode = pWrapper->pDnode;
   SVnodesMgmt *pMgmt = calloc(1, sizeof(SVnodesMgmt));
@@ -298,8 +294,8 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
   }
 
   vnodeOpt.nthreads = tsNumOfCommitThreads;
-  vnodeOpt.putToQueryQFp = (VndPutToQueryQFp)vmPutMsgToQueryQueue;
-  vnodeOpt.sendReqFp = (VndSendReqFp)vmSendReqToDnode;
+  vnodeOpt.putToQueryQFp = vmPutMsgToQueryQueue;
+  vnodeOpt.sendReqFp = dndSendReqToDnode;
   if (vnodeInit(&vnodeOpt) != 0) {
     dError("failed to init vnode since %s", terrstr());
     goto _OVER;

source/dnode/mgmt/vnode/src/vmWorker.c

@@ -146,7 +146,9 @@ int32_t vmProcessFetchMsg(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
   }
 }
 
-int32_t vmPutMsgToQueryQueue(SVnodesMgmt *pMgmt, SRpcMsg *pRpc) {
+int32_t vmPutMsgToQueryQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc) {
+  SVnodesMgmt *pMgmt = pWrapper->pMgmt;
+
   int32_t   code = -1;
   SMsgHead *pHead = pRpc->pCont;
   // pHead->vgId = htonl(pHead->vgId);
@@ -162,11 +164,20 @@ int32_t vmPutMsgToQueryQueue(SVnodesMgmt *pMgmt, SRpcMsg *pRpc) {
   return code;
 }
 
-int32_t vmPutMsgToApplyQueue(SVnodesMgmt *pMgmt, int32_t vgId, SRpcMsg *pMsg) {
-  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
+int32_t vmPutMsgToApplyQueue(SMgmtWrapper *pWrapper, int32_t vgId, SRpcMsg *pRpc) {
+  SVnodesMgmt *pMgmt = pWrapper->pMgmt;
+
+  int32_t   code = -1;
+  SMsgHead *pHead = pRpc->pCont;
+  // pHead->vgId = htonl(pHead->vgId);
+
+  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) return -1;
 
-  int32_t code = taosWriteQitem(pVnode->pApplyQ, pMsg);
+  SNodeMsg *pMsg = taosAllocateQitem(sizeof(SNodeMsg));
+  if (pMsg != NULL) {
+    code = vmWriteMsgToQueue(pVnode->pApplyQ, pMsg, false);
+  }
   vmReleaseVnode(pMgmt, pVnode);
   return code;
 }

source/dnode/mnode/impl/inc/mndInt.h

@@ -110,7 +110,7 @@ typedef struct SMnode {
   char             *path;
   int64_t           checkTime;
   SSdb             *pSdb;
-  SDnode           *pDnode;
+  SMgmtWrapper     *pWrapper;
   SArray           *pSteps;
   SShowMgmt         showMgmt;
   SProfileMgmt      profileMgmt;
@@ -120,9 +120,9 @@ typedef struct SMnode {
   SGrantInfo        grant;
   MndMsgFp          msgFp[TDMT_MAX];
   SendReqFp         sendReqFp;
-  SendMnodeReqFp    sendReqToMnodeFp;
-  PutToQueueFp      putReqToMWriteQFp;
-  PutToQueueFp      putReqToMReadQFp;
+  SendMnodeReqFp    sendMnodeReqFp;
+  PutToQueueFp      putToWriteQFp;
+  PutToQueueFp      putToReadQFp;
 } SMnode;
 
 int32_t mndSendReqToDnode(SMnode *pMnode, SEpSet *pEpSet, SRpcMsg *rpcMsg);

source/dnode/mnode/impl/src/mndSubscribe.c

@@ -420,7 +420,7 @@ static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg) {
         .pCont = pRebMsg,
         .contLen = sizeof(SMqDoRebalanceMsg),
     };
-    pMnode->putReqToMWriteQFp(pMnode->pDnode, &rpcMsg);
+    (*pMnode->putToWriteQFp)(pMnode->pWrapper, &rpcMsg);
   } else {
     taosHashCleanup(pRebMsg->rebSubHash);
     rpcFreeCont(pRebMsg);

source/dnode/mnode/impl/src/mnode.c

@@ -49,7 +49,7 @@ int32_t mndSendReqToDnode(SMnode *pMnode, SEpSet *pEpSet, SRpcMsg *pMsg) {
     return -1;
   }
 
-  return (*pMnode->sendReqFp)(pMnode->pDnode, pEpSet, pMsg);
+  return (*pMnode->sendReqFp)(pMnode->pWrapper, pEpSet, pMsg);
 }
 
 int32_t mndSendReqToMnode(SMnode *pMnode, SRpcMsg *pMsg) {
@@ -58,7 +58,7 @@ int32_t mndSendReqToMnode(SMnode *pMnode, SRpcMsg *pMsg) {
     return -1;
   }
 
-  return (*pMnode->sendReqToMnodeFp)(pMnode->pDnode, pMsg);
+  return (*pMnode->sendMnodeReqFp)(pMnode->pWrapper, pMsg);
 }
 
 static void *mndBuildTimerMsg(int32_t *pContLen) {
@@ -80,7 +80,7 @@ static void mndPullupTrans(void *param, void *tmrId) {
     int32_t contLen = 0;
     void   *pReq = mndBuildTimerMsg(&contLen);
     SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRANS_TIMER, .pCont = pReq, .contLen = contLen};
-    pMnode->putReqToMWriteQFp(pMnode->pDnode, &rpcMsg);
+    pMnode->putToWriteQFp(pMnode->pWrapper, &rpcMsg);
   }
 
   taosTmrReset(mndPullupTrans, TRNAS_TIMER_MS, pMnode, pMnode->timer, &pMnode->transTimer);
@@ -96,7 +96,7 @@ static void mndCalMqRebalance(void *param, void *tmrId) {
         .pCont = pReq,
         .contLen = contLen,
     };
-    pMnode->putReqToMReadQFp(pMnode->pDnode, &rpcMsg);
+    pMnode->putToReadQFp(pMnode->pWrapper, &rpcMsg);
   }
 
   taosTmrReset(mndCalMqRebalance, MQ_TIMER_MS, pMnode, pMnode->timer, &pMnode->mqTimer);
@@ -108,7 +108,7 @@ static void mndPullupTelem(void *param, void *tmrId) {
     int32_t contLen = 0;
     void   *pReq = mndBuildTimerMsg(&contLen);
     SRpcMsg rpcMsg = {.msgType = TDMT_MND_TELEM_TIMER, .pCont = pReq, .contLen = contLen};
-    pMnode->putReqToMReadQFp(pMnode->pDnode, &rpcMsg);
+    pMnode->putToReadQFp(pMnode->pWrapper, &rpcMsg);
   }
 
   taosTmrReset(mndPullupTelem, TELEM_TIMER_MS, pMnode, pMnode->timer, &pMnode->telemTimer);
@@ -286,14 +286,14 @@ static int32_t mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) {
   pMnode->replica = pOption->replica;
   pMnode->selfIndex = pOption->selfIndex;
   memcpy(&pMnode->replicas, pOption->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
-  pMnode->pDnode = pOption->pDnode;
-  pMnode->putReqToMWriteQFp = pOption->putReqToMWriteQFp;
-  pMnode->putReqToMReadQFp = pOption->putReqToMReadQFp;
+  pMnode->pWrapper = pOption->pWrapper;
+  pMnode->putToWriteQFp = pOption->putToWriteQFp;
+  pMnode->putToReadQFp = pOption->putToReadQFp;
   pMnode->sendReqFp = pOption->sendReqFp;
-  pMnode->sendReqToMnodeFp = pOption->sendReqToMnodeFp;
+  pMnode->sendMnodeReqFp = pOption->sendMnodeReqFp;
 
-  if (pMnode->sendReqFp == NULL || pMnode->sendReqToMnodeFp == NULL  ||
-      pMnode->putReqToMWriteQFp == NULL || pMnode->dnodeId < 0 || pMnode->clusterId < 0) {
+  if (pMnode->sendReqFp == NULL || pMnode->sendMnodeReqFp == NULL  ||
+      pMnode->putToWriteQFp == NULL || pMnode->dnodeId < 0 || pMnode->clusterId < 0) {
     terrno = TSDB_CODE_MND_INVALID_OPTIONS;
     return -1;
   }

source/dnode/vnode/inc/vnode.h

@@ -30,10 +30,8 @@ extern "C" {
 #endif
 
 /* ------------------------ TYPES EXPOSED ------------------------ */
-typedef struct SVnode SVnode;
-typedef int32_t (*VndPutToQueryQFp)(void *pMgmt, struct SRpcMsg *pReq);
-typedef int32_t (*VndSendReqFp)(void *pMgmt, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
-
+typedef struct SMgmtWrapper SMgmtWrapper;
+typedef struct SVnode       SVnode;
 typedef struct {
   // TODO
   int32_t reserved;
@@ -62,9 +60,11 @@ typedef struct {
 } SVnodeCfg;
 
 typedef struct {
-  uint16_t         nthreads;  // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
-  VndPutToQueryQFp putToQueryQFp;
-  VndSendReqFp     sendReqFp;
+  uint16_t       nthreads;  // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
+  PutToQueueFp   putToQueryQFp;
+  SendReqFp      sendReqFp;
+  SendMnodeReqFp sendMnodeReqFp;
+  SendRspFp      sendRspFp;
 } SVnodeOpt;
 
 typedef struct {

source/dnode/vnode/src/inc/vnd.h

@@ -53,8 +53,8 @@ typedef struct SVnodeMgr {
   pthread_cond_t  hasTask;
   TD_DLIST(SVnodeTask) queue;
   // For vnode Mgmt
-  VndPutToQueryQFp putToQueryQFp;
-  VndSendReqFp     sendReqFp;
+  PutToQueueFp putToQueryQFp;
+  SendReqFp    sendReqFp;
 } SVnodeMgr;
 
 extern SVnodeMgr vnodeMgr;