xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #17443 from taosdata/fix/commit_txn 

fix: commit txn
This commit is contained in:
Hongze Cheng 2022-10-19 13:15:15 +08:00 committed by GitHub
parent 87586431f9 20b3e2168b
commit eb7de2e571
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
26 changed files with 873 additions and 795 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/util/tutil.h
View File

@ -83,6 +83,12 @@ static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen,
}
}
#define TSDB_CHECK_CODE(CODE, LINO, LABEL) \
if (CODE) { \
LINO = __LINE__; \
goto LABEL; \
}
#ifdef __cplusplus
}
#endif

15
source/dnode/vnode/src/inc/tsdb.h
View File

@ -32,12 +32,6 @@ extern "C" {
#define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TSD ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); }} while(0)
// clang-format on
#define TSDB_CHECK_CODE(CODE, LINO, LABEL) \
if (CODE) { \
LINO = __LINE__; \
goto LABEL; \
}
typedef struct TSDBROW TSDBROW;
typedef struct TABLEID TABLEID;
typedef struct TSDBKEY TSDBKEY;
@ -247,18 +241,17 @@ 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);
int32_t tDFileSetCmprFn(const void *p1, const void *p2);
int32_t tsdbFSCommit1(STsdb *pTsdb, STsdbFS *pFS);
int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFS);
int32_t tsdbFSCommit(STsdb *pTsdb);
int32_t tsdbFSRollback(STsdb *pTsdb);
int32_t tsdbFSPrepareCommit(STsdb *pTsdb, STsdbFS *pFS);
int32_t tsdbFSRef(STsdb *pTsdb, STsdbFS *pFS);
void tsdbFSUnref(STsdb *pTsdb, STsdbFS *pFS);
int32_t tsdbFSRollback(STsdbFS *pFS);
int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet);
int32_t tsdbFSUpsertDelFile(STsdbFS *pFS, SDelFile *pDelFile);
// tsdbReaderWriter.c ==============================================================================================

2
source/dnode/vnode/src/inc/vnd.h
View File

@ -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);

9
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -97,10 +97,11 @@ 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);
int metaFinishCommit(SMeta* pMeta);
int metaCreateSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
int metaAlterSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
int metaDropSTable(SMeta* pMeta, int64_t verison, SVDropStbReq* pReq, SArray* tbUidList);
@ -149,10 +150,12 @@ 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);
int32_t tsdbFinishCommit(STsdb* pTsdb);
int32_t tsdbRollbackCommit(STsdb* pTsdb);
int32_t tsdbDoRetention(STsdb* pTsdb, int64_t now);
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq* pMsg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
@ -200,7 +203,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);

1
source/dnode/vnode/src/meta/metaCommit.c
View File

