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

fix:set task status killed when vnode receive subscribe msg from mnode

This commit is contained in:
wangmm0220 2023-05-19 18:06:05 +08:00
parent ff2b545b27
commit c7c255e967
1 changed files with 11 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/dnode/vnode/src/tq/tq.c
View File

@ -370,6 +370,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
bool exec = tqIsHandleExec(pHandle); bool exec = tqIsHandleExec(pHandle);
if(!exec) { if(!exec) {
tqSetHandleExec(pHandle); tqSetHandleExec(pHandle);
qSetTaskCode(pHandle->execHandle.task, TDB_CODE_SUCCESS);
tqDebug("tmq poll: consumer:0x%" PRIx64 "vgId:%d, topic:%s, set handle exec, pHandle:%p", consumerId, vgId, req.subKey, pHandle); tqDebug("tmq poll: consumer:0x%" PRIx64 "vgId:%d, topic:%s, set handle exec, pHandle:%p", consumerId, vgId, req.subKey, pHandle);
taosWUnLockLatch(&pTq->lock); taosWUnLockLatch(&pTq->lock);
break; break;
@ -394,7 +395,6 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
consumerId, req.epoch, pHandle->subKey, vgId, buf, req.reqId); consumerId, req.epoch, pHandle->subKey, vgId, buf, req.reqId);
code = tqExtractDataForMq(pTq, pHandle, &req, pMsg); code = tqExtractDataForMq(pTq, pHandle, &req, pMsg);
qSetTaskCode(pHandle->execHandle.task, TDB_CODE_SUCCESS);
tqSetHandleIdle(pHandle); tqSetHandleIdle(pHandle);
tqDebug("tmq poll: consumer:0x%" PRIx64 "vgId:%d, topic:%s, , set handle idle, pHandle:%p", consumerId, vgId, req.subKey, pHandle); tqDebug("tmq poll: consumer:0x%" PRIx64 "vgId:%d, topic:%s, , set handle idle, pHandle:%p", consumerId, vgId, req.subKey, pHandle);
return code; return code;
@ -571,16 +571,18 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
// atomic_store_32(&pHandle->epoch, 0); // atomic_store_32(&pHandle->epoch, 0);
} }
taosWLockLatch(&pTq->lock);
// kill executing task // kill executing task
if(tqIsHandleExec(pHandle)) {
qTaskInfo_t pTaskInfo = pHandle->execHandle.task; qTaskInfo_t pTaskInfo = pHandle->execHandle.task;
if (pTaskInfo != NULL) { if (pTaskInfo != NULL) {
qKillTask(pTaskInfo, TSDB_CODE_TSC_QUERY_KILLED); qKillTask(pTaskInfo, TSDB_CODE_TSC_QUERY_KILLED);
} }
taosWLockLatch(&pTq->lock);
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
qStreamCloseTsdbReader(pTaskInfo); qStreamCloseTsdbReader(pTaskInfo);
} }
}
// remove if it has been register in the push manager, and return one empty block to consumer // remove if it has been register in the push manager, and return one empty block to consumer
tqUnregisterPushHandle(pTq, pHandle); tqUnregisterPushHandle(pTq, pHandle);
taosWUnLockLatch(&pTq->lock); taosWUnLockLatch(&pTq->lock);