Introduce a thread pool for batch execution of parallel CPU-bound tasks.

This introduced a new `BatchExecutor` class that uses a pool of physical CPU-pinned (if possible) workers to execute CPU-heavy workloads in parallel. Currently this is only used for the Soroban apply stage (which is already parallel), but in the followups we'll be able to use this for more apply flow stages (pre/post apply), as well as tasks outside of apply (like parallel tx set building).

Pinning the threads helps a lot with making per-thread runtime more even on machines with hyper-threading (which is probably pretty much every CPU out there). Pooling the threads is less impactful, but it helps a bit with the allocator churn.

Author: dmkozh

Builds/VisualStudio/stellar-core.vcxproj

@@ -626,6 +626,7 @@ exit /b 0
     <ClCompile Include="..\..\src\main\AppConnector.cpp" />
     <ClCompile Include="..\..\src\main\BannedAccountsPersistor.cpp" />
     <ClCompile Include="..\..\src\main\Diagnostics.cpp" />
+    <ClCompile Include="..\..\src\main\NtpProbe.cpp" />
     <ClCompile Include="..\..\src\main\QueryServer.cpp" />
     <ClCompile Include="..\..\src\main\SettingsUpgradeUtils.cpp" />
     <ClCompile Include="..\..\src\main\test\ApplicationUtilsTests.cpp" />
@@ -759,6 +760,7 @@ exit /b 0
     <ClCompile Include="..\..\src\transactions\TransactionUtils.cpp" />
     <ClCompile Include="..\..\src\transactions\TrustFlagsOpFrameBase.cpp" />
     <ClCompile Include="..\..\src\util\Backtrace.cpp" />
+    <ClCompile Include="..\..\src\util\BatchExecutor.cpp" />
     <ClCompile Include="..\..\src\util\BinaryFuseFilter.cpp" />
     <ClCompile Include="..\..\src\util\DebugMetaUtils.cpp" />
     <ClCompile Include="..\..\src\util\FileSystemException.cpp" />
@@ -1092,6 +1094,7 @@ exit /b 0
     <ClInclude Include="..\..\src\main\AppConnector.h" />
     <ClInclude Include="..\..\src\main\BannedAccountsPersistor.h" />
     <ClInclude Include="..\..\src\main\Diagnostics.h" />
+    <ClInclude Include="..\..\src\main\NtpProbe.h" />
     <ClInclude Include="..\..\src\main\QueryServer.h" />
     <ClInclude Include="..\..\src\main\SettingsUpgradeUtils.h" />
     <ClInclude Include="..\..\src\overlay\BanManager.h" />
@@ -1182,6 +1185,7 @@ exit /b 0
     <ClInclude Include="..\..\src\transactions\TransactionUtils.h" />
     <ClInclude Include="..\..\src\transactions\TrustFlagsOpFrameBase.h" />
     <ClInclude Include="..\..\src\util\Backtrace.h" />
+    <ClInclude Include="..\..\src\util\BatchExecutor.h" />
     <ClInclude Include="..\..\src\util\BinaryFuseFilter.h" />
     <ClInclude Include="..\..\src\util\BufferedAsioCerealOutputArchive.h" />
     <ClInclude Include="..\..\src\util\DebugMetaUtils.h" />

Builds/VisualStudio/stellar-core.vcxproj.filters

@@ -462,9 +462,6 @@
     <ClCompile Include="..\..\src\work\test\WorkTests.cpp">
       <Filter>work\tests</Filter>
     </ClCompile>
-    <ClCompile Include="..\..\src\test\fuzz\FuzzMain.cpp">
-      <Filter>test\fuzz</Filter>
-    </ClCompile>
     <ClCompile Include="..\..\src\test\fuzz\FuzzRegressionTests.cpp">
       <Filter>test\fuzz</Filter>
     </ClCompile>
@@ -1459,6 +1456,12 @@
     <ClCompile Include="..\..\src\test\CovMark.cpp">
       <Filter>test</Filter>
     </ClCompile>
+    <ClCompile Include="..\..\src\util\BatchExecutor.cpp">
+      <Filter>util</Filter>
+    </ClCompile>
+    <ClCompile Include="..\..\src\main\NtpProbe.cpp">
+      <Filter>main</Filter>
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <ClInclude Include="..\..\lib\util\cpptoml.h">
@@ -2590,6 +2593,12 @@
     <ClInclude Include="..\..\src\test\CovMark.h">
       <Filter>test</Filter>
     </ClInclude>
+    <ClInclude Include="..\..\src\util\BatchExecutor.h">
+      <Filter>util</Filter>
+    </ClInclude>
+    <ClInclude Include="..\..\src\main\NtpProbe.h">
+      <Filter>main</Filter>
+    </ClInclude>
   </ItemGroup>
   <ItemGroup>
     <None Include="..\..\COPYING" />

