fix: flushdb and flushall bug

include/pika_admin.h

@@ -185,17 +185,14 @@ class FlushallCmd : public Cmd {
       : Cmd(name, arity, flag, static_cast<uint32_t>(AclCategory::KEYSPACE)) {}
   void Do() override;
   void DoThroughDB() override;
-  void DoUpdateCache() override;
+  void DoUpdateCache(std::shared_ptr<DB> db);
   void Split(const HintKeys& hint_keys) override{};
   void Merge() override{};
   Cmd* Clone() override { return new FlushallCmd(*this); }
-  void Execute() override;
   void FlushAllWithoutLock();
-  void DoBinlog(std::shared_ptr<SyncMasterDB> sync_db_);
 
  private:
   void DoInitial() override;
-  std::string ToRedisProtocol() override;
   void DoWithoutLock(std::shared_ptr<DB> db);
 };
 
@@ -212,7 +209,6 @@ class FlushdbCmd : public Cmd {
   void Merge() override{};
   Cmd* Clone() override { return new FlushdbCmd(*this); }
   void FlushAllDBsWithoutLock();
-  void Execute() override;
   std::string GetFlushDBname() { return db_name_; }
 
  private:
@@ -265,7 +261,6 @@ class InfoCmd : public Cmd {
   void Split(const HintKeys& hint_keys) override {};
   void Merge() override {};
   Cmd* Clone() override { return new InfoCmd(*this); }
-  void Execute() override;
 
  private:
   InfoSection info_section_;
@@ -333,7 +328,6 @@ class ConfigCmd : public Cmd {
   void Split(const HintKeys& hint_keys) override {};
   void Merge() override {};
   Cmd* Clone() override { return new ConfigCmd(*this); }
-  void Execute() override;
 
  private:
   std::vector<std::string> config_args_v_;

include/pika_db.h

@@ -139,12 +139,9 @@ class DB : public std::enable_shared_from_this<DB>, public pstd::noncopyable {
   void Init();
   bool TryUpdateMasterOffset();
   /*
-   * FlushDB & FlushSubDB use
+   * FlushDB used
    */
-  bool FlushDB();
-  bool FlushSubDB(const std::string& db_name);
   bool FlushDBWithoutLock();
-  bool FlushSubDBWithoutLock(const std::string& db_name);
   bool ChangeDb(const std::string& new_path);
   pstd::Status GetBgSaveUUID(std::string* snapshot_uuid);
   void PrepareRsync();

include/pika_transaction.h

@@ -31,7 +31,6 @@ class ExecCmd : public Cmd {
       : Cmd(name, arity, flag, static_cast<uint32_t>(AclCategory::TRANSACTION)) {}
   void Do() override;
   Cmd* Clone() override { return new ExecCmd(*this); }
-  void Execute() override;
   void Split(const HintKeys& hint_keys) override {}
   void Merge() override {}
   std::vector<std::string> current_key() const override { return {}; }
@@ -79,7 +78,6 @@ class WatchCmd : public Cmd {
       : Cmd(name, arity, flag, static_cast<uint32_t>(AclCategory::TRANSACTION)) {}
 
   void Do() override;
-  void Execute() override;
   void Split(const HintKeys& hint_keys) override {}
   Cmd* Clone() override { return new WatchCmd(*this); }
   void Merge() override {}

src/pika_admin.cc

@@ -526,39 +526,8 @@ void FlushallCmd::DoInitial() {
     return;
   }
 }
-void FlushallCmd::Do() {
-  if (!db_) {
-    LOG(INFO) << "Flushall, but DB not found";
-  } else {
-    db_->FlushDB();
-  }
-}
-
-void FlushallCmd::DoThroughDB() {
-  Do();
-}
-
-void FlushallCmd::DoUpdateCache() {
-  // clear cache
-  if (PIKA_CACHE_NONE != g_pika_conf->cache_model()) {
-    g_pika_server->ClearCacheDbAsync(db_);
-  }
-}
-
-// flushall convert flushdb writes to every db binlog
-std::string FlushallCmd::ToRedisProtocol() {
-  std::string content;
-  content.reserve(RAW_ARGS_LEN);
-  RedisAppendLen(content, 1, "*");
-
-  // to flushdb cmd
-  std::string flushdb_cmd("flushdb");
-  RedisAppendLenUint64(content, flushdb_cmd.size(), "$");
-  RedisAppendContent(content, flushdb_cmd);
-  return content;
-}
 
-void FlushallCmd::Execute() {
+void FlushallCmd::Do() {
   std::lock_guard l_trw(g_pika_server->GetDBLock());
   for (const auto& db_item : g_pika_server->GetDB()) {
     if (db_item.second->IsKeyScaning()) {
@@ -580,6 +549,17 @@ void FlushallCmd::Execute() {
   }
 }
 
+void FlushallCmd::DoThroughDB() {
+  Do();
+}
+
+void FlushallCmd::DoUpdateCache(std::shared_ptr<DB> db) {
+  // clear cache
+  if (PIKA_CACHE_NONE != g_pika_conf->cache_model()) {
+    g_pika_server->ClearCacheDbAsync(db);
+  }
+}
+
 void FlushallCmd::FlushAllWithoutLock() {
   for (const auto& db_item : g_pika_server->GetDB()) {
     std::shared_ptr<DB> db = db_item.second;
@@ -589,45 +569,18 @@ void FlushallCmd::FlushAllWithoutLock() {
       return;
     }
     DoWithoutLock(db);
-    DoBinlog(g_pika_rm->GetSyncMasterDBs()[p_info]);
   }
   if (res_.ok()) {
     res_.SetRes(CmdRes::kOk);
   }
 }
 
-void FlushallCmd::DoBinlog(std::shared_ptr<SyncMasterDB> sync_db) {
-  if (res().ok() && is_write() && g_pika_conf->write_binlog()) {
-    std::shared_ptr<net::NetConn> conn_ptr = GetConn();
-    std::shared_ptr<std::string> resp_ptr = GetResp();
-    // Consider that dummy cmd appended by system, both conn and resp are null.
-    if ((!conn_ptr || !resp_ptr) && (name_ != kCmdDummy)) {
-      if (!conn_ptr) {
-        LOG(WARNING) << sync_db->SyncDBInfo().ToString() << " conn empty.";
-      }
-      if (!resp_ptr) {
-        LOG(WARNING) << sync_db->SyncDBInfo().ToString() << " resp empty.";
-      }
-      res().SetRes(CmdRes::kErrOther);
-      return;
-    }
-
-    Status s = sync_db->ConsensusProposeLog(shared_from_this());
-    if (!s.ok()) {
-      LOG(WARNING) << sync_db->SyncDBInfo().ToString() << " Writing binlog failed, maybe no space left on device "
-                   << s.ToString();
-      res().SetRes(CmdRes::kErrOther, s.ToString());
-      return;
-    }
-  }
-}
-
 void FlushallCmd::DoWithoutLock(std::shared_ptr<DB> db) {
   if (!db) {
     LOG(INFO) << "Flushall, but DB not found";
   } else {
     db->FlushDBWithoutLock();
-    DoUpdateCache();
+    DoUpdateCache(db);
   }
 }
 
@@ -646,18 +599,19 @@ void FlushdbCmd::DoInitial() {
 
 void FlushdbCmd::Do() {
   if (!db_) {
-    LOG(INFO) << "Flushdb, but DB not found";
+    res_.SetRes(CmdRes::kInvalidDB);
   } else {
-    if (db_name_ == "all") {
-      db_->FlushDB();
+    if (db_->IsKeyScaning()) {
+      res_.SetRes(CmdRes::kErrOther, "The keyscan operation is executing, Try again later");
     } else {
-      //Floyd does not support flushdb by type
-      LOG(ERROR) << "cannot flushdb by type in floyd";
+      std::lock_guard s_prw(g_pika_rm->GetDBLock());
+      std::lock_guard l_prw(db_->GetDBLock());
+      FlushAllDBsWithoutLock();
+      res_.SetRes(CmdRes::kOk);
     }
   }
 }
 
-
 void FlushdbCmd::DoThroughDB() {
   Do();
 }
@@ -676,7 +630,6 @@ void FlushdbCmd::FlushAllDBsWithoutLock() {
     return;
   }
   DoWithoutLock();
-  DoBinlog();
 }
 
 void FlushdbCmd::DoWithoutLock() {
@@ -689,22 +642,6 @@ void FlushdbCmd::DoWithoutLock() {
       //Floyd does not support flushdb by type
       LOG(ERROR) << "cannot flushdb by type in floyd";
     }
-    DoUpdateCache();
-  }
-}
-
-void FlushdbCmd::Execute() {
-  if (!db_) {
-    res_.SetRes(CmdRes::kInvalidDB);
-  } else {
-    if (db_->IsKeyScaning()) {
-      res_.SetRes(CmdRes::kErrOther, "The keyscan operation is executing, Try again later");
-    } else {
-      std::lock_guard l_prw(db_->GetDBLock());
-      std::lock_guard s_prw(g_pika_rm->GetDBLock());
-      FlushAllDBsWithoutLock();
-      res_.SetRes(CmdRes::kOk);
-    }
   }
 }
 
@@ -1465,11 +1402,6 @@ std::string InfoCmd::CacheStatusToString(int status) {
   }
 }
 
-void InfoCmd::Execute() {
-  std::shared_ptr<DB> db = g_pika_server->GetDB(db_name_);
-  Do();
-}
-
 void ConfigCmd::DoInitial() {
   if (!CheckArg(argv_.size())) {
     res_.SetRes(CmdRes::kWrongNum, kCmdNameConfig);
@@ -2793,10 +2725,6 @@ void ConfigCmd::ConfigResetstat(std::string& ret) {
   ret = "+OK\r\n";
 }
 
-void ConfigCmd::Execute() {
-  Do();
-}
-
 void MonitorCmd::DoInitial() {
   if (argv_.size() != 1) {
     res_.SetRes(CmdRes::kWrongNum, kCmdNameMonitor);

src/pika_db.cc

@@ -201,6 +201,7 @@ DisplayCacheInfo DB::GetCacheInfo() {
 }
 
 bool DB::FlushDBWithoutLock() {
+  std::lock_guard l(bgsave_protector_);
   if (bgsave_info_.bgsaving) {
     return false;
   }
@@ -225,34 +226,6 @@ bool DB::FlushDBWithoutLock() {
   return true;
 }
 
-bool DB::FlushSubDBWithoutLock(const std::string& db_name) {
-  std::lock_guard l(bgsave_protector_);
-  if (bgsave_info_.bgsaving) {
-    return false;
-  }
-
-  LOG(INFO) << db_name_ << " Delete old " + db_name + " db...";
-  storage_.reset();
-
-  std::string dbpath = db_path_;
-  if (dbpath[dbpath.length() - 1] != '/') {
-    dbpath.append("/");
-  }
-
-  std::string sub_dbpath = dbpath + db_name;
-  std::string del_dbpath = dbpath + db_name + "_deleting";
-  pstd::RenameFile(sub_dbpath, del_dbpath);
-
-  storage_ = std::make_shared<storage::Storage>(g_pika_conf->db_instance_num(),
-      g_pika_conf->default_slot_num(), g_pika_conf->classic_mode());
-  rocksdb::Status s = storage_->Open(g_pika_server->storage_options(), db_path_);
-  assert(storage_);
-  assert(s.ok());
-  LOG(INFO) << db_name_ << " open new " + db_name + " db success";
-  g_pika_server->PurgeDir(del_dbpath);
-  return true;
-}
-
 void DB::DoBgSave(void* arg) {
   std::unique_ptr<BgTaskArg> bg_task_arg(static_cast<BgTaskArg*>(arg));
 
@@ -575,11 +548,6 @@ void DB::ClearBgsave() {
   bgsave_info_.Clear();
 }
 
-bool DB::FlushSubDB(const std::string& db_name) {
-  std::lock_guard rwl(dbs_rw_);
-  return FlushSubDBWithoutLock(db_name);
-}
-
 void DB::UpdateCacheInfo(CacheInfo& cache_info) {
   std::unique_lock<std::shared_mutex> lock(cache_info_rwlock_);
 
@@ -628,9 +596,3 @@ void DB::ResetDisplayCacheInfo(int status) {
   cache_info_.waitting_load_keys_num = 0;
   cache_usage_ = 0;
 }
-
-bool DB::FlushDB() {
-  std::lock_guard rwl(dbs_rw_);
-  std::lock_guard l(bgsave_protector_);
-  return FlushDBWithoutLock();
-}

src/pika_transaction.cc

@@ -42,6 +42,22 @@ void MultiCmd::DoInitial() {
 void ExecCmd::Do() {
   auto conn = GetConn();
   auto client_conn = std::dynamic_pointer_cast<PikaClientConn>(conn);
+  if (client_conn == nullptr) {
+    res_.SetRes(CmdRes::kErrOther, name());
+    return;
+  }
+  if (!client_conn->IsInTxn()) {
+    res_.SetRes(CmdRes::kErrOther, "EXEC without MULTI");
+    return;
+  }
+  if (IsTxnFailedAndSetState()) {
+    client_conn->ExitTxn();
+    return;
+  }
+  SetCmdsVec();
+  Lock();
+  conn = GetConn();
+  client_conn = std::dynamic_pointer_cast<PikaClientConn>(conn);
   std::vector<CmdRes> res_vec = {};
   std::vector<std::shared_ptr<std::string>> resp_strs;
   for (size_t i = 0; i < cmds_.size(); ++i) {
@@ -83,26 +99,6 @@ void ExecCmd::Do() {
   for (auto &r : res_vec) {
     res_.AppendStringRaw(r.message());
   }
-}
-
-void ExecCmd::Execute() {
-  auto conn = GetConn();
-  auto client_conn = std::dynamic_pointer_cast<PikaClientConn>(conn);
-  if (client_conn == nullptr) {
-    res_.SetRes(CmdRes::kErrOther, name());
-    return;
-  }
-  if (!client_conn->IsInTxn()) {
-    res_.SetRes(CmdRes::kErrOther, "EXEC without MULTI");
-    return;
-  }
-  if (IsTxnFailedAndSetState()) {
-    client_conn->ExitTxn();
-    return;
-  }
-  SetCmdsVec();
-  Lock();
-  Do();
   Unlock();
   ServeToBLrPopWithKeys();
   list_cmd_.clear();
@@ -245,10 +241,6 @@ void WatchCmd::Do() {
   res_.SetRes(CmdRes::kOk);
 }
 
-void WatchCmd::Execute() {
-  Do();
-}
-
 void WatchCmd::DoInitial() {
   if (!CheckArg(argv_.size())) {
     res_.SetRes(CmdRes::kWrongNum, name());