feat:kv更换成skiplist

Merge pull request #11 from 578223592/main

CMakeLists.txt

@@ -19,6 +19,7 @@ include_directories(${PROJECT_SOURCE_DIR}/example)
 include_directories(${PROJECT_SOURCE_DIR}/src/raftCore/include)
 include_directories(${PROJECT_SOURCE_DIR}/src/raftRpcPro/include)
 include_directories(${PROJECT_SOURCE_DIR}/src/raftClerk/include)
+include_directories(${PROJECT_SOURCE_DIR}/src/skipList/include)
 
 
 # 设置项目库文件搜索路径 -L

bin/test.conf

@@ -1,6 +1,6 @@
 node0ip=127.0.1.1
-node0port=26951
+node0port=26603
 node1ip=127.0.1.1
-node1port=26952
+node1port=26604
 node2ip=127.0.1.1
-node2port=26953
+node2port=26605

src/CMakeLists.txt

@@ -1,5 +1,7 @@
 
 
+add_subdirectory(skipList)
+
 add_subdirectory(common) #需要注意如果rpc中需要依靠common中的代码，则需要在rpc在前，其cmake文件中的set才对rpc文件夹可见
 # 使用include_directories()函数应该也是可行的
 add_subdirectory(rpc)

src/raftCore/include/kvServer.h

@@ -9,6 +9,7 @@
 #include "raft.h"
 #include <unordered_map>
 #include "kvServerRPC.pb.h"
+#include "skipList.h"
 #include <iostream>
 #include <unordered_map>
 #include <boost/serialization/serialization.hpp>
@@ -32,6 +33,8 @@ class KvServer : raftKVRpcProctoc::kvServerRpc {
     int m_maxRaftState; // snapshot if log grows this big
 
     // Your definitions here.
+    std::string m_serializedKVData; // todo ： 序列化后的kv数据，理论上可以不用，但是目前没有找到特别好的替代方法
+    SkipList<std::string, std::string> m_skipList;
     std::unordered_map<std::string, std::string> m_kvDB;
 
     std::unordered_map<int, LockQueue<Op> *> waitApplyCh;
@@ -108,21 +111,27 @@ class KvServer : raftKVRpcProctoc::kvServerRpc {
     template<class Archive>
     void serialize(Archive &ar, const unsigned int version) //这里面写需要序列话和反序列化的字段
     {
-        ar & m_kvDB;
+        ar & m_serializedKVData;
+
+        // ar & m_kvDB;
         ar & m_lastRequestId;
     }
 
     std::string getSnapshotData() {
+        m_serializedKVData = m_skipList.dump_file();
         std::stringstream ss;
         boost::archive::text_oarchive oa(ss);
         oa << *this;
+        m_serializedKVData.clear();
         return ss.str();
     }
 
     void parseFromString(const std::string &str) {
         std::stringstream ss(str);
         boost::archive::text_iarchive ia(ss);
         ia >> *this;
+        m_skipList.load_file(m_serializedKVData);
+        m_serializedKVData.clear();
     }
 
     /////////////////serialiazation end ///////////////////////////////

src/raftCore/include/raft.h

@@ -170,8 +170,6 @@ class Raft : public raftRpcProctoc::raftRpc
         std::unordered_map<std::string,int> umap;
     public:
 
-
-
     };
 };
 

src/raftCore/kvServer.cpp

@@ -1,4 +1,4 @@
-#include "kvServer.h"   //todo  ： 这里为什么只能用相对路径
+#include "kvServer.h"
 
 #include <rpcprovider.h>
 
@@ -10,22 +10,27 @@ void KvServer::DprintfKVDB() {
     }
     std::lock_guard<std::mutex> lg(m_mtx);
     Defer ec1([this]() -> void {
-        for (const auto &item: m_kvDB) {
-            DPrintf("[DBInfo ----]Key : %s, Value : %s", &item.first, &item.second);
-        }
+        // for (const auto &item: m_kvDB) {
+        //     DPrintf("[DBInfo ----]Key : %s, Value : %s", &item.first, &item.second);
+        // }
+        m_skipList.display_list();
     });
+
 }
 
 void KvServer::ExecuteAppendOpOnKVDB(Op op) {
     //if op.IfDuplicate {   //get请求是可重复执行的，因此可以不用判复
     //	return
     //}
     m_mtx.lock();
-    if (m_kvDB.find(op.Key) != m_kvDB.end()) {
-        m_kvDB[op.Key] = m_kvDB[op.Key] + op.Value;
-    } else {
-        m_kvDB.insert(std::make_pair(op.Key, op.Value));
-    }
+
+    m_skipList.insert_set_element(op.Key,op.Value);
+
+    // if (m_kvDB.find(op.Key) != m_kvDB.end()) {
+    //     m_kvDB[op.Key] = m_kvDB[op.Key] + op.Value;
+    // } else {
+    //     m_kvDB.insert(std::make_pair(op.Key, op.Value));
+    // }
     m_lastRequestId[op.ClientId] = op.RequestId;
     m_mtx.unlock();
 
@@ -38,10 +43,14 @@ void KvServer::ExecuteGetOpOnKVDB(Op op, std::string *value, bool *exist) {
     m_mtx.lock();
     *value = "";
     *exist = false;
-    if (m_kvDB.find(op.Key) != m_kvDB.end()) {
+    if(m_skipList.search_element(op.Key, *value)) {
         *exist = true;
-        *value = m_kvDB[op.Key];
+        // *value = m_skipList.se //value已经完成赋值了
     }
+    // if (m_kvDB.find(op.Key) != m_kvDB.end()) {
+    //     *exist = true;
+    //     *value = m_kvDB[op.Key];
+    // }
     m_lastRequestId[op.ClientId] = op.RequestId;
     m_mtx.unlock();
 
@@ -56,7 +65,8 @@ void KvServer::ExecuteGetOpOnKVDB(Op op, std::string *value, bool *exist) {
 
 void KvServer::ExecutePutOpOnKVDB(Op op) {
     m_mtx.lock();
-    m_kvDB[op.Key] = op.Value;
+    m_skipList.insert_set_element(op.Key,op.Value);
+    // m_kvDB[op.Key] = op.Value;
     m_lastRequestId[op.ClientId] = op.RequestId;
     m_mtx.unlock();
 
@@ -357,7 +367,8 @@ void KvServer::Get(google::protobuf::RpcController *controller, const ::raftKVRp
     done->Run();
 }
 
-KvServer::KvServer(int me, int maxraftstate, std::string nodeInforFileName, short port) {
+KvServer::KvServer(int me, int maxraftstate, std::string nodeInforFileName, short port):
+m_skipList(6){
     std::shared_ptr<Persister> persister = std::make_shared<Persister>(me);
 
     m_me = me;
@@ -418,7 +429,8 @@ KvServer::KvServer(int me, int maxraftstate, std::string nodeInforFileName, shor
     //////////////////////////////////
 
     // You may need initialization code here.
-    m_kvDB; //kvdb初始化
+    // m_kvDB; //kvdb初始化
+    m_skipList;
     waitApplyCh;
     m_lastRequestId;
     m_lastSnapShotRaftLogIndex = 0; //todo:感覺這個函數沒什麼用，不如直接調用raft節點中的snapshot值？？？