From 55a84dab92f833a7716abba68d3c90a4748f93ab Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 15 Nov 2022 18:15:27 +0800 Subject: [PATCH 1/5] test: add asan case --- tests/parallel_test/cases.task | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 9b1d829c9a..abf5d76733 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -12,7 +12,7 @@ ,,y,script,./test.sh -f tsim/user/privilege_sysinfo.sim ,,y,script,./test.sh -f tsim/db/alter_option.sim ,,,script,./test.sh -f tsim/db/alter_replica_13.sim -,,,script,./test.sh -f tsim/db/alter_replica_31.sim +,,y,script,./test.sh -f tsim/db/alter_replica_31.sim ,,y,script,./test.sh -f tsim/db/basic1.sim ,,y,script,./test.sh -f tsim/db/basic2.sim ,,y,script,./test.sh -f tsim/db/basic3.sim @@ -149,27 +149,27 @@ ,,y,script,./test.sh -f tsim/parser/selectResNum.sim ,,y,script,./test.sh -f tsim/parser/set_tag_vals.sim ,,y,script,./test.sh -f tsim/parser/single_row_in_tb.sim -,,,script,./test.sh -f tsim/parser/sliding.sim +,,y,script,./test.sh -f tsim/parser/sliding.sim ,,y,script,./test.sh -f tsim/parser/slimit_alter_tags.sim ,,y,script,./test.sh -f tsim/parser/slimit.sim ,,y,script,./test.sh -f tsim/parser/slimit1.sim ,,y,script,./test.sh -f tsim/parser/stableOp.sim ,,y,script,./test.sh -f tsim/parser/tags_dynamically_specifiy.sim -,,,script,./test.sh -f tsim/parser/tags_filter.sim -,,,script,./test.sh -f tsim/parser/tbnameIn.sim +,,y,script,./test.sh -f tsim/parser/tags_filter.sim +,,y,script,./test.sh -f tsim/parser/tbnameIn.sim ,,y,script,./test.sh -f tsim/parser/timestamp.sim ,,y,script,./test.sh -f tsim/parser/top_groupby.sim ,,y,script,./test.sh -f tsim/parser/topbot.sim ,,,script,./test.sh -f tsim/parser/union.sim ,,y,script,./test.sh -f tsim/parser/union_sysinfo.sim -,,,script,./test.sh -f tsim/parser/where.sim +,,y,script,./test.sh -f tsim/parser/where.sim ,,y,script,./test.sh -f tsim/query/charScalarFunction.sim ,,y,script,./test.sh -f tsim/query/explain.sim ,,y,script,./test.sh -f tsim/query/interval-offset.sim ,,y,script,./test.sh -f tsim/query/interval.sim ,,y,script,./test.sh -f tsim/query/scalarFunction.sim ,,y,script,./test.sh -f tsim/query/scalarNull.sim -,,,script,./test.sh -f tsim/query/session.sim +,,y,script,./test.sh -f tsim/query/session.sim ,,,script,./test.sh -f tsim/query/udf.sim ,,y,script,./test.sh -f tsim/qnode/basic1.sim ,,y,script,./test.sh -f tsim/snode/basic1.sim @@ -215,17 +215,17 @@ ,,y,script,./test.sh -f tsim/stream/fillHistoryBasic2.sim ,,y,script,./test.sh -f tsim/stream/fillHistoryBasic3.sim ,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim -,,,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim +,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim ,,,script,./test.sh -f tsim/stream/distributeSession0.sim ,,,script,./test.sh -f tsim/stream/session0.sim -,,,script,./test.sh -f tsim/stream/session1.sim +,,y,script,./test.sh -f tsim/stream/session1.sim ,,y,script,./test.sh -f tsim/stream/state0.sim ,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim -,,,script,./test.sh -f tsim/stream/triggerSession0.sim +,,y,script,./test.sh -f tsim/stream/triggerSession0.sim ,,y,script,./test.sh -f tsim/stream/partitionby.sim ,,y,script,./test.sh -f tsim/stream/partitionby1.sim ,,y,script,./test.sh -f tsim/stream/schedSnode.sim -,,,script,./test.sh -f tsim/stream/windowClose.sim +,,y,script,./test.sh -f tsim/stream/windowClose.sim ,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim ,,y,script,./test.sh -f tsim/stream/sliding.sim ,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim @@ -251,7 +251,7 @@ ,,y,script,./test.sh -f tsim/tmq/basic3Of2Cons.sim ,,y,script,./test.sh -f tsim/tmq/basic4Of2Cons.sim ,,y,script,./test.sh -f tsim/tmq/basic2Of2ConsOverlap.sim -,,,script,./test.sh -f tsim/tmq/topic.sim +,,y,script,./test.sh -f tsim/tmq/topic.sim ,,y,script,./test.sh -f tsim/tmq/snapshot.sim ,,y,script,./test.sh -f tsim/tmq/snapshot1.sim ,,y,script,./test.sh -f tsim/stable/alter_comment.sim @@ -278,7 +278,7 @@ ,,,script,./test.sh -f tsim/sma/drop_sma.sim ,,,script,./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim ,,y,script,./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim -,,,script,./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim +,,y,script,./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim ,,n,script,./test.sh -f tsim/valgrind/checkError1.sim ,,n,script,./test.sh -f tsim/valgrind/checkError2.sim ,,n,script,./test.sh -f tsim/valgrind/checkError3.sim @@ -297,8 +297,8 @@ ,,y,script,./test.sh -f tsim/vnode/stable_dnode2_stop.sim ,,y,script,./test.sh -f tsim/vnode/stable_dnode2.sim ,,y,script,./test.sh -f tsim/vnode/stable_dnode3.sim -,,,script,./test.sh -f tsim/vnode/stable_replica3_dnode6.sim -,,,script,./test.sh -f tsim/vnode/stable_replica3_vnode3.sim +,,y,script,./test.sh -f tsim/vnode/stable_replica3_dnode6.sim +,,y,script,./test.sh -f tsim/vnode/stable_replica3_vnode3.sim ,,y,script,./test.sh -f tsim/sync/3Replica1VgElect.sim ,,y,script,./test.sh -f tsim/sync/3Replica5VgElect.sim ,,y,script,./test.sh -f tsim/sync/oneReplica1VgElect.sim From a9bf05bc018d338103c01874dc41b64fdd4ac054 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 15 Nov 2022 18:32:08 +0800 Subject: [PATCH 2/5] refactor: stream meta ref count --- include/libs/stream/tstream.h | 7 +++- source/dnode/mnode/impl/src/mndStream.c | 1 + source/dnode/snode/src/snode.c | 23 ++++++++---- source/dnode/vnode/src/tq/tq.c | 48 ++++++++++++++++++------- source/libs/executor/src/scanoperator.c | 2 ++ source/libs/stream/src/streamMeta.c | 41 +++++++++++++++++++++ source/libs/wal/src/walRef.c | 4 +-- 7 files changed, 103 insertions(+), 23 deletions(-) diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 0354078b7b..ecd1b6f916 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -338,7 +338,7 @@ typedef struct SStreamTask { int32_t recoverWaitingUpstream; int64_t checkReqId; SArray* checkReqIds; // shuffle - + int32_t refCnt; } SStreamTask; int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo); @@ -565,6 +565,7 @@ typedef struct SStreamMeta { TXN txn; FTaskExpand* expandFunc; int32_t vgId; + SRWLatch lock; } SStreamMeta; SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandFunc, int32_t vgId); @@ -575,6 +576,10 @@ int32_t streamMetaAddSerializedTask(SStreamMeta* pMeta, int64_t startVer, c int32_t streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId); SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId); +SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int32_t taskId); +void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask); +void streamMetaRemoveTask1(SStreamMeta* pMeta, int32_t taskId); + int32_t streamMetaBegin(SStreamMeta* pMeta); int32_t streamMetaCommit(SStreamMeta* pMeta); int32_t streamMetaRollBack(SStreamMeta* pMeta); diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index c649180285..2cfb81a817 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -738,6 +738,7 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) { } sdbRelease(pMnode->pSdb, pStream); + mndTransDrop(pTrans); return TSDB_CODE_ACTION_IN_PROGRESS; } diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c index f53350c10b..aa55204ae5 100644 --- a/source/dnode/snode/src/snode.c +++ b/source/dnode/snode/src/snode.c @@ -36,13 +36,14 @@ void sndEnqueueStreamDispatch(SSnode *pSnode, SRpcMsg *pMsg) { int32_t taskId = req.taskId; - SStreamTask *pTask = streamMetaGetTask(pSnode->pMeta, taskId); + SStreamTask *pTask = streamMetaAcquireTask(pSnode->pMeta, taskId); if (pTask) { SRpcMsg rsp = { .info = pMsg->info, .code = 0, }; streamProcessDispatchReq(pTask, &req, &rsp, false); + streamMetaReleaseTask(pSnode->pMeta, pTask); rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); return; @@ -63,6 +64,7 @@ int32_t sndExpandTask(SSnode *pSnode, SStreamTask *pTask, int64_t ver) { ASSERT(pTask->taskLevel == TASK_LEVEL__AGG); ASSERT(taosArrayGetSize(pTask->childEpInfo) != 0); + pTask->refCnt = 1; pTask->schedStatus = TASK_SCHED_STATUS__INACTIVE; pTask->inputQueue = streamQueueOpen(); pTask->outputQueue = streamQueueOpen(); @@ -166,15 +168,17 @@ int32_t sndProcessTaskDeployReq(SSnode *pSnode, char *msg, int32_t msgLen) { int32_t sndProcessTaskDropReq(SSnode *pSnode, char *msg, int32_t msgLen) { SVDropStreamTaskReq *pReq = (SVDropStreamTaskReq *)msg; - return streamMetaRemoveTask(pSnode->pMeta, pReq->taskId); + streamMetaRemoveTask1(pSnode->pMeta, pReq->taskId); + return 0; } int32_t sndProcessTaskRunReq(SSnode *pSnode, SRpcMsg *pMsg) { SStreamTaskRunReq *pReq = pMsg->pCont; int32_t taskId = pReq->taskId; - SStreamTask *pTask = streamMetaGetTask(pSnode->pMeta, taskId); + SStreamTask *pTask = streamMetaAcquireTask(pSnode->pMeta, taskId); if (pTask) { streamProcessRunReq(pTask); + streamMetaReleaseTask(pSnode->pMeta, pTask); return 0; } else { return -1; @@ -191,13 +195,14 @@ int32_t sndProcessTaskDispatchReq(SSnode *pSnode, SRpcMsg *pMsg, bool exec) { tDecodeStreamDispatchReq(&decoder, &req); int32_t taskId = req.taskId; - SStreamTask *pTask = streamMetaGetTask(pSnode->pMeta, taskId); + SStreamTask *pTask = streamMetaAcquireTask(pSnode->pMeta, taskId); if (pTask) { SRpcMsg rsp = { .info = pMsg->info, .code = 0, }; streamProcessDispatchReq(pTask, &req, &rsp, exec); + streamMetaReleaseTask(pSnode->pMeta, pTask); return 0; } else { return -1; @@ -215,13 +220,14 @@ int32_t sndProcessTaskRetrieveReq(SSnode *pSnode, SRpcMsg *pMsg) { tDecodeStreamRetrieveReq(&decoder, &req); tDecoderClear(&decoder); int32_t taskId = req.dstTaskId; - SStreamTask *pTask = streamMetaGetTask(pSnode->pMeta, taskId); + SStreamTask *pTask = streamMetaAcquireTask(pSnode->pMeta, taskId); if (pTask) { SRpcMsg rsp = { .info = pMsg->info, .code = 0, }; streamProcessRetrieveReq(pTask, &req, &rsp); + streamMetaReleaseTask(pSnode->pMeta, pTask); tDeleteStreamRetrieveReq(&req); return 0; } else { @@ -232,9 +238,10 @@ int32_t sndProcessTaskRetrieveReq(SSnode *pSnode, SRpcMsg *pMsg) { int32_t sndProcessTaskDispatchRsp(SSnode *pSnode, SRpcMsg *pMsg) { SStreamDispatchRsp *pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); int32_t taskId = ntohl(pRsp->upstreamTaskId); - SStreamTask *pTask = streamMetaGetTask(pSnode->pMeta, taskId); + SStreamTask *pTask = streamMetaAcquireTask(pSnode->pMeta, taskId); if (pTask) { streamProcessDispatchRsp(pTask, pRsp, pMsg->code); + streamMetaReleaseTask(pSnode->pMeta, pTask); return 0; } else { return -1; @@ -274,15 +281,17 @@ int32_t sndProcessTaskRecoverFinishReq(SSnode *pSnode, SRpcMsg *pMsg) { tDecoderClear(&decoder); // find task - SStreamTask *pTask = streamMetaGetTask(pSnode->pMeta, req.taskId); + SStreamTask *pTask = streamMetaAcquireTask(pSnode->pMeta, req.taskId); if (pTask == NULL) { return -1; } // do process request if (streamProcessRecoverFinishReq(pTask, req.childId) < 0) { + streamMetaReleaseTask(pSnode->pMeta, pTask); return -1; } + streamMetaReleaseTask(pSnode->pMeta, pTask); return 0; } diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 1cc771dbb4..419d4c3b9b 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -882,6 +882,7 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver) { ASSERT(taosArrayGetSize(pTask->childEpInfo) != 0); } + pTask->refCnt = 1; pTask->schedStatus = TASK_SCHED_STATUS__INACTIVE; pTask->inputQueue = streamQueueOpen(); @@ -975,13 +976,15 @@ int32_t tqProcessStreamTaskCheckReq(STQ* pTq, SRpcMsg* pMsg) { .upstreamNodeId = req.upstreamNodeId, .upstreamTaskId = req.upstreamTaskId, }; - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, taskId); if (pTask && atomic_load_8(&pTask->taskStatus) == TASK_STATUS__NORMAL) { rsp.status = 1; } else { rsp.status = 0; } + streamMetaReleaseTask(pTq->pStreamMeta, pTask); + tqDebug("tq recv task check req(reqId: %" PRId64 ") %d at node %d check req from task %d at node %d, status %d", rsp.reqId, rsp.downstreamTaskId, rsp.downstreamNodeId, rsp.upstreamTaskId, rsp.upstreamNodeId, rsp.status); @@ -1027,12 +1030,14 @@ int32_t tqProcessStreamTaskCheckRsp(STQ* pTq, int64_t version, char* msg, int32_ tqDebug("tq recv task check rsp(reqId: %" PRId64 ") %d at node %d check req from task %d at node %d, status %d", rsp.reqId, rsp.downstreamTaskId, rsp.downstreamNodeId, rsp.upstreamTaskId, rsp.upstreamNodeId, rsp.status); - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, rsp.upstreamTaskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, rsp.upstreamTaskId); if (pTask == NULL) { return -1; } - return streamProcessTaskCheckRsp(pTask, &rsp, version); + code = streamProcessTaskCheckRsp(pTask, &rsp, version); + streamMetaReleaseTask(pTq->pStreamMeta, pTask); + return code; } int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) { @@ -1077,15 +1082,17 @@ int32_t tqProcessTaskRecover1Req(STQ* pTq, SRpcMsg* pMsg) { int32_t msgLen = pMsg->contLen; SStreamRecoverStep1Req* pReq = (SStreamRecoverStep1Req*)msg; - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, pReq->taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, pReq->taskId); if (pTask == NULL) { return -1; } + ASSERT(pReq->taskId == pTask->taskId); // check param int64_t fillVer1 = pTask->startVer; if (fillVer1 <= 0) { ASSERT(0); + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return -1; } @@ -1096,10 +1103,11 @@ int32_t tqProcessTaskRecover1Req(STQ* pTq, SRpcMsg* pMsg) { SStreamRecoverStep2Req req; code = streamBuildSourceRecover2Req(pTask, &req); if (code < 0) { + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return -1; } - ASSERT(pReq->taskId == pTask->taskId); + streamMetaReleaseTask(pTq->pStreamMeta, pTask); // serialize msg int32_t len = sizeof(SStreamRecoverStep1Req); @@ -1127,7 +1135,7 @@ int32_t tqProcessTaskRecover1Req(STQ* pTq, SRpcMsg* pMsg) { int32_t tqProcessTaskRecover2Req(STQ* pTq, int64_t version, char* msg, int32_t msgLen) { int32_t code; SStreamRecoverStep2Req* pReq = (SStreamRecoverStep2Req*)msg; - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, pReq->taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, pReq->taskId); if (pTask == NULL) { return -1; } @@ -1135,27 +1143,33 @@ int32_t tqProcessTaskRecover2Req(STQ* pTq, int64_t version, char* msg, int32_t m // do recovery step 2 code = streamSourceRecoverScanStep2(pTask, version); if (code < 0) { + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return -1; } // restore param code = streamRestoreParam(pTask); if (code < 0) { + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return -1; } // set status normal code = streamSetStatusNormal(pTask); if (code < 0) { + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return -1; } // dispatch recover finish req to all related downstream task code = streamDispatchRecoverFinishReq(pTask); if (code < 0) { + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return -1; } + streamMetaReleaseTask(pTq->pStreamMeta, pTask); + return 0; } @@ -1172,15 +1186,17 @@ int32_t tqProcessTaskRecoverFinishReq(STQ* pTq, SRpcMsg* pMsg) { tDecoderClear(&decoder); // find task - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, req.taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, req.taskId); if (pTask == NULL) { return -1; } // do process request if (streamProcessRecoverFinishReq(pTask, req.childId) < 0) { + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return -1; } + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return 0; } @@ -1354,9 +1370,10 @@ int32_t tqProcessSubmitReq(STQ* pTq, SSubmitReq* pReq, int64_t ver) { int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) { SStreamTaskRunReq* pReq = pMsg->pCont; int32_t taskId = pReq->taskId; - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, taskId); if (pTask) { streamProcessRunReq(pTask); + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return 0; } else { return -1; @@ -1373,13 +1390,14 @@ int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec) { tDecodeStreamDispatchReq(&decoder, &req); int32_t taskId = req.taskId; - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, taskId); if (pTask) { SRpcMsg rsp = { .info = pMsg->info, .code = 0, }; streamProcessDispatchReq(pTask, &req, &rsp, exec); + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return 0; } else { return -1; @@ -1389,10 +1407,11 @@ int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec) { int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) { SStreamDispatchRsp* pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); int32_t taskId = ntohl(pRsp->upstreamTaskId); - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, taskId); tqDebug("recv dispatch rsp, code: %x", pMsg->code); if (pTask) { streamProcessDispatchRsp(pTask, pRsp, pMsg->code); + streamMetaReleaseTask(pTq->pStreamMeta, pTask); return 0; } else { return -1; @@ -1401,7 +1420,8 @@ int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) { int32_t tqProcessTaskDropReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) { SVDropStreamTaskReq* pReq = (SVDropStreamTaskReq*)msg; - return streamMetaRemoveTask(pTq->pStreamMeta, pReq->taskId); + streamMetaRemoveTask1(pTq->pStreamMeta, pReq->taskId); + return 0; } int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) { @@ -1414,13 +1434,14 @@ int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) { tDecodeStreamRetrieveReq(&decoder, &req); tDecoderClear(&decoder); int32_t taskId = req.dstTaskId; - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, taskId); if (pTask) { SRpcMsg rsp = { .info = pMsg->info, .code = 0, }; streamProcessRetrieveReq(pTask, &req, &rsp); + streamMetaReleaseTask(pTq->pStreamMeta, pTask); tDeleteStreamRetrieveReq(&req); return 0; } else { @@ -1452,13 +1473,14 @@ int32_t vnodeEnqueueStreamMsg(SVnode* pVnode, SRpcMsg* pMsg) { int32_t taskId = req.taskId; - SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId); + SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, taskId); if (pTask) { SRpcMsg rsp = { .info = pMsg->info, .code = 0, }; streamProcessDispatchReq(pTask, &req, &rsp, false); + streamMetaReleaseTask(pTq->pStreamMeta, pTask); rpcFreeCont(pMsg->pCont); taosFreeQitem(pMsg); return 0; diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index d2e59ea8e1..77d2aeba28 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -2385,6 +2385,8 @@ static void destroyStreamScanOperatorInfo(void* param) { taosMemoryFree(pStreamScan->pPseudoExpr); } + cleanupExprSupp(&pStreamScan->tbnameCalSup); + updateInfoDestroy(pStreamScan->pUpdateInfo); blockDataDestroy(pStreamScan->pRes); blockDataDestroy(pStreamScan->pUpdateRes); diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c index da082b7f74..5ec8828c05 100644 --- a/source/libs/stream/src/streamMeta.c +++ b/source/libs/stream/src/streamMeta.c @@ -169,6 +169,47 @@ SStreamTask* streamMetaGetTask(SStreamMeta* pMeta, int32_t taskId) { } } +SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int32_t taskId) { + taosRLockLatch(&pMeta->lock); + + SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, &taskId, sizeof(int32_t)); + if (ppTask) { + SStreamTask* pTask = *ppTask; + if (atomic_load_8(&pTask->taskStatus) != TASK_STATUS__DROPPING) { + atomic_add_fetch_32(&pTask->refCnt, 1); + taosRUnLockLatch(&pMeta->lock); + return pTask; + } else { + taosRUnLockLatch(&pMeta->lock); + return NULL; + } + } + taosRUnLockLatch(&pMeta->lock); + return NULL; +} + +void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask) { + int32_t left = atomic_sub_fetch_32(&pTask->refCnt, 1); + ASSERT(left >= 0); + if (left == 0) { + ASSERT(atomic_load_8(&pTask->taskStatus) == TASK_STATUS__DROPPING); + tFreeSStreamTask(pTask); + } +} + +void streamMetaRemoveTask1(SStreamMeta* pMeta, int32_t taskId) { + SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, &taskId, sizeof(int32_t)); + if (ppTask) { + SStreamTask* pTask = *ppTask; + taosHashRemove(pMeta->pTasks, &taskId, sizeof(int32_t)); + atomic_store_8(&pTask->taskStatus, TASK_STATUS__DROPPING); + + taosWLockLatch(&pMeta->lock); + streamMetaReleaseTask(pMeta, pTask); + taosWUnLockLatch(&pMeta->lock); + } +} + int32_t streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId) { SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, &taskId, sizeof(int32_t)); if (ppTask) { diff --git a/source/libs/wal/src/walRef.c b/source/libs/wal/src/walRef.c index f9994fd315..eae5d9f1a7 100644 --- a/source/libs/wal/src/walRef.c +++ b/source/libs/wal/src/walRef.c @@ -32,7 +32,7 @@ SWalRef *walOpenRef(SWal *pWal) { return pRef; } -#if 0 +#if 1 void walCloseRef(SWal *pWal, int64_t refId) { SWalRef **ppRef = taosHashGet(pWal->pRefHash, &refId, sizeof(int64_t)); if (ppRef == NULL) return; @@ -67,7 +67,7 @@ int32_t walRefVer(SWalRef *pRef, int64_t ver) { return 0; } -#if 0 +#if 1 void walUnrefVer(SWalRef *pRef) { pRef->refId = -1; pRef->refFile = -1; From c5aa7e072dd8780c9d8d34d08fac44dc1a1944b3 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Tue, 15 Nov 2022 19:17:28 +0800 Subject: [PATCH 3/5] refactor(sync): optimize, make LRU Cache hit more --- source/libs/sync/inc/syncInt.h | 1 + source/libs/sync/src/syncAppendEntries.c | 100 ++++++++++++++++++++--- source/libs/sync/src/syncCommit.c | 5 ++ source/libs/sync/src/syncMain.c | 81 ++++++++++++------ source/libs/sync/src/syncRaftEntry.c | 2 + source/libs/sync/src/syncRaftLog.c | 14 +++- source/libs/sync/src/syncReplication.c | 21 ++++- source/libs/sync/src/syncUtil.c | 36 ++++---- 8 files changed, 206 insertions(+), 54 deletions(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index e81f63d871..0e0042d526 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -225,6 +225,7 @@ int32_t syncNodeRestartHeartbeatTimer(SSyncNode* pSyncNode); int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg); int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg); SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode); +int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHandle** h); // raft state change -------------- void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term); diff --git a/source/libs/sync/src/syncAppendEntries.c b/source/libs/sync/src/syncAppendEntries.c index 4551702b34..df2ccee546 100644 --- a/source/libs/sync/src/syncAppendEntries.c +++ b/source/libs/sync/src/syncAppendEntries.c @@ -192,13 +192,34 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) { SSyncRaftEntry* pAppendEntry = syncEntryBuildFromAppendEntries(pMsg); ASSERT(pAppendEntry != NULL); - SyncIndex appendIndex = pMsg->prevLogIndex + 1; + SyncIndex appendIndex = pMsg->prevLogIndex + 1; + + LRUHandle* hLocal = NULL; + LRUHandle* hAppend = NULL; + + int32_t code = 0; SSyncRaftEntry* pLocalEntry = NULL; - int32_t code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, appendIndex, &pLocalEntry); + SLRUCache* pCache = ths->pLogStore->pCache; + hLocal = taosLRUCacheLookup(pCache, &appendIndex, sizeof(appendIndex)); + if (hLocal) { + pLocalEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, hLocal); + code = 0; + + sNTrace(ths, "hit cache index:%" PRId64 ", bytes:%u, %p", appendIndex, pLocalEntry->bytes, pLocalEntry); + + } else { + sNTrace(ths, "miss cache index:%" PRId64, appendIndex); + + code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, appendIndex, &pLocalEntry); + } + if (code == 0) { + // get local entry success + if (pLocalEntry->term == pAppendEntry->term) { // do nothing sNTrace(ths, "log match, do nothing, index:%" PRId64, appendIndex); + } else { // truncate code = ths->pLogStore->syncLogTruncate(ths->pLogStore, appendIndex); @@ -207,8 +228,18 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) { snprintf(logBuf, sizeof(logBuf), "ignore, truncate error, append-index:%" PRId64, appendIndex); syncLogRecvAppendEntries(ths, pMsg, logBuf); - syncEntryDestory(pLocalEntry); - syncEntryDestory(pAppendEntry); + if (hLocal) { + taosLRUCacheRelease(ths->pLogStore->pCache, hLocal, false); + } else { + syncEntryDestory(pLocalEntry); + } + + if (hAppend) { + taosLRUCacheRelease(ths->pLogStore->pCache, hAppend, false); + } else { + syncEntryDestory(pAppendEntry); + } + goto _IGNORE; } @@ -219,10 +250,22 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) { snprintf(logBuf, sizeof(logBuf), "ignore, append error, append-index:%" PRId64, appendIndex); syncLogRecvAppendEntries(ths, pMsg, logBuf); - syncEntryDestory(pLocalEntry); - syncEntryDestory(pAppendEntry); + if (hLocal) { + taosLRUCacheRelease(ths->pLogStore->pCache, hLocal, false); + } else { + syncEntryDestory(pLocalEntry); + } + + if (hAppend) { + taosLRUCacheRelease(ths->pLogStore->pCache, hAppend, false); + } else { + syncEntryDestory(pAppendEntry); + } + goto _IGNORE; } + + syncCacheEntry(ths->pLogStore, pAppendEntry, &hAppend); } } else { @@ -248,20 +291,42 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) { snprintf(logBuf, sizeof(logBuf), "ignore, log not exist, append error, append-index:%" PRId64, appendIndex); syncLogRecvAppendEntries(ths, pMsg, logBuf); - syncEntryDestory(pLocalEntry); - syncEntryDestory(pAppendEntry); + if (hLocal) { + taosLRUCacheRelease(ths->pLogStore->pCache, hLocal, false); + } else { + syncEntryDestory(pLocalEntry); + } + + if (hAppend) { + taosLRUCacheRelease(ths->pLogStore->pCache, hAppend, false); + } else { + syncEntryDestory(pAppendEntry); + } + goto _IGNORE; } + syncCacheEntry(ths->pLogStore, pAppendEntry, &hAppend); + } else { - // error + // get local entry success char logBuf[128]; snprintf(logBuf, sizeof(logBuf), "ignore, get local entry error, append-index:%" PRId64 " err:%d", appendIndex, terrno); syncLogRecvAppendEntries(ths, pMsg, logBuf); - syncEntryDestory(pLocalEntry); - syncEntryDestory(pAppendEntry); + if (hLocal) { + taosLRUCacheRelease(ths->pLogStore->pCache, hLocal, false); + } else { + syncEntryDestory(pLocalEntry); + } + + if (hAppend) { + taosLRUCacheRelease(ths->pLogStore->pCache, hAppend, false); + } else { + syncEntryDestory(pAppendEntry); + } + goto _IGNORE; } } @@ -269,8 +334,17 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) { // update match index pReply->matchIndex = pAppendEntry->index; - syncEntryDestory(pLocalEntry); - syncEntryDestory(pAppendEntry); + if (hLocal) { + taosLRUCacheRelease(ths->pLogStore->pCache, hLocal, false); + } else { + syncEntryDestory(pLocalEntry); + } + + if (hAppend) { + taosLRUCacheRelease(ths->pLogStore->pCache, hAppend, false); + } else { + syncEntryDestory(pAppendEntry); + } } else { // no append entries, do nothing diff --git a/source/libs/sync/src/syncCommit.c b/source/libs/sync/src/syncCommit.c index d2320fc6be..60bec6ab65 100644 --- a/source/libs/sync/src/syncCommit.c +++ b/source/libs/sync/src/syncCommit.c @@ -116,7 +116,12 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) { LRUHandle* h = taosLRUCacheLookup(pCache, &index, sizeof(index)); if (h) { pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h); + + sNTrace(pSyncNode, "hit cache index:%" PRId64 ", bytes:%u, %p", index, pEntry->bytes, pEntry); + } else { + sNTrace(pSyncNode, "miss cache index:%" PRId64, index); + int32_t code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, index, &pEntry); if (code != 0) { sNError(pSyncNode, "advance commit index error, read wal index:%" PRId64, index); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index e802f60f30..a59078fff6 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -383,15 +383,33 @@ bool syncIsReadyForRead(int64_t rid) { } else { if (!pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore)) { + SyncIndex lastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore); SSyncRaftEntry* pEntry = NULL; - int32_t code = pSyncNode->pLogStore->syncLogGetEntry( - pSyncNode->pLogStore, pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore), &pEntry); + SLRUCache* pCache = pSyncNode->pLogStore->pCache; + LRUHandle* h = taosLRUCacheLookup(pCache, &lastIndex, sizeof(lastIndex)); + int32_t code = 0; + if (h) { + pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h); + code = 0; + + sNTrace(pSyncNode, "hit cache index:%" PRId64 ", bytes:%u, %p", lastIndex, pEntry->bytes, pEntry); + + } else { + sNTrace(pSyncNode, "miss cache index:%" PRId64, lastIndex); + + code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, lastIndex, &pEntry); + } + if (code == 0 && pEntry != NULL) { if (pEntry->originalRpcType == TDMT_SYNC_NOOP && pEntry->term == pSyncNode->pRaftStore->currentTerm) { ready = true; } - syncEntryDestory(pEntry); + if (h) { + taosLRUCacheRelease(pCache, h, false); + } else { + syncEntryDestory(pEntry); + } } } } @@ -1761,10 +1779,24 @@ SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) { return 0; } - SyncTerm preTerm = 0; - SyncIndex preIndex = index - 1; + SyncTerm preTerm = 0; + SyncIndex preIndex = index - 1; + SSyncRaftEntry* pPreEntry = NULL; - int32_t code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, preIndex, &pPreEntry); + SLRUCache* pCache = pSyncNode->pLogStore->pCache; + LRUHandle* h = taosLRUCacheLookup(pCache, &preIndex, sizeof(preIndex)); + int32_t code = 0; + if (h) { + pPreEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h); + code = 0; + + sNTrace(pSyncNode, "hit cache index:%" PRId64 ", bytes:%u, %p", preIndex, pPreEntry->bytes, pPreEntry); + + } else { + sNTrace(pSyncNode, "miss cache index:%" PRId64, preIndex); + + code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, preIndex, &pPreEntry); + } SSnapshot snapshot = {.data = NULL, .lastApplyIndex = SYNC_INDEX_INVALID, @@ -1774,7 +1806,13 @@ SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) { if (code == 0) { ASSERT(pPreEntry != NULL); preTerm = pPreEntry->term; - taosMemoryFree(pPreEntry); + + if (h) { + taosLRUCacheRelease(pCache, h, false); + } else { + syncEntryDestory(pPreEntry); + } + return preTerm; } else { if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) { @@ -1820,9 +1858,6 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) { } taosTmrReset(syncNodeEqPingTimer, pNode->pingTimerMS, pNode, syncEnv()->pTimerManager, &pNode->pPingTimer); - } else { - sTrace("==syncNodeEqPingTimer== pingTimerLogicClock:%" PRId64 ", pingTimerLogicClockUser:%" PRId64, - pNode->pingTimerLogicClock, pNode->pingTimerLogicClockUser); } } @@ -1856,16 +1891,6 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) { } taosMemoryFree(pElectTimer); - -#if 0 - // reset timer ms - if (syncIsInit() && pNode->electBaseLine > 0) { - pNode->electTimerMS = syncUtilElectRandomMS(pNode->electBaseLine, 2 * pNode->electBaseLine); - taosTmrReset(syncNodeEqElectTimer, pNode->electTimerMS, pNode, syncEnv()->pTimerManager, &pNode->pElectTimer); - } else { - sError("sync env is stop, syncNodeEqElectTimer"); - } -#endif } static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { @@ -1979,7 +2004,10 @@ static int32_t syncNodeEqNoop(SSyncNode* pNode) { static void deleteCacheEntry(const void* key, size_t keyLen, void* value) { taosMemoryFree(value); } -static int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHandle** h) { +int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHandle** h) { + SSyncLogStoreData* pData = pLogStore->data; + sNTrace(pData->pSyncNode, "in cache index:%" PRId64 ", bytes:%u, %p", pEntry->index, pEntry->bytes, pEntry); + int32_t code = 0; int32_t entryLen = sizeof(*pEntry) + pEntry->dataLen; LRUStatus status = taosLRUCacheInsert(pLogStore->pCache, &pEntry->index, sizeof(pEntry->index), pEntry, entryLen, @@ -2000,7 +2028,6 @@ static int32_t syncNodeAppendNoop(SSyncNode* ths) { ASSERT(pEntry != NULL); LRUHandle* h = NULL; - syncCacheEntry(ths->pLogStore, pEntry, &h); if (ths->state == TAOS_SYNC_STATE_LEADER) { int32_t code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry); @@ -2008,6 +2035,8 @@ static int32_t syncNodeAppendNoop(SSyncNode* ths) { sError("append noop error"); return -1; } + + syncCacheEntry(ths->pLogStore, pEntry, &h); } if (h) { @@ -2143,7 +2172,6 @@ int32_t syncNodeOnClientRequest(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIn } LRUHandle* h = NULL; - syncCacheEntry(ths->pLogStore, pEntry, &h); if (ths->state == TAOS_SYNC_STATE_LEADER) { // append entry @@ -2183,6 +2211,8 @@ int32_t syncNodeOnClientRequest(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIn } } + syncCacheEntry(ths->pLogStore, pEntry, &h); + // if mulit replica, start replicate right now if (ths->replicaNum > 1) { syncNodeReplicate(ths); @@ -2349,7 +2379,12 @@ int32_t syncNodeDoCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endInde LRUHandle* h = taosLRUCacheLookup(pCache, &i, sizeof(i)); if (h) { pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h); + + sNTrace(ths, "hit cache index:%" PRId64 ", bytes:%u, %p", i, pEntry->bytes, pEntry); + } else { + sNTrace(ths, "miss cache index:%" PRId64, i); + code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, i, &pEntry); // ASSERT(code == 0); // ASSERT(pEntry != NULL); diff --git a/source/libs/sync/src/syncRaftEntry.c b/source/libs/sync/src/syncRaftEntry.c index 540054e473..a759409163 100644 --- a/source/libs/sync/src/syncRaftEntry.c +++ b/source/libs/sync/src/syncRaftEntry.c @@ -92,6 +92,8 @@ SSyncRaftEntry* syncEntryBuildNoop(SyncTerm term, SyncIndex index, int32_t vgId) void syncEntryDestory(SSyncRaftEntry* pEntry) { if (pEntry != NULL) { taosMemoryFree(pEntry); + + sTrace("free entry: %p", pEntry); } } diff --git a/source/libs/sync/src/syncRaftLog.c b/source/libs/sync/src/syncRaftLog.c index b00ba3918c..2b1cee51d7 100644 --- a/source/libs/sync/src/syncRaftLog.c +++ b/source/libs/sync/src/syncRaftLog.c @@ -37,7 +37,8 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) { return NULL; } - pLogStore->pCache = taosLRUCacheInit(10 * 1024 * 1024, 1, .5); + // pLogStore->pCache = taosLRUCacheInit(10 * 1024 * 1024, 1, .5); + pLogStore->pCache = taosLRUCacheInit(100 * 1024 * 1024, 1, .5); if (pLogStore->pCache == NULL) { taosMemoryFree(pLogStore); terrno = TSDB_CODE_WAL_OUT_OF_MEMORY; @@ -321,6 +322,17 @@ static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIn return 0; } + // delete from cache + for (SyncIndex index = fromIndex; index <= wallastVer; ++index) { + SLRUCache* pCache = pData->pSyncNode->pLogStore->pCache; + LRUHandle* h = taosLRUCacheLookup(pCache, &index, sizeof(index)); + if (h) { + sNTrace(pData->pSyncNode, "cache delete index:%" PRId64, index); + + taosLRUCacheRelease(pData->pSyncNode->pLogStore->pCache, h, true); + } + } + int32_t code = walRollback(pWal, fromIndex); if (code != 0) { int32_t err = terrno; diff --git a/source/libs/sync/src/syncReplication.c b/source/libs/sync/src/syncReplication.c index 6a7a2c18c1..802595c55a 100644 --- a/source/libs/sync/src/syncReplication.c +++ b/source/libs/sync/src/syncReplication.c @@ -73,7 +73,20 @@ int32_t syncNodeReplicateOne(SSyncNode* pSyncNode, SRaftId* pDestId, bool snapsh SyncAppendEntries* pMsg = NULL; SSyncRaftEntry* pEntry = NULL; - int32_t code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, nextIndex, &pEntry); + SLRUCache* pCache = pSyncNode->pLogStore->pCache; + LRUHandle* h = taosLRUCacheLookup(pCache, &nextIndex, sizeof(nextIndex)); + int32_t code = 0; + if (h) { + pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h); + code = 0; + + sNTrace(pSyncNode, "hit cache index:%" PRId64 ", bytes:%u, %p", nextIndex, pEntry->bytes, pEntry); + + } else { + sNTrace(pSyncNode, "miss cache index:%" PRId64, nextIndex); + + code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, nextIndex, &pEntry); + } if (code == 0) { ASSERT(pEntry != NULL); @@ -99,7 +112,11 @@ int32_t syncNodeReplicateOne(SSyncNode* pSyncNode, SRaftId* pDestId, bool snapsh } } - syncEntryDestory(pEntry); + if (h) { + taosLRUCacheRelease(pCache, h, false); + } else { + syncEntryDestory(pEntry); + } // prepare msg ASSERT(pMsg != NULL); diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index b9a271ab9d..fb1b07b0b6 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -212,7 +212,11 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo } char cfgStr[1024]; - syncCfg2SimpleStr(&(pNode->pRaftCfg->cfg), cfgStr, sizeof(cfgStr)); + if (pNode->pRaftCfg != NULL) { + syncCfg2SimpleStr(&(pNode->pRaftCfg->cfg), cfgStr, sizeof(cfgStr)); + } else { + return; + } char peerStr[1024] = "{"; syncPeerState2Str(pNode, peerStr, sizeof(peerStr)); @@ -230,17 +234,19 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo // restore error code terrno = errCode; - taosPrintLog(flags, level, dflag, - "vgId:%d, sync %s " - "%s" - ", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64 - ", snap-tm:%" PRIu64 ", sby:%d, aq:%d, bch:%d, r-num:%d, lcfg:%" PRId64 - ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s", - pNode->vgId, syncStr(pNode->state), eventLog, currentTerm, pNode->commitIndex, logBeginIndex, - logLastIndex, pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm, - pNode->pRaftCfg->isStandBy, aqItems, pNode->pRaftCfg->batchSize, pNode->replicaNum, - pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum, - pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, peerStr, cfgStr); + if (pNode != NULL && pNode->pRaftCfg != NULL) { + taosPrintLog(flags, level, dflag, + "vgId:%d, sync %s " + "%s" + ", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64 + ", snap-tm:%" PRIu64 ", sby:%d, aq:%d, bch:%d, r-num:%d, lcfg:%" PRId64 + ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s", + pNode->vgId, syncStr(pNode->state), eventLog, currentTerm, pNode->commitIndex, logBeginIndex, + logLastIndex, pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm, + pNode->pRaftCfg->isStandBy, aqItems, pNode->pRaftCfg->batchSize, pNode->replicaNum, + pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum, + pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, peerStr, cfgStr); + } } void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotSender* pSender, @@ -364,9 +370,9 @@ void syncLogSendAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntries syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port); sNTrace(pSyncNode, - "send sync-append-entries-reply to %s:%d, {term:%" PRId64 ", pterm:%" PRId64 ", success:%d, match:%" PRId64 - "}, %s", - host, port, pMsg->term, pMsg->privateTerm, pMsg->success, pMsg->matchIndex, s); + "send sync-append-entries-reply to %s:%d, {term:%" PRId64 ", pterm:%" PRId64 + ", success:%d, lsend-index:%" PRId64 ", match:%" PRId64 "}, %s", + host, port, pMsg->term, pMsg->privateTerm, pMsg->success, pMsg->lastSendIndex, pMsg->matchIndex, s); } void syncLogRecvAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s) { From 20e31aacca9c3e5ca2d93c90853fa3524191efc4 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 15 Nov 2022 19:48:16 +0800 Subject: [PATCH 4/5] fix(wal): set commit ver and snapshot ver after restore --- source/libs/wal/src/walWrite.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 216dd5fcb1..1a75d5ed81 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -70,8 +70,8 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) { taosArrayClear(pWal->fileInfoSet); pWal->vers.firstVer = -1; pWal->vers.lastVer = ver; - pWal->vers.commitVer = ver - 1; - pWal->vers.snapshotVer = ver - 1; + pWal->vers.commitVer = ver; + pWal->vers.snapshotVer = ver; pWal->vers.verInSnapshotting = -1; taosThreadMutexUnlock(&pWal->mutex); From d5080b5ce631d38b47c666948a1e347df53ad44e Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 15 Nov 2022 19:55:40 +0800 Subject: [PATCH 5/5] test: add asan case --- tests/parallel_test/cases.task | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index bfd93994eb..e4edc43600 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -228,7 +228,7 @@ ,,y,script,./test.sh -f tsim/stream/partitionby.sim ,,y,script,./test.sh -f tsim/stream/partitionby1.sim ,,y,script,./test.sh -f tsim/stream/schedSnode.sim -,,y,script,./test.sh -f tsim/stream/windowClose.sim +,,,script,./test.sh -f tsim/stream/windowClose.sim ,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim ,,y,script,./test.sh -f tsim/stream/sliding.sim ,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim