chore: add -Wthread-analysis and annotate (part 1/2)

CMakeLists.txt

@@ -18,6 +18,10 @@ if (HAS_USE_AFTER_FREE_WARN)
     set(CMAKE_CXX_FLAGS "-Wno-use-after-free ${CMAKE_CXX_FLAGS}")
 endif()
 
+if (CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
+    set(CMAKE_CXX_FLAGS "-Wthread-safety ${CMAKE_CXX_FLAGS}")
+endif()
+
 # We can not use here CHECK_CXX_COMPILER_FLAG because systems that do not support sanitizers
 # fail during linking time.
 set(CMAKE_REQUIRED_FLAGS "-fsanitize=address")

src/server/cluster/cluster_family.cc

@@ -698,7 +698,7 @@ void ClusterFamily::DflySlotMigrationStatus(CmdArgList args, ConnectionContext*
   auto* rb = static_cast<RedisReplyBuilder*>(cntx->reply_builder());
   CmdArgParser parser(args);
 
-  lock_guard lk(migration_mu_);
+  util::fb2::LockGuard lk(migration_mu_);
 
   string_view node_id;
   if (parser.HasNext()) {

src/server/cluster/cluster_family.h

@@ -62,7 +62,8 @@ class ClusterFamily {
   void DflyClusterFlushSlots(CmdArgList args, ConnectionContext* cntx);
 
  private:  // Slots migration section
-  void DflySlotMigrationStatus(CmdArgList args, ConnectionContext* cntx);
+  void DflySlotMigrationStatus(CmdArgList args, ConnectionContext* cntx)
+      ABSL_LOCKS_EXCLUDED(migration_mu_);
 
   // DFLYMIGRATE is internal command defines several steps in slots migrations process
   void DflyMigrate(CmdArgList args, ConnectionContext* cntx);

src/server/cluster/incoming_slot_migration.h

@@ -43,17 +43,17 @@ class IncomingSlotMigration {
     return slots_;
   }
 
-  const std::string GetSourceID() const {
+  const std::string& GetSourceID() const {
     return source_id_;
   }
 
-  void ReportError(dfly::GenericError err) {
-    std::lock_guard lk(error_mu_);
-    last_error_ = err;
+  void ReportError(dfly::GenericError err) ABSL_LOCKS_EXCLUDED(error_mu_) {
+    util::fb2::LockGuard lk(error_mu_);
+    last_error_ = std::move(err);
   }
 
-  std::string GetErrorStr() const {
-    std::lock_guard lk(error_mu_);
+  std::string GetErrorStr() const ABSL_LOCKS_EXCLUDED(error_mu_) {
+    util::fb2::LockGuard lk(error_mu_);
     return last_error_.Format();
   }
 

src/server/cluster/outgoing_slot_migration.cc

@@ -18,6 +18,7 @@
 #include "server/journal/streamer.h"
 #include "server/main_service.h"
 #include "server/server_family.h"
+#include "util/fibers/synchronization.h"
 
 ABSL_FLAG(int, slot_migration_connection_timeout_ms, 2000, "Timeout for network operations");
 
@@ -112,7 +113,7 @@ OutgoingMigration::~OutgoingMigration() {
 }
 
 bool OutgoingMigration::ChangeState(MigrationState new_state) {
-  std::lock_guard lk(state_mu_);
+  util::fb2::LockGuard lk(state_mu_);
   if (state_ == MigrationState::C_FINISHED) {
     return false;
   }
@@ -136,7 +137,7 @@ void OutgoingMigration::Finish(bool is_error) {
   bool should_cancel_flows = false;
 
   {
-    std::lock_guard lk(state_mu_);
+    util::fb2::LockGuard lk(state_mu_);
     switch (state_) {
       case MigrationState::C_FINISHED:
         return;  // Already finished, nothing else to do
@@ -164,7 +165,7 @@ void OutgoingMigration::Finish(bool is_error) {
 }
 
 MigrationState OutgoingMigration::GetState() const {
-  std::lock_guard lk(state_mu_);
+  util::fb2::LockGuard lk(state_mu_);
   return state_;
 }
 

src/server/cluster/outgoing_slot_migration.h

@@ -32,9 +32,9 @@ class OutgoingMigration : private ProtocolClient {
 
   // mark migration as FINISHED and cancel migration if it's not finished yet
   // can be called from any thread, but only after Start()
-  void Finish(bool is_error = false);
+  void Finish(bool is_error = false) ABSL_LOCKS_EXCLUDED(state_mu_);
 
-  MigrationState GetState() const;
+  MigrationState GetState() const ABSL_LOCKS_EXCLUDED(state_mu_);
 
   const std::string& GetHostIp() const {
     return server().host;
@@ -56,7 +56,7 @@ class OutgoingMigration : private ProtocolClient {
     return last_error_.Format();
   }
 
-  size_t GetKeyCount() const;
+  size_t GetKeyCount() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(state_mu_);
 
   static constexpr long kInvalidAttempt = -1;
   static constexpr std::string_view kUnknownMigration = "UNKNOWN_MIGRATION";

src/server/common.h

@@ -362,13 +362,13 @@ class UniquePicksGenerator : public PicksGenerator {
 };
 
 // Helper class used to guarantee atomicity between serialization of buckets
-class ThreadLocalMutex {
+class ABSL_LOCKABLE ThreadLocalMutex {
  public:
   ThreadLocalMutex();
   ~ThreadLocalMutex();
 
-  void lock();
-  void unlock();
+  void lock() ABSL_EXCLUSIVE_LOCK_FUNCTION();
+  void unlock() ABSL_UNLOCK_FUNCTION();
 
  private:
   EngineShard* shard_;

src/server/config_registry.cc

@@ -1,12 +1,13 @@
 // Copyright 2023, Roman Gershman.  All rights reserved.
 // See LICENSE for licensing terms.
 //
-#include "src/server/config_registry.h"
+#include "server/config_registry.h"
 
 #include <absl/flags/reflection.h>
 #include <absl/strings/str_replace.h>
 
 #include "base/logging.h"
+#include "server/common.h"
 
 extern "C" {
 #include "redis/util.h"
@@ -25,7 +26,7 @@ string NormalizeConfigName(string_view name) {
 auto ConfigRegistry::Set(string_view config_name, string_view value) -> SetResult {
   string name = NormalizeConfigName(config_name);
 
-  unique_lock lk(mu_);
+  util::fb2::LockGuard lk(mu_);
   auto it = registry_.find(name);
   if (it == registry_.end())
     return SetResult::UNKNOWN;
@@ -48,26 +49,27 @@ auto ConfigRegistry::Set(string_view config_name, string_view value) -> SetResul
 optional<string> ConfigRegistry::Get(string_view config_name) {
   string name = NormalizeConfigName(config_name);
 
-  unique_lock lk(mu_);
-  if (!registry_.contains(name))
-    return nullopt;
-  lk.unlock();
+  {
+    util::fb2::LockGuard lk(mu_);
+    if (!registry_.contains(name))
+      return nullopt;
+  }
 
   absl::CommandLineFlag* flag = absl::FindCommandLineFlag(name);
   CHECK(flag);
   return flag->CurrentValue();
 }
 
 void ConfigRegistry::Reset() {
-  unique_lock lk(mu_);
+  util::fb2::LockGuard lk(mu_);
   registry_.clear();
 }
 
 vector<string> ConfigRegistry::List(string_view glob) const {
   string normalized_glob = NormalizeConfigName(glob);
 
   vector<string> res;
-  unique_lock lk(mu_);
+  util::fb2::LockGuard lk(mu_);
   for (const auto& [name, _] : registry_) {
     if (stringmatchlen(normalized_glob.data(), normalized_glob.size(), name.data(), name.size(), 1))
       res.push_back(name);
@@ -81,7 +83,7 @@ void ConfigRegistry::RegisterInternal(string_view config_name, bool is_mutable,
   absl::CommandLineFlag* flag = absl::FindCommandLineFlag(name);
   CHECK(flag) << "Unknown config name: " << name;
 
-  unique_lock lk(mu_);
+  util::fb2::LockGuard lk(mu_);
   auto [it, inserted] = registry_.emplace(name, Entry{std::move(cb), is_mutable});
   CHECK(inserted) << "Duplicate config name: " << name;
 }

src/server/config_registry.h

@@ -34,16 +34,17 @@ class ConfigRegistry {
   };
 
   // Returns true if the value was updated.
-  SetResult Set(std::string_view config_name, std::string_view value);
+  SetResult Set(std::string_view config_name, std::string_view value) ABSL_LOCKS_EXCLUDED(mu_);
 
-  std::optional<std::string> Get(std::string_view config_name);
+  std::optional<std::string> Get(std::string_view config_name) ABSL_LOCKS_EXCLUDED(mu_);
 
   void Reset();
 
-  std::vector<std::string> List(std::string_view glob) const;
+  std::vector<std::string> List(std::string_view glob) const ABSL_LOCKS_EXCLUDED(mu_);
 
  private:
-  void RegisterInternal(std::string_view name, bool is_mutable, WriteCb cb);
+  void RegisterInternal(std::string_view name, bool is_mutable, WriteCb cb)
+      ABSL_LOCKS_EXCLUDED(mu_);
 
   mutable util::fb2::Mutex mu_;
 

src/server/db_slice.cc

@@ -206,7 +206,7 @@ unsigned PrimeEvictionPolicy::Evict(const PrimeTable::HotspotBuckets& eb, PrimeT
     if (auto journal = db_slice_->shard_owner()->journal(); journal) {
       RecordExpiry(cntx_.db_index, key);
     }
-    // Safe we already acquired std::unique_lock lk(db_slice_->GetSerializationMutex());
+    // Safe we already acquired util::fb2::LockGuard lk(db_slice_->GetSerializationMutex());
     // on the flows that call this function
     db_slice_->PerformDeletion(DbSlice::Iterator(last_slot_it, StringOrView::FromView(key)), table);
 
@@ -492,7 +492,7 @@ OpResult<DbSlice::PrimeItAndExp> DbSlice::FindInternal(const Context& cntx, std:
   if (caching_mode_ && IsValid(res.it)) {
     if (!change_cb_.empty()) {
       FetchedItemsRestorer fetched_restorer(&fetched_items_);
-      std::unique_lock lk(local_mu_);
+      util::fb2::LockGuard lk(local_mu_);
       auto bump_cb = [&](PrimeTable::bucket_iterator bit) {
         CallChangeCallbacks(cntx.db_index, key, bit);
       };
@@ -588,7 +588,7 @@ OpResult<DbSlice::AddOrFindResult> DbSlice::AddOrFindInternal(const Context& cnt
   CHECK(status == OpStatus::KEY_NOTFOUND || status == OpStatus::OUT_OF_MEMORY) << status;
 
   FetchedItemsRestorer fetched_restorer(&fetched_items_);
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
 
   // It's a new entry.
   CallChangeCallbacks(cntx.db_index, key, {key});
@@ -704,7 +704,7 @@ void DbSlice::ActivateDb(DbIndex db_ind) {
 }
 
 bool DbSlice::Del(Context cntx, Iterator it) {
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
 
   if (!IsValid(it)) {
     return false;
@@ -828,7 +828,7 @@ void DbSlice::FlushDb(DbIndex db_ind) {
   // We should not flush if serialization of a big value is in progress because this
   // could lead to UB or assertion failures (while DashTable::Traverse is iterating over
   // a logical bucket).
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
   // clear client tracking map.
   client_tracking_map_.clear();
 
@@ -850,7 +850,7 @@ void DbSlice::FlushDb(DbIndex db_ind) {
 }
 
 void DbSlice::AddExpire(DbIndex db_ind, Iterator main_it, uint64_t at) {
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
   uint64_t delta = at - expire_base_[0];  // TODO: employ multigen expire updates.
   auto& db = *db_arr_[db_ind];
   size_t table_before = db.expire.mem_usage();
@@ -860,7 +860,7 @@ void DbSlice::AddExpire(DbIndex db_ind, Iterator main_it, uint64_t at) {
 }
 
 bool DbSlice::RemoveExpire(DbIndex db_ind, Iterator main_it) {
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
   if (main_it->second.HasExpire()) {
     auto& db = *db_arr_[db_ind];
     size_t table_before = db.expire.mem_usage();
@@ -1088,7 +1088,7 @@ bool DbSlice::CheckLock(IntentLock::Mode mode, DbIndex dbid, uint64_t fp) const
 
 void DbSlice::PreUpdate(DbIndex db_ind, Iterator it, std::string_view key) {
   FetchedItemsRestorer fetched_restorer(&fetched_items_);
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
   CallChangeCallbacks(db_ind, key, ChangeReq{it.GetInnerIt()});
   it.GetInnerIt().SetVersion(NextVersion());
 }
@@ -1224,7 +1224,7 @@ void DbSlice::FlushChangeToEarlierCallbacks(DbIndex db_ind, Iterator it, uint64_
 
 //! Unregisters the callback.
 void DbSlice::UnregisterOnChange(uint64_t id) {
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
   auto it = find_if(change_cb_.begin(), change_cb_.end(),
                     [id](const auto& cb) { return cb.first == id; });
   CHECK(it != change_cb_.end());
@@ -1382,13 +1382,13 @@ void DbSlice::CreateDb(DbIndex db_ind) {
 void DbSlice::RegisterWatchedKey(DbIndex db_indx, std::string_view key,
                                  ConnectionState::ExecInfo* exec_info) {
   // Because we might insert while another fiber is preempted
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
   db_arr_[db_indx]->watched_keys[key].push_back(exec_info);
 }
 
 void DbSlice::UnregisterConnectionWatches(const ConnectionState::ExecInfo* exec_info) {
   // Because we might remove while another fiber is preempted and miss a notification
-  std::unique_lock lk(local_mu_);
+  util::fb2::LockGuard lk(local_mu_);
   for (const auto& [db_indx, key] : exec_info->watched_keys) {
     auto& watched_keys = db_arr_[db_indx]->watched_keys;
     if (auto it = watched_keys.find(key); it != watched_keys.end()) {

src/server/db_slice.h

@@ -507,7 +507,7 @@ class DbSlice {
   template <typename Cb, typename DashTable>
   PrimeTable::Cursor Traverse(DashTable* pt, PrimeTable::Cursor cursor, Cb&& cb)
       ABSL_LOCKS_EXCLUDED(local_mu_) {
-    std::unique_lock lk(local_mu_);
+    util::fb2::LockGuard lk(local_mu_);
     return pt->Traverse(cursor, std::forward<Cb>(cb));
   }