BucketSnapshotManager manages all concurrency

Author: SirTyson

src/bucket/Bucket.h

@@ -165,6 +165,6 @@ class Bucket : public std::enable_shared_from_this<Bucket>,
     static uint32_t
     getBucketVersion(std::shared_ptr<Bucket const> const& bucket);
 
-    friend class SearchableBucketSnapshot;
+    friend class BucketSnapshot;
 };
 }

src/bucket/BucketList.cpp

@@ -6,8 +6,8 @@
 #include "bucket/Bucket.h"
 #include "bucket/BucketInputIterator.h"
 #include "bucket/BucketManager.h"
+#include "bucket/BucketSnapshot.h"
 #include "bucket/LedgerCmp.h"
-#include "bucket/SearchableBucketSnapshot.h"
 #include "crypto/Hex.h"
 #include "crypto/Random.h"
 #include "crypto/SHA.h"
@@ -454,12 +454,6 @@ BucketList::getLevel(uint32_t i)
     return mLevels.at(i);
 }
 
-uint32_t
-BucketList::getLedgerSeq() const
-{
-    return mLedgerSeq;
-}
-
 void
 BucketList::resolveAnyReadyFutures()
 {
@@ -532,10 +526,6 @@ BucketList::addBatch(Application& app, uint32_t currLedger,
 {
     ZoneScoped;
     releaseAssert(currLedger > 0);
-    releaseAssert(threadIsMain());
-
-    std::lock_guard<std::recursive_mutex> lock(
-        app.getBucketManager().getBucketListMutex());
 
     std::vector<std::shared_ptr<Bucket>> shadows;
     for (auto& level : mLevels)
@@ -647,8 +637,6 @@ BucketList::addBatch(Application& app, uint32_t currLedger,
     {
         resolveAnyReadyFutures();
     }
-
-    mLedgerSeq = currLedger;
 }
 
 void

src/bucket/BucketList.h

@@ -431,9 +431,6 @@ class BucketList
     // nullopt and be initialized at the start of the next cycle.
     std::optional<EvictionStatistics> mEvictionStatistics;
 
-    // Current ledgerSeq for BucketList state
-    uint32_t mLedgerSeq{};
-
   public:
     // Number of bucket levels in the bucketlist. Every bucketlist in the system
     // will have this many levels and it effectively gets wired-in to the
@@ -481,12 +478,8 @@ class BucketList
     // Return level `i` of the BucketList.
     BucketLevel const& getLevel(uint32_t i) const;
 
-    // Return level `i` of the BucketList.
     BucketLevel& getLevel(uint32_t i);
 
-    // Not thread safe, must hold mBucketListMutex
-    uint32_t getLedgerSeq() const;
-
     // Return a cumulative hash of the entire bucketlist; this is the hash of
     // the concatenation of each level's hash, each of which in turn is the hash
     // of the concatenation of the hashes of the `curr` and `snap` buckets.
@@ -535,7 +528,6 @@ class BucketList
     // should have spilled due to passing through `currLedger`. The `currLedger`
     // and `currProtocolVersion` values should be taken from the ledger at which
     // this batch is being added.
-    // Not thread safe, must hold mBucketListMutex.
     void addBatch(Application& app, uint32_t currLedger,
                   uint32_t currLedgerProtocol,
                   std::vector<LedgerEntry> const& initEntries,

src/bucket/SearchableBucketListSnapshot.cpp src/bucket/BucketListSnapshot.cpp

@@ -2,7 +2,7 @@
 // under the Apache License, Version 2.0. See the COPYING file at the root
 // of this distribution or at http://www.apache.org/licenses/LICENSE-2.0
 
-#include "bucket/SearchableBucketListSnapshot.h"
+#include "bucket/BucketListSnapshot.h"
 #include "bucket/BucketInputIterator.h"
 #include "crypto/SecretKey.h"
 #include "ledger/LedgerTxn.h"
@@ -13,14 +13,40 @@
 namespace stellar
 {
 
+BucketListSnapshot::BucketListSnapshot(BucketList const& bl, uint32_t ledgerSeq)
+    : mLedgerSeq(ledgerSeq)
+{
+    releaseAssert(threadIsMain());
+
+    for (uint32_t i = 0; i < BucketList::kNumLevels; ++i)
+    {
+        auto const& level = bl.getLevel(i);
+        mLevels.emplace_back(BucketLevelSnapshot(level));
+    }
+}
+
+std::vector<BucketLevelSnapshot> const&
+BucketListSnapshot::getLevels() const
+{
+    return mLevels;
+}
+
+uint32_t
+BucketListSnapshot::getLedgerSeq() const
+{
+    return mLedgerSeq;
+}
+
 void
 SearchableBucketListSnapshot::loopAllBuckets(
-    std::function<bool(SearchableBucketSnapshot const&)> f) const
+    std::function<bool(BucketSnapshot const&)> f) const
 {
-    for (auto const& lev : mLevels)
+    releaseAssert(mSnapshot);
+
+    for (auto const& lev : mSnapshot->getLevels())
     {
         // Return true if we should exit loop early
-        auto processBucket = [f](SearchableBucketSnapshot const& b) {
+        auto processBucket = [f](BucketSnapshot const& b) {
             if (b.isEmpty())
             {
                 return false;
@@ -40,7 +66,7 @@ std::shared_ptr<LedgerEntry>
 SearchableBucketListSnapshot::getLedgerEntry(LedgerKey const& k)
 {
     ZoneScoped;
-    mSnapshotManager.maybeUpdateSnapshot(*this);
+    mSnapshotManager.maybeUpdateSnapshot(mSnapshot);
 
     if (threadIsMain())
     {
@@ -58,7 +84,7 @@ SearchableBucketListSnapshot::getLedgerEntryInternal(LedgerKey const& k)
 {
     std::shared_ptr<LedgerEntry> result{};
 
-    auto f = [&](SearchableBucketSnapshot const& b) {
+    auto f = [&](BucketSnapshot const& b) {
         auto be = b.getBucketEntry(k);
         if (be.has_value())
         {
@@ -86,7 +112,7 @@ SearchableBucketListSnapshot::loadKeysInternal(
 
     // Make a copy of the key set, this loop is destructive
     auto keys = inKeys;
-    auto f = [&](SearchableBucketSnapshot const& b) {
+    auto f = [&](BucketSnapshot const& b) {
         b.loadKeys(keys, entries);
         return keys.empty();
     };
@@ -100,7 +126,7 @@ SearchableBucketListSnapshot::loadKeys(
     std::set<LedgerKey, LedgerEntryIdCmp> const& inKeys)
 {
     ZoneScoped;
-    mSnapshotManager.maybeUpdateSnapshot(*this);
+    mSnapshotManager.maybeUpdateSnapshot(mSnapshot);
 
     if (threadIsMain())
     {
@@ -128,11 +154,11 @@ SearchableBucketListSnapshot::loadPoolShareTrustLinesByAccountAndAsset(
 
     // This query should only be called during TX apply
     releaseAssert(threadIsMain());
-    mSnapshotManager.maybeUpdateSnapshot(*this);
+    mSnapshotManager.maybeUpdateSnapshot(mSnapshot);
 
     LedgerKeySet trustlinesToLoad;
 
-    auto trustLineLoop = [&](SearchableBucketSnapshot const& b) {
+    auto trustLineLoop = [&](BucketSnapshot const& b) {
         for (auto const& poolID : b.getPoolIDsByAsset(asset))
         {
             LedgerKey trustlineKey(TRUSTLINE);
@@ -159,7 +185,7 @@ SearchableBucketListSnapshot::loadInflationWinners(size_t maxWinners,
                                                    int64_t minBalance)
 {
     ZoneScoped;
-    mSnapshotManager.maybeUpdateSnapshot(*this);
+    mSnapshotManager.maybeUpdateSnapshot(mSnapshot);
 
     // This is a legacy query, should only be called by main thread during
     // catchup
@@ -170,7 +196,7 @@ SearchableBucketListSnapshot::loadInflationWinners(size_t maxWinners,
     UnorderedMap<AccountID, int64_t> voteCount;
     UnorderedSet<AccountID> seen;
 
-    auto countVotesInBucket = [&](SearchableBucketSnapshot const& b) {
+    auto countVotesInBucket = [&](BucketSnapshot const& b) {
         for (BucketInputIterator in(b.getRawBucket()); in; ++in)
         {
             BucketEntry const& be = *in;
@@ -246,22 +272,16 @@ SearchableBucketListSnapshot::loadInflationWinners(size_t maxWinners,
     return winners;
 }
 
-SearchableBucketLevelSnapshot::SearchableBucketLevelSnapshot(
-    BucketLevel const& level)
+BucketLevelSnapshot::BucketLevelSnapshot(BucketLevel const& level)
     : curr(level.getCurr()), snap(level.getSnap())
 {
 }
 
-// This is not thread safe, must call while holding
-// BucketManager::mBucketListMutex.
 SearchableBucketListSnapshot::SearchableBucketListSnapshot(
-    BucketSnapshotManager const& snapshotManager, BucketList const& bl)
+    BucketSnapshotManager const& snapshotManager)
     : mSnapshotManager(snapshotManager)
 {
-    for (uint32_t i = 0; i < BucketList::kNumLevels; ++i)
-    {
-        auto const& level = bl.getLevel(i);
-        mLevels.emplace_back(SearchableBucketLevelSnapshot(level));
-    }
+    // Initialize snapshot from SnapshotManager
+    mSnapshotManager.maybeUpdateSnapshot(mSnapshot);
 }
 }

src/bucket/SearchableBucketListSnapshot.h src/bucket/BucketListSnapshot.h

@@ -6,8 +6,8 @@
 
 #include "bucket/BucketList.h"
 #include "bucket/BucketManagerImpl.h"
+#include "bucket/BucketSnapshot.h"
 #include "bucket/BucketSnapshotManager.h"
-#include "bucket/SearchableBucketSnapshot.h"
 
 namespace medida
 {
@@ -17,38 +17,51 @@ class Timer;
 namespace stellar
 {
 
-struct SearchableBucketLevelSnapshot
+struct BucketLevelSnapshot
 {
-    SearchableBucketSnapshot curr;
-    SearchableBucketSnapshot snap;
+    BucketSnapshot curr;
+    BucketSnapshot snap;
 
-    SearchableBucketLevelSnapshot(BucketLevel const& level);
+    BucketLevelSnapshot(BucketLevel const& level);
+};
+
+class BucketListSnapshot
+{
+  private:
+    std::vector<BucketLevelSnapshot> mLevels;
+
+    // ledgerSeq that this BucketList snapshot is based off of
+    uint32_t mLedgerSeq;
+
+  public:
+    BucketListSnapshot(BucketList const& bl, uint32_t ledgerSeq);
+
+    std::vector<BucketLevelSnapshot> const& getLevels() const;
+    uint32_t getLedgerSeq() const;
 };
 
 // A lightweight wrapper around the BucketList for thread safe BucketListDB
 // lookups
 class SearchableBucketListSnapshot : public NonMovableOrCopyable
 {
-    std::vector<SearchableBucketLevelSnapshot> mLevels;
     BucketSnapshotManager const& mSnapshotManager;
-
-    // ledgerSeq that this BucketList snapshot is based off of
-    uint32_t mLedgerSeq;
+    std::unique_ptr<BucketListSnapshot const> mSnapshot{};
 
     // Loops through all buckets, starting with curr at level 0, then snap at
     // level 0, etc. Calls f on each bucket. Exits early if function
     // returns true
-    void loopAllBuckets(
-        std::function<bool(SearchableBucketSnapshot const&)> f) const;
-
-    SearchableBucketListSnapshot(BucketSnapshotManager const& snapshotManager,
-                                 BucketList const& bl);
+    void loopAllBuckets(std::function<bool(BucketSnapshot const&)> f) const;
 
     std::vector<LedgerEntry>
     loadKeysInternal(std::set<LedgerKey, LedgerEntryIdCmp> const& inKeys);
 
     std::shared_ptr<LedgerEntry> getLedgerEntryInternal(LedgerKey const& k);
 
+    SearchableBucketListSnapshot(BucketSnapshotManager const& snapshotManager);
+
+    friend std::unique_ptr<SearchableBucketListSnapshot>
+    BucketSnapshotManager::getSearchableBucketListSnapshot() const;
+
   public:
     std::vector<LedgerEntry>
     loadKeys(std::set<LedgerKey, LedgerEntryIdCmp> const& inKeys);
@@ -61,7 +74,5 @@ class SearchableBucketListSnapshot : public NonMovableOrCopyable
                                                       int64_t minBalance);
 
     std::shared_ptr<LedgerEntry> getLedgerEntry(LedgerKey const& k);
-
-    friend class BucketSnapshotManager;
 };
 }

src/bucket/BucketManager.h

@@ -209,8 +209,6 @@ class BucketManager : NonMovableOrCopyable
     virtual void maybeSetIndex(std::shared_ptr<Bucket> b,
                                std::unique_ptr<BucketIndex const>&& index) = 0;
 
-    virtual std::recursive_mutex& getBucketListMutex() const = 0;
-
     // Scans BucketList for non-live entries to evict starting at the entry
     // pointed to by EvictionIterator. Scans until `maxEntriesToEvict` entries
     // have been evicted or maxEvictionScanSize bytes have been scanned.

src/bucket/BucketManagerImpl.cpp

@@ -6,9 +6,9 @@
 #include "bucket/Bucket.h"
 #include "bucket/BucketInputIterator.h"
 #include "bucket/BucketList.h"
+#include "bucket/BucketListSnapshot.h"
 #include "bucket/BucketOutputIterator.h"
 #include "bucket/BucketSnapshotManager.h"
-#include "bucket/SearchableBucketListSnapshot.h"
 #include "crypto/Hex.h"
 #include "history/HistoryManager.h"
 #include "historywork/VerifyBucketWork.h"
@@ -90,7 +90,8 @@ BucketManagerImpl::initialize()
     {
         mBucketList = std::make_unique<BucketList>();
         mSnapshotManager = std::make_unique<BucketSnapshotManager>(
-            mApp.getMetrics(), *mBucketList, mBucketListMutex);
+            mApp.getMetrics(),
+            std::make_unique<BucketListSnapshot>(*mBucketList, 0));
     }
 }
 
@@ -833,12 +834,11 @@ BucketManagerImpl::addBatch(Application& app, uint32_t currLedger,
     auto timer = mBucketAddBatch.TimeScope();
     mBucketObjectInsertBatch.Mark(initEntries.size() + liveEntries.size() +
                                   deadEntries.size());
-    {
-        std::lock_guard<std::recursive_mutex> lock(mBucketListMutex);
-        mBucketList->addBatch(app, currLedger, currLedgerProtocol, initEntries,
-                              liveEntries, deadEntries);
-    }
+    mBucketList->addBatch(app, currLedger, currLedgerProtocol, initEntries,
+                          liveEntries, deadEntries);
     mBucketListSizeCounter.set_count(mBucketList->getSize());
+    mSnapshotManager->updateCurrentSnapshot(
+        std::make_unique<BucketListSnapshot>(*mBucketList, currLedger));
 }
 
 #ifdef BUILD_TESTS
@@ -918,12 +918,6 @@ BucketManagerImpl::scanForEvictionLegacySQL(AbstractLedgerTxn& ltx,
     }
 }
 
-std::recursive_mutex&
-BucketManagerImpl::getBucketListMutex() const
-{
-    return mBucketListMutex;
-}
-
 medida::Meter&
 BucketManagerImpl::getBloomMissMeter() const
 {
@@ -985,7 +979,6 @@ BucketManagerImpl::assumeState(HistoryArchiveState const& has,
     releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
 
     {
-        std::lock_guard<std::recursive_mutex> lock(mBucketListMutex);
         for (uint32_t i = 0; i < BucketList::kNumLevels; ++i)
         {
             auto curr =
@@ -1037,6 +1030,8 @@ BucketManagerImpl::assumeState(HistoryArchiveState const& has,
         }
     }
 
+    mSnapshotManager->updateCurrentSnapshot(
+        std::make_unique<BucketListSnapshot>(*mBucketList, has.currentLedger));
     cleanupStaleFiles();
 }