Merge pull request #9752 from taosdata/feature/vnode

Feature/vnode

include/util/tarray.h

@@ -23,10 +23,23 @@ extern "C" {
 #include "os.h"
 #include "talgo.h"
 
+#if 0
+#define TARRAY(TYPE)             \
+  struct {                       \
+    int32_t      tarray_size_;   \
+    int32_t      tarray_neles_;  \
+    struct TYPE* td_array_data_; \
+  }
+
+#define TARRAY_SIZE(ARRAY) (ARRAY)->tarray_size_
+#define TARRAY_NELES(ARRAY) (ARRAY)->tarray_neles_
+#define TARRAY_ELE_AT(ARRAY, IDX) ((ARRAY)->td_array_data_ + idx)
+#endif
+
 #define TARRAY_MIN_SIZE 8
 #define TARRAY_GET_ELEM(array, index) ((void*)((char*)((array)->pData) + (index) * (array)->elemSize))
-#define TARRAY_ELEM_IDX(array, ele)   (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
-#define TARRAY_GET_START(array)       ((array)->pData)
+#define TARRAY_ELEM_IDX(array, ele) (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
+#define TARRAY_GET_START(array) ((array)->pData)
 
 typedef struct SArray {
   size_t size;
@@ -57,7 +70,7 @@ int32_t taosArrayEnsureCap(SArray* pArray, size_t tsize);
  * @param nEles
  * @return
  */
-void *taosArrayAddBatch(SArray *pArray, const void *pData, int nEles);
+void* taosArrayAddBatch(SArray* pArray, const void* pData, int nEles);
 
 /**
  *
@@ -65,7 +78,7 @@ void *taosArrayAddBatch(SArray *pArray, const void *pData, int nEles);
  * @param pData           position array list
  * @param numOfElems      the number of removed position
  */
-void taosArrayRemoveBatch(SArray *pArray, const int32_t* pData, int32_t numOfElems);
+void taosArrayRemoveBatch(SArray* pArray, const int32_t* pData, int32_t numOfElems);
 
 /**
  *
@@ -73,7 +86,7 @@ void taosArrayRemoveBatch(SArray *pArray, const int32_t* pData, int32_t numOfEle
  * @param comparFn
  * @param fp
  */
-void taosArrayRemoveDuplicate(SArray *pArray, __compar_fn_t comparFn, void (*fp)(void*));
+void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*));
 
 /**
  *  add all element from the source array list into the destination
@@ -242,19 +255,18 @@ int32_t taosArraySearchIdx(const SArray* pArray, const void* key, __compar_fn_t
  */
 char* taosArraySearchString(const SArray* pArray, const char* key, __compar_fn_t comparFn, int flags);
 
-
 /**
  * sort the pointer data in the array
  * @param pArray
- * @param compar 
- * @param param  
+ * @param compar
+ * @param param
  * @return
  */
 
-void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void *param);
+void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param);
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif  /*_TD_UTIL_ARRAY_H*/
+#endif /*_TD_UTIL_ARRAY_H*/

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

@@ -83,8 +83,18 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
         if (metaCreateTable(pVnode->pMeta, pCreateTbReq) < 0) {
           // TODO: handle error
         }
+        if (pCreateTbReq->type == TD_SUPER_TABLE) {
+          free(pCreateTbReq->stbCfg.pSchema);
+          free(pCreateTbReq->stbCfg.pTagSchema);
+        } else if (pCreateTbReq->type == TD_CHILD_TABLE) {
+          free(pCreateTbReq->ctbCfg.pTag);
+        } else {
+          free(pCreateTbReq->ntbCfg.pSchema);
+        }
       }
-      
+      taosArrayDestroy(vCreateTbBatchReq.pArray);
+      break;
+
     case TDMT_VND_DROP_STB:
     case TDMT_VND_DROP_TABLE:
       // if (metaDropTable(pVnode->pMeta, vReq.dtReq.uid) < 0) {

source/libs/tkv/CMakeLists.txt

 aux_source_directory(src TKV_SRC)
 add_library(tkv STATIC ${TKV_SRC})
+# target_include_directories(
+#     tkv
+#     PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/tkv"
+#     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+# )
 target_include_directories(
     tkv
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/tkv"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+    PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+    PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/src/inc"
 )
 target_link_libraries(
     tkv
     PUBLIC os
+    PUBLIC util
 )
