From ddf7dcc94f637ce0a2e3df75e61da629326bbdc4 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 18 Jan 2022 16:02:12 +0800 Subject: [PATCH 1/5] fix mem leak --- include/common/tmsg.h | 11 ++++--- source/client/src/clientHb.c | 12 ++++--- source/client/test/clientTests.cpp | 4 ++- source/dnode/mnode/impl/inc/mndDef.h | 18 +++++++++-- source/dnode/mnode/impl/src/mndConsumer.c | 39 ++++++++++++++--------- source/dnode/mnode/impl/src/mndProfile.c | 5 +++ source/dnode/mnode/impl/src/mnode.c | 11 +++++++ 7 files changed, 73 insertions(+), 27 deletions(-) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index f32fdcbae7..dfd376f1e9 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -188,16 +188,19 @@ void* tDeserializeSClientHbRsp(void* buf, SClientHbRsp* pRsp); static FORCE_INLINE void tFreeClientHbReq(void *pReq) { SClientHbReq* req = (SClientHbReq*)pReq; - taosHashCleanup(req->info); - free(pReq); + if (req->info) taosHashCleanup(req->info); } int tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pReq); void* tDeserializeSClientHbBatchReq(void* buf, SClientHbBatchReq* pReq); -static FORCE_INLINE void tFreeClientHbBatchReq(void* pReq) { +static FORCE_INLINE void tFreeClientHbBatchReq(void* pReq, bool deep) { SClientHbBatchReq *req = (SClientHbBatchReq*)pReq; - //taosArrayDestroyEx(req->reqs, tFreeClientHbReq); + if (deep) { + taosArrayDestroyEx(req->reqs, tFreeClientHbReq); + } else { + taosArrayDestroy(req->reqs); + } free(pReq); } diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index 6d7fc9f81a..97ef77abcc 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -60,15 +60,17 @@ SClientHbBatchReq* hbGatherAllInfo(SAppHbMgr *pAppHbMgr) { pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter); } +#if 0 pIter = taosHashIterate(pAppHbMgr->getInfoFuncs, NULL); while (pIter != NULL) { FGetConnInfo getConnInfoFp = (FGetConnInfo)pIter; SClientHbKey connKey; taosHashCopyKey(pIter, &connKey); - getConnInfoFp(connKey, NULL); + SArray* pArray = getConnInfoFp(connKey, NULL); pIter = taosHashIterate(pAppHbMgr->getInfoFuncs, pIter); } +#endif return pBatchReq; } @@ -99,12 +101,12 @@ static void* hbThreadFunc(void* param) { //TODO: error handling break; } - void *bufCopy = buf; - tSerializeSClientHbBatchReq(&bufCopy, pReq); + void *abuf = buf; + tSerializeSClientHbBatchReq(&abuf, pReq); SMsgSendInfo *pInfo = malloc(sizeof(SMsgSendInfo)); if (pInfo == NULL) { terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; - tFreeClientHbBatchReq(pReq); + tFreeClientHbBatchReq(pReq, false); free(buf); break; } @@ -120,7 +122,7 @@ static void* hbThreadFunc(void* param) { int64_t transporterId = 0; SEpSet epSet = getEpSet_s(&pAppInstInfo->mgmtEp); asyncSendMsgToServer(pAppInstInfo->pTransporter, &epSet, &transporterId, pInfo); - tFreeClientHbBatchReq(pReq); + tFreeClientHbBatchReq(pReq, false); atomic_add_fetch_32(&pAppHbMgr->reportCnt, 1); } diff --git a/source/client/test/clientTests.cpp b/source/client/test/clientTests.cpp index 415d6a57ce..13915fd85d 100644 --- a/source/client/test/clientTests.cpp +++ b/source/client/test/clientTests.cpp @@ -53,7 +53,9 @@ TEST(testCase, connect_Test) { if (pConn == NULL) { printf("failed to connect to server, reason:%s\n", taos_errstr(NULL)); } - sleep(3); + while (1) { + sleep(3); + } taos_close(pConn); } diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index de101b0f06..7da0f3826e 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -325,6 +325,19 @@ typedef struct SMqTopicConsumer { } SMqTopicConsumer; #endif +typedef struct SMqConsumerEp { + int32_t vgId; + SEpSet epset; + int64_t consumerId; +} SMqConsumerEp; + +typedef struct SMqCgroupTopicPair { + char key[TSDB_CONSUMER_GROUP_LEN + TSDB_TOPIC_FNAME_LEN]; + SArray* assigned; + SArray* unassignedConsumer; + SArray* unassignedVg; +} SMqCgroupTopicPair; + typedef struct SMqCGroup { char name[TSDB_CONSUMER_GROUP_LEN]; int32_t status; // 0 - uninitialized, 1 - wait rebalance, 2- normal @@ -351,10 +364,11 @@ typedef struct SMqTopicObj { // TODO: add cache and change name to id typedef struct SMqConsumerTopic { + char name[TSDB_TOPIC_FNAME_LEN]; int32_t epoch; - char name[TSDB_TOPIC_NAME_LEN]; //TODO: replace with something with ep SList *vgroups; // SList + SArray *pVgInfo; // SArray } SMqConsumerTopic; typedef struct SMqConsumerObj { @@ -362,7 +376,7 @@ typedef struct SMqConsumerObj { SRWLatch lock; char cgroup[TSDB_CONSUMER_GROUP_LEN]; SArray *topics; // SArray - SHashObj *topicHash; + SHashObj *topicHash; //SHashObj } SMqConsumerObj; typedef struct SMqSubConsumerObj { diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 54e640d8b7..d27bf53a90 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -204,34 +204,37 @@ void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) { static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) { SMnode *pMnode = pMsg->pMnode; char *msgStr = pMsg->rpcMsg.pCont; - SCMSubscribeReq *pSubscribe; - tDeserializeSCMSubscribeReq(msgStr, pSubscribe); - int64_t consumerId = pSubscribe->consumerId; - char *consumerGroup = pSubscribe->consumerGroup; + SCMSubscribeReq subscribe; + tDeserializeSCMSubscribeReq(msgStr, &subscribe); + int64_t consumerId = subscribe.consumerId; + char *consumerGroup = subscribe.consumerGroup; int32_t cgroupLen = strlen(consumerGroup); SArray *newSub = NULL; - int newTopicNum = pSubscribe->topicNum; + int newTopicNum = subscribe.topicNum; if (newTopicNum) { newSub = taosArrayInit(newTopicNum, sizeof(SMqConsumerTopic)); } + SMqConsumerTopic *pConsumerTopics = calloc(newTopicNum, sizeof(SMqConsumerTopic)); + if (pConsumerTopics == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } for (int i = 0; i < newTopicNum; i++) { char *newTopicName = taosArrayGetP(newSub, i); - SMqConsumerTopic *pConsumerTopic = malloc(sizeof(SMqConsumerTopic)); - if (pConsumerTopic == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - // TODO: free - return -1; - } + SMqConsumerTopic *pConsumerTopic = &pConsumerTopics[i]; + strcpy(pConsumerTopic->name, newTopicName); pConsumerTopic->vgroups = tdListNew(sizeof(int64_t)); - taosArrayPush(newSub, pConsumerTopic); - free(pConsumerTopic); } + + taosArrayAddBatch(newSub, pConsumerTopics, newTopicNum); + free(pConsumerTopics); taosArraySortString(newSub, taosArrayCompareString); SArray *oldSub = NULL; int oldTopicNum = 0; + // create consumer if not exist SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId); if (pConsumer == NULL) { // create consumer @@ -249,6 +252,7 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) { } STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg); if (pTrans == NULL) { + //TODO: free memory return -1; } @@ -286,6 +290,7 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) { } if (pOldTopic != NULL) { + //cancel subscribe of that old topic ASSERT(pNewTopic == NULL); char *oldTopicName = pOldTopic->name; SList *vgroups = pOldTopic->vgroups; @@ -298,13 +303,14 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) { SMqCGroup *pGroup = taosHashGet(pTopic->cgroups, consumerGroup, cgroupLen); while ((pn = tdListNext(&iter)) != NULL) { int32_t vgId = *(int64_t *)pn->data; + // acquire and get epset SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId); - // TODO release + // TODO what time to release? if (pVgObj == NULL) { // TODO handle error continue; } - // acquire and get epset + //build reset msg void *pMqVgSetReq = mndBuildMqVGroupSetReq(pMnode, oldTopicName, vgId, consumerId, consumerGroup); // TODO:serialize if (pMsg == NULL) { @@ -323,10 +329,12 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) { return -1; } } + //delete data in mnode taosHashRemove(pTopic->cgroups, consumerGroup, cgroupLen); mndReleaseTopic(pMnode, pTopic); } else if (pNewTopic != NULL) { + // save subscribe info to mnode ASSERT(pOldTopic == NULL); char *newTopicName = pNewTopic->name; @@ -351,6 +359,7 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) { // add into cgroups taosHashPut(pTopic->cgroups, consumerGroup, cgroupLen, pGroup, sizeof(SMqCGroup)); } + /*taosHashPut(pTopic->consumers, &pConsumer->consumerId, sizeof(int64_t), pConsumer, sizeof(SMqConsumerObj));*/ // put the consumer into list // rebalance will be triggered by timer diff --git a/source/dnode/mnode/impl/src/mndProfile.c b/source/dnode/mnode/impl/src/mndProfile.c index 902eaa5c1c..3444a2409a 100644 --- a/source/dnode/mnode/impl/src/mndProfile.c +++ b/source/dnode/mnode/impl/src/mndProfile.c @@ -357,6 +357,11 @@ static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) { } } } + for (int i = 0; i < sz; i++) { + SClientHbReq* pHbReq = taosArrayGet(pArray, i); + tFreeClientHbReq(pHbReq); + } + taosArrayDestroy(pArray); int32_t tlen = tSerializeSClientHbBatchRsp(NULL, &batchRsp); void* buf = rpcMallocCont(tlen); void* abuf = buf; diff --git a/source/dnode/mnode/impl/src/mnode.c b/source/dnode/mnode/impl/src/mnode.c index d70c93e758..cab30702ea 100644 --- a/source/dnode/mnode/impl/src/mnode.c +++ b/source/dnode/mnode/impl/src/mnode.c @@ -69,6 +69,17 @@ static void mndTransReExecute(void *param, void *tmrId) { taosTmrReset(mndTransReExecute, 3000, pMnode, pMnode->timer, &pMnode->transTimer); } +static void mndCalMqRebalance(void* param, void* tmrId) { + SMnode* pMnode = param; + if (mndIsMaster(pMnode)) { + // iterate cgroup, cal rebalance + // sync with raft + // write sdb + } + + taosTmrReset(mndCalMqRebalance, 3000, pMnode, pMnode->timer, &pMnode->transTimer); +} + static int32_t mndInitTimer(SMnode *pMnode) { if (pMnode->timer == NULL) { pMnode->timer = taosTmrInit(5000, 200, 3600000, "MND"); From d14a599a32bda1bdad49e14b77c598a40d6531d5 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 18 Jan 2022 16:15:24 +0800 Subject: [PATCH 2/5] fix mem leak --- source/dnode/mnode/impl/src/mndProfile.c | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndProfile.c b/source/dnode/mnode/impl/src/mndProfile.c index 3444a2409a..3773750ed3 100644 --- a/source/dnode/mnode/impl/src/mndProfile.c +++ b/source/dnode/mnode/impl/src/mndProfile.c @@ -357,15 +357,13 @@ static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) { } } } - for (int i = 0; i < sz; i++) { - SClientHbReq* pHbReq = taosArrayGet(pArray, i); - tFreeClientHbReq(pHbReq); - } - taosArrayDestroy(pArray); + taosArrayDestroyEx(pArray, tFreeClientHbReq); + int32_t tlen = tSerializeSClientHbBatchRsp(NULL, &batchRsp); void* buf = rpcMallocCont(tlen); void* abuf = buf; tSerializeSClientHbBatchRsp(&abuf, &batchRsp); + taosArrayDestroy(batchRsp.rsps); pReq->contLen = tlen; pReq->pCont = buf; return 0; From cc8a02aeb496bcb2901d3c9569f80c4f56265fb2 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 18 Jan 2022 16:17:58 +0800 Subject: [PATCH 3/5] fix mem leak --- source/libs/wal/src/walMeta.c | 1 + 1 file changed, 1 insertion(+) diff --git a/source/libs/wal/src/walMeta.c b/source/libs/wal/src/walMeta.c index d630080086..a3894ceedd 100644 --- a/source/libs/wal/src/walMeta.c +++ b/source/libs/wal/src/walMeta.c @@ -149,6 +149,7 @@ int walCheckAndRepairMeta(SWal* pWal) { } } + closedir(dir); regfree(&logRegPattern); regfree(&idxRegPattern); From 623acecafdbfae6aa1471214040a001ea90b63fd Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 18 Jan 2022 16:24:54 +0800 Subject: [PATCH 4/5] init hb count --- source/client/src/clientHb.c | 3 +++ source/client/test/clientTests.cpp | 4 +--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index 97ef77abcc..0f4ff6f725 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -157,6 +157,9 @@ SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo) { } // init stat pAppHbMgr->startTime = taosGetTimestampMs(); + pAppHbMgr->connKeyCnt = 0; + pAppHbMgr->reportCnt = 0; + pAppHbMgr->reportBytes = 0; // init app info pAppHbMgr->pAppInstInfo = pAppInstInfo; diff --git a/source/client/test/clientTests.cpp b/source/client/test/clientTests.cpp index 13915fd85d..415d6a57ce 100644 --- a/source/client/test/clientTests.cpp +++ b/source/client/test/clientTests.cpp @@ -53,9 +53,7 @@ TEST(testCase, connect_Test) { if (pConn == NULL) { printf("failed to connect to server, reason:%s\n", taos_errstr(NULL)); } - while (1) { - sleep(3); - } + sleep(3); taos_close(pConn); } From 66f610f8df52cf8897452fb06da972066a369ad4 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 18 Jan 2022 16:34:09 +0800 Subject: [PATCH 5/5] make jenkins happy --- source/dnode/mnode/impl/inc/mndDef.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 7da0f3826e..a2d6bbf4e6 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -333,7 +333,7 @@ typedef struct SMqConsumerEp { typedef struct SMqCgroupTopicPair { char key[TSDB_CONSUMER_GROUP_LEN + TSDB_TOPIC_FNAME_LEN]; - SArray* assigned; + SArray* assigned; // SArray SArray* unassignedConsumer; SArray* unassignedVg; } SMqCgroupTopicPair;