feat: kvctl add and some bugs fixed

Author: ZYunfeii

CMakeLists.txt

@@ -23,7 +23,12 @@ set(server_src "${server_src};${CMAKE_CURRENT_BINARY_DIR}/../log/log.cc")
 file(GLOB_RECURSE client_src "${CMAKE_CURRENT_BINARY_DIR}/../client/*.cc")
 set(client_src "${client_src};${CMAKE_CURRENT_BINARY_DIR}/../log/log.cc")
 
+set(kvctl_src "${client_src};${CMAKE_CURRENT_BINARY_DIR}/../kvctl/main.cc")
+list(REMOVE_ITEM kvctl_src "${CMAKE_CURRENT_BINARY_DIR}/../client/client_main.cc")
+
 add_executable(kvserver ${server_src})
 add_executable(kvclient ${client_src})
+add_executable(kvctl    ${kvctl_src})
 target_link_libraries(kvserver grpc_proto ${_REFLECTION} ${_GRPC_GRPCPP} ${_PROTOBUF_LIBPROTOBUF})
 target_link_libraries(kvclient grpc_proto ${_REFLECTION} ${_GRPC_GRPCPP} ${_PROTOBUF_LIBPROTOBUF})
+target_link_libraries(kvctl grpc_proto gflags ${_REFLECTION} ${_GRPC_GRPCPP} ${_PROTOBUF_LIBPROTOBUF})

Dockerfile

@@ -0,0 +1,11 @@
+FROM grpc/cxx
+LABEL maintainer="[email protected]"
+ENV PROJECT_DIR=/server 
+
+COPY ./ $PROJECT_DIR/
+
+RUN apt-get update && apt-get install -y cmake && apt-get install -y build-essential && apt-get clean
+
+WORKDIR $PROJECT_DIR/build
+
+RUN cmake .. && make && cd ..

README.md

@@ -1,8 +1,49 @@
 # A KV high-performance mini-database based on memory and C++17
+**This project is inspired by Redis source code.**
+
+# Command line tools
+Developed command line tool **kvctl**.
+value type:string
+```shell
+yunfei@ubuntu:~/MiniKV/build$ ./kvctl --key qjx --operate set --value hello
+yunfei@ubuntu:~/MiniKV/build$ ./kvctl --key qjx --operate get
+
+hello
+```
+
+value type:list
+```shell
+yunfei@ubuntu:~/MiniKV/build$ ./kvctl --key zyf --operate set --value hello --encoding list
+yunfei@ubuntu:~/MiniKV/build$ ./kvctl --key zyf --operate set --value world --encoding list
+
+yunfei@ubuntu:~/MiniKV/build$ ./kvctl --key zyf --operate get
+
+world hello 
+```
+
 
 # build
+**Dependencies: grpc, protobuf, gflags**
 In the project dir, do:
 ```shell
 cd build && cmake .. && make
 ```
-then you can get `kvserver` and `kvclient`.
+then you can get `kvserver` and `kvclient`.
+
+# run
+```shell
+./kvserver
+```
+
+# About
+- This project is based on the gRPC framework to implement the memory-based KV cache middleware, which realizes the client and server respectively. Support list, string type KV cache, data snapshot and progressive Rehash.
+
+- Communication between client and server is realized based on gRPC, and concurrency security is realized based on read-write lock; The client supports the addition, deletion, modification and query of keys.
+
+- Support data snapshot, background thread timing persistence, based on user-defined data frame format; Service start automaticly to read snapshot.
+
+- The underlying storage structure mimics the design of Redis hash table, solves hash conflicts with zipper method, and realizes automatic memory management with intelligent pointer.
+
+- Imitate Redis to implement the expired key based on the expired hash table, and clean up the expired key through lazy deletion.
+
+- Simulate the design of Redis double hash table, and implement progressive rehash after the background thread calculates the load factor regularly.

client/kvclient.cc

