Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix(client): fix potential memory leak. #20508

Merged
merged 18 commits into from Mar 20, 2023
Merged

fix(client): fix potential memory leak. #20508

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
8d1c695
fix(query): set the tag filter flag.
hjxilinx Mar 17, 2023
4f15a32
fix(client): fix potential memory leak.
hjxilinx Mar 17, 2023
f24e06f
fix tag filter
yihaoDeng Mar 18, 2023
afaa77c
fix tag filter
yihaoDeng Mar 18, 2023
8a8f1da
refactor: update the logs.
hjxilinx Mar 18, 2023
4265fcb
Merge remote-tracking branch 'origin/fix/liaohj' into fix/liaohj
hjxilinx Mar 18, 2023
ae5155e
fix(tmq): do some internal refactor.
hjxilinx Mar 19, 2023
e5e4f50
fix(tmq): adjust the total rows counting.
hjxilinx Mar 19, 2023
a9e5416
fix(tmq): adjust log.
hjxilinx Mar 19, 2023
49df168
fix(tmq): adjust logs.
hjxilinx Mar 19, 2023
fad23f8
fix(tmq): adjust log.
hjxilinx Mar 19, 2023
7e712a1
fix(tmq): adjust log.
hjxilinx Mar 19, 2023
c678740
refactor: do some internal refactor.
hjxilinx Mar 19, 2023
a5e9fad
fix(query): set correct lastkeyInstt.
hjxilinx Mar 19, 2023
192570c
fix tag filter
yihaoDeng Mar 20, 2023
4cb1c51
fix(tmq): update some logs.
hjxilinx Mar 20, 2023
a24d537
Merge remote-tracking branch 'origin/fix/liaohj' into fix/liaohj
hjxilinx Mar 20, 2023
2517f9b
fix(tmq): add some logs.
hjxilinx Mar 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
13 changes: 11 additions & 2 deletions include/libs/executor/executor.h
View file
Open in desktop
Expand Up @@ -67,17 +67,26 @@ typedef enum {
* Create the exec task for stream mode
* @param pMsg
* @param SReadHandle
* @param vgId
* @return
*/
qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers);
qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers, int32_t vgId);

/**
* Create the exec task for queue mode
* @param pMsg
* @param SReadHandle
* @return
*/
qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchema);
qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols, SSchemaWrapper** pSchema);

/**
* set the task Id, usually used by message queue process
* @param tinfo
* @param taskId
* @param queryId
*/
void qSetTaskId(qTaskInfo_t tinfo, uint64_t taskId, uint64_t queryId);

