add purgelogsto command

include/pika_admin.h

@@ -75,4 +75,13 @@ private:
   virtual void DoInitial(PikaCmdArgsType &argvs, const CmdInfo* const ptr_info);
 };
 
+class PurgelogstoCmd : public Cmd {
+public:
+  PurgelogstoCmd() : num_(0){
+  }
+  virtual void Do();
+private:
+  uint32_t num_;
+  virtual void DoInitial(PikaCmdArgsType &argvs, const CmdInfo* const ptr_info);
+};
 #endif

include/pika_binlog.h

@@ -3,6 +3,7 @@
 
 #include <cstdio>
 #include <list>
+#include <string>
 #include <deque>
 #include <pthread.h>
 
@@ -30,7 +31,7 @@ class Version;
 class Binlog
 {
  public:
-  Binlog(const char* Binlog_path);
+  Binlog(const std::string& Binlog_path);
   ~Binlog();
 
   void Lock()         { mutex_.Lock(); }

include/pika_command.h

@@ -17,6 +17,7 @@ const std::string kCmdNameAuth = "auth";
 const std::string kCmdNameBgsave = "bgsave";
 const std::string kCmdNameBgsaveoff = "bgsaveoff";
 const std::string kCmdNameCompact = "compact";
+const std::string kCmdNamePurgelogsto = "purgelogsto";
 
 //Kv
 const std::string kCmdNameSet = "set";
@@ -184,6 +185,8 @@ public:
     kOutOfRange,
     kInvalidPwd,
     kNoneBgsave,
+    kPurgeExist,
+    kInvalidParameter,
     kWrongNum,
     kErrOther,
   };
@@ -226,6 +229,10 @@ public:
       return "-ERR invalid password\r\n";
     case kNoneBgsave:
       return "-ERR No BGSave Works now\r\n";
+    case kPurgeExist:
+      return "-ERR binlog may in use or non_exist or already in purging...\r\n";
+    case kInvalidParameter:
+      return "-ERR Invalid Argument\r\n";
     case kWrongNum:
       result = "-ERR wrong number of arguments for '";
       result.append(message_);

include/pika_define.h

@@ -76,9 +76,10 @@ static const size_t kHeaderSize = 1 + 3;
  */
 static const int64_t kPoolSize = 1073741824;
 
-static std::string kBinlog = "/binlog";
+static const std::string kBinlogPrefix = "binlog";
+static const size_t kBinlogPrefixLen = 6;
 
-static std::string kManifest = "/manifest";
+static const std::string kManifest = "manifest";
 
 /*
  * define common character

include/pika_server.h

@@ -80,7 +80,6 @@ public:
 
   void DeleteSlave(int fd); // hb_fd
   bool FindSlave(std::string& ip_port);
-  Status GetSmallestValidLog(uint32_t* max);
   void MayUpdateSlavesMap(int64_t sid, int32_t hb_fd);
   void BecomeMaster(); 
 
@@ -140,9 +139,25 @@ public:
   }
   void Bgsave();
   bool Bgsaveoff();
-  void ClearBgsave();
   bool RunBgsaveEngine();
+  void ClearBgsave() {
+    bgsave_info_.Clear();
+    bgsaving_ = false;
+  }
+
+/*
+ * PurgeLog used
+ */
+  struct PurgeArg {
+    PikaServer *p;
+    uint32_t to;  
+  };
+  bool PurgeLogs(uint32_t to);
+  bool PurgeFiles(uint32_t to);
 
+  void ClearPurge() {
+    purging_ = false;
+  }
 
 
 private:
@@ -176,7 +191,6 @@ private:
   /*
    * Bgsave use
    */
-  slash::Mutex bgsave_protector_;
   std::atomic<bool> bgsaving_;
   pink::BGThread bgsave_thread_;
   nemo::BackupEngine *bgsave_engine_;
@@ -186,6 +200,15 @@ private:
   bool InitBgsaveEnv(const std::string& bgsave_path);
   bool InitBgsaveEngine();
 
+  /*
+   * Purgelogs use
+   */
+  std::atomic<bool> purging_;
+  pink::BGThread purge_thread_;
+  
+  static void DoPurgeLogs(void* arg);
+  bool GetPurgeWindow(uint32_t &max);
+
   PikaServer(PikaServer &ps);
   void operator =(const PikaServer &ps);
 };

src/pika.cc

