Populate Soroban in-memory state in a background thread

Author: drebelsky

src/bucket/BucketSnapshot.cpp

@@ -363,7 +363,11 @@ LiveBucketSnapshot::scanForEntriesOfType(
         return Loop::INCOMPLETE;
     }
 
-    auto& stream = getStream();
+    // We open up a stream locally here because otherwise, we might race with
+    // other methods accessing the stream while populating in memory soroban
+    // state
+    XDRInputFileStream stream;
+    stream.open(mBucket->getFilename().string());
     stream.seek(range->first);
 
     BucketEntry be;

src/catchup/ApplyLedgerWork.cpp

@@ -23,6 +23,16 @@ BasicWork::State
 ApplyLedgerWork::onRun()
 {
     ZoneScoped;
+    if (mApp.getLedgerManager().isBooting())
+    {
+        mApp.postOnBackgroundThread(
+            [this] {
+                mApp.getLedgerManager().waitForBoot();
+                wakeUp();
+            },
+            "Wait for LedgerManager to finish booting");
+        return BasicWork::State::WORK_WAITING;
+    }
     mApp.getLedgerManager().applyLedger(mLedgerCloseData,
                                         /* externalize */ false);
     return BasicWork::State::WORK_SUCCESS;

src/catchup/LedgerApplyManagerImpl.cpp

@@ -153,8 +153,11 @@ LedgerApplyManagerImpl::processLedger(LedgerCloseData const& ledgerData,
     // 2. LedgerApplyManager received  ledger that should be immediately applied
     // by LedgerManager: check if we have any sequential ledgers. If so, attempt
     // to apply mSyncingLedgers and possibly get back in sync
-    if (!mCatchupWork && lastReceivedLedgerSeq == *mLastQueuedToApply + 1)
+    if (((!mCatchupWork && lastReceivedLedgerSeq == *mLastQueuedToApply + 1) ||
+         mTMP_TODO) &&
+        !mApp.getLedgerManager().isBooting())
     {
+        mTMP_TODO = false;
         tryApplySyncingLedgers();
         return ProcessLedgerResult::PROCESSED_ALL_LEDGERS_SEQUENTIALLY;
     }
@@ -173,6 +176,13 @@ LedgerApplyManagerImpl::processLedger(LedgerCloseData const& ledgerData,
               "Close of ledger {} buffered. mSyncingLedgers has {} ledgers",
               ledgerData.getLedgerSeq(), mSyncingLedgers.size());
 
+    if (mApp.getLedgerManager().isBooting())
+    {
+        mTMP_TODO = true;
+        return ProcessLedgerResult::WAIT_TO_APPLY_BUFFERED_OR_CATCHUP;
+    }
+    mTMP_TODO = false;
+
     // First: if CatchupWork has started, just buffer and return early.
     if (mCatchupWork)
     {
@@ -513,7 +523,7 @@ LedgerApplyManagerImpl::tryApplySyncingLedgers()
         // If we have too many ledgers queued to apply, just stop scheduling
         // more and let the node gracefully go into catchup.
         releaseAssert(mLastQueuedToApply >= lcl);
-        if (nextToApply - lcl >= MAX_EXTERNALIZE_LEDGER_APPLY_DRIFT)
+        if (nextToApply - lcl >= MAX_EXTERNALIZE_LEDGER_APPLY_DRIFT && false)
         {
             CLOG_INFO(History,
                       "Next ledger to apply is {}, but LCL {} is too far "

src/catchup/LedgerApplyManagerImpl.h

@@ -29,6 +29,7 @@ class LedgerApplyManagerImpl : public LedgerApplyManager
 
     Application& mApp;
     std::shared_ptr<CatchupWork> mCatchupWork;
+    bool mTMP_TODO{false};
 
     // key is ledgerSeq
     // mSyncingLedgers has the following invariants:

src/ledger/LedgerManager.h

@@ -198,6 +198,8 @@ class LedgerManager
     virtual void beginApply() = 0;
     virtual State getState() const = 0;
     virtual std::string getStateHuman() const = 0;
+    virtual bool isBooting() const = 0;
+    virtual void waitForBoot() = 0;
 
     bool
     isSynced() const

src/ledger/LedgerManagerImpl.cpp

@@ -388,6 +388,23 @@ LedgerManagerImpl::getStateHuman() const
     return std::string(stateStrings[getState()]);
 }
 
+bool
+LedgerManagerImpl::isBooting() const
+{
+    std::lock_guard guard{mBootingLock};
+    return mIsBooting;
+}
+
+void
+LedgerManagerImpl::waitForBoot()
+{
+    std::unique_lock<std::mutex> lk{mBootingLock};
+    if (mIsBooting)
+    {
+        mBootingCV.wait(lk, [this] { return !mIsBooting; });
+    }
+}
+
 LedgerHeader
 LedgerManager::genesisLedger()
 {
@@ -526,7 +543,6 @@ LedgerManagerImpl::loadLastKnownLedgerInternal(bool skipBuildingFullState)
 
     // Step 2. Restore LedgerHeader from DB based on the ledger hash derived
     // earlier, or verify we're at genesis if in no-history mode
-    std::optional<LedgerHeader> latestLedgerHeader;
     auto currentLedger =
         LedgerHeaderUtils::loadByHash(getDatabase(), lastLedgerHash);
     if (!currentLedger)
@@ -543,9 +559,7 @@ LedgerManagerImpl::loadLastKnownLedgerInternal(bool skipBuildingFullState)
     CLOG_INFO(Ledger, "Loaded LCL header from database: {}",
               ledgerAbbrev(*currentLedger));
     setLedgerTxnHeader(*currentLedger, mApp);
-    latestLedgerHeader = *currentLedger;
-
-    releaseAssert(latestLedgerHeader.has_value());
+    LedgerHeader latestLedgerHeader = *currentLedger;
 
     auto missing = mApp.getBucketManager().checkForMissingBucketsFiles(has);
     auto pubmissing =
@@ -562,12 +576,12 @@ LedgerManagerImpl::loadLastKnownLedgerInternal(bool skipBuildingFullState)
     // (standalone offline commands, in-memory setup) do not need to
     // spin up expensive merge processes.
     auto assumeStateWork = mApp.getWorkScheduler().executeWork<AssumeStateWork>(
-        has, latestLedgerHeader->ledgerVersion,
+        has, latestLedgerHeader.ledgerVersion,
         /* restartMerges */ !skipBuildingFullState);
     if (assumeStateWork->getState() == BasicWork::State::WORK_SUCCESS)
     {
         CLOG_INFO(Ledger, "Assumed bucket-state for LCL: {}",
-                  ledgerAbbrev(*latestLedgerHeader));
+                  ledgerAbbrev(latestLedgerHeader));
     }
     else
     {
@@ -577,12 +591,12 @@ LedgerManagerImpl::loadLastKnownLedgerInternal(bool skipBuildingFullState)
 
     // Step 4. Restore LedgerManager's LCL state
     advanceLastClosedLedgerState(
-        advanceBucketListSnapshotAndMakeLedgerState(*latestLedgerHeader, has));
+        advanceBucketListSnapshotAndMakeLedgerState(latestLedgerHeader, has));
 
     // Maybe truncate checkpoint files if we're restarting after a crash
     // in applyLedger (in which case any modifications to the ledger state have
     // been rolled back)
-    mApp.getHistoryManager().restoreCheckpoint(latestLedgerHeader->ledgerSeq);
+    mApp.getHistoryManager().restoreCheckpoint(latestLedgerHeader.ledgerSeq);
 
     // Prime module cache with LCL state, not apply-state. This is acceptable
     // here because we just started and there is no apply-state yet and no apply
@@ -591,11 +605,30 @@ LedgerManagerImpl::loadLastKnownLedgerInternal(bool skipBuildingFullState)
     if (!skipBuildingFullState)
     {
         mApplyState.compileAllContractsInLedger(
-            snapshot, latestLedgerHeader->ledgerVersion);
-        mApplyState.populateInMemorySorobanState(
-            snapshot, latestLedgerHeader->ledgerVersion);
+            snapshot, latestLedgerHeader.ledgerVersion);
+        mBootingLock.lock();
+        mIsBooting = true;
+        mBootingLock.unlock();
+        mApp.postOnBackgroundThread(
+            [=] {
+                mApplyState.populateInMemorySorobanState(
+                    snapshot, latestLedgerHeader.ledgerVersion);
+                mApp.postOnMainThread(
+                    [=] {
+                        mBootingLock.lock();
+                        mIsBooting = false;
+                        mBootingLock.unlock();
+                        mBootingCV.notify_all();
+                        mApplyState.markEndOfSetupPhase();
+                    },
+                    "Finish populating in-memory Soroban state");
+            },
+            "Populate in-memory Soroban state");
+    }
+    else
+    {
+        mApplyState.markEndOfSetupPhase();
     }
-    mApplyState.markEndOfSetupPhase();
 }
 
 void
@@ -979,7 +1012,9 @@ void
 LedgerManagerImpl::ApplyState::populateInMemorySorobanState(
     SearchableSnapshotConstPtr snap, uint32_t ledgerVersion)
 {
-    assertSetupPhase();
+    // We don't use assertSetupPhase() because we don't expect the thread
+    // invariant to hold
+    releaseAssert(mPhase == Phase::SETTING_UP_STATE);
     mInMemorySorobanState.initializeStateFromSnapshot(snap, ledgerVersion);
 }
 
@@ -1859,9 +1894,29 @@ LedgerManagerImpl::setLastClosedLedger(
         // a snapshot _from_ the LCL state.
         auto const& snapshot = mLastClosedLedgerState->getBucketSnapshot();
         mApplyState.compileAllContractsInLedger(snapshot, ledgerVersion);
-        mApplyState.populateInMemorySorobanState(snapshot, ledgerVersion);
+        mBootingLock.lock();
+        mIsBooting = true;
+        mBootingLock.unlock();
+        mApp.postOnBackgroundThread(
+            [=] {
+                mApplyState.populateInMemorySorobanState(snapshot,
+                                                         ledgerVersion);
+                mApp.postOnMainThread(
+                    [=] {
+                        mBootingLock.lock();
+                        mIsBooting = false;
+                        mBootingLock.unlock();
+                        mBootingCV.notify_all();
+                        mApplyState.markEndOfSetupPhase();
+                    },
+                    "Finish populating in-memory Soroban state");
+            },
+            "Populate in-memory Soroban state");
+    }
+    else
+    {
+        mApplyState.markEndOfSetupPhase();
     }
-    mApplyState.markEndOfSetupPhase();
 }
 
 void

src/ledger/LedgerManagerImpl.h

@@ -409,6 +409,9 @@ class LedgerManagerImpl : public LedgerManager
                                     Config const& config);
 
     State mState;
+    bool mIsBooting{false};
+    std::condition_variable mBootingCV;
+    std::mutex mutable mBootingLock;
 
 #ifdef BUILD_TESTS
     std::vector<TransactionMetaFrame> mLastLedgerTxMeta;
@@ -479,6 +482,8 @@ class LedgerManagerImpl : public LedgerManager
     void beginApply() override;
     State getState() const override;
     std::string getStateHuman() const override;
+    bool isBooting() const override;
+    void waitForBoot() override;
 
     void valueExternalized(LedgerCloseData const& ledgerData,
                            bool isLatestSlot) override;