Added concurrency locks for BucketListDB

Author: SirTyson

src/bucket/BucketManager.h

@@ -208,7 +208,7 @@ class BucketManager : NonMovableOrCopyable
                                std::unique_ptr<BucketIndex const>&& index) = 0;
 
     virtual std::unique_ptr<SearchableBucketListSnapshot const>
-    getSearchableBucketListSnapshot(bool isMainThread = false) const = 0;
+    getSearchableBucketListSnapshot() const = 0;
 
     // Scans BucketList for non-live entries to evict starting at the entry
     // pointed to by EvictionIterator. Scans until `maxEntriesToEvict` entries

src/bucket/BucketManagerImpl.cpp

@@ -7,7 +7,7 @@
 #include "bucket/BucketInputIterator.h"
 #include "bucket/BucketList.h"
 #include "bucket/BucketOutputIterator.h"
-#include "bucket/SearcahbleBucketListSnapshot.h"
+#include "bucket/SearchableBucketListSnapshot.h"
 #include "crypto/Hex.h"
 #include "history/HistoryManager.h"
 #include "historywork/VerifyBucketWork.h"
@@ -180,7 +180,7 @@ std::string const&
 BucketManagerImpl::getTmpDir()
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     if (!mWorkDir)
     {
         TmpDir t = mTmpDirManager->tmpDir("bucket");
@@ -269,7 +269,7 @@ BucketManagerImpl::getMergeTimer()
 MergeCounters
 BucketManagerImpl::readMergeCounters()
 {
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     return mMergeCounters;
 }
 
@@ -359,7 +359,7 @@ MergeCounters::operator==(MergeCounters const& other) const
 void
 BucketManagerImpl::incrMergeCounters(MergeCounters const& delta)
 {
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     mMergeCounters += delta;
 }
 
@@ -385,7 +385,7 @@ BucketManagerImpl::adoptFileAsBucket(std::string const& filename,
 {
     ZoneScoped;
     releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
 
     if (mergeKey)
     {
@@ -469,7 +469,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(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     CLOG_TRACE(Bucket, "BucketManager::noteEmptyMergeOutput({})", mergeKey);
     mLiveFutures.erase(mergeKey);
 }
@@ -478,7 +478,7 @@ std::shared_ptr<Bucket>
 BucketManagerImpl::getBucketIfExists(uint256 const& hash)
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     auto i = mSharedBuckets.find(hash);
     if (i != mSharedBuckets.end())
     {
@@ -495,7 +495,7 @@ std::shared_ptr<Bucket>
 BucketManagerImpl::getBucketByHash(uint256 const& hash)
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     if (isZero(hash))
     {
         return std::make_shared<Bucket>();
@@ -528,7 +528,7 @@ std::shared_future<std::shared_ptr<Bucket>>
 BucketManagerImpl::getMergeFuture(MergeKey const& key)
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     MergeCounters mc;
     auto i = mLiveFutures.find(key);
     if (i == mLiveFutures.end())
@@ -574,7 +574,7 @@ BucketManagerImpl::putMergeFuture(
 {
     ZoneScoped;
     releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     CLOG_TRACE(
         Bucket,
         "BucketManager::putMergeFuture storing future for running merge {}",
@@ -586,7 +586,7 @@ BucketManagerImpl::putMergeFuture(
 void
 BucketManagerImpl::clearMergeFuturesForTesting()
 {
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     mLiveFutures.clear();
 }
 #endif
@@ -685,7 +685,7 @@ BucketManagerImpl::cleanupStaleFiles()
         return;
     }
 
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     auto referenced = getAllReferencedBuckets();
     std::transform(std::begin(mSharedBuckets), std::end(mSharedBuckets),
                    std::inserter(referenced, std::end(referenced)),
@@ -715,7 +715,7 @@ void
 BucketManagerImpl::forgetUnreferencedBuckets()
 {
     ZoneScoped;
-    std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
+    std::lock_guard<std::recursive_mutex> lock(mBucketFileMutex);
     auto referenced = getAllReferencedBuckets();
     auto blReferenced = getBucketListReferencedBuckets();
 
@@ -822,8 +822,11 @@ BucketManagerImpl::addBatch(Application& app, uint32_t currLedger,
     auto timer = mBucketAddBatch.TimeScope();
     mBucketObjectInsertBatch.Mark(initEntries.size() + liveEntries.size() +
                                   deadEntries.size());
-    mBucketList->addBatch(app, currLedger, currLedgerProtocol, initEntries,
-                          liveEntries, deadEntries);
+    {
+        std::lock_guard<std::recursive_mutex> lock(mBucketSnapshotMutex);
+        mBucketList->addBatch(app, currLedger, currLedgerProtocol, initEntries,
+                              liveEntries, deadEntries);
+    }
     mBucketListSizeCounter.set_count(mBucketList->getSize());
 }
 
@@ -905,11 +908,15 @@ BucketManagerImpl::scanForEvictionLegacySQL(AbstractLedgerTxn& ltx,
 }
 
 std::unique_ptr<SearchableBucketListSnapshot const>
-BucketManagerImpl::getSearchableBucketListSnapshot(bool isMainThread) const
+BucketManagerImpl::getSearchableBucketListSnapshot() const
 {
-    // TODO: Lock this
     releaseAssertOrThrow(mApp.getConfig().isUsingBucketListDB() && mBucketList);
-    return std::make_unique<SearchableBucketListSnapshot>(mApp, *mBucketList);
+
+    std::lock_guard<std::recursive_mutex> lock(mBucketSnapshotMutex);
+
+    // Note: cannot use make_unique due to private constructor
+    return std::unique_ptr<SearchableBucketListSnapshot>(
+        new SearchableBucketListSnapshot(mApp, *mBucketList));
 }
 
 medida::Meter&
@@ -972,47 +979,55 @@ BucketManagerImpl::assumeState(HistoryArchiveState const& has,
     ZoneScoped;
     releaseAssertOrThrow(mApp.getConfig().MODE_ENABLES_BUCKETLIST);
 
-    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))
+        std::lock_guard<std::recursive_mutex> lock(mBucketSnapshotMutex);
+        for (uint32_t i = 0; i < BucketList::kNumLevels; ++i)
         {
-            throw std::runtime_error(
-                "Missing bucket files while assuming saved BucketList state");
-        }
+            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");
+            }
 
-        auto const& nextFuture = has.currentBuckets.at(i).next;
-        std::shared_ptr<Bucket> nextBucket = nullptr;
-        if (nextFuture.hasOutputHash())
-        {
-            nextBucket =
-                getBucketByHash(hexToBin256(nextFuture.getOutputHash()));
-            if (!nextBucket)
+            auto const& nextFuture = has.currentBuckets.at(i).next;
+            std::shared_ptr<Bucket> nextBucket = nullptr;
+            if (nextFuture.hasOutputHash())
             {
-                throw std::runtime_error("Missing future bucket files while "
-                                         "assuming saved BucketList state");
+                nextBucket =
+                    getBucketByHash(hexToBin256(nextFuture.getOutputHash()));
+                if (!nextBucket)
+                {
+                    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())
-        {
-            releaseAssert(curr->isEmpty() || curr->isIndexed());
-            releaseAssert(snap->isEmpty() || snap->isIndexed());
-            if (nextBucket)
+            // Buckets on the BucketList should always be indexed when
+            // BucketListDB enabled
+            if (mApp.getConfig().isUsingBucketListDB())
             {
-                releaseAssert(nextBucket->isEmpty() || nextBucket->isIndexed());
+                releaseAssert(curr->isEmpty() || curr->isIndexed());
+                releaseAssert(snap->isEmpty() || snap->isIndexed());
+                if (nextBucket)
+                {
+                    releaseAssert(nextBucket->isEmpty() ||
+                                  nextBucket->isIndexed());
+                }
             }
+
+            mBucketList->getLevel(i).setCurr(curr);
+            mBucketList->getLevel(i).setSnap(snap);
+            mBucketList->getLevel(i).setNext(nextFuture);
         }
 
-        mBucketList->getLevel(i).setCurr(curr);
-        mBucketList->getLevel(i).setSnap(snap);
-        mBucketList->getLevel(i).setNext(nextFuture);
+        mBucketList->restartMerges(mApp, maxProtocolVersion, has.currentLedger);
     }
 
-    mBucketList->restartMerges(mApp, maxProtocolVersion, has.currentLedger);
     cleanupStaleFiles();
 }
 

src/bucket/BucketManagerImpl.h

@@ -41,7 +41,6 @@ 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;
@@ -53,6 +52,16 @@ 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;
+
+    // Lock for logical BucketList changes and snapshots (i.e. addBatch,
+    // getSearchableSnapshot). This lock is not required for raw Bucket file
+    // management.
+    mutable std::recursive_mutex mBucketSnapshotMutex;
+
     bool const mDeleteEntireBucketDirInDtor;
 
     // Records bucket-merges that are currently _live_ in some FutureBucket, in
@@ -135,7 +144,7 @@ class BucketManagerImpl : public BucketManager
                                   uint32_t ledgerSeq) override;
 
     std::unique_ptr<SearchableBucketListSnapshot const>
-    getSearchableBucketListSnapshot(bool isMainThread) const override;
+    getSearchableBucketListSnapshot() const override;
 
     medida::Meter& getBloomMissMeter() const override;
     medida::Meter& getBloomLookupMeter() const override;

src/bucket/SearchableBucketListSnapshot.cpp

@@ -48,6 +48,19 @@ SearchableBucketListSnapshot::getPointLoadTimer(LedgerEntryType t) const
     return iter->second;
 }
 
+bool
+SearchableBucketListSnapshot::isWithinAllowedLedgerDrift(
+    uint32_t allowedLedgerDrift) const
+{
+    auto currLCL = mApp.getLedgerManager().getLastClosedLedgerNum();
+
+    // Edge case: genesis ledger
+    auto minimumLCL =
+        allowedLedgerDrift > currLCL ? 0 : currLCL - allowedLedgerDrift;
+
+    return mLCL >= minimumLCL;
+}
+
 void
 SearchableBucketListSnapshot::loopAllBuckets(
     std::function<bool(SearchableBucketSnapshot const&)> f) const
@@ -77,6 +90,9 @@ SearchableBucketListSnapshot::getLedgerEntry(LedgerKey const& k) const
     ZoneScoped;
     auto timer = getPointLoadTimer(k.type()).TimeScope();
 
+    // Snapshots not currently supported, all access must be up to date
+    releaseAssert(isWithinAllowedLedgerDrift(0));
+
     std::shared_ptr<LedgerEntry> result{};
 
     auto f = [&](SearchableBucketSnapshot const& b) {
@@ -106,6 +122,9 @@ SearchableBucketListSnapshot::loadKeys(
     ZoneScoped;
     auto timer = recordBulkLoadMetrics("prefetch", inKeys.size()).TimeScope();
 
+    // Snapshots not currently supported, all access must be up to date
+    releaseAssert(isWithinAllowedLedgerDrift(0));
+
     std::vector<LedgerEntry> entries;
 
     // Make a copy of the key set, this loop is destructive
@@ -126,6 +145,9 @@ SearchableBucketListSnapshot::loadPoolShareTrustLinesByAccountAndAsset(
     ZoneScoped;
     auto timer = recordBulkLoadMetrics("poolshareTrustlines", 0).TimeScope();
 
+    // Snapshots not currently supported, all access must be up to date
+    releaseAssert(isWithinAllowedLedgerDrift(0));
+
     UnorderedMap<LedgerKey, LedgerEntry> liquidityPoolToTrustline;
     UnorderedSet<LedgerKey> deadTrustlines;
     LedgerKeySet liquidityPoolKeysToSearch;
@@ -172,6 +194,9 @@ SearchableBucketListSnapshot::loadInflationWinners(size_t maxWinners,
     ZoneScoped;
     auto timer = recordBulkLoadMetrics("inflationWinners", 0).TimeScope();
 
+    // Snapshots not currently supported, all access must be up to date
+    releaseAssert(isWithinAllowedLedgerDrift(0));
+
     UnorderedMap<AccountID, int64_t> voteCount;
     UnorderedSet<AccountID> seen;
 
@@ -257,6 +282,8 @@ SearchableBucketLevelSnapshot::SearchableBucketLevelSnapshot(
 {
 }
 
+// This is not thread safe, must call while holding
+// BucketManager::mBucketSnapshotMutex.
 SearchableBucketListSnapshot::SearchableBucketListSnapshot(Application& app,
                                                            BucketList const& bl)
     : mApp(app)

src/bucket/SearchableBucketListSnapshot.h

@@ -5,6 +5,7 @@
 // of this distribution or at http://www.apache.org/licenses/LICENSE-2.0
 
 #include "bucket/BucketList.h"
+#include "bucket/BucketManagerImpl.h"
 #include "bucket/SearchableBucketSnapshot.h"
 
 namespace stellar
@@ -48,11 +49,13 @@ class SearchableBucketListSnapshot : public NonMovable
 
     medida::Timer& getPointLoadTimer(LedgerEntryType t) const;
 
-  public:
-    // TODO: Private constructor so only BucketManager can create this class
-    // with a mutex check
+    // Return true is this snapshot is from a ledger within [lcl -
+    // allowedLedgerDrift, lcl]
+    bool isWithinAllowedLedgerDrift(uint32_t allowedLedgerDrift) const;
+
     SearchableBucketListSnapshot(Application& app, BucketList const& bl);
 
+  public:
     std::vector<LedgerEntry>
     loadKeys(std::set<LedgerKey, LedgerEntryIdCmp> const& inKeys) const;
 
@@ -64,5 +67,8 @@ class SearchableBucketListSnapshot : public NonMovable
                                                       int64_t minBalance) const;
 
     std::shared_ptr<LedgerEntry> getLedgerEntry(LedgerKey const& k) const;
+
+    friend std::unique_ptr<SearchableBucketListSnapshot const>
+    BucketManagerImpl::getSearchableBucketListSnapshot() const;
 };
 }

src/bucket/SearchableBucketSnapshot.h

@@ -36,12 +36,15 @@ class SearchableBucketSnapshot : public NonMovable
                                                 std::streamoff pos,
                                                 size_t pageSize) const;
 
-  public:
+    // Warning: constructor not thread safe
     SearchableBucketSnapshot(std::shared_ptr<Bucket const> const b);
+
+    // Only allow copy constructor, is threadsafe
     SearchableBucketSnapshot(SearchableBucketSnapshot const& b);
     SearchableBucketSnapshot&
     operator=(SearchableBucketSnapshot const& b) = delete;
 
+  public:
     bool isEmpty() const;
     std::shared_ptr<Bucket const> getRawBucket() const;
 
@@ -63,5 +66,7 @@ class SearchableBucketSnapshot : public NonMovable
         AccountID const& accountID, UnorderedSet<LedgerKey>& deadTrustlines,
         UnorderedMap<LedgerKey, LedgerEntry>& liquidityPoolKeyToTrustline,
         LedgerKeySet& liquidityPoolKeys) const;
+
+    friend struct SearchableBucketLevelSnapshot;
 };
 }

src/ledger/LedgerManagerImpl.cpp

@@ -427,12 +427,14 @@ LedgerManagerImpl::getDatabase()
 uint32_t
 LedgerManagerImpl::getLastMaxTxSetSize() const
 {
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     return mLastClosedLedger.header.maxTxSetSize;
 }
 
 uint32_t
 LedgerManagerImpl::getLastMaxTxSetSizeOps() const
 {
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     auto n = mLastClosedLedger.header.maxTxSetSize;
     return protocolVersionStartsFrom(mLastClosedLedger.header.ledgerVersion,
                                      ProtocolVersion::V_11)
@@ -484,6 +486,7 @@ LedgerManagerImpl::maxSorobanTransactionResources()
 int64_t
 LedgerManagerImpl::getLastMinBalance(uint32_t ownerCount) const
 {
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     auto const& lh = mLastClosedLedger.header;
     if (protocolVersionIsBefore(lh.ledgerVersion, ProtocolVersion::V_9))
         return (2 + ownerCount) * lh.baseReserve;
@@ -494,18 +497,21 @@ LedgerManagerImpl::getLastMinBalance(uint32_t ownerCount) const
 uint32_t
 LedgerManagerImpl::getLastReserve() const
 {
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     return mLastClosedLedger.header.baseReserve;
 }
 
 uint32_t
 LedgerManagerImpl::getLastTxFee() const
 {
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     return mLastClosedLedger.header.baseFee;
 }
 
 LedgerHeaderHistoryEntry const&
 LedgerManagerImpl::getLastClosedLedgerHeader() const
 {
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     return mLastClosedLedger;
 }
 
@@ -524,12 +530,14 @@ LedgerManagerImpl::getLastClosedLedgerHAS()
 uint32_t
 LedgerManagerImpl::getLastClosedLedgerNum() const
 {
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     return mLastClosedLedger.header.ledgerSeq;
 }
 
 SorobanNetworkConfig&
 LedgerManagerImpl::getSorobanNetworkConfigInternal()
 {
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     releaseAssert(mSorobanNetworkConfig);
     return *mSorobanNetworkConfig;
 }
@@ -1275,6 +1283,7 @@ LedgerManagerImpl::advanceLedgerPointers(LedgerHeader const& header,
                    ledgerAbbrev(header, ledgerHash));
     }
 
+    std::lock_guard<std::mutex> lock(mLedgerPointerLock);
     mLastClosedLedger.hash = ledgerHash;
     mLastClosedLedger.header = header;
 }

src/ledger/LedgerManagerImpl.h

@@ -51,6 +51,8 @@ class LedgerManagerImpl : public LedgerManager
     LedgerHeaderHistoryEntry mLastClosedLedger;
     std::optional<SorobanNetworkConfig> mSorobanNetworkConfig;
 
+    mutable std::mutex mLedgerPointerLock;
+
     medida::Timer& mTransactionApply;
     medida::Histogram& mTransactionCount;
     medida::Histogram& mOperationCount;