fix: fix dead lock bug (#2529)

.github/workflows/pika.yml

@@ -141,6 +141,7 @@ jobs:
       - name: Start codis, pika master and pika slave
         working-directory: ${{ github.workspace }}/build
         run: |
+          rm -rf ../deps
           chmod +x ../tests/integration/start_master_and_slave.sh
           ../tests/integration/start_master_and_slave.sh
           chmod +x ../tests/integration/start_codis.sh

codis/config/proxy.toml

@@ -105,10 +105,10 @@ session_break_on_failure = false
 # Slowlog-log-slower-than(us), from receive command to send response, 0 is allways print slow log
 slowlog_log_slower_than = 100000
 
-# quick command list e.g. get, set
-quick_cmd_list = ""
-# slow command list e.g. hgetall, mset
-slow_cmd_list = ""
+# quick command list
+quick_cmd_list = "get,set"
+# slow command list
+slow_cmd_list = "mget, mset"
 
 # Set metrics server (such as http://localhost:28000), proxy will report json formatted metrics to specified server in a predefined period.
 metrics_report_server = ""

include/pika_db.h

@@ -101,9 +101,6 @@ class DB : public std::enable_shared_from_this<DB>, public pstd::noncopyable {
   std::shared_mutex& GetDBLock() {
     return dbs_rw_;
   }
-  std::shared_mutex& GetDBLocks() {
-    return db_rwlock_;
-  }
   void DBLock() {
     dbs_rw_.lock();
   }
@@ -132,9 +129,6 @@ 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
    */
@@ -167,7 +161,6 @@ 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,7 +32,6 @@ 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 HandleDBSyncRequest(void* arg);
   static void HandleBinlogSyncRequest(void* arg);
   static void HandleRemoveSlaveNodeRequest(void* arg);

src/pika_admin.cc

@@ -709,7 +709,7 @@ void FlushdbCmd::Execute() {
     if (db_->IsKeyScaning()) {
       res_.SetRes(CmdRes::kErrOther, "The keyscan operation is executing, Try again later");
     } else {
-      std::lock_guard l_prw(db_->GetDBLocks());
+      std::lock_guard l_prw(db_->GetDBLock());
       std::lock_guard s_prw(g_pika_rm->GetDBLock());
       FlushAllDBsWithoutLock();
       res_.SetRes(CmdRes::kOk);
@@ -832,9 +832,9 @@ void ShutdownCmd::DoInitial() {
 // no return
 void ShutdownCmd::Do() {
   DLOG(WARNING) << "handle \'shutdown\'";
-  db_->DbRWUnLock();
+  db_->DBUnlockShared();
   g_pika_server->Exit();
-  db_->DbRWLockReader();
+  db_->DBLockShared();
   res_.SetRes(CmdRes::kNone);
 }
 
@@ -1356,11 +1356,11 @@ void InfoCmd::InfoData(std::string& info) {
     }
     background_errors.clear();
     memtable_usage = table_reader_usage = 0;
-    db_item.second->DbRWLockReader();
+    db_item.second->DBLockShared();
     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->DbRWUnLock();
+    db_item.second->DBUnlockShared();
     total_memtable_usage += memtable_usage;
     total_table_reader_usage += table_reader_usage;
     for (const auto& item : background_errors) {
@@ -1394,9 +1394,9 @@ void InfoCmd::InfoRocksDB(std::string& info) {
       continue;
     }
     std::string rocksdb_info;
-    db_item.second->DbRWLockReader();
+    db_item.second->DBLockShared();
     db_item.second->storage()->GetRocksDBInfo(rocksdb_info);
-    db_item.second->DbRWUnLock();
+    db_item.second->DBUnlockShared();
     tmp_stream << rocksdb_info;
   }
   info.append(tmp_stream.str());
@@ -3141,9 +3141,9 @@ void DiskRecoveryCmd::Do() {
     }
     db_item.second->SetBinlogIoErrorrelieve();
     background_errors_.clear();
-    db_item.second->DbRWLockReader();
+    db_item.second->DBLockShared();
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_BACKGROUND_ERRORS, &background_errors_);
-    db_item.second->DbRWUnLock();
+    db_item.second->DBUnlockShared();
     for (const auto &item: background_errors_) {
       if (item.second != 0) {
         rocksdb::Status s = db_item.second->storage()->GetDBByType(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_->DbRWLockReader();
+    db_->DBLockShared();
   }
   DEFER {
     if (!IsSuspend()) {
-      db_->DbRWUnLock();
+      db_->DBUnlockShared();
     }
   };
   if (IsNeedCacheDo()

src/pika_db.cc

@@ -71,8 +71,6 @@ 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_; }
 
@@ -196,8 +194,6 @@ 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_;
@@ -357,7 +353,7 @@ bool DB::InitBgsaveEngine() {
   }
 
   {
-    std::lock_guard lock(db_rwlock_);
+    std::lock_guard lock(dbs_rw_);
     LogOffset bgsave_offset;
     // term, index are 0
     db->Logger()->GetProducerStatus(&(bgsave_offset.b_offset.filenum), &(bgsave_offset.b_offset.offset));
@@ -546,7 +542,7 @@ bool DB::ChangeDb(const std::string& new_path) {
   tmp_path += "_bak";
   pstd::DeleteDirIfExist(tmp_path);
 
-  std::lock_guard l(db_rwlock_);
+  std::lock_guard l(dbs_rw_);
   LOG(INFO) << "DB: " << db_name_ << ", Prepare change db from: " << tmp_path;
   storage_.reset();
 
@@ -577,7 +573,7 @@ void DB::ClearBgsave() {
 }
 
 bool DB::FlushSubDB(const std::string& db_name) {
-  std::lock_guard rwl(db_rwlock_);
+  std::lock_guard rwl(dbs_rw_);
   return FlushSubDBWithoutLock(db_name);
 }
 
@@ -631,7 +627,7 @@ void DB::ResetDisplayCacheInfo(int status) {
 }
 
 bool DB::FlushDB() {
-  std::lock_guard rwl(db_rwlock_);
+  std::lock_guard rwl(dbs_rw_);
   std::lock_guard l(bgsave_protector_);
   return FlushDBWithoutLock();
 }

src/pika_repl_bgworker.cc

@@ -221,7 +221,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()->DbRWLockReader();
+    c_ptr->GetDB()->DBLockShared();
   }
   if (c_ptr->IsNeedCacheDo()
       && PIKA_CACHE_NONE != g_pika_conf->cache_model()
@@ -238,7 +238,7 @@ void PikaReplBgWorker::HandleBGWorkerWriteDB(void* arg) {
     c_ptr->Do();
   }
   if (!c_ptr->IsSuspend()) {
-    c_ptr->GetDB()->DbRWUnLock();
+    c_ptr->GetDB()->DBUnlockShared();
   }
   record_lock.Unlock(c_ptr->current_key());
   if (g_pika_conf->slowlog_slower_than() >= 0) {

src/pika_server.cc

@@ -346,9 +346,9 @@ bool PikaServer::IsKeyScaning() {
 bool PikaServer::IsCompacting() {
   std::shared_lock db_rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-    db_item.second->DbRWLockReader();
+    db_item.second->DBLockShared();
     std::string task_type = db_item.second->storage()->GetCurrentTaskType();
-    db_item.second->DbRWUnLock();
+    db_item.second->DBUnlockShared();
     if (strcasecmp(task_type.data(), "no") != 0) {
       return true;
     }
@@ -445,27 +445,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->DbRWLockReader();
+    db_item.second->DBLockShared();
     db_item.second->storage()->SetMaxCacheStatisticKeys(max_cache_statistic_keys);
-    db_item.second->DbRWUnLock();
+    db_item.second->DBUnlockShared();
   }
 }
 
 void PikaServer::DBSetSmallCompactionThreshold(uint32_t small_compaction_threshold) {
   std::shared_lock rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-    db_item.second->DbRWLockReader();
+    db_item.second->DBLockShared();
     db_item.second->storage()->SetSmallCompactionThreshold(small_compaction_threshold);
-    db_item.second->DbRWUnLock();
+    db_item.second->DBUnlockShared();
   }
 }
 
 void PikaServer::DBSetSmallCompactionDurationThreshold(uint32_t small_compaction_duration_threshold) {
   std::shared_lock rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-    db_item.second->DbRWLockReader();
+    db_item.second->DBLockShared();
     db_item.second->storage()->SetSmallCompactionDurationThreshold(small_compaction_duration_threshold);
-    db_item.second->DbRWUnLock();
+    db_item.second->DBUnlockShared();
   }
 }
 
@@ -1381,9 +1381,7 @@ 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;
     }
@@ -1562,9 +1560,9 @@ void PikaServer::DisableCompact() {
   /* cancel in-progress manual compactions */
   std::shared_lock rwl(dbs_rw_);
   for (const auto& db_item : dbs_) {
-      db_item.second->DbRWLockWriter();
+      db_item.second->DBLock();
       db_item.second->SetCompactRangeOptions(true);
-      db_item.second->DbRWUnLock();
+      db_item.second->DBUnlock();
   }
 }
 

src/pika_transaction.cc

@@ -146,22 +146,22 @@ void ExecCmd::Lock() {
     g_pika_rm->DBLock();
   }
 
-  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]);
+  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]);
     }
-    need_lock_slot->DbRWLockReader();
+    need_lock_db->DBLockShared();
   });
 }
 
 void ExecCmd::Unlock() {
-  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]);
+  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]);
     }
