Parallelize Soroban pre-apply step.

While the current logic intertwines reads and writes, in fact it can be cleanly separated into a read-only validation step, and a sequential commit step that simply bumps the sequence numbers and removes pre-authorized tx signers. This is possible because that while the writes change the entries that take part in validation, none of these changes are relevant during the validation. Specifically, sequence number bump is only observable by a single transaction (the one that has the respective account as a source), and the pre-authorized tx signer by definition belongs to a single transaction. There is also a subtle caveat to the latter operation: it increases the available balance of the signer owner (or its sponsor), but since at the pre-apply time the fees have already been charged, we're only checking that the account available balance is non-negative, which is an invariant that must always hold in the current protocol.

The change is not protocol-gated because it's not a protocol change for the *current* protocol. It was technically a protocol change prior to p26 where we had a bug that actually did allow overcharging the fee bump source accounts and thus making their available balance to go negative. However, the bug has been fixed without the behavior ever triggering on-chain, and thus this replay-only behavior change should be non-observable.

This change significantly speeds up the pre-apply step. On the local high TPL benchmarks I'm getting 30-60ms improvement locally compared to the main branch version.

Author: dmkozh

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(std::make_shared<LedgerEntry const>());
+        }
+        // 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/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,

src/transactions/ParallelApplyUtils.cpp

@@ -8,14 +8,18 @@
 #include "ledger/LedgerTxn.h"
 #include "ledger/NetworkConfig.h"
 #include "main/AppConnector.h"
+#include "transactions/OperationFrame.h"
 #include "transactions/ParallelApplyStage.h"
 #include "transactions/TransactionFrameBase.h"
+#include "transactions/TransactionUtils.h"
+#include "util/BatchExecutor.h"
 #include "util/GlobalChecks.h"
 #include "xdr/Stellar-ledger-entries.h"
 #include "xdrpp/printer.h"
 #include <fmt/core.h>
 #include <fmt/std.h>
 #include <thread>
+#include <unordered_map>
 
 namespace
 {
@@ -174,8 +178,21 @@ updateMaxOfRoTTLBump(UnorderedMap<LedgerKey, uint32_t>& roTTLBumps,
     }
 }
 
+void
+commitPreParallelApplyWrites(AppConnector& app, AbstractLedgerTxn& ltx,
+                             std::vector<TxBundle const*> const& txBundles)
+{
+    ZoneScoped;
+    for (auto const* txBundle : txBundles)
+    {
+        txBundle->getTx()->preParallelApplyWrite(
+            app, ltx, txBundle->getEffects().getMeta(),
+            txBundle->getResPayload());
+    }
 }
 