\ No newline at end of file

source/libs/tkv/inc/tPage.h → source/libs/tkv/inc/tkv.h

@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef _TD_TKV_PAGE_H_
-#define _TD_TKV_PAGE_H_
+#ifndef _TD_TKV_H_
+#define _TD_TKV_H_
 
 #include "os.h"
 
@@ -22,20 +22,18 @@
 extern "C" {
 #endif
 
-typedef struct {
-  uint16_t dbver;
-  uint16_t pgsize;
-  uint32_t cksm;
-} SPgHdr;
+// Forward declaration
+typedef struct TDB     TDB;
+typedef struct TDB_ENV TDB_ENV;
 
+// SKey
 typedef struct {
-  SPgHdr   chdr;
-  uint16_t used;     // number of used slots
-  uint16_t loffset;  // the offset of the starting location of the last slot used
-} SSlottedPgHdr;
+  void *   bdata;
+  uint32_t size;
+} TDB_KEY, TDB_VALUE;
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif /*_TD_TKV_PAGE_H_*/
\ No newline at end of file
+#endif /*_TD_TKV_H_*/
\ No newline at end of file

source/libs/tkv/src/inc/tkvBufPool.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_TKV_BUF_POOL_H_
+#define _TD_TKV_BUF_POOL_H_
+
+#include "tkvPage.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct STkvBufPool STkvBufPool;
+
+int       tbpOpen(STkvBufPool **ppTkvBufPool);
+int       tbpClose(STkvBufPool *pTkvBufPool);
+STkvPage *tbpNewPage(STkvBufPool *pTkvBufPool);
+int       tbpDelPage(STkvBufPool *pTkvBufPool);
+STkvPage *tbpFetchPage(STkvBufPool *pTkvBufPool, pgid_t pgid);
+int       tbpUnpinPage(STkvBufPool *pTkvBufPool, pgid_t pgid);
+void      tbpFlushPages(STkvBufPool *pTkvBufPool);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_TKV_BUF_POOL_H_*/
\ No newline at end of file

source/libs/tkv/inc/tkvDef.h → source/libs/tkv/src/inc/tkvDB.h

@@ -13,21 +13,19 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef _TD_TKV_DEF_H_
-#define _TD_TKV_DEF_H_
-
-#include "os.h"
+#ifndef _TD_TKV_DB_H_
+#define _TD_TKV_DB_H_
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-// pgid_t
-typedef int32_t pgid_t;
-#define TKV_IVLD_PGID ((pgid_t)-1)
+struct TDB {
+  // TODO
+};
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif /*_TD_TKV_DEF_H_*/
\ No newline at end of file
+#endif /*_TD_TKV_DB_H_*/
\ No newline at end of file

include/libs/tkv/tkv.h → source/libs/tkv/src/inc/tkvDef.h

@@ -13,52 +13,30 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef _TD_TKV_H_
-#define _TD_TKV_H_
+#ifndef _TD_TKV_DEF_H_
+#define _TD_TKV_DEF_H_
 
-#if 0
 #include "os.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-// Types exported
-typedef struct STkvDb        STkvDb;
-typedef struct STkvOpts      STkvOpts;
-typedef struct STkvCache     STkvCache;
-typedef struct STkvReadOpts  STkvReadOpts;
-typedef struct STkvWriteOpts STkvWriteOpts;
+// pgid_t
+typedef int32_t pgid_t;
+#define TKV_IVLD_PGID ((pgid_t)-1)
 
-// DB operations
-STkvDb *tkvOpen(const STkvOpts *options, const char *path);
-void    tkvClose(STkvDb *db);
-void    tkvPut(STkvDb *db, const STkvWriteOpts *, const char *key, size_t keylen, const char *val, size_t vallen);
-char *  tkvGet(STkvDb *db, const STkvReadOpts *, const char *key, size_t keylen, size_t *vallen);
-void    tkvCommit(STkvDb *db);
+// framd_id_t
+typedef int32_t frame_id_t;
 
