Support reusable module cache with multithreaded pre-compilation

Author: graydon

docs/metrics.md

@@ -238,4 +238,8 @@ soroban.config.ledger-max-read-entry         | counter   | soroban config settin
 soroban.config.ledger-max-read-ledger-byte   | counter   | soroban config setting `ledger_max_read_bytes`
 soroban.config.ledger-max-write-entry        | counter   | soroban config setting `ledger_max_write_ledger_entries`
 soroban.config.ledger-max-write-ledger-byte  | counter   | soroban config setting `ledger_max_write_bytes`
-soroban.config.bucket-list-target-size-byte  | counter   | soroban config setting `bucket_list_target_size_bytes`
+soroban.config.bucket-list-target-size-byte  | counter   | soroban config setting `bucket_list_target_size_bytes`
+soroban.module-cache.num-entries             | counter   | current number of entries in module cache
+soroban.module-cache.compilation-time        | timer     | times each contract compilation when adding to module cache
+soroban.module-cache.rebuild-time            | timer     | times each rebuild of module cache (including all compilations)
+soroban.module-cache.rebuild-bytes           | counter   | bytes of WASM bytecode compiled in last rebuild of module cache

docs/stellar-core_example.cfg

@@ -382,6 +382,12 @@ CATCHUP_RECENT=0
 # merging and vertification.
 WORKER_THREADS=11
 
+# COMPILATION_THREADS (integer) default 6
+# Number of threads launched temporarily when compiling contracts at
+# startup. These are short lived, CPU-bound threads that are not
+# in competition with the worker threads.
+COMPILATION_THREADS=6
+
 # QUORUM_INTERSECTION_CHECKER (boolean) default true
 # Enable/disable computation of quorum intersection monitoring
 QUORUM_INTERSECTION_CHECKER=true

src/crypto/ByteSlice.h

@@ -4,6 +4,8 @@
 // under the Apache License, Version 2.0. See the COPYING file at the root
 // of this distribution or at http://www.apache.org/licenses/LICENSE-2.0
 
+#include "rust/RustBridge.h"
+#include <cstdint>
 #include <string>
 #include <vector>
 #include <xdrpp/message.h>
@@ -71,6 +73,13 @@ class ByteSlice
         : mData(bytes.data()), mSize(bytes.size())
     {
     }
+    ByteSlice(::rust::Vec<uint8_t> const& bytes)
+        : mData(bytes.data()), mSize(bytes.size())
+    {
+    }
+    ByteSlice(RustBuf const& bytes) : ByteSlice(bytes.data)
+    {
+    }
     ByteSlice(char const* str) : ByteSlice((void const*)str, strlen(str))
     {
     }

src/ledger/LedgerManager.h

@@ -7,6 +7,7 @@
 #include "catchup/LedgerApplyManager.h"
 #include "history/HistoryManager.h"
 #include "ledger/NetworkConfig.h"
+#include "rust/RustBridge.h"
 #include <memory>
 
 namespace stellar
@@ -200,6 +201,12 @@ class LedgerManager
     virtual void manuallyAdvanceLedgerHeader(LedgerHeader const& header) = 0;
 
     virtual SorobanMetrics& getSorobanMetrics() = 0;
+    virtual ::rust::Box<rust_bridge::SorobanModuleCache> getModuleCache() = 0;
+
+    // Compiles all contracts in the current ledger, for ledger protocols
+    // starting at minLedgerVersion and running through to
+    // Config::CURRENT_LEDGER_PROTOCOL_VERSION (to enable upgrades).
+    virtual void compileAllContractsInLedger(uint32_t minLedgerVersion) = 0;
 
     virtual ~LedgerManager()
     {

src/ledger/LedgerManagerImpl.cpp

@@ -20,12 +20,15 @@
 #include "history/HistoryManager.h"
 #include "ledger/FlushAndRotateMetaDebugWork.h"
 #include "ledger/LedgerHeaderUtils.h"
+#include "ledger/LedgerManager.h"
 #include "ledger/LedgerTxn.h"
 #include "ledger/LedgerTxnEntry.h"
 #include "ledger/LedgerTxnHeader.h"
+#include "ledger/SharedModuleCacheCompiler.h"
 #include "main/Application.h"
 #include "main/Config.h"
 #include "main/ErrorMessages.h"
+#include "rust/RustBridge.h"
 #include "transactions/MutableTransactionResult.h"
 #include "transactions/OperationFrame.h"
 #include "transactions/TransactionFrameBase.h"
@@ -41,8 +44,10 @@
 #include "util/XDRCereal.h"
 #include "util/XDRStream.h"
 #include "work/WorkScheduler.h"
+#include "xdr/Stellar-ledger-entries.h"
 #include "xdrpp/printer.h"
 
+#include <cstdint>
 #include <fmt/format.h>
 
 #include "xdr/Stellar-ledger-entries.h"
@@ -58,6 +63,8 @@
 #include <Tracy.hpp>
 
 #include <chrono>
+#include <memory>
+#include <mutex>
 #include <optional>
 #include <regex>
 #include <sstream>
@@ -125,6 +132,27 @@ LedgerManager::ledgerAbbrev(LedgerHeaderHistoryEntry const& he)
     return ledgerAbbrev(he.header, he.hash);
 }
 
