feat:add multi key slot migrate (OpenAtomFoundation#2486)

* add multi key slot migrate * config set migrate thread --------- Co-authored-by: chejinge <chejinge@360.cn>
cheniujh · Mar 13, 2024 · 6170b0b · 6170b0b
1 parent 50b2de8
commit 6170b0b
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 1 deletion.
diff --git a/conf/pika.conf b/conf/pika.conf
@@ -341,6 +341,12 @@ max-bytes-for-level-multiplier : 10
 # slotmigrate  [yes | no]
 slotmigrate : no
 
+# slotmigrate thread num
+slotmigrate-thread-num : 8
+
+# thread-migrate-keys-num  1/8 of the write_buffer_size_
+thread-migrate-keys-num : 64
+
 # BlockBasedTable block_size, default 4k
 # block-size: 4096
 

diff --git a/include/pika_conf.h b/include/pika_conf.h
@@ -514,6 +514,14 @@ class PikaConf : public pstd::BaseConf {
     std::lock_guard l(rwlock_);
     slotmigrate_ = (value == "yes");
   }
+  void SetSlotMigrateThreadNum(const int value) {
+    std::lock_guard l(rwlock_);
+    slotmigrate_thread_num_ = value;
+  }
+  void SetThreadMigrateKeysNum(const int value) {
+    std::lock_guard l(rwlock_);
+    thread_migrate_keys_num_ = value;
+  }
   void SetExpireLogsNums(const int value) {
     std::lock_guard l(rwlock_);
     TryPushDiffCommands("expire-logs-nums", std::to_string(value));

diff --git a/src/pika_admin.cc b/src/pika_admin.cc
@@ -1673,6 +1673,18 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     EncodeString(&config_body, g_pika_conf->slotmigrate() ? "yes" : "no");
   }
 
+  if (pstd::stringmatch(pattern.data(), "slotmigrate-thread-num", 1)) {
+    elements += 2;
+    EncodeString(&config_body, "slotmigrate-thread-num");
+    EncodeNumber(&config_body, g_pika_conf->slotmigrate_thread_num());
+  }
+
+  if (pstd::stringmatch(pattern.data(), "thread-migrate-keys-num", 1)) {
+    elements += 2;
+    EncodeString(&config_body, "thread-migrate-keys-num");
+    EncodeNumber(&config_body, g_pika_conf->thread_migrate_keys_num());
+  }
+
   if (pstd::stringmatch(pattern.data(), "dump-path", 1) != 0) {
     elements += 2;
     EncodeString(&config_body, "dump-path");
@@ -2148,6 +2160,8 @@ void ConfigCmd::ConfigSet(std::shared_ptr<DB> db) {
         "requirepass",
         "masterauth",
         "slotmigrate",
+        "slotmigrate-thread-num",
+        "thread-migrate-keys-num",
         "userpass",
         "userblacklist",
         "dump-prefix",
@@ -2257,6 +2271,22 @@ void ConfigCmd::ConfigSet(std::shared_ptr<DB> db) {
     }
     g_pika_conf->SetRootConnectionNum(static_cast<int>(ival));
     res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "slotmigrate-thread-num") {
+    if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival <= 0) {
+      res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'expire-logs-nums'\r\n");
+      return;
+    }
+    long int migrate_thread_num = (0 > ival || 24 < ival) ? 8 : ival;
+    g_pika_conf->SetSlotMigrateThreadNum(static_cast<int>(ival));
+    res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "thread-migrate-keys-num") {
+    if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival <= 0) {
+      res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'expire-logs-nums'\r\n");
+      return;
+    }
+    long int thread_migrate_keys_num = (8 > ival || 128 < ival) ? 64 : ival;
+    g_pika_conf->SetThreadMigrateKeysNum(static_cast<int>(ival));
+    res_.AppendStringRaw("+OK\r\n");
   } else if (set_item == "slowlog-write-errorlog") {
     bool is_write_errorlog;
     if (value == "yes") {

diff --git a/src/pika_conf.cc b/src/pika_conf.cc
@@ -281,7 +281,7 @@ int PikaConf::Load() {
 
   // arena_block_size
   GetConfInt64Human("slotmigrate-thread-num_", &slotmigrate_thread_num_);
-  if (slotmigrate_thread_num_ < 1 || slotmigrate_thread_num_ > 24) {
+  if (slotmigrate_thread_num_ < 0 || slotmigrate_thread_num_ > 24) {
     slotmigrate_thread_num_ = 8;  // 1/8 of the write_buffer_size_
   }