Switch BucketManager to AppConnector (#4975)

Author: marta-lokhova

src/bucket/BucketManager.cpp

@@ -52,7 +52,7 @@ namespace stellar
 {
 
 std::unique_ptr<BucketManager>
-BucketManager::create(Application& app)
+BucketManager::create(AppConnector& app)
 {
     auto bucketManagerPtr =
         std::unique_ptr<BucketManager>(new BucketManager(app));
@@ -99,7 +99,7 @@ BucketManager::initialize()
     mLiveBucketList = std::make_unique<LiveBucketList>();
     mHotArchiveBucketList = std::make_unique<HotArchiveBucketList>();
     mSnapshotManager = std::make_unique<BucketSnapshotManager>(
-        mApp,
+        mAppConnector,
         std::make_unique<BucketListSnapshot<LiveBucket>>(*mLiveBucketList,
                                                          LedgerHeader()),
         std::make_unique<BucketListSnapshot<HotArchiveBucket>>(
@@ -131,40 +131,41 @@ BucketManager::getTmpDirManager()
     return *mTmpDirManager;
 }
 
-BucketManager::BucketManager(Application& app)
-    : mApp(app)
+BucketManager::BucketManager(AppConnector& appConnector)
+    : mAppConnector(appConnector)
     , mLiveBucketList(nullptr)
     , mHotArchiveBucketList(nullptr)
     , mSnapshotManager(nullptr)
     , mTmpDirManager(nullptr)
     , mWorkDir(nullptr)
     , mLockedBucketDir(nullptr)
-    , mBucketLiveObjectInsertBatch(app.getMetrics().NewMeter(
+    , mBucketLiveObjectInsertBatch(appConnector.getMetrics().NewMeter(
           {"bucket", "batch", "objectsadded"}, "object"))
-    , mBucketArchiveObjectInsertBatch(app.getMetrics().NewMeter(
+    , mBucketArchiveObjectInsertBatch(appConnector.getMetrics().NewMeter(
           {"bucket", "batch-archive", "objectsadded"}, "object"))
     , mBucketAddLiveBatch(
-          app.getMetrics().NewTimer({"bucket", "batch", "addtime"}))
-    , mBucketAddArchiveBatch(
-          app.getMetrics().NewTimer({"bucket", "batch-archive", "addtime"}))
-    , mBucketSnapMerge(app.getMetrics().NewTimer({"bucket", "snap", "merge"}))
+          appConnector.getMetrics().NewTimer({"bucket", "batch", "addtime"}))
+    , mBucketAddArchiveBatch(appConnector.getMetrics().NewTimer(
+          {"bucket", "batch-archive", "addtime"}))
+    , mBucketSnapMerge(
+          appConnector.getMetrics().NewTimer({"bucket", "snap", "merge"}))
     , mSharedBucketsSize(
-          app.getMetrics().NewCounter({"bucket", "memory", "shared"}))
+          appConnector.getMetrics().NewCounter({"bucket", "memory", "shared"}))
     , mLiveBucketListSizeCounter(
-          app.getMetrics().NewCounter({"bucketlist", "size", "bytes"}))
-    , mArchiveBucketListSizeCounter(
-          app.getMetrics().NewCounter({"bucketlist-archive", "size", "bytes"}))
-    , mCacheHitMeter(app.getMetrics().NewMeter({"bucketlistDB", "cache", "hit"},
-                                               "bucketlistDB"))
-    , mCacheMissMeter(app.getMetrics().NewMeter(
+          appConnector.getMetrics().NewCounter({"bucketlist", "size", "bytes"}))
+    , mArchiveBucketListSizeCounter(appConnector.getMetrics().NewCounter(
+          {"bucketlist-archive", "size", "bytes"}))
+    , mCacheHitMeter(appConnector.getMetrics().NewMeter(
+          {"bucketlistDB", "cache", "hit"}, "bucketlistDB"))
+    , mCacheMissMeter(appConnector.getMetrics().NewMeter(
           {"bucketlistDB", "cache", "miss"}, "bucketlistDB"))
-    , mLiveBucketIndexCacheEntries(
-          app.getMetrics().NewCounter({"bucketlistDB", "cache", "entries"}))
-    , mLiveBucketIndexCacheBytes(
-          app.getMetrics().NewCounter({"bucketlistDB", "cache", "bytes"}))
-    , mBucketListEvictionCounters(app)
+    , mLiveBucketIndexCacheEntries(appConnector.getMetrics().NewCounter(
+          {"bucketlistDB", "cache", "entries"}))
+    , mLiveBucketIndexCacheBytes(appConnector.getMetrics().NewCounter(
+          {"bucketlistDB", "cache", "bytes"}))
+    , mBucketListEvictionCounters(appConnector)
     , mEvictionStatistics(std::make_shared<EvictionStatistics>())
-    , mConfig(app.getConfig())
+    , mConfig(appConnector.getConfig())
 {
     for (uint32_t t =
              static_cast<uint32_t>(LedgerEntryTypeAndDurability::ACCOUNT);
@@ -174,10 +175,10 @@ BucketManager::BucketManager(Application& app)
         auto type = static_cast<LedgerEntryTypeAndDurability>(t);
         auto typeString = toString(type);
         mBucketListEntryCountCounters.emplace(
-            type, app.getMetrics().NewCounter(
+            type, appConnector.getMetrics().NewCounter(
                       {"bucketlist", "entryCounts", typeString}));
         mBucketListEntrySizeCounters.emplace(
-            type, app.getMetrics().NewCounter(
+            type, appConnector.getMetrics().NewCounter(
                       {"bucketlist", "entrySizes", typeString}));
     }
 }
@@ -334,7 +335,7 @@ medida::Meter&
 BucketManager::getBloomMissMeter() const
 {
     BUCKET_TYPE_ASSERT(BucketT);
-    return mApp.getMetrics().NewMeter(
+    return mAppConnector.getMetrics().NewMeter(
         {BucketT::METRIC_STRING, "bloom", "misses"}, "bloom");
 }
 
@@ -343,7 +344,7 @@ medida::Meter&
 BucketManager::getBloomLookupMeter() const
 {
     BUCKET_TYPE_ASSERT(BucketT);
-    return mApp.getMetrics().NewMeter(
+    return mAppConnector.getMetrics().NewMeter(
         {BucketT::METRIC_STRING, "bloom", "lookups"}, "bloom");
 }
 
@@ -843,8 +844,8 @@ BucketManager::getAllReferencedBuckets(HistoryArchiveState const& has) const
     }
 
     // retain buckets that are referenced by a state in the publish queue.
-    for (auto const& h :
-         HistoryManager::getBucketsReferencedByPublishQueue(mApp.getConfig()))
+    for (auto const& h : HistoryManager::getBucketsReferencedByPublishQueue(
+             mAppConnector.getConfig()))
     {
         auto rhash = hexToBin256(h);
         auto rit = referenced.emplace(rhash);
@@ -1158,7 +1159,7 @@ BucketManager::startBackgroundEvictionScan(uint32_t ledgerSeq,
         });
 
     mEvictionFuture = task->get_future();
-    mApp.postOnEvictionBackgroundThread(
+    mAppConnector.postOnEvictionBackgroundThread(
         bind(&task_t::operator(), task),
         "SearchableLiveBucketListSnapshot: eviction scan");
 }
@@ -1295,7 +1296,7 @@ BucketManager::checkForMissingBucketsFiles(HistoryArchiveState const& has)
 }
 
 void
-BucketManager::assumeState(HistoryArchiveState const& has,
+BucketManager::assumeState(Application& app, HistoryArchiveState const& has,
                            uint32_t maxProtocolVersion, bool restartMerges)
 {
     ZoneScoped;
@@ -1355,11 +1356,11 @@ BucketManager::assumeState(HistoryArchiveState const& has,
 
     if (restartMerges)
     {
-        mLiveBucketList->restartMerges(mApp, maxProtocolVersion,
+        mLiveBucketList->restartMerges(app, maxProtocolVersion,
                                        has.currentLedger);
         if (has.hasHotArchiveBuckets())
         {
-            mHotArchiveBucketList->restartMerges(mApp, maxProtocolVersion,
+            mHotArchiveBucketList->restartMerges(app, maxProtocolVersion,
                                                  has.currentLedger);
         }
     }
@@ -1462,14 +1463,14 @@ BucketManager::loadCompleteLedgerState(HistoryArchiveState const& has)
 }
 
 std::shared_ptr<LiveBucket>
-BucketManager::mergeBuckets(HistoryArchiveState const& has)
+BucketManager::mergeBuckets(asio::io_context& ctx,
+                            HistoryArchiveState const& has)
 {
     ZoneScoped;
     releaseAssert(threadIsMain());
     std::map<LedgerKey, LedgerEntry> ledgerMap = loadCompleteLedgerState(has);
     BucketMetadata meta;
     MergeCounters mc;
-    auto& ctx = mApp.getClock().getIOContext();
     meta.ledgerVersion = mConfig.LEDGER_PROTOCOL_VERSION;
     LiveBucketOutputIterator out(getTmpDir(), /*keepTombstoneEntries=*/false,
                                  meta, mc, ctx, /*doFsync=*/true);
@@ -1656,7 +1657,7 @@ BucketManager::visitLedgerEntries(
 
 std::shared_ptr<BasicWork>
 BucketManager::scheduleVerifyReferencedBucketsWork(
-    HistoryArchiveState const& has)
+    Application& app, HistoryArchiveState const& has)
 {
     releaseAssert(threadIsMain());
     std::set<Hash> hashes = getAllReferencedBuckets(has);
@@ -1699,7 +1700,7 @@ BucketManager::scheduleVerifyReferencedBucketsWork(
 
             seq.emplace_back(
                 std::make_shared<VerifyBucketWork<HotArchiveBucket>>(
-                    mApp, filename, hash, indexIter->second, nullptr));
+                    app, filename, hash, indexIter->second, nullptr));
         }
         else
         {
@@ -1708,10 +1709,10 @@ BucketManager::scheduleVerifyReferencedBucketsWork(
             auto [indexIter, _] = liveIndexMap.emplace(i++, nullptr);
 
             seq.emplace_back(std::make_shared<VerifyBucketWork<LiveBucket>>(
-                mApp, filename, hash, indexIter->second, nullptr));
+                app, filename, hash, indexIter->second, nullptr));
         }
     }
-    return mApp.getWorkScheduler().scheduleWork<WorkSequence>(
+    return app.getWorkScheduler().scheduleWork<WorkSequence>(
         "verify-referenced-buckets", seq);
 }
 
@@ -1734,7 +1735,7 @@ BucketManager::reportBucketEntryCountMetrics()
             countCounter =
                 mBucketListEntryCountCounters
                     .emplace(type,
-                             mApp.getMetrics().NewCounter(
+                             mAppConnector.getMetrics().NewCounter(
                                  {"bucketlist", "entryCounts", typeString}))
                     .first;
         }
@@ -1747,7 +1748,7 @@ BucketManager::reportBucketEntryCountMetrics()
             sizeCounter =
                 mBucketListEntrySizeCounters
                     .emplace(type,
-                             mApp.getMetrics().NewCounter(
+                             mAppConnector.getMetrics().NewCounter(
                                  {"bucketlist", "entrySizes", typeString}))
                     .first;
         }

src/bucket/BucketManager.h

@@ -32,7 +32,7 @@ namespace stellar
 
 class TmpDir;
 class AbstractLedgerTxn;
-class Application;
+class AppConnector;
 class Bucket;
 class LiveBucketList;
 class HotArchiveBucketList;
@@ -74,12 +74,10 @@ 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.
-    Application& mApp;
+    // BucketManager uses AppConnector for thread-safe access to Application
+    // services. AppConnector methods are either explicitly main-thread only
+    // (with releaseAssert) or documented as thread-safe.
+    AppConnector& mAppConnector;
     std::unique_ptr<LiveBucketList> mLiveBucketList;
     std::unique_ptr<HotArchiveBucketList> mHotArchiveBucketList;
     std::unique_ptr<BucketSnapshotManager> mSnapshotManager;
@@ -181,13 +179,13 @@ class BucketManager : NonMovableOrCopyable
 #endif
 
   protected:
-    BucketManager(Application& app);
+    BucketManager(AppConnector& appConnector);
     void calculateSkipValues(LedgerHeader& currentHeader);
     std::string bucketFilename(std::string const& bucketHexHash);
     std::string bucketFilename(Hash const& hash);
 
   public:
-    static std::unique_ptr<BucketManager> create(Application& app);
+    static std::unique_ptr<BucketManager> create(AppConnector& app);
     virtual ~BucketManager();
 
     void initialize();
@@ -361,7 +359,7 @@ class BucketManager : NonMovableOrCopyable
 
     // Assume state from `has` in BucketList: find and attach all buckets in
     // `has`, set current BL.
-    void assumeState(HistoryArchiveState const& has,
+    void assumeState(Application& app, HistoryArchiveState const& has,
                      uint32_t maxProtocolVersion, bool restartMerges);
 
     void shutdown();
@@ -382,7 +380,8 @@ class BucketManager : NonMovableOrCopyable
 
     // Merge the bucket list of the provided HAS into a single "super bucket"
     // consisting of only live entries, and return it.
-    std::shared_ptr<LiveBucket> mergeBuckets(HistoryArchiveState const& has);
+    std::shared_ptr<LiveBucket> mergeBuckets(asio::io_context& ctx,
+                                             HistoryArchiveState const& has);
 
     // Visits all the active ledger entries or subset thereof.
     //
@@ -410,7 +409,8 @@ class BucketManager : NonMovableOrCopyable
     // Schedule a Work class that verifies the hashes of all referenced buckets
     // on background threads.
     std::shared_ptr<BasicWork>
-    scheduleVerifyReferencedBucketsWork(HistoryArchiveState const& has);
+    scheduleVerifyReferencedBucketsWork(Application& app,
+                                        HistoryArchiveState const& has);
 
     Config const& getConfig() const;
     void reportBucketEntryCountMetrics();

src/bucket/BucketSnapshotManager.cpp

@@ -37,7 +37,7 @@ copyHistoricalSnapshots(
 }
 
 BucketSnapshotManager::BucketSnapshotManager(
-    Application& app, SnapshotPtrT<LiveBucket>&& snapshot,
+    AppConnector& app, SnapshotPtrT<LiveBucket>&& snapshot,
     SnapshotPtrT<HotArchiveBucket>&& hotArchiveSnapshot,
     uint32_t numLiveHistoricalSnapshots)
     : mAppConnector(app)

src/bucket/BucketSnapshotManager.h

@@ -25,7 +25,6 @@ class Timer;
 namespace stellar
 {
 
-class Application;
 class LiveBucketList;
 template <class BucketT> class BucketListSnapshot;
 class SearchableLiveBucketListSnapshot;
@@ -75,7 +74,8 @@ class BucketSnapshotManager : NonMovableOrCopyable
     // numHistoricalLedgers is the number of historical snapshots that the
     // snapshot manager will maintain. If numHistoricalLedgers is 5, snapshots
     // will be capable of querying state from ledger [lcl, lcl - 5].
-    BucketSnapshotManager(Application& app, SnapshotPtrT<LiveBucket>&& snapshot,
+    BucketSnapshotManager(AppConnector& app,
+                          SnapshotPtrT<LiveBucket>&& snapshot,
                           SnapshotPtrT<HotArchiveBucket>&& hotArchiveSnapshot,
                           uint32_t numHistoricalLedgers);
 

src/bucket/BucketUtils.cpp

@@ -6,6 +6,7 @@
 #include "bucket/HotArchiveBucket.h"
 #include "bucket/LiveBucket.h"
 #include "ledger/LedgerTypeUtils.h"
+#include "main/AppConnector.h"
 #include "main/Application.h"
 #include "util/types.h"
 #include "xdr/Stellar-ledger-entries.h"
@@ -167,7 +168,7 @@ EvictionResultCandidates::isValid(uint32_t currLedgerSeq,
                currSas.startingEvictionScanLevel;
 }
 
-EvictionCounters::EvictionCounters(Application& app)
+EvictionCounters::EvictionCounters(AppConnector& app)
     : entriesEvicted(app.getMetrics().NewCounter(
           {"state-archival", "eviction", "entries-evicted"}))
     , bytesScannedForEviction(app.getMetrics().NewCounter(

src/bucket/BucketUtils.h

@@ -23,6 +23,7 @@ namespace stellar
 {
 
 class Application;
+class AppConnector;
 class LiveBucket;
 class HotArchiveBucket;
 template <class BucketT> class BucketListSnapshot;
@@ -156,7 +157,7 @@ struct EvictionCounters
     medida::Counter& evictionCyclePeriod;
     medida::Counter& averageEvictedEntryAge;
 
-    EvictionCounters(Application& app);
+    EvictionCounters(AppConnector& app);
 };
 
 class EvictionStatistics

src/bucket/test/BucketManagerTests.cpp

@@ -104,7 +104,8 @@ TEST_CASE("skip list", "[bucket][bucketmanager]")
     class BucketManagerTest : public BucketManager
     {
       public:
-        BucketManagerTest(Application& app) : BucketManager(app)
+        BucketManagerTest(Application& app)
+            : BucketManager(app.getAppConnector())
         {
         }
         void

src/catchup/AssumeStateWork.cpp

@@ -92,7 +92,7 @@ AssumeStateWork::doWork()
                               &liveBuckets = mLiveBuckets,
                               &hotArchiveBuckets =
                                   mHotArchiveBuckets](Application& app) {
-            app.getBucketManager().assumeState(has, maxProtocolVersion,
+            app.getBucketManager().assumeState(app, has, maxProtocolVersion,
                                                restartMerges);
 
             // Drop bucket references once assume state complete since buckets

src/main/AppConnector.cpp

@@ -97,6 +97,20 @@ AppConnector::postOnOverlayThread(std::function<void()>&& f,
     mApp.postOnOverlayThread(std::move(f), message);
 }
 
+void
+AppConnector::postOnBackgroundThread(std::function<void()>&& f,
+                                     std::string const& jobName)
+{
+    mApp.postOnBackgroundThread(std::move(f), jobName);
+}
+
+void
+AppConnector::postOnEvictionBackgroundThread(std::function<void()>&& f,
+                                             std::string const& jobName)
+{
+    mApp.postOnEvictionBackgroundThread(std::move(f), jobName);
+}
+
 Config const&
 AppConnector::getConfig() const
 {