Revert "fix lock (#2372)"

This reverts commit b743303.

include/pika_db.h

@@ -129,6 +129,9 @@ class DB : public std::enable_shared_from_this<DB>, public pstd::noncopyable {
   void SetCompactRangeOptions(const bool is_canceled);
 
   std::shared_ptr<pstd::lock::LockMgr> LockMgr();
+  void DbRWLockWriter();
+  void DbRWLockReader();
+  void DbRWUnLock();
   /*
    * Cache used
    */
@@ -161,6 +164,7 @@ class DB : public std::enable_shared_from_this<DB>, public pstd::noncopyable {
   std::string log_path_;
   std::string bgsave_sub_path_;
   pstd::Mutex key_info_protector_;
+  std::shared_mutex db_rwlock_;
   std::atomic<bool> binlog_io_error_;
   std::shared_mutex dbs_rw_;
   // class may be shared, using shared_ptr would be a better choice

include/pika_repl_server_conn.h

@@ -32,6 +32,9 @@ class PikaReplServerConn : public net::PbConn {
                                      const InnerMessage::InnerRequest::TrySync& try_sync_request,
                                      const std::shared_ptr<net::PbConn>& conn,
                                      InnerMessage::InnerResponse::TrySync* try_sync_response);
+  static void BuildConsensusMeta(const bool& reject, const std::vector<LogOffset>& hints, const uint32_t& term,
+                                 InnerMessage::InnerResponse* response);
+
   static void HandleDBSyncRequest(void* arg);
   static void HandleBinlogSyncRequest(void* arg);
   static void HandleRemoveSlaveNodeRequest(void* arg);

src/pika_admin.cc

@@ -823,9 +823,9 @@ void ShutdownCmd::DoInitial() {
 // no return
 void ShutdownCmd::Do() {
   DLOG(WARNING) << "handle \'shutdown\'";
-  db_->DBUnlockShared();
+  db_->DbRWUnLock();
   g_pika_server->Exit();
-  db_->DBLockShared();
+  db_->DbRWLockReader();
   res_.SetRes(CmdRes::kNone);
 }
 
@@ -1347,11 +1347,11 @@ void InfoCmd::InfoData(std::string& info) {
     }
     background_errors.clear();
     memtable_usage = table_reader_usage = 0;
-    db_item.second->DBLockShared();
+    db_item.second->DbRWLockReader();
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_CUR_SIZE_ALL_MEM_TABLES, &memtable_usage);
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_ESTIMATE_TABLE_READER_MEM, &table_reader_usage);
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_BACKGROUND_ERRORS, &background_errors);
-    db_item.second->DBUnlockShared();
+    db_item.second->DbRWUnLock();
     total_memtable_usage += memtable_usage;
     total_table_reader_usage += table_reader_usage;
     for (const auto& item : background_errors) {
@@ -1385,9 +1385,9 @@ void InfoCmd::InfoRocksDB(std::string& info) {
       continue;
     }
     std::string rocksdb_info;
-    db_item.second->DBLockShared();
+    db_item.second->DbRWLockReader();
     db_item.second->storage()->GetRocksDBInfo(rocksdb_info);
-    db_item.second->DBUnlockShared();
+    db_item.second->DbRWUnLock();
     tmp_stream << rocksdb_info;
   }
   info.append(tmp_stream.str());
@@ -3158,9 +3158,9 @@ void DiskRecoveryCmd::Do() {
     }
     db_item.second->SetBinlogIoErrorrelieve();
     background_errors_.clear();
