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

fix mem leak

This commit is contained in:
Liu Jicong 2023-01-03 11:18:48 +08:00
parent d33bcc9977
commit 617d8d4013
7 changed files with 28 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -912,10 +912,12 @@ void tmqFreeImpl(void* handle) {
tmq_t* tmq = (tmq_t*)handle;
// TODO stop timer
tmqClearUnhandleMsg(tmq);
if (tmq->mqueue) taosCloseQueue(tmq->mqueue);
if (tmq->mqueue) {
tmqClearUnhandleMsg(tmq);
taosCloseQueue(tmq->mqueue);
}
if (tmq->delayedTask) taosCloseQueue(tmq->delayedTask);
if (tmq->qall) taosFreeQall(tmq->qall);
taosFreeQall(tmq->qall);
tsem_destroy(&tmq->rspSem);

6
source/dnode/mgmt/mgmt_snode/src/smWorker.c
View File

@ -58,11 +58,7 @@ static void smProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
dTrace("msg:%p, get from snode-stream queue", pMsg);
int32_t code = sndProcessStreamMsg(pMgmt->pSnode, pMsg);
if (code < 0) {
if (pMsg) {
dGError("snd, msg:%p failed to process stream msg %s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr(code));
} else {
dGError("snd, msg:%p failed to process stream empty msg since %s", pMsg, terrstr(code));
}
dGError("snd, msg:%p failed to process stream msg %s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr(code));
smSendRsp(pMsg, terrno);
}

16
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -86,12 +86,8 @@ static void vmProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);
if (code != 0) {
if (terrno != 0) code = terrno;
if (pMsg) {
dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),
terrstr(code));
} else {
dGError("vgId:%d, msg:%p failed to process stream empty msg since %s", pVnode->vgId, pMsg, terrstr(code));
}
dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),
terrstr(code));
vmSendRsp(pMsg, code);
}
@ -146,16 +142,16 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
return -1;
}
SMsgHead *pHead = pMsg->pCont;
int32_t code = 0;
SMsgHead *pHead = pMsg->pCont;
int32_t code = 0;
pHead->contLen = ntohl(pHead->contLen);
pHead->vgId = ntohl(pHead->vgId);
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL) {
dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg, terrstr(),
TMSG_INFO(pMsg->msgType), qtype, pHead->contLen);
dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg,
terrstr(), TMSG_INFO(pMsg->msgType), qtype, pHead->contLen);
return terrno != 0 ? terrno : -1;
}

3
source/dnode/vnode/src/tq/tqMeta.c
View File

@ -284,8 +284,7 @@ int32_t tqMetaRestoreHandle(STQ* pTq) {
handle.pRef = walOpenRef(pTq->pVnode->pWal);
if (handle.pRef == NULL) {
ASSERT(0);
return -1;
continue;
}
walRefVer(handle.pRef, handle.snapshotVer);

17
source/dnode/vnode/src/tq/tqOffset.c
View File

@ -46,20 +46,25 @@ int32_t tqOffsetRestoreFromFile(STqOffsetStore* pStore, const char* fname) {
}
int32_t size = htonl(head.size);
void* memBuf = taosMemoryCalloc(1, size);
if (memBuf == NULL) {
return -1;
}
if ((code = taosReadFile(pFile, memBuf, size)) != size) {
ASSERT(0);
// TODO handle error
taosMemoryFree(memBuf);
return -1;
}
STqOffset offset;
SDecoder decoder;
tDecoderInit(&decoder, memBuf, size);
if (tDecodeSTqOffset(&decoder, &offset) < 0) {
ASSERT(0);
taosMemoryFree(memBuf);
tDecoderClear(&decoder);
return -1;
}
tDecoderClear(&decoder);
if (taosHashPut(pStore->pHash, offset.subKey, strlen(offset.subKey), &offset, sizeof(STqOffset)) < 0) {
ASSERT(0);
// TODO
return -1;
}
taosMemoryFree(memBuf);
}
@ -124,7 +129,7 @@ int32_t tqOffsetCommitFile(STqOffsetStore* pStore) {
const char* sysErrStr = strerror(errno);
tqError("vgId:%d, cannot open file %s when commit offset since %s", pStore->pTq->pVnode->config.vgId, fname,
sysErrStr);
ASSERT(0);
taosMemoryFree(fname);
return -1;
}
taosMemoryFree(fname);

2
source/dnode/vnode/src/tq/tqSink.c
View File

@ -64,7 +64,7 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
.startTs = startTs,
.endTs = endTs,
};
strncpy(req.tbname, name, TSDB_TABLE_NAME_LEN);
strncpy(req.tbname, name, TSDB_TABLE_NAME_LEN - 1);
taosMemoryFree(name);
/*tqDebug("stream delete msg, active: vgId:%d, ts:%" PRId64 " name:%s", pVnode->config.vgId, ts, name);*/
taosArrayPush(deleteReq->deleteReqs, &req);

4
source/dnode/vnode/src/tq/tqSnapshot.c
View File

@ -175,6 +175,8 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
if (code) goto _err;
}
int vgId = TD_VID(pWriter->pTq->pVnode);
taosMemoryFree(pWriter);
*ppWriter = NULL;
@ -186,7 +188,7 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
return code;
_err:
tqError("vgId:%d, tq snapshot writer close failed since %s", TD_VID(pWriter->pTq->pVnode), tstrerror(code));
tqError("vgId:%d, tq snapshot writer close failed since %s", vgId, tstrerror(code));
return code;
}