merge with upstream

bill-wangbiao · Mar 23, 2016 · 862dfce · 862dfce
2 parents 085a869 + f52979d
commit 862dfce
Show file tree

Hide file tree

Showing 5 changed files with 67 additions and 3 deletions.
diff --git a/include/pika_conf.h b/include/pika_conf.h
@@ -37,6 +37,7 @@ class PikaConf : public slash::BaseConf {
   const std::vector<std::string>& vuser_blacklist() {
     RWLock l(&rwlock_, false); return user_blacklist_;
   }
+  std::string compression()       { RWLock l(&rwlock_, false); return compression_; }
   int target_file_size_base()         { RWLock l(&rwlock_, false); return target_file_size_base_; }
   std::string conf_path()       { RWLock l(&rwlock_, false); return conf_path_; }
   bool readonly() {
@@ -100,7 +101,7 @@ class PikaConf : public slash::BaseConf {
   std::string pidfile_;
 
   //char pidfile_[PIKA_WORD_SIZE];
-  //char compression_[PIKA_WORD_SIZE];
+  std::string compression_;
   //int maxconnection_;
 
   //int expire_logs_days_;

diff --git a/include/pika_server.h b/include/pika_server.h
@@ -157,6 +157,9 @@ class PikaServer
   void ClearPurge() {
     purging_ = false;
   }
+  //flushall
+  bool FlushAll();
+  void PurgeDir(std::string& path);
 
 /*
  * client related
@@ -215,6 +218,11 @@ class PikaServer
   static void DoPurgeLogs(void* arg);
   bool GetPurgeWindow(uint32_t &max);
 
+  /*
+   * Flushall use 
+   */
+  static void DoPurgeDir(void* arg);
+
   PikaServer(PikaServer &ps);
   void operator =(const PikaServer &ps);
 };

diff --git a/src/pika_admin.cc b/src/pika_admin.cc
@@ -264,7 +264,11 @@ void FlushallCmd::DoInitial(PikaCmdArgsType &argv, const CmdInfo* const ptr_info
 }
 void FlushallCmd::Do() {
   slash::RWLock(g_pika_server->rwlock(), true);
-  res_.SetRes(CmdRes::kOk);
+  if (g_pika_server->FlushAll()) {
+    res_.SetRes(CmdRes::kOk);
+  } else {
+    res_.SetRes(CmdRes::kErrOther, "There are some bgthread using db now, can not flushall");
+  }
 }
 
 void ReadonlyCmd::DoInitial(PikaCmdArgsType &argv, const CmdInfo* const ptr_info) {

diff --git a/src/pika_conf.cc b/src/pika_conf.cc
@@ -30,6 +30,7 @@ int PikaConf::Load()
   GetConfStr("userblacklist", &user_blacklist);
   SetUserBlackList(std::string(user_blacklist));
   GetConfStr("dump_path", &bgsave_path_);
+  GetConfStr("compression", &compression_);
 
   GetConfBool("slave-read-only", &readonly_);
 
@@ -119,6 +120,7 @@ int PikaConf::ConfigRewrite() {
   SetConfStr("userblacklist", suser_blacklist());
   SetConfStr("dump_path", bgsave_path_);
   SetConfInt("target_file_size_base", target_file_size_base_);
+  SetConfStr("compression", compression_);
   SetConfBool("slave_read_only", readonly_);
 
   return WriteBack();

diff --git a/src/pika_server.cc b/src/pika_server.cc
@@ -33,9 +33,12 @@ PikaServer::PikaServer() :
 
   option.write_buffer_size = g_pika_conf->write_buffer_size();
   option.target_file_size_base = g_pika_conf->target_file_size_base();
+	if (g_pika_conf->compression() == "none") {
+		 option.compression = false;
+  }
   std::string db_path = g_pika_conf->db_path();
   LOG(WARNING) << "Prepare DB...";
-  db_ = std::unique_ptr<nemo::Nemo>(new nemo::Nemo(db_path, option));
+  db_ = std::shared_ptr<nemo::Nemo>(new nemo::Nemo(db_path, option));
   assert(db_);
   LOG(WARNING) << "DB Success";
 
@@ -532,6 +535,52 @@ bool PikaServer::GetPurgeWindow(uint32_t &max) {
   return false;
 }
 
+bool PikaServer::FlushAll() {
+  if (bgsaving_ /*|| bgscaning_*/) {
+    return false;
+  }
+  std::string dbpath = g_pika_conf->db_path();
+  if (dbpath[dbpath.length() - 1] == '/') {
+    dbpath.erase(dbpath.length() - 1);
+  }
+  int pos = dbpath.find_last_of('/');
+  dbpath = dbpath.substr(0, pos);
+  dbpath.append("/deleting");
+  slash::RenameFile(g_pika_conf->db_path(), dbpath.c_str());
+
+  LOG(WARNING) << "Delete old db...";
+  db_.reset();
+
+  nemo::Options option;
+  option.write_buffer_size = g_pika_conf->write_buffer_size();
+  option.target_file_size_base = g_pika_conf->target_file_size_base();
+  if (g_pika_conf->compression() == "none") {
+    option.compression = false;
+  }
+  LOG(WARNING) << "Prepare open new db...";
+  db_ = std::shared_ptr<nemo::Nemo>(new nemo::Nemo(g_pika_conf->db_path(), option));
+  LOG(WARNING) << "open new db success";
+  PurgeDir(dbpath);
+  return true; 
+}
+
+void PikaServer::PurgeDir(std::string& path) {
+  std::string *dir_path = new std::string(path);
+  // Start new thread if needed
+  if (!purge_thread_.is_running()) {
+    purge_thread_.StartThread();
+  }
+  purge_thread_.Schedule(&DoPurgeDir, static_cast<void*>(dir_path));
+}
+
+void PikaServer::DoPurgeDir(void* arg) {
+  std::string path = *(static_cast<std::string*>(arg));
+  DLOG(INFO) << "Delete dir: " << path << " start";
+  slash::DeleteDir(path);
+  DLOG(INFO) << "Delete dir: " << path << " done";
+  delete static_cast<std::string*>(arg);
+}
+
 void PikaServer::ClientKillAll() {
   for (size_t idx = 0; idx != PIKA_MAX_WORKER_THREAD_NUM; idx++) {
     pika_worker_thread_[idx]->ThreadClientKill();