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

fix: memory error

This commit is contained in:
Liu Jicong 2022-04-29 16:35:59 +08:00
parent f542785687
commit 4402833a82
2 changed files with 22 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
example/src/tmq.c
View File

@ -14,7 +14,9 @@
*/
#include <assert.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include "taos.h"
@ -24,7 +26,7 @@ static void msg_process(TAOS_RES* msg) {
char buf[1024];
memset(buf, 0, 1024);
printf("topic: %s\n", tmq_get_topic_name(msg));
printf("vg:%d\n", tmq_get_vgroup_id(msg));
printf("vg: %d\n", tmq_get_vgroup_id(msg));
while (1) {
TAOS_ROW row = taos_fetch_row(msg);
if (row == NULL) break;
@ -141,7 +143,7 @@ int32_t create_topic() {
}
void tmq_commit_cb_print(tmq_t* tmq, tmq_resp_err_t resp, tmq_topic_vgroup_list_t* offsets, void* param) {
printf("commit %d\n", resp);
printf("commit %d tmq %p offsets %p param %p\n", resp, tmq, offsets, param);
}
tmq_t* build_consumer() {
@ -232,6 +234,7 @@ void sync_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
msg_process(tmqmessage);
taos_free_result(tmqmessage);
tmq_commit(tmq, NULL, 1);
/*if ((++msg_count % MIN_COMMIT_COUNT) == 0) tmq_commit(tmq, NULL, 0);*/
}
}

26
source/client/src/tmq.c
View File

@ -377,7 +377,15 @@ int32_t tmqCommitCb(void* param, const SDataBuf* pMsg, int32_t code) {
if (pParam->tmq->commitCb) {
pParam->tmq->commitCb(pParam->tmq, pParam->rspErr, NULL, pParam->tmq->commitCbUserParam);
}
if (!pParam->async) tsem_post(&pParam->rspSem);
if (!pParam->async)
tsem_post(&pParam->rspSem);
else {
tsem_destroy(&pParam->rspSem);
/*if (pParam->pArray) {*/
/*taosArrayDestroy(pParam->pArray);*/
/*}*/
taosMemoryFree(pParam);
}
return 0;
}
@ -560,7 +568,7 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
tscError("failed to malloc request");
}
SMqCommitCbParam* pParam = taosMemoryMalloc(sizeof(SMqCommitCbParam));
SMqCommitCbParam* pParam = taosMemoryCalloc(1, sizeof(SMqCommitCbParam));
if (pParam == NULL) {
return -1;
}
@ -575,6 +583,7 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
};
SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
sendInfo->requestObjRefId = 0;
sendInfo->param = pParam;
sendInfo->fp = tmqCommitCb;
SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
@ -585,13 +594,12 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
if (!async) {
tsem_wait(&pParam->rspSem);
resp = pParam->rspErr;
}
tsem_destroy(&pParam->rspSem);
taosMemoryFree(pParam);
tsem_destroy(&pParam->rspSem);
taosMemoryFree(pParam);
if (pArray) {
taosArrayDestroy(pArray);
if (pArray) {
taosArrayDestroy(pArray);
}
}
return resp;
@ -1313,7 +1321,7 @@ const char* tmq_err2str(tmq_resp_err_t err) {
const char* tmq_get_topic_name(TAOS_RES* res) {
if (TD_RES_TMQ(res)) {
SMqRspObj* pRspObj = (SMqRspObj*)res;
return pRspObj->topic;
return strchr(pRspObj->topic, '.') + 1;
} else {
return NULL;
}