Minor fixes

Author: SirTyson

src/bucket/BucketList.cpp

@@ -63,7 +63,7 @@ BucketLevel::getNext()
 void
 BucketLevel::setNext(FutureBucket const& fb)
 {
-    releaseAssert(threadIsMain());
+    assertThreadIsMain();
     mNextCurr = fb;
 }
 
@@ -82,7 +82,7 @@ BucketLevel::getSnap() const
 void
 BucketLevel::setCurr(std::shared_ptr<Bucket> b)
 {
-    releaseAssert(threadIsMain());
+    assertThreadIsMain();
     mNextCurr.clear();
     mCurr = b;
 }
@@ -117,7 +117,7 @@ BucketList::shouldMergeWithEmptyCurr(uint32_t ledger, uint32_t level)
 void
 BucketLevel::setSnap(std::shared_ptr<Bucket> b)
 {
-    releaseAssert(threadIsMain());
+    assertThreadIsMain();
     mSnap = b;
 }
 

src/bucket/BucketList.h

@@ -478,6 +478,7 @@ 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);
 
     // Return a cumulative hash of the entire bucketlist; this is the hash of

src/bucket/BucketListSnapshot.cpp

@@ -16,7 +16,7 @@ namespace stellar
 BucketListSnapshot::BucketListSnapshot(BucketList const& bl, uint32_t ledgerSeq)
     : mLedgerSeq(ledgerSeq)
 {
-    releaseAssert(threadIsMain());
+    assertThreadIsMain();
 
     for (uint32_t i = 0; i < BucketList::kNumLevels; ++i)
     {
@@ -25,6 +25,11 @@ BucketListSnapshot::BucketListSnapshot(BucketList const& bl, uint32_t ledgerSeq)
     }
 }
 
