Merge pull request #9750 from taosdata/feature/dnode3

add message queue for vnode

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,15 +70,23 @@ typedef struct SVnodeCfg {
   SWalCfg walCfg;
 } SVnodeCfg;
 
+typedef struct {
+  int32_t           sver;
+  char             *timezone;
+  char             *locale;
+  char             *charset;
+  uint16_t          nthreads;  // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
+  PutReqToVQueryQFp putReqToVQueryQFp;
+} SVnodeOpt;
+
 /* ------------------------ SVnode ------------------------ */
 /**
  * @brief Initialize the vnode module
  *
- * @param nthreads number of commit threads. 0 for no threads and
- *        a schedule queue should be given (TODO)
+ * @param pOption Option of the vnode mnodule
  * @return int 0 for success and -1 for failure
  */
-int vnodeInit(uint16_t nthreads);
+int vnodeInit(const SVnodeOpt *pOption);
 
 /**
  * @brief clear a vnode
@@ -89,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

@@ -22,8 +22,8 @@
 #include "dndTransport.h"
 #include "dndVnodes.h"
 #include "sync.h"
-#include "wal.h"
 #include "tfs.h"
+#include "wal.h"
 
 EStat dndGetStat(SDnode *pDnode) { return pDnode->stat; }
 
@@ -153,6 +153,8 @@ static void dndCleanupEnv(SDnode *pDnode) {
   taosStopCacheRefreshWorker();
 }
 
+static void dndPutMsgToVQueryQ(SDnode *pDnode, SRpcMsg *pRpcMsg) { dndProcessVnodeQueryMsg(pDnode, pRpcMsg, NULL); }
+
 SDnode *dndInit(SDnodeOpt *pOption) {
   taosIgnSIGPIPE();
   taosBlockSIGPIPE();
@@ -196,7 +198,15 @@ SDnode *dndInit(SDnodeOpt *pOption) {
     return NULL;
   }
 
-  if (vnodeInit(pDnode->opt.numOfCommitThreads) != 0) {
+  SVnodeOpt vnodeOpt = {
+      .sver = pDnode->opt.sver,
+      .timezone = pDnode->opt.timezone,
+      .locale = pDnode->opt.locale,
+      .charset = pDnode->opt.charset,
+      .nthreads = pDnode->opt.numOfCommitThreads,
+      .putReqToVQueryQFp = dndPutMsgToVQueryQ,
+  };
+  if (vnodeInit(&vnodeOpt) != 0) {
     dError("failed to init vnode env");
     dndCleanup(pDnode);
     return NULL;

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

@@ -57,6 +57,8 @@ typedef struct SVnodeMgr {
   pthread_cond_t  hasTask;
   TD_DLIST(SVnodeTask) queue;
   // For vnode Mgmt
+  SDnode*           pDnode;
+  PutReqToVQueryQFp putReqToVQueryQFp;
 } SVnodeMgr;
 
 extern SVnodeMgr vnodeMgr;
@@ -75,10 +77,13 @@ struct SVnode {
   SVnodeFS*   pFs;
   tsem_t      canCommit;
   SQHandle*   pQuery;
+  SDnode*     pDnode;
 };
 
 int vnodeScheduleTask(SVnodeTask* task);
 
+void vnodePutReqToVQueryQ(SVnode *pVnode, struct SRpcMsg *pReq);
+
 #ifdef __cplusplus
 }
 #endif

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

@@ -19,17 +19,18 @@ SVnodeMgr vnodeMgr = {.vnodeInitFlag = TD_MOD_UNINITIALIZED};
 
 static void* loop(void* arg);
 
-int vnodeInit(uint16_t nthreads) {
+int vnodeInit(const SVnodeOpt *pOption) {
   if (TD_CHECK_AND_SET_MODE_INIT(&(vnodeMgr.vnodeInitFlag)) == TD_MOD_INITIALIZED) {
     return 0;
   }
 
   vnodeMgr.stop = false;
+  vnodeMgr.putReqToVQueryQFp = pOption->putReqToVQueryQFp;
 
   // Start commit handers
-  if (nthreads > 0) {
-    vnodeMgr.nthreads = nthreads;
-    vnodeMgr.threads = (pthread_t*)calloc(nthreads, sizeof(pthread_t));
+  if (pOption->nthreads > 0) {
+    vnodeMgr.nthreads = pOption->nthreads;
+    vnodeMgr.threads = (pthread_t*)calloc(pOption->nthreads, sizeof(pthread_t));
     if (vnodeMgr.threads == NULL) {
       return -1;
     }
@@ -38,7 +39,7 @@ int vnodeInit(uint16_t nthreads) {
     pthread_cond_init(&(vnodeMgr.hasTask), NULL);
     TD_DLIST_INIT(&(vnodeMgr.queue));
 
-    for (uint16_t i = 0; i < nthreads; i++) {
+    for (uint16_t i = 0; i < pOption->nthreads; i++) {
       pthread_create(&(vnodeMgr.threads[i]), NULL, loop, NULL);
       pthread_setname_np(vnodeMgr.threads[i], "VND Commit Thread");
     }
@@ -89,6 +90,12 @@ int vnodeScheduleTask(SVnodeTask* pTask) {
   return 0;
 }
 
+void vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq) {
+  assert(vnodeMgr.putReqToVQueryQFp);
+  assert(pVnode->pDnode);
+  (*vnodeMgr.putReqToVQueryQFp)(pVnode->pDnode, pReq);
+}
+
 /* ------------------------ STATIC METHODS ------------------------ */
 static void* loop(void* arg) {
   SVnodeTask* pTask;