MB-30234: Revert "Always close replica-checkpoint on memory-snapshot-end"

This reverts commit 075614a (MB-30019)

Reverting this commit as this has resulted in a throughput degradation

Change-Id: I12ba07f8155c998c0c797bf88baa6fa569a404bf
Reviewed-on: http://review.couchbase.org/96014
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Dave Finlay <[email protected]>
Tested-by: Dave Finlay <[email protected]>

Author: Sriram Ganesan

engines/ep/CMakeLists.txt

@@ -280,7 +280,6 @@ if (COUCHBASE_KV_BUILD_UNIT_TESTS)
                    tests/module_tests/checkpoint_remover_test.cc
                    tests/module_tests/checkpoint_test.h
                    tests/module_tests/checkpoint_test.cc
-                   tests/module_tests/checkpoint_utils.h
                    tests/module_tests/collections/collection_dockey_test.cc
                    tests/module_tests/collections/evp_store_collections_dcp_test.cc
                    tests/module_tests/collections/evp_store_collections_eraser_test.cc

engines/ep/src/dcp/stream.cc

@@ -2722,17 +2722,23 @@ void PassiveStream::processSetVBucketState(SetVBucketState* state) {
 void PassiveStream::handleSnapshotEnd(VBucketPtr& vb, uint64_t byseqno) {
     if (byseqno == cur_snapshot_end.load()) {
         auto& ckptMgr = *vb->checkpointManager;
-
         if (cur_snapshot_type.load() == Snapshot::Disk &&
                 vb->isBackfillPhase()) {
             vb->setBackfillPhase(false);
+            const auto id = ckptMgr.getOpenCheckpointId() + 1;
+            ckptMgr.checkAndAddNewCheckpoint(id, *vb);
+        } else {
+            size_t mem_threshold = engine->getEpStats().mem_high_wat.load();
+            size_t mem_used =
+                    engine->getEpStats().getEstimatedTotalMemoryUsed();
+            /* We want to add a new replica checkpoint if the mem usage is above
+               high watermark (85%) */
+            if (mem_threshold < mem_used) {
+                const auto id = ckptMgr.getOpenCheckpointId() + 1;
+                ckptMgr.checkAndAddNewCheckpoint(id, *vb);
+            }
         }
 
-        // MB-30019: we always want to close the open checkpoint on replica
-        // vbuckets when the Consumer receives the snapshot-end mutation
-        ckptMgr.checkAndAddNewCheckpoint(ckptMgr.getOpenCheckpointId() + 1,
-                                         *vb);
-
         if (cur_snapshot_ack) {
             {
                 LockHolder lh(streamMutex);

engines/ep/src/dcp/stream.h

@@ -742,7 +742,7 @@ class PassiveStream : public Stream {
 
     bool transitionState(StreamState newState);
 
-    virtual ENGINE_ERROR_CODE processMutation(MutationResponse* mutation);
+    ENGINE_ERROR_CODE processMutation(MutationResponse* mutation);
 
     ENGINE_ERROR_CODE processDeletion(MutationResponse* deletion);
 
@@ -782,7 +782,7 @@ class PassiveStream : public Stream {
 
     void handleSnapshotEnd(VBucketPtr& vb, uint64_t byseqno);
 
-    virtual void processMarker(SnapshotMarker* marker);
+    void processMarker(SnapshotMarker* marker);
 
     void processSetVBucketState(SetVBucketState* state);
 

engines/ep/tests/mock/mock_stream.h

@@ -272,14 +272,6 @@ class MockPassiveStream : public PassiveStream {
         return PassiveStream::messageReceived(std::move(dcpResponse));
     }
 
-    void processMarker(SnapshotMarker* marker) override {
-        PassiveStream::processMarker(marker);
-    }
-
-    ENGINE_ERROR_CODE processMutation(MutationResponse* mutation) override {
-        return PassiveStream::processMutation(mutation);
-    }
-
     size_t getNumBufferItems() const {
         LockHolder lh(buffer.bufMutex);
         return buffer.messages.size();

engines/ep/tests/module_tests/checkpoint_remover_test.cc

@@ -16,9 +16,6 @@
  */
 
 #include "checkpoint_remover_test.h"
-
-#include "checkpoint_utils.h"
-
 #include <engines/ep/src/checkpoint_remover.h>
 
 #include "../mock/mock_dcp.h"
@@ -31,6 +28,12 @@ size_t CheckpointRemoverTest::getMaxCheckpointItems(VBucket& vb) {
     return vb.checkpointManager->getCheckpointConfig().getCheckpointMaxItems();
 }
 
+const CheckpointList&
+CheckpointManagerTestIntrospector::public_getCheckpointList(
+        CheckpointManager& checkpointManager) {
+    return checkpointManager.checkpointList;
+}
+
 /**
  * Check that the VBucketMap.getActiveVBucketsSortedByChkMgrMem() returns the
  * correct ordering of vBuckets, sorted from largest memory usage to smallest.
@@ -204,4 +207,4 @@ TEST_F(CheckpointRemoverEPTest, CursorDropMemoryFreed) {
 
     // There should only be the one checkpoint cursor now for persistence
     ASSERT_EQ(1, checkpointManager->getNumOfCursors());
-}
+}

engines/ep/tests/module_tests/checkpoint_remover_test.h

@@ -50,3 +50,13 @@ class CheckpointRemoverEPTest : public CheckpointRemoverTest {
         return dynamic_cast<EPBucket&>(*store);
     }
 };
+
+/**
+ * Stateless class used to gain privileged access into CheckpointManager for
+ * testing purposes. This is a friend class of CheckpointManager.
+ */
+class CheckpointManagerTestIntrospector {
+public:
+    static const CheckpointList& public_getCheckpointList(
+            CheckpointManager& checkpointManager);
+};

engines/ep/tests/module_tests/checkpoint_test.cc

@@ -15,8 +15,6 @@
  *   limitations under the License.
  */
 
-#include "checkpoint_test.h"
-
 #include "config.h"
 
 #include <algorithm>
@@ -25,7 +23,7 @@
 #include <vector>
 
 #include "checkpoint.h"
-#include "checkpoint_utils.h"
+#include "checkpoint_test.h"
 #include "configuration.h"
 #include "ep_vb.h"
 #include "failover-table.h"
@@ -34,8 +32,6 @@
 #include "tests/module_tests/test_helpers.h"
 #include "thread_gate.h"
 
-#include "../mock/mock_dcp_consumer.h"
-
 #include <engines/ep/src/ep_types.h>
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
@@ -1239,113 +1235,3 @@ TYPED_TEST(CheckpointTest,
     // Test - second item (duplicate key) should return false.
     EXPECT_FALSE(this->queueNewItem("key"));
 }
-
-/*
- * We always want to close the current open checkpoint on replica-vbuckets
- * when the Consumer receives the snapshotEnd mutation of a memory-snapshot.
- */
-TEST_F(SingleThreadedCheckpointTest,
-       MB30019_CloseReplicaCheckpointOnMemorySnapshotEnd) {
-    setVBucketStateAndRunPersistTask(vbid, vbucket_state_replica);
-    auto vb = store->getVBuckets().getBucket(vbid);
-    auto* ckptMgr = vb->checkpointManager.get();
-    ASSERT_NE(nullptr, ckptMgr);
-
-    // We must have only 1 open checkpoint
-    ASSERT_EQ(1, ckptMgr->getNumCheckpoints());
-    // We must have only one cursor (the persistence cursor), as there is no
-    // DCP producer for vbid
-    ASSERT_EQ(1, ckptMgr->getNumOfCursors());
-    // We must have only the checkpoint-open and the vbucket-state meta-items
-    // in the open checkpoint
-    ASSERT_EQ(2, ckptMgr->getNumItems());
-    ASSERT_EQ(0, ckptMgr->getNumOpenChkItems());
-
-    auto consumer =
-            std::make_shared<MockDcpConsumer>(*engine, cookie, "test-consumer");
-    auto passiveStream = std::static_pointer_cast<MockPassiveStream>(
-            consumer->makePassiveStream(
-                    *engine,
-                    consumer,
-                    "test-passive-stream",
-                    0 /* flags */,
-                    0 /* opaque */,
-                    vbid,
-                    0 /* startSeqno */,
-                    std::numeric_limits<uint64_t>::max() /* endSeqno */,
-                    0 /* vbUuid */,
-                    0 /* snapStartSeqno */,
-                    0 /* snapEndSeqno */,
-                    0 /* vb_high_seqno */));
-
-    const size_t snapshotEnd = 3;
-    // 1) the consumer receives the snapshot-marker
-    SnapshotMarker snapshotMarker(
-            0 /* opaque */,
-            vbid,
-            0 /* startSeqno */,
-            snapshotEnd /* endSeqno */,
-            dcp_marker_flag_t::MARKER_FLAG_MEMORY /* flags */);
-    passiveStream->processMarker(&snapshotMarker);
-
-    // 2) the consumer receives the mutations until (snapshotEnd -1)
-    size_t i = 1;
-    for (; i < snapshotEnd; i++) {
-        // Queue item
-        queued_item qi(new Item(makeStoredDocKey("key_" + std::to_string(i)),
-                                0 /*flags*/,
-                                0 /*expiry*/,
-                                "value",
-                                5 /*valueSize*/,
-                                PROTOCOL_BINARY_RAW_BYTES,
-                                0 /*cas*/,
-                                i /*bySeqno*/,
-                                vb->getId()));
-
-        MutationResponse mutation(std::move(qi), 0 /* opaque */);
-
-        // PassiveStream::processMutation does 2 things:
-        //     1) setWithMeta (which enqueues the item into the checkpoint)
-        //     2) calls PassiveStream::handleSnapshotEnd (which must close the
-        //             open checkpoint if the current mutation is the
-        //             snapshot-end)
-        passiveStream->processMutation(&mutation);
-    }
-    // We must have 2 items in the checkpoint now
-    ASSERT_EQ(snapshotEnd - 1, ckptMgr->getNumOpenChkItems());
-    // We still must have only 1 open checkpoint, as the consumer has not
-    // received the snapshot-end mutation
-    ASSERT_EQ(1, ckptMgr->getNumCheckpoints());
-
-    // 3) the consumer receives the snapshotEnd mutation
-    queued_item qi(
-            new Item(makeStoredDocKey("key_" + std::to_string(snapshotEnd)),
-                     0 /*flags*/,
-                     0 /*expiry*/,
-                     "value",
-                     5 /*valueSize*/,
-                     PROTOCOL_BINARY_RAW_BYTES,
-                     0 /*cas*/,
-                     i /*bySeqno*/,
-                     vb->getId()));
-    MutationResponse mutation(std::move(qi), 0 /* opaque */);
-    passiveStream->processMutation(&mutation);
-
-    // The consumer has received the snapshotEnd mutation, now we expect
-    // that a new (empty) open checkpoint has been created. So we must have
-    // 2 checkpoints in total (the closed and the new open one).
-    ASSERT_EQ(2, ckptMgr->getNumCheckpoints());
-
-    // Also, the new open checkpoint must be empty (all mutations are in the
-    // closed one)
-    const auto& ckptList =
-            CheckpointManagerTestIntrospector::public_getCheckpointList(
-                    *ckptMgr);
-    ASSERT_EQ(ckptList.back()->getId(), ckptList.front()->getId() + 1);
-    ASSERT_EQ(checkpoint_state::CHECKPOINT_CLOSED,
-              ckptList.front()->getState_UNLOCKED());
-    ASSERT_EQ(snapshotEnd, ckptList.front()->getNumItems());
-    ASSERT_EQ(checkpoint_state::CHECKPOINT_OPEN,
-              ckptList.back()->getState_UNLOCKED());
-    ASSERT_EQ(0, ckptList.back()->getNumItems());
-}

engines/ep/tests/module_tests/checkpoint_test.h

@@ -21,7 +21,6 @@
 
 #include "checkpoint_config.h"
 #include "configuration.h"
-#include "evp_store_single_threaded_test.h"
 
 #include <gtest/gtest.h>
 
@@ -61,9 +60,4 @@ class CheckpointTest : public ::testing::Test {
     std::shared_ptr<Callback<uint16_t> > callback;
     std::unique_ptr<V> vbucket;
     std::unique_ptr<CheckpointManager> manager;
-};
-
-/*
- * Test fixture for single-threaded Checkpoint tests
- */
-class SingleThreadedCheckpointTest : public SingleThreadedKVBucketTest {};
+};