Parallelize Soroban pre-apply step.

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/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 across 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