Parallelize merge of per-thread changes into global map.

src/ledger/ImmutableLedgerView.cpp

@@ -188,6 +188,12 @@ CheckValidLedgerViewWrapper::CheckValidLedgerViewWrapper(
 {
 }
 
+CheckValidLedgerViewWrapper::CheckValidLedgerViewWrapper(
+    std::unique_ptr<AbstractLedgerView const> getter)
+    : mGetter(std::move(getter))
+{
+}
+
 LedgerHeaderWrapper
 CheckValidLedgerViewWrapper::getLedgerHeader() const
 {
@@ -355,6 +361,70 @@ ImmutableLedgerView::executeWithMaybeInnerSnapshot(
         "ImmutableLedgerView::executeWithMaybeInnerSnapshot is illegal: "
         "ImmutableLedgerView has no nested snapshots");
 }
+SorobanPreApplyLedgerView::SorobanPreApplyLedgerView(
+    std::shared_ptr<LedgerHeader const> header, AbstractLedgerTxn& ltx,
+    ApplyLedgerView const& lclView)
+    : mHeader(std::move(header)), mLtx(ltx), mLclView(lclView)
+{
+}
+
+LedgerHeaderWrapper
+SorobanPreApplyLedgerView::getLedgerHeader() const
+{
+    return LedgerHeaderWrapper(mHeader);
+}
+
+LedgerEntryWrapper
+SorobanPreApplyLedgerView::getAccount(AccountID const& account) const
+{
+    return load(accountKey(account));
+}
+
+LedgerEntryWrapper
+SorobanPreApplyLedgerView::getAccount(LedgerHeaderWrapper const& header,
+                                      TransactionFrame const& tx) const
+{
+    return getAccount(tx.getSourceID());
+}
+
+LedgerEntryWrapper
+SorobanPreApplyLedgerView::getAccount(LedgerHeaderWrapper const& header,
+                                      TransactionFrame const& tx,
+                                      AccountID const& accountID) const
+{
+    return getAccount(accountID);
+}
+
+LedgerEntryWrapper
+SorobanPreApplyLedgerView::load(LedgerKey const& key) const
+{
+    auto entryPair = mLtx.getNewestVersionBelowRoot(key);
+    if (entryPair.first)
+    {
+        // Modified in this ledger, so the ltx has the authoritative version.
+        // A null entry means it has been deleted.
+        if (!entryPair.second)
+        {
+            return LedgerEntryWrapper(nullptr);
+        }
+        // Alias the entry owned by the ltx instead of copying it: the aliasing
+        // constructor shares ownership with the InternalLedgerEntry while
+        // pointing at the LedgerEntry nested inside it.
+        return LedgerEntryWrapper(std::shared_ptr<LedgerEntry const>(
+            entryPair.second, &entryPair.second->ledgerEntry()));
+    }
+    // Not modified in this ledger, so the last closed ledger snapshot is
+    // up to date.
+    return LedgerEntryWrapper(mLclView.loadLiveEntry(key));
+}
+
+void
+SorobanPreApplyLedgerView::executeWithMaybeInnerSnapshot(
+    std::function<void(CheckValidLedgerViewWrapper const& ledgerView)> f) const
+{
+    throw std::runtime_error("SorobanPreApplyLedgerView::"
+                             "executeWithMaybeInnerSnapshot is not supported");
+}
 
 // === Live BucketList wrapper methods ===
 

src/ledger/ImmutableLedgerView.h

@@ -219,6 +219,39 @@ class ApplyLedgerView : private ImmutableLedgerView,
     using ImmutableLedgerView::scanLiveEntriesOfType;
 };
 
