more

include/dnode/vnode/meta/meta.h

@@ -71,7 +71,7 @@ typedef struct STbCfg {
 } STbCfg;
 
 // SMeta operations
-SMeta *metaOpen(const char *path, const SMetaCfg *pOptions);
+SMeta *metaOpen(const char *path, const SMetaCfg *pMetaCfg);
 void   metaClose(SMeta *pMeta);
 void   metaRemove(const char *path);
 int    metaCreateTable(SMeta *pMeta, STbCfg *pTbCfg);
@@ -79,8 +79,8 @@ int    metaDropTable(SMeta *pMeta, tb_uid_t uid);
 int    metaCommit(SMeta *pMeta);
 
 // Options
-void metaOptionsInit(SMetaCfg *pOptions);
-void metaOptionsClear(SMetaCfg *pOptions);
+void metaOptionsInit(SMetaCfg *pMetaCfg);
+void metaOptionsClear(SMetaCfg *pMetaCfg);
 
 // STbCfg
 #define META_INIT_STB_CFG(NAME, TTL, KEEP, SUID, PSCHEMA, PTAGSCHEMA)                   \

include/dnode/vnode/tq/tq.h

@@ -264,7 +264,7 @@ typedef struct STQ {
 
 // open in each vnode
 STQ* tqOpen(const char* path, STqCfg* tqConfig, TqLogReader* tqLogReader, SMemAllocatorFactory *allocFac);
-void tqDestroy(STQ*);
+void tqClose(STQ*);
 
 // void* will be replace by a msg type
 int tqPushMsg(STQ*, void* msg, int64_t version);

include/dnode/vnode/tsdb/tsdb.h

@@ -16,6 +16,8 @@
 #ifndef _TD_TSDB_H_
 #define _TD_TSDB_H_
 
+#include "mallocator.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -25,7 +27,7 @@ typedef struct STsdb    STsdb;
 typedef struct STsdbCfg STsdbCfg;
 
 // STsdb
-STsdb *tsdbOpen(const char *path, const STsdbCfg *pTsdbCfg);
+STsdb *tsdbOpen(const char *path, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF);
 void   tsdbClose(STsdb *);
 void   tsdbRemove(const char *path);
 int    tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg);

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

@@ -32,6 +32,8 @@ int   vnodeBufPoolRecycle(SVnode *pVnode);
 void *vnodeMalloc(SVnode *pVnode, uint64_t size);
 bool  vnodeBufPoolIsFull(SVnode *pVnode);
 
+SMemAllocatorFactory *vBufPoolGetMAF(SVnode *pVnode);
+
 #ifdef __cplusplus
 }
 #endif

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

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_VNODE_MAF_H_
+#define _TD_VNODE_MAF_H_
+
+#include "vnode.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int  vnodeOpenMAF(SVnode *pVnode);
+void vnodeCloseMAF(SVnode *pVnode);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_VNODE_MAF_H_*/
\ No newline at end of file

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

