fix: db cannot resume after disk has no space

conf/pika.conf

@@ -217,6 +217,22 @@ slave-priority : 100
 # [NOTICE]: compact-interval is prior than compact-cron.
 #compact-interval :
 
+# The minimum disk usage ratio for checking resume.
+# If the disk usage ratio is lower than min-check-resume-ratio, it will not check resume, only higher will check resume.
+# Its default value is 0.7.
+#min-check-resume-ratio : 0.7
+
+# The minimum free disk space to trigger db resume.
+# If the db has a background error, only the free disk size is larger than this configuration can trigger manually resume db.
+# Its default value is 256MB.
+# [NOTICE]: least-free-disk-resume-size should not smaller than write-buffer-size!
+#least-free-disk-resume-size : 256M
+
+# Manually trying to resume db interval is configured by manually-resume-interval.
+# If db has a background error, it will try to manually call resume() to resume db if satisfy the least free disk to resume.
+# Its default value is 60 seconds.
+#manually-resume-interval : 60
+
 # This window-size determines the amount of data that can be transmitted in a single synchronization process.
 # [Tip] In the scenario of high network latency. Increasing this size can improve synchronization efficiency.
 # Its default value is 9000. the [maximum] value is 90000.

include/pika_conf.h

@@ -86,6 +86,18 @@ class PikaConf : public pstd::BaseConf {
     std::shared_lock l(rwlock_);
     return compact_interval_;
   }