-    db_item.second->DBLockShared();
+    db_item.second->DbRWLockReader();
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_BACKGROUND_ERRORS, &background_errors_);
-    db_item.second->DBUnlockShared();
+    db_item.second->DbRWUnLock();
     for (const auto &item: background_errors_) {
       if (item.second != 0) {
         rocksdb::Status s = db_item.second->storage()->GetDBByIndex(item.first)->Resume();

src/pika_command.cc

@@ -889,11 +889,11 @@ void Cmd::InternalProcessCommand(const HintKeys& hint_keys) {
 
 void Cmd::DoCommand(const HintKeys& hint_keys) {
   if (!IsSuspend()) {
-    db_->DBLockShared();
+    db_->DbRWLockReader();
   }
   DEFER {
     if (!IsSuspend()) {
-      db_->DBUnlockShared();
+      db_->DbRWUnLock();
     }
   };
   if (IsNeedCacheDo()

src/pika_db.cc

@@ -72,6 +72,8 @@ void DB::SetBinlogIoError() { return binlog_io_error_.store(true); }
 void DB::SetBinlogIoErrorrelieve() { return binlog_io_error_.store(false); }
 bool DB::IsBinlogIoError() { return binlog_io_error_.load(); }
 std::shared_ptr<pstd::lock::LockMgr> DB::LockMgr() { return lock_mgr_; }
+void DB::DbRWLockReader() { db_rwlock_.lock_shared(); }
+void DB::DbRWUnLock() { db_rwlock_.unlock(); }
 std::shared_ptr<PikaCache> DB::cache() const { return cache_; }
 std::shared_ptr<storage::Storage> DB::storage() const { return storage_; }
 
@@ -195,6 +197,8 @@ void DB::SetCompactRangeOptions(const bool is_canceled) {
   storage_->SetCompactRangeOptions(is_canceled);
 }
 
+void DB::DbRWLockWriter() { db_rwlock_.lock(); }
+
 DisplayCacheInfo DB::GetCacheInfo() {
   std::lock_guard l(key_info_protector_);
   return cache_info_;
@@ -356,7 +360,7 @@ bool DB::InitBgsaveEngine() {
   }
 
   {
-    std::lock_guard lock(dbs_rw_);
+    std::lock_guard lock(db_rwlock_);
     LogOffset bgsave_offset;
     // term, index are 0
     db->Logger()->GetProducerStatus(&(bgsave_offset.b_offset.filenum), &(bgsave_offset.b_offset.offset));
@@ -544,7 +548,7 @@ bool DB::ChangeDb(const std::string& new_path) {
   tmp_path += "_bak";
   pstd::DeleteDirIfExist(tmp_path);
 
-  std::lock_guard l(dbs_rw_);
+  std::lock_guard l(db_rwlock_);
   LOG(INFO) << "DB: " << db_name_ << ", Prepare change db from: " << tmp_path;
   storage_.reset();
 
@@ -576,7 +580,7 @@ void DB::ClearBgsave() {
 }
 
 bool DB::FlushSubDB(const std::string& db_name) {
-  std::lock_guard rwl(dbs_rw_);
+  std::lock_guard rwl(db_rwlock_);
   return FlushSubDBWithoutLock(db_name);
 }
 
@@ -630,7 +634,7 @@ void DB::ResetDisplayCacheInfo(int status) {
 }
 
 bool DB::FlushDB() {
-  std::lock_guard rwl(dbs_rw_);
+  std::lock_guard rwl(db_rwlock_);
   std::lock_guard l(bgsave_protector_);
   return FlushDBWithoutLock();
 }

src/pika_repl_bgworker.cc

@@ -219,7 +219,7 @@ void PikaReplBgWorker::HandleBGWorkerWriteDB(void* arg) {
   pstd::lock::MultiRecordLock record_lock(c_ptr->GetDB()->LockMgr());
   record_lock.Lock(c_ptr->current_key());
   if (!c_ptr->IsSuspend()) {
-    c_ptr->GetDB()->DBLockShared();
+    c_ptr->GetDB()->DbRWLockReader();
   }
   if (c_ptr->IsNeedCacheDo()
       && PIKA_CACHE_NONE != g_pika_conf->cache_model()
@@ -236,7 +236,7 @@ void PikaReplBgWorker::HandleBGWorkerWriteDB(void* arg) {
     c_ptr->Do();
   }
   if (!c_ptr->IsSuspend()) {
-    c_ptr->GetDB()->DBUnlockShared();
+    c_ptr->GetDB()->DbRWUnLock();
   }
 
   if (c_ptr->res().ok()

src/pika_repl_server_conn.cc

@@ -216,6 +216,23 @@ bool PikaReplServerConn::TrySyncOffsetCheck(const std::shared_ptr<SyncMasterDB>&
   return true;
 }
 
+void PikaReplServerConn::BuildConsensusMeta(const bool& reject, const std::vector<LogOffset>& hints,
+                                            const uint32_t& term, InnerMessage::InnerResponse* response) {
+  InnerMessage::ConsensusMeta* consensus_meta = response->mutable_consensus_meta();
+  consensus_meta->set_term(term);
+  consensus_meta->set_reject(reject);
+  if (!reject) {
+    return;
+  }
+  for (const auto& hint : hints) {
+    InnerMessage::BinlogOffset* offset = consensus_meta->add_hint();
+    offset->set_filenum(hint.b_offset.filenum);
+    offset->set_offset(hint.b_offset.offset);
+    offset->set_term(hint.l_offset.term);
+    offset->set_index(hint.l_offset.index);
+  }
+}
+
 void PikaReplServerConn::HandleDBSyncRequest(void* arg) {
   std::unique_ptr<ReplServerTaskArg> task_arg(static_cast<ReplServerTaskArg*>(arg));
   const std::shared_ptr<InnerMessage::InnerRequest> req = task_arg->req;

src/pika_server.cc

@@ -347,9 +347,9 @@ bool PikaServer::IsKeyScaning() {
 bool PikaServer::IsCompacting() {
   std::shared_lock db_rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-    db_item.second->DBLockShared();
+    db_item.second->DbRWLockReader();
     std::string task_type = db_item.second->storage()->GetCurrentTaskType();
-    db_item.second->DBUnlockShared();
+    db_item.second->DbRWUnLock();
     if (strcasecmp(task_type.data(), "no") != 0) {
       return true;
     }
@@ -446,27 +446,27 @@ void PikaServer::PrepareDBTrySync() {
 void PikaServer::DBSetMaxCacheStatisticKeys(uint32_t max_cache_statistic_keys) {
   std::shared_lock rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-    db_item.second->DBLockShared();
+    db_item.second->DbRWLockReader();
     db_item.second->storage()->SetMaxCacheStatisticKeys(max_cache_statistic_keys);
-    db_item.second->DBUnlockShared();
+    db_item.second->DbRWUnLock();
   }
 }
 
 void PikaServer::DBSetSmallCompactionThreshold(uint32_t small_compaction_threshold) {
   std::shared_lock rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-    db_item.second->DBLockShared();
+    db_item.second->DbRWLockReader();
     db_item.second->storage()->SetSmallCompactionThreshold(small_compaction_threshold);
-    db_item.second->DBUnlockShared();
+    db_item.second->DbRWUnLock();
   }
 }
 
 void PikaServer::DBSetSmallCompactionDurationThreshold(uint32_t small_compaction_duration_threshold) {
   std::shared_lock rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-    db_item.second->DBLockShared();
+    db_item.second->DbRWLockReader();
     db_item.second->storage()->SetSmallCompactionDurationThreshold(small_compaction_duration_threshold);
-    db_item.second->DBUnlockShared();
+    db_item.second->DbRWUnLock();
   }
 }
 
@@ -1386,7 +1386,9 @@ storage::Status PikaServer::RewriteStorageOptions(const storage::OptionType& opt
   storage::Status s;
   std::shared_lock db_rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
+    db_item.second->DbRWLockWriter();
     s = db_item.second->storage()->SetOptions(option_type, storage::ALL_DB, options_map);
+    db_item.second->DbRWUnLock();
     if (!s.ok()) {
       return s;
     }
@@ -1565,9 +1567,9 @@ void PikaServer::DisableCompact() {
   /* cancel in-progress manual compactions */
   std::shared_lock rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-      db_item.second->DBLock();
+      db_item.second->DbRWLockWriter();
       db_item.second->SetCompactRangeOptions(true);
-      db_item.second->DBUnlock();
+      db_item.second->DbRWUnLock();
   }
 }
 

src/pika_transaction.cc

@@ -145,22 +145,22 @@ void ExecCmd::Lock() {
     g_pika_rm->DBLock();
   }
 
-  std::for_each(r_lock_dbs_.begin(), r_lock_dbs_.end(), [this](auto& need_lock_db) {
-    if (lock_db_keys_.count(need_lock_db) != 0) {
-      pstd::lock::MultiRecordLock record_lock(need_lock_db->LockMgr());
-      record_lock.Lock(lock_db_keys_[need_lock_db]);
+  std::for_each(r_lock_dbs_.begin(), r_lock_dbs_.end(), [this](auto& need_lock_slot) {
+    if (lock_db_keys_.count(need_lock_slot) != 0) {
+      pstd::lock::MultiRecordLock record_lock(need_lock_slot->LockMgr());
+      record_lock.Lock(lock_db_keys_[need_lock_slot]);
     }
-    need_lock_db->DBLockShared();
+    need_lock_slot->DbRWLockReader();
   });
 }
 
 void ExecCmd::Unlock() {
-  std::for_each(r_lock_dbs_.begin(), r_lock_dbs_.end(), [this](auto& need_lock_db) {
-    if (lock_db_keys_.count(need_lock_db) != 0) {
-      pstd::lock::MultiRecordLock record_lock(need_lock_db->LockMgr());
-      record_lock.Unlock(lock_db_keys_[need_lock_db]);
+  std::for_each(r_lock_dbs_.begin(), r_lock_dbs_.end(), [this](auto& need_lock_slot) {
+    if (lock_db_keys_.count(need_lock_slot) != 0) {
+      pstd::lock::MultiRecordLock record_lock(need_lock_slot->LockMgr());
+      record_lock.Unlock(lock_db_keys_[need_lock_slot]);
     }
-    need_lock_db->DBUnlockShared();
+    need_lock_slot->DbRWUnLock();
   });
   if (is_lock_rm_dbs_) {
     g_pika_rm->DBUnlock();

src/storage/src/redis_hashes.cc

@@ -122,6 +122,7 @@ Status Redis::HDel(const Slice& key, const std::vector<std::string>& fields, int
   std::string meta_value;
   int32_t del_cnt = 0;
   uint64_t version = 0;
+  ScopeRecordLock l(lock_mgr_, key);
   ScopeSnapshot ss(db_, &snapshot);
   read_options.snapshot = snapshot;
 
@@ -280,6 +281,7 @@ Status Redis::HGetallWithTTL(const Slice& key, std::vector<FieldValue>* fvs, int
 Status Redis::HIncrby(const Slice& key, const Slice& field, int64_t value, int64_t* ret) {
   *ret = 0;
   rocksdb::WriteBatch batch;
+  ScopeRecordLock l(lock_mgr_, key);
 
   uint64_t version = 0;
   uint32_t statistic = 0;
@@ -357,6 +359,7 @@ Status Redis::HIncrby(const Slice& key, const Slice& field, int64_t value, int64
 Status Redis::HIncrbyfloat(const Slice& key, const Slice& field, const Slice& by, std::string* new_value) {
   new_value->clear();
   rocksdb::WriteBatch batch;
+  ScopeRecordLock l(lock_mgr_, key);
 
   uint64_t version = 0;
   uint32_t statistic = 0;
@@ -550,6 +553,7 @@ Status Redis::HMSet(const Slice& key, const std::vector<FieldValue>& fvs) {
   }
 
   rocksdb::WriteBatch batch;
+  ScopeRecordLock l(lock_mgr_, key);
 
   uint64_t version = 0;
   std::string meta_value;
@@ -613,6 +617,7 @@ Status Redis::HMSet(const Slice& key, const std::vector<FieldValue>& fvs) {
 
 Status Redis::HSet(const Slice& key, const Slice& field, const Slice& value, int32_t* res) {
   rocksdb::WriteBatch batch;
+  ScopeRecordLock l(lock_mgr_, key);
 
   uint64_t version = 0;
   uint32_t statistic = 0;
@@ -677,6 +682,7 @@ Status Redis::HSet(const Slice& key, const Slice& field, const Slice& value, int
 
 Status Redis::HSetnx(const Slice& key, const Slice& field, const Slice& value, int32_t* ret) {
   rocksdb::WriteBatch batch;
+  ScopeRecordLock l(lock_mgr_, key);
 
   uint64_t version = 0;
   std::string meta_value;
@@ -1024,6 +1030,8 @@ Status Redis::PKHRScanRange(const Slice& key, const Slice& field_start, const st
 
 Status Redis::HashesExpire(const Slice& key, int64_t ttl) {
   std::string meta_value;
+  ScopeRecordLock l(lock_mgr_, key);
+
   BaseMetaKey base_meta_key(key);
   Status s = db_->Get(default_read_options_, handles_[kHashesMetaCF], base_meta_key.Encode(), &meta_value);
   if (s.ok()) {
@@ -1047,6 +1055,8 @@ Status Redis::HashesExpire(const Slice& key, int64_t ttl) {
 
 Status Redis::HashesDel(const Slice& key) {
   std::string meta_value;
+  ScopeRecordLock l(lock_mgr_, key);
+
   BaseMetaKey base_meta_key(key);
   Status s = db_->Get(default_read_options_, handles_[kHashesMetaCF], base_meta_key.Encode(), &meta_value);
   if (s.ok()) {
@@ -1067,6 +1077,8 @@ Status Redis::HashesDel(const Slice& key) {
 
 Status Redis::HashesExpireat(const Slice& key, int64_t timestamp) {
   std::string meta_value;
+  ScopeRecordLock l(lock_mgr_, key);
+
   BaseMetaKey base_meta_key(key);
   Status s = db_->Get(default_read_options_, handles_[kHashesMetaCF], base_meta_key.Encode(), &meta_value);
   if (s.ok()) {
@@ -1089,6 +1101,8 @@ Status Redis::HashesExpireat(const Slice& key, int64_t timestamp) {
 
 Status Redis::HashesPersist(const Slice& key) {
   std::string meta_value;
+  ScopeRecordLock l(lock_mgr_, key);
+
   BaseMetaKey base_meta_key(key);
   Status s = db_->Get(default_read_options_, handles_[kHashesMetaCF], base_meta_key.Encode(), &meta_value);
   if (s.ok()) {