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

fix:[TD-31146] invalid read if tmq is freed

This commit is contained in:
wangmm0220 2024-08-01 11:29:21 +08:00
parent 7298feac14
commit c75281b9ff
1 changed files with 15 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
source/client/src/clientTmq.c
View File

@ -1452,22 +1452,22 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
tmq_t* tmq = NULL; tmq_t* tmq = NULL;
SMqPollCbParam* pParam = (SMqPollCbParam*)param; SMqPollCbParam* pParam = (SMqPollCbParam*)param;
if (pParam == NULL || pMsg == NULL) { if (pParam == NULL || pMsg == NULL) {
goto FAIL2; return TSDB_CODE_TSC_INTERNAL_ERROR;
} }
int64_t refId = pParam->refId; int64_t refId = pParam->refId;
int32_t vgId = pParam->vgId; int32_t vgId = pParam->vgId;
uint64_t requestId = pParam->requestId; uint64_t requestId = pParam->requestId;
tmq = taosAcquireRef(tmqMgmt.rsetId, refId); tmq = taosAcquireRef(tmqMgmt.rsetId, refId);
if (tmq == NULL) { if (tmq == NULL) {
code = TSDB_CODE_TMQ_CONSUMER_CLOSED; return TSDB_CODE_TMQ_CONSUMER_CLOSED;
goto FAIL2;
} }
SMqPollRspWrapper* pRspWrapper = NULL; SMqPollRspWrapper* pRspWrapper = NULL;
code = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM, 0, (void**)&pRspWrapper); int32_t ret = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM, 0, (void**)&pRspWrapper);
if (code) { if (ret) {
code = ret;
tscWarn("consumer:0x%" PRIx64 " msg discard from vgId:%d, since out of memory", tmq->consumerId, vgId); tscWarn("consumer:0x%" PRIx64 " msg discard from vgId:%d, since out of memory", tmq->consumerId, vgId);
goto FAIL1; goto END;
} }
if (code != 0) { if (code != 0) {
@ -1550,6 +1550,7 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
} }
END: END:
if (pRspWrapper){
pRspWrapper->code = code; pRspWrapper->code = code;
pRspWrapper->vgId = vgId; pRspWrapper->vgId = vgId;
(void)strcpy(pRspWrapper->topicName, pParam->topicName); (void)strcpy(pRspWrapper->topicName, pParam->topicName);
@ -1557,18 +1558,15 @@ END:
if(code != 0){ if(code != 0){
tscError("consumer:0x%" PRIx64 " put poll res into mqueue failed, code:%d", tmq->consumerId, code); tscError("consumer:0x%" PRIx64 " put poll res into mqueue failed, code:%d", tmq->consumerId, code);
} }
}
int32_t total = taosQueueItemSize(tmq->mqueue); int32_t total = taosQueueItemSize(tmq->mqueue);
tscDebug("consumer:0x%" PRIx64 " put poll res into mqueue, type:%d, vgId:%d, total in queue:%d, reqId:0x%" PRIx64, tscDebug("consumer:0x%" PRIx64 " put poll res into mqueue, type:%d, vgId:%d, total in queue:%d, reqId:0x%" PRIx64,
tmq->consumerId, rspType, vgId, total, requestId); tmq->consumerId, rspType, vgId, total, requestId);
FAIL1:
(void)taosReleaseRef(tmqMgmt.rsetId, refId);
FAIL2:
if (tmq) (void)tsem2_post(&tmq->rspSem); if (tmq) (void)tsem2_post(&tmq->rspSem);
if (pMsg) taosMemoryFreeClear(pMsg->pData); if (pMsg) taosMemoryFreeClear(pMsg->pData);
if (pMsg) taosMemoryFreeClear(pMsg->pEpSet); if (pMsg) taosMemoryFreeClear(pMsg->pEpSet);
(void)taosReleaseRef(tmqMgmt.rsetId, refId);
return code; return code;
} }