Parallel ledger close implementation

Author: marta-lokhova

src/bucket/BucketListBase.cpp

@@ -9,6 +9,7 @@
 #include "bucket/LiveBucket.h"
 #include "crypto/SHA.h"
 #include "ledger/LedgerTxn.h"
+#include "ledger/NetworkConfig.h"
 #include "main/Application.h"
 #include "util/GlobalChecks.h"
 #include "util/Logging.h"
@@ -57,7 +58,6 @@ template <typename BucketT>
 void
 BucketLevel<BucketT>::setNext(FutureBucket<BucketT> const& fb)
 {
-    releaseAssert(threadIsMain());
     mNextCurr = fb;
 }
 
@@ -79,7 +79,6 @@ template <typename BucketT>
 void
 BucketLevel<BucketT>::setCurr(std::shared_ptr<BucketT> b)
 {
-    releaseAssert(threadIsMain());
     mNextCurr.clear();
     mCurr = b;
 }

src/bucket/BucketListSnapshotBase.cpp

@@ -19,8 +19,6 @@ BucketListSnapshot<BucketT>::BucketListSnapshot(
     BucketListBase<BucketT> const& bl, LedgerHeader header)
     : mHeader(std::move(header))
 {
-    releaseAssert(threadIsMain());
-
     for (uint32_t i = 0; i < BucketListBase<BucketT>::kNumLevels; ++i)
     {
         auto const& level = bl.getLevel(i);

src/bucket/BucketManager.cpp

@@ -62,6 +62,7 @@ void
 BucketManager::initialize()
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
     std::string d = mConfig.BUCKET_DIR_PATH;
 
     if (!fs::exists(d))
@@ -729,7 +730,7 @@ BucketManager::getBucketListReferencedBuckets() const
 }
 
 std::set<Hash>
-BucketManager::getAllReferencedBuckets() const
+BucketManager::getAllReferencedBuckets(HistoryArchiveState const& has) const
 {
     ZoneScoped;
     auto referenced = getBucketListReferencedBuckets();
@@ -740,8 +741,7 @@ BucketManager::getAllReferencedBuckets() const
 
     // retain any bucket referenced by the last closed ledger as recorded in the
     // database (as merges complete, the bucket list drifts from that state)
-    auto lclHas = mApp.getLedgerManager().getLastClosedLedgerHAS();
-    auto lclBuckets = lclHas.allBuckets();
+    auto lclBuckets = has.allBuckets();
     for (auto const& h : lclBuckets)
     {
         auto rit = referenced.emplace(hexToBin256(h));
@@ -752,39 +752,38 @@ BucketManager::getAllReferencedBuckets() const
     }
 
     // retain buckets that are referenced by a state in the publish queue.
-    auto pub = mApp.getHistoryManager().getBucketsReferencedByPublishQueue();
+    for (auto const& h :
+         HistoryManager::getBucketsReferencedByPublishQueue(mApp.getConfig()))
     {
-        for (auto const& h : pub)
+        auto rhash = hexToBin256(h);
+        auto rit = referenced.emplace(rhash);
+        if (rit.second)
         {
-            auto rhash = hexToBin256(h);
-            auto rit = referenced.emplace(rhash);
-            if (rit.second)
-            {
-                CLOG_TRACE(Bucket, "{} referenced by publish queue", h);
-
-                // Project referenced bucket `rhash` -- which might be a merge
-                // input captured before a merge finished -- through our weak
-                // map of merge input/output relationships, to find any outputs
-                // we'll want to retain in order to resynthesize the merge in
-                // the future, rather than re-run it.
-                mFinishedMerges.getOutputsUsingInput(rhash, referenced);
-            }
+            CLOG_TRACE(Bucket, "{} referenced by publish queue", h);
+
+            // Project referenced bucket `rhash` -- which might be a merge
+            // input captured before a merge finished -- through our weak
+            // map of merge input/output relationships, to find any outputs
+            // we'll want to retain in order to resynthesize the merge in
+            // the future, rather than re-run it.
+            mFinishedMerges.getOutputsUsingInput(rhash, referenced);
         }
     }
     return referenced;
 }
 
 void
-BucketManager::cleanupStaleFiles()
+BucketManager::cleanupStaleFiles(HistoryArchiveState const& has)
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
     if (mConfig.DISABLE_BUCKET_GC)
     {
         return;
     }
 
     std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
-    auto referenced = getAllReferencedBuckets();
+    auto referenced = getAllReferencedBuckets(has);
     std::transform(std::begin(mSharedLiveBuckets), std::end(mSharedLiveBuckets),
                    std::inserter(referenced, std::end(referenced)),
                    [](std::pair<Hash, std::shared_ptr<LiveBucket>> const& p) {
@@ -818,11 +817,11 @@ BucketManager::cleanupStaleFiles()
 }
 
 void
-BucketManager::forgetUnreferencedBuckets()
+BucketManager::forgetUnreferencedBuckets(HistoryArchiveState const& has)
 {
     ZoneScoped;
     std::lock_guard<std::recursive_mutex> lock(mBucketMutex);
-    auto referenced = getAllReferencedBuckets();
+    auto referenced = getAllReferencedBuckets(has);
     auto blReferenced = getBucketListReferencedBuckets();
 
     auto bucketMapLoop = [&](auto& bucketMap, auto& futureMap) {
@@ -867,7 +866,7 @@ BucketManager::forgetUnreferencedBuckets()
                     Bucket,
                     "BucketManager::forgetUnreferencedBuckets dropping {}",
                     filename);
-                if (!filename.empty() && !mApp.getConfig().DISABLE_BUCKET_GC)
+                if (!filename.empty() && !mConfig.DISABLE_BUCKET_GC)
                 {
                     CLOG_TRACE(Bucket, "removing bucket file: {}", filename);
                     std::filesystem::remove(filename);
@@ -1076,12 +1075,11 @@ BucketManager::startBackgroundEvictionScan(uint32_t ledgerSeq)
 }
 
 void
-BucketManager::resolveBackgroundEvictionScan(AbstractLedgerTxn& ltx,
-                                             uint32_t ledgerSeq,
-                                             LedgerKeySet const& modifiedKeys)
+BucketManager::resolveBackgroundEvictionScan(
+    AbstractLedgerTxn& ltx, uint32_t ledgerSeq,
+    LedgerKeySet const& modifiedKeys, SorobanNetworkConfig& networkConfig)
 {
     ZoneScoped;
-    releaseAssert(threadIsMain());
     releaseAssert(mEvictionStatistics);
 
     if (!mEvictionFuture.valid())
@@ -1091,9 +1089,6 @@ BucketManager::resolveBackgroundEvictionScan(AbstractLedgerTxn& ltx,
 
     auto evictionCandidates = mEvictionFuture.get();
 
-    auto const& networkConfig =
-        mApp.getLedgerManager().getSorobanNetworkConfig();
-
     // If eviction related settings changed during the ledger, we have to
     // restart the scan
     if (!evictionCandidates.isValid(ledgerSeq,
@@ -1209,6 +1204,7 @@ BucketManager::assumeState(HistoryArchiveState const& has,
                            uint32_t maxProtocolVersion, bool restartMerges)
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
     releaseAssertOrThrow(mConfig.MODE_ENABLES_BUCKETLIST);
 
     // TODO: Assume archival bucket state
@@ -1257,7 +1253,7 @@ BucketManager::assumeState(HistoryArchiveState const& has,
         mLiveBucketList->restartMerges(mApp, maxProtocolVersion,
                                        has.currentLedger);
     }
-    cleanupStaleFiles();
+    cleanupStaleFiles(has);
 }
 
 void
@@ -1358,7 +1354,7 @@ std::shared_ptr<LiveBucket>
 BucketManager::mergeBuckets(HistoryArchiveState const& has)
 {
     ZoneScoped;
-
+    releaseAssert(threadIsMain());
     std::map<LedgerKey, LedgerEntry> ledgerMap = loadCompleteLedgerState(has);
     BucketMetadata meta;
     MergeCounters mc;
@@ -1548,9 +1544,11 @@ BucketManager::visitLedgerEntries(
 }
 
 std::shared_ptr<BasicWork>
-BucketManager::scheduleVerifyReferencedBucketsWork()
+BucketManager::scheduleVerifyReferencedBucketsWork(
+    HistoryArchiveState const& has)
 {
-    std::set<Hash> hashes = getAllReferencedBuckets();
+    releaseAssert(threadIsMain());
+    std::set<Hash> hashes = getAllReferencedBuckets(has);
     std::vector<std::shared_ptr<BasicWork>> seq;
     for (auto const& h : hashes)
     {

src/bucket/BucketManager.h

@@ -70,6 +70,12 @@ class BucketManager : NonMovableOrCopyable
 
     static std::string const kLockFilename;
 
+    // NB: ideally, BucketManager should have no access to mApp, as it's too
+    // dangerous in the context of parallel application. BucketManager is quite
+    // bloated, with lots of legacy code, so to ensure safety, annotate all
+    // functions using mApp with `releaseAssert(threadIsMain())` and avoid
+    // accessing mApp in the background. Safety invariant: lock acquisition must
+    // always be LCL lock -> BucketManger lock, and never the other direction
     Application& mApp;
     std::unique_ptr<LiveBucketList> mLiveBucketList;
     std::unique_ptr<HotArchiveBucketList> mHotArchiveBucketList;
@@ -126,7 +132,7 @@ class BucketManager : NonMovableOrCopyable
 
     std::atomic<bool> mIsShutdown{false};
 
-    void cleanupStaleFiles();
+    void cleanupStaleFiles(HistoryArchiveState const& has);
     void deleteTmpDirAndUnlockBucketDir();
     void deleteEntireBucketDir();
 
@@ -262,7 +268,7 @@ class BucketManager : NonMovableOrCopyable
     // not immediately cause the buckets to delete themselves, if someone else
     // is using them via a shared_ptr<>, but the BucketManager will no longer
     // independently keep them alive.
-    void forgetUnreferencedBuckets();
+    void forgetUnreferencedBuckets(HistoryArchiveState const& has);
 
     // Feed a new batch of entries to the bucket list. This interface expects to
     // be given separate init (created) and live (updated) entry vectors. The
@@ -295,7 +301,8 @@ class BucketManager : NonMovableOrCopyable
     void startBackgroundEvictionScan(uint32_t ledgerSeq);
     void resolveBackgroundEvictionScan(AbstractLedgerTxn& ltx,
                                        uint32_t ledgerSeq,
-                                       LedgerKeySet const& modifiedKeys);
+                                       LedgerKeySet const& modifiedKeys,
+                                       SorobanNetworkConfig& networkConfig);
 
     medida::Meter& getBloomMissMeter() const;
     medida::Meter& getBloomLookupMeter() const;
@@ -320,7 +327,8 @@ class BucketManager : NonMovableOrCopyable
 
     // Return the set of buckets referenced by the BucketList, LCL HAS,
     // and publish queue.
-    std::set<Hash> getAllReferencedBuckets() const;
+    std::set<Hash>
+    getAllReferencedBuckets(HistoryArchiveState const& has) const;
 
     // Check for missing bucket files that would prevent `assumeState` from
     // succeeding
@@ -377,7 +385,8 @@ class BucketManager : NonMovableOrCopyable
 
     // Schedule a Work class that verifies the hashes of all referenced buckets
     // on background threads.
-    std::shared_ptr<BasicWork> scheduleVerifyReferencedBucketsWork();
+    std::shared_ptr<BasicWork>
+    scheduleVerifyReferencedBucketsWork(HistoryArchiveState const& has);
 
     Config const& getConfig() const;
 

src/bucket/BucketSnapshotManager.cpp

@@ -75,7 +75,6 @@ BucketSnapshotManager::recordBulkLoadMetrics(std::string const& label,
 {
     // For now, only keep metrics for the main thread. We can decide on what
     // metrics make sense when more background services are added later.
-    releaseAssert(threadIsMain());
 
     if (numEntries != 0)
     {
@@ -170,8 +169,6 @@ BucketSnapshotManager::updateCurrentSnapshot(
     SnapshotPtrT<LiveBucket>&& liveSnapshot,
     SnapshotPtrT<HotArchiveBucket>&& hotArchiveSnapshot)
 {
-    releaseAssert(threadIsMain());
-
     auto updateSnapshot = [numHistoricalSnapshots = mNumHistoricalSnapshots](
                               auto& currentSnapshot, auto& historicalSnapshots,
                               auto&& newSnapshot) {

src/bucket/LiveBucketList.h

@@ -9,6 +9,9 @@
 
 namespace stellar
 {
+
+class SorobanNetworkConfig;
+
 // The LiveBucketList stores the current canonical state of the ledger. It is
 // made up of LiveBucket buckets, which in turn store individual entries of type
 // BucketEntry. When an entry is "evicted" from the ledger, it is removed from

src/bucket/SearchableBucketList.cpp

@@ -110,7 +110,6 @@ SearchableLiveBucketListSnapshot::loadPoolShareTrustLinesByAccountAndAsset(
     ZoneScoped;
 
     // This query should only be called during TX apply
-    releaseAssert(threadIsMain());
     mSnapshotManager.maybeUpdateSnapshot(mSnapshot, mHistoricalSnapshots);
     releaseAssert(mSnapshot);
 
@@ -157,7 +156,6 @@ SearchableLiveBucketListSnapshot::loadInflationWinners(size_t maxWinners,
 
     // This is a legacy query, should only be called by main thread during
     // catchup
-    releaseAssert(threadIsMain());
     auto timer = mSnapshotManager.recordBulkLoadMetrics("inflationWinners", 0)
                      .TimeScope();
 

src/bucket/test/BucketListTests.cpp

@@ -868,7 +868,8 @@ TEST_CASE_VERSIONS("network config snapshots BucketList size", "[bucketlist]")
     for_versions_from(20, *app, [&] {
         LedgerManagerForBucketTests& lm = app->getLedgerManager();
 
-        auto& networkConfig = app->getLedgerManager().getSorobanNetworkConfig();
+        auto& networkConfig =
+            app->getLedgerManager().getMutableSorobanNetworkConfig();
 
         uint32_t windowSize = networkConfig.stateArchivalSettings()
                                   .bucketListSizeWindowSampleSize;

src/bucket/test/BucketManagerTests.cpp

@@ -237,7 +237,8 @@ TEST_CASE_VERSIONS("bucketmanager ownership", "[bucket][bucketmanager]")
             CHECK(fs::exists(indexFilename));
 
             b.reset();
-            app->getBucketManager().forgetUnreferencedBuckets();
+            app->getBucketManager().forgetUnreferencedBuckets(
+                app->getLedgerManager().getLastClosedLedgerHAS());
             CHECK(!fs::exists(filename));
             CHECK(!fs::exists(indexFilename));
         };
@@ -260,7 +261,8 @@ TEST_CASE_VERSIONS("bucketmanager ownership", "[bucket][bucketmanager]")
 
         // This shouldn't change if we forget unreferenced buckets since
         // it's referenced by bucketlist.
-        app->getBucketManager().forgetUnreferencedBuckets();
+        app->getBucketManager().forgetUnreferencedBuckets(
+            app->getLedgerManager().getLastClosedLedgerHAS());
         CHECK(b1.use_count() == 3);
 
         // But if we mutate the curr bucket of the bucketlist, it should.
@@ -343,7 +345,8 @@ TEST_CASE_VERSIONS("bucketmanager reattach to finished merge",
                 LedgerTestUtils::generateValidLedgerEntriesWithExclusions(
                     {CONFIG_SETTING}, 10),
                 {});
-            bm.forgetUnreferencedBuckets();
+            bm.forgetUnreferencedBuckets(
+                app->getLedgerManager().getLastClosedLedgerHAS());
         } while (!LiveBucketList::levelShouldSpill(ledger, level - 1));
 
         // Check that the merge on level isn't committed (we're in
@@ -433,7 +436,8 @@ TEST_CASE_VERSIONS("bucketmanager reattach to running merge",
                     {CONFIG_SETTING}, 100),
                 {});
 
-            bm.forgetUnreferencedBuckets();
+            bm.forgetUnreferencedBuckets(
+                app->getLedgerManager().getLastClosedLedgerHAS());
 
             HistoryArchiveState has(ledger, bl,
                                     app->getConfig().NETWORK_PASSPHRASE);
@@ -517,8 +521,10 @@ TEST_CASE("bucketmanager do not leak empty-merge futures",
         bl.resolveAnyReadyFutures();
         std::this_thread::sleep_for(std::chrono::seconds(1));
     }
-    bm.forgetUnreferencedBuckets();
-    auto bmRefBuckets = bm.getAllReferencedBuckets();
+    bm.forgetUnreferencedBuckets(
+        app->getLedgerManager().getLastClosedLedgerHAS());
+    auto bmRefBuckets = bm.getAllReferencedBuckets(
+        app->getLedgerManager().getLastClosedLedgerHAS());
     auto bmDirBuckets = bm.getBucketHashesInBucketDirForTesting();
 
     // Remove the 0 bucket in case it's "referenced"; it's never a file.
@@ -574,16 +580,18 @@ TEST_CASE_VERSIONS(
                     {CONFIG_SETTING}, 100),
                 {});
             clock.crank(false);
-            bm.forgetUnreferencedBuckets();
+            bm.forgetUnreferencedBuckets(
+                app->getLedgerManager().getLastClosedLedgerHAS());
         }
         // We should have published nothing and have the first
         // checkpoint still queued.
         REQUIRE(hm.getPublishSuccessCount() == 0);
-        REQUIRE(hm.getMinLedgerQueuedToPublish() == 7);
+        REQUIRE(HistoryManager::getMinLedgerQueuedToPublish(app->getConfig()) ==
+                7);
 
         auto oldReattachments =
             bm.readMergeCounters().mFinishedMergeReattachments;
-        auto HASs = hm.getPublishQueueStates();
+        auto HASs = HistoryManager::getPublishQueueStates(app->getConfig());
         REQUIRE(HASs.size() == 5);
         for (auto& has : HASs)
         {