diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index 450d7a30ed1f0863c4bfafcd0725a5151ff00146..4dd7e070b331f5eded04dd69f08d9e6a971d9048 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -349,6 +349,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) { } rpcFreeCont(req.pCont); + req.pCont = NULL; if (code != 0) { mError("trans:%d, failed to propose, code:0x%x", pMgmt->transId, code); return code; diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index e81f63d871e053032aa289076e08a1a1d860f3f6..485186fda6ce3ab1314c690a3fe330902ccc7d8d 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -70,7 +70,7 @@ typedef struct SSyncTimer { uint64_t counter; int32_t timerMS; SRaftId destId; - void* pData; + SSyncHbTimerData hbData; } SSyncTimer; typedef struct SElectTimer { diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index e802f60f30d978ee6ef9cd95cb542d7e7fc10ec5..4fa4b7e49fdb53fca947f727fb85e6beb8762476 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -385,7 +385,7 @@ bool syncIsReadyForRead(int64_t rid) { if (!pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore)) { SSyncRaftEntry* pEntry = NULL; int32_t code = pSyncNode->pLogStore->syncLogGetEntry( - pSyncNode->pLogStore, pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore), &pEntry); + pSyncNode->pLogStore, pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore), &pEntry); if (code == 0 && pEntry != NULL) { if (pEntry->originalRpcType == TDMT_SYNC_NOOP && pEntry->term == pSyncNode->pRaftStore->currentTerm) { ready = true; @@ -442,7 +442,9 @@ int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) { pMsg->newLeaderId.vgId = pSyncNode->vgId; pMsg->newNodeInfo = newLeader; - return syncNodePropose(pSyncNode, &rpcMsg, false); + int32_t ret = syncNodePropose(pSyncNode, &rpcMsg, false); + rpcFreeCont(rpcMsg.pCont); + return ret; } SSyncState syncGetState(int64_t rid) { @@ -645,13 +647,12 @@ static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRa static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { int32_t ret = 0; if (syncIsInit()) { - SSyncHbTimerData* pData = taosMemoryMalloc(sizeof(SSyncHbTimerData)); + SSyncHbTimerData* pData = &pSyncTimer->hbData; pData->pSyncNode = pSyncNode; pData->pTimer = pSyncTimer; pData->destId = pSyncTimer->destId; pData->logicClock = pSyncTimer->logicClock; - pSyncTimer->pData = pData; taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, pData, syncEnv()->pTimerManager, &pSyncTimer->pTimer); } else { sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId); @@ -664,7 +665,6 @@ static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { atomic_add_fetch_64(&pSyncTimer->logicClock, 1); taosTmrStop(pSyncTimer->pTimer); pSyncTimer->pTimer = NULL; - // taosMemoryFree(pSyncTimer->pData); return ret; } @@ -1086,15 +1086,8 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) { int32_t ret = 0; if (syncIsInit()) { pSyncNode->electTimerMS = ms; - - SElectTimer* pElectTimer = taosMemoryMalloc(sizeof(SElectTimer)); - pElectTimer->logicClock = pSyncNode->electTimerLogicClock; - pElectTimer->pSyncNode = pSyncNode; - pElectTimer->pData = NULL; - - taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pElectTimer, syncEnv()->pTimerManager, + taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, syncEnv()->pTimerManager, &pSyncNode->pElectTimer); - } else { sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId); } @@ -1827,20 +1820,17 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) { } static void syncNodeEqElectTimer(void* param, void* tmrId) { - if (!syncIsInit()) return; - - SElectTimer* pElectTimer = param; - SSyncNode* pNode = pElectTimer->pSyncNode; + SSyncNode* pNode = param; + if (!syncIsInit()) return; if (pNode == NULL) return; if (pNode->syncEqMsg == NULL) return; SRpcMsg rpcMsg = {0}; - int32_t code = syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pElectTimer->logicClock, pNode->electTimerMS, pNode); - + int32_t code = + syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pNode->electTimerLogicClock, pNode->electTimerMS, pNode); if (code != 0) { sError("failed to build elect msg"); - taosMemoryFree(pElectTimer); return; } @@ -1851,12 +1841,8 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) { if (code != 0) { sError("failed to sync enqueue elect msg since %s", terrstr()); rpcFreeCont(rpcMsg.pCont); - taosMemoryFree(pElectTimer); - return; } - taosMemoryFree(pElectTimer); - #if 0 // reset timer ms if (syncIsInit() && pNode->electBaseLine > 0) { diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 2e9b942df4defd98edb870b3bef260369b3fc8cf..2369b238ea9563c94ef8ebae31ce04da841a1758 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -242,8 +242,8 @@ ,,y,script,./test.sh -f tsim/stream/fillIntervalValue.sim ,,y,script,./test.sh -f tsim/trans/lossdata1.sim ,,y,script,./test.sh -f tsim/trans/create_db.sim -,,,script,./test.sh -f tsim/tmq/basic1.sim -,,,script,./test.sh -f tsim/tmq/basic2.sim +,,y,script,./test.sh -f tsim/tmq/basic1.sim +,,y,script,./test.sh -f tsim/tmq/basic2.sim ,,,script,./test.sh -f tsim/tmq/basic3.sim ,,,script,./test.sh -f tsim/tmq/basic4.sim ,,,script,./test.sh -f tsim/tmq/basic1Of2Cons.sim