+static std::vector<uint32_t>
+getModuleCacheProtocols()
+{
+    std::vector<uint32_t> ledgerVersions;
+    for (uint32_t i = (uint32_t)REUSABLE_SOROBAN_MODULE_CACHE_PROTOCOL_VERSION;
+         i <= Config::CURRENT_LEDGER_PROTOCOL_VERSION; i++)
+    {
+        ledgerVersions.push_back(i);
+    }
+    auto extra = getenv("SOROBAN_TEST_EXTRA_PROTOCOL");
+    if (extra)
+    {
+        uint32_t proto = static_cast<uint32_t>(atoi(extra));
+        if (proto > 0)
+        {
+            ledgerVersions.push_back(proto);
+        }
+    }
+    return ledgerVersions;
+}
+
 LedgerManagerImpl::LedgerManagerImpl(Application& app)
     : mApp(app)
     , mSorobanMetrics(app.getMetrics())
@@ -157,6 +185,8 @@ LedgerManagerImpl::LedgerManagerImpl(Application& app)
     , mCatchupDuration(
           app.getMetrics().NewTimer({"ledger", "catchup", "duration"}))
     , mState(LM_BOOTING_STATE)
+    , mModuleCache(::rust_bridge::new_module_cache())
+    , mModuleCacheProtocols(getModuleCacheProtocols())
 
 {
     setupLedgerCloseMetaStream();
@@ -405,6 +435,9 @@ LedgerManagerImpl::loadLastKnownLedger(bool restoreBucketlist)
         updateNetworkConfig(ltx);
         mSorobanNetworkConfigReadOnly = mSorobanNetworkConfigForApply;
     }
+
+    // Prime module cache with ledger content.
+    compileAllContractsInLedger(latestLedgerHeader->ledgerVersion);
 }
 
 Database&
@@ -560,6 +593,118 @@ LedgerManagerImpl::getSorobanMetrics()
     return mSorobanMetrics;
 }
 