int32_t qSetStreamOpOpen(qTaskInfo_t tinfo);
/**
Expand Down
5 changes: 0 additions & 5 deletions source/client/src/clientMain.c
View file
Open in desktop
Expand Up @@ -271,8 +271,6 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
SReqResultInfo *pResultInfo;
if (msg->resIter == -1) {
pResultInfo = tmqGetNextResInfo(res, true);
tscDebug("consumer:0x%" PRIx64 ", vgId:%d, numOfRows:%" PRId64 ", total rows:%" PRId64, msg->rsp.head.consumerId,
msg->vgId, pResultInfo->numOfRows, pResultInfo->totalRows);
} else {
pResultInfo = tmqGetCurResInfo(res);
}
Expand All @@ -287,9 +285,6 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
return NULL;
}

tscDebug("consumer:0x%" PRIx64 " vgId:%d, numOfRows:%" PRId64 ", total rows:%" PRId64, msg->rsp.head.consumerId,
msg->vgId, pResultInfo->numOfRows, pResultInfo->totalRows);

doSetOneRowPtr(pResultInfo);
pResultInfo->current += 1;
return pResultInfo->row;
Expand Down
131 changes: 77 additions & 54 deletions source/client/src/clientTmq.c
View file
Open in desktop
Expand Up @@ -580,7 +580,10 @@ static int32_t tmqCommitMsgImpl(tmq_t* tmq, const TAOS_RES* msg, int8_t async, t
int32_t code = -1;

taosThreadMutexLock(&tmq->lock);
for (int32_t i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
int32_t numOfTopics = taosArrayGetSize(tmq->clientTopics);

tscDebug("consumer:0x%" PRIx64 " user invoked commit offset for %d", tmq->consumerId, numOfTopics);
for (int32_t i = 0; i < numOfTopics; i++) {
SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
if (strcmp(pTopic->topicName, topic) != 0) {
continue;
Expand Down Expand Up @@ -874,7 +877,7 @@ int32_t tmqHandleAllDelayedTask(tmq_t* pTmq) {
return 0;
}

static void tmqFreeRspWrapper(SMqRspWrapper* rspWrapper) {
static void* tmqFreeRspWrapper(SMqRspWrapper* rspWrapper) {
if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__END_RSP) {
// do nothing
} else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__EP_RSP) {
Expand Down Expand Up @@ -905,6 +908,8 @@ static void tmqFreeRspWrapper(SMqRspWrapper* rspWrapper) {
taosArrayDestroy(pRsp->taosxRsp.createTableLen);
taosArrayDestroyP(pRsp->taosxRsp.createTableReq, taosMemoryFree);
}

return NULL;
}

void tmqClearUnhandleMsg(tmq_t* tmq) {
Expand Down Expand Up @@ -1333,10 +1338,10 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
tDecoderClear(&decoder);
memcpy(&pRspWrapper->dataRsp, pMsg->pData, sizeof(SMqRspHead));

tscDebug("consumer:0x%" PRIx64 " recv poll rsp, vgId:%d, req:%" PRId64 ", rsp:%" PRId64
" type %d, reqId:0x%" PRIx64,
tmq->consumerId, vgId, pRspWrapper->dataRsp.reqOffset.version, pRspWrapper->dataRsp.rspOffset.version, rspType, requestId);

char buf[80];
tFormatOffset(buf, 80, &pRspWrapper->dataRsp.rspOffset);
tscDebug("consumer:0x%" PRIx64 " recv poll rsp, vgId:%d, req:%" PRId64 ", rsp:%s type %d, reqId:0x%" PRIx64,
tmq->consumerId, vgId, pRspWrapper->dataRsp.reqOffset.version, buf, rspType, requestId);
} else if (rspType == TMQ_MSG_TYPE__POLL_META_RSP) {
SDecoder decoder;
tDecoderInit(&decoder, POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), pMsg->len - sizeof(SMqRspHead));
Expand Down Expand Up @@ -1375,6 +1380,11 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
return -1;
}

typedef struct SVgroupSaveInfo {
STqOffsetVal offset;
int64_t numOfRows;
} SVgroupSaveInfo;

static void initClientTopicFromRsp(SMqClientTopic* pTopic, SMqSubTopicEp* pTopicEp, SHashObj* pVgOffsetHashMap,
tmq_t* tmq) {
pTopic->schema = pTopicEp->schema;
Expand All @@ -1394,11 +1404,13 @@ static void initClientTopicFromRsp(SMqClientTopic* pTopic, SMqSubTopicEp* pTopic
SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j);

makeTopicVgroupKey(vgKey, pTopic->topicName, pVgEp->vgId);
STqOffsetVal* pOffset = taosHashGet(pVgOffsetHashMap, vgKey, strlen(vgKey));
SVgroupSaveInfo* pInfo = taosHashGet(pVgOffsetHashMap, vgKey, strlen(vgKey));

int64_t numOfRows = 0;
STqOffsetVal offsetNew = {.type = tmq->resetOffsetCfg};
if (pOffset != NULL) {
offsetNew = *pOffset;
if (pInfo != NULL) {
offsetNew = pInfo->offset;
numOfRows = pInfo->numOfRows;
}

SMqClientVg clientVg = {
Expand All @@ -1409,7 +1421,7 @@ static void initClientTopicFromRsp(SMqClientTopic* pTopic, SMqSubTopicEp* pTopic
.vgStatus = TMQ_VG_STATUS__IDLE,
.vgSkipCnt = 0,
.emptyBlockReceiveTs = 0,
.numOfRows = 0,
.numOfRows = numOfRows,
};

taosArrayPush(pTopic->vgs, &clientVg);
Expand Down Expand Up @@ -1461,7 +1473,9 @@ static bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, const SMqAskEpRsp* pRsp) {
tFormatOffset(buf, 80, &pVgCur->currentOffset);
tscDebug("consumer:0x%" PRIx64 ", epoch:%d vgId:%d vgKey:%s, offset:%s", tmq->consumerId, epoch,
pVgCur->vgId, vgKey, buf);
taosHashPut(pVgOffsetHashMap, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(STqOffsetVal));

SVgroupSaveInfo info = {.offset = pVgCur->currentOffset, .numOfRows = pVgCur->numOfRows};
taosHashPut(pVgOffsetHashMap, vgKey, strlen(vgKey), &info, sizeof(SVgroupSaveInfo));
}
}
}
Expand Down Expand Up @@ -1602,10 +1616,11 @@ SMqMetaRspObj* tmqBuildMetaRspFromWrapper(SMqPollRspWrapper* pWrapper) {
return pRspObj;
}

SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper, SMqClientVg* pVg) {
SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper, SMqClientVg* pVg, int64_t* numOfRows) {
SMqRspObj* pRspObj = taosMemoryCalloc(1, sizeof(SMqRspObj));
pRspObj->resType = RES_TYPE__TMQ;

(*numOfRows) = 0;
tstrncpy(pRspObj->topic, pWrapper->topicHandle->topicName, TSDB_TOPIC_FNAME_LEN);
tstrncpy(pRspObj->db, pWrapper->topicHandle->db, TSDB_DB_FNAME_LEN);

Expand All @@ -1624,8 +1639,8 @@ SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper, SMqClientVg* pVg)
for(int32_t i = 0; i < pRspObj->rsp.blockNum; ++i) {
SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(pRspObj->rsp.blockData, i);
int64_t rows = htobe64(pRetrieve->numOfRows);
pRspObj->resInfo.totalRows += rows;
pVg->numOfRows += rows;
(*numOfRows) += rows;
}

return pRspObj;
Expand Down Expand Up @@ -1788,29 +1803,28 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
tscDebug("consumer:0x%" PRIx64 " start to handle the rsp, total:%d", tmq->consumerId, tmq->qall->numOfItems);

while (1) {
SMqRspWrapper* rspWrapper = NULL;
taosGetQitem(tmq->qall, (void**)&rspWrapper);
SMqRspWrapper* pRspWrapper = NULL;
taosGetQitem(tmq->qall, (void**)&pRspWrapper);

if (rspWrapper == NULL) {
if (pRspWrapper == NULL) {
taosReadAllQitems(tmq->mqueue, tmq->qall);
taosGetQitem(tmq->qall, (void**)&rspWrapper);
taosGetQitem(tmq->qall, (void**)&pRspWrapper);

if (rspWrapper == NULL) {
if (pRspWrapper == NULL) {
return NULL;
}
}

tscDebug("consumer:0x%"PRIx64" handle rsp, type:%d", tmq->consumerId, rspWrapper->tmqRspType);
tscDebug("consumer:0x%"PRIx64" handle rsp, type:%d", tmq->consumerId, pRspWrapper->tmqRspType);

if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__END_RSP) {
taosFreeQitem(rspWrapper);
if (pRspWrapper->tmqRspType == TMQ_MSG_TYPE__END_RSP) {
taosFreeQitem(pRspWrapper);
terrno = TSDB_CODE_TQ_NO_COMMITTED_OFFSET;
tscError("consumer:0x%" PRIx64 " unexpected rsp from poll, code:%s", tmq->consumerId, tstrerror(terrno));
return NULL;
} else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_RSP) {
SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)rspWrapper;
} else if (pRspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_RSP) {
SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)pRspWrapper;

/*atomic_sub_fetch_32(&tmq->readyRequest, 1);*/
int32_t consumerEpoch = atomic_load_32(&tmq->epoch);
SMqDataRsp* pDataRsp = &pollRspWrapper->dataRsp;

