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;