parallelize classic-entry collection and TTL extraction, instrument post-tx apply

Three mechanical wins on the apply path (60-ledger SAC benchmark mean
close 117.1 -> 112.3ms, tracy-less build):

- collectModifiedClassicEntries now runs in two parallel phases on the
  apply pool: tx bundles are chunked across workers that bin classic
  footprint keys by global-map shard (hashing each key once), then one
  worker per shard dedups and copies the modified entries out of the
  ltx (pure read: getNewestVersionBelowRoot terminates at the root
  without touching its caches). setup_collect_classic 5.6 -> 0.8ms.

- extractDirtyTTLShardEntries extracts the 16 global-map shards on
  parallel workers; the extracted vector is shared between the early
  in-memory updater and the TTL shard writer, which now makes its own
  sorted copy off the apply thread. par gap 7.1 -> 5.5ms.

- post_tx_set_apply split into post_tx_refunds (8.1ms: per-tx fee
  refund writes through the ltx) and post_tx_results (1.6ms: result
  XDR serialization), exposing refunds as part of the serial-ltx cost
  pool for the next optimization round.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Copilot

src/ledger/LedgerManagerImpl.cpp

@@ -341,7 +341,7 @@ LedgerManagerImpl::ApplyState::updateInMemorySorobanState(
 void
 LedgerManagerImpl::ApplyState::startEarlyInMemorySorobanStateUpdate(
     std::vector<std::shared_future<std::shared_ptr<LiveBucket>>> threadShards,
-    std::vector<BucketEntry> ttlEntries,
+    std::shared_ptr<std::vector<BucketEntry> const> ttlEntries,
     SorobanNetworkConfig const& sorobanConfig, uint32_t ledgerVersion)
 {
     threadInvariant();
@@ -356,8 +356,7 @@ LedgerManagerImpl::ApplyState::startEarlyInMemorySorobanStateUpdate(
 
     // The TTL entries are shared (read-only) between the byproduct scan and
     // the state application.
-    auto sharedTtl = std::make_shared<std::vector<BucketEntry> const>(
-        std::move(ttlEntries));
+    auto sharedTtl = std::move(ttlEntries);
 
     // Byproduct scan: modified-key set (for eviction) and new contract code
     // (for the module cache). Fast; runs in parallel with the state
@@ -3016,21 +3015,26 @@ LedgerManagerImpl::applySorobanStages(AppConnector& app, AbstractLedgerTxn& ltx,
         // so read-only TTL bumps are fully reconciled: write the TTL shard
         // optimistically (overlapping the ltx commit and post-apply work).
         // The thread-shard futures are snapshotted first: the early
-        // in-memory updater consumes the TTL changes directly (by value)
+        // in-memory updater shares the extracted TTL entries with the shard
+        // writer (which makes its own sorted copy off the apply thread)
         // rather than waiting on the TTL shard's file write.
         auto threadShardFutures = mPendingLedgerShards;
-        auto ttlEntries = globalParState.extractDirtyTTLShardEntries();
-        auto ttlEntriesForUpdater = ttlEntries;
-        if (!ttlEntries.empty())
+        auto ttlEntries = std::make_shared<std::vector<BucketEntry> const>(
+            globalParState.extractDirtyTTLShardEntries(app));
+        if (!ttlEntries->empty())
         {
             auto& bm = mApp.getBucketManager();
             auto& workerCtx = mApp.getWorkerIOContext();
             bool doFsync = !mApp.getConfig().DISABLE_XDR_FSYNC;
             uint32_t protocol = header.ledgerVersion;
             mPendingLedgerShards.emplace_back(
                 std::async(std::launch::async,
-                           [&bm, &workerCtx, protocol, doFsync,
-                            entries = std::move(ttlEntries)]() mutable {
+                           [&bm, &workerCtx, protocol, doFsync, ttlEntries]() {
+                               // Copy + sort off the apply thread; the shared
+                               // vector itself stays unsorted for the early
+                               // updater (TTL keys are unique, so apply order
+                               // doesn't matter there).
+                               auto entries = *ttlEntries;
                                std::sort(entries.begin(), entries.end(),
                                          BucketEntryIdCmp<LiveBucket>{});
                                return LiveBucket::freshShard(
@@ -3051,10 +3055,10 @@ LedgerManagerImpl::applySorobanStages(AppConnector& app, AbstractLedgerTxn& ltx,
         // legacy seal-time updateState instead. Also skipped when this
         // ledger produced no soroban changes (e.g. empty tx sets in tests).
         if (bypassLtxForSorobanEntries(mApp.getConfig()) &&
-            (!threadShardFutures.empty() || !ttlEntriesForUpdater.empty()))
+            (!threadShardFutures.empty() || !ttlEntries->empty()))
         {
             mApplyState.startEarlyInMemorySorobanStateUpdate(
-                std::move(threadShardFutures), std::move(ttlEntriesForUpdater),
+                std::move(threadShardFutures), std::move(ttlEntries),
                 sorobanConfig, header.ledgerVersion);
         }
 
@@ -3232,6 +3236,8 @@ LedgerManagerImpl::applyTransactions(
         std::chrono::duration<double, std::milli>(txSubEnd - txSubStart)
             .count();
     mLastPhaseTimings.applySeqClassicMs = 0;
+    mLastPhaseTimings.postTxRefundsMs = 0;
+    mLastPhaseTimings.postTxResultsMs = 0;
 #endif
     std::vector<ApplyStage> applyStages;
     for (auto const& phase : phases)
@@ -3471,6 +3477,9 @@ LedgerManagerImpl::processPostTxSetApply(
             {
                 for (auto const& txBundle : stage)
                 {
+#ifdef BUILD_TESTS
+                    auto refundStart = std::chrono::steady_clock::now();
+#endif
                     if (ledgerCloseMeta)
                     {
                         // Use child LTX for meta change tracking.
@@ -3496,13 +3505,26 @@ LedgerManagerImpl::processPostTxSetApply(
                                 .getMeta()
                                 .getTxEventManager());
                     }
+#ifdef BUILD_TESTS
+                    auto refundEnd = std::chrono::steady_clock::now();
+                    mLastPhaseTimings.postTxRefundsMs +=
+                        std::chrono::duration<double, std::milli>(refundEnd -
+                                                                  refundStart)
+                            .count();
+#endif
 
                     // setPostTxApplyFeeProcessing can update the feeCharged in
                     // the result, so this needs to be done after
                     processResultAndMeta(ledgerCloseMeta, txBundle.getTxNum(),
                                          txBundle.getEffects().getMeta(),
                                          *txBundle.getTx(),
                                          txBundle.getResPayload(), txResultSet);
+#ifdef BUILD_TESTS
+                    mLastPhaseTimings.postTxResultsMs +=
+                        std::chrono::duration<double, std::milli>(
+                            std::chrono::steady_clock::now() - refundEnd)
+                            .count();
+#endif
                 }
             }
         }

src/ledger/LedgerManagerImpl.h

@@ -276,12 +276,13 @@ class LedgerManagerImpl : public LedgerManager
         // runs at seal via joinEarlyInMemorySorobanStateUpdate.
         // threadShards: the per-cluster shard futures (long since written
         // by updater start). ttlEntries: this ledger's reconciled TTL
-        // changes, fed directly (by value) so the updater need not wait for
-        // the TTL shard's file write.
+        // changes, fed directly (shared with the TTL shard writer, which
+        // makes its own sorted copy off the apply thread) so the updater
+        // need not wait for the TTL shard's file write.
         void startEarlyInMemorySorobanStateUpdate(
             std::vector<std::shared_future<std::shared_ptr<LiveBucket>>>
                 threadShards,
-            std::vector<BucketEntry> ttlEntries,
+            std::shared_ptr<std::vector<BucketEntry> const> ttlEntries,
             SorobanNetworkConfig const& sorobanConfig, uint32_t ledgerVersion);
 
         bool
@@ -659,6 +660,11 @@ class LedgerManagerImpl : public LedgerManager
         double applyParallelPhaseTotalMs = 0;
         double applySeqClassicMs = 0;
         double postTxSetApplyMs = 0;
+        // Sub-timings of postTxSetApplyMs: the per-tx post-apply fee
+        // processing (refunds, written through the ltx) and the result/meta
+        // collection (result XDR serialization + result set append).
+        double postTxRefundsMs = 0;
+        double postTxResultsMs = 0;
         double applyTxTailMs = 0;
         double destroyApplyStagesMs = 0;
         double applyUpgradesMs = 0;

src/simulation/ApplyLoad.cpp

@@ -284,6 +284,10 @@ logPhaseTimingsTable(
         extract(&LedgerManagerImpl::LedgerClosePhaseTimings::applySeqClassicMs);
     auto postTxSetApply =
         extract(&LedgerManagerImpl::LedgerClosePhaseTimings::postTxSetApplyMs);
+    auto postTxRefunds =
+        extract(&LedgerManagerImpl::LedgerClosePhaseTimings::postTxRefundsMs);
+    auto postTxResults =
+        extract(&LedgerManagerImpl::LedgerClosePhaseTimings::postTxResultsMs);
     auto applyTxTail =
         extract(&LedgerManagerImpl::LedgerClosePhaseTimings::applyTxTailMs);
     auto destroyApplyStages = extract(
@@ -410,6 +414,8 @@ logPhaseTimingsTable(
         {"|   *** par gap ***", computePhaseStats(parGap)},
         {"| apply_seq_classic", computePhaseStats(applySeqClassic)},
         {"| post_tx_set_apply", computePhaseStats(postTxSetApply)},
+        {"|   post_tx_refunds", computePhaseStats(postTxRefunds)},
+        {"|   post_tx_results", computePhaseStats(postTxResults)},
         {"| tail", computePhaseStats(applyTxTail)},
         {"| ~apply_stages", computePhaseStats(destroyApplyStages)},
         {"| *** tx gap ***", computePhaseStats(txGap)},

src/transactions/ParallelApplyUtils.cpp

@@ -503,7 +503,7 @@ GlobalParallelApplyLedgerState::
         mSetupCommitWritesMs += setupLap();
         mSetupSeqWriteMs += gSeqPreApplyWriteMs;
 #endif
-        collectModifiedClassicEntries(ltx, stages);
+        collectModifiedClassicEntries(app, ltx, stages);
         return;
     }
 
@@ -648,53 +648,120 @@ GlobalParallelApplyLedgerState::commitBufferedPreParallelApplyWrites(
 
 void
 GlobalParallelApplyLedgerState::collectModifiedClassicEntries(
-    AbstractLedgerTxn& ltx, std::vector<ApplyStage> const& stages)
+    AppConnector& app, AbstractLedgerTxn& ltx,
+    std::vector<ApplyStage> const& stages)
 {
     ZoneScoped;
 #ifdef BUILD_TESTS
     auto _collectT0 = std::chrono::steady_clock::now();
 #endif
 
-    std::unordered_set<LedgerKey> classicKeys;
+    // Collect the classic footprint keys and copy their modified entries out
+    // of the ltx, in two parallel phases on the (currently idle) apply pool:
+    // first the tx bundles are chunked across workers, each binning classic
+    // keys by target global-map shard (computing each key's hash once);
+    // then one worker per shard deduplicates its keys and copies the
+    // modified entries into that shard. The ltx reads are pure lookups
+    // (getNewestVersionBelowRoot walks per-level entry maps and terminates
+    // at the root without touching its caches), and distinct shard workers
+    // touch disjoint submaps, so no synchronization is needed.
+    std::vector<TxBundle const*> bundles;
     for (auto const& stage : stages)
     {
         for (auto const& txBundle : stage)
         {
-            auto const& footprint =
-                txBundle.getTx()->sorobanResources().footprint;
-            for (auto const& key : footprint.readWrite)
-            {
-                if (!isSorobanEntry(key))
-                {
-                    classicKeys.emplace(key);
-                }
-            }
-            for (auto const& key : footprint.readOnly)
-            {
-                if (!isSorobanEntry(key))
-                {
-                    classicKeys.emplace(key);
-                }
-            }
+            bundles.push_back(&txBundle);
         }
     }
 
-    for (auto const& lk : classicKeys)
+    auto& threadPool = app.getApplyThreadPool();
+    threadPool.ensureWorkerCount(1);
+    size_t const numChunks = std::max<size_t>(
+        1, std::min<size_t>(kGlobalEntryMapShards, bundles.size()));
+    std::vector<
+        std::array<std::vector<ParallelApplyLedgerKey>, kGlobalEntryMapShards>>
+        bins(numChunks);
     {
-        auto entryPair = ltx.getNewestVersionBelowRoot(lk);
-        if (!entryPair.first)
+        std::vector<std::future<void>> futures;
+        futures.reserve(numChunks);
+        size_t const chunkSize = (bundles.size() + numChunks - 1) / numChunks;
+        for (size_t c = 0; c < numChunks; ++c)
         {
-            continue;
+            futures.emplace_back(
+                threadPool.submit([c, chunkSize, &bundles, &bins]() {
+                    size_t const begin = c * chunkSize;
+                    size_t const end =
+                        std::min(begin + chunkSize, bundles.size());
+                    auto binKey = [&](LedgerKey const& key) {
+                        if (isSorobanEntry(key))
+                        {
+                            return;
+                        }
+                        ParallelApplyLedgerKey pk(key);
+                        bins[c][globalMapShardOf(pk)].push_back(std::move(pk));
+                    };
+                    for (size_t i = begin; i < end; ++i)
+                    {
+                        auto const& fp =
+                            bundles[i]->getTx()->sorobanResources().footprint;
+                        for (auto const& key : fp.readWrite)
+                        {
+                            binKey(key);
+                        }
+                        for (auto const& key : fp.readOnly)
+                        {
+                            binKey(key);
+                        }
+                    }
+                }));
         }
+        for (auto& f : futures)
+        {
+            releaseAssert(f.valid());
+            f.get();
+        }
+    }
 
-        GlobalParApplyLedgerEntryOpt entry = scopeAdoptEntryOpt(
-            entryPair.second
-                ? std::make_optional(entryPair.second->ledgerEntry())
-                : std::nullopt);
-
-        ParallelApplyLedgerKey pk(lk);
-        globalMapShardFor(pk).emplace(std::move(pk),
-                                      GlobalParallelApplyEntry{entry, false});
+    {
+        std::vector<std::future<void>> futures;
+        futures.reserve(kGlobalEntryMapShards);
+        for (size_t s = 0; s < kGlobalEntryMapShards; ++s)
+        {
+            futures.emplace_back(threadPool.submit([this, s, &bins, &ltx]() {
+                auto& shard = mGlobalEntryMapShards[s];
+                for (auto& bin : bins)
+                {
+                    for (auto& pk : bin[s])
+                    {
+                        // Skip duplicates (and keys already collected by the
+                        // pre-apply phases), matching the no-op-on-duplicate
+                        // emplace semantics of the serial version.
+                        if (shard.find(pk) != shard.end())
+                        {
+                            continue;
+                        }
+                        auto entryPair =
+                            ltx.getNewestVersionBelowRoot(pk.ledgerKey());
+                        if (!entryPair.first)
+                        {
+                            continue;
+                        }
+                        GlobalParApplyLedgerEntryOpt entry = scopeAdoptEntryOpt(
+                            entryPair.second
+                                ? std::make_optional(
+                                      entryPair.second->ledgerEntry())
+                                : std::nullopt);
+                        shard.emplace(std::move(pk), GlobalParallelApplyEntry{
+                                                         entry, false});
+                    }
+                }
+            }));
+        }
+        for (auto& f : futures)
+        {
+            releaseAssert(f.valid());
+            f.get();
+        }
     }
 
 #ifdef BUILD_TESTS
@@ -1086,13 +1153,35 @@ ThreadParallelApplyLedgerState::extractDirtySorobanShardEntries() const
 }
 
 std::vector<BucketEntry>
-GlobalParallelApplyLedgerState::extractDirtyTTLShardEntries() const
+GlobalParallelApplyLedgerState::extractDirtyTTLShardEntries(
+    AppConnector& app) const
 {
+    ZoneScoped;
+    // Extract the shards on parallel workers (the apply pool is idle at this
+    // point: all stages have joined), then concatenate.
+    auto& threadPool = app.getApplyThreadPool();
+    threadPool.ensureWorkerCount(1);
+    std::array<std::vector<BucketEntry>, kGlobalEntryMapShards> perShard;
+    std::vector<std::future<void>> futures;
+    futures.reserve(kGlobalEntryMapShards);
+    for (size_t i = 0; i < kGlobalEntryMapShards; ++i)
+    {
+        futures.emplace_back(threadPool.submit([this, i, &perShard]() {
+            perShard[i] = extractDirtyEntriesForShard(mGlobalEntryMapShards[i],
+                                                      *this, /*ttlOnly=*/true);
+        }));
+    }
+    size_t total = 0;
+    for (size_t i = 0; i < kGlobalEntryMapShards; ++i)
+    {
+        releaseAssert(futures[i].valid());
+        futures[i].get();
+        total += perShard[i].size();
+    }
     std::vector<BucketEntry> entries;
-    for (auto const& shard : mGlobalEntryMapShards)
+    entries.reserve(total);
+    for (auto& shardEntries : perShard)
     {
-        auto shardEntries =
-            extractDirtyEntriesForShard(shard, *this, /*ttlOnly=*/true);
         entries.insert(entries.end(),
                        std::make_move_iterator(shardEntries.begin()),
                        std::make_move_iterator(shardEntries.end()));

src/transactions/ParallelApplyUtils.h

@@ -284,7 +284,8 @@ class GlobalParallelApplyLedgerState
         AppConnector& app, AbstractLedgerTxn& ltx,
         std::vector<TxBundle const*> const& txBundles);
 
-    void collectModifiedClassicEntries(AbstractLedgerTxn& ltx,
+    void collectModifiedClassicEntries(AppConnector& app,
+                                       AbstractLedgerTxn& ltx,
                                        std::vector<ApplyStage> const& stages);
 
     bool maybeMergeRoTTLBumps(ParallelApplyLedgerKey const& key,
@@ -343,8 +344,9 @@ class GlobalParallelApplyLedgerState
     // stages) as (unsorted) BucketEntries for a level-0 shard write. Must
     // be called after the last stage's commitChangesFromThreads (so
     // read-only TTL bumps have been max-merged) and before
-    // commitChangesToLedgerTxn (which moves entries out).
-    std::vector<BucketEntry> extractDirtyTTLShardEntries() const;
+    // commitChangesToLedgerTxn (which moves entries out). Extracts the map
+    // shards on the (idle) apply pool in parallel.
+    std::vector<BucketEntry> extractDirtyTTLShardEntries(AppConnector& app) const;
 
     // Consumes the global entry map: moves entries into the LedgerTxn
     // instead of copying. Must only be called once, as the final operation