#include #include #include "syncEnv.h" #include "syncIO.h" #include "syncInt.h" #include "syncMessage.h" #include "syncRaftEntry.h" #include "syncRaftLog.h" #include "syncRaftStore.h" #include "syncUtil.h" #include "syncRaftLog.h" void logTest() { sTrace("--- sync log test: trace"); sDebug("--- sync log test: debug"); sInfo("--- sync log test: info"); sWarn("--- sync log test: warn"); sError("--- sync log test: error"); sFatal("--- sync log test: fatal"); } uint16_t ports[] = {7010, 7110, 7210, 7310, 7410}; int32_t replicaNum = 1; int32_t myIndex = 0; SRaftId ids[TSDB_MAX_REPLICA]; SSyncInfo syncInfo; SSyncFSM* pFsm; SWal* pWal; SSyncNode* pSyncNode; SSyncNode* syncNodeInit() { syncInfo.vgId = 1234; syncInfo.rpcClient = gSyncIO->clientRpc; syncInfo.FpSendMsg = syncIOSendMsg; syncInfo.queue = gSyncIO->pMsgQ; syncInfo.FpEqMsg = syncIOEqMsg; syncInfo.pFsm = pFsm; snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./"); int code = walInit(); assert(code == 0); SWalCfg walCfg; memset(&walCfg, 0, sizeof(SWalCfg)); walCfg.vgId = syncInfo.vgId; walCfg.fsyncPeriod = 1000; walCfg.retentionPeriod = 1000; walCfg.rollPeriod = 1000; walCfg.retentionSize = 1000; walCfg.segSize = 1000; walCfg.level = TAOS_WAL_FSYNC; pWal = walOpen("./wal_test", &walCfg); assert(pWal != NULL); syncInfo.pWal = pWal; SSyncCfg* pCfg = &syncInfo.syncCfg; pCfg->myIndex = myIndex; pCfg->replicaNum = replicaNum; for (int i = 0; i < replicaNum; ++i) { pCfg->nodeInfo[i].nodePort = ports[i]; snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1"); // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn); } pSyncNode = syncNodeOpen(&syncInfo); assert(pSyncNode != NULL); gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing; gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply; gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote; gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply; gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries; gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply; gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing; gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply; gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout; gSyncIO->pSyncNode = pSyncNode; return pSyncNode; } SSyncNode* syncInitTest() { return syncNodeInit(); } void logStoreTest() { logStorePrint2((char*)"logStoreTest", pSyncNode->pLogStore); assert(pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore) == SYNC_INDEX_INVALID); for (int i = 0; i < 5; ++i) { int32_t dataLen = 10; SSyncRaftEntry* pEntry = syncEntryBuild(dataLen); assert(pEntry != NULL); pEntry->msgType = 1; pEntry->originalRpcType = 2; pEntry->seqNum = 3; pEntry->isWeak = true; pEntry->term = 100; pEntry->index = pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore) + 1; snprintf(pEntry->data, dataLen, "value%d", i); // syncEntryPrint2((char*)"write entry:", pEntry); pSyncNode->pLogStore->appendEntry(pSyncNode->pLogStore, pEntry); syncEntryDestory(pEntry); if (i == 0) { assert(pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore) == SYNC_INDEX_BEGIN); } } logStorePrint2((char*)"after appendEntry", pSyncNode->pLogStore); pSyncNode->pLogStore->truncate(pSyncNode->pLogStore, 3); logStorePrint2((char*)"after truncate 3", pSyncNode->pLogStore); } void initRaftId(SSyncNode* pSyncNode) { for (int i = 0; i < replicaNum; ++i) { ids[i] = pSyncNode->replicasId[i]; char* s = syncUtilRaftId2Str(&ids[i]); printf("raftId[%d] : %s\n", i, s); free(s); } } SRpcMsg *step0() { SRpcMsg *pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg)); memset(pMsg, 0, sizeof(SRpcMsg)); pMsg->msgType = 9999; pMsg->contLen = 32; pMsg->pCont = malloc(pMsg->contLen); snprintf((char *)(pMsg->pCont), pMsg->contLen, "hello, world"); return pMsg; } SyncClientRequest *step1(const SRpcMsg *pMsg) { SyncClientRequest *pRetMsg = syncClientRequestBuild2(pMsg, 123, true); return pRetMsg; } SRpcMsg *step2(const SyncClientRequest *pMsg) { SRpcMsg *pRetMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg)); syncClientRequest2RpcMsg(pMsg, pRetMsg); return pRetMsg; } SyncClientRequest *step3(const SRpcMsg *pMsg) { SyncClientRequest *pRetMsg = syncClientRequestFromRpcMsg2(pMsg); return pRetMsg; } SSyncRaftEntry *step4(const SyncClientRequest *pMsg) { SSyncRaftEntry *pRetMsg = syncEntryBuild2((SyncClientRequest *)pMsg, 100, 0); return pRetMsg; } char *step5(const SSyncRaftEntry *pMsg, uint32_t *len) { char *pRetMsg = syncEntrySerialize(pMsg, len); return pRetMsg; } SSyncRaftEntry *step6(const char *pMsg, uint32_t len) { SSyncRaftEntry *pRetMsg = syncEntryDeserialize(pMsg, len); return pRetMsg; } SRpcMsg *step7(const SSyncRaftEntry *pMsg) { SRpcMsg *pRetMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg)); syncEntry2OriginalRpc(pMsg, pRetMsg); return pRetMsg; } int main(int argc, char **argv) { // taosInitLog((char *)"syncTest.log", 100000, 10); tsAsyncLog = 0; sDebugFlag = 143 + 64; void logTest(); myIndex = 0; if (argc >= 2) { myIndex = atoi(argv[1]); } int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); assert(ret == 0); ret = syncEnvStart(); assert(ret == 0); taosRemoveDir("./wal_test"); // step0 SRpcMsg *pMsg0 = step0(); syncRpcMsgPrint2((char *)"==step0==", pMsg0); // step1 SyncClientRequest *pMsg1 = step1(pMsg0); syncClientRequestPrint2((char *)"==step1==", pMsg1); // step2 SRpcMsg *pMsg2 = step2(pMsg1); syncRpcMsgPrint2((char *)"==step2==", pMsg2); // step3 SyncClientRequest *pMsg3 = step3(pMsg2); syncClientRequestPrint2((char *)"==step3==", pMsg3); // step4 SSyncRaftEntry *pMsg4 = step4(pMsg3); syncEntryPrint2((char *)"==step4==", pMsg4); // log, relog SSyncNode* pSyncNode = syncNodeInit(); assert(pSyncNode != NULL); SSyncRaftEntry *pEntry = pMsg4; pSyncNode->pLogStore->appendEntry(pSyncNode->pLogStore, pEntry); SSyncRaftEntry *pEntry2 = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, pEntry->index); syncEntryPrint2((char *)"==pEntry2==", pEntry2); // step5 uint32_t len; char * pMsg5 = step5(pMsg4, &len); char * s = syncUtilprintBin(pMsg5, len); printf("==step5== [%s] \n", s); free(s); // step6 SSyncRaftEntry *pMsg6 = step6(pMsg5, len); syncEntryPrint2((char *)"==step6==", pMsg6); // step7 SRpcMsg *pMsg7 = step7(pMsg6); syncRpcMsgPrint2((char *)"==step7==", pMsg7); return 0; }