+// An ledger view used by the read-only phase of the Soroban pre-apply.
+//
+// It's a thin wrapper around the LTX representing the current ledger state,
+// and the LCL view, which allows the pre-apply phase to observe the changes
+// that happened in the classic phase.
+//
+// Lookups are first attempted in the LTX *newest version* only (which is thread
+// safe as long as we don't mutate the LTX), and only then in the LCL view.
+class SorobanPreApplyLedgerView : public AbstractLedgerView
+{
+  public:
+    SorobanPreApplyLedgerView(std::shared_ptr<LedgerHeader const> header,
+                              AbstractLedgerTxn& ltx,
+                              ApplyLedgerView const& lclView);
+
+    LedgerHeaderWrapper getLedgerHeader() const override;
+    LedgerEntryWrapper getAccount(AccountID const& account) const override;
+    LedgerEntryWrapper getAccount(LedgerHeaderWrapper const& header,
+                                  TransactionFrame const& tx) const override;
+    LedgerEntryWrapper getAccount(LedgerHeaderWrapper const& header,
+                                  TransactionFrame const& tx,
+                                  AccountID const& accountID) const override;
+    LedgerEntryWrapper load(LedgerKey const& key) const override;
+    void executeWithMaybeInnerSnapshot(
+        std::function<void(CheckValidLedgerViewWrapper const&)> f)
+        const override;
+
+  private:
+    std::shared_ptr<LedgerHeader const> mHeader;
+    AbstractLedgerTxn& mLtx;
+    ApplyLedgerView mLclView;
+};
+
 // A helper class to create and query read-only snapshots
 // Automatically decides whether to create a BucketList (recommended), or SQL
 // snapshot (deprecated, but currently supported)
@@ -235,6 +268,8 @@ class CheckValidLedgerViewWrapper : public NonMovableOrCopyable
     CheckValidLedgerViewWrapper(AbstractLedgerTxn& ltx);
     CheckValidLedgerViewWrapper(Application& app);
     explicit CheckValidLedgerViewWrapper(ImmutableLedgerView const& ledgerView);
+    explicit CheckValidLedgerViewWrapper(
+        std::unique_ptr<AbstractLedgerView const> getter);
 #ifdef BUILD_TESTS
     // Set by overlay-only mode call sites so commonValid skips the seqnum
     // equality check: on-disk seqnums are frozen at genesis while

src/ledger/LedgerEntryScope.cpp

@@ -417,6 +417,14 @@ LedgerEntryScope<S>::scopeAdoptEntryOpt(
     return ScopedLedgerEntryOpt(mScopeID, entry);
 }
 
+template <StaticLedgerEntryScope S>
+ScopedLedgerEntryOpt<S>
+LedgerEntryScope<S>::scopeAdoptEntryOpt(
+    std::optional<LedgerEntry>&& entry) const
+{
+    return ScopedLedgerEntryOpt(mScopeID, std::move(entry));
+}
+
 template <StaticLedgerEntryScope S>
 template <StaticLedgerEntryScope OtherScope>
 ScopedLedgerEntry<S>
@@ -456,6 +464,41 @@ LedgerEntryScope<S>::scopeAdoptEntryOptFromImpl(
     return ScopedLedgerEntryOpt<S>{mScopeID, entry.mEntry};
 }
 
+template <StaticLedgerEntryScope S>
+template <StaticLedgerEntryScope OtherScope>
+ScopedLedgerEntry<S>
+LedgerEntryScope<S>::scopeAdoptEntryFromImpl(
+    ScopedLedgerEntry<OtherScope>&& entry,
+    LedgerEntryScope<OtherScope> const& scope) const
+{
+    if (scope.mActive)
+    {
+        throw std::runtime_error(fmt::format(
+            "scopeAdoptEntryFrom: adopting entry with scope ID {} from "
+            "still-active scope ID '{}'",
+            entry.mScopeID, scope.mScopeID));
+    }
+    return EntryT{mScopeID, std::move(entry.mEntry)};
+}
+
+template <StaticLedgerEntryScope S>
+template <StaticLedgerEntryScope OtherScope>
+ScopedLedgerEntryOpt<S>
+LedgerEntryScope<S>::scopeAdoptEntryOptFromImpl(
+    ScopedLedgerEntryOpt<OtherScope>&& entry,
+    LedgerEntryScope<OtherScope> const& scope) const
+{
+    if (scope.mActive)
+    {
+        throw std::runtime_error(
+            fmt::format("scopeAdoptEntryOptFrom: adopting entry with "
+                        "scope ID {} from "
+                        "still-active scope ID '{}'",
+                        entry.mScopeID, scope.mScopeID));
+    }
+    return ScopedLedgerEntryOpt<S>{mScopeID, std::move(entry.mEntry)};
+}
+
 /////////////////////////////////
 // DeactivateScopeGuard
 /////////////////////////////////
@@ -495,6 +538,20 @@ FOREACH_STATIC_LEDGER_ENTRY_SCOPE(INSTANTIATE_SCOPE_CLASSES)
         scopeAdoptEntryOptFromImpl<StaticLedgerEntryScope::SOURCE_SCOPE>( \
             ScopedLedgerEntryOpt<StaticLedgerEntryScope::SOURCE_SCOPE> const&, \
             LedgerEntryScope<StaticLedgerEntryScope::SOURCE_SCOPE> const&) \