+::rust::Box<rust_bridge::SorobanModuleCache>
+LedgerManagerImpl::getModuleCache()
+{
+    std::lock_guard<std::recursive_mutex> guard(mLedgerStateMutex);
+    finishAnyPendingCompilation();
+    return mModuleCache->shallow_clone();
+}
+
+void
+LedgerManagerImpl::finishAnyPendingCompilation()
+{
+    std::lock_guard<std::recursive_mutex> guard(mLedgerStateMutex);
+    if (mCompiler)
+    {
+        auto newCache = mCompiler->wait();
+        mSorobanMetrics.mModuleCacheRebuildBytes.set_count(
+            (int64)mCompiler->getBytesCompiled());
+        mSorobanMetrics.mModuleCacheNumEntries.set_count(
+            (int64)mCompiler->getContractsCompiled());
+        mSorobanMetrics.mModuleCacheRebuildTime.Update(
+            mCompiler->getCompileTime());
+        mModuleCache.swap(newCache);
+        mCompiler.reset();
+        mApp.getAppConnector().setModuleCache(mModuleCache->shallow_clone());
+    }
+}
+
+void
+LedgerManagerImpl::compileAllContractsInLedger(uint32_t minLedgerVersion)
+{
+    startCompilingAllContracts(minLedgerVersion);
+    finishAnyPendingCompilation();
+}
+
+void
+LedgerManagerImpl::startCompilingAllContracts(uint32_t minLedgerVersion)
+{
+    std::lock_guard<std::recursive_mutex> guard(mLedgerStateMutex);
+    // Always stop a previous compilation before starting a new one. Can only
+    // have one running at any time.
+    finishAnyPendingCompilation();
+    std::vector<uint32_t> versions;
+    for (auto const& v : mModuleCacheProtocols)
+    {
+        if (v >= minLedgerVersion)
+        {
+            versions.push_back(v);
+        }
+    }
+    mCompiler = std::make_unique<SharedModuleCacheCompiler>(mApp, versions);
+    mCompiler->start();
+}
+
+void
+LedgerManagerImpl::maybeRebuildModuleCache(uint32_t minLedgerVersion)
+{
+    std::lock_guard<std::recursive_mutex> guard(mLedgerStateMutex);
+    // There is (currently) a grow-only arena underlying the module cache, so as
+    // entries are uploaded and evicted that arena will still grow. To cap this
+    // growth, we periodically rebuild the module cache from scratch.
+    //
+    // We could pick various size caps, but we want to avoid rebuilding
+    // spuriously when there just happens to be "a fairly large" cache due to
+    // having a fairly large live BL. I.e. we want to allow it to get as big as
+    // we can -- or as big as the "natural" BL-limits-dictated size -- while
+    // still rebuilding fairly often in DoS-attempt scenarios or just generally
+    // if there's regular upload/expiry churn that would otherwise cause
+    // unbounded growth.
+    //
+    // Unfortunately we do not know exactly how much memory is used by each byte
+    // of contract we compile, and the size estimates from the cost model have
+    // to assume a worst case which is almost a factor of _40_ larger than the
+    // byte-size of the contracts. So for example if we assume 100MB of
+    // contracts, the cost model says we ought to budget for 4GB of memory, just
+    // in case _all 100MB of contracts_ are "the worst case contract" that's
+    // just a continuous stream of function definitions.
+    //
+    // So: we take this multiplier, times the size of the contracts we _last_
+    // drew from the BL when doing a full recompile, times two, as a cap on the
+    // _current_ (post-rebuild, currently-growing) cache's budget-tracked
+    // memory. This should avoid rebuilding spuriously, while still treating
+    // events that double the size of the contract-set in the live BL as an
+    // event that warrants a rebuild.
+
+    // We try to fish the current cost multiplier out of the soroban network
+    // config's memory cost model, but fall back to a conservative default in
+    // case there is no mem cost param for VmInstantiation (This should never
+    // happen but just in case).
+    uint64_t linearTerm = 5000;
+
+    // linearTerm is in 1/128ths in the cost model, to reduce rounding error.
+    uint64_t scale = 128;
+
+    auto const& cfg = getSorobanNetworkConfigForApply();
+    auto const& memParams = cfg.memCostParams();
+    if (memParams.size() > (size_t)stellar::VmInstantiation)
+    {
+        auto const& param = memParams[(size_t)stellar::VmInstantiation];
+        linearTerm = param.linearTerm;
+    }
+    auto lastBytesCompiled = mSorobanMetrics.mModuleCacheRebuildBytes.count();
+    uint64_t limit = 2 * lastBytesCompiled * linearTerm / scale;
+    if (mModuleCache->get_mem_bytes_consumed() > limit)
+    {
+        CLOG_DEBUG(Ledger,
+                   "Rebuilding module cache: worst-case estimate {} "
+                   "model-bytes consumed of {} limit",
+                   mModuleCache->get_mem_bytes_consumed(), limit);
+        startCompilingAllContracts(minLedgerVersion);
+    }
+}
+
 void
 LedgerManagerImpl::publishSorobanMetrics()
 {
@@ -803,6 +948,9 @@ LedgerManagerImpl::closeLedger(LedgerCloseData const& ledgerData,
         return;
     }
 
+    // Complete any pending wasm-module-compilation before closing the ledger.
+    finishAnyPendingCompilation();
+
 #ifdef BUILD_TESTS
     mLastLedgerTxMeta.clear();
 #endif
@@ -1149,11 +1297,14 @@ LedgerManagerImpl::setLastClosedLedger(
     advanceLedgerPointers(advanceLedgerStateSnapshot(lastClosed.header, has));
 
     LedgerTxn ltx2(mApp.getLedgerTxnRoot());
-    if (protocolVersionStartsFrom(ltx2.loadHeader().current().ledgerVersion,
-                                  SOROBAN_PROTOCOL_VERSION))
+    auto lv = ltx2.loadHeader().current().ledgerVersion;
+    if (protocolVersionStartsFrom(lv, SOROBAN_PROTOCOL_VERSION))
     {
-        mApp.getLedgerManager().updateNetworkConfig(ltx2);
+        updateNetworkConfig(ltx2);
     }
+    // This should not be additionally conditionalized on lv >= anything,
+    // since we want to support SOROBAN_TEST_EXTRA_PROTOCOL > lv.
+    compileAllContractsInLedger(lv);
 }
 
 void
@@ -1804,6 +1955,8 @@ LedgerManagerImpl::transferLedgerEntriesToBucketList(
                 ledgerCloseMeta->populateEvictedEntries(evictedState);
             }
 
+            evictFromModuleCache(lh.ledgerVersion, evictedState);
+
             ltxEvictions.commit();
         }
 
