Address gaps in concurrency thread safety

Author: marta-lokhova

src/bucket/BucketManager.h

@@ -101,7 +101,7 @@ class BucketManager : NonMovableOrCopyable
     // mLedgerStateMutex to prevent deadlocks. Code must NOT hold mBucketMutex
     // while trying to acquire LedgerManagerImpl::mLedgerStateMutex, as this
     // will cause a deadlock.
-    mutable RecursiveMutex mBucketMutex;
+    mutable ANNOTATED_RECURSIVE_MUTEX(mBucketMutex);
 
 #ifdef THREAD_SAFETY
   private:

src/bucket/BucketSnapshotManager.h

@@ -37,7 +37,7 @@ class BucketSnapshotManager : NonMovableOrCopyable
     AppConnector& mAppConnector;
 
     // Lock must be held when accessing any member variables holding snapshots
-    mutable SharedMutex mSnapshotMutex;
+    mutable ANNOTATED_SHARED_MUTEX(mSnapshotMutex);
 
     // Snapshot that is maintained and periodically updated by BucketManager on
     // the main thread. When background threads need to generate or refresh a

src/bucket/LiveBucketIndex.cpp

@@ -102,7 +102,7 @@ LiveBucketIndex::maybeInitializeCache(size_t totalBucketListAccountsSizeBytes,
     }
 
     // Cache is already initialized
-    if (std::shared_lock<std::shared_mutex> lock(mCacheMutex); mCache)
+    if (SharedLockShared lock(mCacheMutex); mCache)
     {
         return;
     }
@@ -123,7 +123,7 @@ LiveBucketIndex::maybeInitializeCache(size_t totalBucketListAccountsSizeBytes,
         return;
     }
 
