MB-40531: [BP] Prefer paging from replicas if possible

Backport of http://review.couchbase.org/c/kv_engine/+/134806

Prior to this change, the pager would try to page out X% of items in
replica vbuckets, but if this did not bring memory usage below the low
water mark, it would flip phase to ACTIVE_AND_PENDING_ONLY - meaning
the next run of the pager would page immediately try to page from
active and pending vbuckets.

Now, active/pending vbuckets will only be considered if paging _every_
possible item from replicas would not free enough memory to drop below
the low water mark.

Change-Id: I46e578bf163fa2e12834173595c8aa3f52f1593d
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/133197
Tested-by: Build Bot <[email protected]>
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Jim Walker <[email protected]>

Author: jameseh96

engines/ep/src/ep_bucket.h

@@ -216,6 +216,10 @@ class EPBucket : public KVBucket {
 
     void warmupCompleted();
 
+    bool canEvictFromReplicas() override {
+        return true;
+    }
+
 protected:
     // During the warmup phase we might want to enable external traffic
     // at a given point in time.. The LoadStorageKvPairCallback will be

engines/ep/src/ep_vb.cc

@@ -490,6 +490,14 @@ bool EPVBucket::eligibleToPageOut(const HashTable::HashBucketLock& lh,
     return v.eligibleForEviction(eviction);
 }
 
+size_t EPVBucket::getPageableMemUsage() {
+    if (eviction == EvictionPolicy::Full) {
+        return ht.getItemMemory();
+    } else {
+        return ht.getItemMemory() - ht.getMetadataMemory();
+    }
+}
+
 size_t EPVBucket::queueBGFetchItem(const DocKey& key,
                                    std::unique_ptr<VBucketBGFetchItem> fetch,
                                    BgFetcher* bgFetcher) {

engines/ep/src/ep_vb.h

@@ -108,6 +108,8 @@ class EPVBucket : public VBucket {
     bool eligibleToPageOut(const HashTable::HashBucketLock& lh,
                            const StoredValue& v) const override;
 
+    size_t getPageableMemUsage() override;
+
     bool areDeletedItemsAlwaysResident() const override;
 
     void addStats(VBucketStatsDetailLevel detail,

engines/ep/src/ephemeral_bucket.h

@@ -136,6 +136,10 @@ class EphemeralBucket : public KVBucket {
      */
     static void reconfigureForEphemeral(Configuration& config);
 
+    bool canEvictFromReplicas() override {
+        return false;
+    }
+
 protected:
     std::unique_ptr<VBucketCountVisitor> makeVBCountVisitor(
             vbucket_state_t state) override;

engines/ep/src/ephemeral_vb.cc

@@ -153,6 +153,17 @@ bool EphemeralVBucket::eligibleToPageOut(const HashTable::HashBucketLock& lh,
     return true;
 }
 
+size_t EphemeralVBucket::getPageableMemUsage() {
+    if (getState() == vbucket_state_replica) {
+        // Ephemeral buckets are not backed by disk - nothing can be evicted
+        // from a replica as deleting from replicas would cause inconsistency
+        // with the active. When the active vb evicts items deletions will be
+        // streamed to the replica.
+        return 0;
+    }
+    return ht.getItemMemory();
+}
+
 bool EphemeralVBucket::areDeletedItemsAlwaysResident() const {
     // Ephemeral buckets do keep all deleted items resident in memory.
     // (We have nowhere else to store them, given there is no disk).

engines/ep/src/ephemeral_vb.h

@@ -110,6 +110,8 @@ class EphemeralVBucket : public VBucket {
     bool eligibleToPageOut(const HashTable::HashBucketLock& lh,
                            const StoredValue& v) const override;
 
+    size_t getPageableMemUsage() override;
+
     bool areDeletedItemsAlwaysResident() const override;
 
     void addStats(VBucketStatsDetailLevel detail,

engines/ep/src/hash_table.h

@@ -1283,6 +1283,15 @@ class HashTable {
      */
     nlohmann::json dumpStoredValuesAsJson() const;
 
+    /**
+     * Generates a new value that is either the same or higher than the input
+     * value.  It is intended to be used to increment the frequency counter of a
+     * storedValue.
+     * @param value  The value counter to try to generate an increment for.
+     * @returns      The new value that is the same or higher than value.
+     */
+    uint8_t generateFreqValue(uint8_t value);
+
 private:
     // The container for actually holding the StoredValues.
     using table_type = std::vector<StoredValue::UniquePtr>;
@@ -1452,15 +1461,6 @@ class HashTable {
 
     void clear_UNLOCKED(bool deactivate);
 
-    /**
-     * Generates a new value that is either the same or higher than the input
-     * value.  It is intended to be used to increment the frequency counter of a
-     * storedValue.
-     * @param value  The value counter to try to generate an increment for.
-     * @returns      The new value that is the same or higher than value.
-     */
-    uint8_t generateFreqValue(uint8_t value);
-
     /**
      * Update the frequency counter of a given stored value.
      * @param v  reference to a value in the hash table whose frequency counter

engines/ep/src/item_pager.cc

@@ -30,6 +30,7 @@
 #include "kv_bucket_iface.h"
 #include "paging_visitor.h"
 
+#include <folly/lang/Assume.h>
 #include <platform/platform_time.h>
 
 #include <cmath>
@@ -43,23 +44,44 @@
 #include <phosphor/phosphor.h>
 #include <memory>
 
+double EvictionRatios::getForState(vbucket_state_t state) {
+    switch (state) {
+    case vbucket_state_replica:
+        return replica;
+    case vbucket_state_active:
+    case vbucket_state_pending:
+        return activeAndPending;
+    case vbucket_state_dead:
+        return 0;
+    }
+    folly::assume_unreachable();
+}
+
+void EvictionRatios::setForState(vbucket_state_t state, double value) {
+    switch (state) {
+    case vbucket_state_replica:
+        replica = value;
+        return;
+    case vbucket_state_active:
+    case vbucket_state_pending:
+        activeAndPending = value;
+        return;
+    case vbucket_state_dead:
+        // no-op
+        return;
+    }
+    folly::assume_unreachable();
+}
+
 ItemPager::ItemPager(EventuallyPersistentEngine& e, EPStats& st)
     : GlobalTask(&e, TaskId::ItemPager, 10, false),
       engine(e),
       stats(st),
       available(new std::atomic<bool>(true)),
-      phase(REPLICA_ONLY),
       doEvict(false),
       sleepTime(std::chrono::milliseconds(
               e.getConfiguration().getPagerSleepTimeMs())),
       notified(false) {
-    // For the hifi_mfu algorithm if a couchbase/persistent bucket we
-    // want to start visiting the replica vbucket first.  However for
-    // ephemeral we do not evict from replica vbuckets and therefore
-    // we start with active and pending vbuckets.
-    phase = (engine.getConfiguration().getBucketType() == "persistent")
-                    ? REPLICA_ONLY
-                    : ACTIVE_AND_PENDING_ONLY;
 }
 
 bool ItemPager::run(void) {
@@ -72,7 +94,7 @@ bool ItemPager::run(void) {
     // be that we've gone over HWM have been notified to run, then came back
     // down (e.g. 1 byte under HWM), we should still page in this scenario.
     // Notified would be false if we were woken by the periodic scheduler
-    bool wasNotified = notified;
+    const bool wasNotified = notified;
 
     // Clear the notification flag before starting the task's actions
     notified.store(false);
@@ -95,53 +117,82 @@ bool ItemPager::run(void) {
 
         ++stats.pagerRuns;
 
-        double toKill = (current - static_cast<double>(lower)) / current;
+        if (current <= lower) {
+            // early exit - no need to run a paging visitor
+            return true;
+        }
 
-        EP_LOG_DEBUG("Using {} bytes of memory, paging out {} of items.",
-                     stats.getEstimatedTotalMemoryUsed(),
-                     (toKill * 100.0));
+        VBucketFilter replicaFilter;
+        VBucketFilter activePendingFilter;
 
-        // compute active vbuckets evicition bias factor
-        Configuration& cfg = engine.getConfiguration();
-        size_t activeEvictPerc = cfg.getPagerActiveVbPcnt();
-        double bias = static_cast<double>(activeEvictPerc) / 50;
+        for (auto vbid : kvBucket->getVBucketsInState(vbucket_state_replica)) {
+            replicaFilter.addVBucket(vbid);
+        }
+
+        for (auto vbid : kvBucket->getVBucketsInState(vbucket_state_active)) {
+            activePendingFilter.addVBucket(vbid);
+        }
+        for (auto vbid : kvBucket->getVBucketsInState(vbucket_state_pending)) {
+            activePendingFilter.addVBucket(vbid);
+        }
+
+        ssize_t bytesToEvict = current - lower;
+
+        const double replicaEvictableMem = getEvictableBytes(replicaFilter);
+        const double activePendingEvictableMem =
+                getEvictableBytes(activePendingFilter);
+
+        double replicaEvictionRatio = 0.0;
+        double activeAndPendingEvictionRatio = 0.0;
+
+        if (kvBucket->canEvictFromReplicas()) {
+            // try evict from replicas first if we can
+            replicaEvictionRatio =
+                    std::min(1.0, bytesToEvict / replicaEvictableMem);
+
+            bytesToEvict -= replicaEvictableMem;
+        }
+
+        if (bytesToEvict > 0) {
+            // replicas are not sufficient (or are not eligible for eviction if
+            // ephemeral). Not enough memory can be reclaimed from them to
+            // reach the low watermark.
+            // Consider active and pending vbuckets too.
+            // active and pending share an eviction ratio, it need only be
+            // set once
+            activeAndPendingEvictionRatio =
+                    std::min(1.0, bytesToEvict / activePendingEvictableMem);
+        }
+
+        EP_LOG_DEBUG(
+                "Using {} bytes of memory, paging out {}% of active and "
+                "pending items, {}% of replica items.",
+                stats.getEstimatedTotalMemoryUsed(),
+                (activeAndPendingEvictionRatio * 100.0),
+                (replicaEvictionRatio * 100.0));
 
         VBucketFilter filter;
-        // For the hifi_mfu algorithm use the phase to filter which vbuckets
-        // we want to visit (either replica or active/pending vbuckets).
-        vbucket_state_t state;
-        if (phase == REPLICA_ONLY) {
-            state = vbucket_state_replica;
-        } else if (phase == ACTIVE_AND_PENDING_ONLY) {
-            state = vbucket_state_active;
-            auto acceptableVBs = kvBucket->getVBucketsInState(state);
-            for (auto vb : acceptableVBs) {
-                filter.addVBucket(vb);
-            }
-            state = vbucket_state_pending;
-        } else {
-            throw std::invalid_argument(
-                    "ItemPager::run - "
-                    "phase is invalid for hifi_mfu eviction algorithm");
+
+        if (replicaEvictionRatio > 0.0) {
+            filter = filter.filter_union(replicaFilter);
         }
-        auto acceptableVBs = kvBucket->getVBucketsInState(state);
-        for (auto vb : acceptableVBs) {
-            filter.addVBucket(vb);
+
+        if (activeAndPendingEvictionRatio > 0.0) {
+            filter = filter.filter_union(activePendingFilter);
         }
 
-        bool isEphemeral = (cfg.getBucketType() == "ephemeral");
+        // compute active vbuckets evicition bias factor
+        const Configuration& cfg = engine.getConfiguration();
 
         auto pv = std::make_unique<PagingVisitor>(
                 *kvBucket,
                 stats,
-                toKill,
+                EvictionRatios{activeAndPendingEvictionRatio,
+                               replicaEvictionRatio},
                 available,
                 ITEM_PAGER,
                 false,
-                bias,
                 filter,
-                &phase,
-                isEphemeral,
                 cfg.getItemEvictionAgePercentage(),
                 cfg.getItemEvictionFreqCounterAgeThreshold());
 
@@ -165,6 +216,40 @@ void ItemPager::scheduleNow() {
     }
 }
 
+/**
+ * Visitor used to aggregate how much memory could potentially be reclaimed
+ * by evicting every eligible item from specified vbuckets
+ */
+class VBucketEvictableMemVisitor : public VBucketVisitor {
+public:
+    explicit VBucketEvictableMemVisitor(const VBucketFilter& filter)
+        : filter(filter) {
+    }
+
+    void visitBucket(const VBucketPtr& vb) override {
+        if (!filter.empty() && filter(vb->getId())) {
+            totalEvictableMemory += vb->getPageableMemUsage();
+        }
+    }
+
+    size_t getTotalEvictableMemory() const {
+        return totalEvictableMemory;
+    }
+
+private:
+    const VBucketFilter& filter;
+    size_t totalEvictableMemory = 0;
+};
+
+size_t ItemPager::getEvictableBytes(const VBucketFilter& filter) const {
+    KVBucket* kvBucket = engine.getKVBucket();
+
+    VBucketEvictableMemVisitor visitor(filter);
+    kvBucket->visit(visitor);
+
+    return visitor.getTotalEvictableMemory();
+}
+
 ExpiredItemPager::ExpiredItemPager(EventuallyPersistentEngine *e,
                                    EPStats &st, size_t stime,
                                    ssize_t taskTime) :
@@ -218,19 +303,15 @@ bool ExpiredItemPager::run(void) {
 
         VBucketFilter filter;
         Configuration& cfg = engine->getConfiguration();
-        bool isEphemeral =
-                (engine->getConfiguration().getBucketType() == "ephemeral");
         auto pv = std::make_unique<PagingVisitor>(
                 *kvBucket,
                 stats,
-                -1,
+                EvictionRatios{0.0 /* active&pending */,
+                               0.0 /* replica */}, // evict nothing
                 available,
                 EXPIRY_PAGER,
                 true,
-                1,
                 filter,
-                /* pager_phase */ nullptr,
-                isEphemeral,
                 cfg.getItemEvictionAgePercentage(),
                 cfg.getItemEvictionFreqCounterAgeThreshold());