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

refactor: get sdb snapshot in atomic way

This commit is contained in:
Shengliang Guan 2022-06-20 18:37:46 +08:00
parent ca7057ce11
commit ab0d7d31bc
4 changed files with 12 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
source/dnode/mnode/impl/src/mndSync.c
View File

@ -75,12 +75,10 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM
} }
int32_t mndSyncGetSnapshot(struct SSyncFSM *pFsm, SSnapshot *pSnapshot, void *pReaderParam, void **ppReader) { int32_t mndSyncGetSnapshot(struct SSyncFSM *pFsm, SSnapshot *pSnapshot, void *pReaderParam, void **ppReader) {
// TODO: mInfo("start to read snapshot from sdb in atomic way");
SMnode *pMnode = pFsm->data;
// atomic operation return sdbStartRead(pMnode->pSdb, (SSdbIter **)ppReader, &pSnapshot->lastApplyIndex, &pSnapshot->lastApplyTerm,
// step1. sdbGetCommitInfo &pSnapshot->lastConfigIndex);
// step2. create ppReader with pReaderParam
return 0; return 0;
} }
@ -106,14 +104,6 @@ void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta cbM
SMnode *pMnode = pFsm->data; SMnode *pMnode = pFsm->data;
SSyncMgmt *pMgmt = &pMnode->syncMgmt; SSyncMgmt *pMgmt = &pMnode->syncMgmt;
#if 0
// send response
SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen, .conn.applyIndex = cbMeta.index};
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
syncGetAndDelRespRpc(pMnode->syncMgmt.sync, cbMeta.seqNum, &rpcMsg.info);
#endif
pMgmt->errCode = cbMeta.code; pMgmt->errCode = cbMeta.code;
mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64, pMgmt->transId, mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64, pMgmt->transId,
cbMeta.code, cbMeta.index, cbMeta.term); cbMeta.code, cbMeta.index, cbMeta.term);
@ -130,7 +120,7 @@ void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta cbM
int32_t mndSnapshotStartRead(struct SSyncFSM *pFsm, void **ppReader) { int32_t mndSnapshotStartRead(struct SSyncFSM *pFsm, void **ppReader) {
mInfo("start to read snapshot from sdb"); mInfo("start to read snapshot from sdb");
SMnode *pMnode = pFsm->data; SMnode *pMnode = pFsm->data;
return sdbStartRead(pMnode->pSdb, (SSdbIter **)ppReader); return sdbStartRead(pMnode->pSdb, (SSdbIter **)ppReader, NULL, NULL, NULL);
} }
int32_t mndSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) { int32_t mndSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) {

2
source/dnode/mnode/impl/test/sdb/sdbTest.cpp
View File

@ -910,7 +910,7 @@ TEST_F(MndTestSdb, 01_Read_Str) {
int32_t len = 0; int32_t len = 0;
int32_t code = 0; int32_t code = 0;
code = sdbStartRead(pSdb, &pReader); code = sdbStartRead(pSdb, &pReader, NULL, NULL, NULL);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
code = sdbStartWrite(pSdb, &pWritter); code = sdbStartWrite(pSdb, &pWritter);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);

2
source/dnode/mnode/sdb/inc/sdb.h
View File

@ -388,7 +388,7 @@ SSdbRow *sdbAllocRow(int32_t objSize);
void *sdbGetRowObj(SSdbRow *pRow); void *sdbGetRowObj(SSdbRow *pRow);
void sdbFreeRow(SSdb *pSdb, SSdbRow *pRow, bool callFunc); void sdbFreeRow(SSdb *pSdb, SSdbRow *pRow, bool callFunc);
int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter); int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter, int64_t *index, int64_t *term, int64_t *config);
int32_t sdbStopRead(SSdb *pSdb, SSdbIter *pIter); int32_t sdbStopRead(SSdb *pSdb, SSdbIter *pIter);
int32_t sdbDoRead(SSdb *pSdb, SSdbIter *pIter, void **ppBuf, int32_t *len); int32_t sdbDoRead(SSdb *pSdb, SSdbIter *pIter, void **ppBuf, int32_t *len);

6
source/dnode/mnode/sdb/src/sdbFile.c
View File

@ -523,7 +523,7 @@ static void sdbCloseIter(SSdbIter *pIter) {
taosMemoryFree(pIter); taosMemoryFree(pIter);
} }
int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter) { int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter, int64_t *index, int64_t *term, int64_t *config) {
SSdbIter *pIter = sdbCreateIter(pSdb); SSdbIter *pIter = sdbCreateIter(pSdb);
if (pIter == NULL) return -1; if (pIter == NULL) return -1;
@ -552,6 +552,10 @@ int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter) {
} }
*ppIter = pIter; *ppIter = pIter;
if (index != NULL) *index = commitIndex;
if (term != NULL) *term = commitTerm;
if (config != NULL) *config = commitConfig;
mInfo("sdbiter:%p, is created to read snapshot, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " file:%s", mInfo("sdbiter:%p, is created to read snapshot, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " file:%s",
pIter, commitIndex, commitTerm, commitConfig, pIter->name); pIter, commitIndex, commitTerm, commitConfig, pIter->name);
return 0; return 0;