Merge commit neo/87dd4209e into master

paolococchi · paolococchi · commit 178bcae25c7a · 2023-09-14T17:32:32.000+02:00
* 87dd420 Revert "MB-57400: Use an estimated items_remaining count for Prometheus stats" * 6dcc716 Revert "MB-57400: "estimated" items_remaining drops to 0 when cursor settles" * c328ad5 MB-52276: Remove unused 'numItems' args in DCPTest::removeCheckpoint() * 25598d6 MB-52276: Remove unused arg in STActiveStreamTest::setupProducer * 69da985 [BP] MB-52276: Remove broken and unused Checkpoint::numMetaItems * dcaec69 [BP] MB-52276: Remove unused Checkpoint::numItems * 831cc2a [BP] MB-52276: Make ActiveStream::getItemsRemaining consistent * 3af4664 [BP] MB-52276: Make CheckpointCursor::getRemainingItemsCount O(1) * d324f13 MB-52276: Fix StreamTest.BackfillSmallBuffer * 27a56dc [BP] MB-52276: Fix CheckpointCursor::getRemainingItemsCount * f2b1d9e [BP] MB-52276: Checkpoint::getNumItems() accounts all items * 2b6245d [BP] MB-52276: Don't use Checkpoint::numItems at CM::maybeCreateNewCheckpoint * 52d3e96 [BP] MB-52276: Don't use Checkpoint::numItems at CM::getVisibleSnapshotEndSeqno * 88210c5 [BP] MB-52276: Don't use Checkpoint::numItems at CM::getSnapshotInfo * f2652b6 [BP] MB-52276: Don't rely on Checkpoint::numItems at ItemExpel Change-Id: I170fcd4544cbfaaf37e7dc55feff63e5fb7956ca
diff --git a/engines/ep/src/dcp/active_stream.h b/engines/ep/src/dcp/active_stream.h
@@ -291,7 +291,8 @@ class ActiveStream : public Stream,
                           CookieIface& c,
                           const VBucket& vb);
 
-    /* Returns a count of how many items are outstanding to be sent for this
+    /**
+     * Returns a count of how many items are outstanding to be sent for this
      * stream's vBucket.
      */
     size_t getItemsRemaining();
diff --git a/engines/ep/src/dcp/producer.h b/engines/ep/src/dcp/producer.h
@@ -385,6 +385,11 @@ class DcpProducer : public ConnHandler,
     */
     void scheduleCheckpointProcessorTask(std::shared_ptr<ActiveStream> s);
 
+    /**
+     * Schedule the checkpointCreatorTask on the ExecutorPool
+     */
+    void scheduleCheckpointProcessorTask();
+
     /** Searches the streams map for a stream for vbucket ID. Returns the
      *  found stream, or an empty pointer if none found.
      */
@@ -449,11 +454,6 @@ class DcpProducer : public ConnHandler,
      */
     void createCheckpointProcessorTask();
 
-    /**
-     * Schedule the checkpointCreatorTask on the ExecutorPool
-     */
-    void scheduleCheckpointProcessorTask();
-
     struct {
         // The time of the last noop message sent by this producer
         std::chrono::steady_clock::time_point sendTime;
diff --git a/engines/ep/src/ep_engine.cc b/engines/ep/src/ep_engine.cc
@@ -4047,6 +4047,11 @@ struct ConnStatBuilder {
 };
 
 struct ConnAggStatBuilder {
+    /**
+     * Construct with the separator.
+     * @param sep The separator used for determining "type" of DCP connection
+     *            by splitting the connection name with sep.
+     */
     ConnAggStatBuilder(std::string_view sep) : sep(sep) {
     }
 
diff --git a/engines/ep/tests/mock/mock_dcp_producer.h b/engines/ep/tests/mock/mock_dcp_producer.h
@@ -110,13 +110,6 @@ class MockDcpProducer : public DcpProducer {
         DcpProducer::createCheckpointProcessorTask();
     }
 
-    /**
-     * Schedule the checkpointCreator->task on the ExecutorPool
-     */
-    void scheduleCheckpointProcessorTask() {
-        DcpProducer::scheduleCheckpointProcessorTask();
-    }
-
     ActiveStreamCheckpointProcessorTask* getCheckpointSnapshotTask() const;
 
     /**
diff --git a/engines/ep/tests/module_tests/dcp_durability_stream_test.cc b/engines/ep/tests/module_tests/dcp_durability_stream_test.cc
@@ -39,15 +39,14 @@
 
 void DurabilityActiveStreamTest::SetUp() {
     SingleThreadedActiveStreamTest::SetUp();
-    setUp(false /*startCheckpointProcessorTask*/);
+    setUp();
 }
 
 void DurabilityActiveStreamTest::TearDown() {
     SingleThreadedActiveStreamTest::TearDown();
 }
 
