BucketListDB Random Eviction Cache (#4632)

# Description

Resolves #3696

Adds a random eviction cache to `LiveBucket` using `DiskIndex`.
Currently, I've set the default to cache up to 25% of Bucket entries,
which puts total stellar-core memory usage at about 5-6 GB by default.

# Checklist
- [x] Reviewed the
[contributing](https://github.com/stellar/stellar-core/blob/master/CONTRIBUTING.md#submitting-changes)
document
- [x] Rebased on top of master (no merge commits)
- [x] Ran `clang-format` v8.0.0 (via `make format` or the Visual Studio
extension)
- [x] Compiles
- [x] Ran all tests
- [ ] If change impacts performance, include supporting evidence per the
[performance
document](https://github.com/stellar/stellar-core/blob/master/performance-eval/performance-eval.md)

Author: dmkozh

docs/metrics.md

@@ -45,6 +45,8 @@ bucketlistDB-live.bulk.inflationWinners     | timer     | time to load inflation
 bucketlistDB-live.bulk.poolshareTrustlines  | timer     | time to load poolshare trustlines by accountID and assetID
 bucketlistDB-live.bulk.prefetch             | timer     | time to prefetch
 bucketlistDB-<X>.point.<y>                | timer     | time to load single entry of type <Y> on BucketList <X> (live/hotArchive)
+bucketlistDB-cache.hit                    | meter     | number of cache hits on Live BucketList Disk random eviction cache
+bucketlistDB-cache.miss                   | meter     | number of cache misses on Live BucketList Disk random eviction cache
 crypto.verify.hit                         | meter     | number of signature cache hits
 crypto.verify.miss                        | meter     | number of signature cache misses
 crypto.verify.total                       | meter     | sum of both hits and misses

docs/stellar-core_example.cfg

@@ -235,6 +235,14 @@ MAX_DEX_TX_OPERATIONS_IN_TX_SET = 0
 # 0, indiviudal index is always used. Default page size 16 kb.
 BUCKETLIST_DB_INDEX_PAGE_SIZE_EXPONENT = 14
 
+# BUCKETLIST_DB_CACHED_PERCENT (Integer) default 25
+# Percentage of entries cached by BucketListDB when Bucket size is larger
+# than BUCKETLIST_DB_INDEX_CUTOFF. Note that this value does not impact
+# Buckets smaller than BUCKETLIST_DB_INDEX_CUTOFF, as they are always
+# completely held in memory. Roughly speaking, RAM usage for BucketList
+# cache == BucketListSize * (BUCKETLIST_DB_CACHED_PERCENT / 100).
+BUCKETLIST_DB_CACHED_PERCENT = 25
+
 # BUCKETLIST_DB_INDEX_CUTOFF (Integer) default 250
 # Size, in MB, determining whether a bucket should have an individual
 # key index or a key range index. If bucket size is below this value, range

src/bucket/BucketIndexUtils.cpp

@@ -19,7 +19,8 @@ namespace stellar
 std::streamoff
 getPageSizeFromConfig(Config const& cfg)
 {
-    if (cfg.BUCKETLIST_DB_INDEX_PAGE_SIZE_EXPONENT == 0)
+    if (cfg.BUCKETLIST_DB_INDEX_PAGE_SIZE_EXPONENT == 0 ||
+        cfg.BUCKETLIST_DB_CACHED_PERCENT == 100)
     {
         return 0;
     }

src/bucket/BucketManager.cpp

@@ -157,6 +157,10 @@ BucketManager::BucketManager(Application& app)
           app.getMetrics().NewCounter({"bucketlist", "size", "bytes"}))
     , mArchiveBucketListSizeCounter(
           app.getMetrics().NewCounter({"bucketlist-archive", "size", "bytes"}))
+    , mCacheHitMeter(app.getMetrics().NewMeter({"bucketlistDB", "cache", "hit"},
+                                               "bucketlistDB"))
+    , mCacheMissMeter(app.getMetrics().NewMeter(
+          {"bucketlistDB", "cache", "miss"}, "bucketlistDB"))
     , mBucketListEvictionCounters(app)
     , mEvictionStatistics(std::make_shared<EvictionStatistics>())
     , mConfig(app.getConfig())
@@ -351,6 +355,18 @@ BucketManager::readMergeCounters()
     return mMergeCounters;
 }
 
+medida::Meter&
+BucketManager::getCacheHitMeter() const
+{
+    return mCacheHitMeter;
+}
+
+medida::Meter&
+BucketManager::getCacheMissMeter() const
+{
+    return mCacheMissMeter;
+}
+
 void
 BucketManager::incrMergeCounters(MergeCounters const& delta)
 {

src/bucket/BucketManager.h

@@ -100,6 +100,8 @@ class BucketManager : NonMovableOrCopyable
     medida::Counter& mSharedBucketsSize;
     medida::Counter& mLiveBucketListSizeCounter;
     medida::Counter& mArchiveBucketListSizeCounter;
+    medida::Meter& mCacheHitMeter;
+    medida::Meter& mCacheMissMeter;
     EvictionCounters mBucketListEvictionCounters;
     MergeCounters mMergeCounters;
     std::shared_ptr<EvictionStatistics> mEvictionStatistics{};
@@ -197,6 +199,8 @@ class BucketManager : NonMovableOrCopyable
 
     template <class BucketT> medida::Meter& getBloomMissMeter() const;
     template <class BucketT> medida::Meter& getBloomLookupMeter() const;
+    medida::Meter& getCacheHitMeter() const;
+    medida::Meter& getCacheMissMeter() const;
 
     // Reading and writing the merge counters is done in bulk, and takes a lock
     // briefly; this can be done from any thread.

src/bucket/BucketSnapshot.cpp

@@ -39,7 +39,7 @@ BucketSnapshotBase<BucketT>::isEmpty() const
 }
 
 template <class BucketT>
-std::pair<std::shared_ptr<typename BucketT::EntryT>, bool>
+std::pair<std::shared_ptr<typename BucketT::EntryT const>, bool>
 BucketSnapshotBase<BucketT>::getEntryAtOffset(LedgerKey const& k,
                                               std::streamoff pos,
                                               size_t pageSize) const
@@ -58,12 +58,16 @@ BucketSnapshotBase<BucketT>::getEntryAtOffset(LedgerKey const& k,
     {
         if (stream.readOne(be))
         {
-            return {std::make_shared<typename BucketT::EntryT>(be), false};
+            auto entry = std::make_shared<typename BucketT::EntryT const>(be);
+            mBucket->getIndex().maybeAddToCache(entry);
+            return {entry, false};
         }
     }
     else if (stream.readPage(be, k, pageSize))
     {
-        return {std::make_shared<typename BucketT::EntryT>(be), false};
+        auto entry = std::make_shared<typename BucketT::EntryT const>(be);
+        mBucket->getIndex().maybeAddToCache(entry);
+        return {entry, false};
     }
 
     mBucket->getIndex().markBloomMiss();

src/bucket/BucketSnapshot.h

@@ -41,7 +41,7 @@ template <class BucketT> class BucketSnapshotBase : public NonMovable
     // reads until key is found or the end of the page. Returns <BucketEntry,
     // bloomMiss>, where bloomMiss is true if a bloomMiss occurred during the
     // load.
-    std::pair<std::shared_ptr<typename BucketT::EntryT>, bool>
+    std::pair<std::shared_ptr<typename BucketT::EntryT const>, bool>
     getEntryAtOffset(LedgerKey const& k, std::streamoff pos,
                      size_t pageSize) const;
 

src/bucket/BucketUtils.cpp

@@ -337,6 +337,17 @@ BucketEntryCounters::operator!=(BucketEntryCounters const& other) const
     return !(*this == other);
 }
 
+size_t
+BucketEntryCounters::numEntries() const
+{
+    size_t num = 0;
+    for (auto const& [_, count] : entryTypeCounts)
+    {
+        num += count;
+    }
+    return num;
+}
+
 template void
 BucketEntryCounters::count<LiveBucket>(LiveBucket::EntryT const& be);
 template void BucketEntryCounters::count<HotArchiveBucket>(

src/bucket/BucketUtils.h

@@ -196,6 +196,7 @@ struct BucketEntryCounters
     BucketEntryCounters& operator+=(BucketEntryCounters const& other);
     bool operator==(BucketEntryCounters const& other) const;
     bool operator!=(BucketEntryCounters const& other) const;
+    size_t numEntries() const;
 
     template <class Archive>
     void

src/bucket/HotArchiveBucketIndex.h

@@ -74,6 +74,14 @@ class HotArchiveBucketIndex : public NonMovableOrCopyable
         return mDiskIndex.scan(mDiskIndex.begin(), k).first;
     }
 
+    // Hot Archive does not support the cache, so define empty function for
+    // consistency with LiveBucketIndex
+    void
+    maybeAddToCache(
+        std::shared_ptr<HotArchiveBucketEntry const> const& entry) const
+    {
+    }
+
     std::pair<IndexReturnT, IterT> scan(IterT start, LedgerKey const& k) const;
 
     BucketEntryCounters const&

src/bucket/LiveBucketIndex.cpp

@@ -11,6 +11,8 @@
 #include "util/Logging.h"
 #include "xdr/Stellar-ledger-entries.h"
 #include <ios>
+#include <medida/meter.h>
+#include <shared_mutex>
 #include <vector>
 
 namespace stellar
@@ -38,6 +40,8 @@ LiveBucketIndex::getPageSize(Config const& cfg, size_t bucketSize)
 LiveBucketIndex::LiveBucketIndex(BucketManager& bm,
                                  std::filesystem::path const& filename,
                                  Hash const& hash, asio::io_context& ctx)
+    : mCacheHitMeter(bm.getCacheHitMeter())
+    , mCacheMissMeter(bm.getCacheMissMeter())
 {
     ZoneScoped;
     releaseAssert(!filename.empty());
@@ -60,6 +64,21 @@ LiveBucketIndex::LiveBucketIndex(BucketManager& bm,
                    pageSize, filename);
         mDiskIndex = std::make_unique<DiskIndex<LiveBucket>>(
             bm, filename, pageSize, hash, ctx);
+
+        auto percentCached = bm.getConfig().BUCKETLIST_DB_CACHED_PERCENT;
+        if (percentCached > 0)
+        {
+            auto const& counters = mDiskIndex->getBucketEntryCounters();
+            auto cacheSize = (counters.numEntries() * percentCached) / 100;
+
+            // Minimum cache size of 100 if we are going to cache a non-zero
+            // number of entries
+            // We don't want to reserve here, since caches only live as long as
+            // the lifetime of the Bucket and fill relatively slowly
+            mCache = std::make_unique<CacheT>(std::max<size_t>(cacheSize, 100),
+                                              /*separatePRNG=*/false,
+                                              /*reserve=*/false);
+        }
     }
 }
 
@@ -68,6 +87,8 @@ LiveBucketIndex::LiveBucketIndex(BucketManager const& bm, Archive& ar,
                                  std::streamoff pageSize)
 
     : mDiskIndex(std::make_unique<DiskIndex<LiveBucket>>(ar, bm, pageSize))
+    , mCacheHitMeter(bm.getCacheHitMeter())
+    , mCacheMissMeter(bm.getCacheMissMeter())
 {
     // Only disk indexes are serialized
     releaseAssertOrThrow(pageSize != 0);
@@ -108,11 +129,39 @@ LiveBucketIndex::markBloomMiss() const
     mDiskIndex->markBloomMiss();
 }
 
+std::shared_ptr<BucketEntry const>
+LiveBucketIndex::getCachedEntry(LedgerKey const& k) const
+{
+    if (shouldUseCache())
+    {
+        std::shared_lock<std::shared_mutex> lock(mCacheMutex);
+        auto cachePtr = mCache->maybeGet(k);
+        if (cachePtr)
+        {
+            mCacheHitMeter.Mark();
+            return *cachePtr;
+        }
+
+        // In the case of a bloom filter false positive, we might have a cache
+        // "miss" because we're searching for something that doesn't exist. We
+        // don't cache non-existent entries, so we don't meter misses here.
+        // Instead, we track misses when we insert a new entry, since we always
+        // insert a new entry into the cache after a miss.
+    }
+
+    return nullptr;
+}
+
 IndexReturnT
 LiveBucketIndex::lookup(LedgerKey const& k) const
 {
     if (mDiskIndex)
     {
+        if (auto cached = getCachedEntry(k); cached)
+        {
+            return IndexReturnT(cached);
+        }
+
         return mDiskIndex->scan(mDiskIndex->begin(), k).first;
     }
     else
@@ -127,6 +176,11 @@ LiveBucketIndex::scan(IterT start, LedgerKey const& k) const
 {
     if (mDiskIndex)
     {
+        if (auto cached = getCachedEntry(k); cached)
+        {
+            return {IndexReturnT(cached), start};
+        }
+
         return mDiskIndex->scan(getDiskIter(start), k);
     }
 
@@ -207,6 +261,24 @@ LiveBucketIndex::getBucketEntryCounters() const
     return mInMemoryIndex->getBucketEntryCounters();
 }
 
+void
+LiveBucketIndex::maybeAddToCache(
+    std::shared_ptr<BucketEntry const> const& entry) const
+{
+    if (shouldUseCache())
+    {
+        releaseAssertOrThrow(entry);
+        auto k = getBucketLedgerKey(*entry);
+
+        // If we are adding an entry to the cache, we must have missed it
+        // earlier.
+        mCacheMissMeter.Mark();
+
+        std::unique_lock<std::shared_mutex> lock(mCacheMutex);
+        mCache->put(k, entry);
+    }
+}
+
 #ifdef BUILD_TESTS
 bool
 LiveBucketIndex::operator==(LiveBucketIndex const& in) const

src/bucket/LiveBucketIndex.h

@@ -9,13 +9,16 @@
 #include "bucket/InMemoryIndex.h"
 #include "bucket/LedgerCmp.h"
 #include "bucket/LiveBucket.h"
+#include "ledger/LedgerHashUtils.h" // IWYU pragma: keep
 #include "util/NonCopyable.h"
+#include "util/RandomEvictionCache.h"
 #include "util/XDROperators.h" // IWYU pragma: keep
 #include "xdr/Stellar-ledger-entries.h"
 #include <filesystem>
 #include <optional>
 
 #include <cereal/archives/binary.hpp>
+#include <shared_mutex>
 
 namespace asio
 {
@@ -52,9 +55,21 @@ class LiveBucketIndex : public NonMovableOrCopyable
     using IterT =
         std::variant<InMemoryIndex::IterT, DiskIndex<LiveBucket>::IterT>;
 
+    using CacheT =
+        RandomEvictionCache<LedgerKey, std::shared_ptr<BucketEntry const>>;
+
   private:
     std::unique_ptr<DiskIndex<LiveBucket> const> mDiskIndex;
     std::unique_ptr<InMemoryIndex const> mInMemoryIndex;
+    std::unique_ptr<CacheT> mCache;
+
+    // The indexes themselves are thread safe, as they are immutable after
+    // construction. The cache is not, all accesses must first acquire this
+    // mutex.
+    mutable std::shared_mutex mCacheMutex;
+
+    medida::Meter& mCacheHitMeter;
+    medida::Meter& mCacheMissMeter;
 
     static inline DiskIndex<LiveBucket>::IterT
     getDiskIter(IterT const& iter)
@@ -68,6 +83,15 @@ class LiveBucketIndex : public NonMovableOrCopyable
         return std::get<InMemoryIndex::IterT>(iter);
     }
 
+    bool
+    shouldUseCache() const
+    {
+        return mDiskIndex && mCache;
+    }
+
+    // Returns nullptr if cache is not enabled or entry not found
+    std::shared_ptr<BucketEntry const> getCachedEntry(LedgerKey const& k) const;
+
   public:
     inline static const std::string DB_BACKEND_STATE = "bl";
     inline static const uint32_t BUCKET_INDEX_VERSION = 5;
@@ -97,6 +121,8 @@ class LiveBucketIndex : public NonMovableOrCopyable
     std::optional<std::pair<std::streamoff, std::streamoff>>
     getOfferRange() const;
 
+    void maybeAddToCache(std::shared_ptr<BucketEntry const> const& entry) const;
+
     BucketEntryCounters const& getBucketEntryCounters() const;
     uint32_t getPageSize() const;
 
@@ -105,6 +131,8 @@ class LiveBucketIndex : public NonMovableOrCopyable
     void markBloomMiss() const;
 #ifdef BUILD_TESTS
     bool operator==(LiveBucketIndex const& in) const;
+
+    void clearCache() const;
 #endif
 };
 }

src/bucket/readme.md

@@ -97,3 +97,8 @@ lookup speed and memory overhead. The following configuration flags control thes
     on startup. Defaults to true, should only be set to false for testing purposes.
     Validators do not currently support persisted indexes. If NODE_IS_VALIDATOR=true,
     this value is ignored and indexes are never persisted.
+- `BUCKETLIST_DB_CACHED_PERCENT`
+  - Percentage of entries cached by BucketListDB when Bucket size is larger
+    than `BUCKETLIST_DB_INDEX_CUTOFF`. Note that this value does not impact
+    Buckets smaller than `BUCKETLIST_DB_INDEX_CUTOFF`, as they are always
+    completely held in memory.