Bucket query asserts and perf fix (#5323)

# Description

The first commit resolves
#5067.

The 2nd commit addresses a perf issue we saw with the BucketList
refactor. We build snapshot objects in parallel contents. Currently, the
constructor gets a fresh set of metric references for each snapshot,
acquiring a lock to do so. This changes it such that we use a single
shared metrics object across all snapshots. Metrics are thread safe, so
there should be no safety issue with this, and it makes the snapshot
constructor much cheaper.

# Checklist
- [x] Reviewed the
[contributing](https://github.com/stellar/stellar-core/blob/master/CONTRIBUTING.md#submitting-changes)
document
- [x] Rebased on top of master (no merge commits)
- [x] Ran `clang-format` v8.0.0 (via `make format` or the Visual Studio
extension)
- [x] Compiles
- [x] Ran all tests
- [x] If change impacts performance, include supporting evidence per the
[performance
document](https://github.com/stellar/stellar-core/blob/master/performance-eval/performance-eval.md)

Author: SirTyson

src/bucket/BucketListSnapshot.cpp

@@ -53,27 +53,44 @@ BucketListSnapshotData<BucketT>::BucketListSnapshotData(
 {
 }
 
+//
+// BucketSnapshotMetrics
+//
+
+template <class BucketT>
+BucketSnapshotMetrics<BucketT>::BucketSnapshotMetrics(MetricsRegistry& metrics)
+    : mPointTimers([&metrics]() {
+        UnorderedMap<LedgerEntryType, std::reference_wrapper<SimpleTimer>>
+            timers;
+        for (auto t : xdr::xdr_traits<LedgerEntryType>::enum_values())
+        {
+            auto const& label = xdr::xdr_traits<LedgerEntryType>::enum_name(
+                static_cast<LedgerEntryType>(t));
+            auto& metric = metrics.NewSimpleTimer(
+                {BucketT::METRIC_STRING, label}, std::chrono::microseconds{1});
+            timers.emplace(static_cast<LedgerEntryType>(t), metric);
+        }
+        return timers;
+    }())
+    , mBulkLoadMeter(
+          metrics.NewMeter({BucketT::METRIC_STRING, "query", "loads"}, "query"))
+{
+}
+
 //
 // SearchableBucketListSnapshot
 //
 
 template <class BucketT>
 SearchableBucketListSnapshot<BucketT>::SearchableBucketListSnapshot(
     MetricsRegistry& metrics,
+    std::shared_ptr<BucketSnapshotMetrics<BucketT> const> snapshotMetrics,
     std::shared_ptr<BucketListSnapshotData<BucketT> const> data)
     : mData(std::move(data))
     , mMetrics(metrics)
-    , mBulkLoadMeter(
-          metrics.NewMeter({BucketT::METRIC_STRING, "query", "loads"}, "query"))
+    , mSnapshotMetrics(std::move(snapshotMetrics))
 {
-    for (auto t : xdr::xdr_traits<LedgerEntryType>::enum_values())
-    {
-        auto const& label = xdr::xdr_traits<LedgerEntryType>::enum_name(
-            static_cast<LedgerEntryType>(t));
-        auto& metric = metrics.NewSimpleTimer({BucketT::METRIC_STRING, label},
-                                              std::chrono::microseconds{1});
-        mPointTimers.emplace(static_cast<LedgerEntryType>(t), metric);
-    }
+    releaseAssert(mSnapshotMetrics);
 }
 
 template <class BucketT>
@@ -82,9 +99,8 @@ SearchableBucketListSnapshot<BucketT>::SearchableBucketListSnapshot(
     : mData(other.mData)
     // mStreams intentionally left empty — each copy gets its own stream cache
     , mMetrics(other.mMetrics)
-    , mPointTimers(other.mPointTimers)
+    , mSnapshotMetrics(other.mSnapshotMetrics)
     , mBulkTimers(other.mBulkTimers)
-    , mBulkLoadMeter(other.mBulkLoadMeter)
 {
 }
 
@@ -98,20 +114,45 @@ SearchableBucketListSnapshot<BucketT>::operator=(
         mData = other.mData;
         mStreams.clear();
         mMetrics = other.mMetrics;
-        mPointTimers = other.mPointTimers;
+        mSnapshotMetrics = other.mSnapshotMetrics;
         mBulkTimers = other.mBulkTimers;
-        mBulkLoadMeter = other.mBulkLoadMeter;
+#ifdef BUILD_TESTS
+        // Reset thread ownership so the copy can be claimed by another thread.
+        mThreadId.store(std::thread::id{});
+#endif
     }
     return *this;
 }
 
+// Bucket loads are not thread safe and a single snapshot instance should only
+// be queried by one thread. We cache the initial caller's thread id and assert
+// following queries are from the same thread. Note: this only guards the
+// bucket-loading query entry points; access to the immutable underlying
+// snapshot data is thread safe.
+template <class BucketT>
+void
+SearchableBucketListSnapshot<BucketT>::threadInvariant() const
+{
+#ifdef BUILD_TESTS
+    auto const current = std::this_thread::get_id();
+    std::thread::id unclaimed{};
+    // Atomically claim ownership on first use, so any concurrent claimant sees
+    // the CAS fail with `unclaimed` set to the owner's id and asserts.
+    if (!mThreadId.compare_exchange_strong(unclaimed, current))
+    {
+        releaseAssert(unclaimed == current);
+    }
+#endif
+}
+
 // File streams are fairly expensive to create, so they are lazily created and
 // stored in mStreams.
 template <class BucketT>
 XDRInputFileStream&
 SearchableBucketListSnapshot<BucketT>::getStream(
     std::shared_ptr<BucketT const> const& bucket) const
 {
+    threadInvariant();
     BucketT const* key = bucket.get();
     auto it = mStreams.find(key);
     if (it == mStreams.end())
@@ -307,9 +348,10 @@ SearchableBucketListSnapshot<BucketT>::load(LedgerKey const& k) const
 {
     ZoneScoped;
     releaseAssert(mData);
+    threadInvariant();
 
-    auto timerIter = mPointTimers.find(k.type());
-    releaseAssert(timerIter != mPointTimers.end());
+    auto timerIter = mSnapshotMetrics->mPointTimers.find(k.type());
+    releaseAssert(timerIter != mSnapshotMetrics->mPointTimers.end());
     auto timer = timerIter->second.get().TimeScope();
 
     std::shared_ptr<typename BucketT::LoadT const> result{};
@@ -341,9 +383,13 @@ medida::Timer&
 SearchableBucketListSnapshot<BucketT>::getBulkLoadTimer(
     std::string const& label, size_t numEntries) const
 {
+    // mBulkTimers is per-snapshot mutable state lazily populated here, so this
+    // must be single-threaded. Enforced here as well as at the public query
+    // entry points.
+    threadInvariant();
     if (numEntries != 0)
     {
-        mBulkLoadMeter.get().Mark(numEntries);
+        mSnapshotMetrics->mBulkLoadMeter.get().Mark(numEntries);
     }
 
     auto iter = mBulkTimers.find(label);
@@ -370,8 +416,10 @@ SearchableBucketListSnapshot<BucketT>::getSnapshotData() const
 
 SearchableLiveBucketListSnapshot::SearchableLiveBucketListSnapshot(
     MetricsRegistry& metrics,
+    std::shared_ptr<BucketSnapshotMetrics<LiveBucket> const> snapshotMetrics,
     std::shared_ptr<BucketListSnapshotData<LiveBucket> const> data)
-    : SearchableBucketListSnapshot<LiveBucket>(metrics, std::move(data))
+    : SearchableBucketListSnapshot<LiveBucket>(
+          metrics, std::move(snapshotMetrics), std::move(data))
 {
 }
 
@@ -383,6 +431,7 @@ SearchableBucketListSnapshot<BucketT>::loadKeys(
 {
     ZoneScoped;
     releaseAssert(mData);
+    threadInvariant();
     auto timer = getBulkLoadTimer(label, inKeys.size()).TimeScope();
 
     auto keys = inKeys;
@@ -406,6 +455,7 @@ SearchableLiveBucketListSnapshot::loadPoolShareTrustLinesByAccountAndAsset(
 {
     ZoneScoped;
     releaseAssert(mData);
+    threadInvariant();
 
     LedgerKeySet trustlinesToLoad;
 
@@ -448,6 +498,7 @@ SearchableLiveBucketListSnapshot::loadInflationWinners(size_t maxWinners,
 {
     ZoneScoped;
     releaseAssert(mData);
+    threadInvariant();
 
     auto timer = getBulkLoadTimer("inflationWinners", 0).TimeScope();
 
@@ -548,6 +599,7 @@ SearchableLiveBucketListSnapshot::scanForEviction(
     ZoneScoped;
     releaseAssert(mData);
     releaseAssert(stats);
+    threadInvariant();
 
     auto getBucketFromIter =
         [&levels = mData->levels](
@@ -599,6 +651,7 @@ SearchableLiveBucketListSnapshot::scanForEntriesOfType(
 {
     ZoneScoped;
     releaseAssert(mData);
+    threadInvariant();
 
     auto scanBucket = [&](std::shared_ptr<LiveBucket const> const& bucket) {
         if (bucket->isEmpty())
@@ -806,8 +859,11 @@ SearchableLiveBucketListSnapshot::scanForEvictionInBucket(
 
 SearchableHotArchiveBucketListSnapshot::SearchableHotArchiveBucketListSnapshot(
     MetricsRegistry& metrics,
+    std::shared_ptr<BucketSnapshotMetrics<HotArchiveBucket> const>
+        snapshotMetrics,
     std::shared_ptr<BucketListSnapshotData<HotArchiveBucket> const> data)
-    : SearchableBucketListSnapshot<HotArchiveBucket>(metrics, std::move(data))
+    : SearchableBucketListSnapshot<HotArchiveBucket>(
+          metrics, std::move(snapshotMetrics), std::move(data))
 {
 }
 
@@ -817,6 +873,7 @@ SearchableHotArchiveBucketListSnapshot::scanAllEntries(
 {
     ZoneScoped;
     releaseAssert(mData);
+    threadInvariant();
 
     auto scanBucket =
         [&](std::shared_ptr<HotArchiveBucket const> const& bucket) {
@@ -841,6 +898,8 @@ SearchableHotArchiveBucketListSnapshot::scanAllEntries(
 // Explicit template instantiations
 template struct BucketListSnapshotData<LiveBucket>;
 template struct BucketListSnapshotData<HotArchiveBucket>;
+template struct BucketSnapshotMetrics<LiveBucket>;
+template struct BucketSnapshotMetrics<HotArchiveBucket>;
 template class SearchableBucketListSnapshot<LiveBucket>;
 template class SearchableBucketListSnapshot<HotArchiveBucket>;
 

src/bucket/BucketListSnapshot.h

@@ -14,11 +14,13 @@
 #include "util/XDRStream.h"
 #include "xdr/Stellar-ledger-entries.h"
 
+#include <atomic>
 #include <functional>
 #include <list>
 #include <memory>
 #include <set>
 #include <string>
+#include <thread>
 #include <vector>
 
 namespace medida
@@ -64,14 +66,38 @@ template <class BucketT> struct BucketListSnapshotData
     explicit BucketListSnapshotData(BucketListBase<BucketT> const& bl);
 };
 
+// Pre-resolved metric references for snapshot queries. Resolving a metric
+// takes the global MetricsRegistry lock, so metrics are resolved once and
+// shared across snapshots rather than re-resolved in every
+// SearchableBucketListSnapshot constructor.
+template <class BucketT> struct BucketSnapshotMetrics
+{
+    BUCKET_TYPE_ASSERT(BucketT);
+
+    // Tracks load times for each LedgerEntryType. We use
+    // SimpleTimer since medida Timer overhead is too expensive for point
+    // loads.
+    UnorderedMap<LedgerEntryType, std::reference_wrapper<SimpleTimer>> const
+        mPointTimers;
+    std::reference_wrapper<medida::Meter> const mBulkLoadMeter;
+
+    explicit BucketSnapshotMetrics(MetricsRegistry& metrics);
+};
+
 // SearchableBucketListSnapshot provides BucketList lookup functionality.
 // Each snapshot maintains its own stream cache for file I/O and a pointer to
 // immutable snapshot data (ledger header, list of referenced buckets, etc).
 //
 // Thread-safety:
-// - A single snapshot instance must only be used by one thread at a time.
+// - A single snapshot instance must only be used by one thread at a time. In
+//   test builds this is enforced by threadInvariant(): the first thread to
+//   issue a query claims the snapshot, and subsequent queries from a different
+//   thread trigger an assertion.
 // - The underlying snapshot data (shared via shared_ptr) is immutable and
 //   can be safely shared.
+// - To query from another thread, make a copy: copies reset the cached stream
+//   cache and (in test builds) the owning-thread id, so each copy can be
+//   claimed by a different thread.
 template <class BucketT> class SearchableBucketListSnapshot
 {
     BUCKET_TYPE_ASSERT(BucketT);
@@ -84,18 +110,30 @@ template <class BucketT> class SearchableBucketListSnapshot
     mutable UnorderedMap<BucketT const*, std::unique_ptr<XDRInputFileStream>>
         mStreams;
 
+#ifdef BUILD_TESTS
+    // Thread id of the first thread to issue a query on this snapshot instance.
+    // Used by threadInvariant() to assert that a single snapshot is not queried
+    // concurrently from multiple threads. Reset on copy so each copy can be
+    // claimed by a different thread.
+    mutable std::atomic<std::thread::id> mThreadId{};
+#endif
+
+    // Bucket loads are not thread safe and a single snapshot instance should
+    // only be queried by one thread. We cache the initial caller's thread id
+    // and assert following queries are from the same thread (test builds only).
+    void threadInvariant() const;
+
     std::reference_wrapper<MetricsRegistry> mMetrics;
 
-    // Tracks load times for each LedgerEntryType. We use
-    // SimpleTimer since medida Timer overhead is too expensive for point loads.
-    UnorderedMap<LedgerEntryType, std::reference_wrapper<SimpleTimer>>
-        mPointTimers;
+    // Pre-resolved point load timers and bulk load meter, shared across
+    // snapshots (see BucketSnapshotMetrics).
+    std::shared_ptr<BucketSnapshotMetrics<BucketT> const> mSnapshotMetrics;
 
     // Bulk load timers take significantly longer, so the timer overhead is
-    // comparatively negligible.
+    // comparatively negligible. Resolved lazily per label, so kept
+    // per-instance rather than in BucketSnapshotMetrics.
     mutable UnorderedMap<std::string, std::reference_wrapper<medida::Timer>>
         mBulkTimers;
-    std::reference_wrapper<medida::Meter> mBulkLoadMeter;
 
     // Returns (lazily-constructed) file stream for bucket file. Note
     // this might be in some random position left over from a previous read --
@@ -136,6 +174,7 @@ template <class BucketT> class SearchableBucketListSnapshot
 
     SearchableBucketListSnapshot(
         MetricsRegistry& metrics,
+        std::shared_ptr<BucketSnapshotMetrics<BucketT> const> snapshotMetrics,
         std::shared_ptr<BucketListSnapshotData<BucketT> const> data);
 
   public:
@@ -170,6 +209,8 @@ class SearchableLiveBucketListSnapshot
 {
     SearchableLiveBucketListSnapshot(
         MetricsRegistry& metrics,
+        std::shared_ptr<BucketSnapshotMetrics<LiveBucket> const>
+            snapshotMetrics,
         std::shared_ptr<BucketListSnapshotData<LiveBucket> const> data);
 
     Loop scanForEvictionInBucket(
@@ -210,6 +251,8 @@ class SearchableHotArchiveBucketListSnapshot
 {
     SearchableHotArchiveBucketListSnapshot(
         MetricsRegistry& metrics,
+        std::shared_ptr<BucketSnapshotMetrics<HotArchiveBucket> const>
+            snapshotMetrics,
         std::shared_ptr<BucketListSnapshotData<HotArchiveBucket> const> data);
 
   public:
@@ -226,6 +269,8 @@ class SearchableHotArchiveBucketListSnapshot
 
 extern template struct BucketListSnapshotData<LiveBucket>;
 extern template struct BucketListSnapshotData<HotArchiveBucket>;
+extern template struct BucketSnapshotMetrics<LiveBucket>;
+extern template struct BucketSnapshotMetrics<HotArchiveBucket>;
 extern template class SearchableBucketListSnapshot<LiveBucket>;
 extern template class SearchableBucketListSnapshot<HotArchiveBucket>;
 

src/ledger/ImmutableLedgerView.cpp

@@ -225,12 +225,16 @@ ImmutableLedgerData::checkInvariant() const
 ImmutableLedgerData::ImmutableLedgerData(
     LiveBucketList const& liveBL, HotArchiveBucketList const& hotArchiveBL,
     LedgerHeaderHistoryEntry const& lcl, HistoryArchiveState const& has,
-    std::optional<SorobanNetworkConfig> sorobanConfig)
+    std::optional<SorobanNetworkConfig> sorobanConfig, MetricsRegistry& metrics)
     : mLiveBucketData(
           std::make_shared<BucketListSnapshotData<LiveBucket>>(liveBL))
     , mHotArchiveBucketData(
           std::make_shared<BucketListSnapshotData<HotArchiveBucket>>(
               hotArchiveBL))
+    , mLiveSnapshotMetrics(
+          std::make_shared<BucketSnapshotMetrics<LiveBucket>>(metrics))
+    , mHotArchiveSnapshotMetrics(
+          std::make_shared<BucketSnapshotMetrics<HotArchiveBucket>>(metrics))
     , mSorobanConfig(std::move(sorobanConfig))
     , mLastClosedLedgerHeader(lcl)
     , mLastClosedHistoryArchiveState(has)
@@ -275,19 +279,22 @@ ImmutableLedgerData::createAndMaybeLoadConfig(
         // Bootstrap: build a lightweight temporary state just to load config
         // from the current live bucket list.
         auto tempState = std::make_shared<ImmutableLedgerData>(
-            liveBL, hotArchiveBL, lcl, has, /*sorobanConfig*/ std::nullopt);
+            liveBL, hotArchiveBL, lcl, has, /*sorobanConfig*/ std::nullopt,
+            metrics);
         ImmutableLedgerView tempView(tempState, metrics);
         sorobanConfig = SorobanNetworkConfig::loadFromLedger(tempView);
     }
-    return std::make_shared<ImmutableLedgerData>(liveBL, hotArchiveBL, lcl, has,
-                                                 std::move(sorobanConfig));
+    return std::make_shared<ImmutableLedgerData>(
+        liveBL, hotArchiveBL, lcl, has, std::move(sorobanConfig), metrics);
 }
 
 ImmutableLedgerView::ImmutableLedgerView(ImmutableLedgerDataPtr state,
                                          MetricsRegistry& metrics)
     : mState(state)
-    , mLiveSnapshot(metrics, state->mLiveBucketData)
-    , mHotArchiveSnapshot(metrics, state->mHotArchiveBucketData)
+    , mLiveSnapshot(metrics, state->mLiveSnapshotMetrics,
+                    state->mLiveBucketData)
+    , mHotArchiveSnapshot(metrics, state->mHotArchiveSnapshotMetrics,
+                          state->mHotArchiveBucketData)
     , mMetrics(metrics)
 {
 }