@@ -7,6 +7,7 @@ KVClient::KVClient(std::string ip, uint32_t port) : ip_(ip), port_(port) {
 }
 
 int KVClient::setKV(std::string key, std::string val, uint32_t encoding) {
+    if (key.empty()) return MiniKV_KEY_EMPTY;
     kv::ReqKV req;
     kv::SetKVResponse res;
     req.set_key(key);
@@ -22,6 +23,7 @@ int KVClient::setKV(std::string key, std::string val, uint32_t encoding) {
 }
 
 int KVClient::delK(std::string key) {
+    if (key.empty()) return MiniKV_KEY_EMPTY;
     kv::ReqK req;
     kv::DelKVResponse res;
     req.set_key(key);
@@ -35,6 +37,7 @@ int KVClient::delK(std::string key) {
 }
 
 GetRes KVClient::getK(std::string key) {
+    if (key.empty()) return {};
     GetRes ans;
     kv::ReqK req;
     kv::GetKResponse res;
@@ -58,6 +61,7 @@ GetRes KVClient::getK(std::string key) {
 }
 
 int KVClient::setExpires(std::string key, uint64_t millisecond) {
+    if (key.empty()) return MiniKV_KEY_EMPTY;
     auto now = std::chrono::system_clock::now(); 
     auto timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
     uint64_t expires = timestamp + millisecond;

kvctl/main.cc

@@ -0,0 +1,65 @@
+#include "gflags/gflags.h"
+#include "../client/kvclient.h"
+
+DEFINE_string(ip, "localhost", "ip for admin node");
+DEFINE_int32(port, 6789, "port for admin node");
+DEFINE_string(key, "", "key");
+DEFINE_string(value, "", "value");
+DEFINE_string(encoding, "string", "encoding, see in encoding.h");
+DEFINE_string(operate, "set", "operate, e.g. set, del..");
+DEFINE_uint64(expires, 5000, "expire time");
+
+
+DECLARE_string(ip);
+DECLARE_int32(port);
+DECLARE_string(key);
+DECLARE_string(value);
+DECLARE_string(encoding);
+DECLARE_string(operate);
+DECLARE_uint64(expires);
+
+#include <unordered_map>
+std::unordered_map<std::string, int> oper {
+    {"set", KV_SET},
+    {"del", KV_DEL},
+    {"get", KV_GET},
+    {"expire", KV_EXPIRE}
+};
+
+std::unordered_map<std::string, int> enMap {
+    {"string", MiniKV_STRING},
+    {"list", MiniKV_LIST},
+};
+
+int main(int argc, char* argv[]) {
+    gflags::ParseCommandLineFlags(&argc, &argv, true);
+    KVClient* kvclient = new KVClient(FLAGS_ip, FLAGS_port);
+    std::string encoding = FLAGS_encoding;
+    std::string operation  = FLAGS_operate;
+    std::string key = FLAGS_key;
+    std::string val = FLAGS_value;
+    uint64_t expires = FLAGS_expires;
+    
+    switch (oper[operation]) {
+        case KV_SET : {
+            kvclient->setKV(key, val, enMap[encoding]);
+            break;
+        }
+        case KV_DEL : {
+            kvclient->delK(key);
+            break;
+        }
+        case KV_GET : {
+            auto res = kvclient->getK(key);
+            for (int i = 0; i < res.data.size(); ++i) {
+                std::cout << res.data[i] << " ";
+            } std::cout << std::endl;
+            break;
+        }
+        case KV_EXPIRE : {
+            kvclient->setExpires(key, expires);
+        }
+    }
+
+    return 0;
+}

minikv.rdb

@@ -39,23 +39,30 @@ void MiniKVDB::insert(std::string key, std::string val, uint32_t encoding) {
     progressiveRehash(hash2_);
 }
 
-void MiniKVDB::get(std::string key, std::vector<std::string>& res) {
-    // there are 2 cases for empty res:
-    // 1. exceed the time limit
-    // 2. no data
+bool MiniKVDB::expired(std::string key) {
     std::vector<std::string> e;
     expires_->get(key, e);
     if (!e.empty()) {
         uint64_t expires = stoi(e[0]);
         auto now = std::chrono::system_clock::now(); 
         auto timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
         if (timestamp >= expires) {
-            res = std::vector<std::string>();
-            hash1_->del(key); // lazy delete mode
-            expires_->del(key);
-            return;
+            return true;
         }
     }
+    return false;
+}
+
+void MiniKVDB::get(std::string key, std::vector<std::string>& res) {
+    // there are 2 cases for empty res:
+    // 1. exceed the time limit
+    // 2. no data
+    if (expired(key)) {
+        res = std::vector<std::string>();
+        hash1_->del(key); // lazy delete mode
+        expires_->del(key);
+        return;
+    }
     if (!rehashFlag_) {
         hash1_->get(key, res);
         return;
@@ -80,6 +87,7 @@ int MiniKVDB::setExpire(std::string key, uint64_t expires) {
 }
 
 void MiniKVDB::rdbSave() {
+    if (rehashFlag_) return;
     // TODO: save the expire time
     std::unique_lock<std::shared_mutex> lk(smutex_);
     kvlogi("rdb save triggered!");
@@ -91,6 +99,7 @@ void MiniKVDB::rdbSave() {
         }
         for (auto it = d.begin(); it != d.end(); ++it) {
             std::string key = it->get()->key;
+            if (expired(key)) continue;
             uint32_t encoding = it->get()->encoding;
             void* data = it->get()->data.get();
             saveKVWithEncoding(key, encoding, data);

server/db.cc

@@ -53,6 +53,7 @@ typedef class MiniKVDB {
     void rdbSave();
     void rehash();
     void progressiveRehash(std::shared_ptr<HashTable> hash2);
+    bool expired(std::string key);
     // void fixedTimeDeleteExpiredKey();
 public:
     void insert(std::string key, std::string val, uint32_t encoding);

server/db.h

@@ -128,14 +128,13 @@ class HashTable {
     }
     int del(std::string key) {
         int slot = hash(key);
-        bool flag = false;
-        std::for_each(hash_[slot].begin(), hash_[slot].end(), [&key, &slot, &flag, this](std::shared_ptr<Entry> p) {
-            if (p.get()->key == key) {
-                hash_[slot].remove(p);
-                flag = true;
+        for (auto it = hash_[slot].begin(); it != hash_[slot].end(); ++it) {
+            if (it->get()->key == key) {
+                hash_[slot].remove(*it);
+                return MiniKV_DEL_SUCCESS;
             }
-        });
-        return flag ? MiniKV_DEL_SUCCESS : MiniKV_DEL_FAIL;
+        }
+        return MiniKV_DEL_FAIL;
     }
 
     bool needRehash() {

server/hash.h

@@ -11,7 +11,15 @@ enum {
     MiniKV_DEL_FAIL,
     MiniKV_DEL_SUCCESS,
     MiniKV_SET_EXPIRE_SUCCESS,
-    MiniKV_SET_EXPIRE_FAIL
+    MiniKV_SET_EXPIRE_FAIL,
+    MiniKV_KEY_EMPTY
+};
+
+enum {
+    KV_SET = 100,
+    KV_DEL,
+    KV_GET,
+    KV_EXPIRE
 };
 
 #endif