-void DurabilityActiveStreamTest::setUp(bool startCheckpointProcessorTask,
-                                       bool persist) {
+void DurabilityActiveStreamTest::setUp(bool persist) {
     setVBucketState(vbid,
                     vbucket_state_active,
                     {{"topology", nlohmann::json::array({{active, replica}})}});
@@ -62,8 +61,7 @@ void DurabilityActiveStreamTest::setUp(bool startCheckpointProcessorTask,
     // Enable SyncReplication and flow-control (Producer BufferLog)
     setupProducer({{"enable_sync_writes", "true"},
                    {"connection_buffer_size", "52428800"},
-                   {"consumer_name", "test_consumer"}},
-                  startCheckpointProcessorTask);
+                   {"consumer_name", "test_consumer"}});
     ASSERT_TRUE(stream->public_supportSyncReplication());
 }
 
@@ -391,7 +389,6 @@ TEST_P(DurabilityActiveStreamTest, SendDcpAbort) {
 }
 
 TEST_P(DurabilityActiveStreamTest, BackfillDurabilityLevel) {
-    startCheckpointTask();
     auto vb = engine->getVBucket(vbid);
     auto& ckptMgr = *vb->checkpointManager;
     // Get rid of set_vb_state and any other queue_op we are not interested in
@@ -4661,8 +4658,7 @@ void DurabilityPromotionStreamTest::testDiskCheckpointStreamedAsDiskSnapshot() {
     }
 
     // 3) Set up the Producer and ActiveStream
-    DurabilityActiveStreamTest::setUp(true /*startCheckpointProcessorTask*/,
-                                      false /*persist*/);
+    DurabilityActiveStreamTest::setUp(false /*persist*/);
 
     // The vbstate-change must have:
     // - closed the checkpoint snap{2, 4, Disk}
@@ -4931,8 +4927,7 @@ void DurabilityPromotionStreamTest::
 
     // 3) Simulate vbstate-change Replica->Active (we have also a Producer and
     // ActiveStream from this point onward)
-    DurabilityActiveStreamTest::setUp(true /*startCheckpointProcessorTask*/,
-                                      false /*persist*/);
+    DurabilityActiveStreamTest::setUp(false /*persist*/);
     ASSERT_TRUE(producer);
     auto* activeStream = DurabilityActiveStreamTest::stream.get();
     ASSERT_TRUE(activeStream);
@@ -5187,8 +5182,7 @@ void DurabilityPromotionStreamTest::
 
     // 3) Simulate vbstate-change Replica->Active (we have also a Producer and
     // ActiveStream from this point onward)
-    DurabilityActiveStreamTest::setUp(true /*startCheckpointProcessorTask*/,
-                                      false /*persist*/);
+    DurabilityActiveStreamTest::setUp(false /*persist*/);
     ASSERT_TRUE(producer);
     auto* activeStream = DurabilityActiveStreamTest::stream.get();
     ASSERT_TRUE(activeStream);
@@ -5392,8 +5386,7 @@ TEST_P(DurabilityPromotionStreamTest,
 
     // Simulate vbstate-change Replica->Active (we have also a Producer and
     // ActiveStream from this point onward)
-    DurabilityActiveStreamTest::setUp(true /*startCheckpointProcessorTask*/,
-                                      false /*persist*/);
+    DurabilityActiveStreamTest::setUp(false /*persist*/);
     ASSERT_TRUE(producer);
     auto* activeStream = DurabilityActiveStreamTest::stream.get();
     ASSERT_TRUE(activeStream);
@@ -5528,8 +5521,7 @@ TEST_P(DurabilityPromotionStreamTest,
 
     // Simulate vbstate-change Replica->Active (we have also a Producer and
     // ActiveStream from this point onward)
-    DurabilityActiveStreamTest::setUp(true /*startCheckpointProcessorTask*/,
-                                      false /*persist*/);
+    DurabilityActiveStreamTest::setUp(false /*persist*/);
     ASSERT_TRUE(producer);
     auto* activeStream = DurabilityActiveStreamTest::stream.get();
     ASSERT_TRUE(activeStream);
@@ -5622,7 +5614,7 @@ TEST_P(DurabilityPromotionStreamTest, ReplicaDeadActiveCanCommitPrepare) {
 
     // Simulate vbstate-change Dead->Active (we have also a Producer and
     // ActiveStream from this point onward)
-    DurabilityActiveStreamTest::setUp(true /*startCheckpointProcessorTask*/);
+    DurabilityActiveStreamTest::setUp();
     ASSERT_TRUE(producer);
     auto* activeStream = DurabilityActiveStreamTest::stream.get();
     ASSERT_TRUE(activeStream);
diff --git a/engines/ep/tests/module_tests/dcp_durability_stream_test.h b/engines/ep/tests/module_tests/dcp_durability_stream_test.h
@@ -25,7 +25,7 @@ class DurabilityActiveStreamTest
     /**
      * Does the DurabilityActiveStreamTest specific setup
      */
-    void setUp(bool startCheckpointProcessorTask, bool persist = true);
+    void setUp(bool persist = true);
 
     /*
      * Queues a Prepare and verifies that the corresponding DCP_PREPARE
diff --git a/engines/ep/tests/module_tests/dcp_stream_ephemeral_test.cc b/engines/ep/tests/module_tests/dcp_stream_ephemeral_test.cc
@@ -90,7 +90,7 @@ TEST_P(EphemeralStreamTest, EphemeralBackfillSnapshotHasNoDuplicates) {
         store_item(vbid, "key2", "value1");
     }
 
-    removeCheckpoint(numItems);
+    removeCheckpoint();
 
     /* Set up a DCP stream for the backfill */
     setup_dcp_stream();
diff --git a/engines/ep/tests/module_tests/dcp_stream_sync_repl_test.cc b/engines/ep/tests/module_tests/dcp_stream_sync_repl_test.cc
@@ -545,7 +545,7 @@ void DcpStreamSyncReplTest::testBackfillPrepare(DocumentState docState,
     // Store a pending item then remove the checkpoint to force backfill.
     using cb::durability::Level;
     auto prepared = storePending(docState, "1", "X", {level, {}});
-    removeCheckpoint(1);
+    removeCheckpoint();
 
     // Create sync repl DCP stream
     setup_dcp_stream(0,
@@ -623,7 +623,7 @@ void DcpStreamSyncReplTest::testBackfillPrepareCommit(
     EXPECT_EQ(1, pdm.getNumTracked());
     EXPECT_EQ(0, pdm.getHighCompletedSeqno());
 
-    removeCheckpoint(1);
+    removeCheckpoint();
     EXPECT_EQ(1, vb0->getPersistenceSeqno());
 
     EXPECT_EQ(cb::engine_errc::success,
@@ -635,7 +635,7 @@ void DcpStreamSyncReplTest::testBackfillPrepareCommit(
     EXPECT_EQ(0, pdm.getNumTracked());
     EXPECT_EQ(1, pdm.getHighCompletedSeqno());
 
-    removeCheckpoint(1);
+    removeCheckpoint();
     EXPECT_EQ(2, vb0->getPersistenceSeqno());
 
     // Create sync repl DCP stream
@@ -730,7 +730,7 @@ void DcpStreamSyncReplTest::testBackfillPrepareAbort(
                              {},
                              vb0->lockCollections(prepared->getKey())));
     }
-    removeCheckpoint(2);
+    removeCheckpoint();
 
     // Create sync repl DCP stream
     setup_dcp_stream(0,
diff --git a/engines/ep/tests/module_tests/dcp_stream_test.cc b/engines/ep/tests/module_tests/dcp_stream_test.cc
@@ -885,12 +885,12 @@ TEST_P(StreamTest, BackfillSmallBuffer) {
        size of a mutation */
     producer->setBackfillBufferSize(1);
 
+    ASSERT_TRUE(stream->isBackfilling());
+    ASSERT_EQ(stream->getNumBackfillPauses(), 0);
+
     /* We want the backfill task to run in a background thread */
     ExecutorPool::get()->setNumAuxIO(ThreadPoolConfig::AuxIoThreadCount{1});
 
-    EXPECT_EQ(stream->getNumBackfillPauses(), 0);
-    stream->transitionStateToBackfilling();
-
     /* Backfill can only read 1 as its buffer will become full after that */
     cb::waitForPredicate(
             [&] { return (numItems - 1) == stream->getLastBackfilledSeqno(); });
@@ -1404,7 +1404,7 @@ class CacheCallbackTest : public StreamTest {
         StreamTest::SetUp();
         store_item(vbid, key, "value");
 
-        removeCheckpoint(numItems);
+        removeCheckpoint();
 
         /* Set up a DCP stream for the backfill */
         setup_dcp_stream();
@@ -1571,16 +1571,8 @@ void SingleThreadedActiveStreamTest::TearDown() {
     STParameterizedBucketTest::TearDown();
 }
 
-void SingleThreadedActiveStreamTest::startCheckpointTask() {
-    if (!producer->getCheckpointSnapshotTask()) {
-        producer->createCheckpointProcessorTask();
-        producer->scheduleCheckpointProcessorTask();
-    }
-}
-
 void SingleThreadedActiveStreamTest::setupProducer(
-        const std::vector<std::pair<std::string, std::string>>& controls,
-        bool startCheckpointProcessorTask) {
+        const std::vector<std::pair<std::string, std::string>>& controls) {
     uint32_t flags = 0;
 
     // We don't set the startTask flag here because we will create the task
@@ -1592,10 +1584,7 @@ void SingleThreadedActiveStreamTest::setupProducer(
                                                  flags,
                                                  false /*startTask*/);
     producer->createCheckpointProcessorTask();
-
-    if (startCheckpointProcessorTask) {
-        producer->scheduleCheckpointProcessorTask();
-    }
+    producer->scheduleCheckpointProcessorTask();
 
     for (const auto& c : controls) {
         EXPECT_EQ(cb::engine_errc::success,
@@ -3433,7 +3422,7 @@ TEST_P(SingleThreadedActiveToPassiveStreamTest,
 TEST_P(SingleThreadedActiveStreamTest,
        FirstSnapshotHasRequestedStartSnapSeqno) {
     // Replace initial stream with one registered with DCP producer.
-    setupProducer({}, true);
+    setupProducer({});
 
     auto& vb = *store->getVBucket(vbid);
     auto& cm = *vb.checkpointManager;
@@ -3486,7 +3475,7 @@ TEST_P(SingleThreadedActiveStreamTest,
  */
 TEST_P(SingleThreadedActiveStreamTest, MetaOnlyCheckpointsSkipped) {
     // Replace initial stream with one registered with DCP producer.
-    setupProducer({}, true);
+    setupProducer({});
 
     auto& vb = *store->getVBucket(vbid);
     auto& cm = *vb.checkpointManager;
@@ -3759,11 +3748,7 @@ TEST_P(SingleThreadedActiveStreamTest, CompleteBackfillRaceNoStreamEnd) {
 // seqno(s) until another mutation for that vBucket occurs.
 TEST_P(SingleThreadedActiveStreamTest,
        RaceBetweenNotifyAndProcessingExistingItems) {
-    // Replace initial stream with one registered with DCP producer.
-    setupProducer({}, true);
-
     auto vb = engine->getVBucket(vbid);
-    producer->scheduleCheckpointProcessorTask();
     stream = producer->mockActiveStreamRequest(0,
                                                /*opaque*/ 0,
                                                *vb,
@@ -6040,7 +6025,6 @@ void CDCActiveStreamTest::SetUp() {
               producer->control(0, DcpControlKeys::ChangeStreams, "true"));
     ASSERT_TRUE(producer->areChangeStreamsEnabled());
     producer->public_enableSyncReplication();
-    startCheckpointTask();
 
     recreateStream(*vb,
                    true,
diff --git a/engines/ep/tests/module_tests/dcp_stream_test.h b/engines/ep/tests/module_tests/dcp_stream_test.h
@@ -47,11 +47,8 @@ class SingleThreadedActiveStreamTest
     void SetUp() override;
     void TearDown() override;
 
-    void startCheckpointTask();
-
     void setupProducer(const std::vector<std::pair<std::string, std::string>>&
-                               controls = {},
-                       bool startCheckpointProcessorTask = false);
+                               controls = {});
 
     MutationStatus public_processSet(VBucket& vb,
                                      Item& item,
diff --git a/engines/ep/tests/module_tests/dcp_test.cc b/engines/ep/tests/module_tests/dcp_test.cc
@@ -229,10 +229,10 @@ void DCPTest::addItemsAndRemoveCheckpoint(int numItems) {
         std::string key("key" + std::to_string(i));
         store_item(vbid, key, "value");
     }
-    removeCheckpoint(numItems);
+    removeCheckpoint();
 }
 
-void DCPTest::removeCheckpoint(int numItems) {
+void DCPTest::removeCheckpoint() {
     /* Create new checkpoint so that we can remove the current checkpoint
        and force a backfill in the DCP stream */
     auto& ckpt_mgr = *vb0->checkpointManager;
diff --git a/engines/ep/tests/module_tests/dcp_test.h b/engines/ep/tests/module_tests/dcp_test.h
@@ -92,7 +92,7 @@ class DCPTest : virtual public EventuallyPersistentEngineTest {
     /* Add items onto the vbucket and wait for the checkpoint to be removed */
     void addItemsAndRemoveCheckpoint(int numItems);
 
-    void removeCheckpoint(int numItems);
+    void removeCheckpoint();
 
     void runCheckpointProcessor(DcpMessageProducersIface& producers);
 
diff --git a/engines/ep/tests/module_tests/item_pager_test.cc b/engines/ep/tests/module_tests/item_pager_test.cc
@@ -2519,18 +2519,16 @@ TEST_P(MultiPagingVisitorTest, ItemPagerCreatesMultiplePagers) {
         return stats.getPreciseTotalMemoryUsed() > stats.mem_high_wat.load();
     };
 
-    auto itemCount = 0;
     auto i = 0;
     // items need to be persisted to be evicted, but flushing after each item
     // would be slow. Load to the HWM, then flush, then repeat if we dropped
     // below the HWM
     while (!aboveHWM()) {
-        itemCount += populateVbsUntil(
+        populateVbsUntil(
                 vbids, aboveHWM, "keys_" + std::to_string(i++) + "_", 500);
         // populateVbsUntil flushes and removes checkpoints which is required
         // as eviction will not touch dirty items
     }
-    ASSERT_GT(itemCount, 0);
 
     // make sure the item pager has been notified while we're above the HWM
     store->attemptToFreeMemory();
diff --git a/engines/ep/tests/module_tests/kv_bucket_test.cc b/engines/ep/tests/module_tests/kv_bucket_test.cc
@@ -346,12 +346,11 @@ void KVBucketTest::removeCheckpoint(VBucket& vb) {
     const auto& stats = engine->getEpStats();
     const auto pre = stats.itemsRemovedFromCheckpoints;
     auto& manager = *vb.checkpointManager;
-    ASSERT_EQ(1, manager.getNumCheckpoints());
-    // Note: open items + checkpoint_end
-    const auto expectedRemoved = manager.getNumOpenChkItems() + 1;
-    vb.checkpointManager->createNewCheckpoint();
-    flushVBucketToDiskIfPersistent(vb.getId(), 0);
-    EXPECT_EQ(pre + expectedRemoved, stats.itemsRemovedFromCheckpoints);
+    manager.createNewCheckpoint();
+    if (persistent()) {
+        flushVBucket(vb.getId());
+    }
+    EXPECT_GT(stats.itemsRemovedFromCheckpoints, pre);
 }
 
 void KVBucketTest::flushAndRemoveCheckpoints(Vbid vbid) {
diff --git a/engines/ep/tests/module_tests/kv_bucket_test.h b/engines/ep/tests/module_tests/kv_bucket_test.h
@@ -173,7 +173,7 @@ class KVBucketTest : virtual public ::testing::Test {
     void flushVBucketToDiskIfPersistent(Vbid vbid, int expected = 1);
 
     /**
-     * Removes the open checkpoint.
+     * Creates a new open checkpoint and removes all the closed ones.
      */
     void removeCheckpoint(VBucket& vb);
 
diff --git a/engines/ep/tests/module_tests/vbucket_durability_test.cc b/engines/ep/tests/module_tests/vbucket_durability_test.cc

Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ TEST_P(EphemeralStreamTest, EphemeralBackfillSnapshotHasNoDuplicates) {`
`90`	`90`	`store_item(vbid, "key2", "value1");`
`91`	`91`	`}`
`92`	`92`
`93`		`- removeCheckpoint(numItems);`
	`93`	`+ removeCheckpoint();`
`94`	`94`
`95`	`95`	`/* Set up a DCP stream for the backfill */`
`96`	`96`	`setup_dcp_stream();`