-// DB options
-STkvOpts *tkvOptsCreate();
-void      tkvOptsDestroy(STkvOpts *);
-void      tkvOptionsSetCache(STkvOpts *, STkvCache *);
-void      tkvOptsSetCreateIfMissing(STkvOpts *, unsigned char);
-
-// DB cache
-typedef enum { TKV_LRU_CACHE = 0, TKV_LFU_CACHE = 1 } ETkvCacheType;
-STkvCache *tkvCacheCreate(size_t capacity, ETkvCacheType type);
-void       tkvCacheDestroy(STkvCache *);
-
-// STkvReadOpts
-STkvReadOpts *tkvReadOptsCreate();
-void          tkvReadOptsDestroy(STkvReadOpts *);
-
-// STkvWriteOpts
-STkvWriteOpts *tkvWriteOptsCreate();
-void           tkvWriteOptsDestroy(STkvWriteOpts *);
+// pgsize_t
+typedef int32_t pgsize_t;
+#define TKV_MIN_PGSIZE 512
+#define TKV_MAX_PGSIZE 16384
+#define TKV_IS_PGSIZE_VLD(s) (((s) >= TKV_MIN_PGSIZE) && (TKV_MAX_PGSIZE <= TKV_MAX_PGSIZE))
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif
-#endif /*_TD_TKV_H_*/
\ No newline at end of file
+#endif /*_TD_TKV_DEF_H_*/
\ No newline at end of file

source/libs/tkv/inc/tDiskMgr.h → source/libs/tkv/src/inc/tkvDiskMgr.h

@@ -24,13 +24,14 @@ extern "C" {
 
 #include "tkvDef.h"
 
-typedef struct SDiskMgr SDiskMgr;
-
-int     tdmOpen(SDiskMgr **ppDiskMgr, const char *fname, uint16_t pgsize);
-int     tdmClose(SDiskMgr *pDiskMgr);
-int     tdmReadPage(SDiskMgr *pDiskMgr, pgid_t pgid, void *pData);
-int     tdmWritePage(SDiskMgr *pDiskMgr, pgid_t pgid, const void *pData);
-int32_t tdmAllocPage(SDiskMgr *pDiskMgr);
+typedef struct STkvDiskMgr STkvDiskMgr;
+
+int    tdmOpen(STkvDiskMgr **ppDiskMgr, const char *fname, uint16_t pgsize);
+int    tdmClose(STkvDiskMgr *pDiskMgr);
+int    tdmReadPage(STkvDiskMgr *pDiskMgr, pgid_t pgid, void *pData);
+int    tdmWritePage(STkvDiskMgr *pDiskMgr, pgid_t pgid, const void *pData);
+int    tdmFlush(STkvDiskMgr *pDiskMgr);
+pgid_t tdmAllocPage(STkvDiskMgr *pDiskMgr);
 
 #ifdef __cplusplus
 }

source/libs/tkv/src/inc/tkvEnv.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_TKV_ENV_H_
+#define _TD_TKV_ENV_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct TDB_ENV {
+  char *homeDir;
+};
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_TKV_ENV_H_*/
\ No newline at end of file

source/libs/tkv/src/inc/tkvPage.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_TKV_PAGE_H_
+#define _TD_TKV_PAGE_H_
+
+#include "os.h"
+#include "tkvDef.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct STkvPage {
+  pgid_t  pgid;
+  int32_t pinCount;
+  bool    idDirty;
+  char*   pData;
+} STkvPage;
+
+typedef struct {
+  uint16_t dbver;
+  uint16_t pgsize;
+  uint32_t cksm;
+} STkvPgHdr;
+
+// typedef struct {
+//   SPgHdr   chdr;
+//   uint16_t used;     // number of used slots
+//   uint16_t loffset;  // the offset of the starting location of the last slot used
+// } SSlottedPgHdr;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_TKV_PAGE_H_*/
\ No newline at end of file

source/libs/tkv/src/tDiskMgr.c

@@ -13,37 +13,62 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "tDiskMgr.h"
+#include "tkvDiskMgr.h"
 
