MB-41804: Use 'Pagable' mem_used & watermark

Ephemeral buckets cannot evict from replica vbuckets directly, but
memory used in replicas is still included in calculations determining
when the item pager should run to delete items.

Because of this, ephemeral buckets can, in certain situations (see MB)
evict all active documents from a given node, as the remaining memory
is occupied by replicas.

In this situation, with enough nodes, the bucket memory usage can
eventually reach 99% of the quota from _just_ replicas. At this point,
the node will backoff on replication. The node will then never recover
from this state - it cannot delete items from replicas, and can also no
longer stream any deletions from other nodes for the replicas.

Change-Id: I96d7be615bca3b53bf31597a93aada0bcf202ea9
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/136495
Well-Formed: Build Bot <[email protected]>
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: daverigby

engines/ep/src/checkpoint.cc

@@ -89,11 +89,13 @@ std::ostream& operator<<(std::ostream& os, const CheckpointCursor& c) {
     return os;
 }
 
-Checkpoint::Checkpoint(EPStats& st,
-                       uint64_t id,
-                       uint64_t snapStart,
-                       uint64_t snapEnd,
-                       uint16_t vbid)
+Checkpoint::Checkpoint(
+        EPStats& st,
+        uint64_t id,
+        uint64_t snapStart,
+        uint64_t snapEnd,
+        uint16_t vbid,
+        const std::function<void(int64_t delta)>& memOverheadChangedCallback)
     : stats(st),
       checkpointId(id),
       snapStartSeqno(snapStart),
