feat: Refactor config (#255)

* Replaced the original Pconfig

pikiwidb.conf

@@ -10,7 +10,7 @@ port 9221
 # If you want you can bind a single interface, if the bind option is not
 # specified all the interfaces will listen for incoming connections.
 #
-# bind 127.0.0.1
+ip 127.0.0.1
 
 
 # Close the connection after a client is idle for N seconds (0 to disable)
@@ -35,7 +35,7 @@ logfile stdout
 # Set the number of databases. The default database is DB 0, you can select
 # a different one on a per-connection basis using SELECT <dbid> where
 # dbid is a number between 0 and 'databases'-1
-databases 3
+databases 16
 
 ################################ SNAPSHOTTING  #################################
 #
@@ -315,37 +315,27 @@ slowlog-log-slower-than 10000
 # You can reclaim memory used by the slow log with SLOWLOG RESET.
 slowlog-max-len 128
 
-############################### ADVANCED CONFIG ###############################
-
-# Redis calls an internal function to perform many background tasks, like
-# closing connections of clients in timeot, purging expired keys that are
-# never requested, and so forth.
-#
-# Not all tasks are perforemd with the same frequency, but Redis checks for
-# tasks to perform accordingly to the specified "hz" value.
-#
-# By default "hz" is set to 10. Raising the value will use more CPU when
-# Redis is idle, but at the same time will make Redis more responsive when
-# there are many keys expiring at the same time, and timeouts may be
-# handled with more precision.
-#
-# The range is between 1 and 500, however a value over 100 is usually not
-# a good idea. Most users should use the default of 10 and raise this up to
-# 100 only in environments where very low latency is required.
-hz 10
 ############################### BACKENDS CONFIG ###############################
-# PikiwiDB is a in memory database, though it has aof and rdb for dump data to disk, it
-# is very limited. Try use leveldb for real storage, pikiwidb as cache. The cache algorithm
-# is like linux page cache, please google or read your favorite linux book
-# 0 is default, no backend
-# 1 is RocksDB, currently only support RocksDB
-backend 1
-backendpath dump
-# the frequency of dump to backend per second
-backendhz 10
-# the rocksdb number per db
-db-instance-num 5
-# default 86400 * 7
-rocksdb-ttl-second 604800
-# default 86400 * 3
-rocksdb-periodic-second 259200
+# PikiwiDB uses RocksDB as the underlying storage engine, and the data belonging
+# to the same DB is distributed among several RocksDB instances.
+
+# RocksDB instances number per DB
+db-instance-num 3
+# default is 86400 * 7
+small-compaction-threshold 604800
+# default is 86400 * 3
+small-compaction-duration-threshold 259200
+
+############################### ROCKSDB CONFIG ###############################
+rocksdb-max-subcompactions 2
+rocksdb-max-background-jobs 4
+rocksdb-max-write-buffer-number 2
+rocksdb-min-write-buffer-number-to-merge 2
+# default is 64M
+rocksdb-write-buffer-size 67108864
+rocksdb-level0-file-num-compaction-trigger 4
+rocksdb-number-levels 7
+rocksdb-enable-pipelined-write no
+rocksdb-level0-slowdown-writes-trigger 20
+rocksdb-level0-stop-writes-trigger 36
+

src/base_cmd.cc

@@ -30,7 +30,7 @@ std::vector<std::string> BaseCmd::CurrentKey(PClient* client) const { return std
 
 void BaseCmd::Execute(PClient* client) {
   auto dbIndex = client->GetCurrentDB();
-  if (!isExclusive()) {
+  if (!HasFlag(kCmdFlagsExclusive)) {
     PSTORE.GetBackend(dbIndex)->LockShared();
   }
 
@@ -39,7 +39,7 @@ void BaseCmd::Execute(PClient* client) {
   }
   DoCmd(client);
 
-  if (!isExclusive()) {
+  if (!HasFlag(kCmdFlagsExclusive)) {
     PSTORE.GetBackend(dbIndex)->UnLockShared();
   }
 }

src/base_cmd.h

@@ -293,8 +293,6 @@ class BaseCmd : public std::enable_shared_from_this<BaseCmd> {
 
   uint32_t GetCmdID() const;
 
-  bool isExclusive() { return static_cast<bool>(flag_ & kCmdFlagsExclusive); }
-
  protected:
   // Execute a specific command
   virtual void DoCmd(PClient* client) = 0;

src/client.cc

@@ -437,7 +437,7 @@ void PClient::OnConnect() {
     SetName("MasterConnection");
     SetFlag(kClientFlagMaster);
 
-    if (g_config.masterauth.empty()) {
+    if (g_config.master_auth.empty()) {
       SetAuth();
     }
   } else {

src/cmd_admin.cc

@@ -20,14 +20,27 @@ CmdConfigGet::CmdConfigGet(const std::string& name, int16_t arity)
 
 bool CmdConfigGet::DoInitial(PClient* client) { return true; }
 
-void CmdConfigGet::DoCmd(PClient* client) { client->AppendString("config cmd in development"); }
+void CmdConfigGet::DoCmd(PClient* client) {
+  std::vector<std::string> results;
+  for (int i = 0; i < client->argv_.size() - 2; i++) {
+    g_config.Get(client->argv_[i + 2], &results);
+  }
+  client->AppendStringVector(results);
+}
 
 CmdConfigSet::CmdConfigSet(const std::string& name, int16_t arity)
     : BaseCmd(name, arity, kCmdFlagsAdmin, kAclCategoryAdmin) {}
 
 bool CmdConfigSet::DoInitial(PClient* client) { return true; }
 
-void CmdConfigSet::DoCmd(PClient* client) { client->AppendString("config cmd in development"); }
+void CmdConfigSet::DoCmd(PClient* client) {
+  auto s = g_config.Set(client->argv_[2], client->argv_[3]);
+  if (!s.ok()) {
+    client->SetRes(CmdRes::kInvalidParameter);
+  } else {
+    client->SetRes(CmdRes::kOK);
+  }
+}
 
 FlushdbCmd::FlushdbCmd(const std::string& name, int16_t arity)
     : BaseCmd(name, arity, kCmdFlagsAdmin | kCmdFlagsWrite, kAclCategoryWrite | kAclCategoryAdmin) {}
@@ -85,7 +98,7 @@ ShutdownCmd::ShutdownCmd(const std::string& name, int16_t arity)
 bool ShutdownCmd::DoInitial(PClient* client) {
   // For now, only shutdown need check local
   if (client->PeerIP().find("127.0.0.1") == std::string::npos &&
-      client->PeerIP().find(g_config.ip.c_str()) == std::string::npos) {
+      client->PeerIP().find(g_config.ip.ToString()) == std::string::npos) {
     client->SetRes(CmdRes::kErrOther, kCmdNameShutdown + " should be localhost");
     return false;
   }

src/cmd_hash.cc

@@ -151,7 +151,7 @@ void HGetAllCmd::DoCmd(PClient* client) {
   int64_t total_fv = 0;
   int64_t cursor = 0;
   int64_t next_cursor = 0;
-  size_t raw_limit = g_config.max_client_response_size;
+  size_t raw_limit = g_config.max_client_response_size.load();
   std::string raw;
   std::vector<storage::FieldValue> fvs;
   storage::Status s;

src/common.cc

@@ -199,4 +199,23 @@ std::vector<PString> SplitString(const PString& str, char seperator) {
   return results;
 }
 
+std::string MergeString(const std::vector<std::string*>& values, char delimiter) {
+  std::string result(*values.at(0));
+  for (int i = 0; i < values.size() - 1; i++) {
+    result += delimiter;
+    result += *values.at(i + 1);
+  }
+  return result;
+}
+
+std::string MergeString(const std::vector<AtomicString*>& values, char delimiter) {
+  std::string result(*values.at(0));
+  for (int i = 0; i < values.size() - 1; i++) {
+    result += delimiter;
+    std::string s(*values.at(i + 1));
+    result += s;
+  }
+  return result;
+}
+
 }  // namespace pikiwidb

src/common.h

@@ -12,6 +12,7 @@
 #include <cstddef>
 #include <cstdio>
 #include <functional>
+#include <shared_mutex>
 #include <string>
 #include <vector>
 
@@ -76,8 +77,67 @@ enum class PParseResult : int8_t {
 
 PParseResult GetIntUntilCRLF(const char*& ptr, std::size_t nBytes, int& val);
 
+class AtomicString {
+ public:
+  AtomicString() = default;
+  ~AtomicString() = default;
+  AtomicString(std::string str) {
+    std::lock_guard lock(mutex_);
+    str_ = std::move(str);
+  }
+  AtomicString(std::string&& str) {
+    std::lock_guard lock(mutex_);
+    str_ = std::move(str);
+  }
+  AtomicString(const std::string& str) {
+    std::lock_guard lock(mutex_);
+    str_ = str;
+  }
+  AtomicString(const char* c) {
+    std::lock_guard lock(mutex_);
+    str_ = std::string(c);
+  };
+  AtomicString& operator=(const std::string& str) {
+    std::lock_guard lock(mutex_);
+    str_ = str;
+    return *this;
+  }
+  AtomicString& operator=(std::string&& str) {
+    std::lock_guard lock(mutex_);
+    str_ = std::move(str);
+    return *this;
+  }
+  operator std::string() {
+    std::shared_lock<std::shared_mutex> lock(mutex_);
+    return str_;
+  }
+
+  operator std::string() const {
+    std::shared_lock<std::shared_mutex> lock(mutex_);
+    return str_;
+  }
+
+  bool empty() const {
+    std::shared_lock<std::shared_mutex> lock(mutex_);
+    return str_.empty();
+  }
+
+  std::string ToString() const {
+    std::shared_lock<std::shared_mutex> lock(mutex_);
+    return str_;
+  }
+
+ private:
+  mutable std::shared_mutex mutex_;
+  std::string str_;
+};
+
 std::vector<PString> SplitString(const PString& str, char seperator);
 
+std::string MergeString(const std::vector<std::string*>& values, char delimiter);
+
+std::string MergeString(const std::vector<AtomicString*>& values, char delimiter);
+
 // The defer class for C++11
 class ExecuteOnScopeExit {
  public: