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

fix:add sdbFetchCancel to release hash node

This commit is contained in:
wangmm0220 2023-07-17 20:21:23 +08:00
parent 8d8e12ee0d
commit da9e30434a
11 changed files with 39 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/mnode/impl/src/mndCluster.c
View File

@ -77,7 +77,7 @@ static SClusterObj *mndAcquireCluster(SMnode *pMnode, void **ppIter) {
if (pIter == NULL) break;
*ppIter = pIter;
sdbCancelFetch(pSdb, pIter);
return pCluster;
}

1
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -706,6 +706,7 @@ _OVER:
} else {
mndReleaseDnode(pMnode, pDnode);
}
sdbCancelFetch(pSdb, pIter);
mndTransDrop(pTrans);
sdbFreeRaw(pRaw);
return terrno;

1
source/dnode/mnode/impl/src/mndIndex.c
View File

@ -831,6 +831,7 @@ int32_t mndGetIdxsByTagName(SMnode *pMnode, SStbObj *pStb, char *tagName, SIdxOb
if (pIdx->stbUid == pStb->uid && strcasecmp(pIdx->colName, tagName) == 0) {
memcpy((char *)idx, (char *)pIdx, sizeof(SIdxObj));
sdbRelease(pSdb, pIdx);
sdbCancelFetch(pSdb, pIter);
return 0;
}

1
source/dnode/mnode/impl/src/mndMain.c
View File

@ -199,6 +199,7 @@ static void mndSetVgroupOffline(SMnode *pMnode, int32_t dnodeId, int64_t curMs)
pGid->syncCanRead = 0;
roleChanged = true;
}
sdbCancelFetch(pSdb, pIter);
break;
}
}

5
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -168,6 +168,7 @@ SSnodeObj* mndSchedFetchOneSnode(SMnode* pMnode) {
void* pIter = NULL;
// TODO random fetch
pIter = sdbFetch(pMnode->pSdb, SDB_SNODE, pIter, (void**)&pObj);
sdbCancelFetch(pMnode->pSdb, pIter);
return pObj;
}
@ -435,6 +436,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
sdbRelease(pSdb, pVgroup);
qDestroyQueryPlan(pPlan);
sdbCancelFetch(pSdb, pIter);
return -1;
}
@ -444,6 +446,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
if (mndAssignStreamTaskToVgroup(pMnode, pTask, plan, pVgroup) < 0) {
sdbRelease(pSdb, pVgroup);
qDestroyQueryPlan(pPlan);
sdbCancelFetch(pSdb, pIter);
return -1;
}
@ -453,6 +456,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
qDestroyQueryPlan(pPlan);
sdbCancelFetch(pSdb, pIter);
return -1;
}
}
@ -492,6 +496,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
if (code != TSDB_CODE_SUCCESS) {
qDestroyQueryPlan(pPlan);
sdbCancelFetch(pSdb, pIter);
return -1;
}
}

5
source/dnode/mnode/impl/src/mndStb.c
View File

@ -900,7 +900,6 @@ static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {
SMsgHead *pHead = rpcMallocCont(contLen);
if (pHead == NULL) {
sdbCancelFetch(pSdb, pVgroup);
sdbRelease(pSdb, pVgroup);
continue;
}
@ -1289,6 +1288,7 @@ static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName
terrno = TSDB_CODE_MND_INVALID_STREAM_OPTION;
mError("stream:%s, create ast error", pStream->name);
sdbRelease(pSdb, pStream);
sdbCancelFetch(pSdb, pIter);
return -1;
}
@ -1308,6 +1308,7 @@ static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
sdbRelease(pSdb, pStream);
sdbCancelFetch(pSdb, pIter);
return -1;
}
mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
@ -1337,6 +1338,7 @@ static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName,
terrno = TSDB_CODE_SDB_INVALID_DATA_CONTENT;
mError("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d failed since parse AST err",
pSma->name, stbFullName, suid, colId);
sdbCancelFetch(pSdb, pIter);
return -1;
}
@ -1357,6 +1359,7 @@ static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName,
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
sdbRelease(pSdb, pSma);
sdbCancelFetch(pSdb, pIter);
return -1;
}
mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);

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

@ -705,12 +705,14 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
if (numOfStream > MND_STREAM_MAX_NUM) {
mError("too many streams, no more than %d for each database", MND_STREAM_MAX_NUM);
terrno = TSDB_CODE_MND_TOO_MANY_STREAMS;
sdbCancelFetch(pMnode->pSdb, pIter);
goto _OVER;
}
if (pStream->targetStbUid == streamObj.targetStbUid) {
mError("Cannot write the same stable as other stream:%s", pStream->name);
terrno = TSDB_CODE_MND_INVALID_TARGET_TABLE;
sdbCancelFetch(pMnode->pSdb, pIter);
goto _OVER;
}
}

