diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index bef26cb3102173b6353afd83fa4ae05b3de97803..c226d7c8ccbd2cb2e9c248ec895d1220ef40e104 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -26,6 +26,8 @@ extern "C" { extern bool gRaftDetailLog; +#define SYNC_RESP_TTL_MS 5000 + #define SYNC_MAX_BATCH_SIZE 500 #define SYNC_INDEX_BEGIN 0 #define SYNC_INDEX_INVALID -1 diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index bdcfe208d693f9bc6a655b58f4836e4f3396300e..87148a8450da794b6dd0cf980b7415f2e490c014 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -427,13 +427,22 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType)); - SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen}; - rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); - memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen); - syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info); - rpcMsg.info.conn.applyIndex = cbMeta.index; - rpcMsg.info.conn.applyTerm = cbMeta.term; - tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg); + if (cbMeta.code == 0) { + SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen}; + rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); + memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen); + syncGetAndDelRespRpc(pVnode->sync, cbMeta.seqNum, &rpcMsg.info); + rpcMsg.info.conn.applyIndex = cbMeta.index; + rpcMsg.info.conn.applyTerm = cbMeta.term; + tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg); + } else { + SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info}; + vError("vgId:%d, sync commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync), pMsg->msgType, + TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code)); + if (rsp.info.handle != NULL) { + tmsgSendRsp(&rsp); + } + } } static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index abc0f53611663a509bf82097e0d627da0c70be10..918496c894ae5f124a0d9104541aea91099d0fc0 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -1055,19 +1055,12 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) { } // tools - pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, 0); + pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS); ASSERT(pSyncNode->pSyncRespMgr != NULL); // restore state pSyncNode->restoreFinish = false; - // pSyncNode->pSnapshot = NULL; - // if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) { - // pSyncNode->pSnapshot = taosMemoryMalloc(sizeof(SSnapshot)); - // pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, pSyncNode->pSnapshot); - // } - // tsem_init(&(pSyncNode->restoreSem), 0, 0); - // snapshot senders for (int i = 0; i < TSDB_MAX_REPLICA; ++i) { SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, i); diff --git a/source/libs/sync/src/syncRespMgr.c b/source/libs/sync/src/syncRespMgr.c index eaeadd39910bc8a8e0e68f0096bcb88785e6ed0c..2aaa98b2999be6e725e861589c5a79b29f1636d1 100644 --- a/source/libs/sync/src/syncRespMgr.c +++ b/source/libs/sync/src/syncRespMgr.c @@ -122,54 +122,43 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) { int cnt = 0; SSyncNode *pSyncNode = pObj->data; - SArray *delIndexArray = taosArrayInit(0, sizeof(SyncIndex)); + SArray *delIndexArray = taosArrayInit(0, sizeof(uint64_t)); ASSERT(delIndexArray != NULL); while (pStub) { - size_t len; - void * key = taosHashGetKey(pStub, &len); - SyncIndex *pIndex = (SyncIndex *)key; + size_t len; + void *key = taosHashGetKey(pStub, &len); + uint64_t *pSeqNum = (uint64_t *)key; int64_t nowMS = taosGetTimestampMs(); if (nowMS - pStub->createTime > ttl) { - taosArrayPush(delIndexArray, pIndex); + taosArrayPush(delIndexArray, pSeqNum); cnt++; - SSyncRaftEntry *pEntry = NULL; - int32_t code = 0; - if (pSyncNode->pLogStore != NULL) { - code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, *pIndex, &pEntry); - if (code == 0 && pEntry != NULL) { - SFsmCbMeta cbMeta = {0}; - cbMeta.index = pEntry->index; - cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(pSyncNode, cbMeta.index); - cbMeta.isWeak = pEntry->isWeak; - cbMeta.code = TSDB_CODE_SYN_TIMEOUT; - cbMeta.state = pSyncNode->state; - cbMeta.seqNum = pEntry->seqNum; - cbMeta.term = pEntry->term; - cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm; - cbMeta.flag = 0; - - SRpcMsg rpcMsg = pStub->rpcMsg; - rpcMsg.pCont = rpcMallocCont(pEntry->dataLen); - memcpy(rpcMsg.pCont, pEntry->data, pEntry->dataLen); - pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, cbMeta); - - syncEntryDestory(pEntry); - } - } + SFsmCbMeta cbMeta = {0}; + cbMeta.index = SYNC_INDEX_INVALID; + cbMeta.lastConfigIndex = SYNC_INDEX_INVALID; + cbMeta.isWeak = false; + cbMeta.code = TSDB_CODE_SYN_TIMEOUT; + cbMeta.state = pSyncNode->state; + cbMeta.seqNum = *pSeqNum; + cbMeta.term = SYNC_TERM_INVALID; + cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm; + cbMeta.flag = 0; + + pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &(pStub->rpcMsg), cbMeta); } pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, pStub); } int32_t arraySize = taosArrayGetSize(delIndexArray); - sDebug("vgId:%d, resp clean by ttl, cnt:%d, array-size:%d", pSyncNode->vgId, cnt, arraySize); + sDebug("vgId:%d, resp mgr clean by ttl, cnt:%d, array-size:%d", pSyncNode->vgId, cnt, arraySize); for (int32_t i = 0; i < arraySize; ++i) { - SyncIndex *pIndex = taosArrayGet(delIndexArray, i); - taosHashRemove(pObj->pRespHash, pIndex, sizeof(SyncIndex)); + uint64_t *pSeqNum = taosArrayGet(delIndexArray, i); + taosHashRemove(pObj->pRespHash, pSeqNum, sizeof(uint64_t)); + sDebug("vgId:%d, resp mgr clean by ttl, seq:%d", pSyncNode->vgId, *pSeqNum); } taosArrayDestroy(delIndexArray); } diff --git a/source/libs/sync/src/syncTimeout.c b/source/libs/sync/src/syncTimeout.c index 52181a3da8c0d9a21e9ec8a0eb3acdc6e7a1ca82..97de75c108c2829ea16daec6a7a3591de1b50f6c 100644 --- a/source/libs/sync/src/syncTimeout.c +++ b/source/libs/sync/src/syncTimeout.c @@ -16,9 +16,11 @@ #include "syncTimeout.h" #include "syncElection.h" #include "syncReplication.h" +#include "syncRespMgr.h" int32_t syncNodeTimerRoutine(SSyncNode* ths) { syncNodeEventLog(ths, "timer routines ... "); + syncRespClean(ths->pSyncRespMgr); return 0; }