-    need_lock_slot->DbRWUnLock();
+    need_lock_db->DBUnlockShared();
   });
   if (is_lock_rm_dbs_) {
     g_pika_rm->DBUnlock();

tests/integration/main_test.go

@@ -16,7 +16,7 @@ var (
 func TestPikaWithCache(t *testing.T) {
 	GlobalBefore = func(ctx context.Context, client *redis.Client) {
 		Expect(client.SlaveOf(ctx, "NO", "ONE").Err()).NotTo(HaveOccurred())
-		Expect(client.FlushAll(ctx).Err()).NotTo(HaveOccurred())
+		//Expect(client.FlushAll(ctx).Err()).NotTo(HaveOccurred())
 		Expect(client.ConfigSet(ctx, "cache-model", "1").Err()).NotTo(HaveOccurred())
 	}
 	RegisterFailHandler(Fail)
@@ -26,7 +26,7 @@ func TestPikaWithCache(t *testing.T) {
 func TestPikaWithoutCache(t *testing.T) {
 	GlobalBefore = func(ctx context.Context, client *redis.Client) {
 		Expect(client.SlaveOf(ctx, "NO", "ONE").Err()).NotTo(HaveOccurred())
-		Expect(client.FlushAll(ctx).Err()).NotTo(HaveOccurred())
+		//Expect(client.FlushAll(ctx).Err()).NotTo(HaveOccurred())
 		Expect(client.ConfigSet(ctx, "cache-model", "0").Err()).NotTo(HaveOccurred())
 	}
 	RegisterFailHandler(Fail)

tests/integration/replication_test.go

@@ -20,8 +20,8 @@ func cleanEnv(ctx context.Context, clientMaster, clientSlave *redis.Client) {
 	r := clientSlave.Do(ctx, "slaveof", "no", "one")
 	Expect(r.Err()).NotTo(HaveOccurred())
 	Expect(r.Val()).To(Equal("OK"))
-	r = clientSlave.Do(ctx, "clearreplicationid")
-	r = clientMaster.Do(ctx, "clearreplicationid")
+	Expect(clientSlave.Do(ctx, "clearreplicationid").Err()).NotTo(HaveOccurred())
+	Expect(clientMaster.Do(ctx, "clearreplicationid").Err()).NotTo(HaveOccurred())
 	time.Sleep(1 * time.Second)
 }
 
@@ -373,8 +373,8 @@ var _ = Describe("should replication ", func() {
 		BeforeEach(func() {
 			clientMaster = redis.NewClient(PikaOption(MASTERADDR))
 			clientSlave = redis.NewClient(PikaOption(SLAVEADDR))
-			Expect(clientMaster.FlushDB(ctx).Err()).NotTo(HaveOccurred())
-			Expect(clientSlave.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+			//Expect(clientMaster.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+			//Expect(clientSlave.FlushDB(ctx).Err()).NotTo(HaveOccurred())
 			cleanEnv(ctx, clientMaster, clientSlave)
 			if GlobalBefore != nil {
 				GlobalBefore(ctx, clientMaster)
@@ -383,8 +383,8 @@ var _ = Describe("should replication ", func() {
 		})
 		AfterEach(func() {
 			cleanEnv(ctx, clientMaster, clientSlave)
-			Expect(clientMaster.FlushDB(ctx).Err()).NotTo(HaveOccurred())
-			Expect(clientSlave.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+			//Expect(clientMaster.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+			//Expect(clientSlave.FlushDB(ctx).Err()).NotTo(HaveOccurred())
 			Expect(clientSlave.Close()).NotTo(HaveOccurred())
 			Expect(clientMaster.Close()).NotTo(HaveOccurred())
 			log.Println("Replication test case done")
@@ -441,7 +441,8 @@ var _ = Describe("should replication ", func() {
 			set1 := clientMaster.Set(ctx, "a", "b", 0)
 			Expect(set1.Err()).NotTo(HaveOccurred())
 			Expect(set1.Val()).To(Equal("OK"))
-			Expect(clientMaster.Del(ctx, "x").Err()).NotTo(HaveOccurred())
+			time.Sleep(3 * time.Second)
+			Expect(clientMaster.FlushDB(ctx).Err()).NotTo(HaveOccurred())
 			//TODO Use del instead of flushdb after the flushDB problem is fixed
 			Eventually(func() error {
 				return clientMaster.Get(ctx, "x").Err()
@@ -456,9 +457,9 @@ var _ = Describe("should replication ", func() {
 			Expect(clientMaster.Del(ctx, "blist0", "blist1", "blist").Err()).NotTo(HaveOccurred())
 			execute(&ctx, clientMaster, 4, rpoplpushThread)
 			// TODO, the problem was not reproduced locally, record an issue first: https://github.com/OpenAtomFoundation/pika/issues/2492
-			//for i := int64(0); i < clientMaster.LLen(ctx, "blist").Val(); i++ {
-			//	Expect(clientMaster.LIndex(ctx, "blist", i)).To(Equal(clientSlave.LIndex(ctx, "blist", i)))
-			//}
+			for i := int64(0); i < clientMaster.LLen(ctx, "blist").Val(); i++ {
+				Expect(clientMaster.LIndex(ctx, "blist", i)).To(Equal(clientSlave.LIndex(ctx, "blist", i)))
+			}
 			Expect(clientMaster.Del(ctx, "blist0", "blist1", "blist").Err()).NotTo(HaveOccurred())
 			log.Println("rpoplpush test success")