@ -34,6 +34,7 @@ int metaBegin(SMeta *pMeta, int8_t fromSys) {
// commit the meta txn
int metaCommit(SMeta *pMeta) { return tdbCommit(pMeta->pEnv, &pMeta->txn); }
int metaFinishCommit(SMeta *pMeta) { return tdbPostCommit(pMeta->pEnv, &pMeta->txn); }
// abort the meta txn
int metaAbort(SMeta *pMeta) { return tdbAbort(pMeta->pEnv, &pMeta->txn); }

24
source/dnode/vnode/src/meta/metaOpen.c
View File

@ -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;
@ -60,49 +60,49 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
taosMkDir(pMeta->path);
// open env
ret = tdbOpen(pMeta->path, pVnode->config.szPage, pVnode->config.szCache, &pMeta->pEnv);
ret = tdbOpen(pMeta->path, pVnode->config.szPage, pVnode->config.szCache, &pMeta->pEnv, rollback);
if (ret < 0) {
metaError("vgId:%d, failed to open meta env since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// open pTbDb
ret = tdbTbOpen("table.db", sizeof(STbDbKey), -1, tbDbKeyCmpr, pMeta->pEnv, &pMeta->pTbDb);
ret = tdbTbOpen("table.db", sizeof(STbDbKey), -1, tbDbKeyCmpr, pMeta->pEnv, &pMeta->pTbDb, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta table db since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// open pSkmDb
ret = tdbTbOpen("schema.db", sizeof(SSkmDbKey), -1, skmDbKeyCmpr, pMeta->pEnv, &pMeta->pSkmDb);
ret = tdbTbOpen("schema.db", sizeof(SSkmDbKey), -1, skmDbKeyCmpr, pMeta->pEnv, &pMeta->pSkmDb, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta schema db since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// open pUidIdx
ret = tdbTbOpen("uid.idx", sizeof(tb_uid_t), sizeof(SUidIdxVal), uidIdxKeyCmpr, pMeta->pEnv, &pMeta->pUidIdx);
ret = tdbTbOpen("uid.idx", sizeof(tb_uid_t), sizeof(SUidIdxVal), uidIdxKeyCmpr, pMeta->pEnv, &pMeta->pUidIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta uid idx since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// open pNameIdx
ret = tdbTbOpen("name.idx", -1, sizeof(tb_uid_t), NULL, pMeta->pEnv, &pMeta->pNameIdx);
ret = tdbTbOpen("name.idx", -1, sizeof(tb_uid_t), NULL, pMeta->pEnv, &pMeta->pNameIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta name index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// open pCtbIdx
ret = tdbTbOpen("ctb.idx", sizeof(SCtbIdxKey), -1, ctbIdxKeyCmpr, pMeta->pEnv, &pMeta->pCtbIdx);
ret = tdbTbOpen("ctb.idx", sizeof(SCtbIdxKey), -1, ctbIdxKeyCmpr, pMeta->pEnv, &pMeta->pCtbIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta child table index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// open pSuidIdx
ret = tdbTbOpen("suid.idx", sizeof(tb_uid_t), 0, uidIdxKeyCmpr, pMeta->pEnv, &pMeta->pSuidIdx);
ret = tdbTbOpen("suid.idx", sizeof(tb_uid_t), 0, uidIdxKeyCmpr, pMeta->pEnv, &pMeta->pSuidIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta super table index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
@ -119,27 +119,27 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
goto _err;
}
ret = tdbTbOpen("tag.idx", -1, 0, tagIdxKeyCmpr, pMeta->pEnv, &pMeta->pTagIdx);
ret = tdbTbOpen("tag.idx", -1, 0, tagIdxKeyCmpr, pMeta->pEnv, &pMeta->pTagIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta tag index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// open pTtlIdx
ret = tdbTbOpen("ttl.idx", sizeof(STtlIdxKey), 0, ttlIdxKeyCmpr, pMeta->pEnv, &pMeta->pTtlIdx);
ret = tdbTbOpen("ttl.idx", sizeof(STtlIdxKey), 0, ttlIdxKeyCmpr, pMeta->pEnv, &pMeta->pTtlIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta ttl index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// open pSmaIdx
ret = tdbTbOpen("sma.idx", sizeof(SSmaIdxKey), 0, smaIdxKeyCmpr, pMeta->pEnv, &pMeta->pSmaIdx);
ret = tdbTbOpen("sma.idx", sizeof(SSmaIdxKey), 0, smaIdxKeyCmpr, pMeta->pEnv, &pMeta->pSmaIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta sma index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
ret = tdbTbOpen("stream.task.db", sizeof(int64_t), -1, taskIdxKeyCmpr, pMeta->pEnv, &pMeta->pStreamDb);
ret = tdbTbOpen("stream.task.db", sizeof(int64_t), -1, taskIdxKeyCmpr, pMeta->pEnv, &pMeta->pStreamDb, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta stream task index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;

2
source/dnode/vnode/src/meta/metaSnapshot.c
View File

@ -165,6 +165,8 @@ int32_t metaSnapWriterClose(SMetaSnapWriter** ppWriter, int8_t rollback) {
} else {
code = metaCommit(pWriter->pMeta);
if (code) goto _err;
code = metaFinishCommit(pWriter->pMeta);
if (code) goto _err;
}
taosMemoryFree(pWriter);
*ppWriter = NULL;

28
source/dnode/vnode/src/sma/smaOpen.c
View File

@ -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);

6
source/dnode/vnode/src/tq/tqMeta.c
View File

@ -70,17 +70,17 @@ int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle) {
}
int32_t tqMetaOpen(STQ* pTq) {
if (tdbOpen(pTq->path, 16 * 1024, 1, &pTq->pMetaDB) < 0) {
if (tdbOpen(pTq->path, 16 * 1024, 1, &pTq->pMetaDB, 0) < 0) {
ASSERT(0);
return -1;
}
if (tdbTbOpen("tq.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pExecStore) < 0) {
if (tdbTbOpen("tq.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pExecStore, 0) < 0) {
ASSERT(0);
return -1;
}
if (tdbTbOpen("tq.check.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pCheckStore) < 0) {
if (tdbTbOpen("tq.check.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pCheckStore, 0) < 0) {
ASSERT(0);
return -1;
}

81
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -1041,38 +1041,20 @@ _exit:
static int32_t tsdbEndCommit(SCommitter *pCommitter, int32_t eno) {
int32_t code = 0;
int32_t lino = 0;
STsdb *pTsdb = pCommitter->pTsdb;
STsdb *pTsdb = pCommitter->pTsdb;
SMemTable *pMemTable = pTsdb->imem;
ASSERT(eno == 0 &&
"tsdbCommit failure"
"Restart taosd");
code = tsdbFSCommit1(pTsdb, &pCommitter->fs);
TSDB_CHECK_CODE(code, lino, _exit);
// lock
taosThreadRwlockWrlock(&pTsdb->rwLock);
// commit or rollback
code = tsdbFSCommit2(pTsdb, &pCommitter->fs);
if (code) {
taosThreadRwlockUnlock(&pTsdb->rwLock);
if (eno) {
code = eno;
TSDB_CHECK_CODE(code, lino, _exit);
} else {
code = tsdbFSPrepareCommit(pCommitter->pTsdb, &pCommitter->fs);
TSDB_CHECK_CODE(code, lino, _exit);
}
pTsdb->imem = NULL;
// unlock
taosThreadRwlockUnlock(&pTsdb->rwLock);
tsdbUnrefMemTable(pMemTable);
_exit:
tsdbFSDestroy(&pCommitter->fs);
taosArrayDestroy(pCommitter->aTbDataP);
_exit:
if (code) {
if (code || eno) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
} else {
tsdbInfo("vgId:%d tsdb end commit", TD_VID(pTsdb->pVnode));
@ -1646,3 +1628,50 @@ _exit:
}
return code;
}
int32_t tsdbFinishCommit(STsdb *pTsdb) {
int32_t code = 0;
int32_t lino = 0;
SMemTable *pMemTable = pTsdb->imem;
// lock
taosThreadRwlockWrlock(&pTsdb->rwLock);
code = tsdbFSCommit(pTsdb);
if (code) {
taosThreadRwlockUnlock(&pTsdb->rwLock);
TSDB_CHECK_CODE(code, lino, _exit);
}
pTsdb->imem = NULL;
// unlock
taosThreadRwlockUnlock(&pTsdb->rwLock);
if (pMemTable) {
tsdbUnrefMemTable(pMemTable);
}
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
} else {
tsdbInfo("vgId:%d tsdb finish commit", TD_VID(pTsdb->pVnode));
}
return code;
}
int32_t tsdbRollbackCommit(STsdb *pTsdb) {
int32_t code = 0;
int32_t lino = 0;
code = tsdbFSRollback(pTsdb);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
} else {
tsdbInfo("vgId:%d tsdb rollback commit", TD_VID(pTsdb->pVnode));
}
return code;
}

1275
source/dnode/vnode/src/tsdb/tsdbFS.c
View File

File diff suppressed because it is too large Load Diff

4
source/dnode/vnode/src/tsdb/tsdbOpen.c
View File

@ -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;
}

4
source/dnode/vnode/src/tsdb/tsdbRetention.c
View File

@ -86,12 +86,12 @@ int32_t tsdbDoRetention(STsdb *pTsdb, int64_t now) {
}
// do change fs
code = tsdbFSCommit1(pTsdb, &fs);
code = tsdbFSPrepareCommit(pTsdb, &fs);
if (code) goto _err;
taosThreadRwlockWrlock(&pTsdb->rwLock);
code = tsdbFSCommit2(pTsdb, &fs);
code = tsdbFSCommit(pTsdb);
if (code) {
taosThreadRwlockUnlock(&pTsdb->rwLock);
goto _err;

4
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
View File

@ -1380,13 +1380,13 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) {
code = tsdbSnapWriteDelEnd(pWriter);
if (code) goto _err;
code = tsdbFSCommit1(pWriter->pTsdb, &pWriter->fs);
code = tsdbFSPrepareCommit(pWriter->pTsdb, &pWriter->fs);
if (code) goto _err;
// lock
taosThreadRwlockWrlock(&pTsdb->rwLock);
code = tsdbFSCommit2(pWriter->pTsdb, &pWriter->fs);
code = tsdbFSCommit(pWriter->pTsdb);
if (code) {
taosThreadRwlockUnlock(&pTsdb->rwLock);
goto _err;

68
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -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);
@ -215,6 +213,8 @@ int vnodeSyncCommit(SVnode *pVnode) {
}
int vnodeCommit(SVnode *pVnode) {
int32_t code = 0;
int32_t lino = 0;
SVnodeInfo info = {0};
char dir[TSDB_FILENAME_LEN];
@ -234,8 +234,8 @@ int vnodeCommit(SVnode *pVnode) {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
}
if (vnodeSaveInfo(dir, &info) < 0) {
vError("vgId:%d, failed to save vnode info since %s", TD_VID(pVnode), tstrerror(terrno));
return -1;
code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
}
walBeginSnapshot(pVnode->pWal, pVnode->state.applied);
@ -249,8 +249,8 @@ int vnodeCommit(SVnode *pVnode) {
// commit each sub-system
if (metaCommit(pVnode->pMeta) < 0) {
vError("vgId:%d, failed to commit meta since %s", TD_VID(pVnode), tstrerror(terrno));
return -1;
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _exit);
}
if (VND_IS_RSMA(pVnode)) {
@ -259,22 +259,26 @@ int vnodeCommit(SVnode *pVnode) {
return -1;
}
} else {
if (tsdbCommit(pVnode->pTsdb) < 0) {
vError("vgId:%d, failed to commit tsdb since %s", TD_VID(pVnode), tstrerror(terrno));
return -1;
}
code = tsdbCommit(pVnode->pTsdb);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (tqCommit(pVnode->pTq) < 0) {
vError("vgId:%d, failed to commit tq since %s", TD_VID(pVnode), tstrerror(terrno));
return -1;
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _exit);
}
// walCommit (TODO)
// commit info
if (vnodeCommitInfo(dir, &info) < 0) {
vError("vgId:%d, failed to commit vnode info since %s", TD_VID(pVnode), tstrerror(terrno));
return -1;
code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
}
tsdbFinishCommit(pVnode->pTsdb);
if (metaFinishCommit(pVnode->pMeta) < 0) {
code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
}
pVnode->state.committed = info.state.committed;
@ -287,11 +291,31 @@ int vnodeCommit(SVnode *pVnode) {
// apply the commit (TODO)
walEndSnapshot(pVnode->pWal);
vInfo("vgId:%d, commit end", TD_VID(pVnode));
_exit:
if (code) {
vError("vgId:%d %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
} else {
vInfo("vgId:%d, commit end", TD_VID(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;
@ -304,16 +328,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) {

16
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -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;

6
source/libs/stream/src/streamMeta.c
View File

@ -27,7 +27,7 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
char* streamPath = taosMemoryCalloc(1, len);
sprintf(streamPath, "%s/%s", path, "stream");
pMeta->path = strdup(streamPath);
if (tdbOpen(pMeta->path, 16 * 1024, 1, &pMeta->db) < 0) {
if (tdbOpen(pMeta->path, 16 * 1024, 1, &pMeta->db, 0) < 0) {
taosMemoryFree(streamPath);
goto _err;
}
@ -36,11 +36,11 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
taosMulModeMkDir(streamPath, 0755);
taosMemoryFree(streamPath);
if (tdbTbOpen("task.db", sizeof(int32_t), -1, NULL, pMeta->db, &pMeta->pTaskDb) < 0) {
if (tdbTbOpen("task.db", sizeof(int32_t), -1, NULL, pMeta->db, &pMeta->pTaskDb, 0) < 0) {
goto _err;
}
if (tdbTbOpen("checkpoint.db", sizeof(int32_t), -1, NULL, pMeta->db, &pMeta->pCheckpointDb) < 0) {
if (tdbTbOpen("checkpoint.db", sizeof(int32_t), -1, NULL, pMeta->db, &pMeta->pCheckpointDb, 0) < 0) {
goto _err;
}

10
source/libs/stream/src/streamState.c
View File

@ -99,26 +99,26 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath, int
memset(statePath, 0, 300);
tstrncpy(statePath, path, 300);
}
if (tdbOpen(statePath, szPage, pages, &pState->db) < 0) {
if (tdbOpen(statePath, szPage, pages, &pState->db, 0) < 0) {
goto _err;
}
// open state storage backend
if (tdbTbOpen("state.db", sizeof(SStateKey), -1, stateKeyCmpr, pState->db, &pState->pStateDb) < 0) {
if (tdbTbOpen("state.db", sizeof(SStateKey), -1, stateKeyCmpr, pState->db, &pState->pStateDb, 0) < 0) {
goto _err;
}
// todo refactor
if (tdbTbOpen("fill.state.db", sizeof(SWinKey), -1, winKeyCmpr, pState->db, &pState->pFillStateDb) < 0) {
if (tdbTbOpen("fill.state.db", sizeof(SWinKey), -1, winKeyCmpr, pState->db, &pState->pFillStateDb, 0) < 0) {
goto _err;
}
if (tdbTbOpen("session.state.db", sizeof(SStateSessionKey), -1, stateSessionKeyCmpr, pState->db,
&pState->pSessionStateDb) < 0) {
&pState->pSessionStateDb, 0) < 0) {
goto _err;
}
if (tdbTbOpen("func.state.db", sizeof(STupleKey), -1, STupleKeyCmpr, pState->db, &pState->pFuncStateDb) < 0) {
if (tdbTbOpen("func.state.db", sizeof(STupleKey), -1, STupleKeyCmpr, pState->db, &pState->pFuncStateDb, 0) < 0) {
goto _err;
}

6
source/libs/tdb/inc/tdb.h
View File

@ -31,15 +31,17 @@ typedef struct STBC TBC;
typedef struct STxn TXN;
// TDB
int32_t tdbOpen(const char *dbname, int szPage, int pages, TDB **ppDb);
int32_t tdbOpen(const char *dbname, int szPage, int pages, TDB **ppDb, int8_t rollback);
int32_t tdbClose(TDB *pDb);
int32_t tdbBegin(TDB *pDb, TXN *pTxn);
int32_t tdbCommit(TDB *pDb, TXN *pTxn);
int32_t tdbPostCommit(TDB *pDb, TXN *pTxn);
int32_t tdbAbort(TDB *pDb, TXN *pTxn);
int32_t tdbAlter(TDB *pDb, int pages);
// TTB
int32_t tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn, TDB *pEnv, TTB **ppTb);
int32_t tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn, TDB *pEnv, TTB **ppTb,
int8_t rollback);
int32_t tdbTbClose(TTB *pTb);
int32_t tdbTbDrop(TTB *pTb);
int32_t tdbTbInsert(TTB *pTb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn);

19
source/libs/tdb/src/db/tdbDb.c
View File

@ -15,7 +15,7 @@
#include "tdbInt.h"
int32_t tdbOpen(const char *dbname, int32_t szPage, int32_t pages, TDB **ppDb) {
int32_t tdbOpen(const char *dbname, int32_t szPage, int32_t pages, TDB **ppDb, int8_t rollback) {
TDB *pDb;
int dsize;
int zsize;
@ -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(SBtInfo), NULL, pDb, &pDb->pMainDb);
ret = tdbTbOpen(TDB_MAINDB_NAME, -1, sizeof(SBtInfo), NULL, pDb, &pDb->pMainDb, rollback);
if (ret < 0) {
return -1;
}
@ -129,6 +129,21 @@ int32_t tdbCommit(TDB *pDb, TXN *pTxn) {
return 0;
}
int32_t tdbPostCommit(TDB *pDb, TXN *pTxn) {
SPager *pPager;
int ret;
for (pPager = pDb->pgrList; pPager; pPager = pPager->pNext) {
ret = tdbPagerPostCommit(pPager, pTxn);
if (ret < 0) {
tdbError("failed to commit pager since %s. dbName:%s, txnId:%d", tstrerror(terrno), pDb->dbName, pTxn->txnId);
return -1;
}
}
return 0;
}
int32_t tdbAbort(TDB *pDb, TXN *pTxn) {
SPager *pPager;
int ret;

22
source/libs/tdb/src/db/tdbPager.c
View File

@ -305,6 +305,18 @@ int tdbPagerCommit(SPager *pPager, TXN *pTxn) {
return 0;
}
int tdbPagerPostCommit(SPager *pPager, TXN *pTxn) {
if (tdbOsRemove(pPager->jFileName) < 0 && errno != ENOENT) {
tdbError("failed to remove file due to %s. file:%s", strerror(errno), pPager->jFileName);
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
pPager->inTran = 0;
return 0;
}
// recovery dirty pages
int tdbPagerAbort(SPager *pPager, TXN *pTxn) {
SPage *pPage;
@ -657,3 +669,13 @@ int tdbPagerRestore(SPager *pPager, SBTree *pBt) {
return 0;
}
int tdbPagerRollback(SPager *pPager) {
if (tdbOsRemove(pPager->jFileName) < 0 && errno != ENOENT) {
tdbError("failed to remove file due to %s. jFileName:%s", strerror(errno), pPager->jFileName);
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
return 0;
}

15
source/libs/tdb/src/db/tdbTable.c
View File

@ -24,7 +24,8 @@ struct STBC {
SBTC btc;
};
int tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn, TDB *pEnv, TTB **ppTb) {
int tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn, TDB *pEnv, TTB **ppTb,
int8_t rollback) {
TTB *pTb;
SPager *pPager;
int ret;
@ -110,10 +111,14 @@ int tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprF
return -1;
}
ret = tdbPagerRestore(pPager, pTb->pBt);
if (ret < 0) {
tdbOsFree(pTb);
return -1;
if (rollback) {
tdbPagerRollback(pPager);
} else {
ret = tdbPagerRestore(pPager, pTb->pBt);
if (ret < 0) {
tdbOsFree(pTb);
return -1;
}
}
*ppTb = pTb;

2
source/libs/tdb/src/inc/tdbInt.h
View File

@ -190,12 +190,14 @@ 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 tdbPagerPostCommit(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);
int tdbPagerAllocPage(SPager *pPager, SPgno *ppgno);
int tdbPagerRestore(SPager *pPager, SBTree *pBt);
int tdbPagerRollback(SPager *pPager);
// tdbPCache.c ====================================
#define TDB_PCACHE_PAGE \

16
source/libs/tdb/test/tdbExOVFLTest.cpp
View File

@ -140,7 +140,7 @@ static void generateBigVal(char *val, int valLen) {
static TDB *openEnv(char const *envName, int const pageSize, int const pageNum) {
TDB *pEnv = NULL;
int ret = tdbOpen(envName, pageSize, pageNum, &pEnv);
int ret = tdbOpen(envName, pageSize, pageNum, &pEnv, 0);
if (ret) {
pEnv = NULL;
}
@ -162,8 +162,8 @@ static void insertOfp(void) {
// open db
TTB *pDb = NULL;
tdb_cmpr_fn_t compFunc = tKeyCmpr;
// ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb);
ret = tdbTbOpen("ofp_insert.db", 12, -1, compFunc, pEnv, &pDb);
// ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb, 0);
ret = tdbTbOpen("ofp_insert.db", 12, -1, compFunc, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
// open the pool
@ -211,8 +211,8 @@ TEST(TdbOVFLPagesTest, TbGetTest) {
// open db
TTB *pDb = NULL;
tdb_cmpr_fn_t compFunc = tKeyCmpr;
// int ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb);
int ret = tdbTbOpen("ofp_insert.db", 12, -1, compFunc, pEnv, &pDb);
// int ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb, 0);
int ret = tdbTbOpen("ofp_insert.db", 12, -1, compFunc, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
// generate value payload
@ -253,7 +253,7 @@ TEST(TdbOVFLPagesTest, TbDeleteTest) {
// open db
TTB *pDb = NULL;
tdb_cmpr_fn_t compFunc = tKeyCmpr;
ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb);
ret = tdbTbOpen("ofp_insert.db", -1, -1, compFunc, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
// open the pool
@ -354,12 +354,12 @@ TEST(tdb_test, simple_insert1) {
taosRemoveDir("tdb");
// Open Env
ret = tdbOpen("tdb", pageSize, 64, &pEnv);
ret = tdbOpen("tdb", pageSize, 64, &pEnv, 0);
GTEST_ASSERT_EQ(ret, 0);
// Create a database
compFunc = tKeyCmpr;
ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb);
ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
{

26
source/libs/tdb/test/tdbTest.cpp
View File

@ -130,12 +130,12 @@ TEST(tdb_test, DISABLED_simple_insert1) {
taosRemoveDir("tdb");
// Open Env
ret = tdbOpen("tdb", 4096, 64, &pEnv);
ret = tdbOpen("tdb", 4096, 64, &pEnv, 0);
GTEST_ASSERT_EQ(ret, 0);
// Create a database
compFunc = tKeyCmpr;
ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb);
ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
{
@ -250,12 +250,12 @@ TEST(tdb_test, DISABLED_simple_insert2) {
taosRemoveDir("tdb");
// Open Env
ret = tdbOpen("tdb", 1024, 10, &pEnv);
ret = tdbOpen("tdb", 1024, 10, &pEnv, 0);
GTEST_ASSERT_EQ(ret, 0);
// Create a database
compFunc = tDefaultKeyCmpr;
ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb);
ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
{
@ -346,11 +346,11 @@ TEST(tdb_test, DISABLED_simple_delete1) {
pPool = openPool();
// open env
ret = tdbOpen("tdb", 1024, 256, &pEnv);
ret = tdbOpen("tdb", 1024, 256, &pEnv, 0);
GTEST_ASSERT_EQ(ret, 0);
// open database
ret = tdbTbOpen("db.db", -1, -1, tKeyCmpr, pEnv, &pDb);
ret = tdbTbOpen("db.db", -1, -1, tKeyCmpr, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
tdbTxnOpen(&txn, 0, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
@ -435,11 +435,11 @@ TEST(tdb_test, DISABLED_simple_upsert1) {
taosRemoveDir("tdb");
// open env
ret = tdbOpen("tdb", 4096, 64, &pEnv);
ret = tdbOpen("tdb", 4096, 64, &pEnv, 0);
GTEST_ASSERT_EQ(ret, 0);
// open database
ret = tdbTbOpen("db.db", -1, -1, NULL, pEnv, &pDb);
ret = tdbTbOpen("db.db", -1, -1, NULL, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
pPool = openPool();
@ -497,12 +497,12 @@ TEST(tdb_test, multi_thread_query) {
taosRemoveDir("tdb");
// Open Env
ret = tdbOpen("tdb", 4096, 10, &pEnv);
ret = tdbOpen("tdb", 4096, 10, &pEnv, 0);
GTEST_ASSERT_EQ(ret, 0);
// Create a database
compFunc = tKeyCmpr;
ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb);
ret = tdbTbOpen("db.db", -1, -1, compFunc, pEnv, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
char key[64];
@ -614,10 +614,10 @@ TEST(tdb_test, DISABLED_multi_thread1) {
taosRemoveDir("tdb");
// Open Env
ret = tdbOpen("tdb", 512, 1, &pDb);
ret = tdbOpen("tdb", 512, 1, &pDb, 0);
GTEST_ASSERT_EQ(ret, 0);
ret = tdbTbOpen("db.db", -1, -1, NULL, pDb, &pTb);
ret = tdbTbOpen("db.db", -1, -1, NULL, pDb, &pTb, 0);
GTEST_ASSERT_EQ(ret, 0);
auto insert = [](TDB *pDb, TTB *pTb, int nData, int *stop, std::shared_timed_mutex *mu) {
@ -726,4 +726,4 @@ TEST(tdb_test, DISABLED_multi_thread1) {
ret = tdbClose(pDb);
GTEST_ASSERT_EQ(ret, 0);
#endif
}
}

1
source/os/src/osFile.c
View File

@ -343,6 +343,7 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
TdFilePtr pFile = (TdFilePtr)taosMemoryMalloc(sizeof(TdFile));
if (pFile == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
if (fd >= 0) close(fd);
if (fp != NULL) fclose(fp);
return NULL;