Simplify the number of files in tools/pika-migrate

Author: pro-spild

conf/pika.conf

@@ -289,16 +289,6 @@ sync-window-size : 9000
 # Supported Units [K|M|G]. Its default unit is in [bytes] and its default value is 268435456(256MB). The value range is [64MB, 1GB].
 max-conn-rbuf-size : 268435456
 
-###################
-## Migrate Settings
-###################
-
-target-redis-host : 127.0.0.1
-target-redis-port : 6379
-target-redis-pwd  :
-
-sync-batch-num    : 100
-redis-sender-num  : 10
 
 #######################################################################E#######
 #! Critical Settings !#

include/pika_conf.h

@@ -346,14 +346,6 @@ class PikaConf : public pstd::BaseConf {
   int max_conn_rbuf_size() { return max_conn_rbuf_size_.load(); }
   int consensus_level() { return consensus_level_.load(); }
   int replication_num() { return replication_num_.load(); }
-
-  std::string target_redis_host()                   { return target_redis_host_; }
-  int target_redis_port()                           { return target_redis_port_; }
-  std::string target_redis_pwd()                    { return target_redis_pwd_;  }
-  int sync_batch_num()                              { return sync_batch_num_;    }
-  int redis_sender_num()                            { return redis_sender_num_;  }
-  
-
   int rate_limiter_mode() {
     std::shared_lock l(rwlock_);
     return rate_limiter_mode_;
@@ -933,13 +925,6 @@ class PikaConf : public pstd::BaseConf {
   std::map<std::string, std::string> diff_commands_;
   void TryPushDiffCommands(const std::string& command, const std::string& value);
 
-  // migrate configure items
-  std::string target_redis_host_;
-  int target_redis_port_;
-  std::string target_redis_pwd_;
-  int sync_batch_num_;
-  int redis_sender_num_;
-
   //
   // Critical configure items
   //

include/pika_repl_bgworker.h

@@ -47,7 +47,6 @@ class PikaReplBgWorker {
   net::BGThread bg_thread_;
   static int HandleWriteBinlog(net::RedisParser* parser, const net::RedisCmdArgsType& argv);
   static void ParseBinlogOffset(const InnerMessage::BinlogOffset& pb_offset, LogOffset* offset);
-  static void ParseAndSendPikaCommand(const std::shared_ptr<Cmd>& c_ptr);
 };
 
 #endif  // PIKA_REPL_BGWROKER_H_

include/pika_server.h

@@ -46,7 +46,6 @@
 #include "include/pika_statistic.h"
 #include "include/pika_transaction.h"
 #include "include/rsync_server.h"
-#include "include/redis_sender.h"
 
 extern std::unique_ptr<PikaConf> g_pika_conf;
 
@@ -310,12 +309,6 @@ class PikaServer : public pstd::noncopyable {
 
   pstd::Status GetCmdRouting(std::vector<net::RedisCmdArgsType>& redis_cmds, std::vector<Node>* dst, bool* all_local);
 
-  /*
-   * migrate used
-   */
-  int SendRedisCommand(const std::string& command, const std::string& key);
-  void RetransmitData(const std::string& path);
-
   // info debug use
   void ServerStatus(std::string* info);
 
@@ -624,11 +617,6 @@ class PikaServer : public pstd::noncopyable {
    */
   std::unique_ptr<PikaAuxiliaryThread> pika_auxiliary_thread_;
 
-  /*
-   * migrate to redis used
-   */
-  std::vector<std::unique_ptr<RedisSender>> redis_senders_;
-
   /*
    * Async slotsMgrt use
    */

src/pika_conf.cc

@@ -207,12 +207,6 @@ int PikaConf::Load() {
 
   if (classic_mode_.load()) {
     GetConfInt("databases", &databases_);
-    
-    // pika-migrate-tool only support 1 db
-    if (databases_ != 1) {
-      LOG(FATAL) << "pika-migrate-tool only support 1 db";
-    }
-    
     if (databases_ < 1 || databases_ > MAX_DB_NUM) {
       LOG(FATAL) << "config databases error, limit [1 ~ 8], the actual is: " << databases_;
     }
@@ -633,22 +627,6 @@ int PikaConf::Load() {
     sync_window_size_.store(tmp_sync_window_size);
   }
 
-  // redis-migrate conifg args
-  target_redis_host_ = "127.0.0.1";
-  GetConfStr("target-redis-host", &target_redis_host_);
-
-  target_redis_port_ = 6379;
-  GetConfInt("target-redis-port", &target_redis_port_);
-
-  target_redis_pwd_ = "";
-  GetConfStr("target-redis-pwd" , &target_redis_pwd_);
-
-  sync_batch_num_ = 100;
-  GetConfInt("sync-batch-num", &sync_batch_num_);
-
-  redis_sender_num_ = 8;
-  GetConfInt("redis-sender-num", &redis_sender_num_);
-
   // max conn rbuf size
   int tmp_max_conn_rbuf_size = PIKA_MAX_CONN_RBUF;
   GetConfIntHuman("max-conn-rbuf-size", &tmp_max_conn_rbuf_size);

src/pika_db.cc

@@ -473,9 +473,6 @@ bool DB::TryUpdateMasterOffset() {
             << ",  master_ip: " << master_ip << ", master_port: " << master_port << ", filenum: " << filenum
             << ", offset: " << offset << ", term: " << term << ", index: " << index;
 
-  // Retransmit Data to target redis
-  g_pika_server->RetransmitData(dbsync_path_);
-
   pstd::DeleteFile(info_path);
   if (!ChangeDb(dbsync_path_)) {
     LOG(WARNING) << "DB: " << db_name_ << ", Failed to change db";

src/pika_repl_bgworker.cc

@@ -231,7 +231,6 @@ void PikaReplBgWorker::WriteDBInSyncWay(const std::shared_ptr<Cmd>& c_ptr) {
       && PIKA_CACHE_NONE != g_pika_conf->cache_mode()
       && c_ptr->GetDB()->cache()->CacheStatus() == PIKA_CACHE_STATUS_OK) {
     if (c_ptr->is_write()) {
-      ParseAndSendPikaCommand(c_ptr);
       c_ptr->DoThroughDB();
       if (c_ptr->IsNeedUpdateCache()) {
         c_ptr->DoUpdateCache();
@@ -240,7 +239,6 @@ void PikaReplBgWorker::WriteDBInSyncWay(const std::shared_ptr<Cmd>& c_ptr) {
       LOG(WARNING) << "It is impossbile to reach here";
     }
   } else {
-    ParseAndSendPikaCommand(c_ptr);
     c_ptr->Do();
   }
   if (!c_ptr->IsSuspend()) {
@@ -276,33 +274,3 @@ void PikaReplBgWorker::WriteDBInSyncWay(const std::shared_ptr<Cmd>& c_ptr) {
     }
   }
 }
-
-void PikaReplBgWorker::ParseAndSendPikaCommand(const std::shared_ptr<Cmd>& c_ptr) {
-  const PikaCmdArgsType& argv = c_ptr->argv();
-    if (!strcasecmp(argv[0].data(), "pksetexat")) {
-    if (argv.size() != 4) {
-      LOG(WARNING) << "find invaild command, command size: " << argv.size();
-      return;
-    } else {
-      std::string key = argv[1];
-      int timestamp = std::atoi(argv[2].data());
-      std::string value = argv[3];
-
-      int seconds = timestamp - time(NULL);
-      PikaCmdArgsType tmp_argv;
-      tmp_argv.push_back("setex");
-      tmp_argv.push_back(key);
-      tmp_argv.push_back(std::to_string(seconds));
-      tmp_argv.push_back(value);
-
-      std::string command;
-      net::SerializeRedisCommand(tmp_argv, &command);
-      g_pika_server->SendRedisCommand(command, key);
-    }
-  } else {
-    std::string key = argv.size() >= 2 ? argv[1] : argv[0];
-    std::string command;
-    net::SerializeRedisCommand(argv, &command);
-    g_pika_server->SendRedisCommand(command, key);
-  }
-}

src/pika_repl_client_conn.cc

@@ -154,7 +154,6 @@ void PikaReplClientConn::HandleMetaSyncResponse(void* arg) {
   LOG(INFO) << "Finish to handle meta sync response";
 }
 
-static bool alerady_full_sync = false;
 void PikaReplClientConn::HandleDBSyncResponse(void* arg) {
   std::unique_ptr<ReplClientTaskArg> task_arg(static_cast<ReplClientTaskArg*>(arg));
   std::shared_ptr<net::PbConn> conn = task_arg->conn;
@@ -179,16 +178,12 @@ void PikaReplClientConn::HandleDBSyncResponse(void* arg) {
     return;
   }
 
-  if (!alerady_full_sync) {
-    alerady_full_sync = true;
-  } else {
-    LOG(FATAL) << "DBSyncResponse should only be called once";
-  }
-
   slave_db->SetMasterSessionId(session_id);
+
   slave_db->StopRsync();
   slave_db->SetReplState(ReplState::kWaitDBSync);
   LOG(INFO) << "DB: " << db_name << " Need Wait To Sync";
+
   //now full sync is starting, add an unfinished full sync count
   g_pika_conf->AddInternalUsedUnfinishedFullSync(slave_db->DBName());
 }

src/pika_server.cc

@@ -24,8 +24,6 @@
 #include "include/pika_monotonic_time.h"
 #include "include/pika_rm.h"
 #include "include/pika_server.h"
-#include "include/pika_sender.h"
-#include "include/migrator_thread.h"
 
 using pstd::Status;
 extern PikaServer* g_pika_server;
@@ -103,15 +101,6 @@ PikaServer::PikaServer()
     }
   }
 
-  // Create redis sender
-  for (int i = 0; i < g_pika_conf->redis_sender_num(); ++i) {
-    redis_senders_.emplace_back(std::make_unique<RedisSender>(int(i),
-                                                              g_pika_conf->target_redis_host(),
-                                                              g_pika_conf->target_redis_port(),
-                                                              g_pika_conf->target_redis_pwd()));
-  }
-  
-
   acl_ = std::make_unique<::Acl>();
   SetSlowCmdThreadPoolFlag(g_pika_conf->slow_cmd_pool());
   bgsave_thread_.set_thread_name("PikaServer::bgsave_thread_");
@@ -140,10 +129,7 @@ PikaServer::~PikaServer() {
   bgsave_thread_.StopThread();
   key_scan_thread_.StopThread();
   pika_migrate_thread_->StopThread();
-  for (size_t i = 0; i < redis_senders_.size(); ++i) {
-    redis_senders_[i]->Stop();
-  }
-  redis_senders_.clear();
+
   dbs_.clear();
 
   LOG(INFO) << "PikaServer " << pthread_self() << " exit!!!";
@@ -223,15 +209,6 @@ void PikaServer::Start() {
                << (ret == net::kCreateThreadError ? ": create thread error " : ": other error");
   }
 
-  for (size_t i = 0; i < redis_senders_.size(); ++i) {
-    ret = redis_senders_[i]->StartThread();
-    if (ret != net::kSuccess) {
-      dbs_.clear();
-      LOG(FATAL) << "Start RedisSender Error: " << ret
-                 << (ret == net::kCreateThreadError ? ": create thread error " : ": other error");
-    }
-  }
-
   time(&start_time_s_);
   LOG(INFO) << "Pika Server going to start";
   rsync_server_->Start();
@@ -1565,77 +1542,6 @@ Status PikaServer::GetCmdRouting(std::vector<net::RedisCmdArgsType>& redis_cmds,
   return Status::OK();
 }
 
-int PikaServer::SendRedisCommand(const std::string& command, const std::string& key) {
-  // Send command
-  size_t idx = std::hash<std::string>()(key) % redis_senders_.size();
-  redis_senders_[idx]->SendRedisCommand(command);
-  return 0;
-}
-
-void PikaServer::RetransmitData(const std::string& path) {
-  std::shared_ptr<storage::Storage> storage_ = std::make_shared<storage::Storage>();
-  rocksdb::Status s = storage_->Open(g_pika_server->storage_options(), path);
-
-  if (!s.ok()) {
-    LOG(FATAL) << "open received database error: " << s.ToString();
-    return;
-  }
-
-  // Init SenderThread
-  int thread_num = g_pika_conf->redis_sender_num();
-  std::string target_host = g_pika_conf->target_redis_host();
-  int target_port = g_pika_conf->target_redis_port();
-  std::string target_pwd = g_pika_conf->target_redis_pwd();
-
-  LOG(INFO) << "open received database success, start retransmit data to redis("
-    << target_host << ":" << target_port << ")";
-
-
-  std::vector<std::shared_ptr<PikaSender>> pika_senders;
-  std::vector<std::shared_ptr<MigratorThread>> migrators;
-
-  for (int i = 0; i < thread_num; i++) {
-    pika_senders.emplace_back(std::make_shared<PikaSender>(target_host, target_port, target_pwd));
-  }
-  migrators.emplace_back(std::make_shared<MigratorThread>(storage_, &pika_senders, storage::kStrings, thread_num));
-  migrators.emplace_back(std::make_shared<MigratorThread>(storage_, &pika_senders, storage::kLists, thread_num));
-  migrators.emplace_back(std::make_shared<MigratorThread>(storage_, &pika_senders, storage::kHashes, thread_num));
-  migrators.emplace_back(std::make_shared<MigratorThread>(storage_, &pika_senders, storage::kSets, thread_num));
-  migrators.emplace_back(std::make_shared<MigratorThread>(storage_, &pika_senders, storage::kZSets, thread_num));
-
-  for (size_t i = 0; i < pika_senders.size(); i++) {
-    pika_senders[i]->StartThread();
-  }
-  for (size_t i = 0; i < migrators.size(); i++) {
-    migrators[i]->StartThread();
-  }
-
-  for (size_t i = 0; i < migrators.size(); i++) {
-    migrators[i]->JoinThread();
-  }
-  for (size_t i = 0; i < pika_senders.size(); i++) {
-    pika_senders[i]->Stop();
-  }
-  for (size_t i = 0; i < pika_senders.size(); i++) {
-    pika_senders[i]->JoinThread();
-  }
-
-  int64_t replies = 0, records = 0;
-  for (size_t i = 0; i < migrators.size(); i++) {
-    records += migrators[i]->num();
-  }
-  migrators.clear();
-  for (size_t i = 0; i < pika_senders.size(); i++) {
-    replies += pika_senders[i]->elements();
-  }
-  pika_senders.clear();
-
-  LOG(INFO) << "=============== Retransmit Finish =====================";
-  LOG(INFO) << "Total records : " << records << " have been Scaned";
-  LOG(INFO) << "Total replies : " << replies << " received from redis server";
-  LOG(INFO) << "=======================================================";
-}
-
 void PikaServer::ServerStatus(std::string* info) {
   std::stringstream tmp_stream;
   size_t q_size = ClientProcessorThreadPoolCurQueueSize();

tools/CMakeLists.txt

@@ -3,6 +3,7 @@ add_subdirectory(./benchmark_client)
 add_subdirectory(./binlog_sender)
 add_subdirectory(./manifest_generator)
 add_subdirectory(./rdb_to_pika)
+add_subdirectory(./pika_migrate)
 add_subdirectory(./pika_to_txt)
 add_subdirectory(./txt_to_pika)
 add_subdirectory(./pika-port/pika_port_3)