Fix thread safety

Author: YuCai18

src/storage/src/redis.cc

@@ -26,6 +26,12 @@ Redis::Redis(Storage* const s, const DataType& type)
 }
 
 Redis::~Redis() {
+  stop_big_keys_log_ = true;
+  std::lock_guard<std::mutex> lock(bg_thread_mutex_);
+  bg_thread_cond_.notify_all();
+  std::unique_lock<std::mutex> lock(bg_thread_mutex_);
+  bg_thread_exit_cond_.wait_for(lock, std::chrono::milliseconds(200), 
+                                [this]() { return this->bg_threads_count_.load() == 0; });
   std::vector<rocksdb::ColumnFamilyHandle*> tmp_handles = handles_;
   handles_.clear();
   for (auto handle : tmp_handles) {
@@ -36,7 +42,7 @@ Redis::~Redis() {
   if (default_compact_range_options_.canceled) {
     delete default_compact_range_options_.canceled;
   }
-  stop_big_keys_log_ = true;
+
 }
 
 Status Redis::GetScanStartPoint(const Slice& key, const Slice& pattern, int64_t cursor, std::string* start_point) {
@@ -252,7 +258,9 @@ void Redis::CheckAndRecordBigKeys(
 }
 void Redis::StartBigKeysExpireCheckThread() {
   std::thread([this]() {
-    while (!stop_big_keys_log_) {
+    std::lock_guard<std::mutex> lock(bg_thread_mutex_);
+    this->bg_threads_count_++;
+    while (!this->stop_big_keys_log_.load()) {
       {
         std::lock_guard<std::mutex> lock(big_keys_mutex_);
         std::vector<std::string> expired_keys;
@@ -270,12 +278,21 @@ void Redis::StartBigKeysExpireCheckThread() {
           }
         }
       }
+      std::unique_lock<std::mutex> lock(bg_thread_mutex_);
+      bg_thread_cond_.wait_for(lock, 
+                               std::chrono::minutes(storage_->bigkeys_log_interval_),
+                               [this]() { return this->stop_big_keys_log_.load(); });
+      std::lock_guard<std::mutex> lock(bg_thread_mutex_);
+      this->bg_threads_count_--;
+      bg_thread_exit_cond_.notify_all();
       std::this_thread::sleep_for(std::chrono::minutes(storage_->bigkeys_log_interval_));
     }
   }).detach();
 }
 void Redis::StartBigKeysLogThread() {
   std::thread([this]() {
+    std::lock_guard<std::mutex> lock(bg_thread_mutex_);
+    this->bg_threads_count_++;
     while (!stop_big_keys_log_) {
       {
         uint32_t current_log_interval = storage_->bigkeys_log_interval_;
@@ -306,6 +323,13 @@ void Redis::StartBigKeysLogThread() {
           }
         }
       }
+      std::unique_lock<std::mutex> lock(bg_thread_mutex_);
+      bg_thread_cond_.wait_for(lock, 
+                               std::chrono::minutes(storage_->bigkeys_log_interval_),
+                               [this]() { return this->stop_big_keys_log_.load(); });
+      std::lock_guard<std::mutex> lock(bg_thread_mutex_);
+      this->bg_threads_count_--;
+      bg_thread_exit_cond_.notify_all();
       std::this_thread::sleep_for(std::chrono::minutes(storage_->bigkeys_log_interval_));
     }
   }).detach();

src/storage/src/redis.h

@@ -20,7 +20,7 @@
 #include "src/lru_cache.h"
 #include "src/mutex_impl.h"
 #include "storage/storage.h"
-
+#include <condition_variable>
 namespace storage {
 using Status = rocksdb::Status;
 using Slice = rocksdb::Slice;
@@ -132,6 +132,10 @@ class Redis {
   std::unordered_map<std::string, BigKeyInfo> big_keys_info_map_;
   uint64_t big_keys_threshold_;
   std::atomic<bool> stop_big_keys_log_{false};
+  std::mutex bg_thread_mutex_;
+  std::condition_variable bg_thread_cond_;
+  std::condition_variable bg_thread_exit_cond_;
+  std::atomic<int> bg_threads_count_{0};
   void CheckAndRecordBigKeys(const std::string& key, DataType type, uint64_t member_size, uint64_t key_length = 0, uint64_t value_length = 0, bool is_delete = false);
   void StartBigKeysLogThread();
   void StartBigKeysExpireCheckThread();