src/ledger/LedgerManagerImpl.cpp

@@ -40,6 +40,7 @@
 #include "transactions/TransactionFrameBase.h"
 #include "transactions/TransactionMeta.h"
 #include "transactions/TransactionUtils.h"
+#include "util/BatchExecutor.h"
 #include "util/DebugMetaUtils.h"
 #include "util/Decoder.h"
 #include "util/Fs.h"
@@ -2639,42 +2640,37 @@ LedgerManagerImpl::applySorobanStageClustersInParallel(
 {
     ZoneScoped;
 
-    std::vector<std::unique_ptr<ThreadParallelApplyLedgerState>> threadStates;
-    std::vector<std::future<std::unique_ptr<ThreadParallelApplyLedgerState>>>
-        threadFutures;
-
     DeactivateScopeGuard globalStateDeactivateGuard(globalState);
 
+    std::vector<
+        std::function<std::unique_ptr<ThreadParallelApplyLedgerState>()>>
+        tasks;
+    tasks.reserve(stage.numClusters());
     for (size_t i = 0; i < stage.numClusters(); ++i)
     {
-        auto const& cluster = stage.getCluster(i);
-        auto threadStatePtr = std::make_unique<ThreadParallelApplyLedgerState>(
-            app, globalState, cluster, i);
-        threadFutures.emplace_back(std::async(
-            std::launch::async, &LedgerManagerImpl::applyThread, this,
-            std::ref(app), std::move(threadStatePtr), std::cref(cluster),
-            std::cref(config), ledgerInfo, sorobanBasePrngSeed));
+        tasks.emplace_back([this, &app, &globalState, &stage, i, &config,
+                            &ledgerInfo, &sorobanBasePrngSeed]() {
+            auto const& cluster = stage.getCluster(i);
+            auto threadStatePtr =
+                std::make_unique<ThreadParallelApplyLedgerState>(
+                    app, globalState, cluster, i);
+            return applyThread(app, std::move(threadStatePtr), cluster, config,
+                               ledgerInfo, sorobanBasePrngSeed);
+        });
     }
 
-    for (auto& threadFuture : threadFutures)
+    try
     {
-        releaseAssert(threadFuture.valid());
-        try
-        {
-            auto futureResult = threadFuture.get();
-            threadStates.emplace_back(std::move(futureResult));
-        }
-        catch (std::exception const& e)
-        {
-            printErrorAndAbort("Exception on apply thread: ", e.what());
-        }
-        catch (...)
-        {
-            printErrorAndAbort("Unknown exception on apply thread");
-        }
+        return app.getBatchExecutor().executeBatch(std::move(tasks));
+    }
+    catch (std::exception const& e)
+    {
+        printErrorAndAbort("Exception on apply thread: ", e.what());
+    }
+    catch (...)
+    {
+        printErrorAndAbort("Unknown exception on apply thread");
     }
-    threadFutures.clear();
-    return threadStates;
 }
 
 void

src/main/AppConnector.cpp

@@ -169,6 +169,12 @@ AppConnector::threadIsType(Application::ThreadType type) const
     return mApp.threadIsType(type);
 }
 
+BatchExecutor&
+AppConnector::getBatchExecutor()
+{
+    return mApp.getBatchExecutor();
+}
+
 ImmutableLedgerView
 AppConnector::copyImmutableLedgerView()
 {

src/main/AppConnector.h

@@ -22,6 +22,7 @@ class SorobanMetrics;
 class SearchableHotArchiveBucketListSnapshot;
 struct LedgerTxnDelta;
 class CapacityTrackedMessage;
+class BatchExecutor;
 
 // Helper class to isolate access to Application; all function helpers must
 // either be called from main or be thread-safe
@@ -68,6 +69,7 @@ class AppConnector
     checkScheduledAndCache(std::shared_ptr<CapacityTrackedMessage> msgTracker);
     SorobanNetworkConfig const& getLastClosedSorobanNetworkConfig() const;
     bool threadIsType(Application::ThreadType type) const;
+    BatchExecutor& getBatchExecutor();
 
     MetricsRegistry& getMetrics() const;
 

src/main/Application.h

@@ -37,6 +37,7 @@ class WorkScheduler;
 class BanManager;
 class BannedAccountsPersistor;
 class StatusManager;
+class BatchExecutor;
 class AbstractLedgerTxnParent;
 class BasicWork;
 enum class LoadGenMode;
@@ -270,6 +271,13 @@ class Application
     virtual void postOnLedgerCloseThread(std::function<void()>&& f,
                                          std::string jobName) = 0;
 
+    // Get the shared executor for running batches of CPU-bound tasks in
+    // parallel. This is mostly used in the apply path, though it may also be
+    // used for other CPU-bound tasks, as long as they don't overlap with the
+    // apply path (e.g. building the transaction set for the next ledger).
+    // Batches are blocking and must be run one at a time.
+    virtual BatchExecutor& getBatchExecutor() = 0;
+
     // Perform actions necessary to transition from BOOTING_STATE to other
     // states. In particular: either reload or reinitialize the database, and
     // either restart or begin reacquiring SCP consensus (as instructed by

src/main/ApplicationImpl.cpp

@@ -53,6 +53,7 @@
 #include "overlay/OverlayManagerImpl.h"
 #include "process/ProcessManager.h"
 #include "transactions/SignatureChecker.h"
+#include "util/BatchExecutor.h"
 #include "util/GlobalChecks.h"
 #include "util/JitterInjection.h"
 #include "util/LogSlowExecution.h"
@@ -109,6 +110,7 @@ ApplicationImpl::ApplicationImpl(VirtualClock& clock, Config const& cfg)
               : nullptr)
     , mWorkerThreads()
     , mEvictionThread()
