Make Soroban metric batching code lock-free.

The previous attempt at batching Soroban metrics still had some locks and was overall more convoluted than it needs to be. With this change we simply give every thread its own metrics batch to write to, and then cleanly move the batches to the publish step for the merge and Medida export. This is slightly more plumbing, but as a result we get much more straightforward code.

Author: dmkozh

src/ledger/InMemorySorobanState.cpp

@@ -488,7 +488,7 @@ InMemorySorobanState::updateState(
     std::vector<LedgerEntry> const& liveEntries,
     std::vector<LedgerKey> const& deadEntries, LedgerHeader const& lh,
     std::optional<SorobanNetworkConfig const> const& sorobanConfig,
-    SorobanMetrics& metrics)
+    SorobanMetricsRegistry& metrics)
 {
     // After initialization, we must apply every ledger in order to the
     // in-memory state with no gaps.
@@ -575,7 +575,7 @@ InMemorySorobanState::getSize() const
 }
 
 void
-InMemorySorobanState::reportMetrics(SorobanMetrics& metrics) const
+InMemorySorobanState::reportMetrics(SorobanMetricsRegistry& metrics) const
 {
     metrics.mContractCodeStateSize.set_count(mContractCodeStateSize);
     metrics.mContractDataStateSize.set_count(mContractDataStateSize);

src/ledger/InMemorySorobanState.h

@@ -20,7 +20,7 @@ namespace stellar
 class ApplyLedgerView;
 
 class InvariantManagerImpl;
-class SorobanMetrics;
+class SorobanMetricsRegistry;
 
 // TTLData stores both liveUntilLedgerSeq and lastModifiedLedgerSeq for TTL
 // entries. This allows us to construct a LedgerEntry for TTLs without having to
@@ -396,7 +396,7 @@ class InMemorySorobanState
     // CONTRACT_CODE.
     void deleteContractCode(LedgerKey const& ledgerKey);
 
-    void reportMetrics(SorobanMetrics& metrics) const;
+    void reportMetrics(SorobanMetricsRegistry& metrics) const;
 
   public:
     InMemorySorobanState() = default;
@@ -449,7 +449,7 @@ class InMemorySorobanState
                 std::vector<LedgerKey> const& deadEntries,
                 LedgerHeader const& lh,
                 std::optional<SorobanNetworkConfig const> const& sorobanConfig,
-                SorobanMetrics& metrics);
+                SorobanMetricsRegistry& metrics);
 
     // Should only be called in manual ledger close paths.
     void manuallyAdvanceLedgerHeader(LedgerHeader const& lh);

src/ledger/LedgerManager.h

@@ -19,7 +19,8 @@ namespace stellar
 
 class LedgerCloseData;
 class Database;
-class SorobanMetrics;
+class SorobanMetricsRegistry;
+struct SorobanApplyMetrics;
 class InMemorySorobanState;
 
 // This diagram provides a schematic of the flow of (logical) ledgers coming in
@@ -373,11 +374,11 @@ class LedgerManager
     // upgradeApplied should be true if a protocol or network config setting
     // upgrade occurred during the ledger close. If inMemorySnapshotForInvariant
     // is not null, this will kick off a snapshot invariant check.
-    virtual void completeLedgerClose(uint32_t ledgerSeq,
-                                     bool calledViaExternalize,
-                                     LedgerCloseData const& ledgerData,
-                                     ImmutableLedgerDataPtr appliedLedgerState,
-                                     bool upgradeApplied) = 0;
+    virtual void completeLedgerClose(
+        uint32_t ledgerSeq, bool calledViaExternalize,
+        LedgerCloseData const& ledgerData,
+        ImmutableLedgerDataPtr appliedLedgerState, bool upgradeApplied,
+        std::vector<SorobanApplyMetrics>&& sorobanApplyMetrics) = 0;
 
     virtual void assertSetupPhase() const = 0;
 #ifdef BUILD_TESTS
