From 934d0f9784b3cab31f669aecc64bc54a2393f1e6 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 20 Jun 2022 15:23:48 +0800 Subject: [PATCH 1/4] fix:error in windows --- include/common/tmsgdef.h | 1 + source/dnode/mnode/impl/src/mndMain.c | 57 +++++++++----------------- source/dnode/mnode/impl/src/mndTrans.c | 31 ++++++++++++++ 3 files changed, 51 insertions(+), 38 deletions(-) diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index 1b640642d7..d35c5475de 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -144,6 +144,7 @@ enum { TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mq-tmr", SMTimerReq, NULL) TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "telem-tmr", SMTimerReq, SMTimerReq) TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "trans-tmr", NULL, NULL) + TD_DEF_MSG_TYPE(TDMT_MND_TTL_TIMER, "ttl-tmr", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_KILL_TRANS, "kill-trans", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "kill-query", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "kill-conn", NULL, NULL) diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c index 95b721b4dd..7a103438dc 100644 --- a/source/dnode/mnode/impl/src/mndMain.c +++ b/source/dnode/mnode/impl/src/mndMain.c @@ -63,6 +63,21 @@ static void mndPullupTrans(SMnode *pMnode) { tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg); } +static void mndTtlTimer(SMnode *pMnode) { + int32_t contLen = sizeof(SMsgHead) + sizeof(int32_t); + SMsgHead *pHead = rpcMallocCont(contLen); + if (pHead == NULL) { + mError("ttl time malloc err. contLen:%d", contLen); + return; + } + + int32_t t = taosGetTimestampSec(); + *(int32_t*)(POINTER_SHIFT(pHead, sizeof(SMsgHead))) = htonl(t); + + SRpcMsg rpcMsg = {.msgType = TDMT_MND_TTL_TIMER, .pCont = pHead, .contLen = contLen}; + tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg); +} + static void mndCalMqRebalance(SMnode *pMnode) { int32_t contLen = 0; void * pReq = mndBuildTimerMsg(&contLen); @@ -77,54 +92,20 @@ static void mndPullupTelem(SMnode *pMnode) { tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg); } -static void mndPushTtlTime(SMnode *pMnode) { - SSdb *pSdb = pMnode->pSdb; - SVgObj *pVgroup = NULL; - void *pIter = NULL; - - while (1) { - pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); - if (pIter == NULL) break; - - int32_t contLen = sizeof(SMsgHead) + sizeof(int32_t); - SMsgHead *pHead = rpcMallocCont(contLen); - if (pHead == NULL) { - mError("ttl time malloc err. contLen:%d", contLen); - sdbRelease(pSdb, pVgroup); - continue; - } - pHead->contLen = htonl(contLen); - pHead->vgId = htonl(pVgroup->vgId); - - int32_t t = taosGetTimestampSec(); - *(int32_t*)(POINTER_SHIFT(pHead, sizeof(SMsgHead))) = htonl(t); - - SRpcMsg rpcMsg = {.msgType = TDMT_VND_DROP_TTL_TABLE, .pCont = pHead, .contLen = contLen}; - - SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup); - int32_t code = tmsgSendReq(&epSet, &rpcMsg); - if(code != 0){ - mError("ttl time seed err. code:%d", code); - } - mError("ttl time seed succ. time:%d", t); - sdbRelease(pSdb, pVgroup); - } -} - static void *mndThreadFp(void *param) { SMnode *pMnode = param; int64_t lastTime = 0; setThreadName("mnode-timer"); while (1) { - if (lastTime % (864000) == 0) { // sleep 1 day for ttl - mndPushTtlTime(pMnode); - } - lastTime++; taosMsleep(100); if (mndGetStop(pMnode)) break; + if (lastTime % (864000) == 1) { // sleep 1 day for ttl + mndTtlTimer(pMnode); + } + if (lastTime % (tsTransPullupInterval * 10) == 0) { mndPullupTrans(pMnode); } diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c index 31a955b030..07450613c6 100644 --- a/source/dnode/mnode/impl/src/mndTrans.c +++ b/source/dnode/mnode/impl/src/mndTrans.c @@ -21,6 +21,7 @@ #include "mndShow.h" #include "mndSync.h" #include "mndUser.h" +#include "mndVgroup.h" #define TRANS_VER_NUMBER 1 #define TRANS_ARRAY_SIZE 8 @@ -56,6 +57,7 @@ static bool mndCannotExecuteTransAction(SMnode *pMnode) { return !pMnode->dep static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans); static int32_t mndProcessTransReq(SRpcMsg *pReq); +static int32_t mndProcessTtl(SRpcMsg *pReq); static int32_t mndProcessKillTransReq(SRpcMsg *pReq); static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows); @@ -72,6 +74,7 @@ int32_t mndInitTrans(SMnode *pMnode) { .deleteFp = (SdbDeleteFp)mndTransActionDelete, }; + mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtl); mndSetMsgHandle(pMnode, TDMT_MND_TRANS_TIMER, mndProcessTransReq); mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq); @@ -1346,6 +1349,34 @@ static int32_t mndProcessTransReq(SRpcMsg *pReq) { return 0; } +static int32_t mndProcessTtl(SRpcMsg *pReq) { + SMnode *pMnode = pReq->info.node; + SSdb *pSdb = pMnode->pSdb; + SVgObj *pVgroup = NULL; + void *pIter = NULL; + + while (1) { + pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); + if (pIter == NULL) break; + + SMsgHead *pHead = (SMsgHead *)(pReq->pCont); + + pHead->contLen = htonl(pReq->contLen); + pHead->vgId = htonl(pVgroup->vgId); + + SRpcMsg rpcMsg = {.msgType = TDMT_VND_DROP_TTL_TABLE, .pCont = pReq->pCont, .contLen = pReq->contLen}; + + SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup); + int32_t code = tmsgSendReq(&epSet, &rpcMsg); + if(code != 0){ + mError("ttl time seed err. code:%d", code); + } + mError("ttl time seed succ"); + sdbRelease(pSdb, pVgroup); + } + return 0; +} + int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) { SArray *pArray = NULL; if (pTrans->stage == TRN_STAGE_REDO_ACTION) { From f2ad546666dc1e4b1ba8672f879828efbf467f0c Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Tue, 21 Jun 2022 18:37:47 +0800 Subject: [PATCH 2/4] fix: add configure for ttl unit --- include/common/tglobal.h | 3 +++ source/common/src/tglobal.c | 7 +++++++ source/dnode/mnode/impl/src/mndMain.c | 2 +- source/dnode/mnode/impl/src/mndStb.c | 2 +- source/dnode/vnode/src/meta/metaTable.c | 3 +-- source/dnode/vnode/src/vnd/vnodeSvr.c | 2 +- 6 files changed, 14 insertions(+), 5 deletions(-) diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 8c03d3ff42..3f68bb4147 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -138,6 +138,9 @@ extern bool tsSmlDataFormat; extern int32_t tsTransPullupInterval; extern int32_t tsMqRebalanceInterval; +// ttl unit +extern int32_t tsTtlUnit; + #define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize) int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd, diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 269c92a670..cd7cae8471 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -188,6 +188,8 @@ bool tsStartUdfd = true; int32_t tsTransPullupInterval = 2; int32_t tsMqRebalanceInterval = 2; +int32_t tsTtlUnit = 86400; + void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary) { tstrncpy(tsDiskCfg[index].dir, v1, TSDB_FILENAME_LEN); tsDiskCfg[index].level = level; @@ -468,6 +470,9 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, 1) != 0) return -1; if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, 1) != 0) return -1; + if (cfgAddInt32(pCfg, "ttlUnit", tsTtlUnit, 1, 86400*365, 1) != 0) return -1; + + if (cfgAddBool(pCfg, "udf", tsStartUdfd, 0) != 0) return -1; return 0; } @@ -620,6 +625,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsTransPullupInterval = cfgGetItem(pCfg, "transPullupInterval")->i32; tsMqRebalanceInterval = cfgGetItem(pCfg, "mqRebalanceInterval")->i32; + tsTtlUnit = cfgGetItem(pCfg, "ttlUnit")->i32; + tsStartUdfd = cfgGetItem(pCfg, "udf")->bval; if (tsQueryBufferSize >= 0) { diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c index 7a5220d0b2..e0c3b8d2a1 100644 --- a/source/dnode/mnode/impl/src/mndMain.c +++ b/source/dnode/mnode/impl/src/mndMain.c @@ -95,7 +95,7 @@ static void *mndThreadFp(void *param) { taosMsleep(100); if (mndGetStop(pMnode)) break; - if (lastTime % (864000) == 1) { // sleep 1 day for ttl + if (lastTime % (600) == 1) { mndTtlTimer(pMnode); } diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index b5777183db..27c1e558e8 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -805,7 +805,7 @@ static int32_t mndProcessTtl(SRpcMsg *pReq) { if(code != 0){ mError("ttl time seed err. code:%d", code); } - mError("ttl time seed succ. time:%d", t); + mDebug("ttl time seed succ. time:%d", t); sdbRelease(pSdb, pVgroup); } return 0; diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index bf5d5912f9..2bbea593fa 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -400,8 +400,7 @@ static void metaBuildTtlIdxKey(STtlIdxKey *ttlKey, const SMetaEntry *pME){ if (ttlDays <= 0) return; - ttlKey->dtime = ctime / 1000 + ttlDays * 24 * 60 * 60; -// ttlKey->dtime = ctime / 1000 + ttlDays; + ttlKey->dtime = ctime / 1000 + ttlDays * tsTtlUnit; ttlKey->uid = pME->uid; } diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index b5a2da091f..c5023185c8 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -309,7 +309,7 @@ static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t version, void *p if (tbUids == NULL) return TSDB_CODE_OUT_OF_MEMORY; int32_t t = ntohl(*(int32_t *)pReq); - vError("rec ttl time:%d", t); + vDebug("rec ttl time:%d", t); int32_t ret = metaTtlDropTable(pVnode->pMeta, t, tbUids); if (ret != 0) { goto end; From ab37b6e546edf177d787bbe1faa58f9b96556680 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 22 Jun 2022 11:18:10 +0800 Subject: [PATCH 3/4] refactor: adjust code styles --- include/common/tglobal.h | 2 -- source/common/src/tglobal.c | 4 ---- source/dnode/mnode/impl/src/mndMain.c | 3 +-- source/dnode/mnode/impl/src/mndStb.c | 26 +++++++++++++------------- source/dnode/vnode/src/vnd/vnodeSvr.c | 2 +- 5 files changed, 15 insertions(+), 22 deletions(-) diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 3f68bb4147..c20eff7773 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -137,8 +137,6 @@ extern bool tsSmlDataFormat; // internal extern int32_t tsTransPullupInterval; extern int32_t tsMqRebalanceInterval; - -// ttl unit extern int32_t tsTtlUnit; #define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize) diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index cd7cae8471..877b5ed26f 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -187,7 +187,6 @@ bool tsStartUdfd = true; // internal int32_t tsTransPullupInterval = 2; int32_t tsMqRebalanceInterval = 2; - int32_t tsTtlUnit = 86400; void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary) { @@ -469,10 +468,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, 1) != 0) return -1; if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, 1) != 0) return -1; - if (cfgAddInt32(pCfg, "ttlUnit", tsTtlUnit, 1, 86400*365, 1) != 0) return -1; - if (cfgAddBool(pCfg, "udf", tsStartUdfd, 0) != 0) return -1; return 0; } @@ -624,7 +621,6 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsTransPullupInterval = cfgGetItem(pCfg, "transPullupInterval")->i32; tsMqRebalanceInterval = cfgGetItem(pCfg, "mqRebalanceInterval")->i32; - tsTtlUnit = cfgGetItem(pCfg, "ttlUnit")->i32; tsStartUdfd = cfgGetItem(pCfg, "udf")->bval; diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c index e0c3b8d2a1..40e435d2c1 100644 --- a/source/dnode/mnode/impl/src/mndMain.c +++ b/source/dnode/mnode/impl/src/mndMain.c @@ -67,7 +67,6 @@ static void mndTtlTimer(SMnode *pMnode) { int32_t contLen = 0; void *pReq = mndBuildTimerMsg(&contLen); SRpcMsg rpcMsg = {.msgType = TDMT_MND_TTL_TIMER, .pCont = pReq, .contLen = contLen}; - tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg); } @@ -525,7 +524,7 @@ static int32_t mndCheckMnodeState(SRpcMsg *pMsg) { if (mndAcquireRpcRef(pMsg->info.node) == 0) return 0; if (pMsg->msgType != TDMT_MND_MQ_TIMER && pMsg->msgType != TDMT_MND_TELEM_TIMER && - pMsg->msgType != TDMT_MND_TRANS_TIMER) { + pMsg->msgType != TDMT_MND_TRANS_TIMER && pMsg->msgType != TDMT_MND_TTL_TIMER) { mError("msg:%p, failed to check mnode state since %s, type:%s", pMsg, terrstr(), TMSG_INFO(pMsg->msgType)); SEpSet epSet = {0}; diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index 27c1e558e8..e0ac4f8ea8 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -37,7 +37,7 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw); static int32_t mndStbActionInsert(SSdb *pSdb, SStbObj *pStb); static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb); static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew); -static int32_t mndProcessTtl(SRpcMsg *pReq); +static int32_t mndProcessTtlTimer(SRpcMsg *pReq); static int32_t mndProcessCreateStbReq(SRpcMsg *pReq); static int32_t mndProcessAlterStbReq(SRpcMsg *pReq); static int32_t mndProcessDropStbReq(SRpcMsg *pReq); @@ -63,8 +63,7 @@ int32_t mndInitStb(SMnode *pMnode) { mndSetMsgHandle(pMnode, TDMT_VND_ALTER_STB_RSP, mndTransProcessRsp); mndSetMsgHandle(pMnode, TDMT_VND_DROP_STB_RSP, mndTransProcessRsp); mndSetMsgHandle(pMnode, TDMT_MND_TABLE_META, mndProcessTableMetaReq); - mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtl); - + mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtlTimer); mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STB, mndRetrieveStb); mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STB, mndCancelGetNextStb); @@ -775,7 +774,7 @@ int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *p return 0; } -static int32_t mndProcessTtl(SRpcMsg *pReq) { +static int32_t mndProcessTtlTimer(SRpcMsg *pReq) { SMnode *pMnode = pReq->info.node; SSdb *pSdb = pMnode->pSdb; SVgObj *pVgroup = NULL; @@ -785,10 +784,10 @@ static int32_t mndProcessTtl(SRpcMsg *pReq) { pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); if (pIter == NULL) break; - int32_t contLen = sizeof(SMsgHead) + sizeof(int32_t); - SMsgHead *pHead = rpcMallocCont(contLen); + int32_t contLen = sizeof(SMsgHead) + sizeof(int32_t); + SMsgHead *pHead = rpcMallocCont(contLen); if (pHead == NULL) { - mError("ttl time malloc err. contLen:%d", contLen); + sdbCancelFetch(pSdb, pVgroup); sdbRelease(pSdb, pVgroup); continue; } @@ -796,18 +795,19 @@ static int32_t mndProcessTtl(SRpcMsg *pReq) { pHead->vgId = htonl(pVgroup->vgId); int32_t t = taosGetTimestampSec(); - *(int32_t*)(POINTER_SHIFT(pHead, sizeof(SMsgHead))) = htonl(t); + *(int32_t *)((char *)pHead + sizeof(SMsgHead)) = htonl(t); SRpcMsg rpcMsg = {.msgType = TDMT_VND_DROP_TTL_TABLE, .pCont = pHead, .contLen = contLen}; - - SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup); + SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup); int32_t code = tmsgSendReq(&epSet, &rpcMsg); - if(code != 0){ - mError("ttl time seed err. code:%d", code); + if (code != 0) { + mError("failed to send ttl time seed, code:0x%x", code); + } else { + mDebug("send ttl time seed success, time:%d", t); } - mDebug("ttl time seed succ. time:%d", t); sdbRelease(pSdb, pVgroup); } + return 0; } diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index c5023185c8..61f6dd1193 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -309,7 +309,7 @@ static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t version, void *p if (tbUids == NULL) return TSDB_CODE_OUT_OF_MEMORY; int32_t t = ntohl(*(int32_t *)pReq); - vDebug("rec ttl time:%d", t); + vDebug("vgId:%d, recv ttl msg, time:%d", pVnode->config.vgId, t); int32_t ret = metaTtlDropTable(pVnode->pMeta, t, tbUids); if (ret != 0) { goto end; From d9dc806a210624692ab2f59d05163410fde0cecd Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 22 Jun 2022 17:11:18 +0800 Subject: [PATCH 4/4] feat: add configure for ttl --- include/common/tglobal.h | 1 + source/common/src/tglobal.c | 4 ++++ source/dnode/mnode/impl/src/mndMain.c | 2 +- 3 files changed, 6 insertions(+), 1 deletion(-) diff --git a/include/common/tglobal.h b/include/common/tglobal.h index c20eff7773..e083ebcf78 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -138,6 +138,7 @@ extern bool tsSmlDataFormat; extern int32_t tsTransPullupInterval; extern int32_t tsMqRebalanceInterval; extern int32_t tsTtlUnit; +extern int32_t tsTtlPushInterval; #define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize) diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 877b5ed26f..a30eb3fcef 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -188,6 +188,8 @@ bool tsStartUdfd = true; int32_t tsTransPullupInterval = 2; int32_t tsMqRebalanceInterval = 2; int32_t tsTtlUnit = 86400; +int32_t tsTtlPushInterval = 60; + void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary) { tstrncpy(tsDiskCfg[index].dir, v1, TSDB_FILENAME_LEN); @@ -469,6 +471,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, 1) != 0) return -1; if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, 1) != 0) return -1; if (cfgAddInt32(pCfg, "ttlUnit", tsTtlUnit, 1, 86400*365, 1) != 0) return -1; + if (cfgAddInt32(pCfg, "ttlPushInterval", tsTtlPushInterval, 1, 10000, 1) != 0) return -1; if (cfgAddBool(pCfg, "udf", tsStartUdfd, 0) != 0) return -1; return 0; @@ -622,6 +625,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsTransPullupInterval = cfgGetItem(pCfg, "transPullupInterval")->i32; tsMqRebalanceInterval = cfgGetItem(pCfg, "mqRebalanceInterval")->i32; tsTtlUnit = cfgGetItem(pCfg, "ttlUnit")->i32; + tsTtlPushInterval = cfgGetItem(pCfg, "ttlPushInterval")->i32; tsStartUdfd = cfgGetItem(pCfg, "udf")->bval; diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c index d7cc1a5780..9965b803ea 100644 --- a/source/dnode/mnode/impl/src/mndMain.c +++ b/source/dnode/mnode/impl/src/mndMain.c @@ -100,7 +100,7 @@ static void *mndThreadFp(void *param) { taosMsleep(100); if (mndGetStop(pMnode)) break; - if (lastTime % 600 == 1) { + if (lastTime % (tsTransPullupInterval * 10) == 1) { mndTtlTimer(pMnode); }