From a3133b9f6daabe1f06becfbf84d1ad1c00192b83 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Wed, 20 Jul 2022 10:55:28 +0800 Subject: [PATCH] feat(wal): log applied ver --- include/libs/wal/wal.h | 6 +++++- source/dnode/mnode/impl/src/mndConsumer.c | 4 ++++ source/dnode/vnode/src/tq/tqPush.c | 3 ++- source/dnode/vnode/src/vnd/vnodeSvr.c | 2 ++ source/libs/wal/src/walMeta.c | 2 ++ source/libs/wal/src/walRead.c | 10 ++++++++-- source/libs/wal/src/walWrite.c | 6 ++++++ tests/system-test/7-tmq/tmqAutoCreateTbl.py | 2 +- 8 files changed, 30 insertions(+), 5 deletions(-) diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 7e2d09dd63..00a36391fa 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -64,6 +64,7 @@ typedef struct { int64_t verInSnapshotting; int64_t snapshotVer; int64_t commitVer; + int64_t appliedVer; int64_t lastVer; } SWalVer; @@ -172,6 +173,9 @@ int32_t walRollback(SWal *, int64_t ver); int32_t walBeginSnapshot(SWal *, int64_t ver); int32_t walEndSnapshot(SWal *); int32_t walRestoreFromSnapshot(SWal *, int64_t ver); +// for tq +int32_t walApplyVer(SWal *, int64_t ver); + // int32_t walDataCorrupted(SWal*); // read @@ -186,7 +190,6 @@ void walSetReaderCapacity(SWalReader *pRead, int32_t capacity); int32_t walFetchHead(SWalReader *pRead, int64_t ver, SWalCkHead *pHead); int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead); int32_t walSkipFetchBody(SWalReader *pRead, const SWalCkHead *pHead); - typedef struct { int64_t refId; int64_t ver; @@ -206,6 +209,7 @@ int64_t walGetFirstVer(SWal *); int64_t walGetSnapshotVer(SWal *); int64_t walGetLastVer(SWal *); int64_t walGetCommittedVer(SWal *); +int64_t walGetAppliedVer(SWal *); #ifdef __cplusplus } diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 315b7c3afc..f7387f7e88 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -265,6 +265,10 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) { int64_t consumerId = be64toh(pReq->consumerId); SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId); + if (pConsumer == NULL) { + terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST; + return -1; + } atomic_store_32(&pConsumer->hbStatus, 0); diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c index c929c84203..4c0d416ad1 100644 --- a/source/dnode/vnode/src/tq/tqPush.c +++ b/source/dnode/vnode/src/tq/tqPush.c @@ -237,6 +237,8 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_ #endif int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) { + walApplyVer(pTq->pVnode->pWal, ver); + if (msgType == TDMT_VND_SUBMIT) { if (taosHashGetSize(pTq->pStreamTasks) == 0) return 0; @@ -253,4 +255,3 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) return 0; } - diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 8e59d97286..e6d116dfef 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -878,6 +878,8 @@ _exit: tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_INPUT__DATA_SUBMIT); } + vDebug("successful submit in vg %d version %ld", pVnode->config.vgId, version); + return 0; } diff --git a/source/libs/wal/src/walMeta.c b/source/libs/wal/src/walMeta.c index 991b50f7c0..edc811fe82 100644 --- a/source/libs/wal/src/walMeta.c +++ b/source/libs/wal/src/walMeta.c @@ -33,6 +33,8 @@ int64_t FORCE_INLINE walGetLastVer(SWal* pWal) { return pWal->vers.lastVer; } int64_t FORCE_INLINE walGetCommittedVer(SWal* pWal) { return pWal->vers.commitVer; } +int64_t FORCE_INLINE walGetAppliedVer(SWal* pWal) { return pWal->vers.appliedVer; } + static FORCE_INLINE int walBuildMetaName(SWal* pWal, int metaVer, char* buf) { return sprintf(buf, "%s/meta-ver%d", pWal->path, metaVer); } diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c index 8b4225c80c..5bc9cdafa2 100644 --- a/source/libs/wal/src/walRead.c +++ b/source/libs/wal/src/walRead.c @@ -66,9 +66,15 @@ void walCloseReader(SWalReader *pRead) { } int32_t walNextValidMsg(SWalReader *pRead) { - wDebug("vgId:%d wal start to fetch", pRead->pWal->cfg.vgId); int64_t fetchVer = pRead->curVersion; - int64_t endVer = pRead->cond.scanUncommited ? walGetLastVer(pRead->pWal) : walGetCommittedVer(pRead->pWal); + int64_t lastVer = walGetLastVer(pRead->pWal); + int64_t committedVer = walGetCommittedVer(pRead->pWal); + int64_t appliedVer = walGetAppliedVer(pRead->pWal); + int64_t endVer = pRead->cond.scanUncommited ? lastVer : committedVer; + endVer = TMIN(appliedVer, endVer); + + wDebug("vgId:%d wal start to fetch, ver %ld, last ver %ld commit ver %ld, applied ver %ld, end ver %ld", + pRead->pWal->cfg.vgId, fetchVer, lastVer, committedVer, appliedVer, endVer); while (fetchVer <= endVer) { if (walFetchHeadNew(pRead, fetchVer) < 0) { return -1; diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 26dc3cdffb..4fc135a1cf 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -64,6 +64,12 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) { return 0; } +int32_t walApplyVer(SWal *pWal, int64_t ver) { + // TODO: error check + pWal->vers.appliedVer = ver; + return 0; +} + int32_t walCommit(SWal *pWal, int64_t ver) { ASSERT(pWal->vers.commitVer >= pWal->vers.snapshotVer); ASSERT(pWal->vers.commitVer <= pWal->vers.lastVer); diff --git a/tests/system-test/7-tmq/tmqAutoCreateTbl.py b/tests/system-test/7-tmq/tmqAutoCreateTbl.py index 1622ad7621..8fcb57aea6 100644 --- a/tests/system-test/7-tmq/tmqAutoCreateTbl.py +++ b/tests/system-test/7-tmq/tmqAutoCreateTbl.py @@ -225,7 +225,7 @@ class TDTestCase: tdSql.prepare() self.prepareTestEnv() self.tmqCase1() - # self.tmqCase2() TD-17267 + # self.tmqCase2() # TD-17267 def stop(self): -- GitLab