@@ -27,7 +27,7 @@ static void PikaConfInit(const std::string& path) {
 }
 
 static void PikaGlogInit() {
-  FLAGS_log_dir = "./log";
+  FLAGS_log_dir = g_pika_conf->log_path();
   FLAGS_minloglevel = 0;
   FLAGS_alsologtostderr = true;
   FLAGS_max_log_size = 1800;

src/pika_admin.cc

@@ -198,7 +198,6 @@ void CompactCmd::DoInitial(PikaCmdArgsType &argv, const CmdInfo* const ptr_info)
   }
 }
 void CompactCmd::Do() {
-  CmdRes::CmdRet ret;
   nemo::Status s = g_pika_server->db()->Compact(nemo::kALL);
   if (s.ok()) {
     res_.SetRes(CmdRes::kOk);
@@ -206,3 +205,31 @@ void CompactCmd::Do() {
     res_.SetRes(CmdRes::kErrOther, s.ToString());
   }
 }
+
+void PurgelogstoCmd::DoInitial(PikaCmdArgsType &argv, const CmdInfo* const ptr_info) {
+  if (!ptr_info->CheckArg(argv.size())) {
+    res_.SetRes(CmdRes::kWrongNum, kCmdNamePurgelogsto);
+    return;
+  }
+  std::string filename = slash::StringToLower(argv[1]);
+  if (filename.size() <= kBinlogPrefixLen || 
+      kBinlogPrefix != filename.substr(0, kBinlogPrefixLen)) {
+    res_.SetRes(CmdRes::kInvalidParameter);
+    return;
+  }
+  std::string str_num = filename.substr(kBinlogPrefixLen);
+  int64_t num = 0;
+  if (!slash::string2l(str_num.data(), str_num.size(), &num) || num < 0) {
+    res_.SetRes(CmdRes::kInvalidParameter);
+    return;
+  }
+  num_ = num;
+}
+
+void PurgelogstoCmd::Do() {
+  if (g_pika_server->PurgeLogs(num_)) {
+    res_.SetRes(CmdRes::kOk);
+  } else {
+    res_.SetRes(CmdRes::kPurgeExist);
+  }
+}

src/pika_binlog.cc

@@ -75,7 +75,7 @@ Status Version::Init() {
 /*
  * Binlog
  */
-Binlog::Binlog(const char* Binlog_path) :
+Binlog::Binlog(const std::string& Binlog_path) :
     version_(NULL),
     consumer_num_(0),
     item_num_(0),
@@ -94,7 +94,7 @@ Binlog::Binlog(const char* Binlog_path) :
 
   slash::CreateDir(binlog_path_);
 
-  filename = binlog_path_ + kBinlog;
+  filename = binlog_path_ + kBinlogPrefix;
   const std::string manifest = binlog_path_ + kManifest;
   std::string profile;
 
@@ -417,3 +417,4 @@ Status Binlog::SetProducerStatus(uint32_t pro_num, uint64_t pro_offset) {
   InitLogFile();
   return Status::OK();
 }
+

src/pika_command.cc

@@ -23,6 +23,8 @@ void InitCmdInfoTable() {
   cmd_infos.insert(std::pair<std::string, CmdInfo*>(kCmdNameBgsaveoff, bgsaveoffptr));
   CmdInfo* compactptr = new CmdInfo(kCmdNameCompact, 1, kCmdFlagsRead | kCmdFlagsAdmin);
   cmd_infos.insert(std::pair<std::string, CmdInfo*>(kCmdNameCompact, compactptr));
+  CmdInfo* purgelogptr = new CmdInfo(kCmdNamePurgelogsto, 2, kCmdFlagsRead | kCmdFlagsAdmin);
+  cmd_infos.insert(std::pair<std::string, CmdInfo*>(kCmdNamePurgelogsto, purgelogptr));
 
   //Kv
   ////SetCmd
@@ -271,6 +273,8 @@ void InitCmdTable(std::unordered_map<std::string, Cmd*> *cmd_table) {
   cmd_table->insert(std::pair<std::string, Cmd*>(kCmdNameBgsaveoff, bgsaveoffptr));
   Cmd* compactptr = new CompactCmd();
   cmd_table->insert(std::pair<std::string, Cmd*>(kCmdNameCompact, compactptr));
+  Cmd* purgelogptr = new PurgelogstoCmd();
+  cmd_table->insert(std::pair<std::string, Cmd*>(kCmdNamePurgelogsto, purgelogptr));
 
   //Kv
   ////SetCmd

src/pika_conf.cc

@@ -35,6 +35,10 @@ int PikaConf::Load()
   GetConfStr("dump_path", &bgsave_path_);
   GetConfInt("target_file_size_base", &target_file_size_base_);
 
+  if (log_path_[log_path_.length() - 1] != '/') {
+    log_path_ += "/";
+  }
+
   if (bgsave_path_[bgsave_path_.length() - 1] != '/') {
     bgsave_path_ += "/";
   }

src/pika_server.cc

@@ -19,7 +19,8 @@ PikaServer::PikaServer() :
   master_port_(0),
   repl_state_(PIKA_REPL_NO_CONNECT),
   role_(PIKA_ROLE_SINGLE),
-  bgsaving_(false) {
+  bgsaving_(false),
+  purging_(false) {
 
   pthread_rwlock_init(&rwlock_, NULL);
   
@@ -49,7 +50,7 @@ PikaServer::PikaServer() :
   pika_trysync_thread_ = new PikaTrysyncThread();
 
   pthread_rwlock_init(&state_protector_, NULL);
-  logger_ = new Binlog("./log");
+  logger_ = new Binlog(g_pika_conf->log_path());
 }
 
 PikaServer::~PikaServer() {
@@ -284,27 +285,6 @@ Status PikaServer::AddBinlogSender(SlaveItem &slave, uint32_t filenum, uint64_t
   }
 }
 
-Status PikaServer::GetSmallestValidLog(uint32_t* max) {
-  slash::MutexLock l(&slave_mutex_);
-  std::vector<SlaveItem>::iterator iter;
-
-  *max = logger_->version_->pro_num();
-  for (iter = slaves_.begin(); iter != slaves_.end(); iter++) {
-    int tmp = static_cast<PikaBinlogSenderThread*>(iter->sender)->filenum();
-    if (tmp < *max) {
-      *max = tmp;
-    }
-  }
-
-  return Status::OK();
-}
-
-
-void PikaServer::ClearBgsave() {
-  bgsave_info_.Clear();
-  bgsaving_ = false;
-}
-
 bool PikaServer::InitBgsaveEnv(const std::string& bgsave_path) {
   // Prepare for bgsave dir
   bgsave_info_.start_time = time(NULL);
@@ -375,12 +355,10 @@ bool PikaServer::RunBgsaveEngine() {
 
 void PikaServer::Bgsave() {
   // Only one thread can go through
-  bgsave_protector_.Lock();
-  if (bgsaving_) {
+  bool expect = false;
+  if (!bgsaving_.compare_exchange_strong(expect, true)) {
     return;
   }
-  bgsaving_ = true;
-  bgsave_protector_.Unlock();
 
   // Prepare for Bgsaving
   if (!InitBgsaveEnv(g_pika_conf->bgsave_path())
@@ -435,3 +413,84 @@ bool PikaServer::Bgsaveoff() {
   }
   return true;
 }
+
+bool PikaServer::PurgeLogs(uint32_t to) {
+  // Only one thread can go through
+  bool expect = false;
+  if (!purging_.compare_exchange_strong(expect, true)) {
+    return false;
+  }
+  uint32_t max = 0;
+  if (!GetPurgeWindow(max)){
+    return false;
+  }
+  LOG(WARNING) << "max seqnum could be deleted: " << max;
+  if (to > max) {
+    ClearPurge();
+    return false;
+  }
+  PurgeArg *arg = new PurgeArg();
+  arg->p = this;
+  arg->to = to;
+  // Start new thread if needed
+  if (!purge_thread_.is_running()) {
+    purge_thread_.StartThread();
+  }
+  purge_thread_.Schedule(&DoPurgeLogs, static_cast<void*>(arg));
+  return true;
+}
+
+void PikaServer::DoPurgeLogs(void* arg) {
+  PurgeArg *ppurge = static_cast<PurgeArg*>(arg);
+  PikaServer* ps = ppurge->p;
+  
+  ps->PurgeFiles(ppurge->to);
+
+  ps->ClearPurge();
+  delete (PurgeArg*)arg;
+}
+
+bool PikaServer::PurgeFiles(uint32_t to)
+{
+  std::string log_path = g_pika_conf->log_path();
+  std::vector<std::string> children;
+  int ret = slash::GetChildren(log_path, children);
+  if (ret != 0){
+    LOG(ERROR) << "Get all files in log path failed! error:" << ret; 
+    return false;
+  }
+
+  std::string filename;
+  std::vector<std::string>::iterator it;
+  for (it = children.begin(); it != children.end(); ++it) {
+    filename = *it;
+    if (filename.compare(0, kBinlogPrefixLen, kBinlogPrefix) == 0
+        && stoul(filename.substr(kBinlogPrefixLen)) <= to) {
+      slash::Status s = slash::DeleteFile(log_path + filename);
+      LOG(WARNING) << "Purge log file : " << filename;
+      if (!s.ok()) {
+        LOG(ERROR) << "Purge log file : " << filename <<  " failed! error:" << s.ToString();
+      }
+    }
+  }
+  return true;
+}
+
+bool PikaServer::GetPurgeWindow(uint32_t &max) {
+  max = logger_->version_->pro_num();
+  slash::MutexLock l(&slave_mutex_);
+  std::vector<SlaveItem>::iterator it;
+  for (it = slaves_.begin(); it != slaves_.end(); ++it) {
+    PikaBinlogSenderThread *pb = static_cast<PikaBinlogSenderThread*>((*it).sender);
+    uint32_t filenum = pb->filenum();
+    max = filenum < max ? filenum : max;
+  }
+  // remain some more
+  if (max > 10) {
+    max -= 10;
+    return true;
+  }
+  max = 0;
+  return false;
+}
+