diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c index bc6830b8f3..5e708616fd 100644 --- a/source/dnode/mnode/impl/src/mndMain.c +++ b/source/dnode/mnode/impl/src/mndMain.c @@ -519,6 +519,8 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) { } } + syncNodeRelease(pSyncNode); + if (code != 0) { terrno = TSDB_CODE_SYN_INTERNAL_ERROR; } diff --git a/source/libs/sync/inc/syncSnapshot.h b/source/libs/sync/inc/syncSnapshot.h index 3df9c243e7..3b1e4f4560 100644 --- a/source/libs/sync/inc/syncSnapshot.h +++ b/source/libs/sync/inc/syncSnapshot.h @@ -83,9 +83,9 @@ typedef struct SSyncSnapshotReceiver { SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId); void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver); -int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg); -int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver); -bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver); +int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg); +int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver); +bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver); cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver); char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 21d6b55cfa..cefd306f7d 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -158,13 +158,13 @@ int32_t syncSetStandby(int64_t rid) { } if (pSyncNode->state != TAOS_SYNC_STATE_FOLLOWER) { - taosReleaseRef(tsNodeRefId, pSyncNode->rid); if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { terrno = TSDB_CODE_SYN_IS_LEADER; } else { terrno = TSDB_CODE_SYN_STANDBY_NOT_READY; } sError("failed to set standby since it is not follower, state:%s rid:%" PRId64, syncStr(pSyncNode->state), rid); + taosReleaseRef(tsNodeRefId, pSyncNode->rid); return -1; } @@ -620,6 +620,7 @@ int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) { SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid); if (pSyncNode == NULL) { + taosReleaseRef(tsNodeRefId, rid); terrno = TSDB_CODE_SYN_INTERNAL_ERROR; return -1; } diff --git a/source/libs/sync/src/syncRespMgr.c b/source/libs/sync/src/syncRespMgr.c index d2cbabe226..990a92aad7 100644 --- a/source/libs/sync/src/syncRespMgr.c +++ b/source/libs/sync/src/syncRespMgr.c @@ -14,6 +14,7 @@ */ #include "syncRespMgr.h" +#include "syncRaftEntry.h" #include "syncRaftStore.h" SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl) { @@ -116,4 +117,59 @@ void syncRespClean(SSyncRespMgr *pObj) { taosThreadMutexUnlock(&(pObj->mutex)); } -void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) {} +void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) { + SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL); + int cnt = 0; + SSyncNode *pSyncNode = pObj->data; + + SArray *delIndexArray = taosArrayInit(0, sizeof(SyncIndex)); + ASSERT(delIndexArray != NULL); + + while (pStub) { + size_t len; + void *key = taosHashGetKey(pStub, &len); + SyncIndex *pIndex = (SyncIndex *)key; + + int64_t nowMS = taosGetTimestampMs(); + if (nowMS - pStub->createTime > ttl) { + taosArrayPush(delIndexArray, pIndex); + 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); + } + } + } + + 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); + + for (int32_t i = 0; i < arraySize; ++i) { + SyncIndex *pIndex = taosArrayGet(delIndexArray, i); + taosHashRemove(pObj->pRespHash, pIndex, sizeof(SyncIndex)); + } + taosArrayDestroy(delIndexArray); +} diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index cefe676f90..5cdfec72c5 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -22,9 +22,11 @@ #include "wal.h" //---------------------------------- -static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, - SyncSnapshotSend *pBeginMsg); -static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg); +static void snapshotSenderUpdateProgress(SSyncSnapshotSender *pSender, SyncSnapshotRsp *pMsg); +static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg); +static void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver); +static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg); +static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg); //---------------------------------- SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex) { @@ -68,7 +70,9 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) { // close reader if (pSender->pReader != NULL) { int32_t ret = pSender->pSyncNode->pFsm->FpSnapshotStopRead(pSender->pSyncNode->pFsm, pSender->pReader); - ASSERT(ret == 0); + if (ret != 0) { + syncNodeErrorLog(pSender->pSyncNode, "stop reader error"); + } pSender->pReader = NULL; } @@ -79,7 +83,12 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) { bool snapshotSenderIsStart(SSyncSnapshotSender *pSender) { return pSender->start; } -// begin send snapshot by snapshot, pReader +// begin send snapshot by param, snapshot, pReader +// +// action: +// 1. assert reader not start +// 2. update state +// 3. send first snapshot block int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshotParam snapshotParam, SSnapshot snapshot, void *pReader) { ASSERT(!snapshotSenderIsStart(pSender)); @@ -98,7 +107,7 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshotParam snapsho // update term pSender->term = pSender->pSyncNode->pRaftStore->currentTerm; - ++(pSender->privateTerm); + ++(pSender->privateTerm); // increase private term // update state pSender->finish = false; @@ -114,9 +123,7 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshotParam snapsho code = pSender->pSyncNode->pLogStore->syncLogGetEntry(pSender->pSyncNode->pLogStore, pSender->snapshot.lastConfigIndex, &pEntry); - if (code == 0) { - ASSERT(pEntry != NULL); - + if (code == 0 && pEntry != NULL) { SRpcMsg rpcMsg; syncEntry2OriginalRpc(pEntry, &rpcMsg); @@ -207,6 +214,8 @@ int32_t snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) { pSender->start = false; pSender->finish = finish; + // do not update term, maybe print + // event log do { char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender stop"); @@ -243,6 +252,7 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) { pMsg->srcId = pSender->pSyncNode->myRaftId; pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex]; pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm; + pMsg->beginIndex = pSender->snapshotParam.start; pMsg->lastIndex = pSender->snapshot.lastApplyIndex; pMsg->lastTerm = pSender->snapshot.lastApplyTerm; pMsg->lastConfigIndex = pSender->snapshot.lastConfigIndex; @@ -281,11 +291,13 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) { pMsg->srcId = pSender->pSyncNode->myRaftId; pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex]; pMsg->term = pSender->pSyncNode->pRaftStore->currentTerm; + pMsg->beginIndex = pSender->snapshotParam.start; pMsg->lastIndex = pSender->snapshot.lastApplyIndex; pMsg->lastTerm = pSender->snapshot.lastApplyTerm; pMsg->lastConfigIndex = pSender->snapshot.lastConfigIndex; pMsg->lastConfig = pSender->lastConfig; pMsg->seq = pSender->seq; + pMsg->privateTerm = pSender->privateTerm; memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen); // send msg @@ -305,6 +317,12 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) { return 0; } +static void snapshotSenderUpdateProgress(SSyncSnapshotSender *pSender, SyncSnapshotRsp *pMsg) { + ASSERT(pMsg->ack == pSender->seq); + pSender->ack = pMsg->ack; + ++(pSender->seq); +} + cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) { char u64buf[128]; cJSON *pRoot = cJSON_CreateObject(); @@ -371,10 +389,11 @@ char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) { syncUtilU642Addr(destId.addr, host, sizeof(host), &port); snprintf(s, len, - "%s {%p laindex:%ld laterm:%lu lcindex:%ld seq:%d ack:%d finish:%d pterm:%lu replica-index:%d %s:%d}", event, - pSender, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, - pSender->snapshot.lastConfigIndex, pSender->seq, pSender->ack, pSender->finish, pSender->privateTerm, - pSender->replicaIndex, host, port); + "%s {%p s-param:%ld e-param:%ld laindex:%ld laterm:%lu lcindex:%ld seq:%d ack:%d finish:%d pterm:%lu " + "replica-index:%d %s:%d}", + event, pSender, pSender->snapshotParam.start, pSender->snapshotParam.end, pSender->snapshot.lastApplyIndex, + pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex, pSender->seq, pSender->ack, + pSender->finish, pSender->privateTerm, pSender->replicaIndex, host, port); return s; } @@ -429,11 +448,10 @@ bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) { return pReceive // static do start by privateTerm, pBeginMsg // receive first snapshot data // write first block data -static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, - SyncSnapshotSend *pBeginMsg) { +static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg) { // update state pReceiver->term = pReceiver->pSyncNode->pRaftStore->currentTerm; - pReceiver->privateTerm = privateTerm; + pReceiver->privateTerm = pBeginMsg->privateTerm; pReceiver->ack = SYNC_SNAPSHOT_SEQ_BEGIN; pReceiver->fromId = pBeginMsg->srcId; pReceiver->start = true; @@ -445,7 +463,7 @@ static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm p pReceiver->snapshotParam.start = pBeginMsg->beginIndex; pReceiver->snapshotParam.end = pBeginMsg->lastIndex; - // write data + // start writer ASSERT(pReceiver->pWriter == NULL); int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStartWrite(pReceiver->pSyncNode->pFsm, &(pReceiver->snapshotParam), &(pReceiver->pWriter)); @@ -481,10 +499,10 @@ static void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) { // if receiver receive msg from seq = SYNC_SNAPSHOT_SEQ_BEGIN, start receiver // if already start, force close, start again -int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg) { +int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg) { if (!snapshotReceiverIsStart(pReceiver)) { // first start - snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg); + snapshotReceiverDoStart(pReceiver, pBeginMsg); } else { // already start @@ -494,12 +512,14 @@ int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm private snapshotReceiverForceStop(pReceiver); // start again - snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg); + snapshotReceiverDoStart(pReceiver, pBeginMsg); } return 0; } +// just set start = false +// FpSnapshotStopWrite should not be called, assert writer == NULL int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) { if (pReceiver->pWriter != NULL) { int32_t ret = @@ -522,6 +542,7 @@ int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) { return 0; } +// when recv last snapshot block, apply data into snapshot static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) { ASSERT(pMsg->seq == SYNC_SNAPSHOT_SEQ_END); @@ -550,7 +571,7 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap pReceiver->pSyncNode->commitIndex = pReceiver->snapshot.lastApplyIndex; } - // stop writer + // stop writer, apply data code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true); if (code != 0) { syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error"); @@ -579,15 +600,20 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap return 0; } +// apply data block +// update progress static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) { ASSERT(pMsg->seq == pReceiver->ack + 1); if (pReceiver->pWriter != NULL) { if (pMsg->dataLen > 0) { + // apply data block int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotDoWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, pMsg->data, pMsg->dataLen); ASSERT(code == 0); } + + // update progress pReceiver->ack = pMsg->seq; // event log @@ -665,14 +691,23 @@ char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event) uint16_t port; syncUtilU642Addr(fromId.addr, host, sizeof(host), &port); - snprintf(s, len, "%s {%p start:%d ack:%d term:%lu pterm:%lu from:%s:%d laindex:%ld laterm:%lu lcindex:%ld}", event, - pReceiver, pReceiver->start, pReceiver->ack, pReceiver->term, pReceiver->privateTerm, host, port, - pReceiver->snapshot.lastApplyIndex, pReceiver->snapshot.lastApplyTerm, pReceiver->snapshot.lastConfigIndex); + snprintf(s, len, + "%s {%p start:%d ack:%d term:%lu pterm:%lu from:%s:%d s-param:%ld e-param:%ld laindex:%ld laterm:%lu " + "lcindex:%ld}", + event, pReceiver, pReceiver->start, pReceiver->ack, pReceiver->term, pReceiver->privateTerm, host, port, + pReceiver->snapshotParam.start, pReceiver->snapshotParam.end, pReceiver->snapshot.lastApplyIndex, + pReceiver->snapshot.lastApplyTerm, pReceiver->snapshot.lastConfigIndex); return s; } -// receiver do something +// receiver on message +// +// condition 1, recv SYNC_SNAPSHOT_SEQ_BEGIN, start receiver, update privateTerm +// condition 2, recv SYNC_SNAPSHOT_SEQ_END, finish receiver(apply snapshot data, update commit index, maybe reconfig) +// condition 3, recv SYNC_SNAPSHOT_SEQ_FORCE_CLOSE, force close +// condition 4, got data, update ack +// int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { // get receiver SSyncSnapshotReceiver *pReceiver = pSyncNode->pNewNodeReceiver; @@ -683,11 +718,13 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) { if (pMsg->term == pSyncNode->pRaftStore->currentTerm) { if (pMsg->seq == SYNC_SNAPSHOT_SEQ_BEGIN) { + // condition 1 // begin, no data - snapshotReceiverStart(pReceiver, pMsg->privateTerm, pMsg); + snapshotReceiverStart(pReceiver, pMsg); needRsp = true; } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_END) { + // condition 2 // end, finish FSM code = snapshotReceiverFinish(pReceiver, pMsg); if (code == 0) { @@ -697,7 +734,6 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { // maybe update lastconfig if (pMsg->lastConfigIndex >= SYNC_INDEX_BEGIN) { - // int32_t oldReplicaNum = pSyncNode->replicaNum; SSyncCfg oldSyncCfg = pSyncNode->pRaftCfg->cfg; // update new config myIndex @@ -709,11 +745,13 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { } } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_FORCE_CLOSE) { + // condition 3 // force close snapshotReceiverForceStop(pReceiver); needRsp = false; } else if (pMsg->seq > SYNC_SNAPSHOT_SEQ_BEGIN && pMsg->seq < SYNC_SNAPSHOT_SEQ_END) { + // condition 4 // transfering if (pMsg->seq == pReceiver->ack + 1) { snapshotReceiverGotData(pReceiver, pMsg); @@ -752,6 +790,7 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { syncNodeSendMsgById(&(pRspMsg->destId), pSyncNode, &rpcMsg); syncSnapshotRspDestroy(pRspMsg); } + } else { // error log do { @@ -759,6 +798,8 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { syncNodeErrorLog(pSyncNode, eventLog); taosMemoryFree(eventLog); } while (0); + + return -1; } } else { // error log @@ -767,19 +808,19 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { syncNodeErrorLog(pSyncNode, eventLog); taosMemoryFree(eventLog); } while (0); + + return -1; } return 0; } -static void snapshotSenderUpdateProgress(SSyncSnapshotSender *pSender, SyncSnapshotRsp *pMsg) { - ASSERT(pMsg->ack == pSender->seq); - pSender->ack = pMsg->ack; - ++(pSender->seq); -} - -// sender receives ack, set seq = ack + 1, send msg from seq -// if ack == SYNC_SNAPSHOT_SEQ_END, stop sender +// sender on message +// +// condition 1 sender receives SYNC_SNAPSHOT_SEQ_END, close sender +// condition 2 sender receives ack, set seq = ack + 1, send msg from seq +// condition 3 sender receives error msg, just print error log +// int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) { // if already drop replica, do not process if (!syncNodeInRaftGroup(pSyncNode, &(pMsg->srcId)) && pSyncNode->state == TAOS_SYNC_STATE_LEADER) { @@ -794,12 +835,14 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) { // state, term, seq/ack if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { if (pMsg->term == pSyncNode->pRaftStore->currentTerm) { - // receiver ack is finish, close sender + // condition 1 + // receive ack is finish, close sender if (pMsg->ack == SYNC_SNAPSHOT_SEQ_END) { snapshotSenderStop(pSender, true); return 0; } + // condition 2 // send next msg if (pMsg->ack == pSender->seq) { // update sender ack @@ -807,6 +850,7 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) { snapshotSend(pSender); } else if (pMsg->ack == pSender->seq - 1) { + // maybe resend snapshotReSend(pSender); } else { diff --git a/source/libs/sync/test/syncIndexTest.cpp b/source/libs/sync/test/syncIndexTest.cpp index 1cf2847b5c..8627a6c174 100644 --- a/source/libs/sync/test/syncIndexTest.cpp +++ b/source/libs/sync/test/syncIndexTest.cpp @@ -8,7 +8,13 @@ void print(SHashObj *pNextIndex) { printf("----------------\n"); uint64_t *p = (uint64_t *)taosHashIterate(pNextIndex, NULL); while (p) { - printf("%lu \n", *p); + + size_t len; + void* key = taosHashGetKey(p, &len); + + SRaftId *pRaftId = (SRaftId*)key; + + printf("key:<%lu, %d>, value:%lu \n", pRaftId->addr, pRaftId->vgId, *p); p = (uint64_t *)taosHashIterate(pNextIndex, p); } } diff --git a/source/libs/sync/test/syncRespMgrTest.cpp b/source/libs/sync/test/syncRespMgrTest.cpp index 495b82bed7..fd18109280 100644 --- a/source/libs/sync/test/syncRespMgrTest.cpp +++ b/source/libs/sync/test/syncRespMgrTest.cpp @@ -73,9 +73,15 @@ void syncRespMgrGetAndDelTest(uint64_t i) { } } +SSyncNode *createSyncNode() { + SSyncNode *pSyncNode = (SSyncNode*)taosMemoryMalloc(sizeof(SSyncNode)); + memset(pSyncNode, 0, sizeof(SSyncNode)); + return pSyncNode; +} + void test1() { printf("------- test1 ---------\n"); - pMgr = syncRespMgrCreate(NULL, 0); + pMgr = syncRespMgrCreate(createSyncNode(), 0); assert(pMgr != NULL); syncRespMgrInsert(10); @@ -100,7 +106,7 @@ void test1() { void test2() { printf("------- test2 ---------\n"); - pMgr = syncRespMgrCreate(NULL, 0); + pMgr = syncRespMgrCreate(createSyncNode(), 0); assert(pMgr != NULL); syncRespMgrInsert(10); @@ -117,7 +123,7 @@ void test2() { void test3() { printf("------- test3 ---------\n"); - pMgr = syncRespMgrCreate(NULL, 0); + pMgr = syncRespMgrCreate(createSyncNode(), 0); assert(pMgr != NULL); syncRespMgrInsert(10); @@ -132,13 +138,34 @@ void test3() { syncRespMgrDestroy(pMgr); } +void test4() { + printf("------- test4 ---------\n"); + pMgr = syncRespMgrCreate(createSyncNode(), 2); + assert(pMgr != NULL); + + syncRespMgrInsert(5); + syncRespMgrPrint(); + + taosMsleep(3000); + + syncRespMgrInsert(3); + syncRespMgrPrint(); + + printf("====== after clean ttl \n"); + syncRespClean(pMgr); + syncRespMgrPrint(); + + syncRespMgrDestroy(pMgr); +} + int main() { tsAsyncLog = 0; - sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE; + sDebugFlag = DEBUG_DEBUG + DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE; logTest(); test1(); test2(); test3(); + test4(); return 0; }