Expand All @@ -1833,28 +1847,31 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
char buf[80];
tFormatOffset(buf, 80, &pDataRsp->rspOffset);
if (pDataRsp->blockNum == 0) {
tscDebug("consumer:0x%" PRIx64 " empty block received, vgId:%d, offset:%s, reqId:0x%" PRIx64, tmq->consumerId,
pVg->vgId, buf, pollRspWrapper->reqId);
tscDebug("consumer:0x%" PRIx64 " empty block received, vgId:%d, offset:%s, vg total:%"PRId64" total:%"PRId64" reqId:0x%" PRIx64, tmq->consumerId,
pVg->vgId, buf, pVg->numOfRows, tmq->totalRows, pollRspWrapper->reqId);
pRspWrapper = tmqFreeRspWrapper(pRspWrapper);
taosFreeQitem(pollRspWrapper);
rspWrapper = NULL;
continue;
} else { // build rsp
SMqRspObj* pRsp = tmqBuildRspFromWrapper(pollRspWrapper, pVg);
tscDebug("consumer:0x%" PRIx64 " process poll rsp, vgId:%d, offset:%s, blocks:%d, rows:%"PRId64" reqId:0x%" PRIx64,
tmq->consumerId, pVg->vgId, buf, pDataRsp->blockNum, pRsp->resInfo.totalRows, pollRspWrapper->reqId);
int64_t numOfRows = 0;
SMqRspObj* pRsp = tmqBuildRspFromWrapper(pollRspWrapper, pVg, &numOfRows);
tmq->totalRows += numOfRows;

tmq->totalRows += pRsp->resInfo.totalRows;
tscDebug("consumer:0x%" PRIx64 " process poll rsp, vgId:%d, offset:%s, blocks:%d, rows:%" PRId64
" vg total:%" PRId64 " total:%" PRId64 ", reqId:0x%" PRIx64,
tmq->consumerId, pVg->vgId, buf, pDataRsp->blockNum, numOfRows, pVg->numOfRows, tmq->totalRows,
pollRspWrapper->reqId);
taosFreeQitem(pollRspWrapper);
return pRsp;
}
} else {
tscDebug("consumer:0x%" PRIx64 " msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
tmq->consumerId, pDataRsp->head.epoch, consumerEpoch);
tmqFreeRspWrapper(rspWrapper);
SMqClientVg* pVg = pollRspWrapper->vgHandle;
tscDebug("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
tmq->consumerId, pVg->vgId, pDataRsp->head.epoch, consumerEpoch);
pRspWrapper = tmqFreeRspWrapper(pRspWrapper);
taosFreeQitem(pollRspWrapper);
}
} else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_META_RSP) {
SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)rspWrapper;
} else if (pRspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_META_RSP) {
SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)pRspWrapper;
int32_t consumerEpoch = atomic_load_32(&tmq->epoch);