+            const; \
+\
+    template ScopedLedgerEntry<StaticLedgerEntryScope::DEST_SCOPE> \
+    LedgerEntryScope<StaticLedgerEntryScope::DEST_SCOPE>:: \
+        scopeAdoptEntryFromImpl<StaticLedgerEntryScope::SOURCE_SCOPE>( \
+            ScopedLedgerEntry<StaticLedgerEntryScope::SOURCE_SCOPE>&&, \
+            LedgerEntryScope<StaticLedgerEntryScope::SOURCE_SCOPE> const&) \
+            const; \
+\
+    template ScopedLedgerEntryOpt<StaticLedgerEntryScope::DEST_SCOPE> \
+    LedgerEntryScope<StaticLedgerEntryScope::DEST_SCOPE>:: \
+        scopeAdoptEntryOptFromImpl<StaticLedgerEntryScope::SOURCE_SCOPE>( \
+            ScopedLedgerEntryOpt<StaticLedgerEntryScope::SOURCE_SCOPE>&&, \
+            LedgerEntryScope<StaticLedgerEntryScope::SOURCE_SCOPE> const&) \
             const;
 
 FOR_EACH_VALID_SCOPE_ADOPTION(INSTANTIATE_ADOPT_METHODS)

src/ledger/LedgerEntryScope.h

@@ -387,6 +387,7 @@ template <StaticLedgerEntryScope S> class LedgerEntryScope
     EntryT scopeAdoptEntry(LedgerEntry const& entry) const;
     OptionalEntryT
     scopeAdoptEntryOpt(std::optional<LedgerEntry> const& entry) const;
+    OptionalEntryT scopeAdoptEntryOpt(std::optional<LedgerEntry>&& entry) const;
 
     template <StaticLedgerEntryScope OtherScope>
     EntryT
@@ -414,6 +415,32 @@ template <StaticLedgerEntryScope S> class LedgerEntryScope
         return scopeAdoptEntryOptFromImpl(entry, scope);
     }
 
+    template <StaticLedgerEntryScope OtherScope>
+    EntryT
+    scopeAdoptEntryFrom(ScopedLedgerEntry<OtherScope>&& entry,
+                        LedgerEntryScope<OtherScope> const& scope) const
+    {
+        static_assert(
+            IsValidScopeAdoption<S, OtherScope>::value,
+            "Invalid scope adoption: this transition is not allowed. "
+            "Check FOR_EACH_VALID_SCOPE_ADOPTION in LedgerEntryScope.h "
+            "for the list of valid transitions.");
+        return scopeAdoptEntryFromImpl(std::move(entry), scope);
+    }
+
+    template <StaticLedgerEntryScope OtherScope>
+    OptionalEntryT
+    scopeAdoptEntryOptFrom(ScopedLedgerEntryOpt<OtherScope>&& entry,
+                           LedgerEntryScope<OtherScope> const& scope) const
+    {
+        static_assert(
+            IsValidScopeAdoption<S, OtherScope>::value,
+            "Invalid scope adoption: this transition is not allowed. "
+            "Check FOR_EACH_VALID_SCOPE_ADOPTION in LedgerEntryScope.h "
+            "for the list of valid transitions.");
+        return scopeAdoptEntryOptFromImpl(std::move(entry), scope);
+    }
+
   private:
     template <StaticLedgerEntryScope OtherScope>
     EntryT
@@ -424,6 +451,16 @@ template <StaticLedgerEntryScope S> class LedgerEntryScope
     OptionalEntryT
     scopeAdoptEntryOptFromImpl(ScopedLedgerEntryOpt<OtherScope> const& entry,
                                LedgerEntryScope<OtherScope> const& scope) const;
+
+    template <StaticLedgerEntryScope OtherScope>
+    EntryT
+    scopeAdoptEntryFromImpl(ScopedLedgerEntry<OtherScope>&& entry,
+                            LedgerEntryScope<OtherScope> const& scope) const;
+
+    template <StaticLedgerEntryScope OtherScope>
+    OptionalEntryT
+    scopeAdoptEntryOptFromImpl(ScopedLedgerEntryOpt<OtherScope>&& entry,
+                               LedgerEntryScope<OtherScope> const& scope) const;
 };
 
 template <StaticLedgerEntryScope S> class DeactivateScopeGuard

