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

sync refactor

This commit is contained in:
Minghao Li 2022-03-11 19:11:38 +08:00
parent 78b5da0e0f
commit 690dabd617
9 changed files with 516 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
source/libs/sync/inc/syncMessage.h
View File

@ -94,14 +94,24 @@ typedef struct SyncPing {
} SyncPing;
SyncPing* syncPingBuild(uint32_t dataLen);
SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str);
SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId);
void syncPingDestroy(SyncPing* pMsg);
void syncPingSerialize(const SyncPing* pMsg, char* buf, uint32_t bufLen);
void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg);
char* syncPingSerialize2(const SyncPing* pMsg, uint32_t* len);
SyncPing* syncPingDeserialize2(const char* buf, uint32_t len);
void syncPing2RpcMsg(const SyncPing* pMsg, SRpcMsg* pRpcMsg);
void syncPingFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPing* pMsg);
SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg);
cJSON* syncPing2Json(const SyncPing* pMsg);
SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str);
SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId);
char* syncPing2Str(const SyncPing* pMsg);
// for debug ----------------------
void syncPingPrint(const SyncPing* pMsg);
void syncPingPrint2(char* s, const SyncPing* pMsg);
void syncPingLog(const SyncPing* pMsg);
void syncPingLog2(char* s, const SyncPing* pMsg);
// ---------------------------------------------
typedef struct SyncPingReply {
@ -114,18 +124,25 @@ typedef struct SyncPingReply {
char data[];
} SyncPingReply;
#define SYNC_PING_REPLY_FIX_LEN \
(sizeof(uint32_t) + sizeof(uint32_t) + sizeof(SRaftId) + sizeof(SRaftId) + sizeof(uint32_t))
SyncPingReply* syncPingReplyBuild(uint32_t dataLen);
SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str);
SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId);
void syncPingReplyDestroy(SyncPingReply* pMsg);
void syncPingReplySerialize(const SyncPingReply* pMsg, char* buf, uint32_t bufLen);
void syncPingReplyDeserialize(const char* buf, uint32_t len, SyncPingReply* pMsg);
char* syncPingReplySerialize2(const SyncPingReply* pMsg, uint32_t* len); //
SyncPingReply* syncPingReplyDeserialize2(const char* buf, uint32_t len); //
void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg);
void syncPingReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPingReply* pMsg);
SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg); //
cJSON* syncPingReply2Json(const SyncPingReply* pMsg);
SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str);
SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId);
char* syncPingReply2Str(const SyncPingReply* pMsg); //
// for debug ----------------------
void syncPingReplyPrint(const SyncPingReply* pMsg);
void syncPingReplyPrint2(char* s, const SyncPingReply* pMsg);
void syncPingReplyLog(const SyncPingReply* pMsg);
void syncPingReplyLog2(char* s, const SyncPingReply* pMsg);
// ---------------------------------------------
typedef struct SyncClientRequest {

184
source/libs/sync/src/syncMessage.c
View File

@ -217,6 +217,20 @@ SyncPing* syncPingBuild(uint32_t dataLen) {
return pMsg;
}
SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str) {
uint32_t dataLen = strlen(str) + 1;
SyncPing* pMsg = syncPingBuild(dataLen);
pMsg->srcId = *srcId;
pMsg->destId = *destId;
snprintf(pMsg->data, pMsg->dataLen, "%s", str);
return pMsg;
}
SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId) {
SyncPing* pMsg = syncPingBuild2(srcId, destId, "ping");
return pMsg;
}
void syncPingDestroy(SyncPing* pMsg) {
if (pMsg != NULL) {
free(pMsg);
@ -234,6 +248,25 @@ void syncPingDeserialize(const char* buf, uint32_t len, SyncPing* pMsg) {
assert(pMsg->bytes == sizeof(SyncPing) + pMsg->dataLen);
}
char* syncPingSerialize2(const SyncPing* pMsg, uint32_t* len) {
char* buf = malloc(pMsg->bytes);
assert(buf != NULL);
syncPingSerialize(pMsg, buf, pMsg->bytes);
if (len != NULL) {
*len = pMsg->bytes;
}
return buf;
}
SyncPing* syncPingDeserialize2(const char* buf, uint32_t len) {
uint32_t bytes = *((uint32_t*)buf);
SyncPing* pMsg = malloc(bytes);
assert(pMsg != NULL);
syncPingDeserialize(buf, len, pMsg);
assert(len == pMsg->bytes);
return pMsg;
}
void syncPing2RpcMsg(const SyncPing* pMsg, SRpcMsg* pRpcMsg) {
memset(pRpcMsg, 0, sizeof(*pRpcMsg));
pRpcMsg->msgType = pMsg->msgType;
@ -246,6 +279,11 @@ void syncPingFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPing* pMsg) {
syncPingDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}
SyncPing* syncPingFromRpcMsg2(const SRpcMsg* pRpcMsg) {
SyncPing* pMsg = syncPingDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
return pMsg;
}
cJSON* syncPing2Json(const SyncPing* pMsg) {
char u64buf[128];
@ -284,35 +322,75 @@ cJSON* syncPing2Json(const SyncPing* pMsg) {
cJSON_AddItemToObject(pRoot, "destId", pDestId);
cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", pMsg->data);
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
free(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
free(s);
cJSON* pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SyncPing", pRoot);
return pJson;
}
SyncPing* syncPingBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str) {
uint32_t dataLen = strlen(str) + 1;
SyncPing* pMsg = syncPingBuild(dataLen);
char* syncPing2Str(const SyncPing* pMsg) {
cJSON* pJson = syncPing2Json(pMsg);
char* serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
// for debug ----------------------
void syncPingPrint(const SyncPing* pMsg) {
char* serialized = syncPing2Str(pMsg);
printf("syncPingPrint | len:%lu | %s \n", strlen(serialized), serialized);
fflush(NULL);
free(serialized);
}
void syncPingPrint2(char* s, const SyncPing* pMsg) {
char* serialized = syncPing2Str(pMsg);
printf("syncPingPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
fflush(NULL);
free(serialized);
}
void syncPingLog(const SyncPing* pMsg) {
char* serialized = syncPing2Str(pMsg);
sTrace("syncPingLog | len:%lu | %s", strlen(serialized), serialized);
free(serialized);
}
void syncPingLog2(char* s, const SyncPing* pMsg) {
char* serialized = syncPing2Str(pMsg);
sTrace("syncPingLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
free(serialized);
}
// ---- message process SyncPingReply----
SyncPingReply* syncPingReplyBuild(uint32_t dataLen) {
uint32_t bytes = sizeof(SyncPingReply) + dataLen;
SyncPingReply* pMsg = malloc(bytes);
memset(pMsg, 0, bytes);
pMsg->bytes = bytes;
pMsg->msgType = SYNC_PING_REPLY;
pMsg->dataLen = dataLen;
return pMsg;
}
SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str) {
uint32_t dataLen = strlen(str) + 1;
SyncPingReply* pMsg = syncPingReplyBuild(dataLen);
pMsg->srcId = *srcId;
pMsg->destId = *destId;
snprintf(pMsg->data, pMsg->dataLen, "%s", str);
return pMsg;
}
SyncPing* syncPingBuild3(const SRaftId* srcId, const SRaftId* destId) {
SyncPing* pMsg = syncPingBuild2(srcId, destId, "ping");
return pMsg;
}
// ---- message process SyncPingReply----
SyncPingReply* syncPingReplyBuild(uint32_t dataLen) {
uint32_t bytes = SYNC_PING_REPLY_FIX_LEN + dataLen;
SyncPingReply* pMsg = malloc(bytes);
memset(pMsg, 0, bytes);
pMsg->bytes = bytes;
pMsg->msgType = SYNC_PING_REPLY;
pMsg->dataLen = dataLen;
SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId) {
SyncPingReply* pMsg = syncPingReplyBuild2(srcId, destId, "pang");
return pMsg;
}
@ -333,6 +411,25 @@ void syncPingReplyDeserialize(const char* buf, uint32_t len, SyncPingReply* pMsg
assert(pMsg->bytes == sizeof(SyncPing) + pMsg->dataLen);
}
char* syncPingReplySerialize2(const SyncPingReply* pMsg, uint32_t* len) {
char* buf = malloc(pMsg->bytes);
assert(buf != NULL);
syncPingReplySerialize(pMsg, buf, pMsg->bytes);
if (len != NULL) {
*len = pMsg->bytes;
}
return buf;
}
SyncPingReply* syncPingReplyDeserialize2(const char* buf, uint32_t len) {
uint32_t bytes = *((uint32_t*)buf);
SyncPingReply* pMsg = malloc(bytes);
assert(pMsg != NULL);
syncPingReplyDeserialize(buf, len, pMsg);
assert(len == pMsg->bytes);
return pMsg;
}
void syncPingReply2RpcMsg(const SyncPingReply* pMsg, SRpcMsg* pRpcMsg) {
memset(pRpcMsg, 0, sizeof(*pRpcMsg));
pRpcMsg->msgType = pMsg->msgType;
@ -345,6 +442,11 @@ void syncPingReplyFromRpcMsg(const SRpcMsg* pRpcMsg, SyncPingReply* pMsg) {
syncPingReplyDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}
SyncPingReply* syncPingReplyFromRpcMsg2(const SRpcMsg* pRpcMsg) {
SyncPingReply* pMsg = syncPingReplyDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
return pMsg;
}
cJSON* syncPingReply2Json(const SyncPingReply* pMsg) {
char u64buf[128];
@ -383,25 +485,51 @@ cJSON* syncPingReply2Json(const SyncPingReply* pMsg) {
cJSON_AddItemToObject(pRoot, "destId", pDestId);
cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", pMsg->data);
char* s;
s = syncUtilprintBin((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data", s);
free(s);
s = syncUtilprintBin2((char*)(pMsg->data), pMsg->dataLen);
cJSON_AddStringToObject(pRoot, "data2", s);
free(s);
cJSON* pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SyncPingReply", pRoot);
return pJson;
}
SyncPingReply* syncPingReplyBuild2(const SRaftId* srcId, const SRaftId* destId, const char* str) {
uint32_t dataLen = strlen(str) + 1;
SyncPingReply* pMsg = syncPingReplyBuild(dataLen);
pMsg->srcId = *srcId;
pMsg->destId = *destId;
snprintf(pMsg->data, pMsg->dataLen, "%s", str);
return pMsg;
char* syncPingReply2Str(const SyncPingReply* pMsg) {
cJSON* pJson = syncPingReply2Json(pMsg);
char* serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
SyncPingReply* syncPingReplyBuild3(const SRaftId* srcId, const SRaftId* destId) {
SyncPingReply* pMsg = syncPingReplyBuild2(srcId, destId, "pang");
return pMsg;
// for debug ----------------------
void syncPingReplyPrint(const SyncPingReply* pMsg) {
char* serialized = syncPingReply2Str(pMsg);
printf("syncPingReplyPrint | len:%lu | %s \n", strlen(serialized), serialized);
fflush(NULL);
free(serialized);
}
void syncPingReplyPrint2(char* s, const SyncPingReply* pMsg) {
char* serialized = syncPingReply2Str(pMsg);
printf("syncPingReplyPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
fflush(NULL);
free(serialized);
}
void syncPingReplyLog(const SyncPingReply* pMsg) {
char* serialized = syncPingReply2Str(pMsg);
sTrace("syncPingReplyLog | len:%lu | %s", strlen(serialized), serialized);
free(serialized);
}
void syncPingReplyLog2(char* s, const SyncPingReply* pMsg) {
char* serialized = syncPingReply2Str(pMsg);
sTrace("syncPingReplyLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
free(serialized);
}
// ---- message process SyncClientRequest----

38
source/libs/sync/test/CMakeLists.txt
View File

@ -1,6 +1,6 @@
add_executable(syncTest "")
add_executable(syncEnvTest "")
add_executable(syncPingTest "")
add_executable(syncPingTimerTest "")
add_executable(syncEncodeTest "")
add_executable(syncIOTickQTest "")
add_executable(syncIOTickPingTest "")
@ -23,6 +23,8 @@ add_executable(syncAppendEntriesTest "")
add_executable(syncAppendEntriesReplyTest "")
add_executable(syncClientRequestTest "")
add_executable(syncTimeoutTest "")
add_executable(syncPingTest "")
add_executable(syncPingReplyTest "")
target_sources(syncTest
@ -33,9 +35,9 @@ target_sources(syncEnvTest
PRIVATE
"syncEnvTest.cpp"
)
target_sources(syncPingTest
target_sources(syncPingTimerTest
PRIVATE
"syncPingTest.cpp"
"syncPingTimerTest.cpp"
)
target_sources(syncEncodeTest
PRIVATE
@ -125,6 +127,14 @@ target_sources(syncTimeoutTest
PRIVATE
"syncTimeoutTest.cpp"
)
target_sources(syncPingTest
PRIVATE
"syncPingTest.cpp"
)
target_sources(syncPingReplyTest
PRIVATE
"syncPingReplyTest.cpp"
)
target_include_directories(syncTest
@ -137,7 +147,7 @@ target_include_directories(syncEnvTest
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_include_directories(syncPingTest
target_include_directories(syncPingTimerTest
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
@ -252,6 +262,16 @@ target_include_directories(syncTimeoutTest
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_include_directories(syncPingTest
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_include_directories(syncPingReplyTest
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_link_libraries(syncTest
@ -262,7 +282,7 @@ target_link_libraries(syncEnvTest
sync
gtest_main
)
target_link_libraries(syncPingTest
target_link_libraries(syncPingTimerTest
sync
gtest_main
)
@ -354,6 +374,14 @@ target_link_libraries(syncTimeoutTest
sync
gtest_main
)
target_link_libraries(syncPingTest
sync
gtest_main
)
target_link_libraries(syncPingReplyTest
sync
gtest_main
)
enable_testing()

2
source/libs/sync/test/syncAppendEntriesTest.cpp
View File

@ -38,7 +38,7 @@ void test2() {
uint32_t len = pMsg->bytes;
char * serialized = (char *)malloc(len);
syncAppendEntriesSerialize(pMsg, serialized, len);
SyncAppendEntries *pMsg2 = syncAppendEntriesBuild(pMsg2->dataLen);
SyncAppendEntries *pMsg2 = syncAppendEntriesBuild(pMsg->dataLen);
syncAppendEntriesDeserialize(serialized, len, pMsg2);
syncAppendEntriesPrint2((char *)"test2: syncAppendEntriesSerialize -> syncAppendEntriesDeserialize ", pMsg2);

4
source/libs/sync/test/syncClientRequestTest.cpp
View File

@ -20,7 +20,7 @@ SyncClientRequest *createMsg() {
rpcMsg.msgType = 12345;
rpcMsg.contLen = 20;
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
strcpy((char*)rpcMsg.pCont, "hello rpc");
strcpy((char *)rpcMsg.pCont, "hello rpc");
SyncClientRequest *pMsg = syncClientRequestBuild2(&rpcMsg, 123, true);
return pMsg;
}
@ -36,7 +36,7 @@ void test2() {
uint32_t len = pMsg->bytes;
char * serialized = (char *)malloc(len);
syncClientRequestSerialize(pMsg, serialized, len);
SyncClientRequest *pMsg2 = syncClientRequestBuild(pMsg2->dataLen);
SyncClientRequest *pMsg2 = syncClientRequestBuild(pMsg->dataLen);
syncClientRequestDeserialize(serialized, len, pMsg2);
syncClientRequestPrint2((char *)"test2: syncClientRequestSerialize -> syncClientRequestDeserialize ", pMsg2);

95
source/libs/sync/test/syncPingReplyTest.cpp Normal file
View File

@ -0,0 +1,95 @@
#include <gtest/gtest.h>
#include <stdio.h>
#include "syncIO.h"
#include "syncInt.h"
#include "syncMessage.h"
#include "syncUtil.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");
}
SyncPingReply *createMsg() {
SRaftId srcId, destId;
srcId.addr = syncUtilAddr2U64("127.0.0.1", 1234);
srcId.vgId = 100;
destId.addr = syncUtilAddr2U64("127.0.0.1", 5678);
destId.vgId = 100;
SyncPingReply *pMsg = syncPingReplyBuild3(&srcId, &destId);
return pMsg;
}
void test1() {
SyncPingReply *pMsg = createMsg();
syncPingReplyPrint2((char *)"test1:", pMsg);
syncPingReplyDestroy(pMsg);
}
void test2() {
SyncPingReply *pMsg = createMsg();
uint32_t len = pMsg->bytes;
char * serialized = (char *)malloc(len);
syncPingReplySerialize(pMsg, serialized, len);
SyncPingReply *pMsg2 = syncPingReplyBuild(pMsg->dataLen);
syncPingReplyDeserialize(serialized, len, pMsg2);
syncPingReplyPrint2((char *)"test2: syncPingReplySerialize -> syncPingReplyDeserialize ", pMsg2);
free(serialized);
syncPingReplyDestroy(pMsg);
syncPingReplyDestroy(pMsg2);
}
void test3() {
SyncPingReply *pMsg = createMsg();
uint32_t len;
char * serialized = syncPingReplySerialize2(pMsg, &len);
SyncPingReply *pMsg2 = syncPingReplyDeserialize2(serialized, len);
syncPingReplyPrint2((char *)"test3: syncPingReplySerialize3 -> syncPingReplyDeserialize2 ", pMsg2);
free(serialized);
syncPingReplyDestroy(pMsg);
syncPingReplyDestroy(pMsg2);
}
void test4() {
SyncPingReply *pMsg = createMsg();
SRpcMsg rpcMsg;
syncPingReply2RpcMsg(pMsg, &rpcMsg);
SyncPingReply *pMsg2 = (SyncPingReply *)malloc(rpcMsg.contLen);
syncPingReplyFromRpcMsg(&rpcMsg, pMsg2);
syncPingReplyPrint2((char *)"test4: syncPingReply2RpcMsg -> syncPingReplyFromRpcMsg ", pMsg2);
syncPingReplyDestroy(pMsg);
syncPingReplyDestroy(pMsg2);
}
void test5() {
SyncPingReply *pMsg = createMsg();
SRpcMsg rpcMsg;
syncPingReply2RpcMsg(pMsg, &rpcMsg);
SyncPingReply *pMsg2 = syncPingReplyFromRpcMsg2(&rpcMsg);
syncPingReplyPrint2((char *)"test5: syncPingReply2RpcMsg -> syncPingReplyFromRpcMsg2 ", pMsg2);
syncPingReplyDestroy(pMsg);
syncPingReplyDestroy(pMsg2);
}
int main() {
// taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog = 0;
sDebugFlag = 143 + 64;
logTest();
test1();
test2();
test3();
test4();
test5();
return 0;
}

173
source/libs/sync/test/syncPingTest.cpp
View File

@ -1,9 +1,8 @@
#include <gtest/gtest.h>
#include <stdio.h>
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncRaftStore.h"
#include "syncMessage.h"
#include "syncUtil.h"
void logTest() {
@ -15,116 +14,82 @@ void logTest() {
sFatal("--- sync log test: fatal");
}
uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
int32_t replicaNum = 3;
int32_t myIndex = 0;
SRaftId ids[TSDB_MAX_REPLICA];
SSyncInfo syncInfo;
SSyncFSM* pFsm;
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", "./");
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);
}
SSyncNode* 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->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
gSyncIO->pSyncNode = pSyncNode;
return pSyncNode;
SyncPing *createMsg() {
SRaftId srcId, destId;
srcId.addr = syncUtilAddr2U64("127.0.0.1", 1234);
srcId.vgId = 100;
destId.addr = syncUtilAddr2U64("127.0.0.1", 5678);
destId.vgId = 100;
SyncPing *pMsg = syncPingBuild3(&srcId, &destId);
return pMsg;
}
SSyncNode* syncInitTest() { return syncNodeInit(); }
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);
}
void test1() {
SyncPing *pMsg = createMsg();
syncPingPrint2((char *)"test1:", pMsg);
syncPingDestroy(pMsg);
}
int main(int argc, char** argv) {
void test2() {
SyncPing *pMsg = createMsg();
uint32_t len = pMsg->bytes;
char * serialized = (char *)malloc(len);
syncPingSerialize(pMsg, serialized, len);
SyncPing *pMsg2 = syncPingBuild(pMsg->dataLen);
syncPingDeserialize(serialized, len, pMsg2);
syncPingPrint2((char *)"test2: syncPingSerialize -> syncPingDeserialize ", pMsg2);
free(serialized);
syncPingDestroy(pMsg);
syncPingDestroy(pMsg2);
}
void test3() {
SyncPing *pMsg = createMsg();
uint32_t len;
char * serialized = syncPingSerialize2(pMsg, &len);
SyncPing *pMsg2 = syncPingDeserialize2(serialized, len);
syncPingPrint2((char *)"test3: syncPingSerialize3 -> syncPingDeserialize2 ", pMsg2);
free(serialized);
syncPingDestroy(pMsg);
syncPingDestroy(pMsg2);
}
void test4() {
SyncPing *pMsg = createMsg();
SRpcMsg rpcMsg;
syncPing2RpcMsg(pMsg, &rpcMsg);
SyncPing *pMsg2 = (SyncPing *)malloc(rpcMsg.contLen);
syncPingFromRpcMsg(&rpcMsg, pMsg2);
syncPingPrint2((char *)"test4: syncPing2RpcMsg -> syncPingFromRpcMsg ", pMsg2);
syncPingDestroy(pMsg);
syncPingDestroy(pMsg2);
}
void test5() {
SyncPing *pMsg = createMsg();
SRpcMsg rpcMsg;
syncPing2RpcMsg(pMsg, &rpcMsg);
SyncPing *pMsg2 = syncPingFromRpcMsg2(&rpcMsg);
syncPingPrint2((char *)"test5: syncPing2RpcMsg -> syncPingFromRpcMsg2 ", pMsg2);
syncPingDestroy(pMsg);
syncPingDestroy(pMsg2);
}
int main() {
// taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog = 0;
sDebugFlag = 143 + 64;
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);
SSyncNode* pSyncNode = syncInitTest();
assert(pSyncNode != NULL);
syncNodePrint2((char*)"----1", pSyncNode);
initRaftId(pSyncNode);
//---------------------------
sTrace("syncNodeStartPingTimer ...");
ret = syncNodeStartPingTimer(pSyncNode);
assert(ret == 0);
syncNodePrint2((char*)"----2", pSyncNode);
sTrace("sleep ...");
taosMsleep(10000);
sTrace("syncNodeStopPingTimer ...");
ret = syncNodeStopPingTimer(pSyncNode);
assert(ret == 0);
syncNodePrint2((char*)"----3", pSyncNode);
sTrace("sleep ...");
taosMsleep(5000);
sTrace("syncNodeStartPingTimer ...");
ret = syncNodeStartPingTimer(pSyncNode);
assert(ret == 0);
syncNodePrint2((char*)"----4", pSyncNode);
sTrace("sleep ...");
taosMsleep(10000);
sTrace("syncNodeStopPingTimer ...");
ret = syncNodeStopPingTimer(pSyncNode);
assert(ret == 0);
syncNodePrint2((char*)"----5", pSyncNode);
while (1) {
sTrace("while 1 sleep ...");
taosMsleep(1000);
}
test1();
test2();
test3();
test4();
test5();
return 0;
}

130
source/libs/sync/test/syncPingTimerTest.cpp Normal file
View File

@ -0,0 +1,130 @@
#include <gtest/gtest.h>
#include <stdio.h>
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncRaftStore.h"
#include "syncUtil.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 = 3;
int32_t myIndex = 0;
SRaftId ids[TSDB_MAX_REPLICA];
SSyncInfo syncInfo;
SSyncFSM* pFsm;
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", "./");
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);
}
SSyncNode* 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->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
gSyncIO->pSyncNode = pSyncNode;
return pSyncNode;
}
SSyncNode* syncInitTest() { return syncNodeInit(); }
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);
}
}
int main(int argc, char** argv) {
// taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog = 0;
sDebugFlag = 143 + 64;
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);
SSyncNode* pSyncNode = syncInitTest();
assert(pSyncNode != NULL);
syncNodePrint2((char*)"----1", pSyncNode);
initRaftId(pSyncNode);
//---------------------------
sTrace("syncNodeStartPingTimer ...");
ret = syncNodeStartPingTimer(pSyncNode);
assert(ret == 0);
syncNodePrint2((char*)"----2", pSyncNode);
sTrace("sleep ...");
taosMsleep(10000);
sTrace("syncNodeStopPingTimer ...");
ret = syncNodeStopPingTimer(pSyncNode);
assert(ret == 0);
syncNodePrint2((char*)"----3", pSyncNode);
sTrace("sleep ...");
taosMsleep(5000);
sTrace("syncNodeStartPingTimer ...");
ret = syncNodeStartPingTimer(pSyncNode);
assert(ret == 0);
syncNodePrint2((char*)"----4", pSyncNode);
sTrace("sleep ...");
taosMsleep(10000);
sTrace("syncNodeStopPingTimer ...");
ret = syncNodeStopPingTimer(pSyncNode);
assert(ret == 0);
syncNodePrint2((char*)"----5", pSyncNode);
while (1) {
sTrace("while 1 sleep ...");
taosMsleep(1000);
}
return 0;
}

12
source/libs/sync/test/syncTimeoutTest.cpp
View File

@ -29,8 +29,8 @@ void test1() {
void test2() {
SyncTimeout *pMsg = createMsg();
uint32_t len = pMsg->bytes;
char * serialized = (char *)malloc(len);
uint32_t len = pMsg->bytes;
char * serialized = (char *)malloc(len);
syncTimeoutSerialize(pMsg, serialized, len);
SyncTimeout *pMsg2 = syncTimeoutBuild();
syncTimeoutDeserialize(serialized, len, pMsg2);
@ -43,8 +43,8 @@ void test2() {
void test3() {
SyncTimeout *pMsg = createMsg();
uint32_t len;
char * serialized = syncTimeoutSerialize2(pMsg, &len);
uint32_t len;
char * serialized = syncTimeoutSerialize2(pMsg, &len);
SyncTimeout *pMsg2 = syncTimeoutDeserialize2(serialized, len);
syncTimeoutPrint2((char *)"test3: syncTimeoutSerialize3 -> syncTimeoutDeserialize2 ", pMsg2);
@ -55,7 +55,7 @@ void test3() {
void test4() {
SyncTimeout *pMsg = createMsg();
SRpcMsg rpcMsg;
SRpcMsg rpcMsg;
syncTimeout2RpcMsg(pMsg, &rpcMsg);
SyncTimeout *pMsg2 = (SyncTimeout *)malloc(rpcMsg.contLen);
syncTimeoutFromRpcMsg(&rpcMsg, pMsg2);
@ -67,7 +67,7 @@ void test4() {
void test5() {
SyncTimeout *pMsg = createMsg();
SRpcMsg rpcMsg;
SRpcMsg rpcMsg;
syncTimeout2RpcMsg(pMsg, &rpcMsg);
SyncTimeout *pMsg2 = syncTimeoutFromRpcMsg2(&rpcMsg);
syncTimeoutPrint2((char *)"test5: syncTimeout2RpcMsg -> syncTimeoutFromRpcMsg2 ", pMsg2);