tscDebug("consumer:0x%" PRIx64 " process meta rsp", tmq->consumerId);
Expand All @@ -1868,13 +1885,13 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
taosFreeQitem(pollRspWrapper);
return pRsp;
} else {
tscDebug("consumer:0x%" PRIx64 " msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
tmq->consumerId, pollRspWrapper->metaRsp.head.epoch, consumerEpoch);
tmqFreeRspWrapper(rspWrapper);
tscDebug("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
tmq->consumerId, pollRspWrapper->vgHandle->vgId, pollRspWrapper->metaRsp.head.epoch, consumerEpoch);
pRspWrapper = tmqFreeRspWrapper(pRspWrapper);
taosFreeQitem(pollRspWrapper);
}
} else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__TAOSX_RSP) {
SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)rspWrapper;
} else if (pRspWrapper->tmqRspType == TMQ_MSG_TYPE__TAOSX_RSP) {
SMqPollRspWrapper* pollRspWrapper = (SMqPollRspWrapper*)pRspWrapper;
int32_t consumerEpoch = atomic_load_32(&tmq->epoch);

if (pollRspWrapper->taosxRsp.head.epoch == consumerEpoch) {
Expand All @@ -1883,10 +1900,10 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);

if (pollRspWrapper->taosxRsp.blockNum == 0) {
rspWrapper = NULL;
tscDebug("consumer:0x%" PRIx64 " taosx empty block received, vgId:%d, reqId:0x%" PRIx64, tmq->consumerId, pVg->vgId,
pollRspWrapper->reqId);
tscDebug("consumer:0x%" PRIx64 " taosx empty block received, vgId:%d, vg total:%" PRId64 " reqId:0x%" PRIx64,
tmq->consumerId, pVg->vgId, pVg->numOfRows, pollRspWrapper->reqId);
pVg->emptyBlockReceiveTs = taosGetTimestampMs();
pRspWrapper = tmqFreeRspWrapper(pRspWrapper);
taosFreeQitem(pollRspWrapper);
continue;
} else {
Expand All @@ -1895,32 +1912,37 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {

// build rsp
void* pRsp = NULL;
int64_t numOfRows = 0;
if (pollRspWrapper->taosxRsp.createTableNum == 0) {
pRsp = tmqBuildRspFromWrapper(pollRspWrapper, pVg);
pRsp = tmqBuildRspFromWrapper(pollRspWrapper, pVg, &numOfRows);
} else {
pRsp = tmqBuildTaosxRspFromWrapper(pollRspWrapper);
}

tmq->totalRows += numOfRows;

char buf[80];
tFormatOffset(buf, 80, &pVg->currentOffset);
tscDebug("consumer:0x%" PRIx64 " process taosx poll rsp, vgId:%d, offset:%s, blocks:%d, reqId:0x%"PRIx64, tmq->consumerId, pVg->vgId,
buf, pollRspWrapper->dataRsp.blockNum, pollRspWrapper->reqId);
tscDebug("consumer:0x%" PRIx64 " process taosx poll rsp, vgId:%d, offset:%s, blocks:%d, rows:%" PRId64
", vg total:%" PRId64 " total:%"PRId64" reqId:0x%" PRIx64,
tmq->consumerId, pVg->vgId, buf, pollRspWrapper->dataRsp.blockNum, numOfRows, pVg->numOfRows,
tmq->totalRows, pollRspWrapper->reqId);

taosFreeQitem(pollRspWrapper);
return pRsp;

} else {
tscDebug("consumer:0x%" PRIx64 " msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
tmq->consumerId, pollRspWrapper->taosxRsp.head.epoch, consumerEpoch);
tmqFreeRspWrapper(rspWrapper);
tscDebug("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d",
tmq->consumerId, pollRspWrapper->vgHandle->vgId, pollRspWrapper->taosxRsp.head.epoch, consumerEpoch);
pRspWrapper = tmqFreeRspWrapper(pRspWrapper);
taosFreeQitem(pollRspWrapper);
}
} else {
tscDebug("consumer:0x%" PRIx64 " not data msg received", tmq->consumerId);

bool reset = false;
tmqHandleNoPollRsp(tmq, rspWrapper, &reset);
taosFreeQitem(rspWrapper);
tmqHandleNoPollRsp(tmq, pRspWrapper, &reset);
taosFreeQitem(pRspWrapper);
if (pollIfReset && reset) {
tscDebug("consumer:0x%" PRIx64 ", reset and repoll", tmq->consumerId);
tmqPollImpl(tmq, timeout);
Expand Down Expand Up @@ -1968,7 +1990,6 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {

if (tmqPollImpl(tmq, timeout) < 0) {
tscDebug("consumer:0x%" PRIx64 " return due to poll error", tmq->consumerId);
/*return NULL;*/
}

rspObj = tmqHandleAllRsp(tmq, timeout, false);
Expand Down Expand Up @@ -1997,6 +2018,8 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
}

int32_t tmq_consumer_close(tmq_t* tmq) {
tscDebug("consumer:0x%" PRIx64" start to close consumer, status:%d", tmq->consumerId, tmq->status);

if (tmq->status == TMQ_CONSUMER_STATUS__READY) {
int32_t rsp = tmq_commit_sync(tmq, NULL);
if (rsp != 0) {
Expand Down