@@ -398,7 +399,7 @@ class LedgerManager
 
     virtual void manuallyAdvanceLedgerHeader(LedgerHeader const& header) = 0;
 
-    virtual SorobanMetrics& getSorobanMetrics() = 0;
+    virtual SorobanMetricsRegistry& getSorobanMetrics() = 0;
     virtual ::rust::Box<rust_bridge::SorobanModuleCache> getModuleCache() = 0;
 
     virtual ~LedgerManager()

src/ledger/LedgerManagerImpl.cpp

@@ -958,7 +958,7 @@ LedgerManagerImpl::getSorobanInMemoryStateSizeForTesting()
 }
 #endif
 
-SorobanMetrics&
+SorobanMetricsRegistry&
 LedgerManagerImpl::getSorobanMetrics()
 {
     return mApplyState.getMetrics().mSorobanMetrics;
@@ -1198,7 +1198,8 @@ LedgerManagerImpl::ApplyState::maybeRebuildModuleCache(
 }
 
 void
-LedgerManagerImpl::publishSorobanMetrics()
+LedgerManagerImpl::publishSorobanMetrics(
+    std::vector<SorobanApplyMetrics>& sorobanApplyMetricsPerThread)
 {
     if (!hasLastClosedSorobanNetworkConfig())
     {
@@ -1235,8 +1236,13 @@ LedgerManagerImpl::publishSorobanMetrics()
         conf.sorobanStateTargetSizeBytes());
     m.mConfigFeeWrite1KB.set_count(conf.feeRent1KB());
 
-    // then publish the actual ledger usage
-    m.publishAndResetLedgerWideMetrics();
+    // then publish the actual ledger usage, merged into a single instance
+    auto& totalSorobanMetrics = sorobanApplyMetricsPerThread[0];
+    for (size_t i = 1; i < sorobanApplyMetricsPerThread.size(); ++i)
+    {
+        totalSorobanMetrics.merge(std::move(sorobanApplyMetricsPerThread[i]));
+    }
+    m.recordApplyMetrics(totalSorobanMetrics);
 }
 
 // called by txherder
@@ -1575,7 +1581,8 @@ void
 LedgerManagerImpl::completeLedgerClose(
     uint32_t ledgerSeq, bool calledViaExternalize,
     LedgerCloseData const& ledgerData,
-    ImmutableLedgerDataPtr appliedLedgerState, bool upgradeApplied)
+    ImmutableLedgerDataPtr appliedLedgerState, bool upgradeApplied,
+    std::vector<SorobanApplyMetrics>&& sorobanApplyMetricsPerThread)
 {
 #ifdef BUILD_TESTS
     if (mCompleteLedgerCloseOverride)
@@ -1598,7 +1605,7 @@ LedgerManagerImpl::completeLedgerClose(
 
     // We can publish Soroban metrics at any point after advancing the LCL
     // state.
-    publishSorobanMetrics();
+    publishSorobanMetrics(sorobanApplyMetricsPerThread);
 
     // Maybe kick off publishing on complete checkpoint files
     auto& hm = mApp.getHistoryManager();
@@ -1813,6 +1820,15 @@ LedgerManagerImpl::applyLedger(LedgerCloseData const& ledgerData,
 #endif
 
     TransactionResultSet txResultSet;
+    // Soroban apply metrics are collected by different worker threads, so
+    // in order to avoid synchronization we give every thread its own metrics
+    // container. At least one thread (the apply thread) is going to exist, so
+    // this starts at 1, and is extended on-demand before spawning more workers.
+    // The metrics set in each thread's slot should be considered to be
+    // arbitrary by the consumers - the metrics must commute, and in the end
+    // we merge all the metric containers together without worrying about their
+    // order or origin.
+    std::vector<SorobanApplyMetrics> sorobanApplyMetricsPerThread(1);
 #ifdef BUILD_TESTS
     if (mApp.getRunInOverlayOnlyMode())
     {
@@ -1854,8 +1870,9 @@ LedgerManagerImpl::applyLedger(LedgerCloseData const& ledgerData,
         // to use
         auto const mutableTxResults = processFeesSeqNums(
             *applicableTxSet, ltx, ledgerCloseMeta, ledgerData);
-        txResultSet = applyTransactions(*applicableTxSet, mutableTxResults, ltx,
-                                        ledgerCloseMeta);
+        txResultSet =
+            applyTransactions(*applicableTxSet, mutableTxResults, ltx,
+                              ledgerCloseMeta, sorobanApplyMetricsPerThread);
     }
 
     auto ledgerSeq = ltx.loadHeader().current().ledgerSeq;
@@ -2081,15 +2098,19 @@ LedgerManagerImpl::applyLedger(LedgerCloseData const& ledgerData,
     if (threadIsMain())
     {
         completeLedgerClose(ledgerSeq, calledViaExternalize, ledgerData,
-                            std::move(appliedLedgerState), upgradeApplied);
+                            std::move(appliedLedgerState), upgradeApplied,
+                            std::move(sorobanApplyMetricsPerThread));
     }
     else
     {
         auto cb = [this, ledgerSeq, calledViaExternalize, ledgerData,
                    appliedLedgerState = std::move(appliedLedgerState),
-                   upgradeApplied]() mutable {
+                   upgradeApplied,
+                   sorobanApplyMetricsPerThread =
+                       std::move(sorobanApplyMetricsPerThread)]() mutable {
             completeLedgerClose(ledgerSeq, calledViaExternalize, ledgerData,
-                                std::move(appliedLedgerState), upgradeApplied);
+                                std::move(appliedLedgerState), upgradeApplied,
+                                std::move(sorobanApplyMetricsPerThread));
         };
         mApp.postOnMainThread(std::move(cb), "completeLedgerClose");
     }
@@ -2604,26 +2625,18 @@ LedgerManagerImpl::applyThread(
     AppConnector& app,
     std::unique_ptr<ThreadParallelApplyLedgerState> threadState,
     Cluster const& cluster, Config const& config, ParallelLedgerInfo ledgerInfo,
-    Hash sorobanBasePrngSeed)
+    Hash sorobanBasePrngSeed, SorobanApplyMetrics& sorobanMetrics)
 {
     for (auto const& txBundle : cluster)
     {
-        // Apply timer; samples go into the thread's metrics batch and are
-        // published at ledger close.
-        std::optional<BatchedTimerScope> txTime;
-        if (!mApp.getConfig().DISABLE_SOROBAN_METRICS_FOR_TESTING)
-        {
-            txTime.emplace(getSorobanMetrics(),
-                           &SorobanMetrics::ApplyMetricsBatch::mTxApplyNsecs);
-        }
-
+        auto applyStart = std::chrono::steady_clock::now();
         Hash txSubSeed = subSha256(sorobanBasePrngSeed, txBundle.getTxNum());
 
         threadState->flushRoTTLBumpsInTxWriteFootprint(txBundle);
 
         auto res = txBundle.getTx()->parallelApply(
             app, *threadState, config, ledgerInfo, txBundle.getResPayload(),
-            getSorobanMetrics(), txSubSeed, txBundle.getEffects());
+            sorobanMetrics, txSubSeed, txBundle.getEffects());
 
         if (res)
         {
@@ -2633,6 +2646,10 @@ LedgerManagerImpl::applyThread(
         {
             releaseAssert(!txBundle.getResPayload().isSuccess());
         }
+        sorobanMetrics.mTxApplyNsecs.push_back(
+            std::chrono::duration_cast<std::chrono::nanoseconds>(
+                std::chrono::steady_clock::now() - applyStart)
+                .count());
     }
 
     threadState->flushRemainingRoTTLBumps();
@@ -2652,11 +2669,18 @@ LedgerManagerImpl::applySorobanStageClustersInParallel(
     AppConnector& app, ApplyStage const& stage,
     GlobalParallelApplyLedgerState const& globalState,
     Hash const& sorobanBasePrngSeed, Config const& config,
-    ParallelLedgerInfo const& ledgerInfo)
+    ParallelLedgerInfo const& ledgerInfo,
+    std::vector<SorobanApplyMetrics>& sorobanApplyMetricsPerThread)
 {
     ZoneScoped;
 
     DeactivateScopeGuard globalStateDeactivateGuard(globalState);
+    // Stages may contain a different number of clusters, so we ensure that
+    // there is a corresponding metrics entry for each cluster.
+    if (sorobanApplyMetricsPerThread.size() < stage.numClusters())
+    {
+        sorobanApplyMetricsPerThread.resize(stage.numClusters());
+    }
 
     std::vector<
         std::function<std::unique_ptr<ThreadParallelApplyLedgerState>()>>
@@ -2665,13 +2689,16 @@ LedgerManagerImpl::applySorobanStageClustersInParallel(
     for (size_t i = 0; i < stage.numClusters(); ++i)
     {
         tasks.emplace_back([this, &app, &globalState, &stage, i, &config,
-                            &ledgerInfo, &sorobanBasePrngSeed]() {
+                            &ledgerInfo, &sorobanBasePrngSeed,
+                            &sorobanApplyMetricsPerThread]() {
             auto const& cluster = stage.getCluster(i);
             auto threadStatePtr =
                 std::make_unique<ThreadParallelApplyLedgerState>(
                     app, globalState, cluster, i);
+            // Give every thread its own metrics entry to write to.
             return applyThread(app, std::move(threadStatePtr), cluster, config,
-                               ledgerInfo, sorobanBasePrngSeed);
+                               ledgerInfo, sorobanBasePrngSeed,
+                               sorobanApplyMetricsPerThread[i]);
         });
     }
 
@@ -2728,14 +2755,16 @@ void
 LedgerManagerImpl::applySorobanStage(
     AppConnector& app, LedgerHeader const& header,
     GlobalParallelApplyLedgerState& globalParState, ApplyStage const& stage,
-    Hash const& sorobanBasePrngSeed)
+    Hash const& sorobanBasePrngSeed,
+    std::vector<SorobanApplyMetrics>& sorobanApplyMetricsPerThread)
 {
     ZoneScoped;
     auto const& config = app.getConfig();
     auto ledgerInfo = getParallelLedgerInfo(app, header);
 
     auto threadStates = applySorobanStageClustersInParallel(
-        app, stage, globalParState, sorobanBasePrngSeed, config, ledgerInfo);
+        app, stage, globalParState, sorobanBasePrngSeed, config, ledgerInfo,
+        sorobanApplyMetricsPerThread);
 
     if (config.invariantsEnabled())
     {
@@ -2754,22 +2783,24 @@ LedgerManagerImpl::applySorobanStage(
 }
 
 void
-LedgerManagerImpl::applySorobanStages(AppConnector& app, AbstractLedgerTxn& ltx,
-                                      std::vector<ApplyStage> const& stages,
-                                      SorobanNetworkConfig const& sorobanConfig,
-                                      Hash const& sorobanBasePrngSeed)
+LedgerManagerImpl::applySorobanStages(
+    AppConnector& app, AbstractLedgerTxn& ltx,
+    std::vector<ApplyStage> const& stages,
+    SorobanNetworkConfig const& sorobanConfig, Hash const& sorobanBasePrngSeed,
+    std::vector<SorobanApplyMetrics>& sorobanApplyMetricsPerThread)
 {
     ZoneScoped;
     GlobalParallelApplyLedgerState globalParState(
         app, mApplyState.copyApplyLedgerView(), ltx, stages,
-        mApplyState.getInMemorySorobanState(), sorobanConfig);
+        mApplyState.getInMemorySorobanState(), sorobanConfig,
+        sorobanApplyMetricsPerThread[0]);
     // LedgerTxn is not passed into applySorobanStage, so there's no risk
     // of the header being updated while we apply the stages.
     auto const& header = ltx.loadHeader().current();
     for (auto const& stage : stages)
     {
         applySorobanStage(app, header, globalParState, stage,
-                          sorobanBasePrngSeed);
+                          sorobanBasePrngSeed, sorobanApplyMetricsPerThread);
     }
     globalParState.commitChangesToLedgerTxn(ltx);
 }
@@ -2837,7 +2868,8 @@ LedgerManagerImpl::applyTransactions(
     ApplicableTxSetFrame const& txSet,
     std::vector<MutableTxResultPtr> const& mutableTxResults,
     AbstractLedgerTxn& ltx,
-    std::unique_ptr<LedgerCloseMetaFrame> const& ledgerCloseMeta)
+    std::unique_ptr<LedgerCloseMetaFrame> const& ledgerCloseMeta,
+    std::vector<SorobanApplyMetrics>& sorobanApplyMetricsPerThread)
 {
     ZoneNamedN(txsZone, "applyTransactions", true);
     size_t numTxs = txSet.sizeTxTotal();
@@ -2896,7 +2928,8 @@ LedgerManagerImpl::applyTransactions(
                 releaseAssert(sorobanConfig.has_value());
                 applyParallelPhase(phase, applyStages, mutableTxResults, index,
                                    ltx, enableTxMeta, *sorobanConfig,
-                                   sorobanBasePrngSeed);
+                                   sorobanBasePrngSeed,
+                                   sorobanApplyMetricsPerThread);
             }
             catch (std::exception const& e)
             {
@@ -2914,7 +2947,7 @@ LedgerManagerImpl::applyTransactions(
             applySequentialPhase(phase, mutableTxResults, index, ltx,
                                  enableTxMeta, sorobanConfig,
                                  sorobanBasePrngSeed, ledgerCloseMeta,
-                                 txResultSet);
+                                 txResultSet, sorobanApplyMetricsPerThread[0]);
         }
     }
 
