xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix:error in windows

This commit is contained in:
wangmm0220 2022-06-20 15:23:48 +08:00
parent ee1b1007d9
commit 934d0f9784
3 changed files with 51 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/common/tmsgdef.h
View File

@ -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)

57
source/dnode/mnode/impl/src/mndMain.c
View File

@ -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);
}

31
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -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) {