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

fix: add post commit to tq & stream

This commit is contained in:
Minglei Jin 2022-11-07 18:28:45 +08:00
parent c17f8244be
commit 1a4cf6964b
7 changed files with 28 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -125,6 +125,10 @@ int32_t tqMetaSaveCheckInfo(STQ* pTq, const char* key, const void* value, int32_
return -1;
}
if (tdbPostCommit(pTq->pMetaDB, &txn) < 0) {
return -1;
}
return 0;
}
@ -147,6 +151,10 @@ int32_t tqMetaDeleteCheckInfo(STQ* pTq, const char* key) {
ASSERT(0);
}
if (tdbPostCommit(pTq->pMetaDB, &txn) < 0) {
ASSERT(0);
}
return 0;
}
@ -226,6 +234,10 @@ int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
ASSERT(0);
}
if (tdbPostCommit(pTq->pMetaDB, &txn) < 0) {
ASSERT(0);
}
tEncoderClear(&encoder);
taosMemoryFree(buf);
return 0;
@ -250,6 +262,10 @@ int32_t tqMetaDeleteHandle(STQ* pTq, const char* key) {
ASSERT(0);
}
if (tdbPostCommit(pTq->pMetaDB, &txn) < 0) {
ASSERT(0);
}
return 0;
}

2
source/dnode/vnode/src/tq/tqSnapshot.c
View File

@ -169,6 +169,8 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
} else {
code = tdbCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
if (code) goto _err;
code = tdbPostCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
if (code) goto _err;
}
taosMemoryFree(pWriter);

2
source/dnode/vnode/src/tq/tqStreamStateSnap.c
View File

@ -169,6 +169,8 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
} else {
code = tdbCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
if (code) goto _err;
code = tdbPostCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
if (code) goto _err;
}
taosMemoryFree(pWriter);

2
source/dnode/vnode/src/tq/tqStreamTaskSnap.c
View File

@ -169,6 +169,8 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
} else {
code = tdbCommit(pWriter->pTq->pMetaStore, &pWriter->txn);
if (code) goto _err;
code = tdbPostCommit(pWriter->pTq->pMetaStore, &pWriter->txn);
if (code) goto _err;
}
taosMemoryFree(pWriter);

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

@ -70,6 +70,7 @@ _err:
void streamMetaClose(SStreamMeta* pMeta) {
tdbCommit(pMeta->db, &pMeta->txn);
tdbPostCommit(pMeta->db, &pMeta->txn);
tdbTbClose(pMeta->pTaskDb);
tdbTbClose(pMeta->pCheckpointDb);
tdbClose(pMeta->db);

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

@ -142,6 +142,7 @@ _err:
void streamStateClose(SStreamState* pState) {
tdbCommit(pState->db, &pState->txn);
tdbPostCommit(pState->db, &pState->txn);
tdbTbClose(pState->pStateDb);
tdbTbClose(pState->pFuncStateDb);
tdbTbClose(pState->pFillStateDb);
@ -168,6 +169,9 @@ int32_t streamStateCommit(SStreamState* pState) {
if (tdbCommit(pState->db, &pState->txn) < 0) {
return -1;
}
if (tdbPostCommit(pState->db, &pState->txn) < 0) {
return -1;
}
memset(&pState->txn, 0, sizeof(TXN));
if (tdbTxnOpen(&pState->txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
0) {

1
source/libs/tdb/src/db/tdbBtree.c
View File

@ -140,6 +140,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, char const *tbname, SPg
// tdbUnrefPage(pPage);
tdbPCacheRelease(pPager->pCache, pPage, &txn);
tdbCommit(pPager->pEnv, &txn);
tdbPostCommit(pPager->pEnv, &txn);
tdbTxnClose(&txn);
}