@@ -2942,7 +2975,8 @@ LedgerManagerImpl::applyParallelPhase(
     TxSetPhaseFrame const& phase, std::vector<stellar::ApplyStage>& applyStages,
     std::vector<stellar::MutableTxResultPtr> const& mutableTxResults,
     uint32_t& index, stellar::AbstractLedgerTxn& ltx, bool enableTxMeta,
-    SorobanNetworkConfig const& sorobanConfig, Hash const& sorobanBasePrngSeed)
+    SorobanNetworkConfig const& sorobanConfig, Hash const& sorobanBasePrngSeed,
+    std::vector<SorobanApplyMetrics>& sorobanApplyMetricsPerThread)
 {
     ZoneScoped;
 
@@ -2991,7 +3025,7 @@ LedgerManagerImpl::applyParallelPhase(
     }
 
     applySorobanStages(mApp.getAppConnector(), ltx, applyStages, sorobanConfig,
-                       sorobanBasePrngSeed);
+                       sorobanBasePrngSeed, sorobanApplyMetricsPerThread);
 
     // meta will be processed in processPostTxSetApply
 }
@@ -3004,7 +3038,7 @@ LedgerManagerImpl::applySequentialPhase(
     std::optional<SorobanNetworkConfig const> const& sorobanConfig,
     Hash const& sorobanBasePrngSeed,
     std::unique_ptr<LedgerCloseMetaFrame> const& ledgerCloseMeta,
-    TransactionResultSet& txResultSet)
+    TransactionResultSet& txResultSet, SorobanApplyMetrics& sorobanMetrics)
 {
     for (auto const& tx : phase)
     {
@@ -3040,7 +3074,7 @@ LedgerManagerImpl::applySequentialPhase(
         }
 
         tx->apply(mApp.getAppConnector(), ltx, tm, mutableTxResult,
-                  sorobanConfig, subSeed);
+                  sorobanConfig, subSeed, sorobanMetrics);
         tx->processPostApply(mApp.getAppConnector(), ltx, tm, mutableTxResult);
 
         tm.maybeSetRefundableFeeMeta(mutableTxResult.getRefundableFeeTracker());