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-09 10:15:40 +08:00
parent 4118ddb9eb
commit 3b4149ba32
4 changed files with 221 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/libs/sync/inc/syncVoteMgr.h
View File

@ -31,8 +31,8 @@ extern "C" {
// SVotesGranted -----------------------------
typedef struct SVotesGranted {
SRaftId (*replicas)[TSDB_MAX_REPLICA];
int32_t replicaNum;
bool isGranted[TSDB_MAX_REPLICA];
int32_t replicaNum;
int32_t votes;
SyncTerm term;
int32_t quorum;
@ -61,7 +61,9 @@ SVotesRespond *votesRespondCreate(SSyncNode *pSyncNode);
void votesRespondDestory(SVotesRespond *pVotesRespond);
bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId);
void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg);
void Reset(SVotesRespond *pVotesRespond, SyncTerm term);
void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term);
cJSON * votesRespond2Json(SVotesRespond *pVotesRespond);
char * votesRespond2Str(SVotesRespond *pVotesRespond);
#ifdef __cplusplus
}

55
source/libs/sync/src/syncVoteMgr.c
View File

@ -153,8 +153,8 @@ bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId) {
void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg) {
assert(pVotesRespond->term == pMsg->term);
for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
if (syncUtilSameId(&(*pVotesRespond->replicas)[i], &pMsg->srcId)) {
assert(pVotesRespond->isRespond[i] == false);
if (syncUtilSameId(&((*(pVotesRespond->replicas))[i]), &pMsg->srcId)) {
// assert(pVotesRespond->isRespond[i] == false);
pVotesRespond->isRespond[i] = true;
return;
}
@ -162,9 +162,52 @@ void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *p
assert(0);
}
void Reset(SVotesRespond *pVotesRespond, SyncTerm term) {
void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term) {
pVotesRespond->term = term;
for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
pVotesRespond->isRespond[i] = false;
}
memset(pVotesRespond->isRespond, 0, sizeof(pVotesRespond->isRespond));
/*
for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
pVotesRespond->isRespond[i] = false;
}
*/
}
cJSON *votesRespond2Json(SVotesRespond *pVotesRespond) {
char u64buf[128];
cJSON *pRoot = cJSON_CreateObject();
cJSON_AddNumberToObject(pRoot, "replicaNum", pVotesRespond->replicaNum);
cJSON *pReplicas = cJSON_CreateArray();
cJSON_AddItemToObject(pRoot, "replicas", pReplicas);
for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pVotesRespond->replicas))[i]));
}
int respondNum = 0;
int *arr = (int *)malloc(sizeof(int) * pVotesRespond->replicaNum);
for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
arr[i] = pVotesRespond->isRespond[i];
if (pVotesRespond->isRespond[i]) {
respondNum++;
}
}
cJSON *pIsRespond = cJSON_CreateIntArray(arr, pVotesRespond->replicaNum);
free(arr);
cJSON_AddItemToObject(pRoot, "isRespond", pIsRespond);
cJSON_AddNumberToObject(pRoot, "respondNum", respondNum);
snprintf(u64buf, sizeof(u64buf), "%lu", pVotesRespond->term);
cJSON_AddStringToObject(pRoot, "term", u64buf);
snprintf(u64buf, sizeof(u64buf), "%p", pVotesRespond->pSyncNode);
cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
cJSON *pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SVotesRespond", pRoot);
return pJson;
}
char *votesRespond2Str(SVotesRespond *pVotesRespond) {
cJSON *pJson = votesRespond2Json(pVotesRespond);
char * serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}

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

@ -13,6 +13,7 @@ add_executable(syncIndexTest "")
add_executable(syncInitTest "")
add_executable(syncUtilTest "")
add_executable(syncVotesGrantedTest "")
add_executable(syncVotesRespondTest "")
target_sources(syncTest
@ -75,6 +76,10 @@ target_sources(syncVotesGrantedTest
PRIVATE
"syncVotesGrantedTest.cpp"
)
target_sources(syncVotesRespondTest
PRIVATE
"syncVotesRespondTest.cpp"
)
target_include_directories(syncTest
@ -152,6 +157,11 @@ target_include_directories(syncVotesGrantedTest
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_include_directories(syncVotesRespondTest
PUBLIC
"${CMAKE_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_link_libraries(syncTest
@ -214,6 +224,10 @@ target_link_libraries(syncVotesGrantedTest
sync
gtest_main
)
target_link_libraries(syncVotesRespondTest
sync
gtest_main
)
enable_testing()

156
source/libs/sync/test/syncVotesRespondTest.cpp Normal file
View File

@ -0,0 +1,156 @@
#include <gtest/gtest.h>
#include <stdio.h>
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncRaftStore.h"
#include "syncUtil.h"
#include "syncVoteMgr.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* 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", "./test_path");
snprintf(syncInfo.walPath, sizeof(syncInfo.walPath), "%s", "./test_wal_path");
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->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);
char* serialized = syncNode2Str(pSyncNode);
printf("%s\n", serialized);
free(serialized);
initRaftId(pSyncNode);
SVotesRespond* pVotesRespond = votesRespondCreate(pSyncNode);
assert(pVotesRespond != NULL);
printf("---------------------------------------\n");
{
char* serialized = votesRespond2Str(pVotesRespond);
assert(serialized != NULL);
printf("%s\n", serialized);
free(serialized);
}
SyncTerm term = 1234;
printf("---------------------------------------\n");
votesRespondReset(pVotesRespond, term);
{
char* serialized = votesRespond2Str(pVotesRespond);
assert(serialized != NULL);
printf("%s\n", serialized);
free(serialized);
}
for (int i = 0; i < replicaNum; ++i) {
SyncRequestVoteReply* reply = SyncRequestVoteReplyBuild();
reply->destId = pSyncNode->myRaftId;
reply->srcId = ids[i];
reply->term = term;
reply->voteGranted = true;
votesRespondAdd(pVotesRespond, reply);
{
char* serialized = votesRespond2Str(pVotesRespond);
assert(serialized != NULL);
printf("%s\n", serialized);
free(serialized);
}
votesRespondAdd(pVotesRespond, reply);
{
char* serialized = votesRespond2Str(pVotesRespond);
assert(serialized != NULL);
printf("%s\n", serialized);
free(serialized);
}
}
printf("---------------------------------------\n");
votesRespondReset(pVotesRespond, 123456789);
{
char* serialized = votesRespond2Str(pVotesRespond);
assert(serialized != NULL);
printf("%s\n", serialized);
free(serialized);
}
votesRespondDestory(pVotesRespond);
return 0;
}