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-07 16:17:41 +08:00
parent f76ca26cfb
commit fa8284af73
7 changed files with 131 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/libs/sync/inc/syncEnv.h
View File

@ -46,11 +46,8 @@ typedef struct SSyncEnv {
extern SSyncEnv* gSyncEnv;
int32_t syncEnvStart();
int32_t syncEnvStop();
tmr_h syncEnvStartTimer(TAOS_TMR_CALLBACK fp, int mseconds, void* param);
void syncEnvStopTimer(tmr_h* pTimer);
#ifdef __cplusplus

10
source/libs/sync/inc/syncRaftStore.h
View File

@ -34,25 +34,15 @@ extern "C" {
typedef struct SRaftStore {
SyncTerm currentTerm;
SRaftId voteFor;
// FileFd fd;
TdFilePtr pFile;
char path[RAFT_STORE_PATH_LEN];
} SRaftStore;
SRaftStore *raftStoreOpen(const char *path);
static int32_t raftStoreInit(SRaftStore *pRaftStore);
int32_t raftStoreClose(SRaftStore *pRaftStore);
int32_t raftStorePersist(SRaftStore *pRaftStore);
static bool raftStoreFileExist(char *path);
int32_t raftStoreSerialize(SRaftStore *pRaftStore, char *buf, size_t len);
int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len);
void raftStorePrint(SRaftStore *pRaftStore);
#ifdef __cplusplus

3
source/libs/sync/src/syncAppendEntries.c
View File

@ -15,7 +15,6 @@
#include "syncAppendEntries.h"
int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
// TLA+ Spec
// HandleAppendEntriesRequest(i, j, m) ==
// LET logOk == \/ m.mprevLogIndex = 0
@ -81,4 +80,4 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
// /\ UNCHANGED <<serverVars, commitIndex, messages>>
// /\ UNCHANGED <<candidateVars, leaderVars>>
//
}
int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {}

3
source/libs/sync/src/syncAppendEntriesReply.c
View File

@ -15,7 +15,6 @@
#include "syncAppendEntriesReply.h"
int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg) {
// TLA+ Spec
// HandleAppendEntriesResponse(i, j, m) ==
// /\ m.mterm = currentTerm[i]
@ -28,4 +27,4 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
// /\ UNCHANGED <<matchIndex>>
// /\ Discard(m)
// /\ UNCHANGED <<serverVars, candidateVars, logVars, elections>>
}
int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg) {}

123
source/libs/sync/src/syncRaftStore.c
View File