@@ -1814,6 +1967,8 @@ LedgerManagerImpl::transferLedgerEntriesToBucketList(
     ltx.getAllEntries(initEntries, liveEntries, deadEntries);
     if (blEnabled)
     {
+        addAnyContractsToModuleCache(lh.ledgerVersion, initEntries);
+        addAnyContractsToModuleCache(lh.ledgerVersion, liveEntries);
         mApp.getBucketManager().addLiveBatch(mApp, lh, initEntries, liveEntries,
                                              deadEntries);
     }
@@ -1872,6 +2027,75 @@ LedgerManagerImpl::ledgerClosed(
         res = advanceLedgerStateSnapshot(lh, has);
     });
 
+    if (protocolVersionStartsFrom(
+            initialLedgerVers, REUSABLE_SOROBAN_MODULE_CACHE_PROTOCOL_VERSION))
+    {
+        maybeRebuildModuleCache(initialLedgerVers);
+    }
+
     return res;
 }
+
+void
+LedgerManagerImpl::evictFromModuleCache(uint32_t ledgerVersion,
+                                        EvictedStateVectors const& evictedState)
+{
+    std::vector<Hash> keys;
+    for (auto const& key : evictedState.deletedKeys)
+    {
+        if (key.type() == CONTRACT_CODE)
+        {
+            keys.emplace_back(key.contractCode().hash);
+        }
+    }
+    for (auto const& entry : evictedState.archivedEntries)
+    {
+        if (entry.data.type() == CONTRACT_CODE)
+        {
+            Hash const& hash = entry.data.contractCode().hash;
+            keys.emplace_back(hash);
+        }
+    }
+    if (keys.size() > 0)
+    {
+        CLOG_DEBUG(Ledger, "evicting {} modules from module cache",
+                   keys.size());
+        for (auto const& hash : keys)
+        {
+            CLOG_DEBUG(Ledger, "evicting {} from module cache", binToHex(hash));
+            ::rust::Slice<uint8_t const> slice{hash.data(), hash.size()};
+            mModuleCache->evict_contract_code(slice);
+            mSorobanMetrics.mModuleCacheNumEntries.dec();
+        }
+    }
+}
+
+void
+LedgerManagerImpl::addAnyContractsToModuleCache(
+    uint32_t ledgerVersion, std::vector<LedgerEntry> const& le)
+{
+    for (auto const& e : le)
+    {
+        if (e.data.type() == CONTRACT_CODE)
+        {
+            for (auto const& v : mModuleCacheProtocols)
+            {
+                if (v >= ledgerVersion)
+                {
+                    auto const& wasm = e.data.contractCode().code;
+                    CLOG_DEBUG(Ledger,
+                               "compiling wasm {} for protocol {} module cache",
+                               binToHex(sha256(wasm)), v);
+                    auto slice =
+                        rust::Slice<const uint8_t>(wasm.data(), wasm.size());
+                    mSorobanMetrics.mModuleCacheNumEntries.inc();
+                    auto timer =
+                        mSorobanMetrics.mModuleCompilationTime.TimeScope();
+                    mModuleCache->compile(v, slice);
+                }
+            }
+        }
+    }
+}
+
 }