add msg queue

include/dnode/vnode/vnode.h

@@ -31,8 +31,12 @@ extern "C" {
 
 /* ------------------------ TYPES EXPOSED ------------------------ */
 typedef struct SVnode SVnode;
+typedef struct SDnode SDnode;
+typedef void (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
+
 typedef struct SVnodeCfg {
   int32_t vgId;
+  SDnode *pDnode;
 
   /** vnode buffer pool options */
   struct {
@@ -66,16 +70,13 @@ typedef struct SVnodeCfg {
   SWalCfg walCfg;
 } SVnodeCfg;
 
-typedef struct SDnode SDnode;
-typedef void (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
 typedef struct {
   int32_t           sver;
-  SDnode           *pDnode;
   char             *timezone;
   char             *locale;
   char             *charset;
-  PutReqToVQueryQFp putReqToVQueryQFp;
   uint16_t          nthreads;  // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
+  PutReqToVQueryQFp putReqToVQueryQFp;
 } SVnodeOpt;
 
 /* ------------------------ SVnode ------------------------ */
@@ -100,7 +101,7 @@ void vnodeClear();
  * @param pVnodeCfg options of the vnode
  * @return SVnode* The vnode object
  */
-SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg, int32_t vid);
+SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg);
 
 /**
  * @brief Close a VNODE

source/dnode/mgmt/impl/src/dndVnodes.c

@@ -381,7 +381,8 @@ static void *dnodeOpenVnodeFunc(void *param) {
              pMgmt->openVnodes, pMgmt->totalVnodes);
     dndReportStartup(pDnode, "open-vnodes", stepDesc);
 
-    SVnode *pImpl = vnodeOpen(pCfg->path, NULL, pCfg->vgId);
+    SVnodeCfg cfg = {.pDnode = pDnode, .vgId = pCfg->vgId};
+    SVnode   *pImpl = vnodeOpen(pCfg->path, &cfg);
     if (pImpl == NULL) {
       dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
       pThread->failed++;
@@ -581,7 +582,8 @@ int32_t dndProcessCreateVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
     return -1;
   }
 
-  SVnode *pImpl = vnodeOpen(wrapperCfg.path, NULL /*pCfg*/, pCreate->vgId);
+  vnodeCfg.pDnode = pDnode;
+  SVnode *pImpl = vnodeOpen(wrapperCfg.path, &vnodeCfg);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to create vnode since %s", pCreate->vgId, terrstr());
     return -1;

source/dnode/mgmt/impl/src/dnode.c

@@ -200,12 +200,11 @@ SDnode *dndInit(SDnodeOpt *pOption) {
 
   SVnodeOpt vnodeOpt = {
       .sver = pDnode->opt.sver,
-      .pDnode = pDnode,
       .timezone = pDnode->opt.timezone,
       .locale = pDnode->opt.locale,
       .charset = pDnode->opt.charset,
-      .putReqToVQueryQFp = dndPutMsgToVQueryQ,
       .nthreads = pDnode->opt.numOfCommitThreads,
+      .putReqToVQueryQFp = dndPutMsgToVQueryQ,
   };
   if (vnodeInit(&vnodeOpt) != 0) {
     dError("failed to init vnode env");

source/dnode/vnode/impl/inc/vnodeDef.h

@@ -77,11 +77,12 @@ struct SVnode {
   SVnodeFS*   pFs;
   tsem_t      canCommit;
   SQHandle*   pQuery;
+  SDnode*     pDnode;
 };
 
 int vnodeScheduleTask(SVnodeTask* task);
 
-void vnodePutReqToVQueryQ(struct SRpcMsg *pReq);
+void vnodePutReqToVQueryQ(SVnode *pVnode, struct SRpcMsg *pReq);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/impl/src/vnodeMain.c

@@ -15,27 +15,29 @@
 
 #include "vnodeDef.h"
 
-static SVnode *vnodeNew(const char *path, const SVnodeCfg *pVnodeCfg, int32_t vid);
+static SVnode *vnodeNew(const char *path, const SVnodeCfg *pVnodeCfg);
 static void    vnodeFree(SVnode *pVnode);
 static int     vnodeOpenImpl(SVnode *pVnode);
 static void    vnodeCloseImpl(SVnode *pVnode);
 
-SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg, int32_t vid) {
+SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg) {
   SVnode *pVnode = NULL;
 
   // Set default options
-  //if (pVnodeCfg == NULL) {
-    pVnodeCfg = &defaultVnodeOptions;
-  //}
+  SVnodeCfg cfg = defaultVnodeOptions;
+  if (pVnodeCfg != NULL) {
+    cfg.vgId = pVnodeCfg->vgId;
+    cfg.pDnode = pVnodeCfg->pDnode;
+  }
 
   // Validate options
-  if (vnodeValidateOptions(pVnodeCfg) < 0) {
+  if (vnodeValidateOptions(&cfg) < 0) {
     // TODO
     return NULL;
   }
 
   // Create the handle
-  pVnode = vnodeNew(path, pVnodeCfg, vid);
+  pVnode = vnodeNew(path, &cfg);
   if (pVnode == NULL) {
     // TODO: handle error
     return NULL;
@@ -62,7 +64,7 @@ void vnodeClose(SVnode *pVnode) {
 void vnodeDestroy(const char *path) { taosRemoveDir(path); }
 
 /* ------------------------ STATIC METHODS ------------------------ */
-static SVnode *vnodeNew(const char *path, const SVnodeCfg *pVnodeCfg, int32_t vid) {
+static SVnode *vnodeNew(const char *path, const SVnodeCfg *pVnodeCfg) {
   SVnode *pVnode = NULL;
 
   pVnode = (SVnode *)calloc(1, sizeof(*pVnode));
@@ -71,7 +73,8 @@ static SVnode *vnodeNew(const char *path, const SVnodeCfg *pVnodeCfg, int32_t vi
     return NULL;
   }
 
-  pVnode->vgId = vid;
+  pVnode->vgId = pVnodeCfg->vgId;
+  pVnode->pDnode = pVnodeCfg->pDnode;
   pVnode->path = strdup(path);
   vnodeOptionsCopy(&(pVnode->config), pVnodeCfg);
 

source/dnode/vnode/impl/src/vnodeMgr.c

@@ -26,7 +26,6 @@ int vnodeInit(const SVnodeOpt *pOption) {
 
   vnodeMgr.stop = false;
   vnodeMgr.putReqToVQueryQFp = pOption->putReqToVQueryQFp;
-  vnodeMgr.pDnode = pOption->pDnode;
 
   // Start commit handers
   if (pOption->nthreads > 0) {
@@ -91,10 +90,10 @@ int vnodeScheduleTask(SVnodeTask* pTask) {
   return 0;
 }
 
-void vnodePutReqToVQueryQ(struct SRpcMsg* pReq) {
-  if (vnodeMgr.putReqToVQueryQFp) {
-    (*vnodeMgr.putReqToVQueryQFp)(vnodeMgr.pDnode, pReq);
-  }
+void vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq) {
+  assert(vnodeMgr.putReqToVQueryQFp);
+  assert(pVnode->pDnode);
+  (*vnodeMgr.putReqToVQueryQFp)(pVnode->pDnode, pReq);
 }
 
 /* ------------------------ STATIC METHODS ------------------------ */