+} // namespace
+
 namespace stellar
 {
 
@@ -319,14 +336,13 @@ GlobalParallelApplyLedgerState::GlobalParallelApplyLedgerState(
     // had their sequence numbers bumped and fees charged. preParallelApply will
     // update sequence numbers so it needs to be called before we check
     // LedgerTxn.
-    preParallelApplyAndCollectModifiedClassicEntries(app, ltx, stages);
+    preApplyAndCollectModifiedClassicEntries(app, ltx, stages);
 }
 
 void
-GlobalParallelApplyLedgerState::
-    preParallelApplyAndCollectModifiedClassicEntries(
-        AppConnector& app, AbstractLedgerTxn& ltx,
-        std::vector<ApplyStage> const& stages)
+GlobalParallelApplyLedgerState::preApplyAndCollectModifiedClassicEntries(
+    AppConnector& app, AbstractLedgerTxn& ltx,
+    std::vector<ApplyStage> const& stages)
 {
     auto fetchInMemoryClassicEntries =
         [&](xdr::xvector<LedgerKey> const& keys) {
@@ -353,34 +369,99 @@ GlobalParallelApplyLedgerState::
             }
         };
 
-    // First call preParallelApply on all transactions,
-    // and then load from footprints. This order is important
-    // because preParallelApply modifies the fee source accounts
-    // and those accounts could show up in the footprint
-    // of a different transaction.
+    std::vector<TxBundle const*> txBundles;
     for (auto const& stage : stages)
     {
         for (auto const& txBundle : stage)
         {
-            // Make sure to call preParallelApply on all txs because this will
-            // modify the fee source accounts sequence numbers.
-            txBundle.getTx()->preParallelApply(
-                app, ltx, txBundle.getEffects().getMeta(),
-                txBundle.getResPayload(), mSorobanConfig);
+            txBundles.emplace_back(&txBundle);
         }
     }
 
-    for (auto const& stage : stages)
+    // Pre-apply all the transactions before loading the footprint entries. This
+    // order is important because the pre-apply modifies the source accounts,
+    // and those accounts could show up in the footprint of a transaction
+    // applied by a different thread, thus breaking the invariant that
+    // transactions are independent of each other acrooss threads.
+    //
+    // The pre-apply process is done in two phases: a parallel read-only phase
+    // where the transactions are validated, and a serial write phase where the
+    // writes are committed to the ledger.
+    //
+    // This phase separatation hinges on the fact that the validation outcome
+    // of any Soroban transaction can't be influenced by the pre-apply writes
+    // performed by another Soroban transaction. Specifically, pre-apply writes
+    // only include:
+    // - The source account sequence number bumps - this is fine because we
+    //   have only a single transaction per source account per ledger
+    // - The removal of one-time pre-authorized tx signers - this is also fine
+    //   because any given transaction in a ledger is unique, and increasing the
+    //   sub-entry count of a source/sponsor account is not relevant at that
+    //   point, as the fees have already been successfully charged.
+
+    auto header =
+        std::make_shared<LedgerHeader const>(ltx.loadHeader().current());
+    readOnlyParallelPreApply(app, txBundles, header, ltx);
+    commitPreParallelApplyWrites(app, ltx, txBundles);
+
+    for (auto const& txBundle : txBundles)
     {
-        for (auto const& txBundle : stage)
-        {
-            auto const& footprint =
-                txBundle.getTx()->sorobanResources().footprint;
+        auto const& footprint = txBundle->getTx()->sorobanResources().footprint;
+        fetchInMemoryClassicEntries(footprint.readWrite);
+        fetchInMemoryClassicEntries(footprint.readOnly);
+    }
+}
 
-            fetchInMemoryClassicEntries(footprint.readWrite);
-            fetchInMemoryClassicEntries(footprint.readOnly);
+void
+GlobalParallelApplyLedgerState::readOnlyParallelPreApply(
+    AppConnector& app, std::vector<TxBundle const*> const& txBundles,
+    std::shared_ptr<LedgerHeader const> header, AbstractLedgerTxn& ltx)
+{
+    ZoneScoped;
+    if (txBundles.empty())
+    {
+        return;
+    }
+
+    // Run pre-apply for [begin, end) transaction indices.
+    auto runRange = [&](size_t begin, size_t end) {
+        // NB: mLCLApplyView is not thread-safe, so we need to copy it into a
+        // thread-local view.
+        CheckValidLedgerViewWrapper ledgerView(
+            std::make_unique<SorobanPreApplyLedgerView>(header, ltx,
+                                                        mLCLApplyView));
+        for (size_t i = begin; i < end; ++i)
+        {
+            auto const* txBundle = txBundles[i];
+            txBundle->getTx()->preParallelApplyReadOnly(
+                app, ledgerView, txBundle->getEffects().getMeta(),
+                txBundle->getResPayload(), mSorobanConfig);
         }
+    };
+
+    size_t taskCount = app.getBatchExecutor().preferredTaskCount();
+    if (taskCount <= 1)
+    {
+        runRange(0, txBundles.size());
+        return;
+    }
+
+    std::vector<std::function<int()>> tasks;
+    tasks.reserve(taskCount);
+    size_t begin = 0;
+    size_t baseChunk = txBundles.size() / taskCount;
+    size_t remainder = txBundles.size() % taskCount;
+    for (size_t i = 0; i < taskCount; ++i)
+    {
+        size_t end = begin + baseChunk + (i < remainder ? 1 : 0);
+        tasks.emplace_back([runRange, begin, end]() {
+            runRange(begin, end);
+            return 0;
+        });
+        begin = end;
     }
+    releaseAssert(begin == txBundles.size());
+    app.getBatchExecutor().executeBatch(std::move(tasks));
 }
 
 void