From ad466ec7a982ab1fae79ecd1dcf07b88acf0e400 Mon Sep 17 00:00:00 2001 From: Minglei Jin Date: Tue, 2 Aug 2022 15:40:06 +0800 Subject: [PATCH] fix: new API tdbAbort for tdb txn aborting --- source/dnode/vnode/src/meta/metaCommit.c | 5 ++ source/libs/tdb/src/db/tdbDb.c | 21 ++++++-- source/libs/tdb/src/db/tdbPager.c | 68 ++++++++++++++++++++++-- source/libs/tdb/src/inc/tdbInt.h | 1 + 4 files changed, 89 insertions(+), 6 deletions(-) diff --git a/source/dnode/vnode/src/meta/metaCommit.c b/source/dnode/vnode/src/meta/metaCommit.c index 456c4fd7ee..b4987aea2b 100644 --- a/source/dnode/vnode/src/meta/metaCommit.c +++ b/source/dnode/vnode/src/meta/metaCommit.c @@ -18,6 +18,7 @@ static FORCE_INLINE void *metaMalloc(void *pPool, size_t size) { return vnodeBufPoolMalloc((SVBufPool *)pPool, size); } static FORCE_INLINE void metaFree(void *pPool, void *p) { vnodeBufPoolFree((SVBufPool *)pPool, p); } +// begin a meta txn int metaBegin(SMeta *pMeta) { tdbTxnOpen(&pMeta->txn, 0, metaMalloc, metaFree, pMeta->pVnode->inUse, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED); @@ -28,4 +29,8 @@ int metaBegin(SMeta *pMeta) { return 0; } +// commit the meta txn int metaCommit(SMeta *pMeta) { return tdbCommit(pMeta->pEnv, &pMeta->txn); } + +// abort the meta txn +int metaAbort(SMeta *pMeta) { return tdbAbort(pMeta->pEnv, &pMeta->txn); } diff --git a/source/libs/tdb/src/db/tdbDb.c b/source/libs/tdb/src/db/tdbDb.c index 298992a560..cc8bdca75d 100644 --- a/source/libs/tdb/src/db/tdbDb.c +++ b/source/libs/tdb/src/db/tdbDb.c @@ -66,7 +66,7 @@ int32_t tdbOpen(const char *dbname, int32_t szPage, int32_t pages, TDB **ppDb) { #ifdef USE_MAINDB // open main db - ret = tdbTbOpen(TDB_MAINDB_NAME, -1, sizeof(SPgno), NULL, pDb, &pDb->pMainDb); + ret = tdbTbOpen(TDB_MAINDB_NAME, -1, sizeof(SBtInfo), NULL, pDb, &pDb->pMainDb); if (ret < 0) { return -1; } @@ -97,7 +97,7 @@ int tdbClose(TDB *pDb) { return 0; } -int tdbBegin(TDB *pDb, TXN *pTxn) { +int32_t tdbBegin(TDB *pDb, TXN *pTxn) { SPager *pPager; int ret; @@ -112,7 +112,7 @@ int tdbBegin(TDB *pDb, TXN *pTxn) { return 0; } -int tdbCommit(TDB *pDb, TXN *pTxn) { +int32_t tdbCommit(TDB *pDb, TXN *pTxn) { SPager *pPager; int ret; @@ -127,6 +127,21 @@ int tdbCommit(TDB *pDb, TXN *pTxn) { return 0; } +int32_t tdbAbort(TDB *pDb, TXN *pTxn) { + SPager *pPager; + int ret; + + for (pPager = pDb->pgrList; pPager; pPager = pPager->pNext) { + ret = tdbPagerAbort(pPager, pTxn); + if (ret < 0) { + ASSERT(0); + return -1; + } + } + + return 0; +} + SPager *tdbEnvGetPager(TDB *pDb, const char *fname) { u32 hash; SPager **ppPager; diff --git a/source/libs/tdb/src/db/tdbPager.c b/source/libs/tdb/src/db/tdbPager.c index d9a44ba570..4de99e8b1b 100644 --- a/source/libs/tdb/src/db/tdbPager.c +++ b/source/libs/tdb/src/db/tdbPager.c @@ -253,7 +253,70 @@ int tdbPagerCommit(SPager *pPager, TXN *pTxn) { // sync the db file tdbOsFSync(pPager->fd); - // remote the journal file + // remove the journal file + tdbOsClose(pPager->jfd); + tdbOsRemove(pPager->jFileName); + pPager->inTran = 0; + + return 0; +} + +// recovery dirty pages +int tdbPagerAbort(SPager *pPager, TXN *pTxn) { + SPage *pPage; + int pgIdx; + SPgno journalSize = 0; + int ret; + + // 0, sync the journal file + ret = tdbOsFSync(pPager->jfd); + if (ret < 0) { + // TODO + ASSERT(0); + return 0; + } + + tdb_fd_t jfd = tdbOsOpen(pPager->jFileName, TDB_O_RDWR, 0755); + if (jfd == NULL) { + return 0; + } + + ret = tdbGetFileSize(jfd, pPager->pageSize, &journalSize); + if (ret < 0) { + return -1; + } + + // 1, read pages from jounal file + // 2, write original pages to buffered ones + + /* TODO: reset the buffered pages instead of releasing them + // loop to reset the dirty pages from file + for (pgIdx = 0, pPage = pPager->pDirty; pPage != NULL && pgIndex < journalSize; pPage = pPage->pDirtyNext, ++pgIdx) { + // read pgno & the page from journal + SPgno pgno; + + int ret = tdbOsRead(jfd, &pgno, sizeof(pgno)); + if (ret < 0) { + return -1; + } + + ret = tdbOsRead(jfd, pageBuf, pPager->pageSize); + if (ret < 0) { + return -1; + } + } + */ + // 3, release the dirty pages + for (pPage = pPager->pDirty; pPage; pPage = pPager->pDirty) { + pPager->pDirty = pPage->pDirtyNext; + pPage->pDirtyNext = NULL; + + pPage->isDirty = 0; + + tdbPCacheRelease(pPager->pCache, pPage, pTxn); + } + + // 4, remove the journal file tdbOsClose(pPager->jfd); tdbOsRemove(pPager->jFileName); pPager->inTran = 0; @@ -475,8 +538,7 @@ int tdbPagerRestore(SPager *pPager, SBTree *pBt) { for (int pgIndex = 0; pgIndex < journalSize; ++pgIndex) { // read pgno & the page from journal - SPgno pgno; - SPage *pPage; + SPgno pgno; int ret = tdbOsRead(jfd, &pgno, sizeof(pgno)); if (ret < 0) { diff --git a/source/libs/tdb/src/inc/tdbInt.h b/source/libs/tdb/src/inc/tdbInt.h index 1f38cea038..49126b80b6 100644 --- a/source/libs/tdb/src/inc/tdbInt.h +++ b/source/libs/tdb/src/inc/tdbInt.h @@ -189,6 +189,7 @@ int tdbPagerOpenDB(SPager *pPager, SPgno *ppgno, bool toCreate, SBTree *pBt); int tdbPagerWrite(SPager *pPager, SPage *pPage); int tdbPagerBegin(SPager *pPager, TXN *pTxn); int tdbPagerCommit(SPager *pPager, TXN *pTxn); +int tdbPagerAbort(SPager *pPager, TXN *pTxn); int tdbPagerFetchPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage)(SPage *, void *, int), void *arg, TXN *pTxn); void tdbPagerReturnPage(SPager *pPager, SPage *pPage, TXN *pTxn);