3
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -1104,6 +1104,7 @@ int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName)
if (taosHashGetSize(pSub->consumerHash) != 0) {
sdbRelease(pSdb, pSub);
terrno = TSDB_CODE_MND_IN_REBALANCE;
sdbCancelFetch(pSdb, pIter);
return -1;
}
int32_t sz = taosArrayGetSize(pSub->unassignedVgs);
@ -1122,12 +1123,14 @@ int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName)
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
sdbRelease(pSdb, pSub);
sdbCancelFetch(pSdb, pIter);
return -1;
}
}
if (mndSetDropSubRedoLogs(pMnode, pTrans, pSub) < 0) {
sdbRelease(pSdb, pSub);
sdbCancelFetch(pSdb, pIter);
goto END;
}

7
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -513,6 +513,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
tEncodeSize(tEncodeSTqCheckInfo, &info, len, code);
if (code < 0) {
sdbRelease(pSdb, pVgroup);
sdbCancelFetch(pSdb, pIter);
goto _OUT;
}
void *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);
@ -522,6 +523,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
if (tEncodeSTqCheckInfo(&encoder, &info) < 0) {
taosMemoryFree(buf);
sdbRelease(pSdb, pVgroup);
sdbCancelFetch(pSdb, pIter);
goto _OUT;
}
tEncoderClear(&encoder);
@ -535,6 +537,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(buf);
sdbRelease(pSdb, pVgroup);
sdbCancelFetch(pSdb, pIter);
goto _OUT;
}
buf = NULL;
@ -697,6 +700,7 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
if (strcmp(name, pTopic->name) == 0) {
mndReleaseConsumer(pMnode, pConsumer);
mndReleaseTopic(pMnode, pTopic);
sdbCancelFetch(pSdb, pIter);
terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s",
dropReq.name, pConsumer->consumerId, pConsumer->cgroup);
@ -710,6 +714,7 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
if (strcmp(name, pTopic->name) == 0) {
mndReleaseConsumer(pMnode, pConsumer);
mndReleaseTopic(pMnode, pTopic);
sdbCancelFetch(pSdb, pIter);
terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb new)",
dropReq.name, pConsumer->consumerId, pConsumer->cgroup);
@ -723,6 +728,7 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
if (strcmp(name, pTopic->name) == 0) {
mndReleaseConsumer(pMnode, pConsumer);
mndReleaseTopic(pMnode, pTopic);
sdbCancelFetch(pSdb, pIter);
terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb remove)",
dropReq.name, pConsumer->consumerId, pConsumer->cgroup);
@ -789,6 +795,7 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
taosMemoryFree(buf);
sdbRelease(pSdb, pVgroup);
mndReleaseTopic(pMnode, pTopic);
sdbCancelFetch(pSdb, pIter);
mndTransDrop(pTrans);
return -1;
}

16
source/dnode/mnode/impl/src/mndUser.c
View File

@ -1444,7 +1444,10 @@ int32_t mndUserRemoveDb(SMnode *pMnode, STrans *pTrans, char *db) {
if (pIter == NULL) break;
code = -1;
if (mndUserDupObj(pUser, &newUser) != 0) break;
if (mndUserDupObj(pUser, &newUser) != 0) {
sdbCancelFetch(pSdb, pIter);
break;
}
bool inRead = (taosHashGet(newUser.readDbs, db, len) != NULL);
bool inWrite = (taosHashGet(newUser.writeDbs, db, len) != NULL);
@ -1453,7 +1456,10 @@ int32_t mndUserRemoveDb(SMnode *pMnode, STrans *pTrans, char *db) {
(void)taosHashRemove(newUser.writeDbs, db, len);
SSdbRaw *pCommitRaw = mndUserActionEncode(&newUser);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) break;
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
sdbCancelFetch(pSdb, pIter);
break;
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
}
@ -1484,6 +1490,7 @@ int32_t mndUserRemoveTopic(SMnode *pMnode, STrans *pTrans, char *topic) {
code = -1;
if (mndUserDupObj(pUser, &newUser) != 0) {
sdbCancelFetch(pSdb, pIter);
break;
}
@ -1491,7 +1498,10 @@ int32_t mndUserRemoveTopic(SMnode *pMnode, STrans *pTrans, char *topic) {
if (inTopic) {
(void)taosHashRemove(newUser.topics, topic, len);
SSdbRaw *pCommitRaw = mndUserActionEncode(&newUser);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) break;
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
sdbCancelFetch(pSdb, pIter);
break;
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
}

1
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -2591,6 +2591,7 @@ static int32_t mndProcessBalanceVgroupMsg(SRpcMsg *pReq) {
pIter = sdbFetch(pMnode->pSdb, SDB_DNODE, pIter, (void **)&pDnode);
if (pIter == NULL) break;
if (!mndIsDnodeOnline(pDnode, curMs)) {
sdbCancelFetch(pMnode->pSdb, pIter);
terrno = TSDB_CODE_MND_HAS_OFFLINE_DNODE;
mError("failed to balance vgroup since %s, dnode:%d", terrstr(), pDnode->id);
sdbRelease(pMnode->pSdb, pDnode);