add raft store

source/libs/sync/inc/syncIO.h

@@ -37,10 +37,10 @@ typedef struct SSyncIO {
   pthread_t   tid;
   int8_t      isStart;
 
-    SEpSet    epSet;
+  SEpSet epSet;
 
   void *syncTimer;
-void *syncTimerManager;
+  void *syncTimerManager;
 
   int32_t (*start)(struct SSyncIO *ths);
   int32_t (*stop)(struct SSyncIO *ths);

source/libs/sync/inc/syncRaft.h

@@ -28,7 +28,7 @@ extern "C" {
 #include "taosdef.h"
 
 typedef struct SRaftId {
-  SyncNodeId  nodeId;
+  SyncNodeId  addr;
   SyncGroupId vgId;
 } SRaftId;
 

source/libs/sync/inc/syncRaftStore.h

@@ -23,20 +23,36 @@ extern "C" {
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
-#include "sync.h"
+#include "cJSON.h"
+#include "syncInt.h"
 #include "syncRaft.h"
 #include "taosdef.h"
 
-void testJson();
+#define RAFT_STORE_BLOCK_SIZE 512
-void testJson2();
+#define RAFT_STORE_PATH_LEN 128
 
-int32_t currentTerm(SyncTerm *pCurrentTerm);
+typedef struct SRaftStore {
+  SyncTerm currentTerm;
+  SRaftId  voteFor;
+  FileFd   fd;
+  char     path[RAFT_STORE_PATH_LEN];
+} SRaftStore;
 
-int32_t persistCurrentTerm(SyncTerm currentTerm);
+SRaftStore *raftStoreOpen(const char *path);
 
-int32_t voteFor(SRaftId *pRaftId);
+static int32_t raftStoreInit(SRaftStore *pRaftStore);
 
-int32_t persistVoteFor(SRaftId *pRaftId);
+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
 }

source/libs/sync/src/syncIO.c

@@ -45,7 +45,7 @@ void *syncConsumer(void *param) {
   SSyncIO *io = param;
 
   STaosQall *qall;
-  SRpcMsg   *pRpcMsg, rpcMsg;
+  SRpcMsg *  pRpcMsg, rpcMsg;
   int        type;
 
   qall = taosAllocateQall();

source/libs/sync/src/syncRaftStore.c

@@ -17,50 +17,116 @@
 #include "cJSON.h"
 #include "sync.h"
 
-FileFd raftStoreFd;
+SRaftStore *raftStoreOpen(const char *path) {
+  int32_t ret;
 
-void testJson() {
+  SRaftStore *pRaftStore = malloc(sizeof(SRaftStore));
-  raftStoreFd = taosOpenFileReadWrite("raft.store");
+  if (pRaftStore == NULL) {
+    sError("raftStoreOpen malloc error");
+    return NULL;
+  }
+  memset(pRaftStore, 0, sizeof(*pRaftStore));
+  snprintf(pRaftStore->path, sizeof(pRaftStore->path), "%s", path);
 
-  uint64_t currentTerm = 100;
+  char storeBuf[RAFT_STORE_BLOCK_SIZE];
-  uint64_t voteFor = 200;
+  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", currentTerm);
+  cJSON_AddNumberToObject(pRoot, "current_term", pRaftStore->currentTerm);
-  cJSON_AddNumberToObject(pRoot, "vote_for", voteFor);
+  cJSON_AddNumberToObject(pRoot, "vote_for_addr", pRaftStore->voteFor.addr);
+  cJSON_AddNumberToObject(pRoot, "vote_for_vgid", pRaftStore->voteFor.vgId);
 
   char *serialized = cJSON_Print(pRoot);
-  int   len = strlen(serialized);
+  int   len2 = strlen(serialized);
-  printf("serialized: %s \n", serialized);
+  assert(len2 < len);
+  memset(buf, 0, len);
+  snprintf(buf, len, "%s", serialized);
+  free(serialized);
 
-  taosWriteFile(raftStoreFd, serialized, len);
+  cJSON_Delete(pRoot);
+  return 0;
 }
 
-void testJson2() {
+int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len) {
-  taosLSeekFile(raftStoreFd, 0, SEEK_SET);
+  assert(len > 0 && len <= RAFT_STORE_BLOCK_SIZE);
-
-  char buf[128];
-  memset(buf, 0, sizeof(buf));
-  taosReadFile(raftStoreFd, buf, sizeof(buf));
-  printf("read file: %s \n", buf);
-
   cJSON *pRoot = cJSON_Parse(buf);
 
-  cJSON   *pCurrentTerm = cJSON_GetObjectItem(pRoot, "current_term");
+  cJSON *pCurrentTerm = cJSON_GetObjectItem(pRoot, "current_term");
-  uint64_t currentTerm = pCurrentTerm->valueint;
+  pRaftStore->currentTerm = pCurrentTerm->valueint;
 
-  cJSON   *pVoteFor = cJSON_GetObjectItem(pRoot, "vote_for");
+  cJSON *pVoteForAddr = cJSON_GetObjectItem(pRoot, "vote_for_addr");
-  uint64_t voteFor = pVoteFor->valueint;
+  pRaftStore->voteFor.addr = pVoteForAddr->valueint;
 
-  printf("read json: currentTerm:%lu, voteFor:%lu \n", currentTerm, voteFor);
+  cJSON *pVoteForVgid = cJSON_GetObjectItem(pRoot, "vote_for_vgid");
+  pRaftStore->voteFor.vgId = pVoteForVgid->valueint;
 
-  taosCloseFile(raftStoreFd);
+  cJSON_Delete(pRoot);
+  return 0;
 }
 
-int32_t currentTerm(SyncTerm *pCurrentTerm) { return 0; }
+void raftStorePrint(SRaftStore *pRaftStore) {
-
+  char storeBuf[RAFT_STORE_BLOCK_SIZE];
-int32_t persistCurrentTerm(SyncTerm currentTerm) { return 0; }
+  raftStoreSerialize(pRaftStore, storeBuf, sizeof(storeBuf));
-
+  printf("%s\n", storeBuf);
-int32_t voteFor(SRaftId *pRaftId) { return 0; }
+}
-
-int32_t persistVoteFor(SRaftId *pRaftId) { return 0; }

source/libs/sync/test/syncTest.cpp

@@ -14,9 +14,23 @@ void *pingFunc(void *param) {
 }
 
 int main() {
+  tsAsyncLog = 0;
+  taosInitLog((char *)"syncTest.log", 100000, 10);
+
+  SRaftStore *pRaftStore = raftStoreOpen("./raft_store.json");
+  assert(pRaftStore != NULL);
+
+  raftStorePrint(pRaftStore);
+
+  pRaftStore->currentTerm = 100;
+  pRaftStore->voteFor.addr = 200;
+  pRaftStore->voteFor.vgId = 300;
+
+  raftStorePrint(pRaftStore);
+
+  raftStorePersist(pRaftStore);
+
 
-  testJson();
-  testJson2();
 
   tsAsyncLog = 0;
   taosInitLog((char *)"syncTest.log", 100000, 10);