+    , mBatchExecutor(std::make_unique<BatchExecutor>())
     , mStopSignals(clock.getIOContext(), SIGINT)
     , mStarted(false)
     , mStopping(false)
@@ -1541,6 +1543,13 @@ ApplicationImpl::getLedgerCloseIOContext()
     return *mLedgerCloseIOContext;
 }
 
+BatchExecutor&
+ApplicationImpl::getBatchExecutor()
+{
+    releaseAssert(mBatchExecutor);
+    return *mBatchExecutor;
+}
+
 void
 ApplicationImpl::postOnMainThread(std::function<void()>&& f, std::string&& name,
                                   Scheduler::ActionType type)

src/main/ApplicationImpl.h

@@ -90,6 +90,8 @@ class ApplicationImpl : public Application
     virtual asio::io_context& getOverlayIOContext() override;
     virtual asio::io_context& getLedgerCloseIOContext() override;
 
+    virtual BatchExecutor& getBatchExecutor() override;
+
     virtual void postOnMainThread(std::function<void()>&& f, std::string&& name,
                                   Scheduler::ActionType type) override;
     virtual void postOnBackgroundThread(std::function<void()>&& f,
@@ -239,6 +241,8 @@ class ApplicationImpl : public Application
     // thread for eviction scans.
     std::unique_ptr<std::thread> mEvictionThread;
 
+    std::unique_ptr<BatchExecutor> mBatchExecutor;
+
     // NOTE: It is important that this map not be updated outside of the
     // constructor. `unordered_map` is safe for multiple threads to read from,
     // so long as there are no concurrent writers.

src/transactions/ParallelApplyUtils.cpp

@@ -308,7 +308,8 @@ GlobalParallelApplyLedgerState::GlobalParallelApplyLedgerState(
                          mInMemorySorobanState.getLedgerSeq());
     releaseAssertOrThrow(ltx.getHeader().ledgerSeq ==
                          mLCLApplyView.getLedgerSeq() + 1);
-
+    releaseAssert(threadIsMain() ||
+                  app.threadIsType(Application::ThreadType::APPLY));
     // From now on, we will be using globalState, liveSnapshots, and the
     // hotArchive to collect all entries. Before we continue though, we need to
     // load into the globalEntryMap any classic entries that have been modified
@@ -327,9 +328,6 @@ GlobalParallelApplyLedgerState::
         AppConnector& app, AbstractLedgerTxn& ltx,
         std::vector<ApplyStage> const& stages)
 {
-    releaseAssert(threadIsMain() ||
-                  app.threadIsType(Application::ThreadType::APPLY));
-
     auto fetchInMemoryClassicEntries =
         [&](xdr::xvector<LedgerKey> const& keys) {
             for (auto const& lk : keys)
@@ -564,9 +562,6 @@ ThreadParallelApplyLedgerState::collectClusterFootprintEntriesFromGlobal(
     AppConnector& app, GlobalParallelApplyLedgerState const& global,
     Cluster const& cluster)
 {
-    releaseAssert(threadIsMain() ||
-                  app.threadIsType(Application::ThreadType::APPLY));
-
     // As part of the initialization of this thread state, we need to
     // collect all the keys that are in the global state map. For any keys
     // we need not in the global state, we will fetch them from the live
@@ -751,7 +746,6 @@ ThreadParallelApplyLedgerState::upsertEntry(
 void
 ThreadParallelApplyLedgerState::eraseEntry(LedgerKey const& key)
 {
-
     auto parAppEntry =
         ThreadParallelApplyEntry::dirty(scopeAdoptEntryOpt(std::nullopt));
     mThreadEntryMap.insert_or_assign(key, parAppEntry);