MB-41804: Merge branch 'couchbase/alice' into mad-hatter

Note: Due to changes in the CheckpointManager memory tracking between
alice and mad-hatter, checks in ReplicaMemoryTrackingStateChange have
been temporarily weakened, and will be restored in a later patch.

* commit '69730e6f4':
  MB-41804: Use 'Pagable' mem_used & watermark

Change-Id: Ieec3dee7137a733b7d5aa3161410ac7286c7fa82

Author: jameseh96

engines/ep/src/checkpoint.cc

@@ -181,7 +181,8 @@ Checkpoint::Checkpoint(EPStats& st,
                        uint64_t visibleSnapEnd,
                        boost::optional<uint64_t> highCompletedSeqno,
                        Vbid vbid,
-                       CheckpointType checkpointType)
+                       CheckpointType checkpointType,
+                       const std::function<void(int64_t delta)>& memOverheadChangedCallback)
     : stats(st),
       checkpointId(id),
       snapStartSeqno(snapStart),
@@ -199,8 +200,10 @@ Checkpoint::Checkpoint(EPStats& st,
       keyIndexMemUsage(0),
       queuedItemsMemUsage(0),
       checkpointType(checkpointType),
-      highCompletedSeqno(highCompletedSeqno) {
+      highCompletedSeqno(highCompletedSeqno),
+      memOverheadChangedCallback(memOverheadChangedCallback) {
     stats.coreLocal.get()->memOverhead.fetch_add(sizeof(Checkpoint));
+    memOverheadChangedCallback(sizeof(Checkpoint));
 }
 
 Checkpoint::~Checkpoint() {
@@ -213,8 +216,9 @@ Checkpoint::~Checkpoint() {
      * of queued_items in the checkpoint.
      */
     auto queueMemOverhead = sizeof(queued_item) * toWrite.size();
-    stats.coreLocal.get()->memOverhead.fetch_sub(
-            sizeof(Checkpoint) + keyIndexMemUsage + queueMemOverhead);
+    auto overhead = sizeof(Checkpoint) + keyIndexMemUsage + queueMemOverhead;
+    stats.coreLocal.get()->memOverhead.fetch_sub(overhead);
+    memOverheadChangedCallback(-ssize_t(overhead));
 }
 
 QueueDirtyStatus Checkpoint::queueDirty(const queued_item& qi,
@@ -384,8 +388,9 @@ QueueDirtyStatus Checkpoint::queueDirty(const queued_item& qi,
              * item to the queue (toWrite).  This is approximated to the
              * addition to metaKeyIndex / keyIndex plus sizeof(queued_item).
              */
-            stats.coreLocal.get()->memOverhead.fetch_add(indexKeyUsage +
-                                                         sizeof(queued_item));
+            auto overhead = indexKeyUsage + sizeof(queued_item);
+            stats.coreLocal.get()->memOverhead.fetch_add(overhead);
+            memOverheadChangedCallback(overhead);
             /**
              *  Update the total metaKeyIndex / keyIndex memory usage which is
              *  used when the checkpoint is destructed to manually account

engines/ep/src/checkpoint.h

@@ -396,6 +396,10 @@ std::string to_string(QueueDirtyStatus value);
  * if the cursor is pointing to a meta-item the position to expel from is moved
  * backwards until either a mutation item or the dummy item is reached.
  *
+ * 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:
@@ -406,7 +410,9 @@ class Checkpoint {
                uint64_t visibleSnapEnd,
                boost::optional<uint64_t> highCompletedSeqno,
                Vbid vbid,
-               CheckpointType checkpointType);
+               CheckpointType checkpointType,
+               const std::function<void(int64_t delta)>&
+                       memOverheadChangedCallback);
 
     ~Checkpoint();
 
@@ -770,5 +776,8 @@ class Checkpoint {
     // de-duplication.
     boost::optional<uint64_t> maxDeletedRevSeqno;
 
+    // Reference to callback owned by checkpoint manager for stat tracking
+    const std::function<void(int64_t delta)>& memOverheadChangedCallback;
+
     friend std::ostream& operator <<(std::ostream& os, const Checkpoint& m);
 };

engines/ep/src/checkpoint_manager.cc

@@ -250,7 +250,8 @@ void CheckpointManager::addOpenCheckpoint(
                                              visibleSnapEnd,
                                              highCompletedSeqno,
                                              vbucketId,
-                                             checkpointType);
+                                             checkpointType,
+                                             overheadChangedCallback);
     // Add an empty-item into the new checkpoint.
     // We need this because every CheckpointCursor will point to this empty-item
     // at creation. So, the cursor will point at the first actual non-meta item
@@ -1451,12 +1452,12 @@ queued_item CheckpointManager::createCheckpointItem(uint64_t id,
     return qi;
 }
 
-uint64_t CheckpointManager::createNewCheckpoint() {
+uint64_t CheckpointManager::createNewCheckpoint(bool force) {
     LockHolder lh(queueLock);
 
     const auto& openCkpt = getOpenCheckpoint_UNLOCKED(lh);
 
-    if (openCkpt.getNumItems() == 0) {
+    if (openCkpt.getNumItems() == 0 && !force) {
         return openCkpt.getId();
     }
 
@@ -1631,6 +1632,34 @@ bool CheckpointManager::isOpenCheckpointDisk() {
     return checkpointList.back()->isDiskCheckpoint();
 }
 
+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::setOverheadChangedCallback(
+        std::function<void(int64_t delta)> callback) {
+    LockHolder lh(queueLock);
+    overheadChangedCallback = std::move(callback);
+
+    overheadChangedCallback(getMemoryOverhead_UNLOCKED());
+}
+
+std::function<void(int64_t delta)>
+CheckpointManager::getOverheadChangedCallback() const {
+    LockHolder lh(queueLock);
+    return overheadChangedCallback;
+}
+
 std::ostream& operator <<(std::ostream& os, const CheckpointManager& m) {
     os << "CheckpointManager[" << &m << "] with numItems:"
        << m.getNumItems() << " checkpoints:" << m.checkpointList.size()

engines/ep/src/checkpoint_manager.h

@@ -372,9 +372,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);
 
     /**
      * Get id of the previous checkpoint that is followed by the checkpoint
@@ -484,6 +487,24 @@ class CheckpointManager {
     /// @return true if the current open checkpoint is a DiskCheckpoint
     bool isOpenCheckpointDisk();
 
+    void updateStatsForStateChange(vbucket_state_t from, vbucket_state_t to);
+
+    /**
+     * 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);
+
+    /**
+     * Gets 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).
+     */
+    std::function<void(int64_t delta)> getOverheadChangedCallback() const;
+
     /**
      * Member std::function variable, to allow us to inject code into
      * removeCursor_UNLOCKED() for unit MB36146
@@ -519,6 +540,16 @@ class CheckpointManager {
                                               VBucket& vb,
                                               const queued_item& qi);
 
+    /**
+     * 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;
 
     // Pair of {sequence number, cursor at checkpoint start} used when

engines/ep/src/ep_bucket.cc

@@ -1276,6 +1276,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(
         Vbid id,
         vbucket_state_t state,

engines/ep/src/ep_bucket.h

@@ -129,6 +129,11 @@ class EPBucket : public KVBucket {
 
     ENGINE_ERROR_CODE getPerVBucketDiskStats(
             const void* cookie, const AddStatFn& add_stat) override;
+
+    size_t getPageableMemCurrent() const override;
+    size_t getPageableMemHighWatermark() const override;
+    size_t getPageableMemLowWatermark() const override;
+
     /**
      * Creates a VBucket object from warmup (can set collection state)
      */

engines/ep/src/ep_engine.cc

@@ -3215,6 +3215,14 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::doMemoryStats(
                     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);

engines/ep/src/ephemeral_bucket.cc

@@ -28,9 +28,12 @@
 #include "replicationthrottle.h"
 #include "rollback_result.h"
 #include "statwriter.h"
+#include <checkpoint_manager.h>
 
 #include <platform/sized_buffer.h>
 
+#include <algorithm>
+
 /**
  * A configuration value changed listener that responds to Ephemeral bucket
  * parameter changes.
@@ -148,6 +151,70 @@ bool EphemeralBucket::initialize() {
     return true;
 }
 
+size_t EphemeralBucket::getPageableMemCurrent() const {
+    // Ephemeral buckets differ from persistent in terms of how memory can
+    // be freed - only active items can (potentially) be directly deleted
+    // (auto-delete or items which have expired) - given that the replica
+    // must be an exact copy of the active.
+    // As such, 'pageable' memory is non-replica memmory.
+
+    // We don't directly track "active_mem_used", but we can roughtly estimate
+    // it from mem_used - replica_ht_mem - replica_checkpoint_mem
+    const auto estimatedActiveMemory =
+            int64_t(stats.getEstimatedTotalMemoryUsed()) -
+            stats.replicaHTMemory - stats.replicaCheckpointOverhead;
+    return std::max(estimatedActiveMemory, int64_t(0));
+}
+
+size_t EphemeralBucket::getPageableMemHighWatermark() const {
+    // Ephemeral buckets can only page out non-replica memory (see comments
+    // in getPageableMemCurrent). As such, set pagable high watermark to
+    // a fraction of the overall high watermark based on what faction of
+    // vBuckets are active. Memory used by any dead vbs should be reclaimed
+    // soon, so ignore them here; they don't need to be allocated a portion
+    // of the quota.
+    const double activeVBCount = vbMap.getVBStateCount(vbucket_state_active);
+    const double pendingVBCount = vbMap.getVBStateCount(vbucket_state_pending);
+    const double replicaVBCount = vbMap.getVBStateCount(vbucket_state_replica);
+    const double totalVBCount = activeVBCount + pendingVBCount + replicaVBCount;
+
+    if (totalVBCount <= 0) {
+        // not an expected situation, bail out and return the full high
+        // watermark
+        return stats.mem_high_wat.load();
+    }
+
+    const double activePendingHighWat =
+            (stats.mem_high_wat.load() / totalVBCount) *
+            (activeVBCount + pendingVBCount);
+
+    return activePendingHighWat;
+}
+
+size_t EphemeralBucket::getPageableMemLowWatermark() const {
+    // Ephemeral buckets can only page out non-replica memory (see comments
+    // in getPageableMemCurrent). As such, set pagable low watermark to
+    // a fraction of the overall low watermark based on what faction of
+    // vBuckets are active. Memory used by any dead vbs should be reclaimed
+    // soon, so ignore them here; they don't need to be allocated a portion
+    // of the quota.
+    const double activeVBCount = vbMap.getVBStateCount(vbucket_state_active);
+    const double pendingVBCount = vbMap.getVBStateCount(vbucket_state_pending);
+    const double replicaVBCount = vbMap.getVBStateCount(vbucket_state_replica);
+    const double totalVBCount = activeVBCount + pendingVBCount + replicaVBCount;
+
+    if (totalVBCount <= 0) {
+        // not an expected situation, bail out and return the full low
+        // watermark
+        return stats.mem_low_wat.load();
+    }
+
+    const double activeLowWat = (stats.mem_low_wat.load() / totalVBCount) *
+                                (activeVBCount + pendingVBCount);
+
+    return activeLowWat;
+}
+
 void EphemeralBucket::attemptToFreeMemory() {
     // Call down to the base class; do to whatever it can to free memory.
     KVBucket::attemptToFreeMemory();
@@ -200,28 +267,40 @@ VBucketPtr EphemeralBucket::makeVBucket(
     // 1. make_shared doesn't accept a Deleter
     // 2. allocate_shared has inconsistencies between platforms in calling
     //    alloc.destroy (libc++ doesn't call it)
-    return VBucketPtr(new EphemeralVBucket(id,
-                                           state,
-                                           stats,
-                                           engine.getCheckpointConfig(),
-                                           shard,
-                                           lastSeqno,
-                                           lastSnapStart,
-                                           lastSnapEnd,
-                                           std::move(table),
-                                           std::move(newSeqnoCb),
-                                           makeSyncWriteResolvedCB(),
-                                           makeSyncWriteCompleteCB(),
-                                           makeSeqnoAckCB(),
-                                           engine.getConfiguration(),
-                                           eviction_policy,
-                                           std::move(manifest),
-                                           initState,
-                                           purgeSeqno,
-                                           maxCas,
-                                           mightContainXattrs,
-                                           replicationTopology),
-                      VBucket::DeferredDeleter(engine));
+    auto* vb = new EphemeralVBucket(id,
+                                    state,
+                                    stats,
+                                    engine.getCheckpointConfig(),
+                                    shard,
+                                    lastSeqno,
+                                    lastSnapStart,
+                                    lastSnapEnd,
+                                    std::move(table),
+                                    std::move(newSeqnoCb),
+                                    makeSyncWriteResolvedCB(),
+                                    makeSyncWriteCompleteCB(),
+                                    makeSeqnoAckCB(),
+                                    engine.getConfiguration(),
+                                    eviction_policy,
+                                    std::move(manifest),
+                                    initState,
+                                    purgeSeqno,
+                                    maxCas,
+                                    mightContainXattrs,
+                                    replicationTopology);
+
+    vb->ht.setMemChangedCallback([this, vb](int64_t delta) {
+        if (vb->getState() == vbucket_state_replica) {
+            this->stats.replicaHTMemory += delta;
+        }
+    });
+    vb->checkpointManager->setOverheadChangedCallback(
+            [this, vb](int64_t delta) {
+                if (vb->getState() == vbucket_state_replica) {
+                    this->stats.replicaCheckpointOverhead += delta;
+                }
+            });
+    return VBucketPtr(vb, VBucket::DeferredDeleter(engine));
 }
 
 void EphemeralBucket::completeStatsVKey(const void* cookie,

engines/ep/src/ephemeral_bucket.h

@@ -62,6 +62,10 @@ class EphemeralBucket : public KVBucket {
         return ENGINE_KEY_ENOENT;
     }
 
+    size_t getPageableMemCurrent() const override;
+    size_t getPageableMemHighWatermark() const override;
+    size_t getPageableMemLowWatermark() const override;
+
     void attemptToFreeMemory() override;
 
     /**