From b7342525c7aeb72d9e1c81aed8c08d37f9697dc2 Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Tue, 18 Oct 2022 13:38:26 +0800 Subject: [PATCH] more code --- source/dnode/vnode/src/inc/tsdb.h | 2 +- source/dnode/vnode/src/inc/vnd.h | 2 ++ source/dnode/vnode/src/inc/vnodeInt.h | 6 ++--- source/dnode/vnode/src/meta/metaOpen.c | 2 +- source/dnode/vnode/src/sma/smaOpen.c | 28 +++++++++++----------- source/dnode/vnode/src/tsdb/tsdbFS.c | 2 +- source/dnode/vnode/src/tsdb/tsdbOpen.c | 4 ++-- source/dnode/vnode/src/vnd/vnodeCommit.c | 30 ++++++++++++++---------- source/dnode/vnode/src/vnd/vnodeOpen.c | 16 +++++++------ 9 files changed, 50 insertions(+), 42 deletions(-) diff --git a/source/dnode/vnode/src/inc/tsdb.h b/source/dnode/vnode/src/inc/tsdb.h index bf110f1ae3..ba028232e8 100644 --- a/source/dnode/vnode/src/inc/tsdb.h +++ b/source/dnode/vnode/src/inc/tsdb.h @@ -247,7 +247,7 @@ void tsdbSmaFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSmaFile *pSmaF, ch // SDelFile void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]); // tsdbFS.c ============================================================================================== -int32_t tsdbFSOpen(STsdb *pTsdb); +int32_t tsdbFSOpen(STsdb *pTsdb, int8_t rollback); int32_t tsdbFSClose(STsdb *pTsdb); int32_t tsdbFSCopy(STsdb *pTsdb, STsdbFS *pFS); void tsdbFSDestroy(STsdbFS *pFS); diff --git a/source/dnode/vnode/src/inc/vnd.h b/source/dnode/vnode/src/inc/vnd.h index aca99ecd2f..988ecc5dd3 100644 --- a/source/dnode/vnode/src/inc/vnd.h +++ b/source/dnode/vnode/src/inc/vnd.h @@ -87,11 +87,13 @@ int32_t vnodeGetBatchMeta(SVnode* pVnode, SRpcMsg* pMsg); int32_t vnodeBegin(SVnode* pVnode); int32_t vnodeShouldCommit(SVnode* pVnode); int32_t vnodeCommit(SVnode* pVnode); +void vnodeRollback(SVnode* pVnode); int32_t vnodeSaveInfo(const char* dir, const SVnodeInfo* pCfg); int32_t vnodeCommitInfo(const char* dir, const SVnodeInfo* pInfo); int32_t vnodeLoadInfo(const char* dir, SVnodeInfo* pInfo); int32_t vnodeSyncCommit(SVnode* pVnode); int32_t vnodeAsyncCommit(SVnode* pVnode); +bool vnodeShouldRollback(SVnode* pVnode); // vnodeSync.c int32_t vnodeSyncOpen(SVnode* pVnode, char* path); diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 712f8bd15b..146c978a96 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -97,7 +97,7 @@ typedef struct SMCtbCursor SMCtbCursor; typedef struct SMStbCursor SMStbCursor; typedef struct STbUidStore STbUidStore; -int metaOpen(SVnode* pVnode, SMeta** ppMeta); +int metaOpen(SVnode* pVnode, SMeta** ppMeta, int8_t rollback); int metaClose(SMeta* pMeta); int metaBegin(SMeta* pMeta, int8_t fromSys); int metaCommit(SMeta* pMeta); @@ -149,7 +149,7 @@ typedef struct { int32_t metaGetStbStats(SMeta* pMeta, int64_t uid, SMetaStbStats* pInfo); // tsdb -int tsdbOpen(SVnode* pVnode, STsdb** ppTsdb, const char* dir, STsdbKeepCfg* pKeepCfg); +int tsdbOpen(SVnode* pVnode, STsdb** ppTsdb, const char* dir, STsdbKeepCfg* pKeepCfg, int8_t rollback); int tsdbClose(STsdb** pTsdb); int32_t tsdbBegin(STsdb* pTsdb); int32_t tsdbCommit(STsdb* pTsdb); @@ -200,7 +200,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem // sma int32_t smaInit(); void smaCleanUp(); -int32_t smaOpen(SVnode* pVnode); +int32_t smaOpen(SVnode* pVnode, int8_t rollback); int32_t smaClose(SSma* pSma); int32_t smaBegin(SSma* pSma); int32_t smaSyncPreCommit(SSma* pSma); diff --git a/source/dnode/vnode/src/meta/metaOpen.c b/source/dnode/vnode/src/meta/metaOpen.c index 515fd31e9d..d605229547 100644 --- a/source/dnode/vnode/src/meta/metaOpen.c +++ b/source/dnode/vnode/src/meta/metaOpen.c @@ -27,7 +27,7 @@ static int taskIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int k static int32_t metaInitLock(SMeta *pMeta) { return taosThreadRwlockInit(&pMeta->lock, NULL); } static int32_t metaDestroyLock(SMeta *pMeta) { return taosThreadRwlockDestroy(&pMeta->lock); } -int metaOpen(SVnode *pVnode, SMeta **ppMeta) { +int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) { SMeta *pMeta = NULL; int ret; int slen; diff --git a/source/dnode/vnode/src/sma/smaOpen.c b/source/dnode/vnode/src/sma/smaOpen.c index ef0d51f0eb..850e4c5697 100644 --- a/source/dnode/vnode/src/sma/smaOpen.c +++ b/source/dnode/vnode/src/sma/smaOpen.c @@ -29,19 +29,19 @@ static int32_t rsmaRestore(SSma *pSma); pKeepCfg->days = smaEvalDays(v, pCfg->retentions, l, pCfg->precision, pCfg->days); \ } while (0) -#define SMA_OPEN_RSMA_IMPL(v, l) \ - do { \ - SRetention *r = (SRetention *)VND_RETENTIONS(v) + l; \ - if (!RETENTION_VALID(r)) { \ - if (l == 0) { \ - goto _err; \ - } \ - break; \ - } \ - smaSetKeepCfg(v, &keepCfg, pCfg, TSDB_TYPE_RSMA_L##l); \ - if (tsdbOpen(v, &SMA_RSMA_TSDB##l(pSma), VNODE_RSMA##l##_DIR, &keepCfg) < 0) { \ - goto _err; \ - } \ +#define SMA_OPEN_RSMA_IMPL(v, l) \ + do { \ + SRetention *r = (SRetention *)VND_RETENTIONS(v) + l; \ + if (!RETENTION_VALID(r)) { \ + if (l == 0) { \ + goto _err; \ + } \ + break; \ + } \ + smaSetKeepCfg(v, &keepCfg, pCfg, TSDB_TYPE_RSMA_L##l); \ + if (tsdbOpen(v, &SMA_RSMA_TSDB##l(pSma), VNODE_RSMA##l##_DIR, &keepCfg, rollback) < 0) { \ + goto _err; \ + } \ } while (0) /** @@ -119,7 +119,7 @@ int smaSetKeepCfg(SVnode *pVnode, STsdbKeepCfg *pKeepCfg, STsdbCfg *pCfg, int ty return 0; } -int32_t smaOpen(SVnode *pVnode) { +int32_t smaOpen(SVnode *pVnode, int8_t rollback) { STsdbCfg *pCfg = &pVnode->config.tsdbCfg; ASSERT(!pVnode->pSma); diff --git a/source/dnode/vnode/src/tsdb/tsdbFS.c b/source/dnode/vnode/src/tsdb/tsdbFS.c index 6fd5629592..f593e26aaa 100644 --- a/source/dnode/vnode/src/tsdb/tsdbFS.c +++ b/source/dnode/vnode/src/tsdb/tsdbFS.c @@ -415,7 +415,7 @@ _err: } // EXPOSED APIS ==================================================================================== -int32_t tsdbFSOpen(STsdb *pTsdb) { +int32_t tsdbFSOpen(STsdb *pTsdb, int8_t rollback) { int32_t code = 0; SVnode *pVnode = pTsdb->pVnode; diff --git a/source/dnode/vnode/src/tsdb/tsdbOpen.c b/source/dnode/vnode/src/tsdb/tsdbOpen.c index 5197823490..efc74b68ba 100644 --- a/source/dnode/vnode/src/tsdb/tsdbOpen.c +++ b/source/dnode/vnode/src/tsdb/tsdbOpen.c @@ -33,7 +33,7 @@ int32_t tsdbSetKeepCfg(STsdb *pTsdb, STsdbCfg *pCfg) { * @param dir * @return int */ -int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKeepCfg) { +int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKeepCfg, int8_t rollback) { STsdb *pTsdb = NULL; int slen = 0; @@ -66,7 +66,7 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee } // open tsdb - if (tsdbFSOpen(pTsdb) < 0) { + if (tsdbFSOpen(pTsdb, rollback) < 0) { goto _err; } diff --git a/source/dnode/vnode/src/vnd/vnodeCommit.c b/source/dnode/vnode/src/vnd/vnodeCommit.c index 07c4c32955..00a544dae3 100644 --- a/source/dnode/vnode/src/vnd/vnodeCommit.c +++ b/source/dnode/vnode/src/vnd/vnodeCommit.c @@ -20,8 +20,6 @@ static int vnodeEncodeInfo(const SVnodeInfo *pInfo, char **ppData); static int vnodeDecodeInfo(uint8_t *pData, SVnodeInfo *pInfo); -static int vnodeStartCommit(SVnode *pVnode); -static int vnodeEndCommit(SVnode *pVnode); static int vnodeCommitImpl(void *arg); static void vnodeWaitCommit(SVnode *pVnode); @@ -241,7 +239,7 @@ int vnodeCommit(SVnode *pVnode) { // preCommit // smaSyncPreCommit(pVnode->pSma); - if(smaAsyncPreCommit(pVnode->pSma) < 0){ + if (smaAsyncPreCommit(pVnode->pSma) < 0) { vError("vgId:%d, failed to async pre-commit sma since %s", TD_VID(pVnode), tstrerror(terrno)); return -1; } @@ -309,6 +307,22 @@ int vnodeCommit(SVnode *pVnode) { return 0; } +bool vnodeShouldRollback(SVnode *pVnode) { + char tFName[TSDB_FILENAME_LEN] = {0}; + snprintf(tFName, TSDB_FILENAME_LEN, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP, + VND_INFO_FNAME_TMP); + + return taosCheckExistFile(tFName); +} + +void vnodeRollback(SVnode *pVnode) { + char tFName[TSDB_FILENAME_LEN] = {0}; + snprintf(tFName, TSDB_FILENAME_LEN, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP, + VND_INFO_FNAME_TMP); + + (void)taosRemoveFile(tFName); +} + static int vnodeCommitImpl(void *arg) { SVnode *pVnode = (SVnode *)arg; @@ -321,16 +335,6 @@ static int vnodeCommitImpl(void *arg) { return 0; } -static int vnodeStartCommit(SVnode *pVnode) { - // TODO - return 0; -} - -static int vnodeEndCommit(SVnode *pVnode) { - // TODO - return 0; -} - static FORCE_INLINE void vnodeWaitCommit(SVnode *pVnode) { tsem_wait(&pVnode->canCommit); } static int vnodeEncodeState(const void *pObj, SJson *pJson) { diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c index 001bb5f7f2..3dbd93bb27 100644 --- a/source/dnode/vnode/src/vnd/vnodeOpen.c +++ b/source/dnode/vnode/src/vnd/vnodeOpen.c @@ -110,6 +110,8 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) { taosThreadMutexInit(&pVnode->mutex, NULL); taosThreadCondInit(&pVnode->poolNotEmpty, NULL); + int8_t rollback = vnodeShouldRollback(pVnode); + // open buffer pool if (vnodeOpenBufPool(pVnode) < 0) { vError("vgId:%d, failed to open vnode buffer pool since %s", TD_VID(pVnode), tstrerror(terrno)); @@ -117,19 +119,19 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) { } // open meta - if (metaOpen(pVnode, &pVnode->pMeta) < 0) { + if (metaOpen(pVnode, &pVnode->pMeta, rollback) < 0) { vError("vgId:%d, failed to open vnode meta since %s", TD_VID(pVnode), tstrerror(terrno)); goto _err; } // open tsdb - if (!VND_IS_RSMA(pVnode) && tsdbOpen(pVnode, &VND_TSDB(pVnode), VNODE_TSDB_DIR, NULL) < 0) { + if (!VND_IS_RSMA(pVnode) && tsdbOpen(pVnode, &VND_TSDB(pVnode), VNODE_TSDB_DIR, NULL, rollback) < 0) { vError("vgId:%d, failed to open vnode tsdb since %s", TD_VID(pVnode), tstrerror(terrno)); goto _err; } // open sma - if (smaOpen(pVnode)) { + if (smaOpen(pVnode, rollback)) { vError("vgId:%d, failed to open vnode sma since %s", TD_VID(pVnode), tstrerror(terrno)); goto _err; } @@ -153,14 +155,12 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) { goto _err; } -#if !VNODE_AS_LIB // open query if (vnodeQueryOpen(pVnode)) { vError("vgId:%d, failed to open vnode query since %s", TD_VID(pVnode), tstrerror(terrno)); terrno = TSDB_CODE_OUT_OF_MEMORY; goto _err; } -#endif // vnode begin if (vnodeBegin(pVnode) < 0) { @@ -169,13 +169,15 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) { goto _err; } -#if !VNODE_AS_LIB // open sync if (vnodeSyncOpen(pVnode, dir)) { vError("vgId:%d, failed to open sync since %s", TD_VID(pVnode), tstrerror(terrno)); goto _err; } -#endif + + if (rollback) { + vnodeRollback(pVnode); + } return pVnode;