fix: db cannot resume after disk has no space (OpenAtomFoundation#1765)

* add auto resume

* update resume code

* add conf code

* format code

* add min-check-resume-ratio

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

@@ -507,6 +507,7 @@ class PikaServer : public pstd::noncopyable {
  * TimingTask use
  */
  void DoTimingTask();
+ void AutoResumeDB();
  void AutoCompactRange();
  void AutoPurge();
  void AutoDeleteExpiredDump();
@@ -537,6 +538,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

@@ -51,6 +51,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
@@ -1322,6 +1323,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
@@ -1515,6 +1518,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