diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index 0be27e6d3a65909deea9bbb5b638094cf598190f..ffe435a8b28c3b09007eb12b58b74b00be934db0 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -158,6 +158,8 @@ typedef enum { int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak); +bool syncEnvIsStart(); + extern int32_t sDebugFlag; //----------------------------------------- diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index e9071588973fb3ee3f8e22fd1ed5e767596c6ba6..310a9a336939f589c7d067f6a3f0a9d4e478a041 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -193,84 +193,89 @@ void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) { // sync integration int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { - SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync); - assert(pSyncNode != NULL); - ESyncState state = syncGetMyRole(pVnode->sync); - SyncTerm currentTerm = syncGetMyTerm(pVnode->sync); + if (syncEnvIsStart()) { - SMsgHead *pHead = pMsg->pCont; + SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync); + assert(pSyncNode != NULL); - char logBuf[512]; - char *syncNodeStr = sync2SimpleStr(pVnode->sync); - snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr); - syncRpcMsgLog2(logBuf, pMsg); - taosMemoryFree(syncNodeStr); + ESyncState state = syncGetMyRole(pVnode->sync); + SyncTerm currentTerm = syncGetMyTerm(pVnode->sync); - SRpcMsg *pRpcMsg = pMsg; + SMsgHead *pHead = pMsg->pCont; - if (pRpcMsg->msgType == TDMT_VND_SYNC_TIMEOUT) { - SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg); - assert(pSyncMsg != NULL); + char logBuf[512]; + char *syncNodeStr = sync2SimpleStr(pVnode->sync); + snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr); + syncRpcMsgLog2(logBuf, pMsg); + taosMemoryFree(syncNodeStr); - syncNodeOnTimeoutCb(pSyncNode, pSyncMsg); - syncTimeoutDestroy(pSyncMsg); + SRpcMsg *pRpcMsg = pMsg; - } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING) { - SyncPing *pSyncMsg = syncPingFromRpcMsg2(pRpcMsg); - assert(pSyncMsg != NULL); + if (pRpcMsg->msgType == TDMT_VND_SYNC_TIMEOUT) { + SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg); + assert(pSyncMsg != NULL); - syncNodeOnPingCb(pSyncNode, pSyncMsg); - syncPingDestroy(pSyncMsg); + syncNodeOnTimeoutCb(pSyncNode, pSyncMsg); + syncTimeoutDestroy(pSyncMsg); - } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING_REPLY) { - SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg); - assert(pSyncMsg != NULL); + } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING) { + SyncPing *pSyncMsg = syncPingFromRpcMsg2(pRpcMsg); + assert(pSyncMsg != NULL); - syncNodeOnPingReplyCb(pSyncNode, pSyncMsg); - syncPingReplyDestroy(pSyncMsg); + syncNodeOnPingCb(pSyncNode, pSyncMsg); + syncPingDestroy(pSyncMsg); - } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) { - SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg); - assert(pSyncMsg != NULL); + } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING_REPLY) { + SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg); + assert(pSyncMsg != NULL); - syncNodeOnClientRequestCb(pSyncNode, pSyncMsg); - syncClientRequestDestroy(pSyncMsg); + syncNodeOnPingReplyCb(pSyncNode, pSyncMsg); + syncPingReplyDestroy(pSyncMsg); - } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) { - SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg); - assert(pSyncMsg != NULL); + } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) { + SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg); + assert(pSyncMsg != NULL); - syncNodeOnRequestVoteCb(pSyncNode, pSyncMsg); - syncRequestVoteDestroy(pSyncMsg); + syncNodeOnClientRequestCb(pSyncNode, pSyncMsg); + syncClientRequestDestroy(pSyncMsg); - } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) { - SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg); - assert(pSyncMsg != NULL); + } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) { + SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg); + assert(pSyncMsg != NULL); - syncNodeOnRequestVoteReplyCb(pSyncNode, pSyncMsg); - syncRequestVoteReplyDestroy(pSyncMsg); + syncNodeOnRequestVoteCb(pSyncNode, pSyncMsg); + syncRequestVoteDestroy(pSyncMsg); - } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) { - SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg); - assert(pSyncMsg != NULL); + } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) { + SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg); + assert(pSyncMsg != NULL); - syncNodeOnAppendEntriesCb(pSyncNode, pSyncMsg); - syncAppendEntriesDestroy(pSyncMsg); + syncNodeOnRequestVoteReplyCb(pSyncNode, pSyncMsg); + syncRequestVoteReplyDestroy(pSyncMsg); - } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) { - SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg); - assert(pSyncMsg != NULL); + } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) { + SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg); + assert(pSyncMsg != NULL); - syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg); - syncAppendEntriesReplyDestroy(pSyncMsg); + syncNodeOnAppendEntriesCb(pSyncNode, pSyncMsg); + syncAppendEntriesDestroy(pSyncMsg); - } else { - vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType); - } + } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) { + SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg); + assert(pSyncMsg != NULL); - syncNodeRelease(pSyncNode); + syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg); + syncAppendEntriesReplyDestroy(pSyncMsg); + } else { + vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType); + } + + syncNodeRelease(pSyncNode); + } else { + vError("==vnodeProcessSyncReq== error syncEnv stop"); + } return 0; } diff --git a/source/libs/sync/inc/syncEnv.h b/source/libs/sync/inc/syncEnv.h index e9550ff989013fa99c6846659e580093d4ae9dc0..4f91990ccf29a93cdd7b9d4fd4f2e16bfc3df480 100644 --- a/source/libs/sync/inc/syncEnv.h +++ b/source/libs/sync/inc/syncEnv.h @@ -39,6 +39,8 @@ extern "C" { #define EMPTY_RAFT_ID ((SRaftId){.addr = 0, .vgId = 0}) typedef struct SSyncEnv { + uint8_t isStart; + // tick timer tmr_h pEnvTickTimer; int32_t envTickTimerMS; diff --git a/source/libs/sync/src/syncEnv.c b/source/libs/sync/src/syncEnv.c index ff2d4d4d27f691fe09240b473f3b8ed159ca6f87..5ad1758a8fa88e44be5a984b0cbec98438370b71 100644 --- a/source/libs/sync/src/syncEnv.c +++ b/source/libs/sync/src/syncEnv.c @@ -26,6 +26,14 @@ static int32_t doSyncEnvStopTimer(SSyncEnv *pSyncEnv); static void syncEnvTick(void *param, void *tmrId); // -------------------------------- +bool syncEnvIsStart() { + if (gSyncEnv == NULL) { + return false; + } + + return atomic_load_8(&(gSyncEnv->isStart)); +} + int32_t syncEnvStart() { int32_t ret = 0; taosSeedRand(taosGetTimestampSec()); @@ -88,11 +96,15 @@ static SSyncEnv *doSyncEnvStart() { // start tmr thread pSyncEnv->pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV"); + + atomic_store_8(&(pSyncEnv->isStart), 1); return pSyncEnv; } static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv) { assert(pSyncEnv == gSyncEnv); + atomic_store_8(&(pSyncEnv->isStart), 0); + if (pSyncEnv != NULL) { taosTmrCleanUp(pSyncEnv->pTimerManager); taosMemoryFree(pSyncEnv);