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 10:50:50 +08:00
parent 88aef2d1ab
commit 6a3869a3bc
7 changed files with 185 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -83,8 +83,6 @@ typedef struct SyncPing {
char data[];
} SyncPing;
//#define SYNC_PING_FIX_LEN (sizeof(uint32_t) + sizeof(uint32_t) + sizeof(SRaftId) + sizeof(SRaftId) + sizeof(uint32_t))
SyncPing* syncPingBuild(uint32_t dataLen);
void syncPingDestroy(SyncPing* pMsg);
void syncPingSerialize(const SyncPing* pMsg, char* buf, uint32_t bufLen);
@ -163,9 +161,19 @@ SyncRequestVote* syncRequestVoteBuild();
void syncRequestVoteDestroy(SyncRequestVote* pMsg);
void syncRequestVoteSerialize(const SyncRequestVote* pMsg, char* buf, uint32_t bufLen);
void syncRequestVoteDeserialize(const char* buf, uint32_t len, SyncRequestVote* pMsg);
char* syncRequestVoteSerialize2(const SyncRequestVote* pMsg, uint32_t* len);
SyncRequestVote* syncRequestVoteDeserialize2(const char* buf, uint32_t len);
void syncRequestVote2RpcMsg(const SyncRequestVote* pMsg, SRpcMsg* pRpcMsg);
void syncRequestVoteFromRpcMsg(const SRpcMsg* pRpcMsg, SyncRequestVote* pMsg);
SyncRequestVote* syncRequestVoteFromRpcMsg2(const SRpcMsg* pRpcMsg);
cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg);
char* syncRequestVote2Str(const SyncRequestVote* pMsg);
// for debug ----------------------
void syncRequestVotePrint(const SyncRequestVote* pMsg);
void syncRequestVotePrint2(char* s, const SyncRequestVote* pMsg);
void syncRequestVoteLog(const SyncRequestVote* pMsg);
void syncRequestVoteLog2(char* s, const SyncRequestVote* pMsg);
// ---------------------------------------------
typedef struct SyncRequestVoteReply {

2
source/libs/sync/inc/syncRaftEntry.h
View File

@ -47,7 +47,7 @@ SSyncRaftEntry* syncEntryDeserialize(const char* buf, uint32_t len);
cJSON* syncEntry2Json(const SSyncRaftEntry* pEntry);
char* syncEntry2Str(const SSyncRaftEntry* pEntry);
// for debug
// for debug ----------------------
void syncEntryPrint(const SSyncRaftEntry* pObj);
void syncEntryPrint2(char* s, const SSyncRaftEntry* pObj);
void syncEntryLog(const SSyncRaftEntry* pObj);

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

@ -439,6 +439,25 @@ void syncRequestVoteDeserialize(const char* buf, uint32_t len, SyncRequestVote*
assert(len == pMsg->bytes);
}
char* syncRequestVoteSerialize2(const SyncRequestVote* pMsg, uint32_t* len) {
char* buf = malloc(pMsg->bytes);
assert(buf != NULL);
syncRequestVoteSerialize(pMsg, buf, pMsg->bytes);
if (len != NULL) {
*len = pMsg->bytes;
}
return buf;
}
SyncRequestVote* syncRequestVoteDeserialize2(const char* buf, uint32_t len) {
uint32_t bytes = *((uint32_t*)buf);
SyncRequestVote* pMsg = malloc(bytes);
assert(pMsg != NULL);
syncRequestVoteDeserialize(buf, len, pMsg);
assert(len == pMsg->bytes);
return pMsg;
}
void syncRequestVote2RpcMsg(const SyncRequestVote* pMsg, SRpcMsg* pRpcMsg) {
memset(pRpcMsg, 0, sizeof(*pRpcMsg));
pRpcMsg->msgType = pMsg->msgType;
@ -451,6 +470,10 @@ void syncRequestVoteFromRpcMsg(const SRpcMsg* pRpcMsg, SyncRequestVote* pMsg) {
syncRequestVoteDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
}
SyncRequestVote* syncRequestVoteFromRpcMsg2(const SRpcMsg* pRpcMsg) {
SyncRequestVote* pMsg = syncRequestVoteDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
}
cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) {
char u64buf[128];
@ -499,6 +522,40 @@ cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) {
return pJson;
}
char* syncRequestVote2Str(const SyncRequestVote* pMsg) {
cJSON* pJson = syncRequestVote2Json(pMsg);
char* serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
// for debug ----------------------
void syncRequestVotePrint(const SyncRequestVote* pMsg) {
char* serialized = syncRequestVote2Str(pMsg);
printf("syncRequestVotePrint | len:%lu | %s \n", strlen(serialized), serialized);
fflush(NULL);
free(serialized);
}
void syncRequestVotePrint2(char* s, const SyncRequestVote* pMsg) {
char* serialized = syncRequestVote2Str(pMsg);
printf("syncRequestVotePrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
fflush(NULL);
free(serialized);
}
void syncRequestVoteLog(const SyncRequestVote* pMsg) {
char* serialized = syncRequestVote2Str(pMsg);
sTrace("syncRequestVoteLog | len:%lu | %s", strlen(serialized), serialized);
free(serialized);
}
void syncRequestVoteLog2(char* s, const SyncRequestVote* pMsg) {
char* serialized = syncRequestVote2Str(pMsg);
sTrace("syncRequestVoteLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
free(serialized);
}
// ---- message process SyncRequestVoteReply----
SyncRequestVoteReply* SyncRequestVoteReplyBuild() {
uint32_t bytes = sizeof(SyncRequestVoteReply);

2
source/libs/sync/src/syncRaftEntry.c
View File

@ -104,7 +104,7 @@ char* syncEntry2Str(const SSyncRaftEntry* pEntry) {
return serialized;
}
// for debug -----------
// for debug ----------------------
void syncEntryPrint(const SSyncRaftEntry* pObj) {
char* serialized = syncEntry2Str(pObj);
printf("syncEntryPrint | len:%lu | %s \n", strlen(serialized), serialized);

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

@ -17,6 +17,7 @@ add_executable(syncVotesRespondTest "")
add_executable(syncIndexMgrTest "")
add_executable(syncLogStoreTest "")
add_executable(syncEntryTest "")
add_executable(syncRequestVote "")
target_sources(syncTest
@ -95,6 +96,10 @@ target_sources(syncEntryTest
PRIVATE
"syncEntryTest.cpp"
)
target_sources(syncRequestVote
PRIVATE
"syncRequestVote.cpp"
)
target_include_directories(syncTest
@ -192,6 +197,11 @@ target_include_directories(syncEntryTest
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_include_directories(syncRequestVote
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_link_libraries(syncTest
@ -270,6 +280,10 @@ target_link_libraries(syncEntryTest
sync
gtest_main
)
target_link_libraries(syncRequestVote
sync
gtest_main
)
enable_testing()

6
source/libs/sync/test/syncEncodeTest.cpp
View File

@ -17,6 +17,8 @@ void logTest() {
#define PING_MSG_LEN 20
#define APPEND_ENTRIES_VALUE_LEN 32
#if 0
void test1() {
sTrace("test1: ---- syncPingSerialize, syncPingDeserialize");
@ -487,11 +489,14 @@ void test12() {
syncAppendEntriesReplyDestroy(pMsg2);
}
#endif
int main() {
// taosInitLog((char*)"syncPingTest.log", 100000, 10);
tsAsyncLog = 0;
sDebugFlag = 143 + 64;
#if 0
test1();
test2();
test3();
@ -504,6 +509,7 @@ int main() {
test10();
test11();
test12();
#endif
return 0;
}

96
source/libs/sync/test/syncRequestVote.cpp Normal file
View File

@ -0,0 +1,96 @@
#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");
}
SyncRequestVote *createMsg() {
SyncRequestVote *pMsg = syncRequestVoteBuild();
pMsg->srcId.addr = syncUtilAddr2U64("127.0.0.1", 1234);
pMsg->srcId.vgId = 100;
pMsg->destId.addr = syncUtilAddr2U64("127.0.0.1", 5678);
pMsg->destId.vgId = 100;
pMsg->currentTerm = 11;
pMsg->lastLogIndex = 22;
pMsg->lastLogTerm = 33;
}
void test1() {
SyncRequestVote *pMsg = createMsg();
syncRequestVotePrint2((char *)"test1:", pMsg);
syncRequestVoteDestroy(pMsg);
}
void test2() {
SyncRequestVote *pMsg = createMsg();
uint32_t len = pMsg->bytes;
char * serialized = (char *)malloc(len);
syncRequestVoteSerialize(pMsg, serialized, len);
SyncRequestVote *pMsg2 = syncRequestVoteBuild();
syncRequestVoteDeserialize(serialized, len, pMsg2);
syncRequestVotePrint2((char *)"test2: syncRequestVoteSerialize -> syncRequestVoteDeserialize ", pMsg);
free(serialized);
syncRequestVoteDestroy(pMsg);
syncRequestVoteDestroy(pMsg2);
}
void test3() {
SyncRequestVote *pMsg = createMsg();
uint32_t len;
char * serialized = syncRequestVoteSerialize2(pMsg, &len);
SyncRequestVote *pMsg2 = syncRequestVoteDeserialize2(serialized, len);
syncRequestVotePrint2((char *)"test3: syncRequestVoteSerialize3 -> syncRequestVoteDeserialize2 ", pMsg);
free(serialized);
syncRequestVoteDestroy(pMsg);
syncRequestVoteDestroy(pMsg2);
}
void test4() {
SyncRequestVote *pMsg = createMsg();
SRpcMsg rpcMsg;
syncRequestVote2RpcMsg(pMsg, &rpcMsg);
SyncRequestVote *pMsg2 = syncRequestVoteBuild();
syncRequestVoteFromRpcMsg(&rpcMsg, pMsg2);
syncRequestVotePrint2((char *)"test4: syncRequestVote2RpcMsg -> syncRequestVoteFromRpcMsg ", pMsg);
syncRequestVoteDestroy(pMsg);
syncRequestVoteDestroy(pMsg2);
}
void test5() {
SyncRequestVote *pMsg = createMsg();
SRpcMsg rpcMsg;
syncRequestVote2RpcMsg(pMsg, &rpcMsg);
SyncRequestVote *pMsg2 = syncRequestVoteFromRpcMsg2(&rpcMsg);
syncRequestVotePrint2((char *)"test5: syncRequestVote2RpcMsg -> syncRequestVoteFromRpcMsg2 ", pMsg);
syncRequestVoteDestroy(pMsg);
syncRequestVoteDestroy(pMsg2);
}
int main() {
// taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog = 0;
sDebugFlag = 143 + 64;
logTest();
test1();
test2();
test3();
test4();
test5();
return 0;
}