+  int64_t least_resume_free_disk_size() {
+    std::shared_lock l(rwlock_);
+    return least_free_disk_to_resume_;
+  }
+  int64_t resume_interval() {
+    std::shared_lock l(rwlock_);
+    return resume_check_interval_;
+  }
+  double min_check_resume_ratio() {
+    std::shared_lock l(rwlock_);
+    return min_check_resume_ratio_;
+  }
   int64_t write_buffer_size() {
     std::shared_lock l(rwlock_);
     return write_buffer_size_;
@@ -477,6 +489,21 @@ class PikaConf : public pstd::BaseConf {
     TryPushDiffCommands("compact-interval", value);
     compact_interval_ = value;
   }
+  void SetLeastResumeFreeDiskSize(const int64_t& value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("least-free-disk-resume-size", std::to_string(value));
+    least_free_disk_to_resume_ = value;
+  }
+  void SetResumeInterval(const int64_t& value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("manually-resume-interval", std::to_string(value));
+    resume_check_interval_ = value;
+  }
+  void SetMinCheckResumeRatio(const double& value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("min-check-resume-ratio", std::to_string(value));
+    min_check_resume_ratio_ = value;
+  }
   void SetSyncWindowSize(const int& value) {
     TryPushDiffCommands("sync-window-size", std::to_string(value));
     sync_window_size_.store(value);
@@ -546,6 +573,9 @@ class PikaConf : public pstd::BaseConf {
   int db_sync_speed_ = 0;
   std::string compact_cron_;
   std::string compact_interval_;
+  int64_t resume_check_interval_ = 60; // seconds
+  int64_t least_free_disk_to_resume_ = 268435456; // 256 MB
+  double min_check_resume_ratio_ = 0.7;
   int64_t write_buffer_size_ = 0;
   int64_t arena_block_size_ = 0;
   int64_t slotmigrate_thread_num_ = 0;

include/pika_server.h

@@ -504,6 +504,7 @@ class PikaServer : public pstd::noncopyable {
    * TimingTask use
    */
   void DoTimingTask();
+  void AutoResumeDB();
   void AutoCompactRange();
   void AutoPurge();
   void AutoDeleteExpiredDump();
@@ -534,6 +535,11 @@ class PikaServer : public pstd::noncopyable {
   bool have_scheduled_crontask_ = false;
   struct timeval last_check_compact_time_;
 
+  /*
+   * ResumeDB used
+   */
+  struct timeval last_check_resume_time_;
+
   /*
    * Communicate with the client used
    */

src/pika_conf.cc

@@ -354,6 +354,22 @@ int PikaConf::Load() {
     }
   }
 
+  // least-free-disk-resume-size
+  GetConfInt64Human("least-free-disk-resume-size", &least_free_disk_to_resume_);
+  if (least_free_disk_to_resume_ <= 0) {
+    least_free_disk_to_resume_ = 268435456;  // 256Mb
+  }
+
+  GetConfInt64("manually-resume-interval", &resume_check_interval_);
+  if (resume_check_interval_ <= 0) {
+    resume_check_interval_ = 60; // seconds
+  }
+
+  GetConfDouble("min-check-resume-ratio", &min_check_resume_ratio_);
+  if (min_check_resume_ratio_ < 0) {
+    min_check_resume_ratio_ = 0.7;
+  }
+
   // write_buffer_size
   GetConfInt64Human("write-buffer-size", &write_buffer_size_);
   if (write_buffer_size_ <= 0) {
@@ -621,6 +637,9 @@ int PikaConf::ConfigRewrite() {
   SetConfInt("db-sync-speed", db_sync_speed_);
   SetConfStr("compact-cron", compact_cron_);
   SetConfStr("compact-interval", compact_interval_);
+  SetConfInt64("least-free-disk-resume-size", least_free_disk_to_resume_);
+  SetConfInt64("manually-resume-interval", resume_check_interval_);
+  SetConfDouble("min-check-resume-ratio", min_check_resume_ratio_);
   SetConfInt("slave-priority", slave_priority_);
   SetConfInt("sync-window-size", sync_window_size_.load());
   SetConfInt("consensus-level", consensus_level_.load());

src/pika_server.cc

@@ -52,6 +52,7 @@ PikaServer::PikaServer()
     : exit_(false),
       slot_state_(INFREE),
       last_check_compact_time_({0, 0}),
+      last_check_resume_time_({0, 0}),
       repl_state_(PIKA_REPL_NO_CONNECT),
       role_(PIKA_ROLE_SINGLE) {
   // Init server ip host
@@ -1295,6 +1296,8 @@ void PikaServer::PubSubNumSub(const std::vector<std::string>& channels,
 /******************************* PRIVATE *******************************/
 
 void PikaServer::DoTimingTask() {
+  // Resume DB if satisfy the condition
+  AutoResumeDB();
   // Maybe schedule compactrange
   AutoCompactRange();
   // Purge log
@@ -1487,6 +1490,57 @@ void PikaServer::AutoKeepAliveRSync() {
   }
 }
 
+void PikaServer::AutoResumeDB() {
+  struct statfs disk_info;
+  int ret = statfs(g_pika_conf->db_path().c_str(), &disk_info);
+  if (ret == -1) {
+    LOG(WARNING) << "statfs error: " << strerror(errno);
+    return;
+  }
+
+  int64_t interval = g_pika_conf->resume_interval();
+  int64_t least_free_size = g_pika_conf->least_resume_free_disk_size();
+  double min_check_resume_ratio = g_pika_conf->min_check_resume_ratio();
+  uint64_t free_size = disk_info.f_bsize * disk_info.f_bfree;
+  uint64_t total_size = disk_info.f_bsize * disk_info.f_blocks;
+  double disk_use_ratio = 1.0 - static_cast<double>(free_size) / static_cast<double>(total_size);
+
+  struct timeval now;
+  gettimeofday(&now, nullptr);
+  // first check or time interval between now and last check is larger than variable "interval"
+  if (last_check_resume_time_.tv_sec == 0 || now.tv_sec - last_check_resume_time_.tv_sec >= interval) {
+    gettimeofday(&last_check_resume_time_, nullptr);
+    if (disk_use_ratio < min_check_resume_ratio || free_size < least_free_size){
+      return;
+    }
+
+    std::map<std::string, uint64_t> background_errors;
+    std::shared_lock db_rwl(g_pika_server->dbs_rw_);
+    // loop every db
+    for (const auto& db_item : g_pika_server->dbs_) {
+      if (!db_item.second) {
+        continue;
+      }
+      std::shared_lock slot_rwl(db_item.second->slots_rw_);
+      // loop every slot
+      for (const auto& slot_item : db_item.second->slots_) {
+        background_errors.clear();
+        slot_item.second->DbRWLockReader();
+        slot_item.second->db()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_BACKGROUND_ERRORS, &background_errors);
+        slot_item.second->DbRWUnLock();
+        for (const auto& item : background_errors) {
+          if (item.second != 0) {
+            rocksdb::Status s = slot_item.second->db()->GetDBByType(item.first)->Resume();
+            if (!s.ok()) {
+              LOG(WARNING) << s.ToString();
+            }
+          }
+        }
+      }
+    }
+  }
+}
+
 void PikaServer::AutoUpdateNetworkMetric() {
   monotime current_time = getMonotonicUs();
   size_t factor = 5e6; // us, 5s