Incorporate clang thread-safety static analysis into core (#4727)

Introduce clang thread safety analysis
https://clang.llvm.org/docs/ThreadSafetyAnalysis.html, and perform it
when building core with thread sanitizer.

- For now, I've only annotated `FlowControl` and `BucketSnapshotManager`
classes, but we should make it a standard for all classes doing
concurrency.
- This is useful to identify data races and deadlocks at compile time.
This change was able to identify two unprotected shared data bugs we had
in the past: #4702 and
#4458
- It also identified undefined behavior, where
[shared_mutex](https://en.cppreference.com/w/cpp/thread/shared_mutex/lock_shared)
is locked twice by the same thread. This PR includes a fix as well.

Author: marta-lokhova

configure.ac

@@ -127,6 +127,14 @@ AS_IF([test "x$enable_threadsanitizer" = "xyes"], [
   sanitizeopts="thread"
 ])
 
+AC_ARG_ENABLE([threadsafety],
+  AS_HELP_STRING([--enable-threadsafety],
+        [build with thread safety static analysis instrumentation (Clang only)]))
+AS_IF([test "x$enable_threadsafety" = "xyes"], [
+  AC_MSG_NOTICE([enabling thread safety static analysis, see https://clang.llvm.org/docs/ThreadSafetyAnalysis.html])
+  CXXFLAGS="$CXXFLAGS -Wthread-safety -Werror=thread-safety"
+])
+
 AC_ARG_ENABLE([memcheck],
   AS_HELP_STRING([--enable-memcheck],
         [build with memcheck (memory-sanitizer) instrumentation]))

src/bucket/BucketSnapshotManager.cpp

@@ -55,7 +55,24 @@ BucketSnapshotManager::BucketSnapshotManager(
 SearchableSnapshotConstPtr
 BucketSnapshotManager::copySearchableLiveBucketListSnapshot() const
 {
-    std::shared_lock<std::shared_mutex> lock(mSnapshotMutex);
+    SharedLockShared guard(mSnapshotMutex);
+    // Can't use std::make_shared due to private constructor
+    return copySearchableLiveBucketListSnapshot(guard);
+}
+
+SearchableHotArchiveSnapshotConstPtr
+BucketSnapshotManager::copySearchableHotArchiveBucketListSnapshot() const
+{
+    SharedLockShared guard(mSnapshotMutex);
+    releaseAssert(mCurrHotArchiveSnapshot);
+    // Can't use std::make_shared due to private constructor
+    return copySearchableHotArchiveBucketListSnapshot(guard);
+}
+
+SearchableSnapshotConstPtr
+BucketSnapshotManager::copySearchableLiveBucketListSnapshot(
+    SharedLockShared const& guard) const
+{
     // Can't use std::make_shared due to private constructor
     return std::shared_ptr<SearchableLiveBucketListSnapshot>(
         new SearchableLiveBucketListSnapshot(
@@ -66,9 +83,9 @@ BucketSnapshotManager::copySearchableLiveBucketListSnapshot() const
 }
 
 SearchableHotArchiveSnapshotConstPtr
-BucketSnapshotManager::copySearchableHotArchiveBucketListSnapshot() const
+BucketSnapshotManager::copySearchableHotArchiveBucketListSnapshot(
+    SharedLockShared const& guard) const
 {
-    std::shared_lock<std::shared_mutex> lock(mSnapshotMutex);
     releaseAssert(mCurrHotArchiveSnapshot);
     // Can't use std::make_shared due to private constructor
     return std::shared_ptr<SearchableHotArchiveBucketListSnapshot>(
@@ -86,12 +103,11 @@ BucketSnapshotManager::maybeCopySearchableBucketListSnapshot(
     // The canonical snapshot held by the BucketSnapshotManager is not being
     // modified. Rather, a thread is checking it's copy against the canonical
     // snapshot, so use a shared lock.
-    std::shared_lock<std::shared_mutex> lock(mSnapshotMutex);
-
+    SharedLockShared guard(mSnapshotMutex);
     if (!snapshot ||
         snapshot->getLedgerSeq() < mCurrLiveSnapshot->getLedgerSeq())
     {
-        snapshot = copySearchableLiveBucketListSnapshot();
+        snapshot = copySearchableLiveBucketListSnapshot(guard);
     }
 }
 
@@ -102,12 +118,12 @@ BucketSnapshotManager::maybeCopySearchableHotArchiveBucketListSnapshot(
     // The canonical snapshot held by the BucketSnapshotManager is not being
     // modified. Rather, a thread is checking it's copy against the canonical
     // snapshot, so use a shared lock.
-    std::shared_lock<std::shared_mutex> lock(mSnapshotMutex);
+    SharedLockShared guard(mSnapshotMutex);
 
     if (!snapshot ||
         snapshot->getLedgerSeq() < mCurrHotArchiveSnapshot->getLedgerSeq())
     {
-        snapshot = copySearchableHotArchiveBucketListSnapshot();
+        snapshot = copySearchableHotArchiveBucketListSnapshot(guard);
     }
 }
 
@@ -142,7 +158,7 @@ BucketSnapshotManager::updateCurrentSnapshot(
 
     // Updating the BucketSnapshotManager canonical snapshot, must lock
     // exclusively for write access.
-    std::unique_lock<std::shared_mutex> lock(mSnapshotMutex);
+    SharedLockExclusive guard(mSnapshotMutex);
     updateSnapshot(mCurrLiveSnapshot, mLiveHistoricalSnapshots, liveSnapshot);
     updateSnapshot(mCurrHotArchiveSnapshot, mHotArchiveHistoricalSnapshots,
                    hotArchiveSnapshot);

src/bucket/BucketSnapshotManager.h

@@ -9,6 +9,7 @@
 #include "bucket/LiveBucket.h"
 #include "main/AppConnector.h"
 #include "util/NonCopyable.h"
+#include "util/ThreadAnnotations.h"
 
 #include <map>
 #include <memory>
@@ -45,28 +46,31 @@ class BucketSnapshotManager : NonMovableOrCopyable
   private:
     AppConnector mAppConnector;
 
+    // Lock must be held when accessing any member variables holding snapshots
+    mutable SharedMutex mSnapshotMutex;
+
     // Snapshot that is maintained and periodically updated by BucketManager on
     // the main thread. When background threads need to generate or refresh a
     // snapshot, they will copy this snapshot.
-    SnapshotPtrT<LiveBucket> mCurrLiveSnapshot{};
-    SnapshotPtrT<HotArchiveBucket> mCurrHotArchiveSnapshot{};
+    SnapshotPtrT<LiveBucket> mCurrLiveSnapshot GUARDED_BY(mSnapshotMutex){};
+    SnapshotPtrT<HotArchiveBucket>
+        mCurrHotArchiveSnapshot GUARDED_BY(mSnapshotMutex){};
 
     // ledgerSeq that the snapshot is based on -> snapshot
-    std::map<uint32_t, SnapshotPtrT<LiveBucket>> mLiveHistoricalSnapshots;
+    std::map<uint32_t, SnapshotPtrT<LiveBucket>>
+        mLiveHistoricalSnapshots GUARDED_BY(mSnapshotMutex);
     std::map<uint32_t, SnapshotPtrT<HotArchiveBucket>>
-        mHotArchiveHistoricalSnapshots;
+        mHotArchiveHistoricalSnapshots GUARDED_BY(mSnapshotMutex);
 
     uint32_t const mNumHistoricalSnapshots;
 
-    // Lock must be held when accessing any member variables holding snapshots
-    mutable std::shared_mutex mSnapshotMutex;
-
   public:
     // Called by main thread to update snapshots whenever the BucketList
     // is updated
     void
     updateCurrentSnapshot(SnapshotPtrT<LiveBucket>&& liveSnapshot,
-                          SnapshotPtrT<HotArchiveBucket>&& hotArchiveSnapshot);
+                          SnapshotPtrT<HotArchiveBucket>&& hotArchiveSnapshot)
+        LOCKS_EXCLUDED(mSnapshotMutex);
 
     // numHistoricalLedgers is the number of historical snapshots that the
     // snapshot manager will maintain. If numHistoricalLedgers is 5, snapshots
@@ -76,18 +80,34 @@ class BucketSnapshotManager : NonMovableOrCopyable
                           uint32_t numHistoricalLedgers);
 
     // Copy the most recent snapshot for the live bucket list
-    SearchableSnapshotConstPtr copySearchableLiveBucketListSnapshot() const;
+    SearchableSnapshotConstPtr copySearchableLiveBucketListSnapshot() const
+        LOCKS_EXCLUDED(mSnapshotMutex);
 
     // Copy the most recent snapshot for the hot archive bucket list
     SearchableHotArchiveSnapshotConstPtr
-    copySearchableHotArchiveBucketListSnapshot() const;
+    copySearchableHotArchiveBucketListSnapshot() const
+        LOCKS_EXCLUDED(mSnapshotMutex);
+
+    // Copy the most recent snapshot for the live bucket list, while holding the
+    // lock
+    SearchableSnapshotConstPtr
+    copySearchableLiveBucketListSnapshot(SharedLockShared const& guard) const
+        REQUIRES_SHARED(mSnapshotMutex);
+
+    // Copy the most recent snapshot for the hot archive bucket list, while
+    // holding the lock
+    SearchableHotArchiveSnapshotConstPtr
+    copySearchableHotArchiveBucketListSnapshot(
+        SharedLockShared const& guard) const REQUIRES_SHARED(mSnapshotMutex);
 
     // `maybeCopy` interface refreshes `snapshot` if a newer snapshot is
     // available. It's a no-op otherwise. This is useful to avoid unnecessary
     // copying.
     void
-    maybeCopySearchableBucketListSnapshot(SearchableSnapshotConstPtr& snapshot);
+    maybeCopySearchableBucketListSnapshot(SearchableSnapshotConstPtr& snapshot)
+        LOCKS_EXCLUDED(mSnapshotMutex);
     void maybeCopySearchableHotArchiveBucketListSnapshot(
-        SearchableHotArchiveSnapshotConstPtr& snapshot);
+        SearchableHotArchiveSnapshotConstPtr& snapshot)
+        LOCKS_EXCLUDED(mSnapshotMutex);
 };
 }

src/bucket/SearchableBucketList.h

@@ -36,7 +36,8 @@ class SearchableLiveBucketListSnapshot
         StateArchivalSettings const& sas, uint32_t ledgerVers) const;
 
     friend SearchableSnapshotConstPtr
-    BucketSnapshotManager::copySearchableLiveBucketListSnapshot() const;
+    BucketSnapshotManager::copySearchableLiveBucketListSnapshot(
+        SharedLockShared const& guard) const;
 };
 
 class SearchableHotArchiveBucketListSnapshot
@@ -54,6 +55,7 @@ class SearchableHotArchiveBucketListSnapshot
     loadKeys(std::set<LedgerKey, LedgerEntryIdCmp> const& inKeys) const;
 
     friend SearchableHotArchiveSnapshotConstPtr
-    BucketSnapshotManager::copySearchableHotArchiveBucketListSnapshot() const;
+    BucketSnapshotManager::copySearchableHotArchiveBucketListSnapshot(
+        SharedLockShared const& guard) const;
 };
 }

src/overlay/FlowControl.cpp

@@ -16,8 +16,7 @@ namespace stellar
 {
 
 size_t
-FlowControl::getOutboundQueueByteLimit(
-    std::lock_guard<std::mutex>& lockGuard) const
+FlowControl::getOutboundQueueByteLimit(MutexLocker& lockGuard) const
 {
 #ifdef BUILD_TESTS
     if (mOutboundQueueLimit)
@@ -44,7 +43,7 @@ FlowControl::FlowControl(AppConnector& connector, bool useBackgroundThread)
 
 bool
 FlowControl::hasOutboundCapacity(StellarMessage const& msg,
-                                 std::lock_guard<std::mutex>& lockGuard) const
+                                 MutexLocker& lockGuard) const
 {
     releaseAssert(!threadIsMain() || !mUseBackgroundThread);
     return mFlowControlCapacity.hasOutboundCapacity(msg) &&
@@ -55,15 +54,15 @@ bool
 FlowControl::noOutboundCapacityTimeout(VirtualClock::time_point now,
                                        std::chrono::seconds timeout) const
 {
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     return mNoOutboundCapacity && now - *mNoOutboundCapacity >= timeout;
 }
 
 void
 FlowControl::setPeerID(NodeID const& peerID)
 {
     releaseAssert(threadIsMain());
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     mNodeID = peerID;
 }
 
@@ -72,7 +71,7 @@ FlowControl::maybeReleaseCapacity(StellarMessage const& msg)
 {
     ZoneScoped;
     releaseAssert(threadIsMain());
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
 
     if (msg.type() == SEND_MORE_EXTENDED)
     {
@@ -103,7 +102,7 @@ FlowControl::processSentMessages(
     ZoneScoped;
     releaseAssert(!threadIsMain() || !mUseBackgroundThread);
 
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     for (int i = 0; i < sentMessages.size(); i++)
     {
         auto const& sentMsgs = sentMessages[i];
@@ -162,7 +161,7 @@ FlowControl::getNextBatchToSend()
     ZoneScoped;
     releaseAssert(!threadIsMain() || !mUseBackgroundThread);
 
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     std::vector<QueuedOutboundMessage> batchToSend;
 
     int sent = 0;
@@ -217,7 +216,7 @@ FlowControl::updateMsgMetrics(std::shared_ptr<StellarMessage const> msg,
 {
     // The lock isn't strictly needed here, but is added for consistency and
     // future-proofing this function
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     auto diff = mAppConnector.now() - timePlaced;
 
     auto updateQueueDelay = [&](auto& queue, auto& metrics) {
@@ -258,7 +257,7 @@ FlowControl::handleTxSizeIncrease(uint32_t increase)
 {
     ZoneScoped;
     releaseAssert(threadIsMain());
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     releaseAssert(increase > 0);
     // Bump flood capacity to accommodate the upgrade
     mFlowControlBytesCapacity.handleTxSizeIncrease(increase);
@@ -269,7 +268,7 @@ FlowControl::beginMessageProcessing(StellarMessage const& msg)
 {
     ZoneScoped;
     releaseAssert(!threadIsMain() || !mUseBackgroundThread);
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
 
     return mFlowControlCapacity.lockLocalCapacity(msg) &&
            mFlowControlBytesCapacity.lockLocalCapacity(msg);
@@ -279,7 +278,7 @@ SendMoreCapacity
 FlowControl::endMessageProcessing(StellarMessage const& msg)
 {
     ZoneScoped;
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
 
     mFloodDataProcessed += mFlowControlCapacity.releaseLocalCapacity(msg);
     mFloodDataProcessedBytes +=
@@ -318,7 +317,7 @@ FlowControl::endMessageProcessing(StellarMessage const& msg)
 }
 
 bool
-FlowControl::canRead(std::lock_guard<std::mutex> const& guard) const
+FlowControl::canRead(MutexLocker const& guard) const
 {
     return mFlowControlBytesCapacity.canRead() &&
            mFlowControlCapacity.canRead();
@@ -327,7 +326,7 @@ FlowControl::canRead(std::lock_guard<std::mutex> const& guard) const
 bool
 FlowControl::canRead() const
 {
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     return canRead(guard);
 }
 
@@ -365,7 +364,7 @@ FlowControl::isSendMoreValid(StellarMessage const& msg,
                              std::string& errorMsg) const
 {
     releaseAssert(threadIsMain());
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
 
     if (msg.type() != SEND_MORE_EXTENDED)
     {
@@ -404,7 +403,7 @@ FlowControl::addMsgAndMaybeTrimQueue(std::shared_ptr<StellarMessage const> msg)
 {
     ZoneScoped;
     releaseAssert(threadIsMain());
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     releaseAssert(msg);
     auto type = msg->type();
     size_t msgQInd = 0;
@@ -461,16 +460,12 @@ FlowControl::addMsgAndMaybeTrimQueue(std::shared_ptr<StellarMessage const> msg)
     auto& om = mOverlayMetrics;
     if (type == TRANSACTION)
     {
-        auto isOverLimit = [&](auto const& queue) {
-            bool overLimit =
-                queue.size() > limit ||
-                mTxQueueByteCount > getOutboundQueueByteLimit(guard);
-            return overLimit;
-        };
+        bool isOverLimit = queue.size() > limit ||
+                           mTxQueueByteCount > getOutboundQueueByteLimit(guard);
 
         // If we are at limit, we're probably really behind, so drop the entire
         // queue
-        if (isOverLimit(queue))
+        if (isOverLimit)
         {
             dropped = queue.size();
             mTxQueueByteCount = 0;
@@ -559,7 +554,7 @@ Json::Value
 FlowControl::getFlowControlJsonInfo(bool compact) const
 {
     releaseAssert(threadIsMain());
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
 
     Json::Value res;
     if (mFlowControlCapacity.getCapacity().mTotalCapacity)
@@ -601,7 +596,7 @@ FlowControl::getFlowControlJsonInfo(bool compact) const
 bool
 FlowControl::maybeThrottleRead()
 {
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     if (!canRead(guard))
     {
         CLOG_DEBUG(Overlay, "Throttle reading from peer {}",
@@ -615,7 +610,7 @@ FlowControl::maybeThrottleRead()
 void
 FlowControl::stopThrottling()
 {
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     releaseAssert(mLastThrottle);
     CLOG_DEBUG(Overlay, "Stop throttling reading from peer {}",
                mAppConnector.getConfig().toShortString(mNodeID));
@@ -627,7 +622,7 @@ FlowControl::stopThrottling()
 bool
 FlowControl::isThrottled() const
 {
-    std::lock_guard<std::mutex> guard(mFlowControlMutex);
+    MutexLocker guard(mFlowControlMutex);
     return static_cast<bool>(mLastThrottle);
 }