-struct SDiskMgr {
-  const char *fname;
-  uint16_t    pgsize;
-  FileFd      fd;
-  int32_t     npgid;
+struct STkvDiskMgr {
+  char *   fname;
+  uint16_t pgsize;
+  FileFd   fd;
+  pgid_t   npgid;
 };
 
 #define PAGE_OFFSET(PGID, PGSIZE) ((PGID) * (PGSIZE))
 
-int tdmOpen(SDiskMgr **ppDiskMgr, const char *fname, uint16_t pgsize) {
-  // TODO
+int tdmOpen(STkvDiskMgr **ppDiskMgr, const char *fname, uint16_t pgsize) {
+  STkvDiskMgr *pDiskMgr;
+
+  pDiskMgr = malloc(sizeof(*pDiskMgr));
+  if (pDiskMgr == NULL) {
+    return -1;
+  }
+
+  pDiskMgr->fname = strdup(fname);
+  if (pDiskMgr->fname == NULL) {
+    free(pDiskMgr);
+    return -1;
+  }
+  pDiskMgr->pgsize = pgsize;
+  pDiskMgr->fd = open(fname, O_CREAT | O_RDWR, 0755);
+  if (pDiskMgr->fd < 0) {
+    free(pDiskMgr->fname);
+    free(pDiskMgr);
+    return -1;
+  }
+
+  *ppDiskMgr = pDiskMgr;
+
   return 0;
 }
 
-int tdmClose(SDiskMgr *pDiskMgr) {
-  // TODO
+int tdmClose(STkvDiskMgr *pDiskMgr) {
+  close(pDiskMgr->fd);
+  free(pDiskMgr->fname);
+  free(pDiskMgr);
   return 0;
 }
 
-int tdmReadPage(SDiskMgr *pDiskMgr, pgid_t pgid, void *pData) {
+int tdmReadPage(STkvDiskMgr *pDiskMgr, pgid_t pgid, void *pData) {
   taosLSeekFile(pDiskMgr->fd, PAGE_OFFSET(pgid, pDiskMgr->pgsize), SEEK_SET);
   taosReadFile(pDiskMgr->fd, pData, pDiskMgr->pgsize);
   return 0;
 }
 
-int tdmWritePage(SDiskMgr *pDiskMgr, pgid_t pgid, const void *pData) {
+int tdmWritePage(STkvDiskMgr *pDiskMgr, pgid_t pgid, const void *pData) {
   taosLSeekFile(pDiskMgr->fd, PAGE_OFFSET(pgid, pDiskMgr->pgsize), SEEK_SET);
   taosWriteFile(pDiskMgr->fd, pData, pDiskMgr->pgsize);
   return 0;
 }
 
-int32_t tdmAllocPage(SDiskMgr *pDiskMgr) { return pDiskMgr->npgid++; }
\ No newline at end of file
+int tdmFlush(STkvDiskMgr *pDiskMgr) { return taosFsyncFile(pDiskMgr->fd); }
+
+int32_t tdmAllocPage(STkvDiskMgr *pDiskMgr) { return pDiskMgr->npgid++; }
\ No newline at end of file

source/libs/tkv/src/tkvBufPool.c

+/*
+ * 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/>.
+ */
+
+#include "thash.h"
+#include "tlist.h"
+
+#include "tkvBufPool.h"
+#include "tkvDiskMgr.h"
+#include "tkvPage.h"
+
+struct SFrameIdWrapper {
+  TD_SLIST_NODE(SFrameIdWrapper);
+  frame_id_t id;
+};
+
+struct STkvBufPool {
+  STkvPage*    pages;
+  STkvDiskMgr* pDiskMgr;
+  SHashObj*    pgTb;  // page_id_t --> frame_id_t
+  TD_SLIST(SFrameIdWrapper) freeList;
+  pthread_mutex_t mutex;
+};
+
+typedef struct STkvLRUReplacer {
+} STkvLRUReplacer;
+
+typedef struct STkvLFUReplacer {
+} STkvLFUReplacer;
+
+typedef struct STkvCLKReplacer {
+} STkvCLKReplacer;
+
+typedef enum { TKV_LRU_REPLACER = 0, TKV_LFU_REPLACER, TVK_CLK_REPLACER } tkv_replacer_t;
+
+typedef struct STkvReplacer {
+  tkv_replacer_t type;
+  union {
+    STkvLRUReplacer lruRep;
+    STkvLFUReplacer lfuRep;
+    STkvCLKReplacer clkRep;
+  };
+} STkvReplacer;
\ No newline at end of file