-    std::unique_lock<std::shared_mutex> lock(mCacheMutex);
+    SharedLockExclusive lock(mCacheMutex);
     if (totalBucketListAccountsSizeBytes < maxBucketListBytesToCache)
     {
         // We can cache the entire bucket
@@ -202,7 +202,7 @@ LiveBucketIndex::getCachedEntry(LedgerKey const& k) const
 {
     if (shouldUseCache() && isCachedType(k))
     {
-        std::shared_lock<std::shared_mutex> lock(mCacheMutex);
+        SharedLockShared lock(mCacheMutex);
         auto cachePtr = mCache->maybeGet(k);
         if (cachePtr)
         {
@@ -323,7 +323,7 @@ LiveBucketIndex::shouldUseCache() const
 {
     if (mDiskIndex)
     {
-        std::shared_lock<std::shared_mutex> lock(mCacheMutex);
+        SharedLockShared lock(mCacheMutex);
         return mCache != nullptr;
     }
 
@@ -353,7 +353,7 @@ LiveBucketIndex::maybeAddToCache(
         // earlier.
         mCacheMissMeter.Mark();
 
-        std::unique_lock<std::shared_mutex> lock(mCacheMutex);
+        SharedLockExclusive lock(mCacheMutex);
         mCache->put(k, entry);
     }
 }
@@ -392,7 +392,7 @@ LiveBucketIndex::getMaxCacheSize() const
 {
     if (shouldUseCache())
     {
-        std::shared_lock<std::shared_mutex> lock(mCacheMutex);
+        SharedLockShared lock(mCacheMutex);
         return mCache->maxSize();
     }
 
@@ -405,7 +405,7 @@ LiveBucketIndex::getCurrentCacheSize() const
 {
     if (shouldUseCache())
     {
-        std::shared_lock<std::shared_mutex> lock(mCacheMutex);
+        SharedLockShared lock(mCacheMutex);
         return mCache->size();
     }
 

src/bucket/LiveBucketIndex.h

@@ -12,13 +12,13 @@
 #include "ledger/LedgerHashUtils.h" // IWYU pragma: keep
 #include "util/NonCopyable.h"
 #include "util/RandomEvictionCache.h"
+#include "util/ThreadAnnotations.h"
 #include "util/XDROperators.h" // IWYU pragma: keep
 #include "xdr/Stellar-ledger-entries.h"
 #include <filesystem>
 #include <optional>
 
 #include <cereal/archives/binary.hpp>
-#include <shared_mutex>
 
 namespace asio
 {
@@ -65,8 +65,8 @@ class LiveBucketIndex : public NonMovableOrCopyable
     // The indexes themselves are thread safe, as they are immutable after
     // construction. The cache is not, all accesses must first acquire this
     // mutex.
-    mutable std::unique_ptr<CacheT> mCache{};
-    mutable std::shared_mutex mCacheMutex;
+    mutable std::unique_ptr<CacheT> mCache GUARDED_BY(mCacheMutex){};
+    mutable ANNOTATED_SHARED_MUTEX(mCacheMutex);
 
     medida::Meter& mCacheHitMeter;
     medida::Meter& mCacheMissMeter;

src/invariant/InvariantManagerImpl.h

@@ -37,7 +37,7 @@ class InvariantManagerImpl : public InvariantManager
         std::string lastFailedWithMessage;
     };
 
-    Mutex mutable mFailureInformationMutex;
+    ANNOTATED_MUTEX(mFailureInformationMutex);
     std::map<std::string, InvariantFailureInformation>
         mFailureInformation GUARDED_BY(mFailureInformationMutex);
 

src/ledger/LedgerManagerImpl.h

@@ -296,11 +296,8 @@ class LedgerManagerImpl : public LedgerManager
     VirtualClock::time_point mLastClose;
 
     // Use mutex to guard ledger state during apply
-    mutable RecursiveMutex mLedgerStateMutex
-#ifdef THREAD_SAFETY
-        ACQUIRED_BEFORE(BucketManager::mBucketMutex)
-#endif
-            ;
+    ANNOTATED_RECURSIVE_MUTEX(mLedgerStateMutex,
+                              ACQUIRED_BEFORE(BucketManager::mBucketMutex));
 
     medida::Timer& mCatchupDuration;
 
@@ -564,6 +561,7 @@ class LedgerManagerImpl : public LedgerManager
     virtual bool
     isApplying() const override
     {
+        releaseAssert(threadIsMain());
         return mCurrentlyApplyingLedger;
     }
     void markApplyStateReset() override;

src/overlay/FlowControl.h

@@ -67,7 +67,7 @@ class FlowControl
     size_t mTxQueueByteCount GUARDED_BY(mFlowControlMutex){0};
 
     // Mutex to synchronize flow control state
-    Mutex mutable mFlowControlMutex;
+    mutable ANNOTATED_MUTEX(mFlowControlMutex);
     // Is this peer currently throttled due to lack of capacity
     std::optional<VirtualClock::time_point>
         mLastThrottle GUARDED_BY(mFlowControlMutex);

src/overlay/Hmac.cpp

@@ -83,6 +83,7 @@ Hmac::setAuthenticatedMessageBody(AuthenticatedMessage& aMsg,
 void
 Hmac::damageRecvMacKey()
 {
+    LOCK_GUARD(mMutex, guard);
     auto bytes = randomBytes(mRecvMacKey.key.size());
     std::copy(bytes.begin(), bytes.end(), mRecvMacKey.key.begin());
 }

src/overlay/Hmac.h

@@ -11,17 +11,23 @@
 
 using namespace stellar;
 
+namespace stellar
+{
+class Peer;
+}
+
 class Hmac
 {
-#ifndef USE_TRACY
-    Mutex mMutex;
-#else
-    TracyLockable(std::mutex, mMutex);
+#ifdef THREAD_SAFETY
+    // Make the peer class a friend for thread safety analysis
+    friend class stellar::Peer;
 #endif
-    HmacSha256Key mSendMacKey;
-    HmacSha256Key mRecvMacKey;
-    uint64_t mSendMacSeq{0};
-    uint64_t mRecvMacSeq{0};
+
+    ANNOTATED_MUTEX(mMutex);
+    HmacSha256Key mSendMacKey GUARDED_BY(mMutex);
+    HmacSha256Key mRecvMacKey GUARDED_BY(mMutex);
+    uint64_t mSendMacSeq GUARDED_BY(mMutex){0};
+    uint64_t mRecvMacSeq GUARDED_BY(mMutex){0};
 
   public:
     bool setSendMackey(HmacSha256Key const& key);

src/overlay/Peer.h

@@ -195,12 +195,10 @@ class Peer : public std::enable_shared_from_this<Peer>,
 #endif
 
     // Mutex to protect PeerState, which can be accessed and modified from
-    // multiple threads
-#ifndef USE_TRACY
-    RecursiveMutex mutable mStateMutex;
-#else
-    mutable TracyLockable(std::recursive_mutex, mStateMutex);
-#endif
+    // multiple threads.
+    // LOCK ORDERING: mStateMutex must be acquired before Hmac::mMutex.
+    mutable ANNOTATED_RECURSIVE_MUTEX(mStateMutex,
+                                      ACQUIRED_BEFORE(Hmac::mMutex));
 
     Hmac mHmac;
     // Does local node have capacity to read from this peer