@@ -104,15 +106,18 @@ Checkpoint::Checkpoint(EPStats& st,
       numItems(0),
       numMetaItems(0),
       memOverhead(0),
+      memOverheadChangedCallback(memOverheadChangedCallback),
       effectiveMemUsage(0) {
     stats.coreLocal.get()->memOverhead.fetch_add(memorySize());
+    memOverheadChangedCallback(memorySize());
 }
 
 Checkpoint::~Checkpoint() {
     LOG(EXTENSION_LOG_INFO,
         "Checkpoint %" PRIu64 " for vbucket %d is purged from memory",
         checkpointId, vbucketId);
     stats.coreLocal.get()->memOverhead.fetch_sub(memorySize());
+    memOverheadChangedCallback(-ssize_t(memorySize()));
 }
 
 size_t Checkpoint::getNumMetaItems() const {
@@ -246,7 +251,7 @@ queue_dirty_t Checkpoint::queueDirty(const queued_item &qi,
         if (rv == queue_dirty_t::NEW_ITEM) {
             size_t newEntrySize = qi->getKey().size() +
                                   sizeof(index_entry) + sizeof(queued_item);
-            memOverhead += newEntrySize;
+            increaseMemoryOverhead(newEntrySize);
             stats.coreLocal.get()->memOverhead.fetch_add(newEntrySize);
         }
     }
@@ -365,7 +370,7 @@ size_t Checkpoint::mergePrevCheckpoint(Checkpoint *pPrevCheckpoint) {
      */
     setSnapshotStartSeqno(getLowSeqno());
 
-    memOverhead += newEntryMemOverhead;
+    increaseMemoryOverhead(newEntryMemOverhead);
     stats.coreLocal.get()->memOverhead.fetch_add(newEntryMemOverhead);
 
     return numNewItems;
@@ -518,8 +523,12 @@ bool CheckpointManager::addNewCheckpoint_UNLOCKED(uint64_t id,
         id, vbucketId, snapStartSeqno);
 
     bool was_empty = checkpointList.empty() ? true : false;
-    auto checkpoint = std::make_unique<Checkpoint>(stats, id, snapStartSeqno,
-                                            snapEndSeqno, vbucketId);
+    auto checkpoint = std::make_unique<Checkpoint>(stats,
+                                                   id,
+                                                   snapStartSeqno,
+                                                   snapEndSeqno,
+                                                   vbucketId,
+                                                   overheadChangedCallback);
     // Add a dummy item into the new checkpoint, so that any cursor referring
     // to the actual first
     // item in this new checkpoint can be safely shifted left by 1 if the
@@ -1544,6 +1553,20 @@ size_t CheckpointManager::getNumOfMetaItemsFromCursor(const CheckpointCursor &cu
     return result;
 }
 
+void CheckpointManager::updateStatsForStateChange(vbucket_state_t from,
+                                                  vbucket_state_t to) {
+    LockHolder lh(queueLock);
+    if (from == vbucket_state_replica && to != vbucket_state_replica) {
+        // vbucket is changing state away from replica, it's memory usage
+        // should no longer be accounted for as a replica.
+        stats.replicaCheckpointOverhead -= getMemoryOverhead_UNLOCKED();
+    } else if (from != vbucket_state_replica && to == vbucket_state_replica) {
+        // vbucket is changing state to _become_ a replica, it's memory usage
+        // _should_ be accounted for as a replica.
+        stats.replicaCheckpointOverhead += getMemoryOverhead_UNLOCKED();
+    }
+}
+
 void CheckpointManager::decrCursorFromCheckpointEnd(const std::string &name) {
     LockHolder lh(queueLock);
     cursor_index::iterator it = connCursors.find(name);
@@ -1896,9 +1919,9 @@ queued_item CheckpointManager::createCheckpointItem(uint64_t id, uint16_t vbid,
     return qi;
 }
 
-uint64_t CheckpointManager::createNewCheckpoint() {
+uint64_t CheckpointManager::createNewCheckpoint(bool force) {
     LockHolder lh(queueLock);
-    if (checkpointList.back()->getNumItems() > 0) {
+    if (force || checkpointList.back()->getNumItems() > 0) {
         uint64_t chk_id = checkpointList.back()->getId();
         addNewCheckpoint_UNLOCKED(chk_id + 1);
     }

engines/ep/src/checkpoint.h

@@ -346,14 +346,21 @@ std::string to_string(queue_dirty_t value);
  * and system events - for meta-items like checkpoint start/end they share the
  * same sequence number as the associated op - for checkpoint_start that is the
  * ID of the following op, for checkpoint_end the ID of the proceeding op.
+ *
+ * Checkpoints call the provided memOverheadChangedCallback on any action that
+ * changes the memory overhead of the checkpoint - that is, the memory required
+ * _beyond_ that of the Items the Checkpoint holds. This occurs at
+ * creation/destruction or when queuing new items.
  */
 class Checkpoint {
 public:
     Checkpoint(EPStats& st,
                uint64_t id,
                uint64_t snapStart,
                uint64_t snapEnd,
-               uint16_t vbid);
+               uint16_t vbid,
+               const std::function<void(int64_t delta)>&
+                       memOverheadChangedCallback);
 
     ~Checkpoint();
 
@@ -587,6 +594,14 @@ class Checkpoint {
     static const StoredDocKey SetVBucketStateKey;
 
 private:
+    /**
+     * Increase the tracked overhead of the checkpoint. See getMemoryOverhead
+     */
+    void increaseMemoryOverhead(size_t delta) {
+        memOverhead += delta;
+        memOverheadChangedCallback(delta);
+    }
+
     EPStats                       &stats;
     uint64_t                       checkpointId;
     uint64_t                       snapStartSeqno;
@@ -605,6 +620,9 @@ class Checkpoint {
     checkpoint_index               metaKeyIndex;
     size_t                         memOverhead;
 
+    // Reference to callback owned by checkpoint manager for stat tracking
+    const std::function<void(int64_t delta)>& memOverheadChangedCallback;
+
     // The following stat is to contain the memory consumption of all
     // the queued items in the given checkpoint.
     size_t                         effectiveMemUsage;
@@ -847,9 +865,12 @@ class CheckpointManager {
 
     /**
      * Create a new open checkpoint by force.
+     *
+     * @param force create a new checkpoint even if the existing one
+     *        contains no non-meta items
      * @return the new open checkpoint id
      */
-    uint64_t createNewCheckpoint();
+    uint64_t createNewCheckpoint(bool force = false);
 
     void resetCursors(checkpointCursorInfoList &cursors);
 
@@ -946,6 +967,27 @@ class CheckpointManager {
         return ++lastBySeqno;
     }
 
+    /**
+     * Sets the callback to be invoked whenever memory usage changes due to a
+     * new queued item or checkpoint removal (or checkpoint expelling, in
+     * versions this is implemented in). This allows changes in checkpoint
+     * memory usage to be monitored.
+     */
+    void setOverheadChangedCallback(
+            std::function<void(int64_t delta)> callback) {
+        LockHolder lh(queueLock);
+        overheadChangedCallback = std::move(callback);
+
+        size_t initialOverhead = 0;
+        for (const auto& checkpoint : checkpointList) {
+            initialOverhead += checkpoint->memorySize();
+        }
+
+        overheadChangedCallback(initialOverhead);
+    }
+
+    void updateStatsForStateChange(vbucket_state_t from, vbucket_state_t to);
+
     void dump() const;
 
     static const std::string pCursorName;
@@ -972,6 +1014,16 @@ class CheckpointManager {
     void updateDiskQueueStats(VBucket& vbucket, size_t curr_remains,
                               size_t new_remains);
 
+    /**
+     * function to invoke whenever memory usage changes due to a new
+     * queued item or checkpoint removal (or checkpoint expelling, in versions
+     * this ins implemented in).
+     * Must be declared before checkpointList to ensure it still exists
+     * when any Checkpoints within the list are destroyed during destruction
+     * of this CheckpointManager.
+     */
+    std::function<void(int64_t delta)> overheadChangedCallback{[](int64_t) {}};
+
     CheckpointList checkpointList;
 
 private:

engines/ep/src/ep_bucket.cc

@@ -885,6 +885,20 @@ ENGINE_ERROR_CODE EPBucket::getPerVBucketDiskStats(const void* cookie,
     return ENGINE_SUCCESS;
 }
 
+size_t EPBucket::getPageableMemCurrent() const {
+    // EP Buckets can (theoretically) page out all memory, active(+pending) or
+    // replica.
+    return stats.getEstimatedTotalMemoryUsed();
+}
+
+size_t EPBucket::getPageableMemHighWatermark() const {
+    return stats.mem_high_wat;
+}
+
+size_t EPBucket::getPageableMemLowWatermark() const {
+    return stats.mem_low_wat;
+}
+
 VBucketPtr EPBucket::makeVBucket(VBucket::id_type id,
                                  vbucket_state_t state,
                                  KVShard* shard,

engines/ep/src/ep_bucket.h

@@ -100,6 +100,11 @@ class EPBucket : public KVBucket {
 
     ENGINE_ERROR_CODE getPerVBucketDiskStats(const void* cookie,
                                              ADD_STAT add_stat) override;
+
+    size_t getPageableMemCurrent() const override;
+    size_t getPageableMemHighWatermark() const override;
+    size_t getPageableMemLowWatermark() const override;
+
     /**
      * Creates a VBucket object.
      */

engines/ep/src/ep_engine.cc

@@ -3142,6 +3142,14 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::doMemoryStats(const void *cookie,
                     add_stat,
                     cookie);
 
+    add_casted_stat(
+            "ht_mem_used_replica", stats.replicaHTMemory, add_stat, cookie);
+
+    add_casted_stat("replica_checkpoint_memory_overhead",
+                    stats.replicaCheckpointOverhead,
+                    add_stat,
+                    cookie);
+
     add_casted_stat("ep_kv_size", stats.getCurrentSize(), add_stat, cookie);
     add_casted_stat(
             "ep_value_size", stats.getTotalValueSize(), add_stat, cookie);