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

change chkpid gen way

This commit is contained in:
yihaoDeng 2023-11-09 13:59:21 +08:00
parent 50666987f0
commit 15b73354f8
1 changed files with 6 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -715,12 +715,10 @@ int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream)
} }
static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) { static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
int32_t code = -1; SMnode * pMnode = pReq->info.node;
int32_t code = -1;
SMnode * pMnode = pReq->info.node; SStreamObj * pStream = NULL;
SStreamObj *pStream = NULL; SDbObj * pDb = NULL;
SDbObj * pDb = NULL;
SCMCreateStreamReq createStreamReq = {0}; SCMCreateStreamReq createStreamReq = {0};
SStreamObj streamObj = {0}; SStreamObj streamObj = {0};
@ -883,13 +881,12 @@ int64_t mndStreamGenChkpId(SMnode *pMnode) {
SStreamObj *pStream = NULL; SStreamObj *pStream = NULL;
void * pIter = NULL; void * pIter = NULL;
SSdb * pSdb = pMnode->pSdb; SSdb * pSdb = pMnode->pSdb;
int64_t maxChkpId = 0;
int64_t maxChkpId = 0;
while (1) { while (1) {
pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream); pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
if (pIter == NULL) break; if (pIter == NULL) break;
maxChkpId = MAX(maxChkpId, pStream->checkpointId); maxChkpId = TMAX(maxChkpId, pStream->checkpointId);
sdbRelease(pSdb, pStream); sdbRelease(pSdb, pStream);
} }
return maxChkpId + 1; return maxChkpId + 1;