From 61d0180b852f93333d086cba388e33c0918675f9 Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Wed, 6 Apr 2022 06:37:37 +0000 Subject: [PATCH] more TDB integration --- source/dnode/vnode/src/meta/metaTDBImpl.c | 125 +++++++++++++++++++--- source/dnode/vnode/src/meta/metaTbUid.c | 2 +- source/dnode/vnode/src/vnd/vnodeCommit.c | 2 +- 3 files changed, 112 insertions(+), 17 deletions(-) diff --git a/source/dnode/vnode/src/meta/metaTDBImpl.c b/source/dnode/vnode/src/meta/metaTDBImpl.c index e934caa7e3..282e0a51f6 100644 --- a/source/dnode/vnode/src/meta/metaTDBImpl.c +++ b/source/dnode/vnode/src/meta/metaTDBImpl.c @@ -16,15 +16,28 @@ #include "metaDef.h" #include "tdbInt.h" +typedef struct SPoolMem { + int64_t size; + struct SPoolMem *prev; + struct SPoolMem *next; +} SPoolMem; + +static SPoolMem *openPool(); +static void clearPool(SPoolMem *pPool); +static void closePool(SPoolMem *pPool); +static void *poolMalloc(void *arg, size_t size); +static void poolFree(void *arg, void *ptr); struct SMetaDB { - TENV *pEnv; - TDB *pTbDB; - TDB *pSchemaDB; - TDB *pNameIdx; - TDB *pStbIdx; - TDB *pNtbIdx; - TDB *pCtbIdx; + TXN txn; + TENV *pEnv; + TDB *pTbDB; + TDB *pSchemaDB; + TDB *pNameIdx; + TDB *pStbIdx; + TDB *pNtbIdx; + TDB *pCtbIdx; + SPoolMem *pPool; }; typedef struct __attribute__((__packed__)) { @@ -167,6 +180,8 @@ int metaOpenDB(SMeta *pMeta) { return -1; } + pMetaDb->pPool = openPool(); + tdbTxnOpen(&pMetaDb->txn, 0, poolMalloc, poolFree, pMetaDb->pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED); tdbBegin(pMetaDb->pEnv, NULL); pMeta->pDB = pMetaDb; @@ -214,7 +229,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) { pVal = pBuf = buf; metaEncodeTbInfo(&pBuf, pTbCfg); vLen = POINTER_DISTANCE(pBuf, buf); - ret = tdbDbInsert(pMetaDb->pTbDB, pKey, kLen, pVal, vLen, NULL); + ret = tdbDbInsert(pMetaDb->pTbDB, pKey, kLen, pVal, vLen, &pMetaDb->txn); if (ret < 0) { return -1; } @@ -236,7 +251,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) { pVal = pBuf = buf; metaEncodeSchemaEx(&pBuf, &schemaWrapper); vLen = POINTER_DISTANCE(pBuf, buf); - ret = tdbDbInsert(pMetaDb->pSchemaDB, pKey, kLen, pVal, vLen, NULL); + ret = tdbDbInsert(pMetaDb->pSchemaDB, pKey, kLen, pVal, vLen, &pMeta->pDB->txn); if (ret < 0) { return -1; } @@ -250,7 +265,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) { kLen = nameLen + 1 + sizeof(uid); pVal = NULL; vLen = 0; - ret = tdbDbInsert(pMetaDb->pNameIdx, pKey, kLen, pVal, vLen, NULL); + ret = tdbDbInsert(pMetaDb->pNameIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn); if (ret < 0) { return -1; } @@ -261,7 +276,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) { kLen = sizeof(uid); pVal = NULL; vLen = 0; - ret = tdbDbInsert(pMetaDb->pStbIdx, pKey, kLen, pVal, vLen, NULL); + ret = tdbDbInsert(pMetaDb->pStbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn); if (ret < 0) { return -1; } @@ -272,7 +287,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) { kLen = sizeof(ctbIdxKey); pVal = NULL; vLen = 0; - ret = tdbDbInsert(pMetaDb->pCtbIdx, pKey, kLen, pVal, vLen, NULL); + ret = tdbDbInsert(pMetaDb->pCtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn); if (ret < 0) { return -1; } @@ -281,12 +296,16 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) { kLen = sizeof(uid); pVal = NULL; vLen = 0; - ret = tdbDbInsert(pMetaDb->pNtbIdx, pKey, kLen, pVal, vLen, NULL); + ret = tdbDbInsert(pMetaDb->pNtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn); if (ret < 0) { return -1; } } + if (pMeta->pDB->pPool->size > 0) { + metaCommit(pMeta); + } + return 0; } @@ -705,6 +724,82 @@ static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg) { } int metaCommit(SMeta *pMeta) { - tdbCommit(pMeta->pDB->pEnv, NULL); + TXN *pTxn = &pMeta->pDB->txn; + + // Commit current txn + tdbCommit(pMeta->pDB->pEnv, pTxn); + tdbTxnClose(pTxn); + clearPool(pMeta->pDB->pPool); + + // start a new txn + tdbTxnOpen(&pMeta->pDB->txn, 0, poolMalloc, poolFree, pMeta->pDB->pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED); + tdbBegin(pMeta->pDB->pEnv, pTxn); return 0; -} \ No newline at end of file +} + +static SPoolMem *openPool() { + SPoolMem *pPool = (SPoolMem *)tdbOsMalloc(sizeof(*pPool)); + + pPool->prev = pPool->next = pPool; + pPool->size = 0; + + return pPool; +} + +static void clearPool(SPoolMem *pPool) { + SPoolMem *pMem; + + do { + pMem = pPool->next; + + if (pMem == pPool) break; + + pMem->next->prev = pMem->prev; + pMem->prev->next = pMem->next; + pPool->size -= pMem->size; + + tdbOsFree(pMem); + } while (1); + + assert(pPool->size == 0); +} + +static void closePool(SPoolMem *pPool) { + clearPool(pPool); + tdbOsFree(pPool); +} + +static void *poolMalloc(void *arg, size_t size) { + void *ptr = NULL; + SPoolMem *pPool = (SPoolMem *)arg; + SPoolMem *pMem; + + pMem = (SPoolMem *)tdbOsMalloc(sizeof(*pMem) + size); + if (pMem == NULL) { + assert(0); + } + + pMem->size = sizeof(*pMem) + size; + pMem->next = pPool->next; + pMem->prev = pPool; + + pPool->next->prev = pMem; + pPool->next = pMem; + pPool->size += pMem->size; + + ptr = (void *)(&pMem[1]); + return ptr; +} + +static void poolFree(void *arg, void *ptr) { + SPoolMem *pPool = (SPoolMem *)arg; + SPoolMem *pMem; + + pMem = &(((SPoolMem *)ptr)[-1]); + + pMem->next->prev = pMem->prev; + pMem->prev->next = pMem->next; + pPool->size -= pMem->size; + + tdbOsFree(pMem); +} diff --git a/source/dnode/vnode/src/meta/metaTbUid.c b/source/dnode/vnode/src/meta/metaTbUid.c index 1f57d1396a..b488630024 100644 --- a/source/dnode/vnode/src/meta/metaTbUid.c +++ b/source/dnode/vnode/src/meta/metaTbUid.c @@ -27,5 +27,5 @@ void metaCloseUidGnrt(SMeta *pMeta) { /* TODO */ tb_uid_t metaGenerateUid(SMeta *pMeta) { // Generate a new table UID - return tGenIdPI32(); + return tGenIdPI64(); } diff --git a/source/dnode/vnode/src/vnd/vnodeCommit.c b/source/dnode/vnode/src/vnd/vnodeCommit.c index 696c5f39f6..1717721bb2 100644 --- a/source/dnode/vnode/src/vnd/vnodeCommit.c +++ b/source/dnode/vnode/src/vnd/vnodeCommit.c @@ -47,7 +47,7 @@ int vnodeSyncCommit(SVnode *pVnode) { static int vnodeCommit(void *arg) { SVnode *pVnode = (SVnode *)arg; - metaCommit(pVnode->pMeta); + // metaCommit(pVnode->pMeta); tqCommit(pVnode->pTq); tsdbCommit(pVnode->pTsdb);