@@ -24,8 +24,8 @@ struct SVBufPool {
   TD_DLIST(SVMemAllocator) free;
   TD_DLIST(SVMemAllocator) incycle;
   SVMemAllocator *inuse;
-  // MAF for submodules
-  // SMemAllocatorFactory maf;
+  // MAF for submodules to use
+  SMemAllocatorFactory *pMAF;
 };
 
 int vnodeOpenBufPool(SVnode *pVnode) {
@@ -125,6 +125,8 @@ bool vnodeBufPoolIsFull(SVnode *pVnode) {
   return vmaIsFull(pVnode->pBufPool->inuse);
 }
 
+SMemAllocatorFactory *vBufPoolGetMAF(SVnode *pVnode) { return pVnode->pBufPool->pMAF; }
+
 #if 0
 
 typedef enum {

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

@@ -102,7 +102,7 @@ static int vnodeOpenImpl(SVnode *pVnode) {
 
   // Open tsdb
   sprintf(dir, "%s/tsdb", pVnode->path);
-  pVnode->pTsdb = tsdbOpen(dir, &(pVnode->config.tsdbCfg));
+  pVnode->pTsdb = tsdbOpen(dir, &(pVnode->config.tsdbCfg), vBufPoolGetMAF(pVnode));
   if (pVnode->pTsdb == NULL) {
     // TODO: handle error
     return -1;
@@ -110,7 +110,7 @@ static int vnodeOpenImpl(SVnode *pVnode) {
 
   // TODO: Open TQ
   sprintf(dir, "%s/tq", pVnode->path);
-  pVnode->pTq = tqOpen(dir, &(pVnode->config.tqCfg), NULL, NULL);
+  pVnode->pTq = tqOpen(dir, &(pVnode->config.tqCfg), NULL, vBufPoolGetMAF(pVnode));
   if (pVnode->pTq == NULL) {
     // TODO: handle error
     return -1;
@@ -131,7 +131,9 @@ static int vnodeOpenImpl(SVnode *pVnode) {
 static void vnodeCloseImpl(SVnode *pVnode) {
   if (pVnode) {
     vnodeCloseBufPool(pVnode);
-    tsdbClose(pVnode->pTsdb);
     metaClose(pVnode->pMeta);
+    tsdbClose(pVnode->pTsdb);
+    tqClose(pVnode->pTq);
+    walClose(pVnode->pWal);
   }
 }
\ No newline at end of file

source/dnode/vnode/meta/src/metaMain.c

@@ -17,27 +17,27 @@
 
 #include "metaDef.h"
 
-static SMeta *metaNew(const char *path, const SMetaCfg *pMetaOptions);
+static SMeta *metaNew(const char *path, const SMetaCfg *pMetaCfg);
 static void   metaFree(SMeta *pMeta);
 static int    metaOpenImpl(SMeta *pMeta);
 static void   metaCloseImpl(SMeta *pMeta);
 
-SMeta *metaOpen(const char *path, const SMetaCfg *pMetaOptions) {
+SMeta *metaOpen(const char *path, const SMetaCfg *pMetaCfg) {
   SMeta *pMeta = NULL;
 
   // Set default options
-  if (pMetaOptions == NULL) {
-    pMetaOptions = &defaultMetaOptions;
+  if (pMetaCfg == NULL) {
+    pMetaCfg = &defaultMetaOptions;
   }
 
   // Validate the options
-  if (metaValidateOptions(pMetaOptions) < 0) {
+  if (metaValidateOptions(pMetaCfg) < 0) {
     // TODO: deal with error
     return NULL;
   }
 
   // Allocate handle
-  pMeta = metaNew(path, pMetaOptions);
+  pMeta = metaNew(path, pMetaCfg);
   if (pMeta == NULL) {
     // TODO: handle error
     return NULL;
@@ -65,7 +65,7 @@ void metaClose(SMeta *pMeta) {
 void metaRemove(const char *path) { taosRemoveDir(path); }
 
 /* ------------------------ STATIC METHODS ------------------------ */
-static SMeta *metaNew(const char *path, const SMetaCfg *pMetaOptions) {
+static SMeta *metaNew(const char *path, const SMetaCfg *pMetaCfg) {
   SMeta *pMeta;
   size_t psize = strlen(path);
 
@@ -80,7 +80,7 @@ static SMeta *metaNew(const char *path, const SMetaCfg *pMetaOptions) {
     return NULL;
   }
 
-  metaOptionsCopy(&(pMeta->options), pMetaOptions);
+  metaOptionsCopy(&(pMeta->options), pMetaCfg);
 
   return pMeta;
 };

source/dnode/vnode/tq/src/tq.c

@@ -66,6 +66,10 @@ STQ* tqOpen(const char* path, STqCfg* tqConfig, TqLogReader* tqLogReader, SMemAl
   return pTq;
 }
 
+void tqClose(STQ*pTq) {
+  // TODO
+}
+
 static int tqProtoCheck(TmqMsgHead *pMsg) {
   return pMsg->protoVer == 0;
 }

source/dnode/vnode/tsdb/src/tsdbMain.c

@@ -15,27 +15,27 @@
 
 #include "tsdbDef.h"
 
-static STsdb *tsdbNew(const char *path, const STsdbCfg *pTsdbOptions);
+static STsdb *tsdbNew(const char *path, const STsdbCfg *pTsdbCfg);
 static void   tsdbFree(STsdb *pTsdb);
 static int    tsdbOpenImpl(STsdb *pTsdb);
 static void   tsdbCloseImpl(STsdb *pTsdb);
 
-STsdb *tsdbOpen(const char *path, const STsdbCfg *pTsdbOptions) {
+STsdb *tsdbOpen(const char *path, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF) {
   STsdb *pTsdb = NULL;
 
   // Set default TSDB Options
-  if (pTsdbOptions == NULL) {
-    pTsdbOptions = &defautlTsdbOptions;
+  if (pTsdbCfg == NULL) {
+    pTsdbCfg = &defautlTsdbOptions;
   }
 
   // Validate the options
-  if (tsdbValidateOptions(pTsdbOptions) < 0) {
+  if (tsdbValidateOptions(pTsdbCfg) < 0) {
     // TODO: handle error
     return NULL;
   }
 
   // Create the handle
-  pTsdb = tsdbNew(path, pTsdbOptions);
+  pTsdb = tsdbNew(path, pTsdbCfg);
   if (pTsdb == NULL) {
     // TODO: handle error
     return NULL;
@@ -62,7 +62,7 @@ void tsdbClose(STsdb *pTsdb) {
 void tsdbRemove(const char *path) { taosRemoveDir(path); }
 
 /* ------------------------ STATIC METHODS ------------------------ */
-static STsdb *tsdbNew(const char *path, const STsdbCfg *pTsdbOptions) {
+static STsdb *tsdbNew(const char *path, const STsdbCfg *pTsdbCfg) {
   STsdb *pTsdb = NULL;
 
   pTsdb = (STsdb *)calloc(1, sizeof(STsdb));
@@ -72,7 +72,7 @@ static STsdb *tsdbNew(const char *path, const STsdbCfg *pTsdbOptions) {
   }
 
   pTsdb->path = strdup(path);
-  tsdbOptionsCopy(&(pTsdb->options), pTsdbOptions);
+  tsdbOptionsCopy(&(pTsdb->options), pTsdbCfg);
 
   return pTsdb;
 }