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

refactor stream backend

This commit is contained in:
yihaoDeng 2023-10-23 16:45:17 +08:00
parent 9d210ec957
commit ca1ffd584d
1 changed files with 25 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
source/dnode/mnode/impl/src/mndStream.c
View File

@ -2450,22 +2450,20 @@ int32_t createStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream) {
return TSDB_CODE_ACTION_IN_PROGRESS; return TSDB_CODE_ACTION_IN_PROGRESS;
} }
int32_t mndResetFromCheckpoint(SMnode *pMnode) { int32_t mndFindTransByName(SMnode *pMnode, char *key, int32_t *transId) {
// find the checkpoint trans id
int32_t transId = 0;
{
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
STrans *pTrans = NULL; STrans *pTrans = NULL;
void *pIter = NULL; void *pIter = NULL;
bool found = false;
while (1) { while (1) {
pIter = sdbFetch(pSdb, SDB_TRANS, pIter, (void **)&pTrans); pIter = sdbFetch(pSdb, SDB_TRANS, pIter, (void **)&pTrans);
if (pIter == NULL) { if (pIter == NULL) {
break; break;
} }
if (strncmp(pTrans->opername, MND_STREAM_CHECKPOINT_NAME, tListLen(pTrans->opername) - 1) == 0) { if (strncmp(pTrans->opername, key, tListLen(pTrans->opername) - 1) == 0) {
transId = pTrans->id; *transId = pTrans->id;
found = true;
sdbRelease(pSdb, pTrans); sdbRelease(pSdb, pTrans);
sdbCancelFetch(pSdb, pIter); sdbCancelFetch(pSdb, pIter);
break; break;
@ -2473,9 +2471,13 @@ int32_t mndResetFromCheckpoint(SMnode *pMnode) {
sdbRelease(pSdb, pTrans); sdbRelease(pSdb, pTrans);
} }
} return found ? 0 : -1;
}
if (transId == 0) { int32_t mndResetFromCheckpoint(SMnode *pMnode) {
// find the checkpoint trans id
int32_t transId = 0;
int32_t code = mndFindTransByName(pMnode, MND_STREAM_CHECKPOINT_NAME, &transId);
if (code == -1) {
mError("failed to find the checkpoint trans, reset not executed"); mError("failed to find the checkpoint trans, reset not executed");
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -2492,7 +2494,6 @@ int32_t mndResetFromCheckpoint(SMnode *pMnode) {
if (pIter == NULL) { if (pIter == NULL) {
break; break;
} }
mDebug("stream:%s (0x%" PRIx64 ") reset checkpoint procedure, create reset trans", pStream->name, pStream->uid); mDebug("stream:%s (0x%" PRIx64 ") reset checkpoint procedure, create reset trans", pStream->name, pStream->uid);
int32_t code = createStreamResetStatusTrans(pMnode, pStream); int32_t code = createStreamResetStatusTrans(pMnode, pStream);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
@ -2500,7 +2501,6 @@ int32_t mndResetFromCheckpoint(SMnode *pMnode) {
return code; return code;
} }
} }
return 0; return 0;
} }