+BucketListSnapshot::BucketListSnapshot(BucketListSnapshot const& snapshot)
+    : mLevels(snapshot.mLevels), mLedgerSeq(snapshot.mLedgerSeq)
+{
+}
+
 std::vector<BucketLevelSnapshot> const&
 BucketListSnapshot::getLevels() const
 {
@@ -153,7 +158,7 @@ SearchableBucketListSnapshot::loadPoolShareTrustLinesByAccountAndAsset(
     ZoneScoped;
 
     // This query should only be called during TX apply
-    releaseAssert(threadIsMain());
+    assertThreadIsMain();
     mSnapshotManager.maybeUpdateSnapshot(mSnapshot);
 
     LedgerKeySet trustlinesToLoad;
@@ -189,7 +194,7 @@ SearchableBucketListSnapshot::loadInflationWinners(size_t maxWinners,
 
     // This is a legacy query, should only be called by main thread during
     // catchup
-    releaseAssert(threadIsMain());
+    assertThreadIsMain();
     auto timer = mSnapshotManager.recordBulkLoadMetrics("inflationWinners", 0)
                      .TimeScope();
 

src/bucket/BucketListSnapshot.h

@@ -25,7 +25,7 @@ struct BucketLevelSnapshot
     BucketLevelSnapshot(BucketLevel const& level);
 };
 
-class BucketListSnapshot
+class BucketListSnapshot : public NonMovable
 {
   private:
     std::vector<BucketLevelSnapshot> mLevels;
@@ -36,15 +36,28 @@ class BucketListSnapshot
   public:
     BucketListSnapshot(BucketList const& bl, uint32_t ledgerSeq);
 
+    // Only allow copies via constructor
+    BucketListSnapshot(BucketListSnapshot const& snapshot);
+    BucketListSnapshot& operator=(BucketListSnapshot const&) = delete;
+
     std::vector<BucketLevelSnapshot> const& getLevels() const;
     uint32_t getLedgerSeq() const;
 };
 
-// A lightweight wrapper around the BucketList for thread safe BucketListDB
-// lookups
+// A lightweight wrapper around BucketListSnapshot for thread safe BucketListDB
+// lookups.
+//
+// Any thread that needs to perform BucketList lookups should retrieve
+// a single SearchableBucketListSnapshot instance from
+// BucketListSnapshotManager. On each lookup, the SearchableBucketListSnapshot
+// instance will check that the current snapshot is up to date via the
+// BucketListSnapshotManager and will be refreshed accordingly. Callers can
+// assume SearchableBucketListSnapshot is always up to date.
 class SearchableBucketListSnapshot : public NonMovableOrCopyable
 {
     BucketSnapshotManager const& mSnapshotManager;
+
+    // Snapshot managed by SnapshotManager
     std::unique_ptr<BucketListSnapshot const> mSnapshot{};
 
     // Loops through all buckets, starting with curr at level 0, then snap at

src/bucket/BucketManagerImpl.cpp

@@ -89,9 +89,13 @@ BucketManagerImpl::initialize()
     if (mApp.getConfig().MODE_ENABLES_BUCKETLIST)
     {
         mBucketList = std::make_unique<BucketList>();
-        mSnapshotManager = std::make_unique<BucketSnapshotManager>(
-            mApp.getMetrics(),
-            std::make_unique<BucketListSnapshot>(*mBucketList, 0));
+
+        if (mApp.getConfig().isUsingBucketListDB())
+        {
+            mSnapshotManager = std::make_unique<BucketSnapshotManager>(
+                mApp.getMetrics(),
+                std::make_unique<BucketListSnapshot>(*mBucketList, 0));
+        }
     }
 }
 
@@ -185,7 +189,7 @@ std::string const&
 BucketManagerImpl::getTmpDir()
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     if (!mWorkDir)
     {
         TmpDir t = mTmpDirManager->tmpDir("bucket");
@@ -268,7 +272,8 @@ BucketManagerImpl::getBucketList()
 BucketSnapshotManager&
 BucketManagerImpl::getBucketSnapshotManager() const
 {
-    releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
+    releaseAssertOrThrow(mApp.getConfig().isUsingBucketListDB());
+    releaseAssert(mSnapshotManager);
     return *mSnapshotManager;
 }
 
@@ -281,7 +286,7 @@ BucketManagerImpl::getMergeTimer()
 MergeCounters
 BucketManagerImpl::readMergeCounters()
 {
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     return mMergeCounters;
 }
 
@@ -371,7 +376,7 @@ MergeCounters::operator==(MergeCounters const& other) const
 void
 BucketManagerImpl::incrMergeCounters(MergeCounters const& delta)
 {
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     mMergeCounters += delta;
 }
 
@@ -397,7 +402,7 @@ BucketManagerImpl::adoptFileAsBucket(std::string const& filename,
 {
     ZoneScoped;
     releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
 
     if (mergeKey)
     {
@@ -481,7 +486,7 @@ BucketManagerImpl::noteEmptyMergeOutput(MergeKey const& mergeKey)
     // because it'd over-identify multiple individual inputs with the empty
     // output, potentially retaining far too many inputs, as lots of different
     // mergeKeys result in an empty output.
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     CLOG_TRACE(Bucket, "BucketManager::noteEmptyMergeOutput({})", mergeKey);
     mLiveFutures.erase(mergeKey);
 }
@@ -490,7 +495,7 @@ std::shared_ptr<Bucket>
 BucketManagerImpl::getBucketIfExists(uint256 const& hash)
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     auto i = mSharedBuckets.find(hash);
     if (i != mSharedBuckets.end())
     {
@@ -507,7 +512,7 @@ std::shared_ptr<Bucket>
 BucketManagerImpl::getBucketByHash(uint256 const& hash)
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     if (isZero(hash))
     {
         return std::make_shared<Bucket>();
@@ -540,7 +545,7 @@ std::shared_future<std::shared_ptr<Bucket>>
 BucketManagerImpl::getMergeFuture(MergeKey const& key)
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     MergeCounters mc;
     auto i = mLiveFutures.find(key);
     if (i == mLiveFutures.end())
@@ -586,7 +591,7 @@ BucketManagerImpl::putMergeFuture(
 {
     ZoneScoped;
     releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     CLOG_TRACE(
         Bucket,
         "BucketManager::putMergeFuture storing future for running merge {}",
@@ -598,7 +603,7 @@ BucketManagerImpl::putMergeFuture(
 void
 BucketManagerImpl::clearMergeFuturesForTesting()
 {
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     mLiveFutures.clear();
 }
 #endif
@@ -697,7 +702,7 @@ BucketManagerImpl::cleanupStaleFiles()
         return;
     }
 
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     auto referenced = getAllReferencedBuckets();
     std::transform(std::begin(mSharedBuckets), std::end(mSharedBuckets),
                    std::inserter(referenced, std::end(referenced)),
@@ -727,7 +732,7 @@ void
 BucketManagerImpl::forgetUnreferencedBuckets()
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
     auto referenced = getAllReferencedBuckets();
     auto blReferenced = getBucketListReferencedBuckets();
 
@@ -837,8 +842,12 @@ BucketManagerImpl::addBatch(Application& app, uint32_t currLedger,
     mBucketList->addBatch(app, currLedger, currLedgerProtocol, initEntries,
                           liveEntries, deadEntries);
     mBucketListSizeCounter.set_count(mBucketList->getSize());
-    mSnapshotManager->updateCurrentSnapshot(
-        std::make_unique<BucketListSnapshot>(*mBucketList, currLedger));
+
+    if (app.getConfig().isUsingBucketListDB())
+    {
+        mSnapshotManager->updateCurrentSnapshot(
+            std::make_unique<BucketListSnapshot>(*mBucketList, currLedger));
+    }
 }
 
 #ifdef BUILD_TESTS
@@ -978,60 +987,57 @@ BucketManagerImpl::assumeState(HistoryArchiveState const& has,
     ZoneScoped;
     releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
 
+    for (uint32_t i = 0; i < BucketList::kNumLevels; ++i)
     {
-        for (uint32_t i = 0; i < BucketList::kNumLevels; ++i)
+        auto curr = getBucketByHash(hexToBin256(has.currentBuckets.at(i).curr));
+        auto snap = getBucketByHash(hexToBin256(has.currentBuckets.at(i).snap));
+        if (!(curr && snap))
         {
-            auto curr =
-                getBucketByHash(hexToBin256(has.currentBuckets.at(i).curr));
-            auto snap =
-                getBucketByHash(hexToBin256(has.currentBuckets.at(i).snap));
-            if (!(curr && snap))
-            {
-                throw std::runtime_error("Missing bucket files while assuming "
-                                         "saved BucketList state");
-            }
+            throw std::runtime_error("Missing bucket files while assuming "
+                                     "saved BucketList state");
+        }
 
-            auto const& nextFuture = has.currentBuckets.at(i).next;
-            std::shared_ptr<Bucket> nextBucket = nullptr;
-            if (nextFuture.hasOutputHash())
+        auto const& nextFuture = has.currentBuckets.at(i).next;
+        std::shared_ptr<Bucket> nextBucket = nullptr;
+        if (nextFuture.hasOutputHash())
+        {
+            nextBucket =
+                getBucketByHash(hexToBin256(nextFuture.getOutputHash()));
+            if (!nextBucket)
             {
-                nextBucket =
-                    getBucketByHash(hexToBin256(nextFuture.getOutputHash()));
-                if (!nextBucket)
-                {
-                    throw std::runtime_error(
-                        "Missing future bucket files while "
-                        "assuming saved BucketList state");
-                }
+                throw std::runtime_error("Missing future bucket files while "
+                                         "assuming saved BucketList state");
             }
+        }
 
-            // Buckets on the BucketList should always be indexed when
-            // BucketListDB enabled
-            if (mApp.getConfig().isUsingBucketListDB())
+        // Buckets on the BucketList should always be indexed when
+        // BucketListDB enabled
+        if (mApp.getConfig().isUsingBucketListDB())
+        {
+            releaseAssert(curr->isEmpty() || curr->isIndexed());
+            releaseAssert(snap->isEmpty() || snap->isIndexed());
+            if (nextBucket)
             {
-                releaseAssert(curr->isEmpty() || curr->isIndexed());
-                releaseAssert(snap->isEmpty() || snap->isIndexed());
-                if (nextBucket)
-                {
-                    releaseAssert(nextBucket->isEmpty() ||
-                                  nextBucket->isIndexed());
-                }
+                releaseAssert(nextBucket->isEmpty() || nextBucket->isIndexed());
             }
-
-            mBucketList->getLevel(i).setCurr(curr);
-            mBucketList->getLevel(i).setSnap(snap);
-            mBucketList->getLevel(i).setNext(nextFuture);
         }
 
-        if (restartMerges)
-        {
-            mBucketList->restartMerges(mApp, maxProtocolVersion,
-                                       has.currentLedger);
-        }
+        mBucketList->getLevel(i).setCurr(curr);
+        mBucketList->getLevel(i).setSnap(snap);
+        mBucketList->getLevel(i).setNext(nextFuture);
+    }
+
+    if (restartMerges)
+    {
+        mBucketList->restartMerges(mApp, maxProtocolVersion, has.currentLedger);
     }
 
-    mSnapshotManager->updateCurrentSnapshot(
-        std::make_unique<BucketListSnapshot>(*mBucketList, has.currentLedger));
+    if (mApp.getConfig().isUsingBucketListDB())
+    {
+        mSnapshotManager->updateCurrentSnapshot(
+            std::make_unique<BucketListSnapshot>(*mBucketList,
+                                                 has.currentLedger));
+    }
     cleanupStaleFiles();
 }
 

src/bucket/BucketManagerImpl.h

@@ -43,6 +43,7 @@ class BucketManagerImpl : public BucketManager
     std::unique_ptr<TmpDirManager> mTmpDirManager;
     std::unique_ptr<TmpDir> mWorkDir;
     std::map<Hash, std::shared_ptr<Bucket>> mSharedBuckets;
+    mutable std::recursive_mutex mBucketMutex;
     std::unique_ptr<std::string> mLockedBucketDir;
     medida::Meter& mBucketObjectInsertBatch;
     medida::Timer& mBucketAddBatch;
@@ -54,11 +55,6 @@ class BucketManagerImpl : public BucketManager
     EvictionCounters mBucketListEvictionCounters;
     MergeCounters mMergeCounters;
 
-    // Lock for managing raw Bucket files or the bucket directory. This lock is
-    // only required for file access, but is not required for logical changes to
-    // the BucketList (i.e. addBatch).
-    mutable std::recursive_mutex mBucketFileMutex;
-
     bool const mDeleteEntireBucketDirInDtor;
 
     // Records bucket-merges that are currently _live_ in some FutureBucket, in