src/ledger/LedgerManagerImpl.cpp

@@ -2724,7 +2724,7 @@ LedgerManagerImpl::applySorobanStage(
             txBundle.getResPayload().getRefundableFeeTracker());
     }
 
-    globalParState.commitChangesFromThreads(app, threadStates, stage);
+    globalParState.commitChangesFromThreads(app, threadStates);
 }
 
 void

src/transactions/FeeBumpTransactionFrame.cpp

@@ -83,39 +83,52 @@ FeeBumpTransactionFrame::FeeBumpTransactionFrame(
 #endif
 
 void
-FeeBumpTransactionFrame::preParallelApply(
-    AppConnector& app, AbstractLedgerTxn& ltx, TransactionMetaBuilder& meta,
-    MutableTransactionResultBase& txResult,
+FeeBumpTransactionFrame::preParallelApplyReadOnly(
+    AppConnector& app, CheckValidLedgerViewWrapper const& ls,
+    TransactionMetaBuilder& meta, MutableTransactionResultBase& txResult,
     SorobanNetworkConfig const& sorobanConfig) const
 {
     try
     {
-        LedgerTxn ltxTx(ltx);
-        removeOneTimeSignerKeyFromFeeSource(ltxTx);
-        meta.pushTxChangesBefore(ltxTx);
-        ltxTx.commit();
+        mInnerTx->preParallelApplyReadOnlyWithOptionallyChargedFee(
+            /*chargeFee=*/false, app, ls, meta, txResult, sorobanConfig,
+            getContentsHash());
     }
     catch (std::exception& e)
     {
-        printErrorAndAbort("Exception in preParallelApply ", e.what());
+        printErrorAndAbort("Exception during read-only preParallelApply: ",
+                           e.what());
     }
     catch (...)
     {
-        printErrorAndAbort("Unknown exception in preParallelApply");
+        printErrorAndAbort(
+            "Unknown exception during read-only preParallelApply");
     }
+}
 
+void
+FeeBumpTransactionFrame::preParallelApplyWrite(
+    AppConnector& app, AbstractLedgerTxn& ltx, TransactionMetaBuilder& meta,
+    MutableTransactionResultBase const& txResult) const
+{
     try
     {
-        mInnerTx->preParallelApply(/*chargeFee=*/false, app, ltx, meta,
-                                   txResult, sorobanConfig, getContentsHash());
+        {
+            LedgerTxn ltxTx(ltx);
+            removeOneTimeSignerKeyFromFeeSource(ltxTx);
+            meta.pushTxChangesBefore(ltxTx);
+            ltxTx.commit();
+        }
+        mInnerTx->preParallelApplyWrite(app, ltx, meta, txResult);
     }
     catch (std::exception& e)
     {
-        printErrorAndAbort("Exception during preParallelApply: ", e.what());
+        printErrorAndAbort("Exception during preParallelApply writes: ",
+                           e.what());
     }
     catch (...)
     {
-        printErrorAndAbort("Unknown exception during preParallelApply");
+        printErrorAndAbort("Unknown exception during preParallelApply writes");
     }
 }
 

src/transactions/FeeBumpTransactionFrame.h

@@ -90,11 +90,14 @@ class FeeBumpTransactionFrame : public TransactionFrameBase
 
     ~FeeBumpTransactionFrame() override = default;
 
-    void
-    preParallelApply(AppConnector& app, AbstractLedgerTxn& ltx,
-                     TransactionMetaBuilder& meta,
-                     MutableTransactionResultBase& txResult,
-                     SorobanNetworkConfig const& sorobanConfig) const override;
+    void preParallelApplyReadOnly(
+        AppConnector& app, CheckValidLedgerViewWrapper const& ls,
+        TransactionMetaBuilder& meta, MutableTransactionResultBase& txResult,
+        SorobanNetworkConfig const& sorobanConfig) const override;
+
+    void preParallelApplyWrite(
+        AppConnector& app, AbstractLedgerTxn& ltx, TransactionMetaBuilder& meta,
+        MutableTransactionResultBase const& txResult) const override;
 
     std::optional<ParallelTxSuccessVal> parallelApply(
         AppConnector& app, ThreadParallelApplyLedgerState const& threadState,