@ -16,8 +16,11 @@
#include "syncRaftStore.h"
#include "cJSON.h"
// to complie success: FileIO interface is modified
// private function
static int32_t raftStoreInit(SRaftStore *pRaftStore);
static bool raftStoreFileExist(char *path);
// public function
SRaftStore *raftStoreOpen(const char *path) {
int32_t ret;
@ -137,121 +140,3 @@ void raftStorePrint(SRaftStore *pRaftStore) {
raftStoreSerialize(pRaftStore, storeBuf, sizeof(storeBuf));
printf("%s\n", storeBuf);
}
#if 0
SRaftStore *raftStoreOpen(const char *path) {
int32_t ret;
SRaftStore *pRaftStore = malloc(sizeof(SRaftStore));
if (pRaftStore == NULL) {
sError("raftStoreOpen malloc error");
return NULL;
}
memset(pRaftStore, 0, sizeof(*pRaftStore));
snprintf(pRaftStore->path, sizeof(pRaftStore->path), "%s", path);
char storeBuf[RAFT_STORE_BLOCK_SIZE];
memset(storeBuf, 0, sizeof(storeBuf));
if (!raftStoreFileExist(pRaftStore->path)) {
ret = raftStoreInit(pRaftStore);
assert(ret == 0);
}
pRaftStore->fd = taosOpenFileReadWrite(pRaftStore->path);
if (pRaftStore->fd < 0) {
return NULL;
}
int len = taosReadFile(pRaftStore->fd, storeBuf, sizeof(storeBuf));
assert(len == RAFT_STORE_BLOCK_SIZE);
ret = raftStoreDeserialize(pRaftStore, storeBuf, len);
assert(ret == 0);
return pRaftStore;
}
static int32_t raftStoreInit(SRaftStore *pRaftStore) {
pRaftStore->fd = taosOpenFileCreateWrite(pRaftStore->path);
if (pRaftStore->fd < 0) {
return -1;
}
pRaftStore->currentTerm = 0;
pRaftStore->voteFor.addr = 0;
pRaftStore->voteFor.vgId = 0;
int32_t ret = raftStorePersist(pRaftStore);
assert(ret == 0);
taosCloseFile(pRaftStore->fd);
return 0;
}
int32_t raftStoreClose(SRaftStore *pRaftStore) {
taosCloseFile(pRaftStore->fd);
free(pRaftStore);
return 0;
}
int32_t raftStorePersist(SRaftStore *pRaftStore) {
int32_t ret;
char storeBuf[RAFT_STORE_BLOCK_SIZE];
ret = raftStoreSerialize(pRaftStore, storeBuf, sizeof(storeBuf));
assert(ret == 0);
taosLSeekFile(pRaftStore->fd, 0, SEEK_SET);
ret = taosWriteFile(pRaftStore->fd, storeBuf, sizeof(storeBuf));
assert(ret == RAFT_STORE_BLOCK_SIZE);
fsync(pRaftStore->fd);
return 0;
}
static bool raftStoreFileExist(char *path) { return taosStatFile(path, NULL, NULL) >= 0; }
int32_t raftStoreSerialize(SRaftStore *pRaftStore, char *buf, size_t len) {
cJSON *pRoot = cJSON_CreateObject();
cJSON_AddNumberToObject(pRoot, "current_term", pRaftStore->currentTerm);
cJSON_AddNumberToObject(pRoot, "vote_for_addr", pRaftStore->voteFor.addr);
cJSON_AddNumberToObject(pRoot, "vote_for_vgid", pRaftStore->voteFor.vgId);
char *serialized = cJSON_Print(pRoot);
int len2 = strlen(serialized);
assert(len2 < len);
memset(buf, 0, len);
snprintf(buf, len, "%s", serialized);
free(serialized);
cJSON_Delete(pRoot);
return 0;
}
int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len) {
assert(len > 0 && len <= RAFT_STORE_BLOCK_SIZE);
cJSON *pRoot = cJSON_Parse(buf);
cJSON *pCurrentTerm = cJSON_GetObjectItem(pRoot, "current_term");
pRaftStore->currentTerm = pCurrentTerm->valueint;
cJSON *pVoteForAddr = cJSON_GetObjectItem(pRoot, "vote_for_addr");
pRaftStore->voteFor.addr = pVoteForAddr->valueint;
cJSON *pVoteForVgid = cJSON_GetObjectItem(pRoot, "vote_for_vgid");
pRaftStore->voteFor.vgId = pVoteForVgid->valueint;
cJSON_Delete(pRoot);
return 0;
}
void raftStorePrint(SRaftStore *pRaftStore) {
char storeBuf[RAFT_STORE_BLOCK_SIZE];
raftStoreSerialize(pRaftStore, storeBuf, sizeof(storeBuf));
printf("%s\n", storeBuf);
}
#endif

3
source/libs/sync/src/syncRequestVote.c
View File

@ -15,7 +15,6 @@
#include "syncRequestVote.h"
int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
// TLA+ Spec
// HandleRequestVoteRequest(i, j, m) ==
// LET logOk == \/ m.mlastLogTerm > LastTerm(log[i])
@ -37,4 +36,4 @@ int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
// mdest |-> j],
// m)
// /\ UNCHANGED <<state, currentTerm, candidateVars, leaderVars, logVars>>
}
int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {}

3
source/libs/sync/src/syncRequestVoteReply.c
View File

@ -15,7 +15,6 @@
#include "syncRequestVoteReply.h"
int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) {
// TLA+ Spec
// HandleRequestVoteResponse(i, j, m) ==
// \* This tallies votes even when the current state is not Candidate, but
@ -32,4 +31,4 @@ int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg)
// /\ UNCHANGED <<votesGranted, voterLog>>
// /\ Discard(m)
// /\ UNCHANGED <<serverVars, votedFor, leaderVars, logVars>>
}
int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) {}