diff --git a/include/xrpl/basics/TaggedCache.h b/include/xrpl/basics/TaggedCache.h index 2eeb4996e72..afc91954138 100644 --- a/include/xrpl/basics/TaggedCache.h +++ b/include/xrpl/basics/TaggedCache.h @@ -18,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -74,11 +75,25 @@ class TaggedCache using shared_pointer_type = SharedPointerType; public: + /** + * A byte budget for the strongly-cached entries. Each entry is charged + * `cost(ptr)` bytes when it becomes strong; growth past `bytes` evicts + * like the entry cap. The cost should include the value's heap payload + * plus per-entry overhead. + */ + struct ByteBudget + { + std::size_t bytes = 0; + std::function cost; + }; + /** * @param cacheHardCap When positive, a hard upper bound on the number of * strongly-cached entries, enforced by demoting the approximately * oldest entry whenever growth would exceed it. 0 disables the cap * (the periodic sweep alone bounds the cache). + * @param byteBudget When set, a hard upper bound on the charged bytes of + * strongly-cached entries, enforced the same way. */ TaggedCache( std::string const& name, @@ -87,7 +102,8 @@ class TaggedCache clock_type& clock, beast::Journal journal, beast::insight::Collector::ptr const& collector = beast::insight::NullCollector::make(), - int cacheHardCap = 0); + int cacheHardCap = 0, + std::optional byteBudget = std::nullopt); public: /** @@ -108,6 +124,13 @@ class TaggedCache int getTrackSize() const; + /** + * Returns the charged bytes of strongly-cached entries; 0 unless a + * byte budget with a cost function is configured. + */ + std::size_t + getCacheBytes() const; + float getHitRate(); @@ -322,6 +345,10 @@ class TaggedCache shared_weak_combo_pointer_type ptr; clock_type::time_point lastAccess; + // Bytes charged against the byte budget while strong; 0 when weak + // or when no budget is configured. + std::uint32_t costBytes{0}; + ValueEntry(clock_type::time_point const& lastAccess, shared_pointer_type const& ptr) : ptr(ptr), lastAccess(lastAccess) { @@ -378,6 +405,7 @@ class TaggedCache KeyValueCacheType::map_type& partition, SweptPointersVector& stuffToSweep, std::atomic& allRemovals, + std::atomic& allBytesRemoved, std::scoped_lock const&); [[nodiscard]] std::thread @@ -387,6 +415,7 @@ class TaggedCache KeyOnlyCacheType::map_type& partition, SweptPointersVector&, std::atomic& allRemovals, + std::atomic& allBytesRemoved, std::scoped_lock const&); beast::Journal journal_; @@ -409,6 +438,13 @@ class TaggedCache // weak-to-strong revivals). 0 disables it (sweep-only sizing). int const cacheHardCap_; + // Byte-denominated bound on strongly-cached entries, enforced the same + // way; each entry is charged by the budget's cost function while strong. + std::optional const byteBudget_; + + // Charged bytes of strongly-cached entries (under mutex_). + std::size_t cacheBytes_{0}; + // Total hard-cap evictions (under mutex_); the first marks saturation // onset for logging. std::uint64_t hardCapEvictions_{0}; @@ -416,6 +452,17 @@ class TaggedCache // Number of items cached int cacheCount_{0}; + // Charge or release an entry's bytes against the byte budget. No-ops + // without a configured budget; callers hold mutex_. + void + chargeEntry(ValueEntry& entry, SharedPointerType const& ptr); + void + dischargeEntry(ValueEntry& entry); + + // True when either the entry cap or the byte budget is exceeded. + [[nodiscard]] bool + overHardCap() const; + // Rotating bucket cursor for evictForHardCap so successive over-cap // evictions sweep the whole partition (CLOCK hand) instead of repeatedly // sampling the head buckets. Advanced under mutex_. diff --git a/include/xrpl/basics/TaggedCache.ipp b/include/xrpl/basics/TaggedCache.ipp index 55a6f986c7e..b882d78bd4a 100644 --- a/include/xrpl/basics/TaggedCache.ipp +++ b/include/xrpl/basics/TaggedCache.ipp @@ -6,6 +6,7 @@ #include #include +#include namespace xrpl { @@ -58,7 +59,8 @@ inline TaggedCache< clock_type& clock, beast::Journal journal, beast::insight::Collector::ptr const& collector, - int cacheHardCap) + int cacheHardCap, + std::optional byteBudget) : journal_(journal) , clock_(clock) , stats_( @@ -69,9 +71,65 @@ inline TaggedCache< , targetSize_(size) , targetAge_(expiration) , cacheHardCap_(cacheHardCap) + , byteBudget_(std::move(byteBudget)) { } +template < + class Key, + class T, + bool IsKeyCache, + class SharedWeakUnionPointer, + class SharedPointerType, + class Hash, + class KeyEqual, + class Mutex> +inline void +TaggedCache:: + chargeEntry(ValueEntry& entry, SharedPointerType const& ptr) +{ + if (byteBudget_ && byteBudget_->cost) + { + entry.costBytes = static_cast(std::min( + byteBudget_->cost(ptr), std::numeric_limits::max())); + cacheBytes_ += entry.costBytes; + } +} + +template < + class Key, + class T, + bool IsKeyCache, + class SharedWeakUnionPointer, + class SharedPointerType, + class Hash, + class KeyEqual, + class Mutex> +inline void +TaggedCache:: + dischargeEntry(ValueEntry& entry) +{ + cacheBytes_ -= entry.costBytes; + entry.costBytes = 0; +} + +template < + class Key, + class T, + bool IsKeyCache, + class SharedWeakUnionPointer, + class SharedPointerType, + class Hash, + class KeyEqual, + class Mutex> +inline bool +TaggedCache:: + overHardCap() const +{ + return (cacheHardCap_ > 0 && cacheCount_ > cacheHardCap_) || + (byteBudget_ && cacheBytes_ > byteBudget_->bytes); +} + template < class Key, class T, @@ -122,6 +180,23 @@ TaggedCache +inline std::size_t +TaggedCache:: + getCacheBytes() const +{ + std::scoped_lock const lock(mutex_); + return cacheBytes_; +} + template < class Key, class T, @@ -173,6 +248,7 @@ TaggedCache(bucketCount, 4 * kEvictSampleBudget); - for (int demotions = 0; cacheCount_ > cacheHardCap_ && demotions < kMaxDemotionsPerCall; - ++demotions) + for (int demotions = 0; overHardCap() && demotions < kMaxDemotionsPerCall; ++demotions) { int sampled = 0; std::size_t bucketsWalked = 0; @@ -292,6 +368,7 @@ TaggedCachesecond.isWeak()) return; + dischargeEntry(oldest->second); if (oldest->second.ptr.useCount() == 1) { // Sole owner: release entirely. @@ -362,16 +439,24 @@ TaggedCache workers; workers.reserve(cache_.partitions()); std::atomic allRemovals = 0; + std::atomic allBytesRemoved = 0; for (std::size_t p = 0; p < cache_.partitions(); ++p) { workers.push_back(sweepHelper( - whenExpire, now, cache_.map()[p], allStuffToSweep[p], allRemovals, lock)); + whenExpire, + now, + cache_.map()[p], + allStuffToSweep[p], + allRemovals, + allBytesRemoved, + lock)); } for (std::thread& worker : workers) worker.join(); cacheCount_ -= allRemovals; + cacheBytes_ -= std::min(allBytesRemoved, cacheBytes_); } // At this point allStuffToSweep will go out of scope outside the lock // and decrement the reference count on each strong pointer. @@ -410,6 +495,7 @@ TaggedCachesecond, data); // The just-inserted entry is the newest; evictForHardCap skips it // and drops the oldest in its partition. - if (cacheHardCap_ > 0 && cacheCount_ > cacheHardCap_) + if (overHardCap()) evictForHardCap(*emplacedIt.ait, emplacedIt.mit); return false; } @@ -494,7 +581,9 @@ TaggedCache 0 && cacheCount_ > cacheHardCap_) + chargeEntry(entry, entry.ptr.getStrong()); + if (overHardCap()) evictForHardCap(*cit.ait, cit.mit); return true; } entry.ptr = data; ++cacheCount_; - if (cacheHardCap_ > 0 && cacheCount_ > cacheHardCap_) + chargeEntry(entry, data); + if (overHardCap()) evictForHardCap(*cit.ait, cit.mit); return false; @@ -802,8 +893,17 @@ TaggedCachesecond, it.mit->second.ptr.getStrong()); + if (overHardCap()) + evictForHardCap(*it.ait, it.mit); + } + else + { it->second.touch(clock_.now()); + } return it->second.ptr.getStrong(); } // End CachedSLEs functions. @@ -837,7 +937,8 @@ TaggedCache 0 && cacheCount_ > cacheHardCap_) + chargeEntry(entry, entry.ptr.getStrong()); + if (overHardCap()) evictForHardCap(*cit.ait, cit.mit); entry.touch(clock_.now()); return entry.ptr.getStrong(); @@ -891,11 +992,13 @@ TaggedCache& allRemovals, + std::atomic& allBytesRemoved, std::scoped_lock const&) { return std::thread([&, this]() { int cacheRemovals = 0; int mapRemovals = 0; + std::uint64_t bytesRemoved = 0; // Keep references to all the stuff we sweep // so that we can destroy them outside the lock. @@ -922,6 +1025,8 @@ TaggedCachesecond.costBytes; + cit->second.costBytes = 0; if (cit->second.ptr.useCount() == 1) { stuffToSweep.emplace_back(std::move(cit->second.ptr)); @@ -951,6 +1056,7 @@ TaggedCache& allRemovals, + std::atomic&, std::scoped_lock const&) { return std::thread([&, this]() { diff --git a/include/xrpl/config/Constants.h b/include/xrpl/config/Constants.h index a947e17fead..1b57f663e0f 100644 --- a/include/xrpl/config/Constants.h +++ b/include/xrpl/config/Constants.h @@ -98,6 +98,7 @@ struct Keys static constexpr auto kBgThreads = "bg_threads"; static constexpr auto kBlockSize = "block_size"; static constexpr auto kCacheAge = "cache_age"; + static constexpr auto kCacheBytes = "cache_bytes"; static constexpr auto kCacheMb = "cache_mb"; static constexpr auto kCacheSize = "cache_size"; static constexpr auto kClientMaxWindowBits = "client_max_window_bits"; diff --git a/include/xrpl/nodestore/Database.h b/include/xrpl/nodestore/Database.h index 902cfc9e039..2930f0ba48f 100644 --- a/include/xrpl/nodestore/Database.h +++ b/include/xrpl/nodestore/Database.h @@ -197,7 +197,7 @@ class Database return fetchSz_; } - void + virtual void getCountsJson(json::Value& obj); /** diff --git a/include/xrpl/nodestore/detail/DatabaseNodeImp.h b/include/xrpl/nodestore/detail/DatabaseNodeImp.h index 33a2e279397..93e810ec5f0 100644 --- a/include/xrpl/nodestore/detail/DatabaseNodeImp.h +++ b/include/xrpl/nodestore/detail/DatabaseNodeImp.h @@ -13,6 +13,7 @@ #include #include #include +#include #include #include @@ -58,12 +59,31 @@ class DatabaseNodeImp : public Database if (cacheSize.has_value() || cacheAge.has_value()) { - cache_ = std::make_shared>( + using Cache = TaggedCache; + + // Serialized sizes are exact, so bound this cache by bytes when + // cache_bytes is set; the count target remains advisory. + std::optional byteBudget; + if (auto const bytes = config.exists(Keys::kCacheBytes) + ? get(config, Keys::kCacheBytes) + : 0) + { + // Charge the blob plus per-entry overhead (map node, weak + // tracking, control block). + byteBudget = Cache::ByteBudget{bytes, [](std::shared_ptr const& obj) { + return (obj ? obj->getData().size() : 0) + 160; + }}; + } + + cache_ = std::make_shared( "DatabaseNodeImp", cacheSize.value_or(0), std::chrono::minutes(cacheAge.value_or(0)), stopwatch(), - j); + j, + beast::insight::NullCollector::make(), + 0, + std::move(byteBudget)); } XRPL_ASSERT( @@ -120,6 +140,17 @@ class DatabaseNodeImp : public Database void sweep() override; + void + getCountsJson(json::Value& obj) override + { + Database::getCountsJson(obj); + if (cache_) + { + obj[jss::node_cache_size] = static_cast(cache_->getCacheSize()); + obj[jss::node_cache_bytes] = std::to_string(cache_->getCacheBytes()); + } + } + private: // Cache for database objects. This cache is not always initialized. Check // for null before using. diff --git a/include/xrpl/protocol/jss.h b/include/xrpl/protocol/jss.h index 5b0f978b4c7..4a9d11205c5 100644 --- a/include/xrpl/protocol/jss.h +++ b/include/xrpl/protocol/jss.h @@ -430,6 +430,8 @@ JSS(no_ripple); // out: AccountLines JSS(no_ripple_peer); // out: AccountLines JSS(node); // out: LedgerEntry JSS(node_binary); // out: LedgerEntry +JSS(node_cache_bytes); // out: GetCounts +JSS(node_cache_size); // out: GetCounts JSS(node_read_bytes); // out: GetCounts JSS(node_read_errors); // out: GetCounts JSS(node_read_retries); // out: GetCounts diff --git a/src/tests/libxrpl/basics/TaggedCache.cpp b/src/tests/libxrpl/basics/TaggedCache.cpp index 2789da8f7f4..e55ac21f0c1 100644 --- a/src/tests/libxrpl/basics/TaggedCache.cpp +++ b/src/tests/libxrpl/basics/TaggedCache.cpp @@ -297,4 +297,43 @@ TEST(TaggedCacheTest, hard_cap_disabled) EXPECT_EQ(uncapped.getCacheSize(), 1000); } +TEST(TaggedCacheTest, byte_budget_enforced_on_insert) +{ + using namespace std::chrono_literals; + beast::Journal const journal{TestSink::instance()}; + + TestStopwatch clock; + clock.set(0); + + using Key = LedgerIndex; + using Value = std::string; + using Cache = TaggedCache; + + // Each 100-byte value is charged exactly; the 4 KiB budget holds ~40 + // entries, enforced as the cache grows, with the sweep unable to fire. + Cache::ByteBudget budget{ + 4096, [](std::shared_ptr const& v) { return v ? v->size() : 0; }}; + Cache capped( + "bytes", + 1'000'000, + 3600s, + clock, + journal, + beast::insight::NullCollector::make(), + 0, + budget); + + bool everExceeded = false; + for (Key k = 1; k <= 200; ++k) + { + capped.insert(k, std::string(100, 'x')); + if (capped.getCacheBytes() > 4096) + everExceeded = true; + } + EXPECT_FALSE(everExceeded); + EXPECT_LE(capped.getCacheBytes(), std::size_t{4096}); + EXPECT_GT(capped.getCacheSize(), 0); + EXPECT_LT(capped.getCacheSize(), 50); +} + } // namespace xrpl diff --git a/src/xrpld/app/ledger/LedgerHistory.cpp b/src/xrpld/app/ledger/LedgerHistory.cpp index 8734faa1fb2..292292b5187 100644 --- a/src/xrpld/app/ledger/LedgerHistory.cpp +++ b/src/xrpld/app/ledger/LedgerHistory.cpp @@ -42,7 +42,11 @@ LedgerHistory::LedgerHistory(beast::insight::Collector::ptr const& collector, Ap app_.config().getValueFor(SizedItem::LedgerSize), std::chrono::seconds{app_.config().getValueFor(SizedItem::LedgerAge)}, stopwatch(), - app_.getJournal("TaggedCache")) + app_.getJournal("TaggedCache"), + beast::insight::NullCollector::make(), + // A ledger byte size is ill-posed (nodes are shared copy-on-write + // across ledgers), so bound this cache by count. + app_.config().getValueFor(SizedItem::LedgerSize)) , consensusValidated_( "ConsensusValidated", 64, diff --git a/src/xrpld/app/ledger/detail/LedgerMaster.cpp b/src/xrpld/app/ledger/detail/LedgerMaster.cpp index 83d76bcd2a1..67e37bdee17 100644 --- a/src/xrpld/app/ledger/detail/LedgerMaster.cpp +++ b/src/xrpld/app/ledger/detail/LedgerMaster.cpp @@ -136,7 +136,9 @@ LedgerMaster::LedgerMaster( 65536, std::chrono::seconds{45}, stopwatch, - app_.getJournal("TaggedCache")) + app_.getJournal("TaggedCache"), + beast::insight::NullCollector::make(), + 65536) , stats_([this] { collectMetrics(); }, collector) { } diff --git a/src/xrpld/app/main/Application.cpp b/src/xrpld/app/main/Application.cpp index 62e04a03ed2..7c3a796f571 100644 --- a/src/xrpld/app/main/Application.cpp +++ b/src/xrpld/app/main/Application.cpp @@ -258,6 +258,10 @@ class ApplicationImp : public Application, public BasicApp std::unique_ptr txQ_; ClosureCounter waitHandlerCounter_; boost::asio::steady_timer sweepTimer_; + + // Set by doSweep when post-trim RSS exceeds 150% of the memory budget; + // accelerates the sweep cadence until RSS retreats below 130%. + std::atomic memoryPressure_{false}; boost::asio::steady_timer entropyTimer_; std::optional relationalDatabase_; @@ -365,10 +369,12 @@ class ApplicationImp : public Application, public BasicApp logs_->journal("TaggedCache")) , cachedSLEs_( "Cached SLEs", - 0, + config_->getValueFor(SizedItem::SleCacheSize), std::chrono::minutes(1), stopwatch(), - logs_->journal("CachedSLEs")) + logs_->journal("CachedSLEs"), + beast::insight::NullCollector::make(), + config_->getValueFor(SizedItem::SleCacheSize)) , networkIDService_(std::make_unique(config_->networkId)) , validatorKeys_(*config_, journal_) , resourceManager_( @@ -916,9 +922,11 @@ class ApplicationImp : public Application, public BasicApp })) { using namespace std::chrono; - sweepTimer_.expires_after( - seconds{config_->sweepInterval.value_or( - config_->getValueFor(SizedItem::SweepInterval))}); + auto interval = seconds{ + config_->sweepInterval.value_or(config_->getValueFor(SizedItem::SweepInterval))}; + if (memoryPressure_) + interval = std::min(interval, seconds{10}); + sweepTimer_.expires_after(interval); sweepTimer_.async_wait(std::move(*optionalCountedHandler)); } } @@ -1087,7 +1095,30 @@ class ApplicationImp : public Application, public BasicApp << "; size after: " << cachedSLEs_.size(); } - mallocTrim("doSweep", journal_); + auto const trim = mallocTrim("doSweep", journal_); + + // Circuit breaker, not a control loop: byte-charged caps do the real + // bounding, and RSS legitimately lags eviction (allocator retention), + // so pressure only accelerates sweeps and warns. Hysteresis: trip at + // 150% of the budget, reset below 130%. + if (auto const budget = config_->cacheMemoryBudget(); budget != 0 && trim.rssAfterKB > 0) + { + auto const rss = static_cast(trim.rssAfterKB) * 1024; + if (rss > budget + budget / 2) + { + if (!memoryPressure_) + { + JLOG(journal_.warn()) + << "memory pressure: RSS " << (rss >> 20) << " MB exceeds 150% of the " + << (budget >> 20) << " MB memory_limit; sweeping every 10s"; + } + memoryPressure_ = true; + } + else if (rss < budget + budget * 3 / 10) + { + memoryPressure_ = false; + } + } // Set timer to do another sweep later. setSweepTimer(); diff --git a/src/xrpld/app/misc/SHAMapStoreImp.cpp b/src/xrpld/app/misc/SHAMapStoreImp.cpp index 5c5fc2877b7..0507b766f11 100644 --- a/src/xrpld/app/misc/SHAMapStoreImp.cpp +++ b/src/xrpld/app/misc/SHAMapStoreImp.cpp @@ -178,6 +178,11 @@ SHAMapStoreImp::makeNodeStore(int readThreads) if (!nscfg.exists(Keys::kCacheAge)) nscfg.set(Keys::kCacheAge, "5"); + // An eighth of the memory budget bounds the serialized-object cache. + if (auto const budget = app_.config().cacheMemoryBudget(); + budget != 0 && !nscfg.exists(Keys::kCacheBytes)) + nscfg.set(Keys::kCacheBytes, std::to_string(budget / 8)); + std::unique_ptr db; if (deleteInterval_ != 0u) diff --git a/src/xrpld/core/Config.h b/src/xrpld/core/Config.h index ba5b642d48f..60834532c26 100644 --- a/src/xrpld/core/Config.h +++ b/src/xrpld/core/Config.h @@ -40,6 +40,7 @@ enum class SizedItem : std::size_t { LgrDbCache, BurstSize, AccountIdCacheSize, + SleCacheSize, }; /** diff --git a/src/xrpld/core/detail/Config.cpp b/src/xrpld/core/detail/Config.cpp index 5edd5570b62..ad2accf866c 100644 --- a/src/xrpld/core/detail/Config.cpp +++ b/src/xrpld/core/detail/Config.cpp @@ -1346,6 +1346,10 @@ Config::getValueFor(SizedItem item) const // Fixed regardless of budget: ~22 MB at 72 bytes per slot, and // the value stays prime for hash distribution. return 300007; + case SizedItem::SleCacheSize: + // Closed-ledger SLEs pulled by RPC and pathfinding, a few KB + // each; previously unbounded. + return std::clamp(gb * 1024, 4096, 262144); } UNREACHABLE("xrpl::Config::getValueFor : invalid item");